Vectrum
Sign In
CodeyPublic
Like0 likesFork0 forks

serve: subagents, and the turn that waits for them

Subagents were left out of the ACP front end because their reports had
nowhere to be shown. They do now: each child gets a tool call row — the
only container the protocol has for work with a beginning, a middle and
an end — which opens when it starts, fills in as it works, and closes
with its report.

A turn stays open while its children are still working. Ending it would
leave their rows nowhere to appear and their reports nobody to be handed
to, and the thread would look finished while the machine was busy. Each
report the parent is entitled to comes back as a message, which is what
the parent's own prompt promised would happen, and that runs the agent
again; the round budget counts deliveries only, so waking up to find
nothing to hand over cannot end a long job's own turn. Cancelling breaks
out of all of it and the next turn picks the reports up.

The factory now takes a request struct: a session's supervisor needs the
same approval policy as the session, because what you allow you allow for
the conversation rather than for one agent in it. Selecting another model
rebuilds the agent and keeps the children — they belong to the
conversation, not to the model the parent happens to be on.

Two things the live run found: the progress clock was per emitter, so
whichever child spoke last silenced the others until its window was up,
and the text buffer was not flushed between report rounds, which glued
the end of one answer to the start of the next.
Browse Files
Jjonicommitted 1 parent 919c67dcommite8d9ead

Showing 10 changed files with 883 additions and 60 deletions.

