From f621400e59e4466af194606344dc003e7e32074f Mon Sep 17 00:00:00 2001 From: Mahmoud Mabrouk Date: Fri, 24 Jul 2026 23:52:33 +0200 Subject: [PATCH] fix(runner): fail the turn instead of answering with a lost conversation Two silent failure modes met in the same place. `fetchSessionRecords` returned null on any error and the caller treated that as "keep the inbound history"; once the client sends only its last message, that history is one message, so an unreadable log produced an agent that answered as if the user had just arrived, with no error anywhere on the wire. Separately, `takePersistFailures` had no caller outside the tests, so a dropped record was counted into a map nothing read, and the next turn reconstructed from a log with a hole in it. Consume the drop count at the turn-end drain, where the docstring already said it happened, and mark the session. Reconstruction now fails the turn for an unreadable log or a session known to have lost a record, so the caller sees an error rather than a confident wrong answer. Requests that still send their own history are unaffected: they never reach this path. Claude-Session: https://claude.ai/code/session_01KM69J7uHafgciiN5zfG7qR --- .../sandbox_agent/reconstruct-history.ts | 24 ++++++++++++++++--- services/runner/src/server.ts | 19 ++++++++++++++- services/runner/src/sessions/persist.ts | 18 ++++++++++++++ .../unit/session-reconstruct-history.test.ts | 20 +++++++++++++--- 4 files changed, 74 insertions(+), 7 deletions(-) diff --git a/services/runner/src/engines/sandbox_agent/reconstruct-history.ts b/services/runner/src/engines/sandbox_agent/reconstruct-history.ts index ba2e0c25bd..208c16dc44 100644 --- a/services/runner/src/engines/sandbox_agent/reconstruct-history.ts +++ b/services/runner/src/engines/sandbox_agent/reconstruct-history.ts @@ -3,8 +3,12 @@ * trusting a full inbound history — the server side of "client sends only the last message". * * Flag-gated (`AGENTA_SESSIONS_RECONSTRUCT`) and a strict no-op until BOTH the flag is on AND the - * client actually sent a minimal history (`carriesMinimalHistory`). Best-effort — any miss (no - * session, no records, fetch failure) leaves the inbound history untouched. + * client actually sent a minimal history (`carriesMinimalHistory`). When it does not apply, the + * inbound history is left untouched. + * + * When it DOES apply it is no longer best-effort, because the client kept no copy of the + * conversation: an unreadable log, or one known to have dropped a record, fails the turn rather + * than letting the agent answer as though the conversation had just started. * * The record log already contains the CURRENT turn by the time this runs: the runner persists the * inbound user message before it starts the engine, and acquiring a sandbox takes seconds. Its @@ -14,6 +18,7 @@ import type { AgentRunRequest } from "../../protocol.ts"; import { fetchSessionRecords } from "../../sessions/records-query.ts"; +import { recordsIncomplete } from "../../sessions/persist.ts"; import { reconstructMessages } from "../../sessions/reconstruct.ts"; import { carriesMinimalHistory } from "./session-identity.ts"; @@ -38,8 +43,21 @@ export async function reconstructHistoryIfNeeded( // The client still asserts the conversation itself — nothing to rebuild. if (!carriesMinimalHistory(request)) return null; + // The client kept no copy of the conversation, so there is no history to fall back to. Answering + // anyway would silently produce an agent that forgot everything, which reads as a correct reply. + // Fail the turn instead: `runTurn`'s catch turns this into an error result the caller can see. + if (recordsIncomplete(sessionId)) { + throw new Error( + `session ${sessionId} lost a durable record; refusing to rebuild an incomplete conversation`, + ); + } + const records = await fetchSessionRecords(sessionId, auth); - if (!records) return null; + if (!records) { + throw new Error( + `session ${sessionId} record log is unreadable; cannot rebuild the conversation`, + ); + } // Drop this turn's own records: the inbound message already carries the current prompt. const currentTurnId = request.turnId?.trim(); diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index 44208229f5..cb24527116 100644 --- a/services/runner/src/server.ts +++ b/services/runner/src/server.ts @@ -75,7 +75,11 @@ import { isEntrypoint } from "./entry.ts"; import { insecureEgressAllowed } from "./tools/ssrf-guard.ts"; import { startAliveWatchdog } from "./sessions/alive.ts"; import { cancelStaleInteractions } from "./sessions/interactions.ts"; -import { buildPersistingEmitter } from "./sessions/persist.ts"; +import { + buildPersistingEmitter, + noteRecordsIncomplete, + takePersistFailures, +} from "./sessions/persist.ts"; import { seedForRun } from "./redaction.ts"; // Server binding (host/port) comes from the typed `RunnerConfig` resolved at boot. The host @@ -1038,6 +1042,19 @@ async function runAndStreamWithApiBaseResolved( if (flushPersist) await flushPersist().catch(() => {}); result = { ok: false, error: message }; } finally { + // The drain is the only place that knows whether this turn's records all landed. A dropped + // record means the log no longer represents the conversation, so mark the session: a later + // turn must fail rather than rebuild model context from a log with a hole in it. + if (sessionOwned && sessionId) { + const dropped = takePersistFailures(sessionId); + if (dropped > 0) { + noteRecordsIncomplete(sessionId); + process.stderr.write( + `[sessions] records INCOMPLETE session=${sessionId} dropped=${dropped}; ` + + `reconstruction disabled for this session\n`, + ); + } + } if (aliveWatchdog) await aliveWatchdog.release().catch(() => {}); } diff --git a/services/runner/src/sessions/persist.ts b/services/runner/src/sessions/persist.ts index 8feae4396b..a12ac3ce51 100644 --- a/services/runner/src/sessions/persist.ts +++ b/services/runner/src/sessions/persist.ts @@ -184,6 +184,24 @@ export function takePersistFailures(sessionId: string): number { return n; } +/** Sessions whose record log is known to have lost at least one record. */ +const incompleteSessions = new Set(); + +/** + * Mark a session's record log as incomplete, permanently for this process. Once a record is + * dropped the log no longer represents the conversation, so it must never be used to rebuild + * model context: the turn would silently run with a hole in its history. Set at the turn-end + * drain; read by the reconstruction seam, which fails the turn instead of reconstructing. + */ +export function noteRecordsIncomplete(sessionId: string): void { + incompleteSessions.add(sessionId); +} + +/** Whether this session has lost a record and can no longer be reconstructed from. */ +export function recordsIncomplete(sessionId: string): boolean { + return incompleteSessions.has(sessionId); +} + /** * A tool call streams as many `tool_call` events with a growing partial-args snapshot for * one id. Idle window after which an open, un-closed tool call is flushed as-is — the diff --git a/services/runner/tests/unit/session-reconstruct-history.test.ts b/services/runner/tests/unit/session-reconstruct-history.test.ts index c8369832ed..084c09e8b0 100644 --- a/services/runner/tests/unit/session-reconstruct-history.test.ts +++ b/services/runner/tests/unit/session-reconstruct-history.test.ts @@ -20,6 +20,7 @@ vi.stubGlobal("fetch", async () => { const { reconstructHistoryIfNeeded } = await import( "../../src/engines/sandbox_agent/reconstruct-history.ts" ); +const { noteRecordsIncomplete } = await import("../../src/sessions/persist.ts"); const auth = () => "Secret t"; const userTurn = { role: "user", content: "hi again" }; @@ -63,12 +64,25 @@ describe("reconstructHistoryIfNeeded", () => { assert.equal(out, null); }); - it("no-op (falls back) when the records fetch fails", async () => { + it("fails the turn when the records fetch fails (the client kept no history to fall back to)", async () => { vi.stubEnv("AGENTA_SESSIONS_RECONSTRUCT", "true"); fetchShouldFail = true; const req = { messages: [userTurn] } as never; - const out = await reconstructHistoryIfNeeded(req, "sess-1", auth); - assert.equal(out, null); + await assert.rejects( + () => reconstructHistoryIfNeeded(req, "sess-1", auth), + /unreadable/, + ); + }); + + it("fails the turn when the session is known to have dropped a record", async () => { + vi.stubEnv("AGENTA_SESSIONS_RECONSTRUCT", "true"); + noteRecordsIncomplete("sess-dropped"); + const req = { messages: [userTurn] } as never; + await assert.rejects( + () => reconstructHistoryIfNeeded(req, "sess-dropped", auth), + /incomplete conversation/, + ); + assert.equal(fetchCalls, 0, "no query when the log is already known bad"); }); it("prepends reconstructed prior turns to the inbound message when enabled", async () => {