From 44569ffddac093ff5c697d738274892e5b0bf556 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 16 Jul 2026 23:58:43 -0700 Subject: [PATCH] fix(agents): prevent stale replies after transcript rewrites (#109676) * fix(agents): scope final replies to current run Co-authored-by: ZengWen-DT * refactor(agents): internalize attempt helper * fix(agents): preserve yielded turn classification * fix(agents): preserve yielded run ownership --------- Co-authored-by: ZengWen-DT --- src/agents/embedded-agent-runner/run-loop.ts | 4 +- .../run.incomplete-turn.test.ts | 86 ++++++++++++++++++- .../run.overflow-compaction.fixture.ts | 2 + .../run/attempt-execution-settle.ts | 3 + .../run/attempt-normalization.ts | 5 ++ .../run/attempt-phase-lifecycle.test.ts | 1 + .../run/attempt-result.ts | 1 + .../run/attempt-stream-settle.ts | 9 +- ...mpt.spawn-workspace.context-engine.test.ts | 7 +- .../attempt.spawn-workspace.test-support.ts | 1 + .../run/terminal-preparation.ts | 29 ++++--- src/agents/embedded-agent-runner/run/types.ts | 2 + ...-agent-subscribe.handlers.messages.test.ts | 2 + ...edded-agent-subscribe.handlers.messages.ts | 1 + ...ent-subscribe.handlers.tools.media.test.ts | 1 + ...embedded-agent-subscribe.handlers.types.ts | 1 + ...-emit-duplicate-block-replies-text.test.ts | 15 ++++ ...ompaction-retries-before-resolving.test.ts | 19 ++++ src/agents/embedded-agent-subscribe.ts | 15 ++++ 19 files changed, 183 insertions(+), 21 deletions(-) diff --git a/src/agents/embedded-agent-runner/run-loop.ts b/src/agents/embedded-agent-runner/run-loop.ts index 2174cf8cf0ff..22ccef0e64c4 100644 --- a/src/agents/embedded-agent-runner/run-loop.ts +++ b/src/agents/embedded-agent-runner/run-loop.ts @@ -388,6 +388,7 @@ export async function runPreparedEmbeddedLoop( sessionIdUsed, sessionFileUsed, currentAttemptAssistant, + currentAttemptCompletedAssistant, attemptAssistant, terminalOutcome, terminalAborted, @@ -516,8 +517,7 @@ export async function runPreparedEmbeddedLoop( } = prepareEmbeddedRunTerminal({ runParams: params, attempt, - attemptAssistant, - currentAttemptAssistant, + currentAttemptCompletedAssistant, provider, model: model.id, activeErrorContext, diff --git a/src/agents/embedded-agent-runner/run.incomplete-turn.test.ts b/src/agents/embedded-agent-runner/run.incomplete-turn.test.ts index e9a031e6ff83..8edd1e90cb6a 100644 --- a/src/agents/embedded-agent-runner/run.incomplete-turn.test.ts +++ b/src/agents/embedded-agent-runner/run.incomplete-turn.test.ts @@ -789,6 +789,87 @@ describe("runEmbeddedAgent incomplete-turn safety", () => { expect(result.meta.finalAssistantVisibleText).toBeUndefined(); }); + it("does not resolve a successful run from a stale transcript assistant", async () => { + const staleAssistant = { + role: "assistant", + stopReason: "stop", + provider: "openai", + model: "gpt-5.5", + content: [{ type: "text", text: "Prior transcript reply." }], + } as unknown as NonNullable; + const completedAssistant = { + role: "assistant", + stopReason: "stop", + provider: "openai", + model: "gpt-5.5", + content: [{ type: "text", text: "Current run reply." }], + } as unknown as NonNullable; + mockedBuildEmbeddedRunPayloads.mockReturnValue([{ text: "Current run reply." }]); + mockedRunEmbeddedAttempt.mockResolvedValueOnce( + makeAttemptResult({ + assistantTexts: ["Current run reply."], + lastAssistant: staleAssistant, + currentAttemptAssistant: staleAssistant, + currentAttemptCompletedAssistant: completedAssistant, + }), + ); + + const result = await runEmbeddedAgent({ + ...overflowBaseRunParams, + provider: "openai", + model: "gpt-5.5", + runId: "run-success-stale-transcript-assistant", + }); + + expect(result.payloads).toEqual([{ text: "Current run reply." }]); + expect(result.meta.finalAssistantVisibleText).toBe("Current run reply."); + expect(result.meta.finalAssistantRawText).toBe("Current run reply."); + expect(mockedBuildEmbeddedRunPayloads).toHaveBeenCalledWith( + expect.objectContaining({ + currentAssistant: completedAssistant, + lastAssistant: completedAssistant, + }), + ); + }); + + it("retains the yielded attempt assistant for paused-turn payload classification", async () => { + const completedAssistant = { + role: "assistant", + stopReason: "stop", + provider: "openai", + model: "gpt-5.5", + content: [{ type: "text", text: "Earlier completed cycle." }], + } as unknown as NonNullable; + const yieldedAssistant = { + role: "assistant", + stopReason: "aborted", + provider: "openai", + model: "gpt-5.5", + content: [{ type: "toolCall", name: "sessions_yield", arguments: {} }], + } as unknown as NonNullable; + mockedRunEmbeddedAttempt.mockResolvedValueOnce( + makeAttemptResult({ + assistantTexts: [], + lastAssistant: yieldedAssistant, + currentAttemptAssistant: undefined, + currentAttemptCompletedAssistant: completedAssistant, + yieldDetected: true, + }), + ); + + const result = await runEmbeddedAgent({ + ...overflowBaseRunParams, + provider: "openai", + model: "gpt-5.5", + runId: "run-yielded-assistant-classification", + }); + + expect(result.meta).toMatchObject({ livenessState: "paused", yielded: true }); + expect(mockedBuildEmbeddedRunPayloads).toHaveBeenCalledWith( + expect.objectContaining({ currentAssistant: null, lastAssistant: yieldedAssistant }), + ); + }); + it("recovers a completed prompt-timeout assistant without collected assistant text", async () => { mockedClassifyFailoverReason.mockReturnValue(null); const finalText = "Completed answer after the timeout race."; @@ -2523,7 +2604,10 @@ describe("runEmbeddedAgent incomplete-turn safety", () => { stopReason: "stop", content: [{ type: "text", text: finalText }], }), - lastAssistant: expect.objectContaining({ stopReason: "toolUse" }), + lastAssistant: expect.objectContaining({ + stopReason: "stop", + content: [{ type: "text", text: finalText }], + }), }), ); expect(result.meta.finalAssistantVisibleText).toBe(finalText); diff --git a/src/agents/embedded-agent-runner/run.overflow-compaction.fixture.ts b/src/agents/embedded-agent-runner/run.overflow-compaction.fixture.ts index 70b0957fb8ef..0c1d3473a550 100644 --- a/src/agents/embedded-agent-runner/run.overflow-compaction.fixture.ts +++ b/src/agents/embedded-agent-runner/run.overflow-compaction.fixture.ts @@ -64,6 +64,8 @@ export function makeAttemptResult( assistantTexts: ["Hello!"], acceptedSessionSpawns, lastAssistant: undefined, + currentAttemptCompletedAssistant: + overrides.currentAttemptCompletedAssistant ?? overrides.currentAttemptAssistant, messagesSnapshot: [], replayMetadata: overrides.replayMetadata ?? diff --git a/src/agents/embedded-agent-runner/run/attempt-execution-settle.ts b/src/agents/embedded-agent-runner/run/attempt-execution-settle.ts index ddba287695af..b1de7325da62 100644 --- a/src/agents/embedded-agent-runner/run/attempt-execution-settle.ts +++ b/src/agents/embedded-agent-runner/run/attempt-execution-settle.ts @@ -133,6 +133,7 @@ export async function runEmbeddedAttemptSettledPhase( let promptCacheChangesForTurn: PromptCacheChange[] | null = null; let lastAssistant: AssistantMessage | undefined; let currentAttemptAssistant: EmbeddedRunAttemptResult["currentAttemptAssistant"]; + let currentAttemptCompletedAssistant: EmbeddedRunAttemptResult["currentAttemptCompletedAssistant"]; let attemptUsage: NormalizedUsage | undefined; let cacheBreak: PromptCacheBreak | null = null; let contextBudgetStatus: EmbeddedRunAttemptResult["contextBudgetStatus"]; @@ -290,6 +291,7 @@ export async function runEmbeddedAttemptSettledPhase( sessionIdUsed = settledStream.sessionIdUsed; lastAssistant = settledStream.lastAssistant; currentAttemptAssistant = settledStream.currentAttemptAssistant; + currentAttemptCompletedAssistant = settledStream.currentAttemptCompletedAssistant; attemptUsage = settledStream.attemptUsage; cacheBreak = settledStream.cacheBreak; sessionRuntimeState.promptCache = settledStream.promptCache; @@ -387,6 +389,7 @@ export async function runEmbeddedAttemptSettledPhase( ...(beforeAgentFinalizeRevisionReason ? { beforeAgentFinalizeRevisionReason } : {}), lastAssistant, currentAttemptAssistant, + currentAttemptCompletedAssistant, attemptUsage, promptCache: sessionRuntimeState.promptCache, contextBudgetStatus, diff --git a/src/agents/embedded-agent-runner/run/attempt-normalization.ts b/src/agents/embedded-agent-runner/run/attempt-normalization.ts index 8ef286eb459e..eb08e254e0d1 100644 --- a/src/agents/embedded-agent-runner/run/attempt-normalization.ts +++ b/src/agents/embedded-agent-runner/run/attempt-normalization.ts @@ -87,6 +87,9 @@ export async function normalizeEmbeddedRunAttempt(input: { currentAttemptAssistant: ReturnType< typeof normalizeEmbeddedRunAttemptResult >["currentAttemptAssistant"]; + currentAttemptCompletedAssistant: ReturnType< + typeof normalizeEmbeddedRunAttemptResult + >["currentAttemptCompletedAssistant"]; attemptAssistant: ReturnType< typeof normalizeEmbeddedRunAttemptResult >["currentAttemptAssistant"]; @@ -129,6 +132,7 @@ export async function normalizeEmbeddedRunAttempt(input: { sessionFileUsed, lastAssistant: sessionLastAssistant, currentAttemptAssistant, + currentAttemptCompletedAssistant, } = attempt; const timedOutDuringToolExecution = attempt.timedOutDuringToolExecution ?? false; const timedOutByRunBudget = attempt.timedOutByRunBudget ?? false; @@ -320,6 +324,7 @@ export async function normalizeEmbeddedRunAttempt(input: { sessionIdUsed, sessionFileUsed, currentAttemptAssistant, + currentAttemptCompletedAssistant, attemptAssistant, terminalOutcome, terminalAborted, diff --git a/src/agents/embedded-agent-runner/run/attempt-phase-lifecycle.test.ts b/src/agents/embedded-agent-runner/run/attempt-phase-lifecycle.test.ts index b0aa5c691f34..b7d5942508c7 100644 --- a/src/agents/embedded-agent-runner/run/attempt-phase-lifecycle.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt-phase-lifecycle.test.ts @@ -61,6 +61,7 @@ describe("embedded attempt phase lifecycle state", () => { }, isCompactionInFlight: () => false, getCompactionCount: () => 0, + getCurrentAttemptAssistant: () => undefined, getUsageTotals: () => undefined, } as never, state: { diff --git a/src/agents/embedded-agent-runner/run/attempt-result.ts b/src/agents/embedded-agent-runner/run/attempt-result.ts index 4a7e53b5b26d..26088ba6c94b 100644 --- a/src/agents/embedded-agent-runner/run/attempt-result.ts +++ b/src/agents/embedded-agent-runner/run/attempt-result.ts @@ -59,6 +59,7 @@ type EmbeddedAttemptResultState = Pick< | "beforeAgentFinalizeRevisionReason" | "lastAssistant" | "currentAttemptAssistant" + | "currentAttemptCompletedAssistant" | "attemptUsage" | "promptCache" | "contextBudgetStatus" diff --git a/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts b/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts index 09ce493b2bd8..aa7238b498c7 100644 --- a/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts +++ b/src/agents/embedded-agent-runner/run/attempt-stream-settle.ts @@ -58,6 +58,7 @@ type StreamSettleResult = { sessionIdUsed: string; lastAssistant: EmbeddedRunAttemptResult["lastAssistant"]; currentAttemptAssistant: EmbeddedRunAttemptResult["currentAttemptAssistant"]; + currentAttemptCompletedAssistant: EmbeddedRunAttemptResult["currentAttemptCompletedAssistant"]; attemptUsage: EmbeddedRunAttemptResult["attemptUsage"]; cacheBreak: PromptCacheBreak | null; lastCallUsage: NormalizedUsage | undefined; @@ -226,6 +227,7 @@ export async function settleEmbeddedAttemptStream(input: { let messagesSnapshot: AgentMessage[] = []; let lastAssistant: AssistantMessage | undefined; let currentAttemptAssistant: AssistantMessage | undefined; + let currentAttemptCompletedAssistant: AssistantMessage | undefined; let attemptUsage: EmbeddedRunAttemptResult["attemptUsage"]; let cacheBreak: PromptCacheBreak | null = null; let lastCallUsage: NormalizedUsage | undefined; @@ -285,6 +287,8 @@ export async function settleEmbeddedAttemptStream(input: { messagesSnapshot, prePromptMessageCount: input.prePromptMessageCount, }); + currentAttemptCompletedAssistant = subscription.getCurrentAttemptAssistant(); + const usageAssistant = currentAttemptCompletedAssistant ?? currentAttemptAssistant; attemptUsage = subscription.getUsageTotals(); cacheBreak = input.cache.observabilityEnabled ? completePromptCacheObservation({ @@ -294,7 +298,7 @@ export async function settleEmbeddedAttemptStream(input: { usage: attemptUsage, }) : null; - lastCallUsage = normalizeUsage(currentAttemptAssistant?.usage); + lastCallUsage = normalizeUsage(usageAssistant?.usage); const promptCacheObservation = input.cache.observabilityEnabled && (cacheBreak || input.cache.changesForTurn || typeof attemptUsage?.cacheRead === "number") @@ -321,7 +325,7 @@ export async function settleEmbeddedAttemptStream(input: { observation: promptCacheObservation, lastCacheTouchAt: resolvePromptCacheTouchTimestamp({ lastCallUsage, - assistantTimestamp: currentAttemptAssistant?.timestamp, + assistantTimestamp: usageAssistant?.timestamp, fallbackLastCacheTouchAt, }), }); @@ -356,6 +360,7 @@ export async function settleEmbeddedAttemptStream(input: { sessionIdUsed, lastAssistant, currentAttemptAssistant, + currentAttemptCompletedAssistant, attemptUsage, cacheBreak, lastCallUsage, diff --git a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts index 1c41fe87d610..5d8afcf95df3 100644 --- a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts +++ b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.context-engine.test.ts @@ -29,7 +29,6 @@ import { buildLoopPromptCacheInfo, assembleAttemptContextEngine, buildContextEnginePromptCacheInfo, - findCurrentAttemptAssistantMessage, finalizeAttemptContextEngineTurn, resolvePromptCacheTouchTimestamp, runAttemptContextEngineBootstrap, @@ -3190,16 +3189,12 @@ describe("runEmbeddedAttempt context engine sessionKey forwarding", () => { total: 1340, }, } as unknown as AgentMessage; - const currentAttemptAssistant = findCurrentAttemptAssistantMessage({ + const promptCache = buildLoopPromptCacheInfo({ messagesSnapshot: [seedMessage, priorAssistant], prePromptMessageCount: 2, - }); - const promptCache = buildContextEnginePromptCacheInfo({ retention: "short", - lastCallUsage: (currentAttemptAssistant as { usage?: undefined } | undefined)?.usage, }); - expect(currentAttemptAssistant).toBeUndefined(); expect(promptCache).toEqual({ retention: "short" }); }); diff --git a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts index e0611070c409..f5e8921f197c 100644 --- a/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts +++ b/src/agents/embedded-agent-runner/run/attempt.spawn-workspace.test-support.ts @@ -122,6 +122,7 @@ function createSubscriptionMock(): SubscriptionMock { // override only the lifecycle method they need. return { assistantTexts: [] as string[], + getCurrentAttemptAssistant: () => undefined, getLastAssistantTextMessageIndex: () => undefined, toolMetas: [] as Array<{ toolName: string; meta?: string; asyncStarted?: boolean }>, runToolLifecycle: async (toolParams: { execute: () => Promise }) => diff --git a/src/agents/embedded-agent-runner/run/terminal-preparation.ts b/src/agents/embedded-agent-runner/run/terminal-preparation.ts index 40459a786030..90428ad22aea 100644 --- a/src/agents/embedded-agent-runner/run/terminal-preparation.ts +++ b/src/agents/embedded-agent-runner/run/terminal-preparation.ts @@ -21,8 +21,7 @@ import type { EmbeddedRunAttemptResult } from "./types.js"; export function prepareEmbeddedRunTerminal(input: { runParams: RunEmbeddedAgentParams; attempt: EmbeddedRunAttemptResult; - attemptAssistant?: AssistantMessage; - currentAttemptAssistant?: AssistantMessage; + currentAttemptCompletedAssistant?: AssistantMessage; provider: string; model: string; activeErrorContext: { provider: string; model: string }; @@ -54,12 +53,12 @@ export function prepareEmbeddedRunTerminal(input: { attemptToolSummary: ReturnType; failureSignal: ReturnType; } { - const { runParams, attempt, attemptAssistant } = input; + const { runParams, attempt } = input; const timedOutDuringPrompt = input.terminalTimedOut && !input.timedOutDuringCompaction && !input.timedOutDuringToolExecution; - // A prior same-model assistant can remain in the session snapshot. Timeout - // recovery must project only output owned by the prompt that just timed out. - const terminalAssistant = timedOutDuringPrompt ? input.currentAttemptAssistant : attemptAssistant; + // Session transcript fallbacks can reference an earlier rewritten turn. + // Terminal delivery and metadata must stay scoped to this model attempt. + const terminalAssistant = input.currentAttemptCompletedAssistant; const usageMeta = buildUsageAgentMetaFields({ usageAccumulator: input.usageAccumulator, lastAssistantUsage: terminalAssistant?.usage as UsageLike | undefined, @@ -90,15 +89,25 @@ export function prepareEmbeddedRunTerminal(input: { : undefined, compactionTokensAfter: input.contextRecoveryState.lastCompactionTokensAfter, }; - const finalAssistantVisibleText = resolveFinalAssistantVisibleText(terminalAssistant); - const finalAssistantRawText = resolveFinalAssistantRawText(terminalAssistant); + const attemptFinalText = attempt.assistantTexts + .toReversed() + .map((text) => text.trim()) + .find((text) => text.length > 0); + const finalAssistantVisibleText = + resolveFinalAssistantVisibleText(terminalAssistant) ?? attemptFinalText; + const finalAssistantRawText = resolveFinalAssistantRawText(terminalAssistant) ?? attemptFinalText; + // A yielded attempt ends before message_end. Its aborted tool-call assistant, + // not an earlier completed cycle, owns paused-turn classification. + const payloadAssistant = attempt.yieldDetected + ? attempt.lastAssistant + : input.currentAttemptCompletedAssistant; const payloads = buildEmbeddedRunPayloads({ assistantTexts: attempt.assistantTexts, assistantMessageIndex: attempt.lastAssistantTextMessageIndex, assistantTranscriptOwned: attempt.assistantTranscriptOwned, toolMetas: attempt.toolMetas, - lastAssistant: timedOutDuringPrompt ? input.currentAttemptAssistant : attempt.lastAssistant, - currentAssistant: input.currentAttemptAssistant ?? null, + lastAssistant: payloadAssistant, + currentAssistant: attempt.yieldDetected ? null : (payloadAssistant ?? null), lastToolError: attempt.lastToolError, config: runParams.config, isCronTrigger: runParams.trigger === "cron", diff --git a/src/agents/embedded-agent-runner/run/types.ts b/src/agents/embedded-agent-runner/run/types.ts index e06665c42a8f..f414c9201413 100644 --- a/src/agents/embedded-agent-runner/run/types.ts +++ b/src/agents/embedded-agent-runner/run/types.ts @@ -272,6 +272,8 @@ export type EmbeddedRunAttemptResult = { acceptedSessionSpawns?: AcceptedSessionSpawn[]; lastAssistant: AssistantMessage | undefined; currentAttemptAssistant?: AssistantMessage | undefined; + /** Completed message_end snapshot owned by this model attempt. */ + currentAttemptCompletedAssistant?: AssistantMessage | undefined; lastToolError?: ToolErrorSummary; didSendViaMessagingTool: boolean; didDeliverSourceReplyViaMessageTool?: boolean; diff --git a/src/agents/embedded-agent-subscribe.handlers.messages.test.ts b/src/agents/embedded-agent-subscribe.handlers.messages.test.ts index 5b91cf650cd6..a6e40a62d980 100644 --- a/src/agents/embedded-agent-subscribe.handlers.messages.test.ts +++ b/src/agents/embedded-agent-subscribe.handlers.messages.test.ts @@ -74,6 +74,7 @@ function createMessageUpdateContext( }, log: { debug: params.debug ?? vi.fn() }, noteLastAssistant: vi.fn(), + noteCompletedAssistant: vi.fn(), stripBlockTags: params.stripBlockTags ?? vi.fn((text: string) => text), consumePartialReplyDirectives: params.consumePartialReplyDirectives ?? @@ -160,6 +161,7 @@ function createMessageEndContext( ...params.state, }, noteLastAssistant: vi.fn(), + noteCompletedAssistant: vi.fn(), recordAssistantUsage: vi.fn(), commitAssistantUsage: vi.fn(), log: { debug: vi.fn(), info: vi.fn(), warn: params.warn ?? vi.fn() }, diff --git a/src/agents/embedded-agent-subscribe.handlers.messages.ts b/src/agents/embedded-agent-subscribe.handlers.messages.ts index 06798f0e1f09..585f1457a2f4 100644 --- a/src/agents/embedded-agent-subscribe.handlers.messages.ts +++ b/src/agents/embedded-agent-subscribe.handlers.messages.ts @@ -1159,6 +1159,7 @@ export function handleMessageEnd( const suppressDeterministicApprovalOutput = shouldSuppressDeterministicApprovalOutput(ctx.state); const suppressMessageToolOnlySourceReplyOutput = hasMessageToolOnlySourceDelivery(ctx); ctx.noteLastAssistant(assistantMessage); + ctx.noteCompletedAssistant(assistantMessage); ctx.recordAssistantUsage((assistantMessage as { usage?: unknown }).usage); ctx.commitAssistantUsage(); if (suppressVisibleAssistantOutput) { diff --git a/src/agents/embedded-agent-subscribe.handlers.tools.media.test.ts b/src/agents/embedded-agent-subscribe.handlers.tools.media.test.ts index a37b31f417b3..a37bed6ff523 100644 --- a/src/agents/embedded-agent-subscribe.handlers.tools.media.test.ts +++ b/src/agents/embedded-agent-subscribe.handlers.tools.media.test.ts @@ -61,6 +61,7 @@ function createMockContext(overrides?: { // Fill in remaining required fields with no-ops. blockChunker: null, noteLastAssistant: vi.fn(), + noteCompletedAssistant: vi.fn(), stripBlockTags: vi.fn((t: string) => t), emitBlockChunk: vi.fn(), flushBlockReplyBuffer: vi.fn(), diff --git a/src/agents/embedded-agent-subscribe.handlers.types.ts b/src/agents/embedded-agent-subscribe.handlers.types.ts index 0298703f0956..579b25fddaad 100644 --- a/src/agents/embedded-agent-subscribe.handlers.types.ts +++ b/src/agents/embedded-agent-subscribe.handlers.types.ts @@ -196,6 +196,7 @@ export type EmbeddedAgentSubscribeContext = { builtinToolNames?: ReadonlySet; trustedLocalMediaToolNames?: ReadonlySet; noteLastAssistant: (msg: AgentMessage) => void; + noteCompletedAssistant: (msg: AgentMessage) => void; shouldEmitToolResult: () => boolean; shouldEmitToolOutput: () => boolean; diff --git a/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.does-not-emit-duplicate-block-replies-text.test.ts b/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.does-not-emit-duplicate-block-replies-text.test.ts index 0eeb98c861d9..55c252035542 100644 --- a/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.does-not-emit-duplicate-block-replies-text.test.ts +++ b/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.does-not-emit-duplicate-block-replies-text.test.ts @@ -41,6 +41,21 @@ describe("subscribeEmbeddedAgentSession", () => { expect(subscription.assistantTexts).toEqual(["Hello world"]); }); + it("keeps the completed assistant independent from transcript mutation", () => { + const { session, emit } = createStubSessionHarness(); + const subscription = subscribeEmbeddedAgentSession({ session, runId: "run" }); + const assistantMessage = { + role: "assistant", + content: [{ type: "text", text: "Current run reply" }], + } as AssistantMessage; + + emit({ type: "message_end", message: assistantMessage }); + assistantMessage.content = [{ type: "text", text: "Rewritten transcript reply" }]; + + expect(subscription.getCurrentAttemptAssistant()?.content).toEqual([ + { type: "text", text: "Current run reply" }, + ]); + }); it("does not duplicate assistantTexts when message_end repeats with trailing whitespace changes", () => { const { session, emit } = createStubSessionHarness(); diff --git a/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.waits-multiple-compaction-retries-before-resolving.test.ts b/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.waits-multiple-compaction-retries-before-resolving.test.ts index 8de7ed182bde..5520f6a97088 100644 --- a/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.waits-multiple-compaction-retries-before-resolving.test.ts +++ b/src/agents/embedded-agent-subscribe.subscribe-embedded-agent-session.waits-multiple-compaction-retries-before-resolving.test.ts @@ -1,5 +1,6 @@ // Compaction retry subscription tests cover retry wait accounting, compaction // event emission, abort-on-unsubscribe, and verbose tool summary behavior. +import type { AssistantMessage } from "openclaw/plugin-sdk/llm"; import { describe, expect, it, vi } from "vitest"; import { onAgentEvent } from "../infra/agent-events.js"; import { createSubscribedSessionHarness } from "./embedded-agent-subscribe.e2e-harness.js"; @@ -75,6 +76,24 @@ describe("subscribeEmbeddedAgentSession", () => { expect(subscription.getLastCompactionTokensAfter()).toBe(6_789); }); + it("clears the completed assistant when compaction schedules a retry", () => { + const { emit, subscription } = createSubscribedSessionHarness({ + runId: "run-compaction-assistant", + }); + const assistant = { + role: "assistant", + content: [{ type: "text", text: "Reply before compaction" }], + } as AssistantMessage; + + emit({ type: "message_end", message: assistant }); + expect(subscription.getCurrentAttemptAssistant()).toEqual(assistant); + expect(subscription.assistantTexts).toEqual(["Reply before compaction"]); + + emit({ type: "compaction_end", willRetry: true }); + expect(subscription.getCurrentAttemptAssistant()).toBeUndefined(); + expect(subscription.assistantTexts).toEqual([]); + }); + it("does not count compaction when result is absent", () => { const { emit, subscription } = createSubscribedSessionHarness({ runId: "run-compaction-no-result", diff --git a/src/agents/embedded-agent-subscribe.ts b/src/agents/embedded-agent-subscribe.ts index 2e495015a0a4..aaee6bc30015 100644 --- a/src/agents/embedded-agent-subscribe.ts +++ b/src/agents/embedded-agent-subscribe.ts @@ -14,6 +14,7 @@ import { createStreamingDirectiveAccumulator } from "../auto-reply/reply/streami import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../auto-reply/tokens.js"; import { formatToolAggregate } from "../auto-reply/tool-meta.js"; import { emitAgentEvent } from "../infra/agent-events.js"; +import type { AssistantMessage } from "../llm/types.js"; import { createSubsystemLogger } from "../logging/subsystem.js"; import { findFinalTagMatches } from "../shared/text/final-tags.js"; import { hasOrphanReasoningCloseBoundary } from "../shared/text/reasoning-tags.js"; @@ -254,6 +255,7 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess total: 0, }; let compactionCount = 0; + let currentAttemptAssistant: AssistantMessage | undefined; const assistantTexts = state.assistantTexts; const toolMetas = state.toolMetas; @@ -1280,6 +1282,9 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess state.deterministicApprovalPromptSent = false; state.lastDeliveredBlockReplyText = undefined; state.toolExecutionSinceLastBlockReply = false; + // A retry is a new model attempt. A silent retry must not inherit the + // completed assistant from the attempt that triggered compaction. + currentAttemptAssistant = undefined; state.replayState = mergeEmbeddedRunReplayState(state.replayState, params.initialReplayState); state.livenessState = "working"; resetAssistantMessageState(0); @@ -1290,6 +1295,13 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess state.lastAssistant = msg; } }; + const noteCompletedAssistant = (msg: AgentMessage) => { + if (msg?.role === "assistant") { + // Context-engine projection may later replace or mutate transcript + // objects. Final delivery needs the model event owned by this run. + currentAttemptAssistant = structuredClone(msg) as AssistantMessage; + } + }; const ctx: EmbeddedAgentSubscribeContext = { params, @@ -1301,6 +1313,7 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess builtinToolNames: params.builtinToolNames, trustedLocalMediaToolNames: params.trustedLocalMediaToolNames, noteLastAssistant, + noteCompletedAssistant, shouldEmitToolResult, shouldEmitToolOutput, emitToolSummary, @@ -1374,6 +1387,8 @@ export function subscribeEmbeddedAgentSession(params: SubscribeEmbeddedAgentSess return { assistantTexts, + getCurrentAttemptAssistant: () => + currentAttemptAssistant ? structuredClone(currentAttemptAssistant) : undefined, getLastAssistantTextMessageIndex: () => state.lastAssistantTextMessageIndex >= 0 ? state.lastAssistantTextMessageIndex : undefined, toolMetas,