[fix] Last-message-only reconstruction: fix ordering, config wiring, and robustness

api/oss/src/utils/env.py

@@ -539,6 +539,7 @@ class AgentaConfig(BaseModel):
     otlp: OTLPConfig = OTLPConfig()
     redaction: RedactionConfig = RedactionConfig()
     services: ServicesConfig = ServicesConfig()
+    sessions: SessionsConfig = SessionsConfig()
     webhooks: WebhooksConfig = WebhooksConfig()
     workers: WorkersConfig = WorkersConfig()
 

api/oss/tests/pytest/unit/sessions/test_records_truncation.py

@@ -18,6 +18,15 @@ def _size(obj) -> int:
     return len(dumps(obj))
 
 
+def test_smart_truncation_flag_is_wired_into_agenta_config():
+    """The publish path reads `env.agenta.sessions.records.smart_truncation`. If `SessionsConfig`
+    is not attached to `AgentaConfig` this raises AttributeError inside the publish try/except and
+    silently drops the record — regression guard for that wiring."""
+    from oss.src.utils.env import env
+
+    assert isinstance(env.agenta.sessions.records.smart_truncation, bool)
+
+
 def test_under_budget_returns_unchanged():
     attrs = {"type": "message", "text": "hi"}
     out = _truncate_attributes(attrs, MAX_ATTRIBUTES_BYTES, _size(attrs))

docs/design/sessions-last-message-only/design.md

@@ -6,19 +6,32 @@ Shrink request/trace payloads by sending only the newest user message per turn a
 reconstructing prior conversation server-side from durable session **records**.
 
 ## Decisions (locked)
-- **Reconstruction lives in the runner (TS)** — at the `buildTurnText`/`priorMessages` seam,
-  reusing the runner's records client + session plumbing; a TS↔TS port of the FE
-  `transcriptToMessages.ts` folding. (Shared infra with JP/Mahmoud — keep additive/flagged.)
+- **Reconstruction lives in the runner (TS)** — reusing the runner's records client + session
+  plumbing; a TS↔TS port of the FE `transcriptToMessages.ts` folding. (Shared infra with
+  JP/Mahmoud — keep additive/flagged.) It runs in the session-owned run handler, BEFORE the turn's
+  prompt is persisted and before the keep-alive history fingerprint — both read `request.messages`,
+  so rebuilding afterwards double-counted the prompt and mismatched the warm fingerprint (fixed in
+  the stacked QA PR).
 - **Harden records durability FIRST**, before the FE trusts reconstruction — so reconstructed
   history is authoritative, not lossy.
 
 ## Current architecture (verified FE / SDK / runner / api)
-Two continuity paths already exist in the runner:
+Continuity in the runner has three cases, not two:
 - **Warm** (pool hit / same harness authored last turn): harness native `resumeSession()`
-  supplies history; `sendLastMessageOnly` is ALREADY true (`runtime-contracts.ts:168`).
-- **Cold** (evicted / different harness / cross-device): runner rebuilds the full transcript
-  from inbound `request.messages` (`transcript.ts:buildTurnText`/`priorMessages`; selected at
-  `run-turn.ts:134`).
+  supplies the model's context; `sendLastMessageOnly` is ALREADY true
+  (`runtime-contracts.ts:168`). The turn only needs the new user text.
+- **Cold, not evicted** (a parked sandbox is reused after a short gap): same as warm — native
+  resume still has the history.
+- **Cold-evicted** (the sandbox was torn down / different harness / cross-device): there is no
+  native session to resume, so the runner must rebuild the full transcript. Today it rebuilds
+  from inbound `request.messages` (`transcript.ts:buildTurnText`/`priorMessages`); with
+  last-message-only that history is no longer sent, so reconstruction from **records** is what
+  fills the gap. This is the only case where records-based reconstruction supplies model context.
+
+Reconstruction still runs on warm turns too, but only to realign the keep-alive history
+fingerprint (the FE sent one message, so the fingerprint's prior-conversation must be rebuilt to
+match what the previous park predicted). The model context on a warm turn still comes from native
+resume, not from records.
 
 The FE always sends full history (`agentRequest.ts:401`) because it can't predict warm/cold.
 Records are **write-only** from the run's perspective: the runner persists every event

hosting/docker-compose/ee/docker-compose.dev.yml

@@ -407,6 +407,12 @@ services:
             AGENTA_RUNNER_HOST: ${AGENTA_RUNNER_HOST:-0.0.0.0}
             AGENTA_RUNNER_TOKEN: ${AGENTA_RUNNER_TOKEN:?AGENTA_RUNNER_TOKEN is required}
             PI_CODING_AGENT_DIR: /pi-agent
