Merge app/v3 into workstreams/main (fourth catch-up)

cmd/codeaf/app.go

@@ -52,6 +52,7 @@ import (
 	"github.com/Agent-Field/codeaf/internal/cellstore"
 	"github.com/Agent-Field/codeaf/internal/clipboard"
 	"github.com/Agent-Field/codeaf/internal/config"
+	"github.com/Agent-Field/codeaf/internal/directory"
 	"github.com/Agent-Field/codeaf/internal/effort"
 	"github.com/Agent-Field/codeaf/internal/enginehost"
 	"github.com/Agent-Field/codeaf/internal/executor"
@@ -552,12 +553,15 @@ func appConnOf(client *remote.Client, release func(), assembled bool) *appConn {
 
 // appBackend is the [appwire.Engine] over this process's conversations.
 type appBackend struct {
-	clientTools appClientTools
-	chats       appChats
-	pair        appPair
-	drives      *driveBook
-	sync        appSync
-	syncPending map[string][]appSyncFact
+	clientTools  appClientTools
+	pluginPauses map[string]appPluginPause
+	pluginOllama appPluginOllama
+	pluginAuth   map[string]*appPluginAttempt
+	chats        appChats
+	pair         appPair
+	drives       *driveBook
+	sync         appSync
+	syncPending  map[string][]appSyncFact
 
 	// devices observes existing sync traffic without starting a roster watch.
 	devices            *appDevices
@@ -1980,14 +1984,18 @@ func (b *appBackend) hold(a *appAgent, conn *appConn, workspace string) appwire.
 	return view
 }
 
-// appAskID is the stable id of one ask: the engine's own question key.
-func appAskID(q session.Question) string {
-	return string(q.Kind) + ":" + q.Token()
+// appAskID scopes connection questions to their agent because plugin completion events carry only this id.
+func appAskID(agent string, q session.Question) string {
+	id := string(q.Kind) + ":" + q.Token()
+	if q.Kind == session.QuestionConnect {
+		return agent + "/" + id
+	}
+	return id
 }
 
 // appAskOf is the engine's question as the app's ask.
 func appAskOf(agent string, q session.Question) appwire.Ask {
-	ask := appwire.Ask{ID: appAskID(q), Agent: agent, Kind: appwire.AskQuestion, Title: strings.TrimSpace(q.Head), Detail: strings.TrimSpace(q.Reason)}
+	ask := appwire.Ask{ID: appAskID(agent, q), Agent: agent, Kind: appwire.AskQuestion, Title: strings.TrimSpace(q.Head), Detail: strings.TrimSpace(q.Reason)}
 	if q.Kind == session.QuestionConsent {
 		ask.Kind, ask.Subkind = appwire.AskApproval, appwire.SubkindPermission
 		ask.Irreversible = q.Stakes == session.StakesIrreversible
@@ -2145,7 +2153,7 @@ func (b *appBackend) questionEvent(a *appAgent, id string, live *appLive, event
 			q.ConsentScopes, q.AlwaysAsk = nil, "critical"
 		}
 	}
-	askID := appAskID(q)
+	askID := appAskID(id, q)
 	switch event.Kind {
 	case session.EventQuestion:
 		if q.Subject.Kind == session.SubjectCall {
@@ -2180,6 +2188,9 @@ func (b *appBackend) questionEvent(a *appAgent, id string, live *appLive, event
 			return
 		}
 		live.box.Event(appwire.EventAskOpened, appwire.AskOpened{Ask: ask})
+		if q.Kind == session.QuestionConnect {
+			live.box.Event(appwire.EventPluginsNeeded, appwire.PluginsNeeded{Agent: id, Plugin: q.Subject.Name, Why: q.Reason, AskID: askID})
+		}
 		if moved {
 			live.box.Event(appwire.EventAgentChanged, appwire.AgentChanged{Agent: view})
 		}
@@ -2240,6 +2251,9 @@ func (b *appBackend) questionEvent(a *appAgent, id string, live *appLive, event
 			}
 		}
 		live.box.Event(appwire.EventAskClosed, appwire.AskClosed{Agent: id, ID: askID, By: by, Option: option, At: at})
+		if q.Kind == session.QuestionConnect {
+			live.box.Event(appwire.EventPluginsNeededDone, appwire.PluginsNeededDone{AskID: askID})
+		}
 		receipt := appwire.Entry{ID: appReceiptEntryID(askID), Agent: id, Kind: appwire.EntryReceipt, At: at, Text: words}
 		if scope != "" {
 			receipt.Text = "Always allowed · " + scope + " · you · " + at.Local().Format("15:04")
@@ -2295,6 +2309,7 @@ func (b *appBackend) died(a *appAgent, live *appLive) {
 	mine := false
 	var view appwire.Agent
 	var asks []appwire.Ask
+	pluginAsks := map[string]bool{}
 	b.locked(func() {
 		if mine = a.live == live; !mine {
 			return
@@ -2304,13 +2319,21 @@ func (b *appBackend) died(a *appAgent, live *appLive) {
 		a.view.State = appwire.StateStopped
 		view = a.view
 		asks = live.asks
+		for id, q := range live.questions {
+			if q.Kind == session.QuestionConnect {
+				pluginAsks[id] = true
+			}
+		}
 		live.asks, live.questions = nil, map[string]session.Question{}
 	})
 	if !mine {
 		return
 	}
 	at := b.now()
 	for _, ask := range asks {
+		if pluginAsks[ask.ID] {
+			live.box.Event(appwire.EventPluginsNeededDone, appwire.PluginsNeededDone{AskID: ask.ID})
+		}
 		live.box.Event(appwire.EventAskClosed, appwire.AskClosed{Agent: view.ID, ID: ask.ID, By: appwire.ClosedByStopped, At: at})
 		live.box.Event(appwire.EventEntryAdded, appwire.EntryAdded{Entry: appwire.Entry{
 			ID: appReceiptEntryID(ask.ID), Agent: view.ID, Kind: appwire.EntryReceipt, At: at, Text: appReceipt(appwire.ClosedByStopped, "", nil, at),
@@ -3388,6 +3411,7 @@ func (b *appBackend) answerAskScope(id, askID, option, text, scope string) error
 			refused = appwire.Errorf(appwire.CodeNotFound, "the ask %s is not open; it may have been answered elsewhere", askID)
 			return
 		}
+		q = b.connectedPluginQuestion(q)
 		key, refused = appValidateQuestionAnswer(askID, q, option, text, scope)
 		if refused != nil {
 			return
@@ -4483,11 +4507,16 @@ func (b *appBackend) Close() error {
 	var warmed chan struct{}
 	var balances *appBalances
 	var stopModelNotices func()
+	var pluginAttempts []*appPluginAttempt
 	b.locked(func() {
 		if already = b.closing; already {
 			return
 		}
 		b.closing = true
+		for id, attempt := range b.pluginAuth {
+			pluginAttempts = append(pluginAttempts, attempt)
+			delete(b.pluginAuth, id)
+		}
 		if b.left != nil {
 			close(b.left)
 		}
@@ -4515,6 +4544,12 @@ func (b *appBackend) Close() error {
 	if already {
 		return nil
 	}
+	for _, attempt := range pluginAttempts {
+		attempt.cancel()
+		if attempt.flow != nil {
+			attempt.flow.Cancel()
+		}
+	}
 	b.closePair()
 	if b.stopSyncedSettings != nil {
 		b.stopSyncedSettings()
@@ -4710,6 +4745,10 @@ func appEngineError(err error) error {
 	if appMissingKey(err) {
 		return &appwire.Error{Code: appwire.CodeEngine, Message: appNoProvider}
 	}
+	// A removed computer reads the removal and its remedy, never the wire's sentinel.
+	if errors.Is(err, directory.ErrRevoked) {
+		return &appwire.Error{Code: appwire.CodeEngine, Message: appRemoved}
+	}
 	return &appwire.Error{Code: appwire.CodeEngine, Message: redact.Secrets(plainWords(err.Error()))}
 }
 

cmd/codeaf/app_chats.go

@@ -24,6 +24,8 @@ import (
 type appChats struct {
 	start           sync.Once
 	byAgent         map[string]appwire.ChatRow
+	taking          map[string]int
+	landed          map[string]uint64
 	take            sync.Mutex
 	op              sync.Mutex
 	mu              sync.Mutex
@@ -352,13 +354,16 @@ func (b *appBackend) readChats(parent context.Context, s *syncsetup.Sync, feed d
 		}
 		out = append(out, row)
 	}
+	b.chats.markTaking(out, true)
 	cacheFacts := func() {
 		b.chats.mu.Lock()
 		defer b.chats.mu.Unlock()
 		b.chats.facts = facts
 	}
 	cacheFacts()
-	b.publishChats(out, version, emit)
+	// markTaking has already cleared a landed take on a row saying the chat is
+	// here, so only a row from before the take keeps its agent where it is.
+	b.publishChatRows(out, version, emit, func(r appwire.ChatRow) bool { return !b.chats.landedTake(r.Cell) })
 	return appwire.ChatsResult{Chats: out, Version: version}, nil
 }
 
@@ -371,6 +376,97 @@ func (c *appChats) newerChats(version uint64) (appwire.ChatsResult, bool) {
 	return appwire.ChatsResult{}, false
 }
 
+// A take is in flight from when agents.take asks for the chat until it ends,
+// so a client that stopped waiting can tell a move still copying from one that
+// never happened. A take that acquired the chat keeps saying so on every row
+// from before it until a row says the chat is here or a later fence holds it.
+// Only a directory reading (prune) can say a chat left the list: the cache
+// may be empty or stale, so rows from it never drop a landed take.
+func (c *appChats) markTaking(rows []appwire.ChatRow, prune bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	listed := make(map[string]bool, len(rows))
+	for i := range rows {
+		r := &rows[i]
+		listed[r.Cell] = true
+		fence, landed := c.landed[r.Cell]
+		if landed && (r.Here || r.Seen.Fence > fence) {
+			delete(c.landed, r.Cell)
+			landed = false
+		}
+		r.Taking = c.taking[r.Cell] > 0 || landed
+	}
+	if !prune {
+		return
+	}
+	// A chat that left the list is no longer anyone's take.
+	for id := range c.landed {
+		if !listed[id] {
+			delete(c.landed, id)
+		}
+	}
+}
+
+func (c *appChats) landedTake(id string) bool {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	_, ok := c.landed[id]
+	return ok
+}
+
+func (c *appChats) landTake(id string, fence uint64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.landed == nil {
+		c.landed = make(map[string]uint64)
+	}
+	c.landed[id] = fence
+}
+
+func (c *appChats) countTake(id string, delta int) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.taking == nil {
+		c.taking = make(map[string]int)
+	}
+	c.taking[id] += delta
+	if c.taking[id] <= 0 {
+		delete(c.taking, id)
+	}
+}
+
+// The take's own start and end change no directory version, so its row is
+// emitted again from the cache.
+func (b *appBackend) publishTaking() {
+	c := &b.chats
+	c.op.Lock()
+	defer c.op.Unlock()
+	cached := func() ([]appwire.ChatRow, uint64, bool) {
+		c.mu.Lock()
+		defer c.mu.Unlock()
+		return slices.Clone(c.rows), c.version, !c.closed && c.source != nil
+	}
+	rows, version, following := cached()
+	if !following {
+		return
+	}
+	c.markTaking(rows, false)
+	// Only taking changed, so the cache's older locations stay off the agents
+	// a take may already have placed here.
+	b.publishChatRows(rows, version, true, nil)
+}
+
+// A finished take stops counting under the inventory lock, and this reading
+// clears its taking with the row that says the chat is here.
+func (b *appBackend) settleTake(ctx context.Context, s *syncsetup.Sync, id string) error {
+	c := &b.chats
+	c.op.Lock()
+	defer c.op.Unlock()
+	c.countTake(id, -1)
+	_, err := b.readChats(ctx, s, c.feed, true)
+	return err
+}
+
 func appChatPresence(row *appwire.ChatRow, fact chatlist.Row, state dirwatch.State, known map[string]bool) {
 	row.Online = nil
 	if fact.DeviceID != "" {
@@ -385,6 +481,11 @@ func appChatPresence(row *appwire.ChatRow, fact chatlist.Row, state dirwatch.Sta
 // Inventory operations hold op through caching and emission. A list refreshes
 // the cache silently; only the follower compares against rows actually emitted.
 func (b *appBackend) publishChats(rows []appwire.ChatRow, version uint64, emit bool) {
+	b.publishChatRows(rows, version, emit, func(appwire.ChatRow) bool { return true })
+}
+
+// locate says which rows re-place their local agents; nil re-places none.
+func (b *appBackend) publishChatRows(rows []appwire.ChatRow, version uint64, emit bool, locate func(appwire.ChatRow) bool) {
 	old, current := b.chats.cacheRows(rows, version, emit)
 	if !current || !emit {
 		return
@@ -399,7 +500,7 @@ func (b *appBackend) publishChats(rows []appwire.ChatRow, version uint64, emit b
 		if !exists || !reflect.DeepEqual(prev, r) {
 			b.emit.Emit(appwire.EventChatChanged, appwire.ChatChanged{Chat: r, Version: version})
 		}
-		if r.LocalAgent != "" {
+		if locate != nil && r.LocalAgent != "" && locate(r) {
 			var a *appAgent
 			b.locked(func() { a = b.agents[r.LocalAgent] })
 			if a != nil {
@@ -494,7 +595,10 @@ func (b *appBackend) refreshChatInventory(ctx context.Context, feed dirwatch.Fol
 			r := &rows[i]
 			appChatPresence(r, c.facts[r.Cell], state, known)
 		}
-		b.publishChats(rows, version, true)
+		c.markTaking(rows, false)
+		// Presence still refreshes local agents' lines, but a cached row from
+		// before a landed take would move its agent back elsewhere.
+		b.publishChatRows(rows, version, true, func(r appwire.ChatRow) bool { return !c.landedTake(r.Cell) })
 	}
 	return true
 }