A two-minute test run was a blank row for two minutes. Tools now report partial output through a sink on the context, the loop turns that into EventToolDelta, and the ACP front end forwards it as a tool_call_update so T3 Code fills the row in as the build talks. The sink is installed only when somebody is observing, so a tool asks once whether anyone is watching and the eval runner stays on the code it was on. It carries deltas; the consumer accumulates. bash writes both streams into it, because what a reader wants is what a terminal would have shown. The protocol has no delta here — a tool_call_update replaces the content of the call it names — so each one carries everything so far, bounded to the last 8,000 bytes, which is what the client would have bounded it to anyway. Throttled at 200ms or 256 bytes, since a build writing a line at a time would otherwise be a notification per line each carrying the whole log. The finished report replaces the live tail when the call ends.
Showing 7 changed files with 498 additions and 5 deletions.
| @@ -169,6 +169,15 @@ const ( | ||
| 169 | 169 | EventNotice EventKind = "notice" |
| 170 | 170 | // EventTextDelta is emitted instead of EventText while streaming. |
| 171 | 171 | EventTextDelta EventKind = "text_delta" |
| 172 | + // EventToolDelta is output a tool has produced but not finished producing: | |
| 173 | + // the lines of a test run as they appear, rather than all of them at the | |
| 174 | + // end. Text carries the new bytes, not everything so far, and Tool and ID | |
| 175 | + // say which call they belong to. | |
| 176 | + // | |
| 177 | + // Only tools that opt into it produce these, and only while someone is | |
| 178 | + // watching — see [tools.WithProgress]. A front end that ignores this kind | |
| 179 | + // still shows the whole output when the call finishes. | |
| 180 | + EventToolDelta EventKind = "tool_delta" | |
| 172 | 181 | // EventPaused is emitted when the loop stops at a turn boundary, and |
| 173 | 182 | // EventResumed when it is let go again. |
| 174 | 183 | EventPaused EventKind = "paused" |
| @@ -496,7 +505,7 @@ func (a *Agent) dispatch(ctx context.Context, turn int, calls []llm.ContentBlock | ||
| 496 | 505 | for _, call := range calls { |
| 497 | 506 | a.emit(Event{Kind: EventToolStart, Turn: turn, Tool: call.Name, ID: call.ID, Input: call.Input}) |
| 498 | 507 | |
| 499 | - out, err := a.authorizedExecute(ctx, call) | |
| 508 | + out, err := a.authorizedExecute(a.watching(ctx, turn, call), call) | |
| 500 | 509 | if ctx.Err() != nil { |
| 501 | 510 | return nil, ctx.Err() |
| 502 | 511 | } |
| @@ -599,6 +608,28 @@ func summarizeWebSearchResult(b llm.ContentBlock) (string, bool) { | ||
| 599 | 608 | return strings.Join(lines, "\n"), false |
| 600 | 609 | } |
| 601 | 610 | |
| 611 | + // watching attaches a progress sink to the context a tool runs under, so a | |
| 612 | + // long command's output reaches the front end while it is still running rather | |
| 613 | + // than in one lump at the end. | |
| 614 | + // | |
| 615 | + // The sink is installed only when somebody is observing: a tool that finds | |
| 616 | + // nothing on the context does no extra copying, which keeps the eval runner | |
| 617 | + // and the tests on exactly the path they had before. The mutex is not | |
| 618 | + // decoration — a command's stdout and stderr are copied by separate | |
| 619 | + // goroutines, and an observer written for the loop's own goroutine has no | |
| 620 | + // reason to expect two calls at once. | |
| 621 | + func (a *Agent) watching(ctx context.Context, turn int, call llm.ContentBlock) context.Context { | |
| 622 | + if a.Observe == nil { | |
| 623 | + return ctx | |
| 624 | + } | |
| 625 | + var mu sync.Mutex | |
| 626 | + return tools.WithProgress(ctx, func(chunk string) { | |
| 627 | + mu.Lock() | |
| 628 | + defer mu.Unlock() | |
| 629 | + a.emit(Event{Kind: EventToolDelta, Turn: turn, Tool: call.Name, ID: call.ID, Text: chunk}) | |
| 630 | + }) | |
| 631 | + } | |
| 632 | + | |
| 602 | 633 | // authorizedExecute runs a tool once the approval policy allows it. A denial |
| 603 | 634 | // comes back as an ordinary tool error: the model should read it and change |
| 604 | 635 | // course, not see Codey crash. |
| @@ -900,3 +900,84 @@ func TestSetToolsIsSafeAgainstAReadingTurn(t *testing.T) { | ||
| 900 | 900 | t.Error("SetTools(nil) replaced the tool set") |
| 901 | 901 | } |
| 902 | 902 | } |
| 903 | + | |
| 904 | + func TestToolProgressReachesTheObserver(t *testing.T) { | |
| 905 | + f := newFakeAPI(t, | |
| 906 | + msg("tool_use", toolUseBlock("call-1", "noisy", `{}`)), | |
| 907 | + msg("end_turn", textBlock("done")), | |
| 908 | + ) | |
| 909 | + | |
| 910 | + var logBuf bytes.Buffer | |
| 911 | + log := metrics.NewWriter(&logBuf, "test") | |
| 912 | + state := NewState() | |
| 913 | + reg := tools.NewRegistry(log).Register( | |
| 914 | + tools.New("noisy", "Reports as it goes", func(ctx context.Context, _ struct{}) (string, error) { | |
| 915 | + if p := tools.ProgressFrom(ctx); p != nil { | |
| 916 | + p("half ") | |
| 917 | + p("way") | |
| 918 | + } | |
| 919 | + return "all of it", nil | |
| 920 | + }), | |
| 921 | + ) | |
| 922 | + client := llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k")) | |
| 923 | + a := New(client, reg, state, log, Options{}) | |
| 924 | + | |
| 925 | + var mu sync.Mutex | |
| 926 | + var deltas []Event | |
| 927 | + a.Observe = func(ev Event) { | |
| 928 | + if ev.Kind == EventToolDelta { | |
| 929 | + mu.Lock() | |
| 930 | + deltas = append(deltas, ev) | |
| 931 | + mu.Unlock() | |
| 932 | + } | |
| 933 | + } | |
| 934 | + | |
| 935 | + if _, err := a.Run(context.Background(), "go"); err != nil { | |
| 936 | + t.Fatalf("Run: %v", err) | |
| 937 | + } | |
| 938 | + | |
| 939 | + mu.Lock() | |
| 940 | + defer mu.Unlock() | |
| 941 | + if len(deltas) != 2 { | |
| 942 | + t.Fatalf("got %d deltas, want 2", len(deltas)) | |
| 943 | + } | |
| 944 | + var text strings.Builder | |
| 945 | + for _, d := range deltas { | |
| 946 | + if d.Tool != "noisy" || d.ID != "call-1" { | |
| 947 | + t.Errorf("a delta could not be matched to its call: %+v", d) | |
| 948 | + } | |
| 949 | + text.WriteString(d.Text) | |
| 950 | + } | |
| 951 | + if text.String() != "half way" { | |
| 952 | + t.Errorf("deltas reassembled to %q", text.String()) | |
| 953 | + } | |
| 954 | + } | |
| 955 | + | |
| 956 | + func TestNoObserverMeansNoProgressSink(t *testing.T) { | |
| 957 | + // A tool asks whether anyone is watching before it does the extra work of | |
| 958 | + // streaming. With no observer the answer must be no, so the eval runner | |
| 959 | + // and the -p path stay on exactly the code they were on before. | |
| 960 | + f := newFakeAPI(t, | |
| 961 | + msg("tool_use", toolUseBlock("call-1", "checker", `{}`)), | |
| 962 | + msg("end_turn", textBlock("done")), | |
| 963 | + ) | |
| 964 | + | |
| 965 | + var logBuf bytes.Buffer | |
| 966 | + log := metrics.NewWriter(&logBuf, "test") | |
| 967 | + state := NewState() | |
| 968 | + watched := true | |
| 969 | + reg := tools.NewRegistry(log).Register( | |
| 970 | + tools.New("checker", "Reports whether anyone is watching", func(ctx context.Context, _ struct{}) (string, error) { | |
| 971 | + watched = tools.ProgressFrom(ctx) != nil | |
| 972 | + return "ok", nil | |
| 973 | + }), | |
| 974 | + ) | |
| 975 | + a := New(llm.New(llm.WithBaseURL(f.srv.URL), llm.WithAPIKey("k")), reg, state, log, Options{}) | |
| 976 | + | |
| 977 | + if _, err := a.Run(context.Background(), "go"); err != nil { | |
| 978 | + t.Fatalf("Run: %v", err) | |
| 979 | + } | |
| 980 | + if watched { | |
| 981 | + t.Error("a sink was installed with no observer to receive it") | |
| 982 | + } | |
| 983 | + } | |
| @@ -21,6 +21,20 @@ const ( | ||
| 21 | 21 | flushBytes = 512 |
| 22 | 22 | ) |
| 23 | 23 | |
| 24 | + // The same idea for a tool that is still running, with a slower hand. Each | |
| 25 | + // update carries the whole output so far rather than a delta, so they cost | |
| 26 | + // more than a text chunk does and are worth less often: a progress view is | |
| 27 | + // read at a glance, not word by word. | |
| 28 | + // | |
| 29 | + // progressMaxBytes is the tail kept. The client bounds what it accepts at | |
| 30 | + // 8,000 characters and keeps the end; matching that here means the bound costs | |
| 31 | + // one truncation rather than two. | |
| 32 | + const ( | |
| 33 | + progressAfter = 200 * time.Millisecond | |
| 34 | + progressBytes = 256 | |
| 35 | + progressMaxBytes = 8000 | |
| 36 | + ) | |
| 37 | + | |
| 24 | 38 | // emitter turns one turn's [agent.Event] stream into session/update |
| 25 | 39 | // notifications. |
| 26 | 40 | // |
| @@ -48,6 +62,13 @@ type emitter struct { | ||
| 48 | 62 | // approval prompt is about: the loop emits tool_start and then asks, one |
| 49 | 63 | // call at a time. |
| 50 | 64 | current string |
| 65 | + | |
| 66 | + // running accumulates the output of tool calls that have not finished, | |
| 67 | + // keyed by call id, with sentLen recording how much of each was last sent. | |
| 68 | + // Both are dropped when the call completes. | |
| 69 | + running map[string]*strings.Builder | |
| 70 | + sentLen map[string]int | |
| 71 | + lastProgress time.Time | |
| 51 | 72 | } |
| 52 | 73 | |
| 53 | 74 | // currentCall is the id of the tool call an approval prompt belongs to. |
| @@ -58,7 +79,14 @@ func (e *emitter) currentCall() string { | ||
| 58 | 79 | } |
| 59 | 80 | |
| 60 | 81 | func newEmitter(send func(acp.Update)) *emitter { |
| 61 | - return &emitter{send: send, now: time.Now, afterTime: flushAfter, afterSize: flushBytes} | |
| 82 | + return &emitter{ | |
| 83 | + send: send, | |
| 84 | + now: time.Now, | |
| 85 | + afterTime: flushAfter, | |
| 86 | + afterSize: flushBytes, | |
| 87 | + running: make(map[string]*strings.Builder), | |
| 88 | + sentLen: make(map[string]int), | |
| 89 | + } | |
| 62 | 90 | } |
| 63 | 91 | |
| 64 | 92 | // handle is the [agent.Agent] observer. |
| @@ -80,6 +108,9 @@ func (e *emitter) handle(ev agent.Event) { | ||
| 80 | 108 | e.Flush() |
| 81 | 109 | e.toolCall(ev) |
| 82 | 110 | |
| 111 | + case agent.EventToolDelta: | |
| 112 | + e.toolProgress(ev) | |
| 113 | + | |
| 83 | 114 | case agent.EventToolResult: |
| 84 | 115 | e.Flush() |
| 85 | 116 | e.toolResult(ev) |
| @@ -156,19 +187,86 @@ func (e *emitter) toolCall(ev agent.Event) { | ||
| 156 | 187 | }) |
| 157 | 188 | } |
| 158 | 189 | |
| 190 | + // toolProgress reports output a tool is still producing. | |
| 191 | + // | |
| 192 | + // The protocol has no delta for this: a tool_call_update replaces the content | |
| 193 | + // of the call it names rather than appending to it, so each one carries | |
| 194 | + // everything so far. That is the shape clients expect — and defend against, by | |
| 195 | + // keeping only a bounded tail — so the bound is applied here too rather than | |
| 196 | + // making somebody else pay for our chattiness. | |
| 197 | + // | |
| 198 | + // Throttling is ours as well. A build writing a line at a time would otherwise | |
| 199 | + // be one notification per line, each carrying the whole log. | |
| 200 | + func (e *emitter) toolProgress(ev agent.Event) { | |
| 201 | + if ev.Text == "" { | |
| 202 | + return | |
| 203 | + } | |
| 204 | + id := e.callID(ev, false) | |
| 205 | + | |
| 206 | + e.mu.Lock() | |
| 207 | + if e.running == nil { | |
| 208 | + e.running = make(map[string]*strings.Builder) | |
| 209 | + } | |
| 210 | + buf, ok := e.running[id] | |
| 211 | + if !ok { | |
| 212 | + buf = &strings.Builder{} | |
| 213 | + e.running[id] = buf | |
| 214 | + } | |
| 215 | + buf.WriteString(ev.Text) | |
| 216 | + | |
| 217 | + grown := buf.Len() - e.sentLen[id] | |
| 218 | + due := grown >= progressBytes || e.now().Sub(e.lastProgress) >= progressAfter | |
| 219 | + var out string | |
| 220 | + if due { | |
| 221 | + out = tailOf(buf.String(), progressMaxBytes) | |
| 222 | + e.sentLen[id] = buf.Len() | |
| 223 | + e.lastProgress = e.now() | |
| 224 | + } | |
| 225 | + e.mu.Unlock() | |
| 226 | + | |
| 227 | + if out == "" { | |
| 228 | + return | |
| 229 | + } | |
| 230 | + e.send(acp.Update{ | |
| 231 | + SessionUpdate: acp.UpdateToolCallUpdate, | |
| 232 | + ToolCallID: id, | |
| 233 | + Status: acp.StatusInProgress, | |
| 234 | + Content: []acp.ToolCallContent{acp.ToolCallText(out)}, | |
| 235 | + }) | |
| 236 | + } | |
| 237 | + | |
| 159 | 238 | func (e *emitter) toolResult(ev agent.Event) { |
| 160 | 239 | status := acp.StatusCompleted |
| 161 | 240 | if ev.IsError { |
| 162 | 241 | status = acp.StatusFailed |
| 163 | 242 | } |
| 243 | + id := e.callID(ev, false) | |
| 244 | + | |
| 245 | + e.mu.Lock() | |
| 246 | + delete(e.running, id) | |
| 247 | + delete(e.sentLen, id) | |
| 248 | + e.mu.Unlock() | |
| 249 | + | |
| 250 | + // The result replaces whatever progress said, which is what we want: the | |
| 251 | + // report a tool ends with is capped head and tail, where the live view was | |
| 252 | + // a tail alone. | |
| 164 | 253 | e.send(acp.Update{ |
| 165 | 254 | SessionUpdate: acp.UpdateToolCallUpdate, |
| 166 | - ToolCallID: e.callID(ev, false), | |
| 255 | + ToolCallID: id, | |
| 167 | 256 | Status: status, |
| 168 | 257 | Content: []acp.ToolCallContent{acp.ToolCallText(ev.Result)}, |
| 169 | 258 | }) |
| 170 | 259 | } |
| 171 | 260 | |
| 261 | + // tailOf keeps the end of a live-growing output, which is the part worth | |
| 262 | + // watching, and says that it did so. | |
| 263 | + func tailOf(s string, max int) string { | |
| 264 | + if len(s) <= max { | |
| 265 | + return s | |
| 266 | + } | |
| 267 | + return "[earlier output truncated]\n\n" + s[len(s)-max:] | |
| 268 | + } | |
| 269 | + | |
| 172 | 270 | func (e *emitter) callID(ev agent.Event, start bool) string { |
| 173 | 271 | if ev.ID != "" { |
| 174 | 272 | return ev.ID |
| @@ -217,3 +217,132 @@ func equal(a, b []string) bool { | ||
| 217 | 217 | } |
| 218 | 218 | return true |
| 219 | 219 | } |
| 220 | + | |
| 221 | + // contentText is the text of a tool call update's content, which is a list | |
| 222 | + // rather than the single block a chunk carries. | |
| 223 | + func contentText(t *testing.T, u acp.Update) string { | |
| 224 | + t.Helper() | |
| 225 | + items, ok := u.Content.([]acp.ToolCallContent) | |
| 226 | + if !ok { | |
| 227 | + t.Fatalf("content is not tool call content: %#v", u.Content) | |
| 228 | + } | |
| 229 | + var b strings.Builder | |
| 230 | + for _, it := range items { | |
| 231 | + if it.Content != nil { | |
| 232 | + b.WriteString(it.Content.Text) | |
| 233 | + } | |
| 234 | + } | |
| 235 | + return b.String() | |
| 236 | + } | |
| 237 | + | |
| 238 | + func TestToolOutputGoesOutWhileTheToolIsStillRunning(t *testing.T) { | |
| 239 | + e, r := newRecorded() | |
| 240 | + | |
| 241 | + e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "bash", ID: "t1", | |
| 242 | + Input: json.RawMessage(`{"command":"go test ./..."}`)}) | |
| 243 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: "t1", Text: strings.Repeat("ok\n", 200)}) | |
| 244 | + | |
| 245 | + if len(r.updates) != 2 { | |
| 246 | + t.Fatalf("the output did not go out until the tool finished: %v", r.kinds()) | |
| 247 | + } | |
| 248 | + live := r.updates[1] | |
| 249 | + if live.SessionUpdate != acp.UpdateToolCallUpdate || live.ToolCallID != "t1" { | |
| 250 | + t.Fatalf("update = %+v", live) | |
| 251 | + } | |
| 252 | + if live.Status != acp.StatusInProgress { | |
| 253 | + t.Errorf("status = %q, want in_progress — a completed row stops updating", live.Status) | |
| 254 | + } | |
| 255 | + if !strings.Contains(contentText(t, live), "ok") { | |
| 256 | + t.Errorf("content = %q", contentText(t, live)) | |
| 257 | + } | |
| 258 | + } | |
| 259 | + | |
| 260 | + func TestProgressUpdatesCarryEverythingSoFar(t *testing.T) { | |
| 261 | + // A tool_call_update replaces the content of the call it names rather than | |
| 262 | + // appending to it, so a delta-shaped update would show only the last line. | |
| 263 | + e, r := newRecorded() | |
| 264 | + | |
| 265 | + e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "bash", ID: "t1", Input: json.RawMessage(`{"command":"x"}`)}) | |
| 266 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: "t1", Text: strings.Repeat("a", progressBytes)}) | |
| 267 | + r.clock = r.clock.Add(progressAfter) | |
| 268 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: "t1", Text: "b"}) | |
| 269 | + | |
| 270 | + last := contentText(t, r.updates[len(r.updates)-1]) | |
| 271 | + if !strings.HasPrefix(last, "aaa") || !strings.HasSuffix(last, "b") { | |
| 272 | + t.Errorf("the update did not carry the whole output: %q…%q", last[:min(8, len(last))], last[max(0, len(last)-4):]) | |
| 273 | + } | |
| 274 | + } | |
| 275 | + | |
| 276 | + func TestProgressIsThrottled(t *testing.T) { | |
| 277 | + e, r := newRecorded() | |
| 278 | + | |
| 279 | + e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "bash", ID: "t1", Input: json.RawMessage(`{"command":"x"}`)}) | |
| 280 | + before := len(r.updates) | |
| 281 | + | |
| 282 | + // A build writing a line at a time, inside one window. | |
| 283 | + for range 50 { | |
| 284 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: "t1", Text: "line\n"}) | |
| 285 | + } | |
| 286 | + | |
| 287 | + // 50 lines of five bytes is 250, just under the size trigger, and no time | |
| 288 | + // has passed: one notification per line would be fifty. | |
| 289 | + if got := len(r.updates) - before; got > 1 { | |
| 290 | + t.Errorf("%d notifications for 50 lines inside one window", got) | |
| 291 | + } | |
| 292 | + } | |
| 293 | + | |
| 294 | + func TestProgressIsBoundedToATail(t *testing.T) { | |
| 295 | + e, r := newRecorded() | |
| 296 | + | |
| 297 | + e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "bash", ID: "t1", Input: json.RawMessage(`{"command":"x"}`)}) | |
| 298 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: "t1", | |
| 299 | + Text: strings.Repeat("x", progressMaxBytes*2) + "END"}) | |
| 300 | + | |
| 301 | + body := contentText(t, r.updates[len(r.updates)-1]) | |
| 302 | + if len(body) > progressMaxBytes+64 { | |
| 303 | + t.Errorf("an unbounded %d bytes went over the wire", len(body)) | |
| 304 | + } | |
| 305 | + if !strings.HasSuffix(body, "END") { | |
| 306 | + t.Error("the tail is the useful part and it was the part dropped") | |
| 307 | + } | |
| 308 | + if !strings.Contains(body, "truncated") { | |
| 309 | + t.Error("the truncation was silent") | |
| 310 | + } | |
| 311 | + } | |
| 312 | + | |
| 313 | + func TestTheFinalResultReplacesTheLiveView(t *testing.T) { | |
| 314 | + e, r := newRecorded() | |
| 315 | + | |
| 316 | + e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "bash", ID: "t1", Input: json.RawMessage(`{"command":"x"}`)}) | |
| 317 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: "t1", Text: strings.Repeat("noise\n", 100)}) | |
| 318 | + e.handle(agent.Event{Kind: agent.EventToolResult, Tool: "bash", ID: "t1", Result: "exit code: 0 (1ms)"}) | |
| 319 | + | |
| 320 | + last := r.updates[len(r.updates)-1] | |
| 321 | + if last.Status != acp.StatusCompleted { | |
| 322 | + t.Fatalf("status = %q", last.Status) | |
| 323 | + } | |
| 324 | + if got := contentText(t, last); got != "exit code: 0 (1ms)" { | |
| 325 | + t.Errorf("content = %q, want the finished report", got) | |
| 326 | + } | |
| 327 | + } | |
| 328 | + | |
| 329 | + func TestProgressForAFinishedCallStartsCleanNextTime(t *testing.T) { | |
| 330 | + // Two commands in one turn must not show each other's output. | |
| 331 | + e, r := newRecorded() | |
| 332 | + | |
| 333 | + for _, id := range []string{"t1", "t2"} { | |
| 334 | + e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "bash", ID: id, Input: json.RawMessage(`{"command":"x"}`)}) | |
| 335 | + e.handle(agent.Event{Kind: agent.EventToolDelta, Tool: "bash", ID: id, Text: strings.Repeat(id, 200)}) | |
| 336 | + e.handle(agent.Event{Kind: agent.EventToolResult, Tool: "bash", ID: id, Result: "done " + id}) | |
| 337 | + r.clock = r.clock.Add(progressAfter) | |
| 338 | + } | |
| 339 | + | |
| 340 | + for _, u := range r.updates { | |
| 341 | + if u.ToolCallID != "t2" || u.SessionUpdate != acp.UpdateToolCallUpdate || u.Status != acp.StatusInProgress { | |
| 342 | + continue | |
| 343 | + } | |
| 344 | + if strings.Contains(contentText(t, u), "t1") { | |
| 345 | + t.Error("the second command's progress carried the first command's output") | |
| 346 | + } | |
| 347 | + } | |
| 348 | + } | |
| @@ -58,9 +58,14 @@ func RunCommand(ctx context.Context, dir string, argv []string, timeout time.Dur | ||
| 58 | 58 | } |
| 59 | 59 | cmd.WaitDelay = 2 * time.Second |
| 60 | 60 | |
| 61 | + // Both streams are captured for the report and, when something is | |
| 62 | + // watching, copied out as they arrive. They share one sink: a reader | |
| 63 | + // watching a test run wants what a terminal would have shown, and that is | |
| 64 | + // both streams interleaved in the order they were written. | |
| 61 | 65 | var stdout, stderr bytes.Buffer |
| 62 | - cmd.Stdout = &stdout | |
| 63 | - cmd.Stderr = &stderr | |
| 66 | + progress := ProgressFrom(ctx) | |
| 67 | + cmd.Stdout = tee(&stdout, progress) | |
| 68 | + cmd.Stderr = tee(&stderr, progress) | |
| 64 | 69 | |
| 65 | 70 | start := time.Now() |
| 66 | 71 | runErr := cmd.Run() |
| @@ -0,0 +1,56 @@ | ||
| 1 | + package tools | |
| 2 | + | |
| 3 | + import ( | |
| 4 | + "context" | |
| 5 | + "io" | |
| 6 | + ) | |
| 7 | + | |
| 8 | + // Progress is where a long-running tool reports output it has produced but not | |
| 9 | + // finished producing. It carries deltas, not the whole output so far: the | |
| 10 | + // consumer accumulates if it needs to, and a front end that only wants to show | |
| 11 | + // the last line does not have to diff two strings to find it. | |
| 12 | + // | |
| 13 | + // It may be called from more than one goroutine — a command's stdout and | |
| 14 | + // stderr are copied by separate ones — so an implementation must be safe for | |
| 15 | + // concurrent use. | |
| 16 | + type Progress func(chunk string) | |
| 17 | + | |
| 18 | + type progressKey struct{} | |
| 19 | + | |
| 20 | + // WithProgress attaches a sink to ctx. The agent loop installs one around each | |
| 21 | + // tool call so the tool does not have to know who is watching, or whether | |
| 22 | + // anyone is. | |
| 23 | + func WithProgress(ctx context.Context, p Progress) context.Context { | |
| 24 | + if p == nil { | |
| 25 | + return ctx | |
| 26 | + } | |
| 27 | + return context.WithValue(ctx, progressKey{}, p) | |
| 28 | + } | |
| 29 | + | |
| 30 | + // ProgressFrom returns the sink on ctx, or nil when nothing is watching. | |
| 31 | + // A tool that finds nothing here should do no extra work at all: the cost of | |
| 32 | + // streaming is only worth paying when it is going somewhere. | |
| 33 | + func ProgressFrom(ctx context.Context) Progress { | |
| 34 | + p, _ := ctx.Value(progressKey{}).(Progress) | |
| 35 | + return p | |
| 36 | + } | |
| 37 | + | |
| 38 | + // progressWriter feeds an io.Writer chain into a sink. | |
| 39 | + type progressWriter struct{ p Progress } | |
| 40 | + | |
| 41 | + func (w progressWriter) Write(b []byte) (int, error) { | |
| 42 | + if len(b) > 0 { | |
| 43 | + w.p(string(b)) | |
| 44 | + } | |
| 45 | + return len(b), nil | |
| 46 | + } | |
| 47 | + | |
| 48 | + // tee returns w, plus a copy into the sink when there is one. A nil sink gives | |
| 49 | + // back exactly what was passed in, so the no-observer path allocates nothing | |
| 50 | + // and copies nothing. | |
| 51 | + func tee(w io.Writer, p Progress) io.Writer { | |
| 52 | + if p == nil { | |
| 53 | + return w | |
| 54 | + } | |
| 55 | + return io.MultiWriter(w, progressWriter{p}) | |
| 56 | + } | |
| @@ -0,0 +1,93 @@ | ||
| 1 | + package tools | |
| 2 | + | |
| 3 | + import ( | |
| 4 | + "context" | |
| 5 | + "strings" | |
| 6 | + "sync" | |
| 7 | + "testing" | |
| 8 | + "time" | |
| 9 | + ) | |
| 10 | + | |
| 11 | + func TestACommandsOutputArrivesBeforeItFinishes(t *testing.T) { | |
| 12 | + // The whole point: a two-minute test run should not be a blank row for two | |
| 13 | + // minutes. The command prints, then sleeps; if nothing is streaming, the | |
| 14 | + // first chunk cannot arrive until the sleep is over. | |
| 15 | + first := make(chan string, 1) | |
| 16 | + var mu sync.Mutex | |
| 17 | + var all strings.Builder | |
| 18 | + | |
| 19 | + ctx := WithProgress(context.Background(), func(chunk string) { | |
| 20 | + mu.Lock() | |
| 21 | + all.WriteString(chunk) | |
| 22 | + mu.Unlock() | |
| 23 | + select { | |
| 24 | + case first <- chunk: | |
| 25 | + default: | |
| 26 | + } | |
| 27 | + }) | |
| 28 | + | |
| 29 | + done := make(chan string, 1) | |
| 30 | + go func() { | |
| 31 | + report, _ := RunCommand(ctx, t.TempDir(), []string{"bash", "-c", "echo early; sleep 1"}, 10*time.Second) | |
| 32 | + done <- report | |
| 33 | + }() | |
| 34 | + | |
| 35 | + select { | |
| 36 | + case chunk := <-first: | |
| 37 | + if !strings.Contains(chunk, "early") { | |
| 38 | + t.Errorf("first chunk = %q", chunk) | |
| 39 | + } | |
| 40 | + case <-done: | |
| 41 | + t.Fatal("the output only arrived once the command had finished") | |
| 42 | + case <-time.After(5 * time.Second): | |
| 43 | + t.Fatal("no output arrived at all") | |
| 44 | + } | |
| 45 | + | |
| 46 | + report := <-done | |
| 47 | + // The report is unaffected: streaming is an addition, not a replacement. | |
| 48 | + if !strings.Contains(report, "early") { | |
| 49 | + t.Errorf("report lost the output: %q", report) | |
| 50 | + } | |
| 51 | + mu.Lock() | |
| 52 | + defer mu.Unlock() | |
| 53 | + if !strings.Contains(all.String(), "early") { | |
| 54 | + t.Errorf("streamed output = %q", all.String()) | |
| 55 | + } | |
| 56 | + } | |
| 57 | + | |
| 58 | + func TestBothStreamsReachTheSameSink(t *testing.T) { | |
| 59 | + var mu sync.Mutex | |
| 60 | + var got strings.Builder | |
| 61 | + ctx := WithProgress(context.Background(), func(chunk string) { | |
| 62 | + mu.Lock() | |
| 63 | + defer mu.Unlock() | |
| 64 | + got.WriteString(chunk) | |
| 65 | + }) | |
| 66 | + | |
| 67 | + RunCommand(ctx, t.TempDir(), []string{"bash", "-c", "echo out; echo err >&2"}, 10*time.Second) | |
| 68 | + | |
| 69 | + mu.Lock() | |
| 70 | + defer mu.Unlock() | |
| 71 | + // A reader watching a build wants what a terminal would have shown. | |
| 72 | + for _, want := range []string{"out", "err"} { | |
| 73 | + if !strings.Contains(got.String(), want) { | |
| 74 | + t.Errorf("the sink never saw %q: %q", want, got.String()) | |
| 75 | + } | |
| 76 | + } | |
| 77 | + } | |
| 78 | + | |
| 79 | + func TestWithNobodyWatchingNothingChanges(t *testing.T) { | |
| 80 | + if p := ProgressFrom(context.Background()); p != nil { | |
| 81 | + t.Fatal("a bare context should carry no sink") | |
| 82 | + } | |
| 83 | + // And a nil sink must not be installed, so ProgressFrom stays the cheap | |
| 84 | + // "is anyone watching" check that tools branch on. | |
| 85 | + if p := ProgressFrom(WithProgress(context.Background(), nil)); p != nil { | |
| 86 | + t.Error("a nil sink was installed") | |
| 87 | + } | |
| 88 | + | |
| 89 | + report, failed := RunCommand(context.Background(), t.TempDir(), []string{"bash", "-c", "echo hello"}, 10*time.Second) | |
| 90 | + if failed || !strings.Contains(report, "hello") { | |
| 91 | + t.Errorf("report = %q, failed = %v", report, failed) | |
| 92 | + } | |
| 93 | + } | |