johannSoCodeyPublic

Phase 5: SSE streaming, interruption, sessions, colour

The fiddly part of streaming is tool arguments: they arrive as
input_json_delta fragments that are individually invalid JSON and only
parseable once concatenated, so they are buffered per block index and
parsed at content_block_stop. A stream that ends mid-fragment is reported
as such rather than dispatched with truncated arguments. Signature deltas
are joined the same way, since a thinking block whose signature is
half-assembled is rejected on the next request.

Ctrl-C cancels the turn in progress and leaves the session intact so the
next message can correct course; a second within two seconds, or one with
nothing running, exits. An interrupted turn is reported as an
interruption, not as an error.

Sessions persist to .harness/sessions and --resume takes an id or
"latest". Restoring sets the usage totals outright rather than adding
them, which would have counted the entire restored history as one turn.

Pricing lives in llm so the REPL's running cost and the eval report cannot
drift apart, and an unpriced model shows nothing rather than $0.0000 — a
made-up number in METRICS.md is worse than a blank.

Diffs render in colour, and the renderer tracks whether streamed text has
left a line open so tool output never lands mid-sentence.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Browse Files

Jjonicommitted 4 weeks ago1 parent e8e53e8commitf02316b

Showing 15 changed files with 1076 additions and 64 deletions.

+89−21cmd/harness/main.goView File

