From 82a98a21d2037453c2f3b2bb133e09b913be37c6 Mon Sep 17 00:00:00 2001 From: Dmitrii Balabanov Date: Tue, 31 Mar 2026 20:01:31 +0300 Subject: [PATCH] refactor: remove EphemeralPrefix in favour of pendingDeliveries injection EphemeralPrefix was added to carry the previous assistant reply into a steering continuation before OnDelivered persisted it. pendingDeliveries/injectPendingDelivery already covers this case: the slot is populated before continueResponse is called and cannot be evicted before then (publish happens only after all continuations complete). The two mechanisms were redundant. Remove EphemeralPrefix from processOptions, its injection site in runTurn, the prevContent parameter from continueResponse/continueWithSteeringMessages, and the corresponding prevContent threading in the Run loop. --- pkg/agent/loop.go | 12 ------------ pkg/agent/steering.go | 15 ++------------- pkg/agent/steering_test.go | 33 ++++++++++++++++----------------- 3 files changed, 18 insertions(+), 42 deletions(-) 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