diff --git a/services/runner/src/engines/sandbox_agent/run-turn.ts b/services/runner/src/engines/sandbox_agent/run-turn.ts index 9190d90854..bbae6d372b 100644 --- a/services/runner/src/engines/sandbox_agent/run-turn.ts +++ b/services/runner/src/engines/sandbox_agent/run-turn.ts @@ -854,7 +854,7 @@ export async function runTurn( run.emitEvent({ type: "error", message: swallowedError }); } - const output = run.finish(); + const output = run.finish(stopReason); await run.flush(); const turnEndedAt = new Date().toISOString(); diff --git a/services/runner/src/engines/sandbox_agent/transcript.ts b/services/runner/src/engines/sandbox_agent/transcript.ts index db5851aea6..d3a2254a6a 100644 --- a/services/runner/src/engines/sandbox_agent/transcript.ts +++ b/services/runner/src/engines/sandbox_agent/transcript.ts @@ -6,6 +6,10 @@ import { resolvePromptText, } from "../../protocol.ts"; import { approvalDecisionOf } from "../../responder.ts"; +import { + APPROVED_EXECUTION_RESULT_UNKNOWN, + DEFERRED_NOT_EXECUTED_PREFIX, +} from "../../tracing/otel.ts"; export type ApprovalRenderHint = "executed" | "lastPending" | "stalePending"; @@ -131,6 +135,25 @@ function capToolResultBody(body: string): string { return `${body.slice(0, TOOL_RESULT_RENDER_MAX_CHARS)} [... ${omitted} chars omitted]`; } +/** + * A pause-terminalization sentinel is runner bookkeeping written when a turn parks, NOT a tool + * failure. Left on the generic `[ error: …]` path the model reads "error" as a refusal — the + * parallel-approval bug (turn 6d34b1ea) — so render each as an explicit, non-error nudge. The two + * carry OPPOSITE guidance: a deferred call never ran and must be re-issued; an approved call may + * already have run and must NOT be retried. Returns undefined for any other tool result. + */ +function pauseSentinelNudge(block: ContentBlock): string | undefined { + if (typeof block.output !== "string") return undefined; + const toolName = block.toolName ?? "tool"; + if (block.output.startsWith(DEFERRED_NOT_EXECUTED_PREFIX)) { + return `[${toolName} was NOT run — the turn paused for another approval first, so it was skipped, not denied. Call ${toolName} again with the same arguments now to run it.]`; + } + if (block.output === APPROVED_EXECUTION_RESULT_UNKNOWN) { + return `[${toolName} was approved and may have already run; its result was not observed before the pause ended the turn. Do NOT assume it failed and do NOT retry a side-effecting call.]`; + } + return undefined; +} + /** * Render one message for the replayed transcript, including resolved tool turns. Under * the cold model, ACP prompt content blocks cannot carry tool calls/results, so resolved @@ -168,10 +191,15 @@ export function messageTranscript( parts.push(`[user DENIED ${toolName}; the call was not executed.]`); } } else { - const body = capToolResultBody(safeJson(block.output)); - parts.push( - `[${block.toolName ?? "tool"} ${block.isError ? "error" : "returned"}: ${body}]`, - ); + const nudge = pauseSentinelNudge(block); + if (nudge !== undefined) { + parts.push(nudge); + } else { + const body = capToolResultBody(safeJson(block.output)); + parts.push( + `[${block.toolName ?? "tool"} ${block.isError ? "error" : "returned"}: ${body}]`, + ); + } } } else if (block.type === "image") { parts.push("[image]"); diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index 93beeb2591..44208229f5 100644 --- a/services/runner/src/server.ts +++ b/services/runner/src/server.ts @@ -1006,11 +1006,14 @@ async function runAndStreamWithApiBaseResolved( turnId, request.runContext?.trace?.span_id, ); - // Record the inbound user turn first so the session record is the full conversation, - // not just agent output. Interaction replies ride tool_result blocks (no text) and are - // already recorded on the interaction, so an empty prompt persists nothing. + // Record the inbound user turn first so the session record is the full conversation, not just + // agent output. Guard on `tailIsFreshUserMessage`: an approval RESUME's tail is the tool_result + // envelope (no text), so `resolvePromptText` falls back to the ORIGINAL prompt and would + // re-persist it as a DUPLICATE user row. The guard writes the prompt only on the turn that first + // introduced it — a real new turn's tail IS a fresh user message; a resume's is not. const promptText = resolvePromptText(request); - if (promptText) persist({ type: "message", text: promptText }, "user"); + if (promptText && tailIsFreshUserMessage(request)) + persist({ type: "message", text: promptText }, "user"); emitFn = persistingEmit; flushPersist = flush; persistError = (message) => persist({ type: "error", message }, "agent"); diff --git a/services/runner/src/tracing/otel.ts b/services/runner/src/tracing/otel.ts index f446286671..d663b3249a 100644 --- a/services/runner/src/tracing/otel.ts +++ b/services/runner/src/tracing/otel.ts @@ -1066,8 +1066,9 @@ export interface SandboxAgentOtel { * lands in both the live sink and the batch `events()` log in build order. */ emitEvent(event: AgentEvent): void; - /** End all open spans. Returns the accumulated assistant text. */ - finish(): string; + /** End all open spans. Returns the accumulated assistant text. `stopReason` (e.g. "paused") + * is stamped on the terminal `done` record so hydration can distinguish a pause from a boundary. */ + finish(stopReason?: string): string; /** * Record a run-level error on the agent span: the user-facing message (F-030) plus the * provider that failed, and an OTel exception event, so a trace carries the same diagnostic @@ -1554,7 +1555,7 @@ export function createSandboxAgentOtel( } } - function finish(): string { + function finish(stopReason?: string): string { const text = stripStartupBanner(accumulated.trim()); // The event log is independent of span emission, so build its tail either way. closeText(); @@ -1578,7 +1579,13 @@ export function createSandboxAgentOtel( } // Stamp the run's trace id on the turn's terminal event so a persisted transcript can link a // replayed turn back to its trace (undefined only in span-less mode with no valid traceparent). - record({ type: "done", ...(runTraceId ? { traceId: runTraceId } : {}) }); + // Mark a paused turn's terminal record so a cold reload can tell a pause from a real turn + // boundary (the FE adoption heuristic and hydration read this). A completed turn omits it. + record({ + type: "done", + ...(stopReason === "paused" ? { stopReason: "paused" } : {}), + ...(runTraceId ? { traceId: runTraceId } : {}), + }); if (!emitSpans) return text; if (llmSpan) { emitMessages( diff --git a/services/runner/tests/unit/transcript.test.ts b/services/runner/tests/unit/transcript.test.ts index 9c26d817bc..c57b5b709d 100644 --- a/services/runner/tests/unit/transcript.test.ts +++ b/services/runner/tests/unit/transcript.test.ts @@ -7,6 +7,10 @@ import { TOOL_RESULT_RENDER_MAX_CHARS, } from "../../src/engines/sandbox_agent/transcript.ts"; import type { AgentRunRequest, ContentBlock } from "../../src/protocol.ts"; +import { + APPROVED_EXECUTION_RESULT_UNKNOWN, + TOOL_NOT_EXECUTED_PAUSED, +} from "../../src/tracing/otel.ts"; const COMMIT_TOOL = "mcp__agenta-tools__commit_revision"; const OTHER_TOOL = "mcp__agenta-tools__summarize"; @@ -103,6 +107,41 @@ describe("messageTranscript", () => { assert.match(transcript, /APPROVED approval_status/); assert.match(transcript, /NOT run yet/); }); + + it("renders a deferred sibling as a retry nudge, not an error", () => { + const transcript = messageTranscript([ + { + type: "tool_result", + toolName: OTHER_TOOL, + output: TOOL_NOT_EXECUTED_PAUSED, + isError: true, + }, + ]); + + assert.match(transcript, new RegExp(`${OTHER_TOOL} was NOT run`)); + assert.match(transcript, /skipped, not denied/); + assert.match(transcript, /Call .* again with the same arguments now/); + // The whole point: it must not read as a tool error/refusal, and must not leak the sentinel. + assert.doesNotMatch(transcript, /error:/); + assert.doesNotMatch(transcript, /DEFERRED_NOT_EXECUTED/); + }); + + it("renders an approved-but-unobserved result as a do-not-retry nudge", () => { + const transcript = messageTranscript([ + { + type: "tool_result", + toolName: OTHER_TOOL, + output: APPROVED_EXECUTION_RESULT_UNKNOWN, + isError: true, + }, + ]); + + assert.match(transcript, new RegExp(`${OTHER_TOOL} was approved`)); + assert.match(transcript, /may have already run/); + assert.match(transcript, /do NOT retry a side-effecting call/i); + assert.doesNotMatch(transcript, /error:/); + assert.doesNotMatch(transcript, /APPROVED_EXECUTION_RESULT_UNKNOWN/); + }); }); describe("tool result render cap", () => { diff --git a/web/oss/src/components/AgentChatSlice/AgentConversation.tsx b/web/oss/src/components/AgentChatSlice/AgentConversation.tsx index 0fc3d7f26b..709f041963 100644 --- a/web/oss/src/components/AgentChatSlice/AgentConversation.tsx +++ b/web/oss/src/components/AgentChatSlice/AgentConversation.tsx @@ -969,7 +969,26 @@ const AgentConversation = ({ loadSessionMessages(sessionId).then((serverMsgs) => { if (cancelled || !serverMsgs || serverMsgs.length === 0) return const prev = messagesRef.current - if (busyRef.current || serverMsgs.length <= prev.length) return + if (busyRef.current) return + // Adopt the server transcript when it is strictly ahead by count, OR when our LOCAL tail + // is stuck paused (mid-approval) while the server has moved past it to a terminal turn — a + // resume that completed on another device. Count alone misses the latter (same bubble + // count) and was silently propped up by the now-removed duplicate user row; the server's + // `paused` flag rides the runner's `done.stopReason` through `transcriptToMessages`. + const serverAheadByCount = serverMsgs.length > prev.length + const localTailPaused = getPendingApprovals(prev).length > 0 + const serverTail = serverMsgs[serverMsgs.length - 1] as + | {role?: string; metadata?: {paused?: boolean}} + | undefined + // The paused-tail exception adopts a resume that completed elsewhere, so the server must + // NOT be behind (>= guards against a lagging snapshot discarding newer local approval + // state) and its tail must be a finished assistant turn — not a shorter, older stream. + const serverTailComplete = + serverMsgs.length >= prev.length && + serverTail?.role === "assistant" && + !serverTail.metadata?.paused && + getPendingApprovals(serverMsgs).length === 0 + if (!serverAheadByCount && !(localTailPaused && serverTailComplete)) return serverMsgs.forEach((m) => { seenIdsRef.current.add(m.id) restoredIdsRef.current.add(m.id) diff --git a/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts b/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts index 840b0a2da5..c0f1f2d152 100644 --- a/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts +++ b/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.test.ts @@ -216,3 +216,23 @@ describe("transcriptToMessages approval hydration", () => { }) }) }) + +describe("transcriptToMessages paused end-marker", () => { + it("flags the message whose turn ended paused (done.stopReason)", () => { + const messages = transcriptToMessages([ + ...approvalRecords(), + record("record-done-paused", {type: "done", stopReason: "paused"}), + ]) + expect(messages).not.toBeNull() + expect(messages?.[0].metadata).toMatchObject({paused: true}) + }) + + it("does not flag a normally completed turn", () => { + const messages = transcriptToMessages([ + ...approvalRecords(), + record("record-done-complete", {type: "done"}), + ]) + expect(messages).not.toBeNull() + expect((messages?.[0].metadata as {paused?: boolean} | undefined)?.paused).toBeUndefined() + }) +}) diff --git a/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.ts b/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.ts index 6483f57397..76627ed686 100644 --- a/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.ts +++ b/web/oss/src/components/AgentChatSlice/assets/transcriptToMessages.ts @@ -38,6 +38,9 @@ interface DraftMessage { traceId?: string /** Token/cost totals from the turn's persisted `usage` event, in the raw stream shape. */ usage?: {input?: number; output?: number; total?: number; cost?: number} + /** The turn's terminal `done` carried `stopReason:"paused"` — it ended mid-approval, not at a + * real boundary. Surfaced on the message so a cold reload's adoption heuristic can compare state. */ + paused?: boolean } interface TranscriptIndex { @@ -280,6 +283,9 @@ export function transcriptToMessages(records: SessionRecord[]): UIMessage[] | nu // fresh message per turn. if (row.session_update === "done" || p.type === "done") { if (current && traceId && !current.traceId) current.traceId = traceId + // A paused turn's `done` carries stopReason:"paused" — mark the closing message so a cold + // reload can tell this turn ended mid-approval, not at a real boundary (read by adoption). + if (current && p.stopReason === "paused") current.paused = true current = null continue } @@ -301,6 +307,7 @@ export function transcriptToMessages(records: SessionRecord[]): UIMessage[] | nu const metadata: Record = {} if (d.traceId) metadata.traceId = d.traceId if (d.usage) metadata.usage = d.usage + if (d.paused) metadata.paused = true return { id: d.id, role: d.role,