+            # Sessions: last-message-only + server-side history reconstruction. The runner reads
+            # these from process.env; because this service has no env_file, they must be forwarded
+            # explicitly here or the feature is a silent no-op. All default OFF (empty → disabled).
+            AGENTA_RECORDS_DURABLE: ${AGENTA_RECORDS_DURABLE:-}
+            AGENTA_SESSIONS_RECONSTRUCT: ${AGENTA_SESSIONS_RECONSTRUCT:-}
+            AGENTA_RECORDS_INGEST_MAX_RETRIES: ${AGENTA_RECORDS_INGEST_MAX_RETRIES:-}
             # The API as reached FROM this container — by its compose service name, over the
             # network (the same `http://api:8000` the workers use), NOT the public
             # `http://localhost/api` (which doesn't resolve inside a container). Used by the

hosting/docker-compose/oss/docker-compose.dev.yml

@@ -392,6 +392,12 @@ services:
             # default of `127.0.0.1` (same-host only) is unreachable cross-container.
             AGENTA_RUNNER_HOST: ${AGENTA_RUNNER_HOST:-0.0.0.0}
             PI_CODING_AGENT_DIR: /pi-agent
+            # Sessions: last-message-only + server-side history reconstruction. The runner reads
+            # these from process.env; because this service has no env_file, they must be forwarded
+            # explicitly here or the feature is a silent no-op. All default OFF (empty → disabled).
+            AGENTA_RECORDS_DURABLE: ${AGENTA_RECORDS_DURABLE:-}
+            AGENTA_SESSIONS_RECONSTRUCT: ${AGENTA_SESSIONS_RECONSTRUCT:-}
+            AGENTA_RECORDS_INGEST_MAX_RETRIES: ${AGENTA_RECORDS_INGEST_MAX_RETRIES:-}
             # API as reached FROM this container: the compose service name over the network
             # (http://api:8000, as the workers use), NOT the public http://localhost/api
             # which does not resolve inside a container. Used by the OTLP tracing-export

services/runner/src/engines/sandbox_agent/reconstruct-history.ts

@@ -3,9 +3,13 @@
  * 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: when the client still sends the whole conversation
- * (`messages.length > 1`), reconstruction is skipped and behaviour is unchanged. Best-effort — any
- * miss (no session, no records, fetch failure) leaves the inbound history untouched.
+ * client sent exactly its trailing user message: a full inbound history (more than one message, or
+ * a non-user tail) is left untouched. Best-effort — any miss (no session, no records, fetch
+ * failure) leaves the inbound history untouched.
+ *
+ * Called from the server handler BEFORE the turn's prompt is persisted and before the keep-alive
+ * history fingerprint, so the record log holds only prior turns (no self-duplication) and the
+ * fingerprint sees the same full history a full-history client would have sent.
  */
 
 import type { AgentRunRequest } from "../../protocol.ts";
