fix(runner): fail the turn instead of answering with a lost conversation

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();

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(() => {});
   }
 

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<string>();
+
+/**
+ * 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

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 () => {