diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index 10d0d8a1b..ad6a5575a 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -97,7 +97,6 @@ type processOptions struct { SystemPromptOverride string // Override the default system prompt (Used by SubTurns) Media []string // media:// refs from inbound message InitialSteeringMessages []providers.Message // Steering messages from refactor/agent - EphemeralPrefix []providers.Message // Messages prepended to LLM context but NOT saved to history DefaultResponse string // Response when LLM returns empty EnableSummary bool // Whether to trigger summarization SendResponse bool // Whether to send response via bus @@ -590,7 +589,6 @@ func (al *AgentLoop) Run(ctx context.Context) error { // steering continuation can supersede the initial reply. // We publish exactly once at the end (matching upstream behavior). finalResponse := response - prevContent := response.Content for al.pendingSteeringCountForScope(target.SessionKey) > 0 { logger.InfoCF("agent", "Continuing queued steering after turn end", @@ -606,7 +604,6 @@ func (al *AgentLoop) Run(ctx context.Context) error { target.SessionKey, target.Channel, target.ChatID, - prevContent, ) if continueErr != nil { logger.WarnCF("agent", "Failed to continue queued steering", @@ -621,7 +618,6 @@ func (al *AgentLoop) Run(ctx context.Context) error { return } finalResponse = continued - prevContent = continued.Content } cancelDrain() @@ -640,7 +636,6 @@ func (al *AgentLoop) Run(ctx context.Context) error { target.SessionKey, target.Channel, target.ChatID, - prevContent, ) if continueErr != nil { logger.WarnCF("agent", "Failed to continue queued steering after shutdown drain", @@ -655,7 +650,6 @@ func (al *AgentLoop) Run(ctx context.Context) error { break } finalResponse = continued - prevContent = continued.Content } if finalResponse.Content != "" { @@ -1964,12 +1958,6 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er activeProvider = ts.agent.LightProvider } pendingMessages := append([]providers.Message(nil), ts.opts.InitialSteeringMessages...) - // Inject ephemeral prefix into LLM context without saving to session history. - // Used to carry the previous turn's assistant reply into a steering continuation - // so the model sees it even before OnDelivered persists it. - if len(ts.opts.EphemeralPrefix) > 0 { - messages = append(messages, ts.opts.EphemeralPrefix...) - } var finalContent string turnLoop: diff --git a/pkg/agent/steering.go b/pkg/agent/steering.go index 27c7cca64..9fbc718d6 100644 --- a/pkg/agent/steering.go +++ b/pkg/agent/steering.go @@ -290,16 +290,8 @@ func (al *AgentLoop) continueWithSteeringMessages( ctx context.Context, agent *AgentInstance, sessionKey, channel, chatID string, - prevContent string, steeringMsgs []providers.Message, ) (agentResponse, error) { - // Pass the previous assistant reply as ephemeral context so the model can - // see it without it being persisted to session history prematurely (that - // happens asynchronously via OnDelivered after channel delivery). - var ephemeral []providers.Message - if prevContent != "" { - ephemeral = []providers.Message{{Role: "assistant", Content: prevContent}} - } return al.runAgentLoop(ctx, agent, processOptions{ SessionKey: sessionKey, Channel: channel, @@ -307,7 +299,6 @@ func (al *AgentLoop) continueWithSteeringMessages( DefaultResponse: defaultResponse, EnableSummary: true, SendResponse: false, - EphemeralPrefix: ephemeral, InitialSteeringMessages: steeringMsgs, SkipInitialSteeringPoll: true, }) @@ -329,13 +320,11 @@ func (al *AgentLoop) agentForSession(sessionKey string) *AgentInstance { } // continueResponse dequeues pending steering messages and runs them through the agent loop. -// prevContent is the assistant reply from the immediately preceding turn; it is injected as -// ephemeral context so the model can reference it before OnDelivered persists it to history. // Returns an agentResponse with OnDelivered set for delayed session persistence. // If no steering messages are pending, returns an empty agentResponse. func (al *AgentLoop) continueResponse( ctx context.Context, - sessionKey, channel, chatID, prevContent string, + sessionKey, channel, chatID string, ) (agentResponse, error) { if active := al.GetActiveTurn(); active != nil { return agentResponse{}, fmt.Errorf("turn %s is still active", active.TurnID) @@ -363,7 +352,7 @@ func (al *AgentLoop) continueResponse( } } - return al.continueWithSteeringMessages(ctx, agent, sessionKey, channel, chatID, prevContent, steeringMsgs) + return al.continueWithSteeringMessages(ctx, agent, sessionKey, channel, chatID, steeringMsgs) } func (al *AgentLoop) InterruptGraceful(hint string) error { diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index ba0f7ec93..31b03eb63 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -298,7 +298,7 @@ func TestAgentLoop_Continue_NoMessages(t *testing.T) { t.Fatal("expected provider to be initialized") } - resp, err := al.continueResponse(context.Background(), "test-session", "test", "chat1", "") + resp, err := al.continueResponse(context.Background(), "test-session", "test", "chat1") if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -331,7 +331,7 @@ func TestAgentLoop_Continue_WithMessages(t *testing.T) { al.Steer(providers.Message{Role: "user", Content: "new direction"}) - resp, err := al.continueResponse(context.Background(), "test-session", "test", "chat1", "") + resp, err := al.continueResponse(context.Background(), "test-session", "test", "chat1") if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -1073,7 +1073,7 @@ func TestAgentLoop_Continue_PreservesSteeringMedia(t *testing.T) { t.Fatalf("Steer failed: %v", err) } - resp, err := al.continueResponse(context.Background(), sessionKey, "test", "chat1", "") + resp, err := al.continueResponse(context.Background(), sessionKey, "test", "chat1") if err != nil { t.Fatalf("continueResponse failed: %v", err) } @@ -1583,16 +1583,15 @@ func (w *wrappingProvider) GetDefaultModel() string { return w.inner.GetDefaultModel() } -// TestContinueResponse_EphemeralPrefixBeforePersistence verifies the two -// invariants of the ephemeral-prefix design: +// TestContinueResponse_PendingDeliveryVisibleBeforePersistence verifies that +// a steering continuation sees the previous assistant reply in its LLM context +// even when OnDelivered has not fired yet (reply not yet in session history). +// The pending reply is injected via pendingDeliveries/injectPendingDelivery, +// not via EphemeralPrefix. // -// 1. The steering continuation's LLM call sees the previous assistant reply -// in its message list even though OnDelivered has not fired yet (i.e. the -// reply is not yet in session history). -// -// 2. Session history does not contain the assistant reply at the time the -// continuation starts — only ephemeral context was injected. -func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) { +// 1. The steering continuation's LLM call sees the previous assistant reply. +// 2. Session history does not contain the assistant reply (OnDelivered not called). +func TestContinueResponse_PendingDeliveryVisibleBeforePersistence(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") if err != nil { t.Fatalf("MkdirTemp: %v", err) @@ -1664,8 +1663,8 @@ func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) { } } - // Step 2: enqueue a steering message and run the continuation, passing - // the previous assistant reply as ephemeral context. + // Step 2: enqueue a steering message and run the continuation. + // pendingDeliveries already holds the first reply; no need to pass it explicitly. if pushErr := al.steering.pushScope(sessionKey, providers.Message{ Role: "user", Content: "follow-up question", @@ -1673,7 +1672,7 @@ func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) { t.Fatalf("pushScope: %v", pushErr) } - continued, err := al.continueResponse(ctx, sessionKey, channel, chatID, firstReply) + continued, err := al.continueResponse(ctx, sessionKey, channel, chatID) if err != nil { t.Fatalf("continueResponse: %v", err) } @@ -1681,7 +1680,7 @@ func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) { t.Fatalf("continuation reply = %q, want %q", continued.Content, continuationReply) } - // Step 3: verify the LLM received the ephemeral assistant reply in context. + // Step 3: verify the LLM received the pending assistant reply in context. captureMu.Lock() msgs := append([]providers.Message(nil), capturedMessages...) captureMu.Unlock() @@ -1701,7 +1700,7 @@ func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) { for i, m := range msgs { roles[i] = fmt.Sprintf("%s:%q", m.Role, m.Content) } - t.Fatalf("ephemeral assistant reply not found in LLM context; messages: %v", roles) + t.Fatalf("pending assistant reply not found in LLM context; messages: %v", roles) } // Step 4: verify the session history still does not contain the first