@@ -33,8 +37,11 @@ export async function reconstructHistoryIfNeeded(
 ): Promise<AgentRunRequest | null> {
   if (!reconstructEnabled() || !sessionId) return null;
   const inbound = request.messages ?? [];
-  // The client already sent the conversation — nothing to rebuild.
-  if (inbound.length > 1) return null;
+  // Reconstruct only for a fresh turn that is exactly its trailing user message. A full inbound
+  // history (client sent the conversation) needs no rebuild; an empty or assistant-only inbound
+  // (e.g. an approval resume) must not be rebuilt, or `resolvePromptText` could replay a historical
+  // prompt as the current turn.
+  if (inbound.length !== 1 || inbound[0]?.role !== "user") return null;
 
   const records = await fetchSessionRecords(sessionId, auth);
   if (!records || records.length === 0) return null;

services/runner/src/engines/sandbox_agent/run-turn.ts

@@ -75,16 +75,15 @@ import {
   nextTurnIndex,
   sessionContinuityStore,
 } from "./session-continuity.ts";
-import { reconstructHistoryIfNeeded } from "./reconstruct-history.ts";
-import { buildTurnText, priorMessages } from "./transcript.ts";
+import { priorMessages } from "./transcript.ts";
 import { resolveRunUsage } from "./usage.ts";
 
 /**
  * Run one turn against an acquired environment: start a fresh otel run, wire this turn's pause
  * controller / decisions / responder into `env.currentTurn`, restart the tool relay,
  * send the prompt, resolve usage, and finish + flush the trace. It does NOT tear down the
  * environment (the caller owns `env.destroy`). On a continuation the prompt is only the new user
- * text (`buildTurnText` does not run); on a cold turn it is `plan.turnText`, exactly as before.
+ * text; on a cold turn it is `plan.turnText`, exactly as before.
  */
 export async function runTurn(
   env: SessionEnvironment,
@@ -145,28 +144,12 @@ export async function runTurn(
   });
 
   try {
-    // Server-side history reconstruction (flag-gated, no-op by default): when the client sent a
-    // minimal history, rebuild prior turns from the durable record log so a cold turn still has
-    // full context. Runs before the current user turn is persisted, so records hold only prior
-    // turns. Reassigns `request` so every downstream reader (turnText, priorMessages, responder,
-    // otel) sees the same reconstructed history.
-    const reconstructed = await reconstructHistoryIfNeeded(
-      request,
-      sessionId,
-      () => runCredential(request),
-      logger,
-    );
-    if (reconstructed) request = reconstructed;
-
+    // History reconstruction (last-message-only) happens upstream in the server handler, BEFORE the
+    // prompt is persisted and the keep-alive fingerprint runs, so `request` already carries the full
+    // conversation here and `plan` (built during acquireEnvironment) reflects it.
     const promptText = resolvePromptText(request);
-    // Cold: replay the full transcript. Continuation or loaded: send only new text. When history
-    // was rebuilt from records, recompute the transcript from it — the prebuilt plan.turnText
-    // predates the reconstruction.
-    const turnText = sendLastMessageOnly(opts)
-      ? promptText
-      : reconstructed
-        ? buildTurnText(request, logger)
-        : plan.turnText;
+    // Cold: replay the full transcript (`plan.turnText`). Continuation or loaded: send only new text.
+    const turnText = sendLastMessageOnly(opts) ? promptText : plan.turnText;
 
     const run = (deps.createOtel ?? createSandboxAgentOtel)({
       harness: plan.harness,

services/runner/src/server.ts

@@ -43,6 +43,7 @@ import {
   type SessionEnvironment,
 } from "./engines/sandbox_agent.ts";
 import type { MountCredentials } from "./engines/sandbox_agent/mount.ts";
+import { reconstructHistoryIfNeeded } from "./engines/sandbox_agent/reconstruct-history.ts";
 import type { TeardownReason } from "./engines/sandbox_agent/teardown.ts";
 import {
   approvalDecisionForToolCall,
@@ -960,6 +961,21 @@ async function runAndStreamWithApiBaseResolved(
   let aliveWatchdog: { release: () => Promise<void> } | undefined;
 
   if (sessionOwned) {
+    // Rebuild prior turns from the durable record log BEFORE this turn's prompt is persisted and
+    // BEFORE the keep-alive history fingerprint runs — both read `request.messages`. Reconstructing
+    // here (not inside runTurn) is what keeps last-message-only equivalent to a full-history client:
+    // at this point the record log holds only PRIOR turns, so the rebuilt history plus the inbound
+    // tail is the full conversation exactly once (no self-duplicated prompt), and the warm-
+    // continuation fingerprint sees the same full history it predicted at the previous park (no
+    // spurious evict to cold). Flag-gated and a no-op unless the client sent a minimal history.
+    const reconstructed = await reconstructHistoryIfNeeded(
+      request,
+      sessionId,
+      () => runCredential(request),
+      (msg) => process.stderr.write(`${msg}\n`),
+    );
+    if (reconstructed) request = reconstructed;
+
     // The request's api base (if any) is already scoped for this call via
     // runWithRequestApiBase in the outer runAndStream — apiBase() below sees it.
     // The runner authenticates session calls AS the invoke caller (the run credential),

services/runner/src/sessions/persist.ts

@@ -392,10 +392,20 @@ export function buildPersistingEmitter(
     );
   };
 
-  const flush = (): Promise<void> => {
+  const flush = async (): Promise<void> => {
     // A paused call ends the turn with its slot still open — persist it before draining.
     flushOpenTool();
-    return drainPersist(sessionId);
+    await drainPersist(sessionId);
+    // Consume the drop signal at the turn-end drain: records that exhausted retries mean the durable
+    // log is incomplete, so next turn's reconstruction may be missing context. Reading here also
+    // clears the per-session counter so it can't accumulate unread. (Only ever non-zero in durable
+    // mode.)
+    const dropped = takePersistFailures(sessionId);
+    if (dropped > 0) {
+      log(
+        `WARN session=${sessionId} durable log incomplete: ${dropped} record(s) dropped this turn; reconstruction may lack context`,
+      );
+    }
   };
 
   return { emit, persist, flush };

services/runner/src/sessions/records-query.ts

@@ -11,6 +11,14 @@ function log(msg: string): void {
   process.stderr.write(`[sessions/records-query] ${msg}\n`);
 }
 
+/** Bound the reconstruction fetch: it sits on the turn's critical path (the prompt waits on it),
+ * so a stalled query must fail fast to the inbound-history fallback rather than hang on Undici's
+ * long request-level timeout. Env-overridable for ops tuning. */
+function queryTimeoutMs(): number {
+  const n = Number(process.env.AGENTA_SESSIONS_RECORDS_QUERY_TIMEOUT_MS);
+  return Number.isFinite(n) && n > 0 ? Math.floor(n) : 5000;
+}
+
 /**
  * Fetch a session's durable record log, ordered for reconstruction (the endpoint returns records
  * by ingest time, then per-turn `record_index`). Returns `null` on failure so the caller can fall
@@ -29,6 +37,7 @@ export async function fetchSessionRecords(
         authorization: auth(),
       },
       body: JSON.stringify({ session_id: sessionId }),
+      signal: AbortSignal.timeout(queryTimeoutMs()),
     });
     if (!res.ok) throw new Error(`HTTP ${res.status}`);
     const body = (await res.json()) as { records?: SessionRecordRow[] };

services/runner/tests/unit/session-persist.test.ts

@@ -482,17 +482,48 @@ describe("durable records (AGENTA_RECORDS_DURABLE)", () => {
     vi.stubEnv("AGENTA_RECORDS_DURABLE", "true");
     vi.stubEnv("AGENTA_RECORDS_INGEST_MAX_RETRIES", "2"); // keep the test fast
     fetchFailCount = 99;
-    const { emit, flush } = buildPersistingEmitter("sess-durable", () => "t");
+    const { emit } = buildPersistingEmitter("sess-durable", () => "t");
     emit({ type: "message", text: "x" });
     emit({ type: "done" });
-    await flush();
+    // Drain directly (not flush): flush's turn-end consumer would read + clear the count.
+    await drainPersist("sess-durable");
 
     assert.equal(postedBodies.length, 0); // both records dropped
     assert.equal(takePersistFailures("sess-durable"), 2);
     // take clears the count.
     assert.equal(takePersistFailures("sess-durable"), 0);
   });
 
+  it("durable: the turn-end flush consumes the drop count (warns + clears)", async () => {
+    vi.stubEnv("AGENTA_RECORDS_DURABLE", "true");
+    vi.stubEnv("AGENTA_RECORDS_INGEST_MAX_RETRIES", "2");
+    fetchFailCount = 99;
+    const warns: string[] = [];
+    const writeSpy = vi
+      .spyOn(process.stderr, "write")
+      .mockImplementation((chunk: string | Uint8Array) => {
+        warns.push(String(chunk));
+        return true;
+      });
+    try {
+      const { emit, flush } = buildPersistingEmitter("sess-flush", () => "t");
+      emit({ type: "message", text: "x" });
+      await flush();
+
+      // The drain surfaced the incomplete durable log at turn end...
+      assert.equal(
+        warns.some(
+          (w) => w.includes("sess-flush") && w.includes("durable log incomplete"),
+        ),
+        true,
+      );
+    } finally {
+      writeSpy.mockRestore();
+    }
+    // ...and cleared the counter, so nothing accumulates unread.
+    assert.equal(takePersistFailures("sess-flush"), 0);
+  });
+
   it("durable: a transient failure recovers within the retry budget (no drop counted)", async () => {
     vi.stubEnv("AGENTA_RECORDS_DURABLE", "true");
     vi.stubEnv("AGENTA_RECORDS_INGEST_MAX_RETRIES", "5");

services/runner/tests/unit/session-reconstruct-history.test.ts

@@ -47,6 +47,23 @@ describe("reconstructHistoryIfNeeded", () => {
     assert.equal(fetchCalls, 0);
   });
 
+  it("no-op when the inbound has no messages (never queries)", async () => {
+    vi.stubEnv("AGENTA_SESSIONS_RECONSTRUCT", "true");
+    const req = { messages: [] } as never;
+    const out = await reconstructHistoryIfNeeded(req, "sess-1", auth);
+    assert.equal(out, null);
+    assert.equal(fetchCalls, 0);
+  });
+
+  it("no-op when the only inbound message is not a user turn (approval resume)", async () => {
+    vi.stubEnv("AGENTA_SESSIONS_RECONSTRUCT", "true");
+    // A resume's tail is a tool_result envelope; rebuilding here could replay a historical prompt.
+    const req = { messages: [{ role: "assistant", content: "..." }] } as never;
+    const out = await reconstructHistoryIfNeeded(req, "sess-1", auth);
+    assert.equal(out, null);
+    assert.equal(fetchCalls, 0);
+  });
+
   it("no-op when there is no session id", async () => {
     vi.stubEnv("AGENTA_SESSIONS_RECONSTRUCT", "true");
     const req = { messages: [userTurn] } as never;