@@ -8,8 +8,10 @@ import (
8	8	"flag"
9	9	"fmt"
10	10	"os"
	11	+ "os/signal"
11	12	"path/filepath"
12	13	"strings"
	14	+ "sync/atomic"
13	15	"time"
14	16
15	17	"harness/internal/agent"
@@ -37,8 +39,10 @@ func run() error {
37	39	contextWindow := flag.Int("context-window", agent.DefaultContextWindow, "model context window, used to decide when to compact")
38	40	compactAt := flag.Float64("compact-at", agent.DefaultCompactAt, "fraction of the window at which to compact")
39	41	noCache := flag.Bool("no-cache", false, "disable prompt cache breakpoints")
	42	+ noStream := flag.Bool("no-stream", false, "wait for each complete response instead of streaming")
40	43	logPath := flag.String("metrics", metrics.DefaultPath, "JSONL event log path, relative to the working root")
41	44	prompt := flag.String("p", "", "run one prompt and exit, instead of starting the REPL")
	45	+ resume := flag.String("resume", "", "resume a saved session by id, or \"latest\"")
42	46	verbose := flag.Bool("v", false, "print full tool output")
43	47	flag.Parse()
44	48
@@ -47,15 +51,6 @@ func run() error {
47	51	return err
48	52	}
49	53
50		- session := time.Now().UTC().Format("20060102T150405Z")
51		- log, err := metrics.Open(filepath.Join(root, *logPath), session)
52		- if err != nil {
53		- return err
54		- }
55		- defer log.Close()
56		- // Flags win over the environment, which wins over the defaults. Every
57		- // option below ignores an empty value, so an unset flag leaves the
58		- // environment in place.
59	54	client := llm.New(
60	55	llm.WithModel(*model),
61	56	llm.WithBaseURL(*baseURL),
@@ -65,6 +60,13 @@ func run() error {
65	60	if !client.HasCredentials() {
66	61	return errNoCredentials
67	62	}
	63	+
	64	+ session := time.Now().UTC().Format("20060102T150405Z")
	65	+ log, err := metrics.Open(filepath.Join(root, *logPath), session)
	66	+ if err != nil {
	67	+ return err
	68	+ }
	69	+ defer log.Close()
68	70	log.Log(metrics.Event{Type: "session_start", Extra: map[string]any{
69	71	"root": root, "model": client.Model, "base_url": client.BaseURL,
70	72	}})
@@ -91,15 +93,29 @@ func run() error {
91	93	ContextWindow: *contextWindow,
92	94	CompactAt: *compactAt,
93	95	NoCache: *noCache,
	96	+ Stream: !*noStream,
94	97	})
95	98	a.Observe = render.Handle
96	99
97		- ctx := context.Background()
	100	+ sessionDir := filepath.Join(root, agent.SessionDir)
	101	+ if *resume != "" {
	102	+ prev, err := agent.ResolveSession(sessionDir, *resume)
	103	+ if err != nil {
	104	+ return err
	105	+ }
	106	+ a.Restore(prev)
	107	+ session = prev.ID
	108	+ fmt.Printf("resumed session %s (%d messages, %d turns)\n", prev.ID, len(prev.Messages), prev.Turns)
	109	+ }
	110	+
	111	+ ctx, stopSignals := interruptible(context.Background())
	112	+ defer stopSignals()
	113	+
98	114	if *prompt != "" {
99		- return once(ctx, a, render, *prompt)
	115	+ return once(ctx, a, render, sessionDir, session, root, client, *prompt)
100	116	}
101	117
102		- fmt.Printf("harness · %s · %s\ntools: %s\ntype a prompt, or ctrl-d to exit\n\n",
	118	+ fmt.Printf("harness · %s · %s\ntools: %s\ntype a prompt, ctrl-c to interrupt a turn, ctrl-d to exit\n\n",
103	119	client.Model, root, strings.Join(toolNames(reg), ", "))
104	120
105	121	in := bufio.NewScanner(os.Stdin)
@@ -111,25 +127,77 @@ func run() error {
111	127	return in.Err()
112	128	}
113	129	line := strings.TrimSpace(in.Text())
114		- if line == "" {
	130	+ switch {
	131	+ case line == "":
115	132	continue
116		- }
117		- if line == "/exit" \|\| line == "/quit" {
	133	+ case line == "/exit", line == "/quit":
118	134	return nil
119	135	}
120		- if err := once(ctx, a, render, line); err != nil {
	136	+ if err := once(ctx, a, render, sessionDir, session, root, client, line); err != nil {
121	137	render.Error(err)
122	138	}
123	139	fmt.Println()
124	140	}
125	141	}
126	142
127		- // once runs a single prompt and prints the session totals afterwards.
128		- func once(ctx context.Context, a agent.Agent, render ui.Renderer, prompt string) error {
129		- _, err := a.Run(ctx, prompt)
	143	+ // once runs a single prompt, prints the session totals, and saves the session.
	144	+ func once(ctx context.Context, a agent.Agent, render ui.Renderer, sessionDir, id, root string, client *llm.Client, prompt string) error {
	145	+ turnCtx, endTurn := beginTurn(ctx)
	146	+ _, runErr := a.Run(turnCtx, prompt)
	147	+ endTurn()
	148	+
	149	+ render.Flush()
130	150	usage, turns := a.State().Usage()
131		- render.Usage(usage, turns)
132		- return err
	151	+ render.Usage(usage, turns, client.Model)
	152	+
	153	+ if err := a.Save(sessionDir, id, root, client.Model); err != nil {
	154	+ fmt.Fprintln(os.Stderr, "harness: saving session:", err)
	155	+ }
	156	+ if errors.Is(runErr, context.Canceled) {
	157	+ // An interrupted turn is not an error: the point is to leave the
	158	+ // session intact so the next message can correct course.
	159	+ fmt.Println("interrupted — the session is intact; say what to do differently")
	160	+ return nil
	161	+ }
	162	+ return runErr
	163	+ }
	164	+
	165	+ // interrupt plumbing. The first ctrl-c cancels the turn in progress; a second
	166	+ // within two seconds, or one with nothing running, exits.
	167	+ var (
	168	+ turnCancel atomic.Pointer[context.CancelFunc]
	169	+ lastSignal atomic.Int64
	170	+ )
	171	+
	172	+ func interruptible(parent context.Context) (context.Context, func()) {
	173	+ ch := make(chan os.Signal, 1)
	174	+ signal.Notify(ch, os.Interrupt)
	175	+ go func() {
	176	+ for range ch {
	177	+ now := time.Now()
	178	+ prev := lastSignal.Swap(now.UnixNano())
	179	+ cancel := turnCancel.Load()
	180	+
	181	+ if cancel == nil \|\| (prev != 0 && now.Sub(time.Unix(0, prev)) < 2*time.Second) {
	182	+ fmt.Fprintln(os.Stderr, "\nharness: exiting")
	183	+ os.Exit(130)
	184	+ }
	185	+ fmt.Fprintln(os.Stderr, "\nharness: interrupting this turn (ctrl-c again to exit)")
	186	+ (*cancel)()
	187	+ }
	188	+ }()
	189	+ return parent, func() { signal.Stop(ch) }
	190	+ }
	191	+
	192	+ // beginTurn makes the running turn cancellable by the signal handler.
	193	+ func beginTurn(parent context.Context) (context.Context, func()) {
	194	+ ctx, cancel := context.WithCancel(parent)
	195	+ turnCancel.Store(&cancel)
	196	+ return ctx, func() {
	197	+ turnCancel.Store(nil)
	198	+ lastSignal.Store(0)
	199	+ cancel()
	200	+ }
133	201	}
134	202
135	203	// errNoCredentials spells out every way to supply one, because "unauthorized"

+25−18evals/report.goView File

@@ -6,8 +6,6 @@ import (
6	6	"sort"
7	7	"strings"
8	8	"text/tabwriter"
9		-
10		- "harness/internal/llm"
11	9	)
12	10
13	11	// report prints the numbers METRICS.md wants: the headline pass rate, and
@@ -61,7 +59,7 @@ func report(w io.Writer, results []Result) {
61	59	mean(results, func(r Result) float64 { return float64(r.Usage.InputTokens) }),
62	60	mean(results, func(r Result) float64 { return float64(r.Usage.CacheReadInputTokens) }),
63	61	mean(results, func(r Result) float64 { return float64(r.Usage.OutputTokens) }))
64		- fmt.Fprintf(w, "mean cost $%.4f\n", mean(results, func(r Result) float64 { return cost(r.Usage) }))
	62	+ fmt.Fprintf(w, "mean cost %s\n", costCell(results))
65	63	fmt.Fprintf(w, "mean wall %.1fs\n", mean(results, func(r Result) float64 { return float64(r.WallMS) / 1000 }))
66	64
67	65	fmt.Fprintln(w, "\ntool calls (failures in brackets)")
@@ -78,27 +76,36 @@ func report(w io.Writer, results []Result) {
78	76	tw.Flush()
79	77
80	78	fmt.Fprintln(w, "\nMETRICS.md row:")
81		- fmt.Fprintf(w, "\| <change> \| %d/%d \| %s \| %.1f \| $%.4f \|\n",
	79	+ fmt.Fprintf(w, "\| <change> \| %d/%d \| %s \| %.1f \| %s \|\n",
82	80	solved, len(order),
83	81	rate(calls["edit_file"]-fails["edit_file"], calls["edit_file"]),
84	82	mean(results, func(r Result) float64 { return float64(r.Turns) }),
85		- mean(results, func(r Result) float64 { return cost(r.Usage) }))
	83	+ costCell(results))
86	84	}
87	85
88		- // Per-MTok rates for Claude Opus 5. Cache reads and writes use the standard
89		- // 0.1x and 1.25x multipliers. Override these if you are running another model.
90		- const (
91		- inputPerMTok = 5.0
92		- outputPerMTok = 25.0
93		- cacheReadPerMTok = inputPerMTok * 0.1
94		- cacheWritePerMTok = inputPerMTok * 1.25
95		- )
	86	+ // cost prices one run. Pricing lives in the llm package so the REPL's running
	87	+ // total and this table cannot drift apart.
	88	+ func cost(r Result) float64 {
	89	+ c, _ := r.Usage.Cost(r.Model)
	90	+ return c
	91	+ }
	92	+
	93	+ // costKnown reports whether every run had priceable usage, so an unpriced
	94	+ // model produces a blank rather than $0.0000.
	95	+ func costKnown(results []Result) bool {
	96	+ for _, r := range results {
	97	+ if _, ok := r.Usage.Cost(r.Model); !ok {
	98	+ return false
	99	+ }
	100	+ }
	101	+ return len(results) > 0
	102	+ }
96	103
97		- func cost(u llm.Usage) float64 {
98		- return (float64(u.InputTokens)*inputPerMTok +
99		- float64(u.OutputTokens)*outputPerMTok +
100		- float64(u.CacheReadInputTokens)*cacheReadPerMTok +
101		- float64(u.CacheCreationInputTokens)*cacheWritePerMTok) / 1_000_000
	104	+ func costCell(results []Result) string {
	105	+ if !costKnown(results) {
	106	+ return "n/a"
	107	+ }
	108	+ return fmt.Sprintf("$%.4f", mean(results, cost))
102	109	}
103	110
104	111	func toolStats(results []Result) (calls, fails map[string]int) {

+4−1evals/runner.goView File

@@ -37,6 +37,7 @@ type Task struct {
37	37	type Result struct {
38	38	Task string `json:"task"`
39	39	Run int `json:"run"`
	40	+ Model string `json:"model"`
40	41	Pass bool `json:"pass"`
41	42	Turns int `json:"turns"`
42	43	Usage llm.Usage `json:"usage"`
@@ -71,6 +72,7 @@ func run() error {
71	72	maxTurns := flag.Int("max-turns", 40, "turn cap per run")
72	73	contextWindow := flag.Int("context-window", agent.DefaultContextWindow, "model context window")
73	74	noCache := flag.Bool("no-cache", false, "disable prompt cache breakpoints, to measure what they buy")
	75	+ noStream := flag.Bool("no-stream", false, "use the non-streaming endpoint")
74	76	timeout := flag.Duration("timeout", 10*time.Minute, "wall-clock cap per run")
75	77	tasksDir := flag.String("dir", "evals/tasks", "directory of task definitions")
76	78	outDir := flag.String("out", "evals/runs", "where to persist transcripts")
@@ -125,6 +127,7 @@ func run() error {
125	127	Effort: *effort,
126	128	ContextWindow: *contextWindow,
127	129	NoCache: *noCache,
	130	+ Stream: !*noStream,
128	131	}, timeout, keep)
129	132	mu.Lock()
130	133	results = append(results, res)
@@ -159,7 +162,7 @@ func run() error {
159	162	// The return value is named so the deferred timing assignment lands in the
160	163	// value actually returned rather than in a copy already made.
161	164	func runOne(task Task, runNo int, client *llm.Client, opts agent.Options, timeout time.Duration, keep bool) (res Result) {
162		- res = Result{Task: task.Name, Run: runNo}
	165	+ res = Result{Task: task.Name, Run: runNo, Model: client.Model}
163	166	start := time.Now()
164	167	defer func() { res.WallMS = time.Since(start).Milliseconds() }()
165	168

+20−2internal/agent/loop.goView File

@@ -50,6 +50,8 @@ type Options struct {
50	50	CompactAt float64
51	51	// NoCache disables prompt cache breakpoints, for measuring what they buy.
52	52	NoCache bool
	53	+ // Stream uses the SSE endpoint so output appears as it is produced.
	54	+ Stream bool
53	55	}
54	56
55	57	// EventKind identifies what happened, for an observer to render.
@@ -62,6 +64,8 @@ const (
62	64	EventToolResult EventKind = "tool_result"
63	65	EventUsage EventKind = "usage"
64	66	EventCompact EventKind = "compact"
	67	+ // EventTextDelta is emitted instead of EventText while streaming.
	68	+ EventTextDelta EventKind = "text_delta"
65	69	)
66	70
67	71	// Event is what the loop reports to an observer. It carries plain data only —
@@ -129,7 +133,7 @@ func (a *Agent) Run(ctx context.Context, prompt string) (string, error) {
129	133	return "", err
130	134	}
131	135
132		- resp, err := a.client.CreateMessage(ctx, a.request())
	136	+ resp, err := a.send(ctx, turn)
133	137	if err != nil {
134	138	return "", err
135	139	}
@@ -150,7 +154,8 @@ func (a *Agent) Run(ctx context.Context, prompt string) (string, error) {
150	154	// all: their signatures must come back unchanged on the next request.
151	155	a.state.Append(resp.Message())
152	156
153		- if text := resp.Text(); text != "" {
	157	+ // While streaming the text has already gone out delta by delta.
	158	+ if text := resp.Text(); text != "" && !a.opts.Stream {
154	159	a.emit(Event{Kind: EventText, Turn: turn, Text: text})
155	160	}
156	161
@@ -185,6 +190,19 @@ func (a *Agent) Run(ctx context.Context, prompt string) (string, error) {
185	190	ErrMaxTurns, a.opts.MaxTurns)
186	191	}
187	192
	193	+ // send performs one model call, streaming when asked to.
	194	+ func (a Agent) send(ctx context.Context, turn int) (llm.Response, error) {
	195	+ req := a.request()
	196	+ if !a.opts.Stream {
	197	+ return a.client.CreateMessage(ctx, req)
	198	+ }
	199	+ return a.client.StreamMessage(ctx, req, func(ev llm.StreamEvent) {
	200	+ if ev.Kind == llm.StreamText {
	201	+ a.emit(Event{Kind: EventTextDelta, Turn: turn, Text: ev.Text})
	202	+ }
	203	+ })
	204	+ }
	205	+
188	206	// dispatch runs each tool call in order and builds one tool_result per
189	207	// tool_use id. A tool that fails still produces a result — a missing one is an
190	208	// API error, and the model can usually recover from the error text.

+50−0internal/agent/loop_test.goView File

+  	}
   	return false
+  }
++ // streamingAPI serves one scripted SSE response.
++ func streamingAPI(t *testing.T, events ...string) *httptest.Server {
++ 	t.Helper()
++ 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ 		for _, e := range events {
++ 			io.WriteString(w, "event: x\ndata: "+e+"\n\n")
++ 		}
++ 	}))
++ 	t.Cleanup(srv.Close)
++ 	return srv
++ }
++ func TestStreamingRunEmitsDeltas(t *testing.T) {
++ 	srv := streamingAPI(t,
++ 		`{"type":"message_start","message":{"id":"m","model":"claude-opus-5","usage":{"input_tokens":50}}}`,
++ 		`{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
++ 		`{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"All "}}`,
++ 		`{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"done."}}`,
++ 		`{"type":"content_block_stop","index":0}`,
++ 		`{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`,
++ 	)
++ 	state := NewState()
++ 	a := New(llm.New(llm.WithBaseURL(srv.URL), llm.WithAPIKey("k")),
++ 		tools.NewRegistry(nil), state, nil, Options{Stream: true})
++ 	var deltas []string
++ 	var finalText int
++ 	a.Observe = func(ev Event) {
++ 		switch ev.Kind {
++ 		case EventTextDelta:
++ 			deltas = append(deltas, ev.Text)
++ 		case EventText:
++ 			finalText++
++ 		}
++ 	}
++ 	got, err := a.Run(context.Background(), "go")
++ 	if err != nil {
++ 		t.Fatalf("Run: %v", err)
++ 	}
++ 	if got != "All done." {
++ 		t.Errorf("answer = %q", got)
++ 	}
++ 	if strings.Join(deltas, "") != "All done." {
++ 		t.Errorf("deltas = %v", deltas)
++ 	}
++ 	if finalText != 0 {
++ 		t.Error("the whole text must not be emitted again after streaming it")
++ 	}
++ }

+102−0internal/agent/session.goaddedView File

++ package agent
++ import (
++ 	"encoding/json"
++ 	"fmt"
++ 	"os"
++ 	"path/filepath"
++ 	"sort"
++ 	"time"
++ 	"harness/internal/llm"
++ )
++ // Session is a conversation saved to disk so --resume can pick it up.
++ type Session struct {
++ 	ID       string        `json:"id"`
++ 	Root     string        `json:"root"`
++ 	Model    string        `json:"model"`
++ 	Created  time.Time     `json:"created"`
++ 	Updated  time.Time     `json:"updated"`
++ 	Turns    int           `json:"turns"`
++ 	Usage    llm.Usage     `json:"usage"`
++ 	Messages []llm.Message `json:"messages"`
++ }
++ // SessionDir is where sessions live, relative to the working root.
++ const SessionDir = ".harness/sessions"
++ // Save writes the session for id, overwriting any previous copy.
++ func (a *Agent) Save(dir, id, root, model string) error {
++ 	if err := os.MkdirAll(dir, 0o755); err != nil {
++ 		return err
++ 	}
++ 	usage, turns := a.state.Usage()
++ 	sess := Session{
++ 		ID:       id,
++ 		Root:     root,
++ 		Model:    model,
++ 		Created:  time.Now().UTC(),
++ 		Updated:  time.Now().UTC(),
++ 		Turns:    turns,
++ 		Usage:    usage,
++ 		Messages: a.state.History(),
++ 	}
++ 	if prev, err := LoadSession(filepath.Join(dir, id+".json")); err == nil {
++ 		sess.Created = prev.Created
++ 	}
++ 	b, err := json.MarshalIndent(sess, "", "  ")
++ 	if err != nil {
++ 		return err
++ 	}
++ 	return os.WriteFile(filepath.Join(dir, id+".json"), b, 0o644)
++ }
++ // LoadSession reads one session file.
++ func LoadSession(path string) (Session, error) {
++ 	var s Session
++ 	b, err := os.ReadFile(path)
++ 	if err != nil {
++ 		return s, err
++ 	}
++ 	if err := json.Unmarshal(b, &s); err != nil {
++ 		return s, fmt.Errorf("session %s: %w", path, err)
++ 	}
++ 	return s, nil
++ }
++ // ResolveSession finds a session by id, or the most recently updated one when
++ // id is "latest" or empty.
++ func ResolveSession(dir, id string) (Session, error) {
++ 	if id != "" && id != "latest" {
++ 		return LoadSession(filepath.Join(dir, id+".json"))
++ 	}
++ 	entries, err := os.ReadDir(dir)
++ 	if err != nil {
++ 		return Session{}, fmt.Errorf("no sessions in %s", dir)
++ 	}
++ 	var files []string
++ 	for _, e := range entries {
++ 		if !e.IsDir() && filepath.Ext(e.Name()) == ".json" {
++ 			files = append(files, filepath.Join(dir, e.Name()))
++ 		}
++ 	}
++ 	if len(files) == 0 {
++ 		return Session{}, fmt.Errorf("no sessions in %s", dir)
++ 	}
++ 	sort.Slice(files, func(i, j int) bool {
++ 		a, _ := os.Stat(files[i])
++ 		b, _ := os.Stat(files[j])
++ 		return a.ModTime().After(b.ModTime())
++ 	})
++ 	return LoadSession(files[0])
++ }
++ // Restore seeds the agent's state from a saved session.
++ func (a *Agent) Restore(s Session) {
++ 	a.state.Replace(s.Messages)
++ 	a.state.RestoreUsage(s.Usage, s.Turns)
++ 	// The last request's size is unknown after a restore; the next turn's
++ 	// response re-establishes it before compaction is consulted again.
++ 	a.lastInputTokens = 0
++ }

+118−0internal/agent/session_test.goaddedView File

++ package agent
++ import (
++ 	"bytes"
++ 	"context"
++ 	"os"
++ 	"path/filepath"
++ 	"strings"
++ 	"testing"
++ 	"time"
++ 	"harness/internal/llm"
++ )
++ func TestSaveAndResumeSession(t *testing.T) {
++ 	dir := t.TempDir()
++ 	f := newFakeAPI(t,
++ 		msg("tool_use", toolUseBlock("tu_1", "boom", `{}`)),
++ 		msg("end_turn", textBlock("first answer")),
++ 	)
++ 	a := newTestAgent(t, f, t.TempDir(), &bytes.Buffer{})
++ 	if _, err := a.Run(context.Background(), "the original task"); err != nil {
++ 		t.Fatalf("Run: %v", err)
++ 	}
++ 	if err := a.Save(dir, "sess-1", "/repo", "claude-opus-5"); err != nil {
++ 		t.Fatalf("Save: %v", err)
++ 	}
++ 	loaded, err := ResolveSession(dir, "sess-1")
++ 	if err != nil {
++ 		t.Fatalf("ResolveSession: %v", err)
++ 	}
++ 	if loaded.Model != "claude-opus-5" || loaded.Root != "/repo" || loaded.Turns != 2 {
++ 		t.Errorf("session = %+v", loaded)
++ 	}
++ 	if len(loaded.Messages) != len(a.State().History()) {
++ 		t.Errorf("%d messages saved, want %d", len(loaded.Messages), len(a.State().History()))
++ 	}
++ 	// A resumed agent continues the same conversation.
++ 	f2 := newFakeAPI(t, msg("end_turn", textBlock("second answer")))
++ 	b := newTestAgent(t, f2, t.TempDir(), &bytes.Buffer{})
++ 	b.Restore(loaded)
++ 	if _, err := b.Run(context.Background(), "and now this"); err != nil {
++ 		t.Fatalf("Run after restore: %v", err)
++ 	}
++ 	sent := f2.rawRequest(0)
++ 	if !strings.Contains(sent, "the original task") {
++ 		t.Errorf("resumed request lost the earlier conversation:\n%s", sent)
++ 	}
++ 	if !strings.Contains(sent, "and now this") {
++ 		t.Error("resumed request is missing the new prompt")
++ 	}
++ 	if _, turns := b.State().Usage(); turns != 3 {
++ 		t.Errorf("turns = %d, want the restored 2 plus 1", turns)
++ 	}
++ }
++ func TestSavePreservesCreatedTime(t *testing.T) {
++ 	dir := t.TempDir()
++ 	f := newFakeAPI(t, msg("end_turn", textBlock("x")), msg("end_turn", textBlock("y")))
++ 	a := newTestAgent(t, f, t.TempDir(), &bytes.Buffer{})
++ 	a.Run(context.Background(), "one")
++ 	a.Save(dir, "s", "/r", "m")
++ 	first, _ := LoadSession(filepath.Join(dir, "s.json"))
++ 	time.Sleep(5 * time.Millisecond)
++ 	a.Run(context.Background(), "two")
++ 	a.Save(dir, "s", "/r", "m")
++ 	second, _ := LoadSession(filepath.Join(dir, "s.json"))
++ 	if !second.Created.Equal(first.Created) {
++ 		t.Errorf("created moved from %v to %v across saves", first.Created, second.Created)
++ 	}
++ 	if !second.Updated.After(first.Updated) {
++ 		t.Error("updated should move forward")
++ 	}
++ }
++ func TestResolveLatestSession(t *testing.T) {
++ 	dir := t.TempDir()
++ 	for _, id := range []string{"older", "newer"} {
++ 		s := Session{ID: id, Messages: []llm.Message{{Role: llm.RoleUser}}}
++ 		writeSession(t, dir, s)
++ 		time.Sleep(10 * time.Millisecond)
++ 	}
++ 	got, err := ResolveSession(dir, "latest")
++ 	if err != nil {
++ 		t.Fatalf("ResolveSession: %v", err)
++ 	}
++ 	if got.ID != "newer" {
++ 		t.Errorf("resolved %q, want the most recently updated", got.ID)
++ 	}
++ 	if _, err := ResolveSession(t.TempDir(), "latest"); err == nil {
++ 		t.Error("an empty session directory should be an error, not an empty session")
++ 	}
++ 	if _, err := ResolveSession(dir, "nonexistent"); err == nil {
++ 		t.Error("an unknown id should be an error")
++ 	}
++ }
++ func writeSession(t *testing.T, dir string, s Session) {
++ 	t.Helper()
++ 	a := &Agent{state: NewState()}
++ 	a.state.Replace(s.Messages)
++ 	if err := a.Save(dir, s.ID, "/r", "m"); err != nil {
++ 		t.Fatal(err)
++ 	}
++ }
++ func TestLoadSessionRejectsGarbage(t *testing.T) {
++ 	dir := t.TempDir()
++ 	os.WriteFile(filepath.Join(dir, "broken.json"), []byte("{not json"), 0o644)
++ 	if _, err := ResolveSession(dir, "broken"); err == nil {
++ 		t.Error("want an error for a corrupt session file")
++ 	}
++ }

+9−0internal/agent/state.goView File

@@ -145,6 +145,15 @@ func (s *State) AddUsage(u llm.Usage) {
145	145	s.turns++
146	146	}
147	147
	148	+ // RestoreUsage sets the running totals outright, which is what resuming a
	149	+ // saved session needs: AddUsage would count the whole restored history as one
	150	+ // more turn.
	151	+ func (s *State) RestoreUsage(u llm.Usage, turns int) {
	152	+ s.mu.Lock()
	153	+ defer s.mu.Unlock()
	154	+ s.usage, s.turns = u, turns
	155	+ }
	156	+
148	157	// Usage is the running total for the session, and the number of model turns it
149	158	// took to get there.
150	159	func (s *State) Usage() (llm.Usage, int) {

+1−15internal/llm/client.goView File

@@ -214,21 +214,7 @@ func (c Client) attempt(ctx context.Context, body []byte) (Response, error) {
214	214	return nil, fmt.Errorf("llm: reading response: %w", err)
215	215	}
216	216	if httpResp.StatusCode != http.StatusOK {
217		- apiErr := &APIError{StatusCode: httpResp.StatusCode, Body: string(raw)}
218		- var envelope struct {
219		- Error struct {
220		- Type string `json:"type"`
221		- Message string `json:"message"`
222		- } `json:"error"`
223		- }
224		- if json.Unmarshal(raw, &envelope) == nil {
225		- apiErr.Type = envelope.Error.Type
226		- apiErr.Message = envelope.Error.Message
227		- }
228		- if ra := httpResp.Header.Get("retry-after"); ra != "" {
229		- apiErr.Message += " (retry-after " + ra + "s)"
230		- }
231		- return nil, apiErr
	217	+ return nil, parseAPIError(httpResp.StatusCode, httpResp.Header.Get("retry-after"), raw)
232	218	}
233	219
234	220	var out Response

+38−0internal/llm/pricing.goaddedView File

++ package llm
++ // Price is per-million-token list pricing.
++ type Price struct {
++ 	InputPerMTok  float64
++ 	OutputPerMTok float64
++ }
++ // Cache multipliers relative to the input rate.
++ const (
++ 	cacheReadMultiplier  = 0.1
++ 	cacheWriteMultiplier = 1.25
++ )
++ // prices covers the models this harness is likely to be pointed at. A model
++ // that is not listed reports no cost rather than a wrong one.
++ var prices = map[string]Price{
++ 	"claude-opus-5":    {InputPerMTok: 5, OutputPerMTok: 25},
++ 	"claude-opus-4-8":  {InputPerMTok: 5, OutputPerMTok: 25},
++ 	"claude-opus-4-7":  {InputPerMTok: 5, OutputPerMTok: 25},
++ 	"claude-sonnet-5":  {InputPerMTok: 2, OutputPerMTok: 10},
++ 	"claude-haiku-4-5": {InputPerMTok: 1, OutputPerMTok: 5},
++ 	"claude-fable-5-1": {InputPerMTok: 10, OutputPerMTok: 50},
++ 	"claude-fable-5":   {InputPerMTok: 10, OutputPerMTok: 50},
++ }
++ // Cost prices one usage record. The boolean reports whether the model's
++ // pricing is known; a made-up number in METRICS.md is worse than a blank.
++ func (u Usage) Cost(model string) (float64, bool) {
++ 	p, ok := prices[model]
++ 	if !ok {
++ 		return 0, false
++ 	}
++ 	return (float64(u.InputTokens)*p.InputPerMTok +
++ 		float64(u.OutputTokens)*p.OutputPerMTok +
++ 		float64(u.CacheReadInputTokens)*p.InputPerMTok*cacheReadMultiplier +
++ 		float64(u.CacheCreationInputTokens)*p.InputPerMTok*cacheWriteMultiplier) / 1_000_000, true
++ }

+278−0internal/llm/stream.goaddedView File

++ package llm
++ import (
++ 	"bufio"
++ 	"bytes"
++ 	"context"
++ 	"encoding/json"
++ 	"fmt"
++ 	"io"
++ 	"net/http"
++ 	"strings"
++ )
++ // StreamEventKind is what a streaming callback is being told about.
++ type StreamEventKind string
++ const (
++ 	StreamText      StreamEventKind = "text"
++ 	StreamThinking  StreamEventKind = "thinking"
++ 	StreamToolStart StreamEventKind = "tool_start"
++ 	StreamToolInput StreamEventKind = "tool_input"
++ 	StreamBlockStop StreamEventKind = "block_stop"
++ )
++ // StreamEvent is one incremental update.
++ type StreamEvent struct {
++ 	Kind     StreamEventKind
++ 	Index    int
++ 	Text     string // the delta, not the accumulated value
++ 	ToolID   string
++ 	ToolName string
++ }
++ // StreamMessage sends a streaming request, calls onEvent for each delta, and
++ // returns the assembled response. The callback may be nil.
++ func (c *Client) StreamMessage(ctx context.Context, req Request, onEvent func(StreamEvent)) (*Response, error) {
++ 	if !c.HasCredentials() {
++ 		return nil, fmt.Errorf("llm: no credentials: set ANTHROPIC_API_KEY, or run `ant auth login`")
++ 	}
++ 	if req.Model == "" {
++ 		req.Model = c.Model
++ 	}
++ 	if req.MaxTokens == 0 {
++ 		req.MaxTokens = c.MaxTokens
++ 	}
++ 	req.Stream = true
++ 	body, err := json.Marshal(req)
++ 	if err != nil {
++ 		return nil, fmt.Errorf("llm: encoding request: %w", err)
++ 	}
++ 	var lastErr error
++ 	for attempt := 0; attempt <= c.MaxRetries; attempt++ {
++ 		if attempt > 0 {
++ 			if err := sleep(ctx, c.backoff(attempt, lastErr)); err != nil {
++ 				return nil, err
++ 			}
++ 		}
++ 		resp, err := c.streamAttempt(ctx, body, onEvent)
++ 		if err == nil {
++ 			return resp, nil
++ 		}
++ 		lastErr = err
++ 		if ctx.Err() != nil {
++ 			return nil, ctx.Err()
++ 		}
++ 		var apiErr *APIError
++ 		if ok := asAPIError(err, &apiErr); ok && !apiErr.Retryable() {
++ 			return nil, err
++ 		}
++ 		// A stream that broke partway is not safely retryable either: the
++ 		// caller may already have seen deltas.
++ 		if _, isAPI := err.(*APIError); !isAPI {
++ 			return nil, err
++ 		}
++ 	}
++ 	return nil, lastErr
++ }
++ func (c *Client) streamAttempt(ctx context.Context, body []byte, onEvent func(StreamEvent)) (*Response, error) {
++ 	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, c.BaseURL+"/v1/messages", bytes.NewReader(body))
++ 	if err != nil {
++ 		return nil, err
++ 	}
++ 	httpReq.Header.Set("content-type", "application/json")
++ 	httpReq.Header.Set("accept", "text/event-stream")
++ 	httpReq.Header.Set("anthropic-version", apiVersion)
++ 	if c.apiKey != "" {
++ 		httpReq.Header.Set("x-api-key", c.apiKey)
++ 	} else {
++ 		httpReq.Header.Set("authorization", "Bearer "+c.authToken)
++ 		httpReq.Header.Set("anthropic-beta", oauthBeta)
++ 	}
++ 	httpResp, err := c.HTTP.Do(httpReq)
++ 	if err != nil {
++ 		return nil, fmt.Errorf("llm: %w", err)
++ 	}
++ 	defer httpResp.Body.Close()
++ 	if httpResp.StatusCode != http.StatusOK {
++ 		raw, _ := io.ReadAll(httpResp.Body)
++ 		return nil, parseAPIError(httpResp.StatusCode, httpResp.Header.Get("retry-after"), raw)
++ 	}
++ 	return readStream(httpResp.Body, onEvent)
++ }
++ // blockBuilder accumulates one content block across its deltas. Tool input
++ // arrives as a series of JSON fragments that are only valid once concatenated,
++ // which is the fiddly part of streaming.
++ type blockBuilder struct {
++ 	block ContentBlock
++ 	json  strings.Builder
++ }
++ func readStream(r io.Reader, onEvent func(StreamEvent)) (*Response, error) {
++ 	emit := func(ev StreamEvent) {
++ 		if onEvent != nil {
++ 			onEvent(ev)
++ 		}
++ 	}
++ 	resp := &Response{Role: RoleAssistant}
++ 	blocks := map[int]*blockBuilder{}
++ 	var order []int
++ 	sc := bufio.NewScanner(r)
++ 	sc.Buffer(make([]byte, 0, 64*1024), 8*1024*1024)
++ 	for sc.Scan() {
++ 		line := sc.Text()
++ 		if !strings.HasPrefix(line, "data:") {
++ 			continue // "event:" lines are redundant; the payload names its type
++ 		}
++ 		payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
++ 		if payload == "" || payload == "[DONE]" {
++ 			continue
++ 		}
++ 		var ev struct {
++ 			Type    string `json:"type"`
++ 			Index   int    `json:"index"`
++ 			Message *struct {
++ 				ID    string `json:"id"`
++ 				Model string `json:"model"`
++ 				Usage Usage  `json:"usage"`
++ 			} `json:"message"`
++ 			ContentBlock *ContentBlock `json:"content_block"`
++ 			Delta        *struct {
++ 				Type        string       `json:"type"`
++ 				Text        string       `json:"text"`
++ 				PartialJSON string       `json:"partial_json"`
++ 				Thinking    string       `json:"thinking"`
++ 				Signature   string       `json:"signature"`
++ 				StopReason  string       `json:"stop_reason"`
++ 				StopDetails *StopDetails `json:"stop_details"`
++ 			} `json:"delta"`
++ 			Usage *Usage `json:"usage"`
++ 			Error *struct {
++ 				Type    string `json:"type"`
++ 				Message string `json:"message"`
++ 			} `json:"error"`
++ 		}
++ 		if err := json.Unmarshal([]byte(payload), &ev); err != nil {
++ 			return nil, fmt.Errorf("llm: malformed stream event: %w", err)
++ 		}
++ 		switch ev.Type {
++ 		case "error":
++ 			if ev.Error != nil {
++ 				return nil, &APIError{StatusCode: 200, Type: ev.Error.Type, Message: ev.Error.Message}
++ 			}
++ 			return nil, fmt.Errorf("llm: stream error")
++ 		case "message_start":
++ 			if ev.Message != nil {
++ 				resp.ID, resp.Model, resp.Usage = ev.Message.ID, ev.Message.Model, ev.Message.Usage
++ 			}
++ 		case "content_block_start":
++ 			if ev.ContentBlock == nil {
++ 				continue
++ 			}
++ 			blocks[ev.Index] = &blockBuilder{block: *ev.ContentBlock}
++ 			order = append(order, ev.Index)
++ 			if ev.ContentBlock.Type == BlockToolUse {
++ 				emit(StreamEvent{Kind: StreamToolStart, Index: ev.Index,
++ 					ToolID: ev.ContentBlock.ID, ToolName: ev.ContentBlock.Name})
++ 			}
++ 		case "content_block_delta":
++ 			b, ok := blocks[ev.Index]
++ 			if !ok || ev.Delta == nil {
++ 				continue
++ 			}
++ 			switch ev.Delta.Type {
++ 			case "text_delta":
++ 				b.block.Text += ev.Delta.Text
++ 				emit(StreamEvent{Kind: StreamText, Index: ev.Index, Text: ev.Delta.Text})
++ 			case "thinking_delta":
++ 				b.block.Thinking += ev.Delta.Thinking
++ 				emit(StreamEvent{Kind: StreamThinking, Index: ev.Index, Text: ev.Delta.Thinking})
++ 			case "signature_delta":
++ 				b.block.Signature += ev.Delta.Signature
++ 			case "input_json_delta":
++ 				// Fragments are not individually parseable; buffer and parse
++ 				// the whole thing at content_block_stop.
++ 				b.json.WriteString(ev.Delta.PartialJSON)
++ 				emit(StreamEvent{Kind: StreamToolInput, Index: ev.Index, Text: ev.Delta.PartialJSON})
++ 			}
++ 		case "content_block_stop":
++ 			b, ok := blocks[ev.Index]
++ 			if !ok {
++ 				continue
++ 			}
++ 			if s := b.json.String(); s != "" {
++ 				if !json.Valid([]byte(s)) {
++ 					return nil, fmt.Errorf("llm: tool %q streamed invalid JSON arguments: %s", b.block.Name, s)
++ 				}
++ 				b.block.Input = json.RawMessage(s)
++ 			}
++ 			emit(StreamEvent{Kind: StreamBlockStop, Index: ev.Index})
++ 		case "message_delta":
++ 			if ev.Delta != nil {
++ 				resp.StopReason = ev.Delta.StopReason
++ 				if ev.Delta.StopDetails != nil {
++ 					resp.StopDetails = ev.Delta.StopDetails
++ 				}
++ 			}
++ 			if ev.Usage != nil {
++ 				// message_delta carries the final output count; the input
++ 				// counts arrived with message_start.
++ 				resp.Usage.OutputTokens = ev.Usage.OutputTokens
++ 				if ev.Usage.InputTokens > 0 {
++ 					resp.Usage.InputTokens = ev.Usage.InputTokens
++ 				}
++ 			}
++ 		}
++ 	}
++ 	if err := sc.Err(); err != nil {
++ 		return nil, fmt.Errorf("llm: reading stream: %w", err)
++ 	}
++ 	for _, i := range order {
++ 		if b := blocks[i]; b != nil {
++ 			// A tool_use with no input at all still needs a valid object.
++ 			if b.block.Type == BlockToolUse && len(b.block.Input) == 0 {
++ 				b.block.Input = json.RawMessage("{}")
++ 			}
++ 			resp.Content = append(resp.Content, b.block)
++ 		}
++ 	}
++ 	if resp.StopReason == "" {
++ 		return nil, fmt.Errorf("llm: stream ended without a stop_reason")
++ 	}
++ 	return resp, nil
++ }
++ // parseAPIError builds an APIError from a non-2xx body.
++ func parseAPIError(status int, retryAfterHeader string, raw []byte) *APIError {
++ 	apiErr := &APIError{StatusCode: status, Body: string(raw)}
++ 	var envelope struct {
++ 		Error struct {
++ 			Type    string `json:"type"`
++ 			Message string `json:"message"`
++ 		} `json:"error"`
++ 	}
++ 	if json.Unmarshal(raw, &envelope) == nil {
++ 		apiErr.Type = envelope.Error.Type
++ 		apiErr.Message = envelope.Error.Message
++ 	}
++ 	if retryAfterHeader != "" {
++ 		apiErr.Message += " (retry-after " + retryAfterHeader + "s)"
++ 	}
++ 	return apiErr
++ }

+209−0internal/llm/stream_test.goaddedView File

++ package llm
++ import (
++ 	"context"
++ 	"encoding/json"
++ 	"io"
++ 	"net/http"
++ 	"net/http/httptest"
++ 	"strings"
++ 	"testing"
++ 	"time"
++ )
++ // sse formats events the way the API does.
++ func sse(events ...string) string {
++ 	var b strings.Builder
++ 	for _, e := range events {
++ 		b.WriteString("event: x\ndata: " + e + "\n\n")
++ 	}
++ 	return b.String()
++ }
++ func streamServer(t *testing.T, payload string) *Client {
++ 	t.Helper()
++ 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ 		if r.Header.Get("accept") != "text/event-stream" {
++ 			t.Errorf("accept = %q", r.Header.Get("accept"))
++ 		}
++ 		var req map[string]any
++ 		body, _ := io.ReadAll(r.Body)
++ 		json.Unmarshal(body, &req)
++ 		if req["stream"] != true {
++ 			t.Errorf("stream flag not set in body: %s", body)
++ 		}
++ 		w.Header().Set("content-type", "text/event-stream")
++ 		io.WriteString(w, payload)
++ 	}))
++ 	t.Cleanup(srv.Close)
++ 	c := New(WithBaseURL(srv.URL), WithAPIKey("k"))
++ 	c.RetryDelay = time.Millisecond
++ 	return c
++ }
++ // The fiddly part: tool arguments arrive as fragments that are individually
++ // invalid JSON and only parseable once concatenated.
++ func TestStreamAccumulatesToolInputFragments(t *testing.T) {
++ 	c := streamServer(t, sse(
++ 		`{"type":"message_start","message":{"id":"msg_1","model":"claude-opus-5","usage":{"input_tokens":100,"cache_read_input_tokens":40}}}`,
++ 		`{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
++ 		`{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me "}}`,
++ 		`{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"look."}}`,
++ 		`{"type":"content_block_stop","index":0}`,
++ 		`{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"tu_1","name":"edit_file","input":{}}}`,
++ 		`{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"path\":"}}`,
++ 		`{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"\"main.go\",\"old_str\""}}`,
++ 		`{"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":":\"a\",\"new_str\":\"b\"}"}}`,
++ 		`{"type":"content_block_stop","index":1}`,
++ 		`{"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":55}}`,
++ 		`{"type":"message_stop"}`,
++ 	))
++ 	var textSeen strings.Builder
++ 	var toolStarts int
++ 	resp, err := c.StreamMessage(context.Background(), Request{}, func(ev StreamEvent) {
++ 		switch ev.Kind {
++ 		case StreamText:
++ 			textSeen.WriteString(ev.Text)
++ 		case StreamToolStart:
++ 			toolStarts++
++ 			if ev.ToolName != "edit_file" || ev.ToolID != "tu_1" {
++ 				t.Errorf("tool start = %+v", ev)
++ 			}
++ 		}
++ 	})
++ 	if err != nil {
++ 		t.Fatalf("StreamMessage: %v", err)
++ 	}
++ 	if textSeen.String() != "Let me look." {
++ 		t.Errorf("streamed text = %q", textSeen.String())
++ 	}
++ 	if toolStarts != 1 {
++ 		t.Errorf("%d tool starts, want 1", toolStarts)
++ 	}
++ 	if resp.StopReason != StopToolUse {
++ 		t.Errorf("stop_reason = %q", resp.StopReason)
++ 	}
++ 	if got := resp.Text(); got != "Let me look." {
++ 		t.Errorf("assembled text = %q", got)
++ 	}
++ 	uses := resp.ToolUses()
++ 	if len(uses) != 1 {
++ 		t.Fatalf("%d tool uses, want 1", len(uses))
++ 	}
++ 	var args struct{ Path, OldStr, NewStr string }
++ 	if err := json.Unmarshal(uses[0].Input, &struct {
++ 		Path   *string `json:"path"`
++ 		OldStr *string `json:"old_str"`
++ 		NewStr *string `json:"new_str"`
++ 	}{&args.Path, &args.OldStr, &args.NewStr}); err != nil {
++ 		t.Fatalf("reassembled input is not valid JSON: %v (%s)", err, uses[0].Input)
++ 	}
++ 	if args.Path != "main.go" || args.OldStr != "a" || args.NewStr != "b" {
++ 		t.Errorf("reassembled args = %+v", args)
++ 	}
++ 	if resp.Usage.InputTokens != 100 || resp.Usage.CacheReadInputTokens != 40 || resp.Usage.OutputTokens != 55 {
++ 		t.Errorf("usage = %+v: input counts come from message_start, output from message_delta", resp.Usage)
++ 	}
++ }
++ func TestStreamPreservesThinkingSignature(t *testing.T) {
++ 	c := streamServer(t, sse(
++ 		`{"type":"message_start","message":{"id":"m","usage":{"input_tokens":1}}}`,
++ 		`{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}`,
++ 		`{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig-"}}`,
++ 		`{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"abc"}}`,
++ 		`{"type":"content_block_stop","index":0}`,
++ 		`{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}`,
++ 	))
++ 	resp, err := c.StreamMessage(context.Background(), Request{}, nil)
++ 	if err != nil {
++ 		t.Fatalf("StreamMessage: %v", err)
++ 	}
++ 	if len(resp.Content) != 1 || resp.Content[0].Signature != "sig-abc" {
++ 		t.Errorf("signature = %q, want the fragments joined", resp.Content[0].Signature)
++ 	}
++ 	out, _ := json.Marshal(resp.Content[0])
++ 	if !strings.Contains(string(out), `"signature":"sig-abc"`) {
++ 		t.Errorf("signature must survive re-marshalling: %s", out)
++ 	}
++ }
++ func TestStreamErrors(t *testing.T) {
++ 	tests := []struct {
++ 		name    string
++ 		payload string
++ 		want    string
++ 	}{
++ 		{
++ 			name: "mid-stream error event",
++ 			payload: sse(
++ 				`{"type":"message_start","message":{"id":"m"}}`,
++ 				`{"type":"error","error":{"type":"overloaded_error","message":"try later"}}`),
++ 			want: "try later",
++ 		},
++ 		{
++ 			name: "truncated tool arguments",
++ 			payload: sse(
++ 				`{"type":"message_start","message":{"id":"m"}}`,
++ 				`{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t","name":"bash"}}`,
++ 				`{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"command\":"}}`,
++ 				`{"type":"content_block_stop","index":0}`,
++ 				`{"type":"message_delta","delta":{"stop_reason":"tool_use"}}`),
++ 			want: "invalid JSON arguments",
++ 		},
++ 		{
++ 			name:    "stream ends without a stop reason",
++ 			payload: sse(`{"type":"message_start","message":{"id":"m"}}`),
++ 			want:    "without a stop_reason",
++ 		},
++ 		{
++ 			name:    "malformed event",
++ 			payload: "data: {not json\n\n",
++ 			want:    "malformed stream event",
++ 		},
++ 	}
++ 	for _, tc := range tests {
++ 		t.Run(tc.name, func(t *testing.T) {
++ 			_, err := streamServer(t, tc.payload).StreamMessage(context.Background(), Request{}, nil)
++ 			if err == nil || !strings.Contains(err.Error(), tc.want) {
++ 				t.Errorf("error = %v, want it to mention %q", err, tc.want)
++ 			}
++ 		})
++ 	}
++ }
++ func TestStreamToolUseWithNoArguments(t *testing.T) {
++ 	c := streamServer(t, sse(
++ 		`{"type":"message_start","message":{"id":"m"}}`,
++ 		`{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"t","name":"pwd"}}`,
++ 		`{"type":"content_block_stop","index":0}`,
++ 		`{"type":"message_delta","delta":{"stop_reason":"tool_use"}}`,
++ 	))
++ 	resp, err := c.StreamMessage(context.Background(), Request{}, nil)
++ 	if err != nil {
++ 		t.Fatalf("StreamMessage: %v", err)
++ 	}
++ 	if got := string(resp.ToolUses()[0].Input); got != "{}" {
++ 		t.Errorf("input = %s, want an empty object", got)
++ 	}
++ }
++ func TestStreamHTTPErrorIsAnAPIError(t *testing.T) {
++ 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
++ 		w.WriteHeader(http.StatusBadRequest)
++ 		io.WriteString(w, `{"error":{"type":"invalid_request_error","message":"bad model"}}`)
++ 	}))
++ 	defer srv.Close()
++ 	c := New(WithBaseURL(srv.URL), WithAPIKey("k"))
++ 	_, err := c.StreamMessage(context.Background(), Request{}, nil)
++ 	apiErr, ok := err.(*APIError)
++ 	if !ok {
++ 		t.Fatalf("error type = %T, want *APIError", err)
++ 	}
++ 	if apiErr.StatusCode != 400 || apiErr.Message != "bad model" {
++ 		t.Errorf("APIError = %+v", apiErr)
++ 	}
++ }

+46−0internal/ui/diff.goaddedView File

++ package ui
++ import (
++ 	"fmt"
++ 	"io"
++ 	"strings"
++ )
++ // diff colours are separate from the palette so a unified diff reads the same
++ // way it does in git.
++ var (
++ 	green = "\x1b[32m"
++ 	red   = "\x1b[31m"
++ )
++ func init() {
++ 	if color.reset == "" { // not a terminal, or NO_COLOR
++ 		green, red = "", ""
++ 	}
++ }
++ // WriteDiff renders a unified diff with colour. Anything that is not a diff is
++ // written through unchanged, so a tool result that is a plain message still
++ // reads normally.
++ func WriteDiff(w io.Writer, s string) {
++ 	for _, line := range strings.Split(strings.TrimRight(s, "\n"), "\n") {
++ 		switch {
++ 		case strings.HasPrefix(line, "+++") || strings.HasPrefix(line, "---"):
++ 			fmt.Fprintf(w, "%s%s%s\n", color.bold, line, color.reset)
++ 		case strings.HasPrefix(line, "@@"):
++ 			fmt.Fprintf(w, "%s%s%s\n", color.cyan, line, color.reset)
++ 		case strings.HasPrefix(line, "+"):
++ 			fmt.Fprintf(w, "%s%s%s\n", green, line, color.reset)
++ 		case strings.HasPrefix(line, "-"):
++ 			fmt.Fprintf(w, "%s%s%s\n", red, line, color.reset)
++ 		default:
++ 			fmt.Fprintf(w, "%s\n", line)
++ 		}
++ 	}
++ }
++ // looksLikeDiff reports whether s is a unified diff, so tool output can be
++ // rendered appropriately without the tool having to say which it is.
++ func looksLikeDiff(s string) bool {
++ 	return strings.HasPrefix(s, "--- ") && strings.Contains(s, "\n@@ ")
++ }

+37−4internal/ui/render.goView File

@@ -31,8 +31,21 @@ var color = func() palette {
31	31	type Renderer struct {
32	32	W io.Writer
33	33	Verbose bool // print full tool results rather than a summary line
	34	+
	35	+ midLine bool // streamed text is mid-line and needs a newline before anything else
	36	+ }
	37	+
	38	+ // endLine closes a partially written line of streamed text.
	39	+ func (r *Renderer) endLine() {
	40	+ if r.midLine {
	41	+ fmt.Fprintln(r.W)
	42	+ r.midLine = false
	43	+ }
34	44	}
35	45
	46	+ // Flush closes any open line. Call it when a turn finishes.
	47	+ func (r *Renderer) Flush() { r.endLine() }
	48	+
36	49	func New(w io.Writer) *Renderer { return &Renderer{W: w} }
37	50
38	51	// Handle is the agent.Observe callback.
@@ -40,16 +53,30 @@ func (r *Renderer) Handle(ev agent.Event) {
40	53	switch ev.Kind {
41	54	case agent.EventText:
42	55	fmt.Fprintf(r.W, "%s\n", strings.TrimSpace(ev.Text))
	56	+ case agent.EventTextDelta:
	57	+ // Streamed text arrives mid-line; the newline comes when the tool call
	58	+ // or the turn ends.
	59	+ fmt.Fprint(r.W, ev.Text)
	60	+ r.midLine = true
43	61	case agent.EventToolStart:
	62	+ r.endLine()
44	63	fmt.Fprintf(r.W, "%s⏺ %s%s%s(%s)%s\n", color.cyan, color.bold, ev.Tool, color.reset, summarizeInput(ev.Input), color.reset)
45	64	case agent.EventToolResult:
	65	+ r.endLine()
46	66	r.tool(ev)
47	67	case agent.EventCompact:
	68	+ r.endLine()
48	69	fmt.Fprintf(r.W, "%s[compacted the conversation]%s\n", color.dim, color.reset)
49	70	}
50	71	}
51	72
52	73	func (r *Renderer) tool(ev agent.Event) {
	74	+ // A diff is the one tool result worth showing in full and in colour: it is
	75	+ // the record of what actually changed.
	76	+ if !ev.IsError && looksLikeDiff(ev.Result) {
	77	+ WriteDiff(r.W, ev.Result)
	78	+ return
	79	+ }
53	80	prefix, c := " ⎿ ", color.dim
54	81	if ev.IsError {
55	82	prefix, c = " ⎿ error: ", color.red
@@ -99,13 +126,19 @@ func firstLines(s string, n int) string {
99	126	return strings.Join(lines[:n], "\n") + fmt.Sprintf("\n… %d more lines", len(lines)-n)
100	127	}
101	128
102		- // Usage prints the running token totals for the session.
103		- func (r *Renderer) Usage(u llm.Usage, turns int) {
104		- fmt.Fprintf(r.W, "%s%d turns · in %d (%d cached) · out %d%s\n",
105		- color.dim, turns, u.InputTokens, u.CacheReadInputTokens, u.OutputTokens, color.reset)
	129	+ // Usage prints the running token and cost totals for the session.
	130	+ func (r *Renderer) Usage(u llm.Usage, turns int, model string) {
	131	+ r.endLine()
	132	+ line := fmt.Sprintf("%d turns · in %d (%d cached) · out %d",
	133	+ turns, u.InputTokens, u.CacheReadInputTokens, u.OutputTokens)
	134	+ if cost, known := u.Cost(model); known {
	135	+ line += fmt.Sprintf(" · $%.4f", cost)
	136	+ }
	137	+ fmt.Fprintf(r.W, "%s%s%s\n", color.dim, line, color.reset)
106	138	}
107	139
108	140	// Error prints a failure.
109	141	func (r *Renderer) Error(err error) {
	142	+ r.endLine()
110	143	fmt.Fprintf(r.W, "%serror: %v%s\n", color.red, err, color.reset)
111	144	}

+50−3internal/ui/render_test.goView File

   func TestUsageLine(t *testing.T) {
   	var buf bytes.Buffer
+- 	New(&buf).Usage(llm.Usage{InputTokens: 1200, OutputTokens: 300, CacheReadInputTokens: 900}, 4)
+- 	if got := buf.String(); !strings.Contains(got, "4 turns") || !strings.Contains(got, "900 cached") {
+- 		t.Errorf("got %q, want turns and cache reads", got)
++ 	New(&buf).Usage(llm.Usage{InputTokens: 1200, OutputTokens: 300, CacheReadInputTokens: 900}, 4, "claude-opus-5")
++ 	got := buf.String()
++ 	for _, want := range []string{"4 turns", "900 cached", "$"} {
++ 		if !strings.Contains(got, want) {
++ 			t.Errorf("usage line %q is missing %q", got, want)
++ 		}
++ 	}
++ }
++ func TestUsageOmitsCostForAnUnknownModel(t *testing.T) {
++ 	var buf bytes.Buffer
++ 	New(&buf).Usage(llm.Usage{InputTokens: 10}, 1, "some-gateway-model")
++ 	if strings.Contains(buf.String(), "$") {
++ 		t.Errorf("got %q: a made-up cost is worse than none", buf.String())
++ 	}
++ }
++ func TestStreamedTextIsClosedBeforeOtherOutput(t *testing.T) {
++ 	var buf bytes.Buffer
++ 	r := New(&buf)
++ 	r.Handle(agent.Event{Kind: agent.EventTextDelta, Text: "Let me "})
++ 	r.Handle(agent.Event{Kind: agent.EventTextDelta, Text: "look."})
++ 	r.Handle(agent.Event{Kind: agent.EventToolStart, Tool: "read_file", Input: []byte(`{"path":"a.go"}`)})
++ 	if !strings.HasPrefix(buf.String(), "Let me look.\n") {
++ 		t.Errorf("streamed text should be flushed onto its own line, got %q", buf.String())
++ 	}
++ }
++ func TestDiffResultIsRenderedAsADiff(t *testing.T) {
++ 	var buf bytes.Buffer
++ 	diff := "--- a.go\n+++ a.go\n@@ -1,2 +1,2 @@\n-old\n+new"
++ 	New(&buf).Handle(agent.Event{Kind: agent.EventToolResult, Tool: "edit_file", Result: diff})
++ 	got := buf.String()
++ 	if strings.Contains(got, "⎿") {
++ 		t.Errorf("a diff should not be summarized like ordinary tool output:\n%s", got)
++ 	}
++ 	for _, want := range []string{"-old", "+new", "@@ -1,2 +1,2 @@"} {
++ 		if !strings.Contains(got, want) {
++ 			t.Errorf("diff missing %q:\n%s", want, got)
++ 		}
++ 	}
++ }
++ func TestLooksLikeDiff(t *testing.T) {
++ 	if !looksLikeDiff("--- a.go\n+++ a.go\n@@ -1 +1 @@\n-x\n+y") {
++ 		t.Error("a unified diff should be recognised")
++ 	}
++ 	if looksLikeDiff("Created a.go (12 bytes, 1 lines).") {
++ 		t.Error("ordinary tool output should not be treated as a diff")
+  	}
+  }