diff --git a/api/oss/src/utils/env.py b/api/oss/src/utils/env.py index beef7e70e6..4f22a5958a 100644 --- a/api/oss/src/utils/env.py +++ b/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() diff --git a/api/oss/tests/pytest/unit/sessions/test_records_truncation.py b/api/oss/tests/pytest/unit/sessions/test_records_truncation.py index f183c77798..80644b36b1 100644 --- a/api/oss/tests/pytest/unit/sessions/test_records_truncation.py +++ b/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)) diff --git a/docs/design/sessions-last-message-only/design.md b/docs/design/sessions-last-message-only/design.md index 081a380a7f..3074d52b27 100644 --- a/docs/design/sessions-last-message-only/design.md +++ b/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 diff --git a/hosting/docker-compose/ee/docker-compose.dev.yml b/hosting/docker-compose/ee/docker-compose.dev.yml index b26c27e0cf..e3c081b9ba 100644 --- a/hosting/docker-compose/ee/docker-compose.dev.yml +++ b/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 diff --git a/hosting/docker-compose/oss/docker-compose.dev.yml b/hosting/docker-compose/oss/docker-compose.dev.yml index 604d62bfa1..3ea341206f 100644 --- a/hosting/docker-compose/oss/docker-compose.dev.yml +++ b/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 diff --git a/services/runner/src/engines/sandbox_agent/reconstruct-history.ts b/services/runner/src/engines/sandbox_agent/reconstruct-history.ts index 1a3e0a4766..d38419b883 100644 --- a/services/runner/src/engines/sandbox_agent/reconstruct-history.ts +++ b/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 { 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; diff --git a/services/runner/src/engines/sandbox_agent/run-turn.ts b/services/runner/src/engines/sandbox_agent/run-turn.ts index fa6875385d..fe2408faa4 100644 --- a/services/runner/src/engines/sandbox_agent/run-turn.ts +++ b/services/runner/src/engines/sandbox_agent/run-turn.ts @@ -75,8 +75,7 @@ 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"; /** @@ -84,7 +83,7 @@ import { resolveRunUsage } from "./usage.ts"; * 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, diff --git a/services/runner/src/server.ts b/services/runner/src/server.ts index 44208229f5..5010d538d5 100644 --- a/services/runner/src/server.ts +++ b/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 } | 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), diff --git a/services/runner/src/sessions/persist.ts b/services/runner/src/sessions/persist.ts index 8feae4396b..33b441fe69 100644 --- a/services/runner/src/sessions/persist.ts +++ b/services/runner/src/sessions/persist.ts @@ -392,10 +392,20 @@ export function buildPersistingEmitter( ); }; - const flush = (): Promise => { + const flush = async (): Promise => { // 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 }; diff --git a/services/runner/src/sessions/records-query.ts b/services/runner/src/sessions/records-query.ts index 9bc2468507..bb4b0aa5c8 100644 --- a/services/runner/src/sessions/records-query.ts +++ b/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[] }; diff --git a/services/runner/tests/unit/session-persist.test.ts b/services/runner/tests/unit/session-persist.test.ts index 2def0261fd..d8de2700b0 100644 --- a/services/runner/tests/unit/session-persist.test.ts +++ b/services/runner/tests/unit/session-persist.test.ts @@ -482,10 +482,11 @@ 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); @@ -493,6 +494,36 @@ describe("durable records (AGENTA_RECORDS_DURABLE)", () => { 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"); diff --git a/services/runner/tests/unit/session-reconstruct-history.test.ts b/services/runner/tests/unit/session-reconstruct-history.test.ts index 08c877037e..f6517355e2 100644 --- a/services/runner/tests/unit/session-reconstruct-history.test.ts +++ b/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;