Merge pull request #156 from digitopvn/codex/issue-137-mid-flight-requests

fix(pipeline): preserve mid-flight follow-up requests

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)

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
+}

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)
 		}
 	}

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)
+	}
+}

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 {

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)

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

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.

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.