From 5c1cf6b15c0e9f66c81ddaeb816d9e94e13adff7 Mon Sep 17 00:00:00 2001 From: sqlrush Date: Wed, 3 Jun 2026 17:01:01 +0800 Subject: [PATCH 1/3] fix(llmapp): order streamed text via sealed control segments Route authoritative assistant text through the ctrl FIFO alongside tool events. Keep TokenStream as preview-only and make View read-only over previewNodes. Spec: spec-1.21.1-stream-ordering-fix.md --- git-hooks/spec-registry.txt | 1 + internal/app/cli/llmapp/bullet_test.go | 65 ---------- internal/app/cli/llmapp/doc.go | 11 +- internal/app/cli/llmapp/loop_adapter.go | 63 +++++---- internal/app/cli/llmapp/model.go | 163 ++++++++++-------------- internal/app/cli/llmapp/model_test.go | 140 ++++++++++++++++++-- internal/app/cli/llmapp/stream.go | 69 ++++------ internal/app/cli/llmapp/stream_test.go | 48 ------- internal/app/cli/llmapp/view_test.go | 6 +- 9 files changed, 276 insertions(+), 290 deletions(-) delete mode 100644 internal/app/cli/llmapp/bullet_test.go delete mode 100644 internal/app/cli/llmapp/stream_test.go diff --git a/git-hooks/spec-registry.txt b/git-hooks/spec-registry.txt index 9c6ce5b..5577eac 100644 --- a/git-hooks/spec-registry.txt +++ b/git-hooks/spec-registry.txt @@ -54,3 +54,4 @@ 1.24 1 47 acceptance tag 1.4a 1 48 sub skip 1.25 1 49 main tag +1.21.1 1 50 sub tag diff --git a/internal/app/cli/llmapp/bullet_test.go b/internal/app/cli/llmapp/bullet_test.go deleted file mode 100644 index f012476..0000000 --- a/internal/app/cli/llmapp/bullet_test.go +++ /dev/null @@ -1,65 +0,0 @@ -// Copyright 2026 opendbx contributors. See LICENSE. -// -// Author: sqlrush - -package llmapp - -import ( - "testing" - - "github.com/sqlrush/opendbx/internal/app/cli/render/block" -) - -// TestMarkAssistantBullet_FirstContentOnly: the bullet lands on the first -// non-empty content Message only; thinking/empty/tool nodes are skipped and -// the bullet is consumed (pending→false) once placed. -func TestMarkAssistantBullet_FirstContentOnly(t *testing.T) { - t.Parallel() - nodes := []block.RenderNode{ - block.Message{Empty: true}, // thinking-only placeholder: skip - block.Message{Text: "first answer"}, // ← gets the bullet - block.Message{Text: "second line same turn"}, // no bullet (per-turn) - } - out, pending := markAssistantBullet(nodes, true) - if pending { - t.Errorf("bullet should be consumed (pending=false) after placement") - } - m0 := out[0].(block.Message) - m1 := out[1].(block.Message) - m2 := out[2].(block.Message) - if m0.Speaker == block.SpeakerAssistant { - t.Errorf("empty placeholder must NOT get the bullet") - } - if m1.Speaker != block.SpeakerAssistant { - t.Errorf("first content node should get SpeakerAssistant, got %v", m1.Speaker) - } - if m2.Speaker == block.SpeakerAssistant { - t.Errorf("second content node must NOT get a bullet (per-turn)") - } -} - -// TestMarkAssistantBullet_StaysPendingNoContent: a drain with no content -// Message (e.g. thinking-only) keeps the bullet pending for a later drain in -// the same turn. -func TestMarkAssistantBullet_StaysPendingNoContent(t *testing.T) { - t.Parallel() - nodes := []block.RenderNode{block.Message{Empty: true}} - _, pending := markAssistantBullet(nodes, true) - if !pending { - t.Errorf("no content node → bullet must stay pending") - } -} - -// TestMarkAssistantBullet_NotPendingNoop: when not pending, nodes are -// untouched. -func TestMarkAssistantBullet_NotPendingNoop(t *testing.T) { - t.Parallel() - nodes := []block.RenderNode{block.Message{Text: "x"}} - out, pending := markAssistantBullet(nodes, false) - if pending { - t.Errorf("pending should remain false") - } - if out[0].(block.Message).Speaker == block.SpeakerAssistant { - t.Errorf("not-pending must not mark any node") - } -} diff --git a/internal/app/cli/llmapp/doc.go b/internal/app/cli/llmapp/doc.go index b70eb18..48fa16a 100644 --- a/internal/app/cli/llmapp/doc.go +++ b/internal/app/cli/llmapp/doc.go @@ -14,11 +14,12 @@ // ctrl chan (pure) and returns loopStartCmd; the goroutine + diagnose. // Loop.Run + provider IO + cancel fire live in Cmds / Cleanup. // -// text/control split (R2 CRIT-2 + spec-1.21 D-6 union extension): -// renderable text → TokenStream (Drained in View); control -// (VisibleContent / ThinkingToken + *llm.ToolUse / *llm.ToolResult / Finish + -// TermCode) → ctrl chan → reader-Cmd → Update. Thinking / tool events -// never enter the FROZEN streaming.Chunk. +// text/control split (R2 CRIT-2 + spec-1.21.1 ordering fix): +// renderable text is previewed through TokenStream but committed through +// sealed text control messages in the same ctrl FIFO as tool events. +// PreviewTick wakes Update to refresh previewNodes; ThinkingToken, +// *llm.ToolUse, *llm.ToolResult, Finish and TermCode remain control-only. +// Thinking / tool events never enter the FROZEN streaming.Chunk. // // cli-tree root (peer to cmd/opendbx); not part of the §3.1 render DAG. package llmapp diff --git a/internal/app/cli/llmapp/loop_adapter.go b/internal/app/cli/llmapp/loop_adapter.go index 992c2d0..fccd7cc 100644 --- a/internal/app/cli/llmapp/loop_adapter.go +++ b/internal/app/cli/llmapp/loop_adapter.go @@ -2,16 +2,17 @@ // // Author: sqlrush -// File loop_adapter.go — bridges diagnose.Loop events into the existing -// llmapp control plumbing (streamControlMsg + readControlMsg + ctrl -// channel), preserving the spec-1.20 R2 CRIT-2 token-vs-control split: +// File loop_adapter.go — bridges diagnose.Loop events into llmapp's +// control plumbing (streamControlMsg + readControlMsg + ctrl channel). +// spec-1.21.1 fixes the old token-vs-control merge race by making text +// authoritative through the same ctrl FIFO as tool events: // -// - EventText visible → TokenStream.AppendChunk + streamControlMsg +// - EventText visible → segBuf + TokenStream preview + PreviewTick // - EventText thinking → streamControlMsg{ThinkingToken}, never TokenStream -// - EventToolCall → streamControlMsg carrying *llm.ToolUse -// - EventToolResult → streamControlMsg carrying *llm.ToolResult -// - EventFinish → streamControlMsg carrying Finish + TermCode + Err -// - EventTurnStart → no-op on the UI (turn boundaries are not yet +// - EventToolCall → seal visible segBuf, then *llm.ToolUse +// - EventToolResult → seal visible segBuf, then *llm.ToolResult +// - EventFinish → seal visible segBuf, then Finish + TermCode + Err +// - EventTurnStart → no-op on the UI (turn boundaries are not yet // rendered; reserved for future progress indicators) // // Backpressure parity (spec-1.21 T-2.1 HIGH-2 / D-6): every ctrl send @@ -47,6 +48,21 @@ func makeEmit(ts *streaming.TokenStream, ctrl chan<- streamControlMsg, stripThin return ctx.Err() } } + + // spec-1.21.1 segment-mode contract: makeEmit is called by the + // diagnose.Loop producer goroutine in event order, so segBuf is + // intentionally goroutine-local and unsynchronized. Do not reuse this + // closure from multiple producers without re-specifying ordering. + var segBuf string + seal := func(ctx context.Context, truncated bool) error { + if segBuf == "" { + return nil + } + text := segBuf + segBuf = "" + return send(ctx, streamControlMsg{SealedText: text, SealedTruncated: truncated}) + } + return func(ctx context.Context, e diagnose.Event) error { switch e.Kind { case diagnose.EventText: @@ -57,27 +73,29 @@ func makeEmit(ts *streaming.TokenStream, ctrl chan<- streamControlMsg, stripThin } return send(ctx, msg) } - visible := e.Text != "" - if visible { - _ = ts.AppendChunk(streaming.Chunk{Token: e.Text}) - sb.addText(e.Text) // spec-1.23 D-3: accumulate the visible final answer + if e.Text == "" { + return nil } - return send(ctx, streamControlMsg{VisibleContent: visible}) + segBuf += e.Text + _ = ts.AppendChunk(streaming.Chunk{Token: e.Text}) + sb.addText(e.Text) // spec-1.23 D-3: accumulate the visible final answer + return send(ctx, streamControlMsg{PreviewTick: true}) case diagnose.EventToolCall: + if err := seal(ctx, false); err != nil { + return err + } sb.addToolCall(e.ToolUse) // spec-1.23 D-3 return send(ctx, streamControlMsg{ToolUse: e.ToolUse}) case diagnose.EventToolResult: + if err := seal(ctx, false); err != nil { + return err + } sb.addToolResult(e.ToolResult, e.Cached) // spec-1.23 D-3 return send(ctx, streamControlMsg{ToolResult: e.ToolResult, Cached: e.Cached}) case diagnose.EventFinish: - // Push a terminal chunk so the TokenStream's per-finish - // branches (Length truncation marker / cancel state) keep - // working under the new emit path (spec-1.6 R2.2 / spec-1.20 - // thinking-only Empty placeholder). - _ = ts.AppendChunk(streaming.Chunk{ - FinishReason: mapToRender(e.Finish), - Err: e.Err, - }) + if err := seal(ctx, e.Finish == llm.FinishLength); err != nil { + return err + } // spec-1.23 D-3: seal the run snapshot atomically with the finish // and ride it on this same control msg — no separate post-Run send, // so no channel-close race (three-route T-2 收敛 fix). @@ -100,7 +118,8 @@ func makeEmit(ts *streaming.TokenStream, ctrl chan<- streamControlMsg, stripThin // goroutine drains the loop to terminal, then closes ctrl and ts so the // reader-Cmd observes a done signal (mirrors the spec-1.20 consumeStream // teardown order — close ts BEFORE ctrl is the convention so the -// streamDoneMsg drain sees a fully-flushed TokenStream). +// streamDoneMsg only performs cleanup; authoritative text has already +// travelled through ctrl as SealedText. func loopStartCmd( ctx context.Context, loop *diagnose.Loop, diff --git a/internal/app/cli/llmapp/model.go b/internal/app/cli/llmapp/model.go index a9031d1..22236c1 100644 --- a/internal/app/cli/llmapp/model.go +++ b/internal/app/cli/llmapp/model.go @@ -95,21 +95,15 @@ type Model struct { buffer string cursor int - history []llm.Message // bounded FIFO (R2 MED-4) - stream *streaming.TokenStream // in-flight render stream (nil = idle) - control chan streamControlMsg // Loop emit → Update (spec-1.21 D-6; legacy R2.2 shape preserved) - cancel context.CancelFunc // cancels in-flight stream (Cmd/Cleanup) - streaming bool - sawContent bool // any visible content this turn (R2 H-5 / R2.2 HIGH-2) - sawThinking bool // any thinking-channel token this turn (spec-1.20.2 D-5) - thinkingBuf string // accumulated thinking text when strip_think=false - - // assistantBulletPending marks that the current assistant turn has not yet - // emitted its ⏺ speaker bullet (spec-1.25 D-5 / CRIT-A). Set at submit; - // the first drained content Message of the turn consumes it (per-turn, not - // per-line). render/streaming stays neutral — the bullet is a post-drain - // render-only decoration applied here in llmapp. - assistantBulletPending bool + history []llm.Message // bounded FIFO (R2 MED-4) + stream *streaming.TokenStream // in-flight preview stream (nil = idle) + control chan streamControlMsg // Loop emit → Update (spec-1.21 D-6; legacy R2.2 shape preserved) + previewNodes []block.RenderNode // in-flight visible text preview; never committed to scrollback + cancel context.CancelFunc // cancels in-flight stream (Cmd/Cleanup) + streaming bool + sawContent bool // any visible content this turn (R2 H-5 / R2.2 HIGH-2) + sawThinking bool // any thinking-channel token this turn (spec-1.20.2 D-5) + thinkingBuf string // accumulated thinking text when strip_think=false // toolUseNames joins ToolResult.ToolUseID → ToolUse.Name within a // single submit (spec-1.21 D-6 / spec-1.9b R3 HIGH-3: rendering a @@ -117,7 +111,8 @@ type Model struct { // per CC null-return). Reset on every submit. toolUseNames map[string]string - scrollback []block.RenderNode + scrollback []block.RenderNode + sealedTruncated bool // current run sealed its final text with Message.Truncated // lastSnapshot is the most recent completed (non-cancelled) diagnosis run, // captured by the snapshotBuilder and sealed at EventFinish. /report reads @@ -228,30 +223,20 @@ func (m *Model) Update(msg scheduler.Msg) (program.Model, scheduler.Cmd) { return m.handleReportWriteFailed(v) case streamDoneMsg: next := *m - // T-10a HIGH-3: the TokenStream contract requires a Drain AFTER Close - // to collect final blocks (Close flushes the last partial into - // emitted but does not consume it). loopStartCmd closes the - // TokenStream before closing ctrl, so by now the final blocks are - // flushed — drain them once more before discarding the stream, - // else the last tokens - // are dropped. Update/View share the scheduler goroutine, so this - // Drain is race-free (same single-owner as the View Drain). + // spec-1.21.1: authoritative text is committed via SealedText + // control messages. A final Drain here is cleanup only, so a closed + // TokenStream cannot append stale or reordered blocks into scrollback. if next.stream != nil { - if nodes := next.stream.Drain(); len(nodes) > 0 { - nodes = filterThinkingOnlyEmpty(nodes, next.sawThinking, next.sawContent) - if len(nodes) > 0 { - nodes, next.assistantBulletPending = markAssistantBullet(nodes, next.assistantBulletPending) - next.scrollback = appendNodes(next.scrollback, nodes) - } - } + _ = next.stream.Drain() } + next.previewNodes = nil next.stream = nil next.control = nil next.cancel = nil next.streaming = false next.sawThinking = false next.thinkingBuf = "" - next.assistantBulletPending = false // turn over; do not leak bullet + next.sealedTruncated = false return &next, nil } return m, nil @@ -325,15 +310,16 @@ func (m *Model) submit() (program.Model, scheduler.Cmd) { next.sawContent = false next.sawThinking = false next.thinkingBuf = "" + next.previewNodes = nil + next.sealedTruncated = false next.history = appendBounded(m.history, llm.Message{ Role: llm.RoleUser, Content: []llm.ContentBlock{{Type: llm.BlockText, Text: userText}}, }, m.maxHistory) // spec-1.25 D-5: user echo is plain (no "> " prefix — CC parity), tagged - // SpeakerUser. Starting the assistant turn arms the ⏺ bullet for the first - // content node drained (CRIT-A per-turn). + // SpeakerUser. Assistant bullets are attached when each sealed text segment + // is committed through the ctrl FIFO (spec-1.21.1). next.scrollback = appendNode(m.scrollback, block.Message{Text: userText, Speaker: block.SpeakerUser}) - next.assistantBulletPending = true next.toolUseNames = map[string]string{} // reset per submit (spec-1.21 D-6) return &next, loopStartCmd(ctx, m.loop, req, userText, ts, ctrl, m.stripThink) @@ -367,8 +353,9 @@ func (m *Model) buildRequest(userText string) llm.Request { // name in next.toolUseNames for the matching ToolResult. // - ToolResult != nil → append block.ToolResult (joined name lookup; // empty name → skip render per CC null-return / spec-1.9b R3 HIGH-3). +// - PreviewTick → drain TokenStream into previewNodes only. +// - SealedText → append authoritative assistant Message segment. // - Finish terminal → appendFinishNode (TermCode-aware DIAGNOSE.*). -// - otherwise → text variant; accumulate sawContent only. // // Every branch re-arms the reader so the next ctrl msg (or done signal) // is pulled — including the terminal finish, which depends on the close @@ -376,24 +363,36 @@ func (m *Model) buildRequest(userText string) llm.Request { func (m *Model) handleControl(msg streamControlMsg) (program.Model, scheduler.Cmd) { next := *m switch { + case msg.PreviewTick: + if next.stream != nil { + if nodes := next.stream.Drain(); len(nodes) > 0 { + nodes = filterThinkingOnlyEmpty(nodes, next.sawThinking, next.sawContent) + if len(nodes) > 0 { + next.previewNodes = appendNodes(next.previewNodes, nodes) + } + } + } + case msg.SealedText != "": + next.sawContent = true + next.sealedTruncated = msg.SealedTruncated + next.scrollback = appendNode(m.scrollback, block.Message{ + Text: msg.SealedText, + Truncated: msg.SealedTruncated, + Speaker: block.SpeakerAssistant, + }) + // The matching TokenStream content was preview-only. Discard any + // residual blocks and clear cached preview so the sealed segment is not + // rendered twice. + if next.stream != nil { + _ = next.stream.Drain() + } + next.previewNodes = nil case msg.Thinking: next.sawThinking = true if msg.ThinkingToken != "" { next.thinkingBuf += msg.ThinkingToken } case msg.ToolUse != nil: - // spec-1.25 D-5 / codex D5-HIGH-1 (DEFERRED): draining pending stream - // text here to order assistant prose before the tool node is NOT done — - // the TokenStream is shared across all loop turns and the loop goroutine - // can race ahead, so a boundary drain pulls FUTURE-turn text before this - // tool node (proven by TestModel_LoopAppendsToolBlocks failing under CI - // timing). A correct fix needs producer-side ordering (per-turn stream - // boundary or text-via-ordered-channel) — tracked as a spec-1.21 - // follow-up. The ⏺ bullet still attaches correctly via markAssistantBullet - // at the View/streamDoneMsg drain. - // - // Mutate the per-submit map in place — next is already a shallow - // copy and toolUseNames is owned by this in-flight submit. if next.toolUseNames == nil { next.toolUseNames = map[string]string{} } @@ -409,12 +408,6 @@ func (m *Model) handleControl(msg streamControlMsg) (program.Model, scheduler.Cm // (spec-1.9b R3 HIGH-3). break } - // codex T-10a P2-1 absorb: transition the matching block.ToolUse - // from StateRunning to StateResolved/StateError so the UI does - // not show the tool as "Running" forever once its result has - // arrived. Spec-1.9 toolcall.go:62-64 contract: caller owns the - // transition. spec-1.21 D-6 specifies this exact hand-off: - // "EventToolResult → 转 Resolved/Error". targetState := block.StateResolved if msg.ToolResult.IsError { targetState = block.StateError @@ -426,16 +419,18 @@ func (m *Model) handleControl(msg streamControlMsg) (program.Model, scheduler.Cm tr.Cached = msg.Cached next.scrollback = appendNode(sb, tr) case msg.Finish.Terminal(): - next.sawContent = m.sawContent || msg.VisibleContent + next.sawContent = m.sawContent next.scrollback = next.appendFinishNode(msg) + if next.stream != nil { + _ = next.stream.Drain() + } + next.previewNodes = nil // spec-1.23 D-3: adopt the sealed run snapshot for /report. A cancelled // run does NOT overwrite a prior good snapshot ("last completed", not // "last attempted" — architect L-2). if msg.Snapshot != nil && msg.Finish != llm.FinishCancelled { next.lastSnapshot = msg.Snapshot } - default: - next.sawContent = m.sawContent || msg.VisibleContent } return &next, m.armReader() } @@ -502,7 +497,12 @@ func (m *Model) appendFinishNode(msg streamControlMsg) []block.RenderNode { // thinking-only truncation reports empty, not "[截断]". return appendNode(base, block.Message{Text: "[LLM.STREAM_EMPTY: 无可见输出]"}) case msg.Finish == llm.FinishLength: - // Had visible content but hit the token cap. + // Had visible content but hit the token cap. If the final visible + // segment was sealed with Message.Truncated, the inline marker already + // carries the truncation signal; otherwise add an explicit terminal node. + if m.sealedTruncated { + return base + } return appendNode(base, block.Message{Text: "[截断: 达到 max_tokens]"}) } return base @@ -524,26 +524,25 @@ func (m *Model) handleCancel() (program.Model, scheduler.Cmd) { } } -// View renders scrollback + the in-flight stream. Per spec-1.6 main-loop -// bridge (Q11 Drain-in-View): Drain the active stream into scrollback each -// frame (View runs on the scheduler goroutine, single-owner of m). +// View renders committed scrollback plus the in-flight preview. The +// TokenStream is NOT drained here: spec-1.21.1 makes handleControl the +// sole owner of permanent scrollback writes, while PreviewTick updates +// previewNodes inside Update. This keeps View read-only and removes the +// old View-vs-control ordering race. func (m *Model) View(cols, rows int) buffer.Buffer { - if m.stream != nil { - if nodes := m.stream.Drain(); len(nodes) > 0 { - nodes = filterThinkingOnlyEmpty(nodes, m.sawThinking, m.sawContent) - nodes, m.assistantBulletPending = markAssistantBullet(nodes, m.assistantBulletPending) - m.scrollback = append(m.scrollback, nodes...) - } - } g, err := buffer.NewGrid(cols, rows) if err != nil { return nil } + nodes := m.scrollback + if len(m.previewNodes) > 0 { + nodes = appendNodes(m.scrollback, m.previewNodes) + } // Paint the most recent nodes bottom-up. ctx := block.Context{Cols: cols, Rows: rows} y := rows - 1 - for i := len(m.scrollback) - 1; i >= 0 && y >= 0; i-- { - nb, rerr := m.scrollback[i].Render(ctx) + for i := len(nodes) - 1; i >= 0 && y >= 0; i-- { + nb, rerr := nodes[i].Render(ctx) if rerr != nil || nb == nil { continue } @@ -627,30 +626,6 @@ func appendNodes(sb []block.RenderNode, nodes []block.RenderNode) []block.Render return next } -// markAssistantBullet tags the FIRST content Message in nodes with -// SpeakerAssistant when a bullet is pending, returning the (possibly -// modified) nodes and the still-pending flag (spec-1.25 D-5 / CRIT-A). -// Per-turn: only the first assistant content node of a turn carries the ⏺ -// bullet (CC AssistantTextMessage is per-message, not per-line). When no -// content Message is found (e.g. thinking-only / tool-only drain), the -// bullet stays pending for a later drain in the same turn. nodes are -// replaced by value (block.Message is a value type), not mutated in place. -func markAssistantBullet(nodes []block.RenderNode, pending bool) ([]block.RenderNode, bool) { - if !pending { - return nodes, false - } - for i, n := range nodes { - msg, ok := n.(block.Message) - if !ok || msg.Text == "" || msg.Empty { - continue - } - msg.Speaker = block.SpeakerAssistant - nodes[i] = msg - return nodes, false // consumed - } - return nodes, true // still pending -} - func filterThinkingOnlyEmpty(nodes []block.RenderNode, sawThinking, sawContent bool) []block.RenderNode { if !sawThinking || sawContent || len(nodes) == 0 { return nodes diff --git a/internal/app/cli/llmapp/model_test.go b/internal/app/cli/llmapp/model_test.go index f740cef..73fd108 100644 --- a/internal/app/cli/llmapp/model_test.go +++ b/internal/app/cli/llmapp/model_test.go @@ -6,6 +6,7 @@ package llmapp import ( "context" + "errors" "strings" "testing" "time" @@ -77,7 +78,7 @@ func TestModel_SubmitClearsBuffer(t *testing.T) { if final.streaming { t.Errorf("still streaming after drain; want idle") } - // scrollback has the user echo + assistant text drained. + // scrollback has the user echo + sealed assistant text. if len(final.scrollback) == 0 { t.Errorf("scrollback empty after stream") } @@ -124,8 +125,8 @@ func TestModel_SawContent_TextThenLength(t *testing.T) { if hasNode(final, "STREAM_EMPTY") { t.Errorf("text+Length wrongly produced STREAM_EMPTY") } - if !hasNode(final, "截断") { - t.Errorf("text+Length should show truncation marker") + if !hasTruncatedMessage(final) { + t.Errorf("text+Length should mark the sealed assistant segment truncated") } } @@ -378,6 +379,98 @@ func TestModel_LoopAppendsToolBlocks(t *testing.T) { } } +func TestModel_LoopTextToolTextOrderAndBullets(t *testing.T) { + t.Parallel() + reg, err := diagnose.NewRegistry(diagnose.EchoTool{}) + if err != nil { + t.Fatalf("registry: %v", err) + } + prov := fake.NewScriptedTurns( + fake.Turn{ + Text: "before tool", + ToolUses: []llm.ToolUse{{ID: "c1", Name: "echo", Input: map[string]any{"msg": "ping"}}}, + Finish: llm.FinishToolUse, + }, + fake.Turn{Text: "after tool", Finish: llm.FinishStop}, + ) + m := New(prov, Options{ModelName: "loop", MaxTokens: 1024, Registry: reg}) + m = typeAndModel(t, m, "go") + final := runStream(t, m) + + want := []string{"message:go", "assistant:before tool", "tooluse", "toolresult", "assistant:after tool"} + if got := scrollbackShape(final); strings.Join(got, "|") != strings.Join(want, "|") { + t.Fatalf("scrollback order = %v; want %v", got, want) + } +} + +func TestModel_PreviewTickViewBetweenDoesNotCommit(t *testing.T) { + t.Parallel() + m := New(fake.New(), Options{ModelName: "fake"}) + m.stream = streaming.NewTokenStream(context.Background()) + if err := m.stream.AppendChunk(streaming.Chunk{Token: "preview\n"}); err != nil { + t.Fatalf("AppendChunk: %v", err) + } + + mm, _ := m.Update(streamControlMsg{PreviewTick: true}) + withPreview := mm.(*Model) + if len(withPreview.previewNodes) == 0 { + t.Fatalf("PreviewTick did not cache preview nodes") + } + if len(withPreview.scrollback) != 0 { + t.Fatalf("PreviewTick must not commit to scrollback; got %d nodes", len(withPreview.scrollback)) + } + if txt := gridText(withPreview.View(80, 6)); !strings.Contains(txt, "preview") { + t.Fatalf("View did not render cached preview: %q", txt) + } + if len(withPreview.scrollback) != 0 { + t.Fatalf("View must be read-only; scrollback mutated to %d nodes", len(withPreview.scrollback)) + } + + mm, _ = withPreview.Update(streamControlMsg{SealedText: "preview\n"}) + sealed := mm.(*Model) + if len(sealed.previewNodes) != 0 { + t.Fatalf("SealedText should clear preview nodes; got %d", len(sealed.previewNodes)) + } + if got := scrollbackShape(sealed); strings.Join(got, "|") != "assistant:preview\n" { + t.Fatalf("sealed scrollback = %v", got) + } +} + +func TestModel_ClosedBeforeControlsKeepsSealedOrder(t *testing.T) { + t.Parallel() + m := New(fake.New(), Options{ModelName: "fake"}) + m.stream = streaming.NewTokenStream(context.Background()) + if err := m.stream.AppendChunk(streaming.Chunk{Token: "stale-from-stream\n"}); err != nil { + t.Fatalf("AppendChunk: %v", err) + } + if err := m.stream.Close(); err != nil && !errors.Is(err, streaming.ErrStreamCancelled) { + t.Fatalf("Close: %v", err) + } + + controls := []streamControlMsg{ + {SealedText: "before"}, + {ToolUse: &llm.ToolUse{ID: "c1", Name: "echo"}}, + {ToolResult: &llm.ToolResult{ToolUseID: "c1", Content: "ok"}}, + {SealedText: "after"}, + {Finish: llm.FinishStop}, + } + cur := program.Model(m) + for _, c := range controls { + next, _ := cur.Update(c) + cur = next + } + next, _ := cur.Update(streamDoneMsg{}) + final := next.(*Model) + + want := []string{"assistant:before", "tooluse", "toolresult", "assistant:after"} + if got := scrollbackShape(final); strings.Join(got, "|") != strings.Join(want, "|") { + t.Fatalf("scrollback order = %v; want %v", got, want) + } + if hasNode(final, "stale-from-stream") { + t.Fatalf("closed TokenStream content leaked into authoritative scrollback: %v", nodeTexts(final)) + } +} + func TestModel_StatusSegments(t *testing.T) { t.Parallel() m := newFakeModel(fake.New()) @@ -400,16 +493,15 @@ func TestModel_HistoryBounded(t *testing.T) { } } -// TestModel_AssistantTextDrained is the T-10a HIGH-3 regression: the -// assistant text must reach scrollback even though the final Drain happens -// only after the stream is Closed. runStream never calls View, so this -// fails if the streamDoneMsg handler does not Drain before nil-ing stream. -func TestModel_AssistantTextDrained(t *testing.T) { +// TestModel_AssistantTextCommitted is the spec-1.21.1 regression: assistant +// text must reach scrollback through SealedText control messages even when +// runStream never calls View. +func TestModel_AssistantTextCommitted(t *testing.T) { t.Parallel() m := typeAndModel(t, newFakeModel(fake.Scripted("hello world", llm.FinishStop)), "q") final := runStream(t, m) if !hasNode(final, "hello world") { - t.Errorf("assistant text not drained into scrollback (HIGH-3); got %v", nodeTexts(final)) + t.Errorf("assistant text not committed into scrollback; got %v", nodeTexts(final)) } } @@ -551,6 +643,36 @@ func messageNodeContains(m *Model, substr string) bool { return false } +func hasTruncatedMessage(m *Model) bool { + for _, n := range m.scrollback { + if msg, ok := n.(block.Message); ok && msg.Truncated { + return true + } + } + return false +} + +func scrollbackShape(m *Model) []string { + out := make([]string, 0, len(m.scrollback)) + for _, n := range m.scrollback { + switch v := n.(type) { + case block.Message: + if v.Speaker == block.SpeakerAssistant { + out = append(out, "assistant:"+v.Text) + } else { + out = append(out, "message:"+v.Text) + } + case block.ToolUse: + out = append(out, "tooluse") + case block.ToolResult: + out = append(out, "toolresult") + case block.Thinking: + out = append(out, "thinking") + } + } + return out +} + func blockCtx() block.Context { return block.Context{Cols: 200, Rows: 10} } func gridText(b buffer.Buffer) string { diff --git a/internal/app/cli/llmapp/stream.go b/internal/app/cli/llmapp/stream.go index b970bbf..64d490b 100644 --- a/internal/app/cli/llmapp/stream.go +++ b/internal/app/cli/llmapp/stream.go @@ -5,40 +5,42 @@ package llmapp import ( - "github.com/sqlrush/opendbx/internal/app/cli/render/streaming" "github.com/sqlrush/opendbx/internal/app/report" "github.com/sqlrush/opendbx/internal/domain/llm" ) // streamControlMsg is the control-bypass Msg (spec-1.20 R2 CRIT-2 base; -// spec-1.21 D-6 extension). Routing is by which optional field is set: +// spec-1.21.1 sealed-segment extension). Routing is by which optional +// field is set: // -// - ToolUse != nil → EventToolCall variant -// - ToolResult != nil → EventToolResult variant -// - Finish.Terminal() → EventFinish variant (TermCode optional DIAGNOSE.*) +// - PreviewTick → EventText visible preview wakeup; Update drains +// TokenStream into Model.previewNodes. It carries no text payload and +// does not affect sawContent / finish semantics. +// - SealedText != "" → authoritative assistant text segment; appended +// to scrollback through the same ctrl FIFO as tool events. +// - ToolUse != nil → EventToolCall variant +// - ToolResult != nil → EventToolResult variant +// - Finish.Terminal() → EventFinish variant (TermCode optional DIAGNOSE.*) // - Thinking=true → EventText thinking-channel variant; ThinkingToken // carries renderable thinking only when strip_think=false. Thinking never // enters the TokenStream / main content plane. -// - default (all nil) → EventText visible-text variant; VisibleContent -// tells Update to accumulate sawContent without inspecting TokenStream -// (otherwise text + FinishLength would be misjudged !sawContent → false -// STREAM_EMPTY). // -// At most one variant is populated per message; mixed shapes are not -// produced by makeEmit. Update dispatches on the variants in priority -// (ToolUse / ToolResult before Finish before text) so a future spec -// adding fields keeps the routing explicit. +// At most one variant is populated per message, except SealedTruncated which +// qualifies SealedText. Update dispatches variants explicitly so future +// additive fields cannot silently fall through. type streamControlMsg struct { - VisibleContent bool - Thinking bool - ThinkingToken string // EventText thinking side channel (spec-1.20.2 D-5) - ToolUse *llm.ToolUse // EventToolCall (spec-1.21 D-6) - ToolResult *llm.ToolResult // EventToolResult (spec-1.21 D-6) - Cached bool // EventToolResult — spec-1.22: result served from dedup cache (render-only; ToolResult content is byte-identical to a fresh run) - Finish llm.FinishReason - TermCode string // EventFinish — DIAGNOSE.* code or "" on natural Stop - Err error - Snapshot *report.RunSnapshot // EventFinish — spec-1.23 D-3: the run snapshot, sealed atomically with the finish (rides this msg → no post-Run send → no channel-close race) + PreviewTick bool + SealedText string + SealedTruncated bool + Thinking bool + ThinkingToken string // EventText thinking side channel (spec-1.20.2 D-5) + ToolUse *llm.ToolUse // EventToolCall (spec-1.21 D-6) + ToolResult *llm.ToolResult // EventToolResult (spec-1.21 D-6) + Cached bool // EventToolResult — spec-1.22: result served from dedup cache (render-only; ToolResult content is byte-identical to a fresh run) + Finish llm.FinishReason + TermCode string // EventFinish — DIAGNOSE.* code or "" on natural Stop + Err error + Snapshot *report.RunSnapshot // EventFinish — spec-1.23 D-3: the run snapshot, sealed atomically with the finish (rides this msg → no post-Run send → no channel-close race) } // readControlMsg asks Update to arm readControlCmd (pull one control msg). @@ -46,24 +48,3 @@ type readControlMsg struct{} // streamDoneMsg signals the ctrl channel was closed (all control drained). type streamDoneMsg struct{} - -// mapToRender maps an llm.FinishReason to the renderable streaming subset -// (spec-1.20 R2 H-2). Non-renderable terminal reasons (ToolUse / Pause / -// Refusal / StopSequence) map to streaming.FinishStop; the precise reason -// travels via the control channel. T1 exhaustive test covers all 9. -func mapToRender(f llm.FinishReason) streaming.FinishReason { - switch f { - case llm.FinishUnset: - return streaming.FinishUnset - case llm.FinishLength: - return streaming.FinishLength - case llm.FinishError: - return streaming.FinishError - case llm.FinishCancelled: - return streaming.FinishCancelled - default: - // Stop / ToolUse / StopSequence / Pause / Refusal → clean stop for - // the render stream; control channel carries the precise reason. - return streaming.FinishStop - } -} diff --git a/internal/app/cli/llmapp/stream_test.go b/internal/app/cli/llmapp/stream_test.go deleted file mode 100644 index 055413f..0000000 --- a/internal/app/cli/llmapp/stream_test.go +++ /dev/null @@ -1,48 +0,0 @@ -// Copyright 2026 opendbx contributors. See LICENSE. -// -// Author: sqlrush - -package llmapp - -import ( - "testing" - - "github.com/sqlrush/opendbx/internal/app/cli/render/streaming" - "github.com/sqlrush/opendbx/internal/domain/llm" -) - -func TestMapToRender_AllNine(t *testing.T) { - t.Parallel() - cases := []struct { - in llm.FinishReason - want streaming.FinishReason - }{ - {llm.FinishUnset, streaming.FinishUnset}, - {llm.FinishStop, streaming.FinishStop}, - {llm.FinishLength, streaming.FinishLength}, - {llm.FinishToolUse, streaming.FinishStop}, - {llm.FinishStopSequence, streaming.FinishStop}, - {llm.FinishPause, streaming.FinishStop}, - {llm.FinishRefusal, streaming.FinishStop}, - {llm.FinishCancelled, streaming.FinishCancelled}, - {llm.FinishError, streaming.FinishError}, - } - for _, tc := range cases { - if got := mapToRender(tc.in); got != tc.want { - t.Errorf("mapToRender(%v) = %v; want %v", tc.in, got, tc.want) - } - } -} - -// finishFromErr was retired in spec-1.21 T-8: the diagnose.Loop owns -// terminal classification end-to-end (classifyTerminal / classifyEmitErr -// / classifyToolErr) and surfaces the already-classified -// FinishReason+Err on EventFinish. The llmapp adapter never re-classifies -// — it just renders. So no TestFinishFromErr / BenchmarkFinishFromErr. - -// BenchmarkMapToRender targets spec-1.20 § 4.4 (pure map; well under 500 ns/op). -func BenchmarkMapToRender(b *testing.B) { - for i := 0; i < b.N; i++ { - _ = mapToRender(llm.FinishLength) - } -} diff --git a/internal/app/cli/llmapp/view_test.go b/internal/app/cli/llmapp/view_test.go index f9acb23..0dbcee6 100644 --- a/internal/app/cli/llmapp/view_test.go +++ b/internal/app/cli/llmapp/view_test.go @@ -45,15 +45,15 @@ func TestModel_View_PreservesWideRunes(t *testing.T) { } } -func TestModel_View_DrainsInFlight(t *testing.T) { +func TestModel_View_InFlightPreviewNoPanic(t *testing.T) { t.Parallel() // Submit with a no-finish chunk so a stream is in flight; View should - // Drain it without panic. + // render without mutating or draining preview state. m := typeAndModel(t, newFakeModel(fake.New(llm.Chunk{Token: "x"})), "q") mm, cmd := m.Update(keyAction(terminal.KeyEnter, 0)) _ = cmd streamingModel := mm.(*Model) - _ = streamingModel.View(80, 10) // exercises the Drain-in-View path + _ = streamingModel.View(80, 10) } func TestModel_View_SmallGrid(t *testing.T) { From e7fb4598a9fb76ae768b57f2819758a2c609f1b5 Mon Sep 17 00:00:00 2001 From: sqlrush Date: Wed, 3 Jun 2026 17:10:23 +0800 Subject: [PATCH 2/3] chore(security): allowlist new stdlib vulns --- tools/vuln-allowlist/allowlist.json | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/tools/vuln-allowlist/allowlist.json b/tools/vuln-allowlist/allowlist.json index 29162e2..041ff6a 100644 --- a/tools/vuln-allowlist/allowlist.json +++ b/tools/vuln-allowlist/allowlist.json @@ -139,6 +139,20 @@ "expiry": "2026-08-14", "reason": "Go stdlib vuln advisory landed 2026-05-15 (post spec-0.11 FROZEN). opendbx Stage 0 locks Go 1.23 per spec-0.9 R2. Triage and patch-bump at Stage 1 toolchain review; reachability not verified for this code path. Batch exemption window matches GO-2026-4602.", "spec_ref": "spec-0.9-ci-github-actions.md § 3.1 T-3.5 (batch 2026-05-15)" + }, + { + "osv_id": "GO-2026-5037", + "module": "stdlib", + "expiry": "2026-08-14", + "reason": "Go stdlib vuln advisory surfaced in govulncheck on 2026-06-03 against the project-pinned Go 1.24 toolchain. This PR does not change stdlib/toolchain use; triage and patch-bump at the next toolchain review. Reachability not verified for this code path. Batch exemption window matches GO-2026-4602.", + "spec_ref": "spec-0.9-ci-github-actions.md § 3.1 T-3.5 (batch 2026-06-03; PR #91 security job)" + }, + { + "osv_id": "GO-2026-5039", + "module": "stdlib", + "expiry": "2026-08-14", + "reason": "Go stdlib vuln advisory surfaced in govulncheck on 2026-06-03 against the project-pinned Go 1.24 toolchain. This PR does not change stdlib/toolchain use; triage and patch-bump at the next toolchain review. Reachability not verified for this code path. Batch exemption window matches GO-2026-4602.", + "spec_ref": "spec-0.9-ci-github-actions.md § 3.1 T-3.5 (batch 2026-06-03; PR #91 security job)" } ] } From 2fe97d20bb8ff379e1b5907d1ead78cef6b99619 Mon Sep 17 00:00:00 2001 From: sqlrush Date: Wed, 3 Jun 2026 20:14:28 +0800 Subject: [PATCH 3/3] fix(llmapp): spec-1.21.1 post-impl R-fix (cancel-path H-1/H-2 + segBuf) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 三路 post-impl review absorb(code-reviewer APPROVE + go-reviewer APPROVE-WITH-FIXES + codex REQUEST-CHANGES 2H)。codex H-1(regression, 按用户 verdict 标准 = HIGH):cancel-mid-segment 丢 partial assistant 文本 + [已取消] marker —— Loop emit-error break 不发 EventFinish,partial 在 previewNodes(PreviewTick 缓存),streamDoneMsg 原 drain-discard 丢弃。修: streamDoneMsg final-flush 从 previewNodes(+ Close 残留)恢复 partial 进 永久 scrollback(markFirstAssistant 带 ⏺)+ finishHandled flag 缺 finish 时 补 [已取消](恢复 pre-1.21.1 final-drain-preserves-text)。codex H-2:统一 isCancelledFinish predicate(FinishCancelled OR FinishError+context.Canceled) 给 marker render + snapshot adoption,防 err-cancel 覆盖 lastSnapshot。 go-reviewer MED-1:segBuf strings.Builder(O(n²)→O(n))。+ 回归测 CancelMidSegmentRecoversPartial / NormalFinishNoSpuriousCancel / H-2 snapshot。 gate 全绿 + -race -count=10。 Spec: spec-1.21.1-stream-ordering-fix.md --- internal/app/cli/llmapp/loop_adapter.go | 18 +++--- internal/app/cli/llmapp/model.go | 79 +++++++++++++++++++++--- internal/app/cli/llmapp/model_test.go | 67 ++++++++++++++++++++ internal/app/cli/llmapp/snapshot_test.go | 17 +++++ 4 files changed, 165 insertions(+), 16 deletions(-) diff --git a/internal/app/cli/llmapp/loop_adapter.go b/internal/app/cli/llmapp/loop_adapter.go index fccd7cc..5133301 100644 --- a/internal/app/cli/llmapp/loop_adapter.go +++ b/internal/app/cli/llmapp/loop_adapter.go @@ -23,6 +23,7 @@ package llmapp import ( "context" + "strings" "time" "github.com/sqlrush/opendbx/internal/app/cli/render/scheduler" @@ -53,13 +54,13 @@ func makeEmit(ts *streaming.TokenStream, ctrl chan<- streamControlMsg, stripThin // diagnose.Loop producer goroutine in event order, so segBuf is // intentionally goroutine-local and unsynchronized. Do not reuse this // closure from multiple producers without re-specifying ordering. - var segBuf string + var segBuf strings.Builder // spec-1.21.1 R-fix MED-1: O(n) accumulation (was string concat) seal := func(ctx context.Context, truncated bool) error { - if segBuf == "" { + if segBuf.Len() == 0 { return nil } - text := segBuf - segBuf = "" + text := segBuf.String() + segBuf.Reset() return send(ctx, streamControlMsg{SealedText: text, SealedTruncated: truncated}) } @@ -76,7 +77,7 @@ func makeEmit(ts *streaming.TokenStream, ctrl chan<- streamControlMsg, stripThin if e.Text == "" { return nil } - segBuf += e.Text + segBuf.WriteString(e.Text) _ = ts.AppendChunk(streaming.Chunk{Token: e.Text}) sb.addText(e.Text) // spec-1.23 D-3: accumulate the visible final answer return send(ctx, streamControlMsg{PreviewTick: true}) @@ -117,9 +118,10 @@ func makeEmit(ts *streaming.TokenStream, ctrl chan<- streamControlMsg, stripThin // loopStartCmd launches a diagnose.Loop.Run on the worker pool. The // goroutine drains the loop to terminal, then closes ctrl and ts so the // reader-Cmd observes a done signal (mirrors the spec-1.20 consumeStream -// teardown order — close ts BEFORE ctrl is the convention so the -// streamDoneMsg only performs cleanup; authoritative text has already -// travelled through ctrl as SealedText. +// teardown order — close ts BEFORE ctrl is the convention). On the normal +// path authoritative text has already travelled through ctrl as SealedText, so +// streamDoneMsg only performs cleanup; on a cancel mid-segment streamDoneMsg +// recovers the residual partial text from previewNodes (spec-1.21.1 R-fix H-1). func loopStartCmd( ctx context.Context, loop *diagnose.Loop, diff --git a/internal/app/cli/llmapp/model.go b/internal/app/cli/llmapp/model.go index 22236c1..9f5650d 100644 --- a/internal/app/cli/llmapp/model.go +++ b/internal/app/cli/llmapp/model.go @@ -104,6 +104,12 @@ type Model struct { sawContent bool // any visible content this turn (R2 H-5 / R2.2 HIGH-2) sawThinking bool // any thinking-channel token this turn (spec-1.20.2 D-5) thinkingBuf string // accumulated thinking text when strip_think=false + // finishHandled marks that a terminal Finish control was processed this + // turn (spec-1.21.1 R-fix H-1). When false at streamDoneMsg, the stream + // ended abruptly (user cancel mid-segment: diagnose Loop breaks on emit + // error without emitting EventFinish) — streamDoneMsg then recovers the + // residual partial text + annotates the cancel. + finishHandled bool // toolUseNames joins ToolResult.ToolUseID → ToolUse.Name within a // single submit (spec-1.21 D-6 / spec-1.9b R3 HIGH-3: rendering a @@ -223,11 +229,33 @@ func (m *Model) Update(msg scheduler.Msg) (program.Model, scheduler.Cmd) { return m.handleReportWriteFailed(v) case streamDoneMsg: next := *m - // spec-1.21.1: authoritative text is committed via SealedText - // control messages. A final Drain here is cleanup only, so a closed - // TokenStream cannot append stale or reordered blocks into scrollback. + // spec-1.21.1 R-fix H-1 — final flush. On the normal path the + // authoritative text was committed via SealedText (which cleared + // previewNodes + drain-discarded the TokenStream), so both are empty here. + // On a cancel/error path where the seal + EventFinish were lost to ctx + // cancellation (diagnose loop.go breaks on emit error WITHOUT emitting + // EventFinish), the partial assistant text survives in previewNodes (the + // PreviewTick-accumulated cache) + a possible final partial line flushed + // by TokenStream.Close — commit it so it is not lost (restores the + // pre-1.21.1 final-drain-preserves-text behavior; codex H-1). The + // recovered segment keeps its ⏺ bullet. + var residual []block.RenderNode + residual = append(residual, next.previewNodes...) if next.stream != nil { - _ = next.stream.Drain() + residual = append(residual, next.stream.Drain()...) + } + residual = filterThinkingOnlyEmpty(residual, next.sawThinking, next.sawContent) + if len(residual) > 0 { + markFirstAssistant(residual) + next.scrollback = appendNodes(next.scrollback, residual) + next.sawContent = true + } + if !next.finishHandled { + // No terminal Finish was processed → the stream ended abruptly + // (user cancel mid-segment). Annotate so the screen is not silently + // truncated (the [已取消] marker would otherwise be lost with the + // dropped EventFinish; codex H-1). + next.scrollback = appendNode(next.scrollback, block.Message{Text: "[已取消]"}) } next.previewNodes = nil next.stream = nil @@ -237,6 +265,7 @@ func (m *Model) Update(msg scheduler.Msg) (program.Model, scheduler.Cmd) { next.sawThinking = false next.thinkingBuf = "" next.sealedTruncated = false + next.finishHandled = false return &next, nil } return m, nil @@ -312,6 +341,7 @@ func (m *Model) submit() (program.Model, scheduler.Cmd) { next.thinkingBuf = "" next.previewNodes = nil next.sealedTruncated = false + next.finishHandled = false // spec-1.21.1 R-fix H-1 next.history = appendBounded(m.history, llm.Message{ Role: llm.RoleUser, Content: []llm.ContentBlock{{Type: llm.BlockText, Text: userText}}, @@ -420,6 +450,7 @@ func (m *Model) handleControl(msg streamControlMsg) (program.Model, scheduler.Cm next.scrollback = appendNode(sb, tr) case msg.Finish.Terminal(): next.sawContent = m.sawContent + next.finishHandled = true // spec-1.21.1 R-fix H-1 next.scrollback = next.appendFinishNode(msg) if next.stream != nil { _ = next.stream.Drain() @@ -427,8 +458,11 @@ func (m *Model) handleControl(msg streamControlMsg) (program.Model, scheduler.Cm next.previewNodes = nil // spec-1.23 D-3: adopt the sealed run snapshot for /report. A cancelled // run does NOT overwrite a prior good snapshot ("last completed", not - // "last attempted" — architect L-2). - if msg.Snapshot != nil && msg.Finish != llm.FinishCancelled { + // "last attempted" — architect L-2). spec-1.21.1 R-fix H-2: use the + // SAME cancel predicate as the marker render (appendFinishNode) so a + // FinishError+context.Canceled shape — which renders [已取消] — also does + // not overwrite the snapshot (previously only FinishCancelled was excluded). + if msg.Snapshot != nil && !isCancelledFinish(msg) { next.lastSnapshot = msg.Snapshot } } @@ -440,6 +474,34 @@ func (m *Model) armReader() scheduler.Cmd { return func() scheduler.Msg { return readControlMsg{} } } +// isCancelledFinish reports whether a finish control message represents a +// user/context cancellation. Both a literal FinishCancelled AND a +// FinishError wrapping context.Canceled count (some providers surface a +// cancel as the latter). spec-1.21.1 R-fix H-2: the marker render +// (appendFinishNode) and the /report snapshot-adoption guard MUST use this +// same predicate, else a FinishError+context.Canceled renders [已取消] yet +// still overwrites lastSnapshot, breaking "last completed not last attempted". +func isCancelledFinish(msg streamControlMsg) bool { + return msg.Finish == llm.FinishCancelled || + (msg.Finish == llm.FinishError && errors.Is(msg.Err, context.Canceled)) +} + +// markFirstAssistant tags the first visible-content Message in nodes with +// SpeakerAssistant so a residual (un-sealed) text segment recovered at +// streamDoneMsg still carries the ⏺ bullet on its first line. spec-1.21.1 +// R-fix H-1. Replaces the element by value (Message is a value type). +func markFirstAssistant(nodes []block.RenderNode) { + for i, n := range nodes { + msg, ok := n.(block.Message) + if !ok || msg.Text == "" || msg.Empty { + continue + } + msg.Speaker = block.SpeakerAssistant + nodes[i] = msg + return + } +} + // appendFinishNode appends a terminal status/marker node for the finish // reason. sawContent is already accumulated on next before this is // called (R2 H-5). spec-1.21 D-6 extension: DIAGNOSE.* TermCodes @@ -476,11 +538,12 @@ func (m *Model) appendFinishNode(msg streamControlMsg) []block.RenderNode { // for a genuinely empty non-Stop end (R2.2 case-order fix — the // !sawContent fallback must not shadow Error/Length). switch { - case msg.Finish == llm.FinishCancelled || (msg.Finish == llm.FinishError && errors.Is(msg.Err, context.Canceled)): + case isCancelledFinish(msg): // T-10a MED: a deliberate user cancel gets its own marker so it is // not silently blank nor mislabeled STREAM_EMPTY (checked before the // !sawContent fallback). msg.Err.Error() is NOT rendered here — the - // cancel is expected, not an error to surface. + // cancel is expected, not an error to surface. spec-1.21.1 R-fix H-2: + // shared predicate with snapshot adoption. return appendNode(base, block.Message{Text: "[已取消]"}) case msg.Finish == llm.FinishError && msg.Err != nil: // NB: the SDK's apierror.Error() formats METHOD/URL/STATUS/body only — diff --git a/internal/app/cli/llmapp/model_test.go b/internal/app/cli/llmapp/model_test.go index 73fd108..011c209 100644 --- a/internal/app/cli/llmapp/model_test.go +++ b/internal/app/cli/llmapp/model_test.go @@ -471,6 +471,73 @@ func TestModel_ClosedBeforeControlsKeepsSealedOrder(t *testing.T) { } } +// TestModel_CancelMidSegmentRecoversPartial — spec-1.21.1 R-fix H-1: on a user +// cancel mid-segment the diagnose Loop breaks on emit error WITHOUT emitting +// EventFinish, so the seal + [已取消] controls are lost. The partial assistant +// text survives in previewNodes (PreviewTick cache); streamDoneMsg must recover +// it into permanent scrollback (not lose it) + annotate the cancel. +func TestModel_CancelMidSegmentRecoversPartial(t *testing.T) { + t.Parallel() + m := New(fake.New(), Options{ModelName: "fake"}) + m.stream = streaming.NewTokenStream(context.Background()) + if err := m.stream.AppendChunk(streaming.Chunk{Token: "partial answer\n"}); err != nil { + t.Fatalf("AppendChunk: %v", err) + } + // PreviewTick caches the partial into previewNodes (TokenStream now drained). + mm, _ := m.Update(streamControlMsg{PreviewTick: true}) + mid := mm.(*Model) + if len(mid.previewNodes) == 0 { + t.Fatalf("PreviewTick should cache the partial") + } + // Abrupt end with NO terminal Finish (cancel mid-segment). + mm, _ = mid.Update(streamDoneMsg{}) + final := mm.(*Model) + + if !hasNode(final, "partial answer") { + t.Errorf("cancel-mid-segment LOST partial assistant text: %v", nodeTexts(final)) + } + if !hasNode(final, "[已取消]") { + t.Errorf("cancel without Finish should annotate [已取消]: %v", nodeTexts(final)) + } + if len(final.previewNodes) != 0 { + t.Errorf("streamDone should clear previewNodes; got %d", len(final.previewNodes)) + } + if got := strings.Join(scrollbackShape(final), "|"); !strings.Contains(got, "assistant:") { + t.Errorf("recovered partial should carry SpeakerAssistant bullet; shape=%v", got) + } +} + +// TestModel_NormalFinishNoSpuriousCancelMarker — guard the H-1 fix does NOT add +// a spurious [已取消] or duplicate text on the normal (Finish-handled) path. +func TestModel_NormalFinishNoSpuriousCancelMarker(t *testing.T) { + t.Parallel() + m := New(fake.New(), Options{ModelName: "fake"}) + m.stream = streaming.NewTokenStream(context.Background()) + cur := program.Model(m) + for _, c := range []streamControlMsg{ + {SealedText: "answer"}, + {Finish: llm.FinishStop}, + } { + next, _ := cur.Update(c) + cur = next + } + next, _ := cur.Update(streamDoneMsg{}) + final := next.(*Model) + if hasNode(final, "[已取消]") { + t.Errorf("normal finish must NOT add [已取消]: %v", nodeTexts(final)) + } + // exactly one "answer" node (no duplicate from a spurious final flush) + n := 0 + for _, txt := range nodeTexts(final) { + if strings.Contains(txt, "answer") { + n++ + } + } + if n != 1 { + t.Errorf("expected exactly 1 'answer' node, got %d: %v", n, nodeTexts(final)) + } +} + func TestModel_StatusSegments(t *testing.T) { t.Parallel() m := newFakeModel(fake.New()) diff --git a/internal/app/cli/llmapp/snapshot_test.go b/internal/app/cli/llmapp/snapshot_test.go index 3343d03..0938efd 100644 --- a/internal/app/cli/llmapp/snapshot_test.go +++ b/internal/app/cli/llmapp/snapshot_test.go @@ -8,6 +8,7 @@ package llmapp import ( + "context" "testing" "time" @@ -69,6 +70,22 @@ func TestModel_CancelledRun_KeepsPriorSnapshot(t *testing.T) { t.Errorf("cancelled run overwrote snapshot: %+v", got) } + // spec-1.21.1 R-fix H-2: a FinishError wrapping context.Canceled is ALSO a + // cancel (it renders [已取消]) and must NOT overwrite the snapshot either — + // the marker render + snapshot guard share isCancelledFinish. + outC, _ := m.handleControl(streamControlMsg{ + Finish: llm.FinishError, + Err: context.Canceled, + Snapshot: &report.RunSnapshot{Prompt: "err-cancelled"}, + }) + mc := outC.(*Model) + if got := mc.lastSnapshot; got != good { + t.Errorf("FinishError+context.Canceled overwrote snapshot: %+v", got) + } + if !hasNode(mc, "[已取消]") { + t.Errorf("FinishError+context.Canceled should render [已取消]: %v", nodeTexts(mc)) + } + out2, _ := m.handleControl(streamControlMsg{Finish: llm.FinishStop, Snapshot: &report.RunSnapshot{Prompt: "done"}}) if got := out2.(*Model).lastSnapshot; got == nil || got.Prompt != "done" { t.Errorf("completed run did not adopt snapshot: %+v", got)