diff --git a/docs/project-changelog.md b/docs/project-changelog.md index 0123c6f6..3b820c20 100644 --- a/docs/project-changelog.md +++ b/docs/project-changelog.md @@ -4,6 +4,23 @@ Significant changes, features, and fixes in reverse chronological order. --- +## 2026-06-12 + +### Mid-flight request preservation (issue #137) + +**Fixes** + +- Busy-session `chat.send` follow-ups that arrive after a no-tool final answer + now force one more pipeline iteration instead of finalizing the earlier answer. +- The earlier final answer is preserved as assistant context and the follow-up is + appended after it, so the next model turn can answer both accepted requests. + +**Tests** + +- Added pipeline regressions for late injected follow-ups after a final answer. + +--- + ## 2026-06-11 ### Trace search and advanced filters (issue #152) diff --git a/internal/pipeline/checkpoint_stage.go b/internal/pipeline/checkpoint_stage.go index 176b219c..a664fa16 100644 --- a/internal/pipeline/checkpoint_stage.go +++ b/internal/pipeline/checkpoint_stage.go @@ -3,6 +3,8 @@ package pipeline import ( "context" "log/slog" + + "github.com/nextlevelbuilder/goclaw/internal/providers" ) // CheckpointStage runs per iteration. Flushes pending messages to session store @@ -36,13 +38,32 @@ func (s *CheckpointStage) Execute(ctx context.Context, state *RunState) error { if len(pending) == 0 { return nil } + persistable := persistableMessages(pending) + if len(persistable) == 0 { + return nil + } - if err := s.deps.FlushMessages(ctx, state.Input.SessionKey, pending); err != nil { + if err := s.deps.FlushMessages(ctx, state.Input.SessionKey, persistable); err != nil { // Non-fatal: messages moved to history by FlushPending, will be flushed by FinalizeStage. slog.Warn("checkpoint flush failed", "err", err, "iteration", state.Iteration) return nil } - state.Compact.CheckpointFlushedMsgs += len(pending) + state.Compact.CheckpointFlushedMsgs += len(persistable) return nil } + +func persistableMessages(messages []providers.Message) []providers.Message { + for _, msg := range messages { + if msg.Transient { + filtered := make([]providers.Message, 0, len(messages)) + for _, candidate := range messages { + if !candidate.Transient { + filtered = append(filtered, candidate) + } + } + return filtered + } + } + return messages +} diff --git a/internal/pipeline/finalize_stage.go b/internal/pipeline/finalize_stage.go index 3236e9f6..16a0dd6d 100644 --- a/internal/pipeline/finalize_stage.go +++ b/internal/pipeline/finalize_stage.go @@ -112,7 +112,7 @@ func (s *FinalizeStage) Execute(ctx context.Context, state *RunState) error { // 4. Flush remaining pending messages to session store pending := state.Messages.FlushPending() if len(pending) > 0 && s.deps.FlushMessages != nil { - if err := s.deps.FlushMessages(ctx, state.Input.SessionKey, pending); err != nil { + if err := s.deps.FlushMessages(ctx, state.Input.SessionKey, persistableMessages(pending)); err != nil { slog.Warn("finalize flush failed", "err", err) } } diff --git a/internal/pipeline/observe_stage.go b/internal/pipeline/observe_stage.go index 0bc6763b..d8b8e437 100644 --- a/internal/pipeline/observe_stage.go +++ b/internal/pipeline/observe_stage.go @@ -1,6 +1,10 @@ package pipeline -import "context" +import ( + "context" + + "github.com/nextlevelbuilder/goclaw/internal/providers" +) // ObserveStage runs per iteration after ToolStage. Drains InjectCh, // accumulates final content when no tool calls, tracks block replies. @@ -18,51 +22,73 @@ func (s *ObserveStage) Name() string { return "observe" } // Execute drains injected messages, accumulates final content + block replies. func (s *ObserveStage) Execute(_ context.Context, state *RunState) error { - // 1. Drain InjectCh (non-blocking) — messages from tool side effects, subagent results - if s.deps.DrainInjectCh != nil { - for _, msg := range s.deps.DrainInjectCh() { - state.Messages.AppendPending(msg) - } - } + injected := s.drainInjectedMessages() resp := state.Think.LastResponse if resp == nil { + appendPendingMessages(state, injected) return nil } - // 2. Track block replies — only count tool-iteration responses where - // EmitBlockReply actually fires (think_stage emits block.reply only when - // tool calls are present). The final answer (no tool calls) must NOT be - // counted, otherwise the gateway dedup check falsely suppresses delivery. + // Track block replies only for tool-iteration responses. Final answers do + // not count, otherwise gateway dedup can suppress delivery. if resp.Content != "" && len(resp.ToolCalls) > 0 { state.Observe.BlockReplies++ state.Observe.LastBlockReply = resp.Content } - // 3. Accumulate final content when no tool calls (final answer) if len(resp.ToolCalls) == 0 { - state.Observe.FinalContent = resp.Content - state.Observe.FinalThinking = resp.Thinking - } - - // 4. Accumulate assistant-generated final images across iterations. - // The LLM may emit image_generation_call in iter N alongside a function_call, - // then respond text-only in iter N+1 — LastResponse.Images would then be empty - // at finalize time and the iter-N image would be lost. Drain Images here so - // FinalizeStage sees every image regardless of which iteration produced it. - // Partial streaming frames are filtered out at source (codex.go only sets - // non-partial entries via imageState.recordFinal); a defensive filter here - // avoids coupling to that invariant. - if len(resp.Images) > 0 { - for _, img := range resp.Images { - if img.Partial { - continue - } - state.Observe.AssistantImages = append(state.Observe.AssistantImages, img) - } - // Clear on response so a re-processing pass (e.g. retry) doesn't double-count. - resp.Images = nil + s.observeFinalResponse(state, resp, injected) + } else { + appendPendingMessages(state, injected) } + s.accumulateAssistantImages(state, resp) return nil } + +func (s *ObserveStage) drainInjectedMessages() []providers.Message { + if s.deps.DrainInjectCh == nil { + return nil + } + return s.deps.DrainInjectCh() +} + +func (s *ObserveStage) observeFinalResponse(state *RunState, resp *providers.ChatResponse, injected []providers.Message) { + if len(injected) == 0 { + state.Observe.FinalContent = resp.Content + state.Observe.FinalThinking = resp.Thinking + return + } + + state.Messages.AppendPending(providers.Message{ + Role: "assistant", + Content: resp.Content, + Thinking: resp.Thinking, + Transient: true, + }) + appendPendingMessages(state, injected) + state.Observe.FinalContent = "" + state.Observe.FinalThinking = "" + state.Observe.ContinueAfterFinal = true +} + +func (s *ObserveStage) accumulateAssistantImages(state *RunState, resp *providers.ChatResponse) { + if len(resp.Images) == 0 { + return + } + for _, img := range resp.Images { + if img.Partial { + continue + } + state.Observe.AssistantImages = append(state.Observe.AssistantImages, img) + } + // Clear on response so a re-processing pass (for example a retry) does not double-count. + resp.Images = nil +} + +func appendPendingMessages(state *RunState, messages []providers.Message) { + for _, msg := range messages { + state.Messages.AppendPending(msg) + } +} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index ae595ddd..dce26cc1 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -91,6 +91,11 @@ func (p *Pipeline) Run(ctx context.Context, state *RunState) (*RunResult, error) } } if state.ExitCode == BreakLoop { + if state.Observe.ContinueAfterFinal { + state.Observe.ContinueAfterFinal = false + state.ExitCode = Continue + continue + } break } if ctx.Err() != nil { diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index 0283d547..ab8ea8e3 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -131,6 +131,60 @@ func TestPipeline_BreakLoopExitsIteration(t *testing.T) { } } +func TestPipeline_LateInjectionAfterFinalForcesAnotherIteration(t *testing.T) { + t.Parallel() + var callCount int + var secondMessages []providers.Message + deps := PipelineDeps{ + Config: PipelineConfig{MaxIterations: 3, MaxTokens: 1000}, + CallLLM: func(_ context.Context, _ *RunState, req providers.ChatRequest) (*providers.ChatResponse, error) { + callCount++ + if callCount == 1 { + return &providers.ChatResponse{Content: "answer A", FinishReason: "stop"}, nil + } + secondMessages = append([]providers.Message(nil), req.Messages...) + return &providers.ChatResponse{Content: "answer A and B", FinishReason: "stop"}, nil + }, + } + injected := true + deps.DrainInjectCh = func() []providers.Message { + if injected { + injected = false + return []providers.Message{{Role: "user", Content: "request B"}} + } + return nil + } + + p := NewPipeline( + nil, + []Stage{NewThinkStage(&deps), NewObserveStage(&deps)}, + nil, + deps, + ) + + result, err := p.Run(context.Background(), buildMinimalRunState()) + if err != nil { + t.Fatalf("Run() error: %v", err) + } + if callCount != 2 { + t.Fatalf("CallLLM count = %d, want 2", callCount) + } + if result.Content != "answer A and B" { + t.Fatalf("result content = %q, want final answer after follow-up", result.Content) + } + if len(secondMessages) < 3 { + t.Fatalf("second call messages = %#v, want system + assistant A + user B", secondMessages) + } + if secondMessages[len(secondMessages)-2].Role != "assistant" || + secondMessages[len(secondMessages)-2].Content != "answer A" || + !secondMessages[len(secondMessages)-2].Transient { + t.Fatalf("second call penultimate message = %#v, want assistant answer A", secondMessages[len(secondMessages)-2]) + } + if secondMessages[len(secondMessages)-1].Role != "user" || secondMessages[len(secondMessages)-1].Content != "request B" { + t.Fatalf("second call final message = %#v, want user request B", secondMessages[len(secondMessages)-1]) + } +} + func TestPipeline_AbortRunExitsIteration(t *testing.T) { t.Parallel() aborter := newMockStageWithResult("aborter", AbortRun) diff --git a/internal/pipeline/stages_test.go b/internal/pipeline/stages_test.go index 05d72be1..75b90e1c 100644 --- a/internal/pipeline/stages_test.go +++ b/internal/pipeline/stages_test.go @@ -1929,7 +1929,6 @@ func TestObserveStage_DrainInjectCh_AddsToPending(t *testing.T) { } stage := NewObserveStage(deps) state := defaultState() - state.Think.LastResponse = &providers.ChatResponse{FinishReason: "stop"} err := stage.Execute(context.Background(), state) if err != nil { @@ -1944,6 +1943,44 @@ func TestObserveStage_DrainInjectCh_AddsToPending(t *testing.T) { } } +func TestObserveStage_LateInjectionAfterFinalContinuesRun(t *testing.T) { + t.Parallel() + injected := []providers.Message{ + {Role: "user", Content: "answer this follow-up too"}, + } + deps := &PipelineDeps{ + DrainInjectCh: func() []providers.Message { return injected }, + } + stage := NewObserveStage(deps) + state := defaultState() + state.Think.LastResponse = &providers.ChatResponse{ + Content: "answer to original request", + Thinking: "reasoning", + FinishReason: "stop", + } + + err := stage.Execute(context.Background(), state) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + if !state.Observe.ContinueAfterFinal { + t.Fatal("ContinueAfterFinal = false, want true") + } + if state.Observe.FinalContent != "" { + t.Fatalf("FinalContent = %q, want empty until follow-up is answered", state.Observe.FinalContent) + } + pending := state.Messages.Pending() + if len(pending) != 2 { + t.Fatalf("pending len = %d, want 2", len(pending)) + } + if pending[0].Role != "assistant" || pending[0].Content != "answer to original request" || !pending[0].Transient { + t.Fatalf("pending[0] = %#v, want assistant original answer", pending[0]) + } + if pending[1].Role != "user" || pending[1].Content != "answer this follow-up too" { + t.Fatalf("pending[1] = %#v, want user follow-up", pending[1]) + } +} + func TestObserveStage_FinalContent_SetWhenNoToolCalls(t *testing.T) { t.Parallel() deps := &PipelineDeps{} @@ -2281,6 +2318,37 @@ func TestCheckpointStage_FlushesAtInterval(t *testing.T) { } } +func TestCheckpointStage_SkipsTransientMessagesWhenPersisting(t *testing.T) { + t.Parallel() + var flushedMsgs []providers.Message + deps := &PipelineDeps{ + Config: PipelineConfig{CheckpointInterval: 5}, + FlushMessages: func(_ context.Context, _ string, msgs []providers.Message) error { + flushedMsgs = msgs + return nil + }, + } + stage := NewCheckpointStage(deps) + state := defaultState() + state.Iteration = 5 + state.Messages.AppendPending(providers.Message{Role: "assistant", Content: "draft", Transient: true}) + state.Messages.AppendPending(providers.Message{Role: "user", Content: "follow-up"}) + + err := stage.Execute(context.Background(), state) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + if len(flushedMsgs) != 1 { + t.Fatalf("flushed %d messages, want 1", len(flushedMsgs)) + } + if flushedMsgs[0].Content != "follow-up" { + t.Fatalf("flushed[0].Content = %q, want follow-up", flushedMsgs[0].Content) + } + if state.Compact.CheckpointFlushedMsgs != 1 { + t.Fatalf("CheckpointFlushedMsgs = %d, want 1", state.Compact.CheckpointFlushedMsgs) + } +} + func TestCheckpointStage_NonFatalOnFlushError(t *testing.T) { t.Parallel() deps := &PipelineDeps{ @@ -2797,6 +2865,36 @@ func TestFinalizeStage_FlushesRemainingPending(t *testing.T) { } } +func TestFinalizeStage_SkipsTransientMessagesWhenPersisting(t *testing.T) { + t.Parallel() + var flushedMsgs []providers.Message + deps := &PipelineDeps{ + FlushMessages: func(_ context.Context, _ string, msgs []providers.Message) error { + flushedMsgs = append(flushedMsgs, msgs...) + return nil + }, + } + stage := NewFinalizeStage(deps) + state := defaultState() + state.Observe.FinalContent = "final answer" + state.Messages.AppendPending(providers.Message{Role: "assistant", Content: "draft answer", Transient: true}) + state.Messages.AppendPending(providers.Message{Role: "user", Content: "follow-up"}) + + err := stage.Execute(context.Background(), state) + if err != nil { + t.Fatalf("Execute() error: %v", err) + } + if len(flushedMsgs) != 2 { + t.Fatalf("flushed %d messages, want follow-up + final answer", len(flushedMsgs)) + } + if flushedMsgs[0].Content != "follow-up" { + t.Fatalf("flushed[0].Content = %q, want follow-up", flushedMsgs[0].Content) + } + if flushedMsgs[1].Content != "final answer" { + t.Fatalf("flushed[1].Content = %q, want final answer", flushedMsgs[1].Content) + } +} + func TestFinalizeStage_CallsBootstrapCleanup_WhenHadBootstrap(t *testing.T) { t.Parallel() cleanupCalled := false diff --git a/internal/pipeline/substates.go b/internal/pipeline/substates.go index 0b38003c..8f72bbe0 100644 --- a/internal/pipeline/substates.go +++ b/internal/pipeline/substates.go @@ -8,7 +8,7 @@ import ( // ContextState: owned by ContextStage, read by ThinkStage. type ContextState struct { - ContextFiles []any // bootstrap.ContextFile — typed in Phase 2, any avoids circular import + ContextFiles []any // bootstrap.ContextFile — typed in Phase 2, any avoids circular import SkillsSummary string TeamContext string // team workspace context injected for team runs MemorySection string // L0 auto-injected memory context for system prompt @@ -70,6 +70,12 @@ type ObserveState struct { BlockReplies int LastBlockReply string + // ContinueAfterFinal is set when a user follow-up arrives after the model + // has produced a final answer but before the run finalizes. The pipeline + // must give the model another turn so accepted messages are not silently + // stored without being answered. + ContinueAfterFinal bool + // AssistantImages accumulates final (non-partial) images from every iteration's // ChatResponse.Images. FinalizeStage persists these to workspace/media/. // Accumulation is required because LastResponse holds only the final iteration's @@ -88,12 +94,12 @@ type CompactState struct { // EvolutionState: owned by skill evolution nudge logic. type EvolutionState struct { - Nudge70Sent bool - Nudge90Sent bool - PostscriptSent bool - BootstrapWrite bool // BOOTSTRAP.md write detected - TeamTaskCreates int // team_tasks tool calls - TeamTaskSpawns int // delegate tool calls (spawns) + Nudge70Sent bool + Nudge90Sent bool + PostscriptSent bool + BootstrapWrite bool // BOOTSTRAP.md write detected + TeamTaskCreates int // team_tasks tool calls + TeamTaskSpawns int // delegate tool calls (spawns) } // RunResult is the final output of a pipeline run. diff --git a/internal/providers/types.go b/internal/providers/types.go index d7f03413..f9419ab6 100644 --- a/internal/providers/types.go +++ b/internal/providers/types.go @@ -156,6 +156,10 @@ type Message struct { // Pointer type so that older messages (stored before this field existed) deserialize as nil, // allowing the frontend to fall back to synthetic timestamps. CreatedAt *time.Time `json:"created_at,omitempty"` + + // Transient messages are runtime-only context for the next provider call. + // They must not be persisted to session history or serialized to providers. + Transient bool `json:"-"` } // ToolCall represents a tool invocation requested by the LLM.