Vectrum
Sign In
CodeyPublic
Like0 likesFork0 forks

agent: the model's reasoning, as it is produced

The loop was throwing thinking away: llm already parsed the deltas, and
send() only forwarded text. It now emits EventThinkingDelta while
streaming and EventThinking on the whole-response path, and the ACP front
end sends them as agent_thought_chunk so T3 Code's reasoning view fills
in.

Reasoning and the answer share one buffer but never share a chunk —
appending one kind flushes the other — because a client that got them
merged would show the model thinking out loud in the middle of its reply.
Thinking is emitted before text for the same reason: that is the order it
happened in.

Redacted blocks are not reported. They are encrypted bytes that go back to
the API untouched, with nothing in them to read, and an empty thought
chunk is a blank row in somebody's UI. They still round-trip in the
history, which is what the API checks.

The TUI ignores both kinds for now, as it ignored every kind it does not
know; showing reasoning there is a UI decision, not a plumbing one.
Browse Files
Jjonicommitted 1 parent 299ab40commit919c67d

Showing 6 changed files with 226 additions and 18 deletions.

+5−6README.mdView File
@@ -1040,12 +1040,11 @@ thing. The per-instance environment is how Codey gets a credential without
10401040 inheriting whatever the desktop app was launched with — and without it, the
10411041 provider card will say Codey is installed and not logged in, which is true.
10421042
1043- Threads stream text, show tool calls as they run, raise an approval prompt you
1044- answer in T3 Code's own panel, and stop when you press stop. The model picker
1045- is filled from the endpoint Codey is pointed at. What does not work: resuming a
1046- thread after a restart (Codey does not advertise session loading, so T3 Code
1047- will not offer it), and the reasoning view, because the turn loop does not
1048- report thinking to a front end yet.
1043+ Threads stream text and reasoning, fill a tool row in as the command talks,
1044+ raise an approval prompt you answer in T3 Code's own panel, and stop when you
1045+ press stop. The model picker is filled from the endpoint Codey is pointed at.
1046+ What does not work is resuming a thread after a restart: Codey does not
1047+ advertise session loading, so T3 Code will not offer it.
10491048
10501049 The card will say Grok and show the xAI glyph. The display name and accent
10511050 colour are yours to set; the icon is not. That is the price of occupying a slot
+22−2internal/agent/loop.goView File
@@ -169,6 +169,14 @@ const (
169169 EventNotice EventKind = "notice"
170170 // EventTextDelta is emitted instead of EventText while streaming.
171171 EventTextDelta EventKind = "text_delta"
172+ // EventThinking is the reasoning the model showed before answering, and
173+ // EventThinkingDelta is the same thing arriving piece by piece while
174+ // streaming. Only models asked for extended thinking produce either.
175+ //
176+ // Redacted thinking is not reported at all: it is encrypted bytes that go
177+ // back to the API untouched, with nothing in them to show.
178+ EventThinking EventKind = "thinking"
179+ EventThinkingDelta EventKind = "thinking_delta"
172180 // EventToolDelta is output a tool has produced but not finished producing:
173181 // the lines of a test run as they appear, rather than all of them at the
174182 // end. Text carries the new bytes, not everything so far, and Tool and ID
@@ -438,7 +446,16 @@ func (a *Agent) drive(ctx context.Context) (string, error) {
438446 // go quiet for a search and see nothing of what it asked or found.
439447 a.emitServerTools(turn, resp)
440448
441- // While streaming the text has already gone out delta by delta.
449+ // While streaming both of these have already gone out delta by delta.
450+ // Thinking goes first because that is the order it was produced in:
451+ // the model reasons and then answers, and a front end that showed them
452+ // the other way round would be describing a conversation that did not
453+ // happen.
454+ if !a.opts.Stream {
455+ if thinking := resp.Thinking(); thinking != "" {
456+ a.emit(Event{Kind: EventThinking, Turn: turn, Text: thinking})
457+ }
458+ }
442459 if text := resp.Text(); text != "" {
443460 last = text
444461 if !a.opts.Stream {
@@ -491,8 +508,11 @@ func (a *Agent) send(ctx context.Context, turn int) (*llm.Response, error) {
491508 return a.client.CreateMessage(ctx, req)
492509 }
493510 return a.client.StreamMessage(ctx, req, func(ev llm.StreamEvent) {
494- if ev.Kind == llm.StreamText {
511+ switch ev.Kind {
512+ case llm.StreamText:
495513 a.emit(Event{Kind: EventTextDelta, Turn: turn, Text: ev.Text})
514+ case llm.StreamThinking:
515+ a.emit(Event{Kind: EventThinkingDelta, Turn: turn, Text: ev.Text})
496516 }
497517 })
498518 }
+67−0internal/agent/loop_test.goView File
@@ -981,3 +981,70 @@ func TestNoObserverMeansNoProgressSink(t *testing.T) {
981981 t.Error("a sink was installed with no observer to receive it")
982982 }
983983 }
984+
985+ func thinkingBlock(s string) string {
986+ return `{"type":"thinking","thinking":"` + s + `","signature":"sig"}`
987+ }
988+
989+ func TestReasoningReachesTheObserver(t *testing.T) {
990+ f := newFakeAPI(t, msg("end_turn", thinkingBlock("the parser is the suspect"), textBlock("it is the parser")))
991+
992+ var logBuf bytes.Buffer
993+ a := newTestAgent(t, f, t.TempDir(), &logBuf)
994+
995+ var thoughts, text strings.Builder
996+ a.Observe = func(ev Event) {
997+ switch ev.Kind {
998+ case EventThinking:
999+ thoughts.WriteString(ev.Text)
1000+ case EventText:
1001+ text.WriteString(ev.Text)
1002+ }
1003+ }
1004+
1005+ if _, err := a.Run(context.Background(), "why does it fail?"); err != nil {
1006+ t.Fatalf("Run: %v", err)
1007+ }
1008+ if got := thoughts.String(); got != "the parser is the suspect" {
1009+ t.Errorf("thinking = %q", got)
1010+ }
1011+ if got := text.String(); got != "it is the parser" {
1012+ t.Errorf("text = %q", got)
1013+ }
1014+ }
1015+
1016+ func TestRedactedReasoningIsNotReported(t *testing.T) {
1017+ // Encrypted bytes that go back to the API untouched. There is nothing in
1018+ // them to show, and an empty thought chunk is a blank row in a UI.
1019+ f := newFakeAPI(t, msg("end_turn",
1020+ `{"type":"redacted_thinking","data":"AAAA"}`, textBlock("done")))
1021+
1022+ var logBuf bytes.Buffer
1023+ a := newTestAgent(t, f, t.TempDir(), &logBuf)
1024+
1025+ var thinking int
1026+ a.Observe = func(ev Event) {
1027+ if ev.Kind == EventThinking {
1028+ thinking++
1029+ }
1030+ }
1031+ if _, err := a.Run(context.Background(), "go"); err != nil {
1032+ t.Fatalf("Run: %v", err)
1033+ }
1034+ if thinking != 0 {
1035+ t.Errorf("%d thinking events for a redacted block", thinking)
1036+ }
1037+ // And it must still be in the history: the API rejects a follow-up request
1038+ // whose redacted block was dropped or altered.
1039+ var kept bool
1040+ for _, m := range a.State().History() {
1041+ for _, b := range m.Content {
1042+ if b.Type == llm.BlockRedactedThinking {
1043+ kept = true
1044+ }
1045+ }
1046+ }
1047+ if !kept {
1048+ t.Error("the redacted block was dropped from the conversation")
1049+ }
1050+ }
+15−0internal/llm/types.goView File
@@ -393,3 +393,18 @@ func (r *Response) Text() string {
393393 }
394394 return s
395395 }
396+
397+ // Thinking concatenates the reasoning the model showed.
398+ //
399+ // Redacted blocks contribute nothing. They are encrypted bytes that have to go
400+ // back to the API untouched on the next request, and there is no text in them
401+ // for anyone to read.
402+ func (r *Response) Thinking() string {
403+ var s string
404+ for _, b := range r.Content {
405+ if b.Type == BlockThinking {
406+ s += b.Thinking
407+ }
408+ }
409+ return s
410+ }
+44−10internal/serve/update.goView File
@@ -47,9 +47,15 @@ type emitter struct {
4747 afterTime time.Duration
4848 afterSize int
4949
50- mu sync.Mutex
51- buf strings.Builder
52- since time.Time
50+ // buf holds text that has not gone out yet, and bufKind says which kind
51+ // of chunk it will become. Reasoning and the answer share the buffer but
52+ // never share a chunk: appending one kind flushes the other first, because
53+ // a client that got them merged would show the model thinking out loud in
54+ // the middle of its answer.
55+ mu sync.Mutex
56+ buf strings.Builder
57+ bufKind string
58+ since time.Time
5359
5460 // anon names tool calls the model did not give an id to — the hosted
5561 // tools, whose result blocks can arrive after a pause split them from
@@ -93,13 +99,20 @@ func newEmitter(send func(acp.Update)) *emitter {
9399 func (e *emitter) handle(ev agent.Event) {
94100 switch ev.Kind {
95101 case agent.EventTextDelta:
96- e.appendText(ev.Text)
102+ e.appendText(acp.UpdateAgentMessageChunk, ev.Text)
97103
98104 case agent.EventText:
99105 // The non-streaming path: the whole message at once.
100106 e.Flush()
101107 e.chunk(acp.UpdateAgentMessageChunk, ev.Text)
102108
109+ case agent.EventThinkingDelta:
110+ e.appendText(acp.UpdateAgentThoughtChunk, ev.Text)
111+
112+ case agent.EventThinking:
113+ e.Flush()
114+ e.chunk(acp.UpdateAgentThoughtChunk, ev.Text)
115+
103116 case agent.EventToolStart:
104117 // Anything buffered is said before the tool runs. Otherwise a sentence
105118 // written just before a thirty-second command would sit in the buffer
@@ -126,24 +139,35 @@ func (e *emitter) handle(ev agent.Event) {
126139 }
127140 }
128141
129- func (e *emitter) appendText(text string) {
142+ func (e *emitter) appendText(kind, text string) {
130143 if text == "" {
131144 return
132145 }
146+
133147 e.mu.Lock()
148+ // A change of kind ends whatever was buffered, wherever it had got to.
149+ var switched string
150+ switchedKind := e.bufKind
151+ if e.bufKind != kind && e.buf.Len() > 0 {
152+ switched = e.takeLocked()
153+ }
134154 if e.buf.Len() == 0 {
135155 e.since = e.now()
136156 }
157+ e.bufKind = kind
137158 e.buf.WriteString(text)
138- ready := e.buf.Len() >= e.afterSize || e.now().Sub(e.since) >= e.afterTime
159+
139160 var out string
140- if ready {
161+ if e.buf.Len() >= e.afterSize || e.now().Sub(e.since) >= e.afterTime {
141162 out = e.takeLocked()
142163 }
143164 e.mu.Unlock()
144165
166+ if switched != "" {
167+ e.chunk(switchedKind, switched)
168+ }
145169 if out != "" {
146- e.chunk(acp.UpdateAgentMessageChunk, out)
170+ e.chunk(kind, out)
147171 }
148172 }
149173
@@ -152,10 +176,11 @@ func (e *emitter) appendText(text string) {
152176 // message is usually short and recent, so nothing else would push it out.
153177 func (e *emitter) Flush() {
154178 e.mu.Lock()
179+ kind := e.bufKind
155180 out := e.takeLocked()
156181 e.mu.Unlock()
157182 if out != "" {
158- e.chunk(acp.UpdateAgentMessageChunk, out)
183+ e.chunk(kind, out)
159184 }
160185 }
161186
@@ -165,8 +190,17 @@ func (e *emitter) takeLocked() string {
165190 return out
166191 }
167192
193+ // chunkKindOr defends the one case takeLocked cannot: a flush with nothing
194+ // buffered has no kind, and an untagged update is one a client drops.
195+ func chunkKindOr(kind string) string {
196+ if kind == "" {
197+ return acp.UpdateAgentMessageChunk
198+ }
199+ return kind
200+ }
201+
168202 func (e *emitter) chunk(kind, text string) {
169- e.send(acp.Update{SessionUpdate: kind, Content: acp.TextContent(text)})
203+ e.send(acp.Update{SessionUpdate: chunkKindOr(kind), Content: acp.TextContent(text)})
170204 }
171205
172206 func (e *emitter) toolCall(ev agent.Event) {
+73−0internal/serve/update_test.goView File
@@ -346,3 +346,76 @@ func TestProgressForAFinishedCallStartsCleanNextTime(t *testing.T) {
346346 }
347347 }
348348 }
349+
350+ // thoughts is everything the recorder saw as reasoning.
351+ func (r *recorder) thoughts() string {
352+ var b strings.Builder
353+ for _, u := range r.updates {
354+ if u.SessionUpdate != acp.UpdateAgentThoughtChunk {
355+ continue
356+ }
357+ if c, ok := u.Content.(*acp.ContentBlock); ok {
358+ b.WriteString(c.Text)
359+ }
360+ }
361+ return b.String()
362+ }
363+
364+ func TestReasoningStreamsAsThoughtNotAsAnswer(t *testing.T) {
365+ e, r := newRecorded()
366+
367+ e.handle(agent.Event{Kind: agent.EventThinkingDelta, Text: "the failing test is "})
368+ e.handle(agent.Event{Kind: agent.EventThinkingDelta, Text: "about the parser"})
369+ e.Flush()
370+
371+ if got := r.thoughts(); got != "the failing test is about the parser" {
372+ t.Errorf("thoughts = %q", got)
373+ }
374+ if got := r.text(); got != "" {
375+ t.Errorf("reasoning leaked into the answer: %q", got)
376+ }
377+ }
378+
379+ func TestThoughtAndAnswerNeverShareAChunk(t *testing.T) {
380+ // The model reasons and then answers. Merged into one chunk, a client
381+ // would show it thinking out loud in the middle of its reply.
382+ e, r := newRecorded()
383+
384+ e.handle(agent.Event{Kind: agent.EventThinkingDelta, Text: "first I check"})
385+ e.handle(agent.Event{Kind: agent.EventTextDelta, Text: "Here is the answer."})
386+ e.Flush()
387+
388+ if got, want := r.kinds(), []string{acp.UpdateAgentThoughtChunk, acp.UpdateAgentMessageChunk}; !equal(got, want) {
389+ t.Fatalf("updates = %v, want %v", got, want)
390+ }
391+ if r.thoughts() != "first I check" {
392+ t.Errorf("thoughts = %q", r.thoughts())
393+ }
394+ if r.text() != "Here is the answer." {
395+ t.Errorf("text = %q", r.text())
396+ }
397+ }
398+
399+ func TestReasoningIsFlushedBeforeAToolRuns(t *testing.T) {
400+ e, r := newRecorded()
401+
402+ e.handle(agent.Event{Kind: agent.EventThinkingDelta, Text: "I should look at the file"})
403+ e.handle(agent.Event{Kind: agent.EventToolStart, Tool: "read_file", ID: "t1",
404+ Input: json.RawMessage(`{"path":"x.go"}`)})
405+
406+ want := []string{acp.UpdateAgentThoughtChunk, acp.UpdateToolCall}
407+ if got := r.kinds(); !equal(got, want) {
408+ t.Errorf("updates = %v, want %v", got, want)
409+ }
410+ }
411+
412+ func TestTheWholeThoughtAtOnceOnTheNonStreamingPath(t *testing.T) {
413+ e, r := newRecorded()
414+
415+ e.handle(agent.Event{Kind: agent.EventThinking, Text: "considered it"})
416+ e.handle(agent.Event{Kind: agent.EventText, Text: "done"})
417+
418+ if r.thoughts() != "considered it" || r.text() != "done" {
419+ t.Errorf("thoughts = %q, text = %q", r.thoughts(), r.text())
420+ }
421+ }