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 b7e7ed7e26..b0a3612e02 100644 --- a/services/runner/src/server.ts +++ b/services/runner/src/server.ts @@ -76,7 +76,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 @@ -1045,6 +1049,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 () => {