+9−0README.mdView File
@@ -1043,6 +1043,15 @@ provider card will say Codey is installed and not logged in, which is true.
10431043 Threads stream text and reasoning, fill a tool row in as the command talks,
10441044 raise an approval prompt you answer in T3 Code's own panel, and stop when you
10451045 press stop. The model picker is filled from the endpoint Codey is pointed at.
1046+
1047+ [Subagents](#subagents) work too. Each one gets a row of its own that opens
1048+ when the child starts, fills in as it works, and closes with its report, and
1049+ the reports reach the parent as messages exactly as they do in the TUI. A turn
1050+ stays open while its children are still working — their output belongs inside
1051+ the turn that started them, and a thread that looked finished while four agents
1052+ were still running would be lying. `-no-agents` in the launch arguments turns
1053+ them off.
1054+
10461055 What does not work is resuming a thread after a restart: Codey does not
10471056 advertise session loading, so T3 Code will not offer it.
10481057
+56−16cmd/codey/agentcmd.goView File
@@ -28,6 +28,7 @@ import (
2828 "codey/internal/llm"
2929 "codey/internal/metrics"
3030 "codey/internal/serve"
31+ "codey/internal/subagent"
3132 "codey/internal/tools"
3233 )
3334
@@ -193,6 +194,8 @@ func sortedModels(models []llm.Model) []llm.Model {
193194 type acpDeps struct {
194195 Permission serve.PermissionMode
195196 Client *llm.Client
197+ // NoAgents leaves the subagent tools out, so the agent works alone.
198+ NoAgents bool
196199 // ClientOpts rebuilds the client for another model. The model lives on the
197200 // client rather than on the agent, so a session that selects one needs a
198201 // client of its own.
@@ -209,21 +212,50 @@ type acpDeps struct {
209212 // the protocol, and one stray line desynchronises a parser that has no way to
210213 // resynchronise.
211214 func serveACP(ctx context.Context, deps acpDeps) error {
212- models, current := offeredModels(ctx, deps.Client)
215+ found := listModels(ctx, deps.Client)
216+
217+ // What a subagent runs under is its parent's settings without the model
218+ // listing: a child never chooses a model for a child of its own, and a
219+ // paragraph about a choice it will not make is one it pays for on every
220+ // request. The parent does get the listing, because it is the one that
221+ // chooses.
222+ subBase := deps.Base
223+ base := deps.Base
224+ base.Models = found
213225
214226 srv := serve.New(serve.Options{
215227 Version: version,
216228 Permission: deps.Permission,
217- Models: models,
218- DefaultModel: current,
229+ Models: offeredModels(found),
230+ DefaultModel: deps.Client.Model(),
219231
220- NewAgent: func(root, model string, state *agent.State) (*agent.Agent, error) {
232+ NewAgent: func(req serve.AgentRequest) (serve.AgentSet, error) {
221233 client := deps.Client
222- if model != "" && model != client.Model() {
223- client = llm.New(append(deps.ClientOpts, llm.WithModel(model))...)
234+ if req.Model != "" && req.Model != client.Model() {
235+ client = llm.New(append(deps.ClientOpts, llm.WithModel(req.Model))...)
236+ }
237+
238+ // One factory for the session and for every child of it: an agent
239+ // that could reach further than its parent would be a way around
240+ // the parent's limits.
241+ newTools := deps.Tools(req.Root)
242+ reg := newTools(req.State, deps.Log)
243+
244+ set := serve.AgentSet{}
245+ if !deps.NoAgents {
246+ sup := subagent.New(ctx, subagent.Options{
247+ Client: client,
248+ Log: deps.Log,
249+ Policy: req.Approve,
250+ Base: subBase,
251+ Tools: newTools,
252+ })
253+ reg.Register(sup.Tools()...)
254+ set.Agents = sup
224255 }
225- reg := deps.Tools(root)(state, deps.Log)
226- return agent.New(client, reg, state, deps.Log, deps.Base), nil
256+
257+ set.Agent = agent.New(client, reg, req.State, deps.Log, base)
258+ return set, nil
227259 },
228260 })
229261
@@ -234,23 +266,31 @@ func serveACP(ctx context.Context, deps acpDeps) error {
234266 return srv.Run(ctx, os.Stdin, os.Stdout)
235267 }
236268
237- // offeredModels is the picker a session is given. An endpoint that will not
238- // list them leaves the picker empty, and the client then shows whatever it
239- // learned from the `models` probe instead.
240- func offeredModels(ctx context.Context, client *llm.Client) ([]acp.ModelInfo, string) {
269+ // listModels asks the endpoint what it serves. An endpoint that will not say
270+ // leaves the list empty, which is the honest answer: no list beats a made-up
271+ // one, and both the picker and the agent's own model choices degrade to the
272+ // model this process was configured with.
273+ func listModels(ctx context.Context, client *llm.Client) []llm.Model {
241274 listCtx, cancel := context.WithTimeout(ctx, modelsTimeout)
242275 defer cancel()
243276
244277 found, err := client.ListModels(listCtx)
245- if err != nil || len(found) == 0 {
246- return nil, client.Model()
278+ if err != nil {
279+ return nil
247280 }
281+ return sortedModels(found)
282+ }
248283
284+ // offeredModels is that list as the client's model picker wants it.
285+ func offeredModels(found []llm.Model) []acp.ModelInfo {
286+ if len(found) == 0 {
287+ return nil
288+ }
249289 out := make([]acp.ModelInfo, 0, len(found))
250- for _, m := range sortedModels(found) {
290+ for _, m := range found {
251291 out = append(out, acp.ModelInfo{ModelID: m.ID, Name: displayName(m)})
252292 }
253- return out, client.Model()
293+ return out
254294 }
255295
256296 func displayName(m llm.Model) string {
+1−0cmd/codey/main.goView File
@@ -383,6 +383,7 @@ func run() error {
383383 }
384384 return serveACP(context.Background(), acpDeps{
385385 Permission: inv.Permission,
386+ NoAgents: *noAgents,
386387 Client: client,
387388 ClientOpts: clientOpts,
388389 Log: log,
+7−2docs/superpowers/plans/2026-09-19-t3code-provider.mdView File
@@ -25,8 +25,13 @@
2525 > advertised and image blocks are converted, which was phase 3 in the plan and
2626 > turned out to be ten lines.
2727 >
28- > Phase 3 remains open. See the `## Inside another editor` section of README.md
29- > for what shipped.
28+ > Phase 3 is now mostly done as well: reasoning streams as `agent_thought_chunk`,
29+ > a tool's output streams into its row while the command is still running
30+ > (which the plan did not list at all, and is the gap you notice first), and
31+ > subagents each get a row that opens, fills and closes with their report.
32+ > What is left of it is session resume, `plan` updates and slash commands.
33+ >
34+ > See the `## Inside another editor` section of README.md for what shipped.
3035
3136 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
3237
+247−0internal/serve/agents.goaddedView File
@@ -0,0 +1,247 @@
1+ package serve
2+
3+ import (
4+ "fmt"
5+ "strings"
6+ "time"
7+
8+ "codey/internal/acp"
9+ "codey/internal/agent"
10+ "codey/internal/subagent"
11+ )
12+
13+ // A subagent is shown as a tool call, because a tool call is the only
14+ // container the protocol has for work with a beginning, a running middle and
15+ // an end. The row opens when the child starts, fills in as it works, and
16+ // closes with its report.
17+ //
18+ // The id is prefixed so it cannot collide with a real tool call: the parent's
19+ // own spawn_agent call is a row too, and the two are different things.
20+ const agentCallPrefix = "agent-"
21+
22+ func agentCallID(id string) string { return agentCallPrefix + id }
23+
24+ // watchAgents subscribes this session to its children. Both callbacks arrive
25+ // on the children's own goroutines, so everything they touch is guarded.
26+ func (s *session) watchAgents() {
27+ if s.agents == nil {
28+ return
29+ }
30+ s.agents.Subscribe(s.onAgentEvent, s.onAgentsChanged)
31+ }
32+
33+ // onAgentEvent is one child saying something.
34+ func (s *session) onAgentEvent(ev subagent.Event) {
35+ line := agentLine(ev.Agent)
36+ if line == "" {
37+ return
38+ }
39+ s.emit.agentProgress(ev, line)
40+ }
41+
42+ // onAgentsChanged is the supervisor telling us something moved. Statuses are
43+ // read from the snapshot rather than carried on the event, so this is also
44+ // where a child that has finished gets its row closed.
45+ func (s *session) onAgentsChanged() {
46+ if s.agents == nil {
47+ return
48+ }
49+ for _, info := range s.agents.Snapshot() {
50+ if info.Status.Live() {
51+ continue
52+ }
53+ s.emit.agentFinished(info)
54+ }
55+
56+ // Wake a turn that is waiting for children to finish. Buffered and
57+ // non-blocking: a signal nobody is waiting for is not worth keeping, and
58+ // the waiter re-reads the snapshot anyway.
59+ select {
60+ case s.agentsMoved <- struct{}{}:
61+ default:
62+ }
63+ }
64+
65+ // agentLine renders one of a child's events as a line for its row. Only the
66+ // things worth watching from outside are rendered: what it said, and what it
67+ // ran. Its usage and its turn boundaries are the parent's business, not the
68+ // reader's.
69+ func agentLine(ev agent.Event) string {
70+ switch ev.Kind {
71+ case agent.EventText:
72+ return strings.TrimSpace(ev.Text) + "\n"
73+ case agent.EventTextDelta:
74+ return ev.Text
75+ case agent.EventToolStart:
76+ return "\n· " + ev.Tool + "\n"
77+ default:
78+ return ""
79+ }
80+ }
81+
82+ // agentProgress opens a child's row if it is new and appends to it.
83+ func (e *emitter) agentProgress(ev subagent.Event, line string) {
84+ id := agentCallID(ev.ID)
85+
86+ e.mu.Lock()
87+ buf, known := e.running[id]
88+ if !known {
89+ buf = &strings.Builder{}
90+ e.running[id] = buf
91+ }
92+ buf.WriteString(line)
93+
94+ out := e.dueLocked(id, buf)
95+ e.mu.Unlock()
96+
97+ if !known {
98+ e.send(acp.Update{
99+ SessionUpdate: acp.UpdateToolCall,
100+ ToolCallID: id,
101+ Title: agentTitle(ev),
102+ Kind: acp.KindThink,
103+ Status: acp.StatusInProgress,
104+ })
105+ }
106+ if out == "" {
107+ return
108+ }
109+ e.send(acp.Update{
110+ SessionUpdate: acp.UpdateToolCallUpdate,
111+ ToolCallID: id,
112+ Status: acp.StatusInProgress,
113+ Content: []acp.ToolCallContent{acp.ToolCallText(out)},
114+ })
115+ }
116+
117+ // agentFinished closes a child's row with its report. It is called on every
118+ // change once the child has ended, so it has to be the one that decides it has
119+ // already been called.
120+ func (e *emitter) agentFinished(info subagent.Info) {
121+ id := agentCallID(info.ID)
122+
123+ e.mu.Lock()
124+ if e.closed == nil {
125+ e.closed = make(map[string]bool)
126+ }
127+ if e.closed[id] {
128+ e.mu.Unlock()
129+ return
130+ }
131+ e.closed[id] = true
132+ _, opened := e.running[id]
133+ e.forgetLocked(id)
134+ e.mu.Unlock()
135+
136+ // An agent that finished without ever saying anything — killed while
137+ // queued, or failed on its first request — has no row yet. Open one, so
138+ // what happened to it is visible rather than merely absent.
139+ if !opened {
140+ e.send(acp.Update{
141+ SessionUpdate: acp.UpdateToolCall,
142+ ToolCallID: id,
143+ Title: agentInfoTitle(info),
144+ Kind: acp.KindThink,
145+ Status: acp.StatusInProgress,
146+ })
147+ }
148+
149+ status := acp.StatusCompleted
150+ if info.Status != subagent.Done {
151+ status = acp.StatusFailed
152+ }
153+ e.send(acp.Update{
154+ SessionUpdate: acp.UpdateToolCallUpdate,
155+ ToolCallID: id,
156+ Title: agentInfoTitle(info),
157+ Status: status,
158+ Content: []acp.ToolCallContent{acp.ToolCallText(agentOutcome(info))},
159+ })
160+ }
161+
162+ func agentTitle(ev subagent.Event) string {
163+ title := fmt.Sprintf("subagent %s (%s)", ev.ID, ev.Preset)
164+ if ev.Model != "" {
165+ title += " on " + ev.Model
166+ }
167+ return title
168+ }
169+
170+ func agentInfoTitle(info subagent.Info) string {
171+ title := fmt.Sprintf("subagent %s (%s)", info.ID, info.Preset)
172+ if info.Model != "" {
173+ title += " on " + info.Model
174+ }
175+ if d := info.Elapsed(); d > 0 {
176+ title += fmt.Sprintf(" — %s", d.Round(time.Second))
177+ }
178+ return title
179+ }
180+
181+ // agentOutcome is what the closed row shows: the report when there is one, and
182+ // why there is not when there is not.
183+ func agentOutcome(info subagent.Info) string {
184+ var b strings.Builder
185+ fmt.Fprintf(&b, "%s (%s) %s after %d turns", info.ID, info.Preset, endedWord(info.Status), info.Turns)
186+ if cost, ok := info.Cost(); ok && cost > 0 {
187+ fmt.Fprintf(&b, ", $%.4f", cost)
188+ }
189+ b.WriteString("\n\n")
190+ switch {
191+ case info.Report != "":
192+ b.WriteString(info.Report)
193+ case info.Err != "":
194+ b.WriteString(info.Err)
195+ default:
196+ b.WriteString("It reported nothing.")
197+ }
198+ return b.String()
199+ }
200+
201+ func endedWord(s subagent.Status) string {
202+ switch s {
203+ case subagent.Done:
204+ return "finished"
205+ case subagent.Failed:
206+ return "failed"
207+ case subagent.Killed:
208+ return "was stopped"
209+ default:
210+ return string(s)
211+ }
212+ }
213+
214+ // reportBlocks folds the reports this session is still entitled to hand over
215+ // into one message for the parent.
216+ //
217+ // The claim is made here rather than when the report was drained: agent_wait
218+ // may have collected it in between, and putting it in again would charge the
219+ // model twice for something already in its context.
220+ func (s *session) reportBlocks() string {
221+ if s.agents == nil {
222+ return ""
223+ }
224+ var parts []string
225+ for _, r := range s.agents.TakeReports() {
226+ if !s.agents.ClaimReport(r.ID) {
227+ continue
228+ }
229+ parts = append(parts, r.Notification())
230+ }
231+ return strings.Join(parts, "\n\n")
232+ }
233+
234+ // agentsLive reports whether any child is still going to do something. A turn
235+ // stays open while one is: their updates belong inside the turn that started
236+ // them, and a report that arrives after the turn closed has nowhere to go.
237+ func (s *session) agentsLive() bool {
238+ if s.agents == nil {
239+ return false
240+ }
241+ for _, info := range s.agents.Snapshot() {
242+ if info.Status.Live() {
243+ return true
244+ }
245+ }
246+ return false
247+ }
+355−0internal/serve/agents_test.goaddedView File
@@ -0,0 +1,355 @@
1+ package serve
2+
3+ import (
4+ "context"
5+ "io"
6+ "strings"
7+ "sync"
8+ "testing"
9+ "time"
10+
11+ "codey/internal/acp"
12+ "codey/internal/agent"
13+ "codey/internal/llm"
14+ "codey/internal/metrics"
15+ "codey/internal/subagent"
16+ "codey/internal/tools"
17+ )
18+
19+ // fakeAgents is a supervisor the test drives by hand: no children, no model
20+ // calls, just the two callbacks and a snapshot the test controls.
21+ type fakeAgents struct {
22+ mu sync.Mutex
23+ observe func(subagent.Event)
24+ changed func()
25+ infos []subagent.Info
26+ reports []subagent.Info
27+ claimed map[string]bool
28+ shutdown bool
29+ }
30+
31+ func newFakeAgents() *fakeAgents { return &fakeAgents{claimed: map[string]bool{}} }
32+
33+ func (f *fakeAgents) Subscribe(observe func(subagent.Event), changed func()) {
34+ f.mu.Lock()
35+ defer f.mu.Unlock()
36+ f.observe, f.changed = observe, changed
37+ }
38+
39+ func (f *fakeAgents) Snapshot() []subagent.Info {
40+ f.mu.Lock()
41+ defer f.mu.Unlock()
42+ return append([]subagent.Info(nil), f.infos...)
43+ }
44+
45+ func (f *fakeAgents) TakeReports() []subagent.Info {
46+ f.mu.Lock()
47+ defer f.mu.Unlock()
48+ out := f.reports
49+ f.reports = nil
50+ return out
51+ }
52+
53+ func (f *fakeAgents) ClaimReport(id string) bool {
54+ f.mu.Lock()
55+ defer f.mu.Unlock()
56+ if f.claimed[id] {
57+ return false
58+ }
59+ f.claimed[id] = true
60+ return true
61+ }
62+
63+ func (f *fakeAgents) Shutdown() {
64+ f.mu.Lock()
65+ defer f.mu.Unlock()
66+ f.shutdown = true
67+ }
68+
69+ // say makes a child speak.
70+ func (f *fakeAgents) say(id, preset, text string) {
71+ f.mu.Lock()
72+ observe := f.observe
73+ f.mu.Unlock()
74+ observe(subagent.Event{ID: id, Preset: preset, Model: "test-model",
75+ Agent: agent.Event{Kind: agent.EventText, Text: text}})
76+ }
77+
78+ // finish ends a child, queues its report, and fires the change callback.
79+ func (f *fakeAgents) finish(info subagent.Info) {
80+ f.mu.Lock()
81+ f.infos = append(f.infos, info)
82+ if info.Status == subagent.Done {
83+ f.reports = append(f.reports, info)
84+ }
85+ changed := f.changed
86+ f.mu.Unlock()
87+ changed()
88+ }
89+
90+ // live adds a child that has not finished.
91+ func (f *fakeAgents) live(info subagent.Info) {
92+ f.mu.Lock()
93+ f.infos = append(f.infos, info)
94+ f.mu.Unlock()
95+ }
96+
97+ // --- the row translation, tested directly --------------------------------
98+
99+ func TestASubagentGetsItsOwnRow(t *testing.T) {
100+ e, r := newRecorded()
101+ sess := &session{emit: e, agentsMoved: make(chan struct{}, 1)}
102+ f := newFakeAgents()
103+ sess.agents = f
104+ sess.watchAgents()
105+
106+ f.say("a1", "explore", strings.Repeat("looking around. ", 40))
107+
108+ if len(r.updates) < 2 {
109+ t.Fatalf("want a row and an update, got %v", r.kinds())
110+ }
111+ open := r.updates[0]
112+ if open.SessionUpdate != acp.UpdateToolCall {
113+ t.Fatalf("first update = %q", open.SessionUpdate)
114+ }
115+ if open.ToolCallID != "agent-a1" {
116+ t.Errorf("toolCallId = %q — it must not collide with a real tool call", open.ToolCallID)
117+ }
118+ if !strings.Contains(open.Title, "a1") || !strings.Contains(open.Title, "explore") {
119+ t.Errorf("title = %q", open.Title)
120+ }
121+ if open.Status != acp.StatusInProgress {
122+ t.Errorf("status = %q", open.Status)
123+ }
124+ if !strings.Contains(contentText(t, r.updates[1]), "looking around") {
125+ t.Errorf("the child's words did not reach its row")
126+ }
127+ }
128+
129+ func TestARowIsOpenedOnceAndClosedOnce(t *testing.T) {
130+ e, r := newRecorded()
131+ sess := &session{emit: e, agentsMoved: make(chan struct{}, 1)}
132+ f := newFakeAgents()
133+ sess.agents = f
134+ sess.watchAgents()
135+
136+ f.say("a1", "explore", "one")
137+ r.clock = r.clock.Add(progressAfter)
138+ f.say("a1", "explore", "two")
139+
140+ info := subagent.Info{ID: "a1", Preset: "explore", Status: subagent.Done, Turns: 3, Report: "found it"}
141+ f.finish(info)
142+ // The change callback fires repeatedly once a child has ended.
143+ f.finish(info)
144+ f.finish(info)
145+
146+ var opens, closes int
147+ for _, u := range r.updates {
148+ switch {
149+ case u.SessionUpdate == acp.UpdateToolCall:
150+ opens++
151+ case u.Status == acp.StatusCompleted || u.Status == acp.StatusFailed:
152+ closes++
153+ }
154+ }
155+ if opens != 1 {
156+ t.Errorf("%d rows opened for one agent", opens)
157+ }
158+ if closes != 1 {
159+ t.Errorf("%d closing updates for one agent", closes)
160+ }
161+ if got := contentText(t, r.updates[len(r.updates)-1]); !strings.Contains(got, "found it") {
162+ t.Errorf("the row closed without the report: %q", got)
163+ }
164+ }
165+
166+ func TestAFailedSubagentClosesAsFailed(t *testing.T) {
167+ e, r := newRecorded()
168+ sess := &session{emit: e, agentsMoved: make(chan struct{}, 1)}
169+ f := newFakeAgents()
170+ sess.agents = f
171+ sess.watchAgents()
172+
173+ f.finish(subagent.Info{ID: "a1", Preset: "plan", Status: subagent.Failed, Err: "the endpoint refused"})
174+
175+ last := r.updates[len(r.updates)-1]
176+ if last.Status != acp.StatusFailed {
177+ t.Errorf("status = %q, want failed", last.Status)
178+ }
179+ if !strings.Contains(contentText(t, last), "the endpoint refused") {
180+ t.Errorf("content = %q", contentText(t, last))
181+ }
182+ // It never said anything, so the row has to be opened before it is closed
183+ // or there is nothing on screen to close.
184+ if r.updates[0].SessionUpdate != acp.UpdateToolCall {
185+ t.Errorf("a silent agent left no row: %v", r.kinds())
186+ }
187+ }
188+
189+ // --- the turn, over the wire ---------------------------------------------
190+
191+ // newAgentsServer wires a server whose sessions all share one fake supervisor.
192+ func newAgentsServer(t *testing.T, f *fakeAPI, agents Agents) *testClient {
193+ t.Helper()
194+ log := metrics.NewWriter(io.Discard, "test")
195+ return newTestServer(t, f, Options{
196+ NewAgent: func(req AgentRequest) (AgentSet, error) {
197+ reg := tools.NewRegistry(log).Register(tools.NewRead(req.Root, req.State))
198+ client := llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k"), llm.WithModel("test-model"))
199+ return AgentSet{
200+ Agent: agent.New(client, reg, req.State, log, agent.Options{}),
201+ Agents: agents,
202+ }, nil
203+ },
204+ })
205+ }
206+
207+ func TestATurnStaysOpenWhileASubagentWorks(t *testing.T) {
208+ // Ending the turn while a child is running would leave its row nowhere to
209+ // appear and its report nobody to be handed to.
210+ f := newFakeAPI(t,
211+ msg("end_turn", textBlock("spawned it")),
212+ msg("end_turn", textBlock("and here is what it found")),
213+ )
214+ agents := newFakeAgents()
215+ agents.live(subagent.Info{ID: "a1", Preset: "explore", Status: subagent.Running})
216+
217+ c := newAgentsServer(t, f, agents)
218+ id := c.handshake(t.TempDir())
219+
220+ done := make(chan acp.PromptResponse, 1)
221+ go func() { done <- c.prompt(id, "find the bug") }()
222+
223+ select {
224+ case <-done:
225+ t.Fatal("the turn ended while a subagent was still running")
226+ case <-time.After(150 * time.Millisecond):
227+ }
228+
229+ // The child finishes and reports. That restarts the turn with its report.
230+ agents.mu.Lock()
231+ agents.infos = nil
232+ agents.mu.Unlock()
233+ agents.finish(subagent.Info{ID: "a1", Preset: "explore", Status: subagent.Done, Turns: 2, Report: "it is the parser"})
234+
235+ select {
236+ case resp := <-done:
237+ if resp.StopReason != acp.StopEndTurn {
238+ t.Errorf("stopReason = %q", resp.StopReason)
239+ }
240+ case <-time.After(5 * time.Second):
241+ t.Fatal("the turn never ended")
242+ }
243+
244+ // The report has to have reached the model, not only the screen.
245+ f.mu.Lock()
246+ defer f.mu.Unlock()
247+ if len(f.requests) != 2 {
248+ t.Fatalf("the parent was asked %d times, want 2", len(f.requests))
249+ }
250+ if !strings.Contains(string(f.requests[1]), "it is the parser") {
251+ t.Error("the subagent's report was never handed to the parent")
252+ }
253+ }
254+
255+ func TestAReportAlreadyClaimedIsNotHandedOverTwice(t *testing.T) {
256+ // agent_wait may have collected it. Putting it in again would charge the
257+ // model twice for something already in its context.
258+ f := newFakeAPI(t, msg("end_turn", textBlock("done")))
259+ agents := newFakeAgents()
260+ agents.claimed["a1"] = true
261+ agents.reports = []subagent.Info{{ID: "a1", Preset: "explore", Status: subagent.Done, Report: "already seen"}}
262+
263+ c := newAgentsServer(t, f, agents)
264+ id := c.handshake(t.TempDir())
265+ c.prompt(id, "go")
266+
267+ f.mu.Lock()
268+ defer f.mu.Unlock()
269+ if len(f.requests) != 1 {
270+ t.Fatalf("the parent was asked %d times, want 1", len(f.requests))
271+ }
272+ }
273+
274+ func TestCancellingATurnStopsWaitingForChildren(t *testing.T) {
275+ f := newFakeAPI(t, msg("end_turn", textBlock("spawned")))
276+ agents := newFakeAgents()
277+ agents.live(subagent.Info{ID: "a1", Preset: "explore", Status: subagent.Running})
278+
279+ c := newAgentsServer(t, f, agents)
280+ id := c.handshake(t.TempDir())
281+
282+ done := make(chan acp.PromptResponse, 1)
283+ go func() { done <- c.prompt(id, "find it") }()
284+
285+ time.Sleep(100 * time.Millisecond)
286+ c.notify(acp.MethodCancel, acp.CancelParams{SessionID: id})
287+
288+ select {
289+ case resp := <-done:
290+ if resp.StopReason != acp.StopCancelled {
291+ t.Errorf("stopReason = %q, want cancelled", resp.StopReason)
292+ }
293+ case <-time.After(5 * time.Second):
294+ t.Fatal("cancelling did not break the wait")
295+ }
296+ }
297+
298+ func TestTheChildrenAreStoppedWhenTheConnectionEnds(t *testing.T) {
299+ f := newFakeAPI(t, msg("end_turn", textBlock("ok")))
300+ agents := newFakeAgents()
301+
302+ toAgent, fromClient := io.Pipe()
303+ fromAgent, toClient := io.Pipe()
304+ go func() { _, _ = io.Copy(io.Discard, fromAgent) }()
305+
306+ log := metrics.NewWriter(io.Discard, "test")
307+ srv := New(Options{
308+ NewAgent: func(req AgentRequest) (AgentSet, error) {
309+ reg := tools.NewRegistry(log).Register(tools.NewRead(req.Root, req.State))
310+ client := llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k"), llm.WithModel("m"))
311+ return AgentSet{Agent: agent.New(client, reg, req.State, log, agent.Options{}), Agents: agents}, nil
312+ },
313+ })
314+
315+ served := make(chan error, 1)
316+ go func() { served <- srv.Run(context.Background(), toAgent, toClient) }()
317+
318+ // Open a session, then hang up.
319+ if _, err := io.WriteString(fromClient, `{"jsonrpc":"2.0","id":1,"method":"session/new","params":{"cwd":`+quote(t.TempDir())+"}}\n"); err != nil {
320+ t.Fatal(err)
321+ }
322+ time.Sleep(100 * time.Millisecond)
323+ _ = fromClient.Close()
324+
325+ select {
326+ case <-served:
327+ case <-time.After(3 * time.Second):
328+ t.Fatal("Run did not return")
329+ }
330+
331+ agents.mu.Lock()
332+ defer agents.mu.Unlock()
333+ if !agents.shutdown {
334+ t.Error("a subagent outlived the connection, still editing a tree nobody is watching")
335+ }
336+ }
337+
338+ func TestSubagentRowsAreBoundedLikeAnyOtherToolOutput(t *testing.T) {
339+ e, r := newRecorded()
340+ sess := &session{emit: e, agentsMoved: make(chan struct{}, 1)}
341+ f := newFakeAgents()
342+ sess.agents = f
343+ sess.watchAgents()
344+
345+ f.say("a1", "explore", strings.Repeat("x", progressMaxBytes*2)+"END")
346+
347+ body := contentText(t, r.updates[len(r.updates)-1])
348+ if len(body) > progressMaxBytes+64 {
349+ t.Errorf("an unbounded %d bytes went over the wire", len(body))
350+ }
351+ // The child's line is newline-terminated, so trim before looking at the end.
352+ if !strings.HasSuffix(strings.TrimSpace(body), "END") {
353+ t.Error("the tail is the useful part and it was the part dropped")
354+ }
355+ }
+151−17internal/serve/serve.goView File
@@ -29,6 +29,7 @@ import (
2929 "codey/internal/agent"
3030 "codey/internal/approval"
3131 "codey/internal/llm"
32+ "codey/internal/subagent"
3233 )
3334
3435 // PermissionMode is how tool calls are gated, chosen on the command line by
@@ -66,14 +67,48 @@ func (m PermissionMode) allows(req approval.Request) bool {
6667 }
6768 }
6869
69- // NewAgentFunc builds the agent for one session.
70- //
71- // root is the directory the client asked to work in, which is not necessarily
72- // the one this process was started in — a client that manages worktrees hands
73- // out a different one per thread. model is empty for the default. state is the
74- // conversation, passed in rather than created here so that selecting another
75- // model can rebuild the agent around a history that survives the switch.
76- type NewAgentFunc func(root, model string, state *agent.State) (*agent.Agent, error)
70+ // AgentRequest is one session asking to be built.
71+ type AgentRequest struct {
72+ // Root is the directory the client asked to work in, which is not
73+ // necessarily the one this process was started in: a client that manages
74+ // worktrees hands out a different one per thread.
75+ Root string
76+ // Model is empty for whatever the process was configured with.
77+ Model string
78+ // State is the conversation. It is passed in rather than created by the
79+ // builder so that selecting another model can rebuild the agent around a
80+ // history that survives the switch.
81+ State *agent.State
82+ // Approve gates tool calls — the session's own and its subagents'. What
83+ // you allow, you allow for the conversation, not for one agent in it.
84+ Approve *approval.Policy
85+ }
86+
87+ // AgentSet is what comes back: the agent, and the subagents it may start.
88+ type AgentSet struct {
89+ Agent *agent.Agent
90+ // Agents is nil when this session may not spawn.
91+ Agents Agents
92+ }
93+
94+ // Agents is a session's subagent supervisor, as this package needs it.
95+ // *subagent.Supervisor is the implementation; the interface is here so a test
96+ // can stand in for one without starting real children.
97+ type Agents interface {
98+ // Subscribe registers the two callbacks, which are called from the
99+ // children's own goroutines.
100+ Subscribe(observe func(subagent.Event), changed func())
101+ // TakeReports drains the reports nobody has been handed yet, and
102+ // ClaimReport says whether this front end is still the one to hand a given
103+ // report over — agent_wait may have collected it in the meantime.
104+ TakeReports() []subagent.Info
105+ ClaimReport(id string) bool
106+ Snapshot() []subagent.Info
107+ Shutdown()
108+ }
109+
110+ // NewAgentFunc builds one session.
111+ type NewAgentFunc func(AgentRequest) (AgentSet, error)
77112
78113 // Options configures a [Server].
79114 type Options struct {
@@ -108,9 +143,28 @@ func New(opts Options) *Server {
108143 // a client says goodbye, and it returns nil.
109144 func (s *Server) Run(ctx context.Context, r io.Reader, w io.Writer) error {
110145 s.conn = acp.NewConn(r, w, s)
146+ defer s.shutdown()
111147 return s.conn.Serve(ctx)
112148 }
113149
150+ // shutdown stops every session's subagents. A child outliving the connection
151+ // would keep editing a tree nobody is watching, and nobody would be left to
152+ // read what it found.
153+ func (s *Server) shutdown() {
154+ s.mu.Lock()
155+ sessions := make([]*session, 0, len(s.sessions))
156+ for _, sess := range s.sessions {
157+ sessions = append(sessions, sess)
158+ }
159+ s.mu.Unlock()
160+
161+ for _, sess := range sessions {
162+ if sess.agents != nil {
163+ sess.agents.Shutdown()
164+ }
165+ }
166+ }
167+
114168 // Initialize answers the handshake.
115169 //
116170 // The protocol version we answer with is ours, not the client's: agreeing to a
@@ -164,10 +218,11 @@ func (s *Server) NewSession(_ context.Context, req acp.NewSessionRequest) (acp.N
164218 }
165219
166220 sess := &session{
167- id: newSessionID(),
168- root: req.Cwd,
169- state: agent.NewState(),
170- model: s.opts.DefaultModel,
221+ id: newSessionID(),
222+ root: req.Cwd,
223+ state: agent.NewState(),
224+ model: s.opts.DefaultModel,
225+ agentsMoved: make(chan struct{}, 1),
171226 }
172227 sess.emit = newEmitter(notifier(s.conn, sess.id))
173228 sess.policy = approval.New(sess.promptFor(s.conn, s.opts.Permission))
@@ -191,14 +246,30 @@ func (s *Server) NewSession(_ context.Context, req acp.NewSessionRequest) (acp.N
191246
192247 // build attaches an agent to a session, or replaces the one it has. The caller
193248 // holds whatever lock the session needs.
249+ //
250+ // A rebuild — which is what selecting another model is — reuses the subagents
251+ // the session already has. They belong to the conversation, not to the model
252+ // the parent happens to be running on, and killing four working children
253+ // because somebody changed a dropdown would be a surprising thing to do.
194254 func (s *Server) build(sess *session) error {
195- a, err := s.opts.NewAgent(sess.root, sess.model, sess.state)
255+ built, err := s.opts.NewAgent(AgentRequest{
256+ Root: sess.root,
257+ Model: sess.model,
258+ State: sess.state,
259+ Approve: sess.policy,
260+ })
196261 if err != nil {
197262 return acp.Errorf(acp.CodeInternal, "cannot start an agent in %s: %v", sess.root, err)
198263 }
199- a.Approve = sess.policy
200- a.Observe = sess.emit.handle
201- sess.agent = a
264+
265+ built.Agent.Approve = sess.policy
266+ built.Agent.Observe = sess.emit.handle
267+ sess.agent = built.Agent
268+
269+ if sess.agents == nil && built.Agents != nil {
270+ sess.agents = built.Agents
271+ sess.watchAgents()
272+ }
202273 return nil
203274 }
204275
@@ -220,7 +291,7 @@ func (s *Server) Prompt(ctx context.Context, req acp.PromptRequest) (acp.PromptR
220291 turnCtx := sess.begin(ctx)
221292 defer sess.end()
222293
223- _, runErr := sess.agent.RunBlocks(turnCtx, blocks)
294+ runErr := sess.runTurn(turnCtx, blocks)
224295
225296 // Whatever happened, say the last of what was written before reporting how
226297 // the turn ended. A buffer dropped on the way out is a sentence the user
@@ -248,6 +319,69 @@ func (s *Server) Prompt(ctx context.Context, req acp.PromptRequest) (acp.PromptR
248319 return acp.PromptResponse{StopReason: acp.StopEndTurn}, nil
249320 }
250321
322+ // maxReportRounds caps how many times one turn may be restarted to hand a
323+ // subagent's report to the parent. A parent that answers every report by
324+ // spawning another agent would otherwise never finish the turn it is in, and
325+ // a cap is a cheaper guard than reasoning about when that is reasonable.
326+ //
327+ // Hitting it ends the turn with the outstanding reports still queued and
328+ // unclaimed, so the next turn hands them over rather than losing them.
329+ const maxReportRounds = 32
330+
331+ // runTurn runs the agent, and then keeps the turn open for as long as this
332+ // session's subagents are still working.
333+ //
334+ // A turn is the protocol's unit of work, and a child's output belongs inside
335+ // the one that started it: end the turn while four agents are still running
336+ // and their rows have nowhere to appear, their reports have nobody to be
337+ // handed to, and the thread looks finished while the machine is busy. So the
338+ // turn ends when the work does.
339+ //
340+ // Each report the parent is entitled to comes back as a message, which is what
341+ // the parent's own prompt promised would happen, and that runs the agent
342+ // again. Cancelling breaks out of all of it: children keep working, and the
343+ // next turn will pick their reports up.
344+ func (s *session) runTurn(ctx context.Context, blocks []llm.ContentBlock) error {
345+ if _, err := s.agent.RunBlocks(ctx, blocks); err != nil {
346+ return err
347+ }
348+
349+ for rounds := 0; ; {
350+ if ctx.Err() != nil {
351+ return ctx.Err()
352+ }
353+ // Counted against report deliveries only. Waking up because a child
354+ // changed status and finding nothing to hand over costs nothing and
355+ // must not eat the budget, or a long job would end its own turn.
356+ if rounds >= maxReportRounds {
357+ return nil
358+ }
359+
360+ if text := s.reportBlocks(); text != "" {
361+ rounds++
362+ // What the agent said before the report arrived is finished. Left
363+ // buffered, the end of one message and the start of the next would
364+ // be glued into one chunk, and the client would show two answers
365+ // as a single run-on sentence.
366+ s.emit.Flush()
367+ if _, err := s.agent.RunBlocks(ctx, []llm.ContentBlock{llm.TextBlock(text)}); err != nil {
368+ return err
369+ }
370+ continue
371+ }
372+ if !s.agentsLive() {
373+ return nil
374+ }
375+
376+ // Nothing to hand over and something still running: wait to be woken.
377+ select {
378+ case <-s.agentsMoved:
379+ case <-ctx.Done():
380+ return ctx.Err()
381+ }
382+ }
383+ }
384+
251385 // Cancel stops the turn in flight. An unknown session is not an error worth
252386 // reporting: a stop arriving after the session closed asks for something that
253387 // already happened.
+11−10internal/serve/serve_test.goView File
@@ -90,20 +90,21 @@ func newTestServer(t *testing.T, f *fakeAPI, opts Options) *testClient {
9090 log := metrics.NewWriter(&logBuf, "test")
9191
9292 if opts.NewAgent == nil {
93- opts.NewAgent = func(root, model string, state *agent.State) (*agent.Agent, error) {
93+ opts.NewAgent = func(req AgentRequest) (AgentSet, error) {
94+ model := req.Model
9495 if model == "" {
9596 model = "test-model"
9697 }
9798 reg := tools.NewRegistry(log).Register(
98- tools.NewRead(root, state),
99- tools.NewWrite(root, state),
100- tools.NewBash(root),
99+ tools.NewRead(req.Root, req.State),
100+ tools.NewWrite(req.Root, req.State),
101+ tools.NewBash(req.Root),
101102 )
102103 client := llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k"), llm.WithModel(model))
103104 // Streaming needs an SSE endpoint; the delta path is covered in
104105 // update_test.go, and a whole-response turn exercises everything
105106 // this file is about.
106- return agent.New(client, reg, state, log, agent.Options{}), nil
107+ return AgentSet{Agent: agent.New(client, reg, req.State, log, agent.Options{})}, nil
107108 }
108109 }
109110 if opts.Version == "" {
@@ -570,12 +571,12 @@ func TestSelectingAModelKeepsTheConversation(t *testing.T) {
570571 c := newTestServer(t, f, Options{
571572 Models: []acp.ModelInfo{{ModelID: "a", Name: "A"}, {ModelID: "b", Name: "B"}},
572573 DefaultModel: "a",
573- NewAgent: func(root, model string, state *agent.State) (*agent.Agent, error) {
574- built = append(built, model)
574+ NewAgent: func(req AgentRequest) (AgentSet, error) {
575+ built = append(built, req.Model)
575576 log := metrics.NewWriter(io.Discard, "test")
576- reg := tools.NewRegistry(log).Register(tools.NewRead(root, state))
577- client := llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k"), llm.WithModel(model))
578- return agent.New(client, reg, state, log, agent.Options{}), nil
577+ reg := tools.NewRegistry(log).Register(tools.NewRead(req.Root, req.State))
578+ client := llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k"), llm.WithModel(req.Model))
579+ return AgentSet{Agent: agent.New(client, reg, req.State, log, agent.Options{})}, nil
579580 },
580581 })
581582
+6−0internal/serve/session.goView File
@@ -24,6 +24,12 @@ type session struct {
2424 state *agent.State
2525 policy *approval.Policy
2626
27+ // agents is this session's subagent supervisor, nil when the session may
28+ // not spawn. agentsMoved is how a child wakes a turn that is waiting for
29+ // it: buffered, so a signal nobody is waiting for costs nothing.
30+ agents Agents
31+ agentsMoved chan struct{}
32+
2733 // mu serializes turns. A second prompt while one is running waits for it,
2834 // which is the same thing the TUI does and the only thing the agent's
2935 // history can survive.
+40−15internal/serve/update.goView File
@@ -70,11 +70,20 @@ type emitter struct {
7070 current string
7171
7272 // running accumulates the output of tool calls that have not finished,
73- // keyed by call id, with sentLen recording how much of each was last sent.
74- // Both are dropped when the call completes.
75- running map[string]*strings.Builder
76- sentLen map[string]int
77- lastProgress time.Time
73+ // keyed by call id, with sentLen recording how much of each was last sent
74+ // and lastSent when. All three are dropped when the call completes.
75+ //
76+ // The clock is per row rather than per emitter: several subagents talk at
77+ // once, and a shared one would let whichever spoke most recently silence
78+ // the others until its own window was up.
79+ running map[string]*strings.Builder
80+ sentLen map[string]int
81+ lastSent map[string]time.Time
82+
83+ // closed records the subagent rows that have already been finished off, so
84+ // the change callback — which fires repeatedly once a child has ended —
85+ // closes each one once.
86+ closed map[string]bool
7887 }
7988
8089 // currentCall is the id of the tool call an approval prompt belongs to.
@@ -92,6 +101,8 @@ func newEmitter(send func(acp.Update)) *emitter {
92101 afterSize: flushBytes,
93102 running: make(map[string]*strings.Builder),
94103 sentLen: make(map[string]int),
104+ lastSent: make(map[string]time.Time),
105+ closed: make(map[string]bool),
95106 }
96107 }
97108
@@ -248,14 +259,7 @@ func (e *emitter) toolProgress(ev agent.Event) {
248259 }
249260 buf.WriteString(ev.Text)
250261
251- grown := buf.Len() - e.sentLen[id]
252- due := grown >= progressBytes || e.now().Sub(e.lastProgress) >= progressAfter
253- var out string
254- if due {
255- out = tailOf(buf.String(), progressMaxBytes)
256- e.sentLen[id] = buf.Len()
257- e.lastProgress = e.now()
258- }
262+ out := e.dueLocked(id, buf)
259263 e.mu.Unlock()
260264
261265 if out == "" {
@@ -269,6 +273,20 @@ func (e *emitter) toolProgress(ev agent.Event) {
269273 })
270274 }
271275
276+ // dueLocked decides whether a row has grown enough, or waited long enough, to
277+ // be worth another notification, and returns what to send when it has. The
278+ // caller holds the mutex.
279+ func (e *emitter) dueLocked(id string, buf *strings.Builder) string {
280+ grown := buf.Len() - e.sentLen[id]
281+ last, seen := e.lastSent[id]
282+ if grown < progressBytes && seen && e.now().Sub(last) < progressAfter {
283+ return ""
284+ }
285+ e.sentLen[id] = buf.Len()
286+ e.lastSent[id] = e.now()
287+ return tailOf(buf.String(), progressMaxBytes)
288+ }
289+
272290 func (e *emitter) toolResult(ev agent.Event) {
273291 status := acp.StatusCompleted
274292 if ev.IsError {
@@ -277,8 +295,7 @@ func (e *emitter) toolResult(ev agent.Event) {
277295 id := e.callID(ev, false)
278296
279297 e.mu.Lock()
280- delete(e.running, id)
281- delete(e.sentLen, id)
298+ e.forgetLocked(id)
282299 e.mu.Unlock()
283300
284301 // The result replaces whatever progress said, which is what we want: the
@@ -292,6 +309,14 @@ func (e *emitter) toolResult(ev agent.Event) {
292309 })
293310 }
294311
312+ // forgetLocked drops a finished row's accumulated state. The caller holds the
313+ // mutex.
314+ func (e *emitter) forgetLocked(id string) {
315+ delete(e.running, id)
316+ delete(e.sentLen, id)
317+ delete(e.lastSent, id)
318+ }
319+
295320 // tailOf keeps the end of a live-growing output, which is the part worth
296321 // watching, and says that it did so.
297322 func tailOf(s string, max int) string {