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.
This commit is contained in:
Dmitrii Balabanov 2026-03-31 20:01:31 +03:00
parent fb7d3a375e
commit 82a98a21d2
3 changed files with 18 additions and 42 deletions

View file

@ -97,7 +97,6 @@ type processOptions struct {
SystemPromptOverride string // Override the default system prompt (Used by SubTurns) SystemPromptOverride string // Override the default system prompt (Used by SubTurns)
Media []string // media:// refs from inbound message Media []string // media:// refs from inbound message
InitialSteeringMessages []providers.Message // Steering messages from refactor/agent 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 DefaultResponse string // Response when LLM returns empty
EnableSummary bool // Whether to trigger summarization EnableSummary bool // Whether to trigger summarization
SendResponse bool // Whether to send response via bus 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. // steering continuation can supersede the initial reply.
// We publish exactly once at the end (matching upstream behavior). // We publish exactly once at the end (matching upstream behavior).
finalResponse := response finalResponse := response
prevContent := response.Content
for al.pendingSteeringCountForScope(target.SessionKey) > 0 { for al.pendingSteeringCountForScope(target.SessionKey) > 0 {
logger.InfoCF("agent", "Continuing queued steering after turn end", logger.InfoCF("agent", "Continuing queued steering after turn end",
@ -606,7 +604,6 @@ func (al *AgentLoop) Run(ctx context.Context) error {
target.SessionKey, target.SessionKey,
target.Channel, target.Channel,
target.ChatID, target.ChatID,
prevContent,
) )
if continueErr != nil { if continueErr != nil {
logger.WarnCF("agent", "Failed to continue queued steering", logger.WarnCF("agent", "Failed to continue queued steering",
@ -621,7 +618,6 @@ func (al *AgentLoop) Run(ctx context.Context) error {
return return
} }
finalResponse = continued finalResponse = continued
prevContent = continued.Content
} }
cancelDrain() cancelDrain()
@ -640,7 +636,6 @@ func (al *AgentLoop) Run(ctx context.Context) error {
target.SessionKey, target.SessionKey,
target.Channel, target.Channel,
target.ChatID, target.ChatID,
prevContent,
) )
if continueErr != nil { if continueErr != nil {
logger.WarnCF("agent", "Failed to continue queued steering after shutdown drain", logger.WarnCF("agent", "Failed to continue queued steering after shutdown drain",
@ -655,7 +650,6 @@ func (al *AgentLoop) Run(ctx context.Context) error {
break break
} }
finalResponse = continued finalResponse = continued
prevContent = continued.Content
} }
if finalResponse.Content != "" { if finalResponse.Content != "" {
@ -1964,12 +1958,6 @@ func (al *AgentLoop) runTurn(ctx context.Context, ts *turnState) (turnResult, er
activeProvider = ts.agent.LightProvider activeProvider = ts.agent.LightProvider
} }
pendingMessages := append([]providers.Message(nil), ts.opts.InitialSteeringMessages...) 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 var finalContent string
turnLoop: turnLoop:

View file

@ -290,16 +290,8 @@ func (al *AgentLoop) continueWithSteeringMessages(
ctx context.Context, ctx context.Context,
agent *AgentInstance, agent *AgentInstance,
sessionKey, channel, chatID string, sessionKey, channel, chatID string,
prevContent string,
steeringMsgs []providers.Message, steeringMsgs []providers.Message,
) (agentResponse, error) { ) (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{ return al.runAgentLoop(ctx, agent, processOptions{
SessionKey: sessionKey, SessionKey: sessionKey,
Channel: channel, Channel: channel,
@ -307,7 +299,6 @@ func (al *AgentLoop) continueWithSteeringMessages(
DefaultResponse: defaultResponse, DefaultResponse: defaultResponse,
EnableSummary: true, EnableSummary: true,
SendResponse: false, SendResponse: false,
EphemeralPrefix: ephemeral,
InitialSteeringMessages: steeringMsgs, InitialSteeringMessages: steeringMsgs,
SkipInitialSteeringPoll: true, 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. // 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. // Returns an agentResponse with OnDelivered set for delayed session persistence.
// If no steering messages are pending, returns an empty agentResponse. // If no steering messages are pending, returns an empty agentResponse.
func (al *AgentLoop) continueResponse( func (al *AgentLoop) continueResponse(
ctx context.Context, ctx context.Context,
sessionKey, channel, chatID, prevContent string, sessionKey, channel, chatID string,
) (agentResponse, error) { ) (agentResponse, error) {
if active := al.GetActiveTurn(); active != nil { if active := al.GetActiveTurn(); active != nil {
return agentResponse{}, fmt.Errorf("turn %s is still active", active.TurnID) 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 { func (al *AgentLoop) InterruptGraceful(hint string) error {

View file

@ -298,7 +298,7 @@ func TestAgentLoop_Continue_NoMessages(t *testing.T) {
t.Fatal("expected provider to be initialized") 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 { if err != nil {
t.Fatalf("unexpected error: %v", err) 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"}) 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 { if err != nil {
t.Fatalf("unexpected error: %v", err) t.Fatalf("unexpected error: %v", err)
} }
@ -1073,7 +1073,7 @@ func TestAgentLoop_Continue_PreservesSteeringMedia(t *testing.T) {
t.Fatalf("Steer failed: %v", err) 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 { if err != nil {
t.Fatalf("continueResponse failed: %v", err) t.Fatalf("continueResponse failed: %v", err)
} }
@ -1583,16 +1583,15 @@ func (w *wrappingProvider) GetDefaultModel() string {
return w.inner.GetDefaultModel() return w.inner.GetDefaultModel()
} }
// TestContinueResponse_EphemeralPrefixBeforePersistence verifies the two // TestContinueResponse_PendingDeliveryVisibleBeforePersistence verifies that
// invariants of the ephemeral-prefix design: // 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 // 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 // 2. Session history does not contain the assistant reply (OnDelivered not called).
// reply is not yet in session history). func TestContinueResponse_PendingDeliveryVisibleBeforePersistence(t *testing.T) {
//
// 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) {
tmpDir, err := os.MkdirTemp("", "agent-test-*") tmpDir, err := os.MkdirTemp("", "agent-test-*")
if err != nil { if err != nil {
t.Fatalf("MkdirTemp: %v", err) 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 // Step 2: enqueue a steering message and run the continuation.
// the previous assistant reply as ephemeral context. // pendingDeliveries already holds the first reply; no need to pass it explicitly.
if pushErr := al.steering.pushScope(sessionKey, providers.Message{ if pushErr := al.steering.pushScope(sessionKey, providers.Message{
Role: "user", Role: "user",
Content: "follow-up question", Content: "follow-up question",
@ -1673,7 +1672,7 @@ func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) {
t.Fatalf("pushScope: %v", pushErr) 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 { if err != nil {
t.Fatalf("continueResponse: %v", err) 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) 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() captureMu.Lock()
msgs := append([]providers.Message(nil), capturedMessages...) msgs := append([]providers.Message(nil), capturedMessages...)
captureMu.Unlock() captureMu.Unlock()
@ -1701,7 +1700,7 @@ func TestContinueResponse_EphemeralPrefixBeforePersistence(t *testing.T) {
for i, m := range msgs { for i, m := range msgs {
roles[i] = fmt.Sprintf("%s:%q", m.Role, m.Content) 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 // Step 4: verify the session history still does not contain the first