diff --git a/docs/adr/0017-turn-enclosure-invariant.md b/docs/adr/0017-turn-enclosure-invariant.md index 7ec769c651..9974bd99a3 100644 --- a/docs/adr/0017-turn-enclosure-invariant.md +++ b/docs/adr/0017-turn-enclosure-invariant.md @@ -4,7 +4,7 @@ Status: accepted (2026-06-15) ## Context -The durable JSONL backend ([ADR 0018](0018-session-persistence.md)) uses the **turn** as its crash-recovery boundary: `load` returns events only up to the last complete `turn/end`, and the first post-load `append` truncates whatever follows as a never-committed crash tail. This is safe only if nothing *legitimately* durable can sit after the last `turn/end`. +A durable session-persistence backend (added in a companion change) uses the **turn** as its crash-recovery boundary: a crash can leave an unclosed final turn, which `load` closes with a synthetic `turn/end {kind:'interrupted'}` while preserving the turn's real events (see [ADR 0018](0018-session-persistence.md)). This recovery is only well-defined if nothing *legitimately* durable sits OUTSIDE a turn — between the last `turn/end` and the next `turn/start` — since such an event would be swept into the next turn's interrupted close. That assumption did not hold. Two paths recorded events outside any turn: @@ -29,7 +29,7 @@ The serializability invariant is enforced at the same source boundary (`Session. ## Consequences -The turn is now the *single* durability/replay boundary, so [ADR 0018](0018-session-persistence.md)'s "last `turn/end` = commit point" rule is complete, not merely sufficient: a backend can discard everything after the last `turn/end` with zero risk of losing between-turn context, because there is no between-turn context. `scanLog` stays simple (no partial-turn boundary walk), and an idle background-task notice survives persist + resume. +The turn is now the *single* durability/replay boundary, so [ADR 0018](0018-session-persistence.md)'s crash-recovery rule is complete, not merely sufficient: an interrupted final turn is closed (with a synthetic `turn/end {interrupted}`) and its real events preserved, with zero risk of conflating between-turn context into it, because there is no between-turn context. `scanLog` stays simple (one possibly-open final turn, never a loose between-turn event), and an idle background-task notice survives persist + resume. Costs: `agent.inject()` while idle now writes three log lines instead of one, and the derived history gains a turn that carries only injected context (no assistant output) — `deriveMessages()` already derives purely by event type, so this renders identically. The `injection` trigger is a new on-disk vocabulary value; like every `SessionEventMap`/`TurnTriggerMap` addition it is part of the frozen format. Event ordering within a turn changed (`turn/start` now precedes `user/message`), which is observable to anything that asserted the old order — the loop's own tests were the only such consumers. diff --git a/docs/adr/0018-session-persistence.md b/docs/adr/0018-session-persistence.md index 8fb0d88852..872829b453 100644 --- a/docs/adr/0018-session-persistence.md +++ b/docs/adr/0018-session-persistence.md @@ -18,8 +18,8 @@ Persistence is an abstract **capability seam** ([ADR 0009](0009-capability-seams Key choices recorded here because they are durable, contested, and surprising: - **The canonical durable log persists every `SessionEvent` verbatim, including `assistant/chunk`.** `deriveMessages()` skips chunks, and a chunk-filtered rollout (Codex's `policy.rs`) is tempting — but `seq = log.length` and the load-validation `events[i].seq === i` require a *contiguous* log; filtering chunks out would leave holes and break both the contract and resume. A chunk-filtered projection is possible later as a derived view with its own renumbering, but it is NOT the canonical log. -- **Append-only with a single exception.** Committed events — those at or below a flushed `turn/end` — are never rewritten. The loop only flushes at `turn/end`, so a crash can leave a half-written final turn below the last checkpoint; `load` returns events only up to the **last complete `turn/end`**, and the first post-load `append` runs a one-time **truncation-repair** (`ftruncate` + `fsync`) that physically discards only that never-committed crash tail before writing. -- **File backend canonical, DB backend a proven drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. `dsh-session-persistence-sqlite` is exactly this: a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL), and it passes the same `runPersistenceContract` suite as the JSONL backend — so the contract holds both backends to identical semantics (lazy materialization, crash-tail-on-load, contiguous-seq), expressed once over file bytes and once over rows. +- **Append-only; a crashed turn is closed, never truncated.** Committed events — those at or below a flushed `turn/end` — are never rewritten. The loop only flushes at `turn/end`, so a crash can leave a durable log whose final turn never closed: real, fully-written events sit after the last `turn/end`. **A single turn can be huge in a long-horizon task** (many steps, large tool output spanning a long autonomous run), so discarding the interrupted turn would silently destroy a large amount of real work — truncating a turn is wrong. Instead, on reload `load` PRESERVES those events and CLOSES the orphaned turn by durably appending the minimal synthetic boundary events: a `step/end` if a step was still open, then a `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason (a marker that records the turn was cut short by a crash, not completed by the model — no loop ever emits it). `load` returns the balanced log, so a resumed session is immediately usable. The ONLY thing discarded is a never-fully-written **torn tail fragment** — a final record whose bytes (JSONL) or row were never completely flushed; that fragment is not a valid event and is dropped before the synthetic closers are written. A parse error or `seq` gap in the COMMITTED region (at or before the last real `turn/end`) is genuine corruption and makes the session unloadable. +- **File backend canonical, DB backend a proven drop-in.** `SessionEvent` maps 1:1 onto a row `(session_id, seq, type, time, data)` — `append` is INSERT (in a transaction asserting the contiguous-seq contract), `load` is SELECT … ORDER BY seq. `dsh-session-persistence-sqlite` is exactly this: a `SessionPersistence` subclass with no interface change (opencode runs this exact shape on SQLite/WAL), and it passes the same `runPersistenceContract` suite as the JSONL backend — so the contract holds both backends to identical semantics (lazy materialization, interrupted-turn close on load, contiguous-seq), expressed once over file bytes and once over rows. - **Metadata is out-of-log.** Format version, cwd, and lineage are storage concerns, not replayable conversation state, so they live in a `SessionMeta` (`SessionHeader & SessionSummary`) owned by `dsh-session` and attached to a `Session` via a new readonly `session.header` — never in `SessionEventMap`, never reaching `deriveMessages()`. The alternative (a merge-extensible `session/meta` event as log line 0) was rejected: an in-log event would ride along with a seeded/forked session for free, but metadata is not replayable state, so the explicit out-of-log header seam is the cleaner cost. - **Resume is an async factory, not a change to synchronous create.** `ctx.agents.resume({ resumeSessionId })` awaits `ctx.sessionPersistence.load`, recreates the live session with the loaded events (so `lastTurnNumber`/`deriveMessages` continue), and starts a fresh agent on the resumed id (NOT `${agentId}-session`). The agent-loop does NOT hard-inject `sessionPersistence` (that would pend non-persistent demos forever); `resume` rejects with a clear error when it is absent. diff --git a/docs/architecture.md b/docs/architecture.md index 829283b729..81e5e6e609 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -95,7 +95,7 @@ A `Session` is an append-only log of typed `SessionEvent`s — the single source Replay/fork = `ctx.sessions.create(id, { seed: seedEvents })`. Trace/telemetry = listen to `session/event`. -**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, truncation-repair of a never-committed crash tail, and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. Resuming a persisted session into a live agent is `ctx.agents.resume({ resumeSessionId })`. A second backend, `dsh-session-persistence-sqlite` (`node:sqlite`, one row per `SessionEvent` — the row shape `(session_id, seq, type, time, data)` maps 1:1 onto it), passes the same `runPersistenceContract` suite, proving the seam is genuinely backend-agnostic. +**Durability seam**: `session/event` is a synchronous notification; persistence plugins buffer (write-behind) and drain at the awaited `session/flush` checkpoint the loop fires at every turn end. The durable backend is a real **capability seam**: the abstract `SessionPersistence` service (`dsh-session-persistence`, `ctx.sessionPersistence`) defines create/append/load/list/update over the existing `SessionEvent` (no parallel persisted type), and `dsh-session-persistence-jsonl` is the first implementation — an append-only JSONL log per session with crash-safe atomic writes, crash recovery that PRESERVES an interrupted turn (closing it with a synthetic `turn/end {interrupted}` rather than truncating — a turn can be huge), and a read/replay path. Session metadata (format version, cwd, lineage) travels separately as `SessionMeta`, attached to a `Session` via `session.header`. Resuming a persisted session into a live agent is `ctx.agents.resume({ resumeSessionId })`. A second backend, `dsh-session-persistence-sqlite` (`node:sqlite`, one row per `SessionEvent` — the row shape `(session_id, seq, type, time, data)` maps 1:1 onto it), passes the same `runPersistenceContract` suite, proving the seam is genuinely backend-agnostic. ## Prompt assembly (dsh-system-prompt) diff --git a/docs/rfc/013-typed-event-schemas.md b/docs/rfc/013-typed-event-schemas.md new file mode 100644 index 0000000000..5b659a7281 --- /dev/null +++ b/docs/rfc/013-typed-event-schemas.md @@ -0,0 +1,65 @@ +# RFC 013: Runtime schemas for the event vocabulary (Zod vs the merge-extensible-map pattern) + +Status: proposed + +## Problem + +The harness models its core vocabulary — content blocks, message sources, finish reasons, turn triggers, turn-end reasons, and session events — as **merge-extensible maps**: a TypeScript `interface` (e.g. `SessionEventMap`, `ContentBlockMap`) that plugins augment via declaration merging, with the public union derived as `Map[keyof Map]`. This is the repo's universal extension pattern, documented in [docs/architecture.md](../architecture.md) ("The same merge-extensible-map pattern is used for `MessageSource`, `FinishReason`, `TurnTrigger`, and `TurnEndReason`") and relied on by the `defineTool` `InferArgs` DSL and the `assertNever` exhaustiveness convention. + +The pattern is **compile-time only**. The types vanish at runtime: there is no schema object to validate an incoming value against, parse untrusted input with, or enumerate at runtime. Two concrete consequences surfaced in review of the session-persistence work (#33): + +1. **Persistence treats `event.data` as opaque JSON.** The JSONL/SQLite backends `JSON.stringify`/`JSON.parse` each event verbatim; the only runtime guard is `isJsonValue` (round-trip serializability — rejects BigInt, functions, cycles, non-finite numbers, …), NOT structural validation. A corrupted-but-still-JSON event datum (wrong field types, missing fields) round-trips silently and is only caught later, if at all, by a consumer's `switch`. +2. **No runtime contract for plugin-added variants.** A plugin that declaration-merges a new `SessionEventMap` key gets compile-time typing for its own code, but nothing validates that the values it produces match the shape it declared — at the producer, at the persistence boundary, or on reload. + +A reviewer asked whether the project should move "all the JSON serialization/deserialization" — and ultimately the event vocabulary itself — to **Zod** (or a similar runtime-schema library), so the durable boundary and the plugin extension points are backed by runtime schemas rather than erased types. + +This RFC scopes that question. It does **not** propose an implementation; it records the tradeoff so the decision is made deliberately rather than incrementally inside a persistence PR. + +## Why this is not a persistence change + +It is tempting to read "use Zod for serialization" as a local change to `dsh-session-persistence-jsonl/src/format.ts`. It is not, for one structural reason: **a plugin cannot declaration-merge a Zod schema.** Declaration merging is a TypeScript compile-time mechanism; a Zod schema is a runtime value. To validate events with Zod you need a **runtime registry** that every event-producing package contributes its schema to (e.g. `ctx.sessionEvents.register('compaction/marker', z.object({…}))`), and every consumer reads from. That registry — not the persistence backend — becomes the source of truth for the vocabulary, replacing the merge-extensible interface. + +So the real proposal is: **replace the compile-time merge-extensible-map pattern with a runtime schema registry, repo-wide.** That is a core-vocabulary redesign. + +## Blast radius (measured) + +A migration of the event/vocabulary surface to runtime schemas touches, at minimum: + +- **Six merge-extensible maps** (~370 LOC of core types): `ContentBlockMap`, `MessageSourceMap`, `FinishReasonMap` (in `dsh-llm`); `TurnTriggerMap`, `TurnEndReasonMap`, `SessionEventMap` (in `dsh-session`). +- **~10 `declare module` augmentation sites** across `dsh-agent`, `dsh-agent-loop`, `dsh-bash`, `dsh-llm`, `dsh-session`, `dsh-session-persistence`, `dsh-system-prompt`, `dsh-tools` — each would move from declaration merging to a runtime `register()` call. +- **The event producers** — 16 `session.append(...)` call sites in the loop — unchanged in shape but now validated at the boundary. +- **~7 switch-consumers** that branch on these unions: `deriveMessages` (`dsh-session`), `BlockAssembler` (`dsh-llm`), the `dsh-invariants` plugin, both LLM adapters (`dsh-llm-deepseek`, `dsh-llm-pi-ai`), and the tool schema layer (`dsh-tools`). The `assertNever`-on-closed-unions vs fall-through-on-extensible-unions convention (a documented lint rule) would need rethinking — runtime variants are not statically exhaustive. +- **The `defineTool` `InferArgs` DSL** (`dsh-tools`), which derives zero-cast `execute` arg types from a compile-time schema spec — the showcase of the current approach. +- **Docs**: architecture.md (the pattern is described as foundational), ADR 0012 (dev-invariants), and any ADR/RFC that references the pattern. + +This is a HUGE change. It is not in scope for the RFC-009 session-persistence work and must not be smuggled in through it. + +## Options + +### A. Status quo — merge-extensible types + `isJsonValue` at the durable boundary +Keep the compile-time pattern. Persistence stays opaque-JSON + serializability guard. Plugins extend via declaration merging; correctness of event *shape* is the producer's responsibility, enforced by TypeScript at compile time and by the `dsh-invariants` plugin's structural checks in dev. + +- **Pros**: zero churn; plugin extension is a one-line `interface` augmentation with full type inference and no runtime registration ceremony; no new runtime dependency; the `defineTool` DSL and `assertNever` exhaustiveness keep working. +- **Cons**: no runtime structural validation at the persistence boundary or at plugin seams; a malformed-but-JSON datum is caught late. + +### B. Header/closed-shape validation only (schemastery), events stay opaque +Tighten only the genuinely-closed shapes that already have hand-rolled type guards — e.g. the JSONL `HeaderLine` guard (`isHeaderLine`) — using **schemastery** (the repo's existing schema library, already used for every plugin `static Config`). Leave the merge-extensible event union as-is. + +- **Pros**: small, fits the existing convention (schemastery, not a new lib); replaces hand-rolled guards on closed shapes with declarative schemas; no core redesign. +- **Cons**: does not address event-data validation (the thing the reviewer actually asked about); only helps the fixed metadata records. + +### C. Runtime schema registry for the whole vocabulary (Zod or schemastery) +Replace the merge-extensible maps with a runtime registry the producers contribute to and the persistence/consumer paths validate against. + +- **Pros**: real runtime validation at the durable boundary and at plugin seams; one source of truth; enables generic tooling (auto-generated docs, fuzzing, wire-format checks). +- **Cons**: the full blast radius above; **Zod is not currently a direct dependency** (only a transitive dep of `@earendil-works/pi-ai`) and the repo's chosen schema lib is **schemastery** — adopting Zod broadly is itself a dependency decision; declaration-merge ergonomics (one-line plugin extension, full inference) are replaced by runtime registration + manual type wiring; the `assertNever` exhaustiveness guarantee weakens (runtime variants aren't statically exhaustive). + +## Recommendation + +Defer. Do **not** change #33. If runtime validation is wanted at the durable boundary in the near term, **Option B** (schemastery on the closed header/metadata shapes) is the proportionate step and stays within the existing convention. **Option C** is a genuine architecture decision that should be evaluated on its own merits — including whether the chosen library is Zod or schemastery — and, if accepted, land as its own change with its own ADR, not as a side effect of persistence serialization. + +## Open questions + +- If a registry is adopted, is the library **schemastery** (already in the tree, already the config schema lib) or **Zod** (richer ecosystem, currently only transitive)? Adopting two schema libraries is a cost in itself. +- Can a hybrid keep compile-time inference (so `defineTool` and plugin DX survive) while adding an *optional* runtime schema per variant, validated only at the persistence/wire boundary rather than on every in-process append? +- Does the `dsh-invariants` plugin already cover enough of the runtime-shape gap in dev that boundary validation is only needed for genuinely untrusted input (reload of an externally-modified log)? diff --git a/docs/rfc/README.md b/docs/rfc/README.md index 55ec59150d..dbd1468489 100644 --- a/docs/rfc/README.md +++ b/docs/rfc/README.md @@ -16,3 +16,4 @@ Proposals for substantial future work — reviewed before implementation, unlike | [010](010-acp-agent-client-protocol.md) | Agent Client Protocol (ACP) support for external editors | proposed | | [011](011-acp-multi-session.md) | Multiplex concurrent ACP sessions over one connection | proposed | | [012](012-optional-code-mode.md) | Optional Code Mode — model writes TypeScript against an SDK of all tools | proposed | +| [013](013-typed-event-schemas.md) | Runtime schemas for the event vocabulary (Zod vs the merge-extensible-map pattern) | proposed | diff --git a/examples/coding-agent/README.md b/examples/coding-agent/README.md index dee1c83b09..44445adc5d 100644 --- a/examples/coding-agent/README.md +++ b/examples/coding-agent/README.md @@ -22,6 +22,16 @@ Type a coding task. The agent's only tools are `bash` (+ `bash_output` / `bash_k … ``` +### Resuming a prior session + +Each run starts a fresh session by default (its event log lands under `./.sessions/`). To **continue** a previous conversation, set `RESUME_SESSION_ID` to that session's id — the `main` agent then rehydrates the persisted log instead of starting fresh, so the model sees the earlier turns as history: + +```sh +RESUME_SESSION_ID= pnpm run demo:coding +``` + +The id is wired through `cordis.yml` (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`); unset, the agent starts a new session. A missing/unreadable id is non-fatal — it logs a warning and starts no `main` agent. + ## What each plugin demonstrates | Entry | Demonstrates | @@ -36,5 +46,6 @@ Type a coding task. The agent's only tools are `bash` (+ `bash_output` / `bash_k - `tests/full-loop.e2e.ts` — the canary: real model runs `echo e2e-ok` through the real bash tool; asserts `tool/call`/`tool/result` session events and the final answer. - `tests/coding-task.e2e.ts` — the swebench-style smoke: a temp dir holds `add.js` (with `a - b` where `a + b` belongs) and a failing `add.test.js`; the agent must fix the bug and verify. The test re-runs `node add.test.js` ITSELF and inspects the files — agent claims are not trusted. +- `tests/resume.e2e.ts` — durable continuity across processes: run 1 tells the real model a secret code and persists the turn to a temp JSONL root, then the whole context is disposed; run 2 is a fresh context over the same root that RESUMES the session id and asks the model to recall the code. The recall can only come from the rehydrated log. Both self-skip without `DEEPSEEK_API_KEY`. diff --git a/examples/coding-agent/cordis.yml b/examples/coding-agent/cordis.yml index 47eaad8a9b..07eb04d123 100644 --- a/examples/coding-agent/cordis.yml +++ b/examples/coding-agent/cordis.yml @@ -61,6 +61,9 @@ agents: - id: main model: deepseek-v4-flash + # Set RESUME_SESSION_ID to continue a prior persisted session (the ids + # live under ./.sessions); unset starts a fresh session each run. + resumeSessionId: !!js process.env.RESUME_SESSION_ID systemPrompt: | You are coding-agent, a CLI coding assistant. diff --git a/examples/coding-agent/tests/harness.ts b/examples/coding-agent/tests/harness.ts index db89b2c183..ca303691e9 100644 --- a/examples/coding-agent/tests/harness.ts +++ b/examples/coding-agent/tests/harness.ts @@ -9,6 +9,7 @@ import AgentLoop, { LoopAgent } from '@deepseek-ai/dsh-agent-loop' import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local' import * as ToolBash from '@deepseek-ai/dsh-tool-bash' import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' +import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' /** * Shared harness for the coding-agent e2e suites: the full plugin stack @@ -20,7 +21,7 @@ export const SYSTEM_PROMPT = 'You are a coding agent. Your only tool is bash; ' + 'do file operations with cat/grep/heredocs, check [exit code: N] markers, ' + 'and report results briefly.' -export async function codingHarness(workdir: string): Promise { +export async function codingHarness(workdir: string, persistenceRoot?: string): Promise { const ctx = new Context() await ctx.plugin(LlmService) await ctx.plugin(SessionStore) @@ -31,6 +32,10 @@ export async function codingHarness(workdir: string): Promise { await ctx.plugin(LlmDeepSeek, { models: ['deepseek-v4-flash'] }) await ctx.plugin(LocalBashExecutor, { cwd: workdir, timeoutMs: 30_000 }) await ctx.plugin(ToolBash) + // Durable JSONL persistence is opt-in: only the resume e2e needs it, and the + // other suites stay file-free. Loaded last so a resume's deferred + // `ctx.inject(['sessionPersistence'])` resolves once this is present. + if (persistenceRoot !== undefined) await ctx.plugin(SessionPersistenceJsonl, { root: persistenceRoot }) return ctx } diff --git a/examples/coding-agent/tests/resume.e2e.ts b/examples/coding-agent/tests/resume.e2e.ts new file mode 100644 index 0000000000..6fd2dcb567 --- /dev/null +++ b/examples/coding-agent/tests/resume.e2e.ts @@ -0,0 +1,69 @@ +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it } from 'vitest' +import type { Context } from 'cordis' +import type { LoopAgent } from '@deepseek-ai/dsh-agent-loop' +import { codingHarness, finalText, SYSTEM_PROMPT, waitForIdle } from './harness.ts' + +/** + * Proves durable conversation continuity end-to-end: run 1 tells the REAL model + * a fact and persists the turn to JSONL; run 2 is a fresh harness (new Context, + * same `.sessions` root) that RESUMES the persisted session id and asks the + * model to recall the fact. The recall can only come from the rehydrated event + * log — a fresh session would have no idea. Key-gated like the other e2es. + */ + +const SECRET = 'plum-galaxy-1791' +const SESSION_ID = 'resume-e2e-session' + +let ctx: Context | undefined +let root: string | undefined + +afterEach(async () => { + // Dispose even on failure/retry: agent-loop teardown stops the loop and the + // JSONL backend flushes; then drop the on-disk session log. + await ctx?.fiber.dispose() + ctx = undefined + if (root !== undefined) await rm(root, { recursive: true, force: true }) + root = undefined +}) + +describe.skipIf(!process.env.DEEPSEEK_API_KEY)('resume: continue a persisted session across processes', () => { + it('recalls a fact stored in a prior, separately-disposed session', async () => { + root = await mkdtemp(join(tmpdir(), 'dsh-resume-e2e-')) + + // Run 1: a fresh agent on a KNOWN session id learns a secret, then we + // dispose the whole context (simulating process exit) so only the JSONL + // log on disk survives. + ctx = await codingHarness(process.cwd(), root) + const first = ctx.agents.create({ + agentId: 'resume-1', + sessionId: SESSION_ID, + agentOptions: { model: 'deepseek-v4-flash', systemPrompt: SYSTEM_PROMPT }, + }) as LoopAgent + first.send([{ type: 'text', text: `Remember this code for later: ${SECRET}. Just acknowledge it.` }]) + await waitForIdle(ctx, first) + await ctx.fiber.dispose() + ctx = undefined + + // Run 2: a brand-new context over the SAME root resumes the persisted + // session. The loaded event log seeds the live session, so the model sees + // run 1's exchange as conversation history. + ctx = await codingHarness(process.cwd(), root) + const resumed = await ctx.agents.resume({ + agentId: 'resume-2', + resumeSessionId: SESSION_ID, + agentOptions: { model: 'deepseek-v4-flash', systemPrompt: SYSTEM_PROMPT }, + }) as LoopAgent + expect(resumed.session.id).toBe(SESSION_ID) + // The prior user turn is in the rehydrated log before the model is asked. + expect(JSON.stringify(resumed.session.deriveMessages())).toContain(SECRET) + + resumed.send([{ type: 'text', text: 'What was the code I asked you to remember? Reply with just the code.' }]) + await waitForIdle(ctx, resumed) + + // The model recalls it — only possible from the resumed history. + expect(finalText([...resumed.session.events])).toContain(SECRET) + }, 180_000) +}) diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index a1efc78d56..9b40b1f770 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -17,7 +17,7 @@ import { SessionId } from '@deepseek-ai/dsh-session' import type { Session } from '@deepseek-ai/dsh-session' import type {} from '@deepseek-ai/dsh-system-prompt' import type {} from '@deepseek-ai/dsh-tools' -import type {} from '@deepseek-ai/dsh-session-persistence' +import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' import { LoopAgent } from './agent.ts' export { LoopAgent } from './agent.ts' @@ -32,7 +32,19 @@ declare module 'cordis' { export interface Config { /** Agents created from configuration at startup. */ - agents: (AgentOptions & { id: string })[] + agents: (AgentOptions & { + id: string + /** + * If set, the config agent RESUMES this persisted session id instead of + * starting a fresh `${id}-session-`. Sourced from an env var in + * cordis.yml (`resumeSessionId: !!js process.env.RESUME_SESSION_ID`), so a + * demo can continue a prior conversation without code changes. Requires a + * `dsh-session-persistence` backend; the resume is deferred until that + * service is available (via `ctx.inject`) and the loaded session's events + * seed the live session so history continues. + */ + resumeSessionId?: string + })[] } /** @@ -53,6 +65,7 @@ export class AgentLoop extends Service implements AgentFactory { id: z.string().required(), model: z.string(), systemPrompt: z.string(), + resumeSessionId: z.string(), })).default([]), }) @@ -61,8 +74,27 @@ export class AgentLoop extends Service implements AgentFactory { // Provide the agent-creation factory to the registry (effect-scoped: the // slot is cleared on dispose). ctx.effect(() => this.ctx.agents.setFactory(this), 'agentLoop.setFactory()') - for (const { id, ...options } of config.agents) { - this.create(id, options) + for (const { id, resumeSessionId, ...options } of config.agents) { + if (resumeSessionId !== undefined && resumeSessionId !== '') { + // Resume a prior session instead of starting fresh. resume() needs + // `ctx.sessionPersistence`, which may load AFTER this plugin (cordis.yml + // lists the backend later). `ctx.inject(['sessionPersistence'], cb)` + // runs `cb` with a child ctx once the service exists; the child reads + // the persistence and hands it to resumeWith (which uses this.ctx — the + // parent — for sessions/registry, all in AgentLoop's static inject). A + // failed resume is contained + logged: startup must not crash. + ctx.effect(() => { + const fiber = this.ctx.inject(['sessionPersistence'], (childCtx: Context) => { + void this.resumeWith(childCtx.sessionPersistence, { agentId: id, resumeSessionId, agentOptions: options }) + .catch((error: unknown) => { + this.ctx.logger.warn(`agent "${id}": config-driven resume of "${resumeSessionId}" failed: ${String(error)}`) + }) + }) + return () => void fiber.dispose() + }, `agentLoop.resume(${id})`) + } else { + this.create(id, options) + } } } @@ -120,7 +152,6 @@ export class AgentLoop extends Service implements AgentFactory { * by the time this runs the service exists. */ async resume(options: ResumeAgentOptions): Promise { - this.assertAgentIdFree(options.agentId) const persistence = this.ctx.sessionPersistence // `sessionPersistence` is declaration-merged onto Context as non-optional, // but the service is only present when a backend plugin is loaded — and @@ -130,6 +161,20 @@ export class AgentLoop extends Service implements AgentFactory { if (persistence === undefined) { throw new Error('cannot resume: session persistence is not configured (load a dsh-session-persistence backend)') } + return this.resumeWith(persistence, options) + } + + /** + * Resume against an EXPLICIT persistence handle. Factored out of {@link resume} + * so the config-driven path can pass the handle it obtained from a + * `ctx.inject(['sessionPersistence'], …)` child context: `this.ctx` (the + * service's own fiber) did not inject `sessionPersistence`, so reading it + * there from inside the inject child trips the cordis inject guard. The + * sessions store + registry are still read through `this.ctx` (both are in + * AgentLoop's static inject, so they resolve fine). + */ + private async resumeWith(persistence: SessionPersistence, options: ResumeAgentOptions): Promise { + this.assertAgentIdFree(options.agentId) const { meta, events } = await persistence.load(SessionId(options.resumeSessionId)) // Re-check the agent id AFTER the await: the pre-load check above can go // stale while load() is pending (a concurrent resume/create may register the diff --git a/packages/agent-loop/tests/config-session-id.spec.ts b/packages/agent-loop/tests/config-session-id.spec.ts index b905d54145..52f0223564 100644 --- a/packages/agent-loop/tests/config-session-id.spec.ts +++ b/packages/agent-loop/tests/config-session-id.spec.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -62,4 +62,76 @@ describe('config-driven session id', () => { await waitForIdle(ctx2, a2) await ctx2.fiber.dispose() }) + + it('config-driven resumeSessionId continues a persisted session (env-var resume)', async () => { + const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-')) + dirs.push(root) + + // Run 1: a programmatically-created agent on a KNOWN session id persists a + // completed turn, so run 2 has a concrete id to resume. + const ctx1 = new Context() + await ctx1.plugin(LlmService) + await ctx1.plugin(SessionStore) + await ctx1.plugin(SystemPrompt) + await ctx1.plugin(ToolRegistry) + await ctx1.plugin(AgentRegistry) + await ctx1.plugin(AgentLoop, { agents: [] }) + await ctx1.plugin(SessionPersistenceJsonl, { root }) + ctx1.llm.registerAdapter(['mock'], new MockAdapter([textResponse('first')])) + const a1 = ctx1.agents.create({ agentId: 'main', sessionId: 'sticky-1' }) as LoopAgent + a1.send([{ type: 'text', text: 'remember me' }], { source: { kind: 'user' } }) + await waitForIdle(ctx1, a1) + await ctx1.fiber.dispose() + + // Run 2: a CONFIG agent with resumeSessionId continues that session. The + // resume is deferred until sessionPersistence loads (ctx.inject), so wait + // for the agent to appear, then assert it is on the resumed id with history. + const ctx2 = new Context() + await ctx2.plugin(LlmService) + await ctx2.plugin(SessionStore) + await ctx2.plugin(SystemPrompt) + await ctx2.plugin(ToolRegistry) + await ctx2.plugin(AgentRegistry) + await ctx2.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'sticky-1' }] }) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + ctx2.llm.registerAdapter(['mock'], new MockAdapter([textResponse('second')])) + + // The deferred resume runs on a microtask after the backend is available. + let resumed: LoopAgent | undefined + for (let i = 0; i < 50 && !resumed; i++) { + await new Promise(r => setTimeout(r, 5)) + resumed = ctx2.agents.get('main') as LoopAgent | undefined + } + expect(resumed).toBeDefined() + // The live session id IS the resumed id (NOT a fresh ${id}-session-), + // and the prior turn's user message is in the derived history. + expect(resumed!.session.id).toBe('sticky-1') + const derived = resumed!.session.deriveMessages() + expect(JSON.stringify(derived)).toContain('remember me') + await ctx2.fiber.dispose() + }) + + it('config-driven resume of a missing session is contained: logs a warning, no agent, no crash', async () => { + const root = await mkdtemp(join(tmpdir(), 'dsh-cfg-resume-miss-')) + dirs.push(root) + const ctx = new Context() + await ctx.plugin(LlmService) + await ctx.plugin(SessionStore) + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(AgentRegistry) + await ctx.plugin(AgentLoop, { agents: [{ id: 'main', model: 'mock', systemPrompt: '', resumeSessionId: 'does-not-exist' }] }) + const warn = vi.spyOn((ctx.agentLoop as unknown as { ctx: { logger: { warn: (...a: unknown[]) => void } } }).ctx.logger, 'warn') + .mockImplementation(() => undefined) + await ctx.plugin(SessionPersistenceJsonl, { root }) + ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('x')])) + + // The deferred resume fails (no such session on disk). It must be contained: + // a warning is logged, no 'main' agent is registered, and the app stays up. + await new Promise(r => setTimeout(r, 200)) + expect(ctx.agents.get('main')).toBeUndefined() + expect(warn).toHaveBeenCalledWith(expect.stringContaining('config-driven resume of "does-not-exist" failed')) + warn.mockRestore() + await ctx.fiber.dispose() + }) }) diff --git a/packages/session-persistence-jsonl/README.md b/packages/session-persistence-jsonl/README.md index 115e5bfe08..8d1d55d9a7 100644 --- a/packages/session-persistence-jsonl/README.md +++ b/packages/session-persistence-jsonl/README.md @@ -24,7 +24,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence - **Lazy materialization.** `create(meta)` writes nothing; the `.jsonl` (header + first batch) is written atomically (temp-write + `fsync` + rename) on the first `append`. A created-but-never-appended session leaves nothing on disk and is absent from `has`/`list`. - **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent appends are line appends at EOF + `fsync`. -- **Truncation-repair.** `load` returns events only up to the last complete `turn/end` and records the byte offset of any never-committed crash tail; the first post-load `append` `ftruncate`s to that offset (+ `fsync`) before writing, atomically discarding only the uncommitted tail. +- **Crash recovery — close, don't truncate.** A crash can leave a log whose final turn never closed (real events after the last `turn/end`). `load` PRESERVES those events (a turn can be huge — they are real work) and closes the orphaned turn by durably appending synthetic boundary events (a `step/end` if a step was open, then `turn/end {kind:'interrupted'}`), returning a balanced log. Only a never-fully-written **torn tail fragment** (a final line with no newline / unparseable) is `ftruncate`d away before the closers are written. See ADR 0018. - **Contiguous-seq.** `load` rejects a mid-log parse error or `seq` gap (unloadable); `append` rejects a batch whose first `seq` does not continue the stored log, and rejects non-JSON-serializable `event.data` naming the offending event type. - **Format version.** Only v1 is supported; `load` rejects an unknown version. A future format change requires a version bump + migration. diff --git a/packages/session-persistence-jsonl/src/format.ts b/packages/session-persistence-jsonl/src/format.ts index e3834b6578..5b498c66f7 100644 --- a/packages/session-persistence-jsonl/src/format.ts +++ b/packages/session-persistence-jsonl/src/format.ts @@ -120,16 +120,23 @@ export function eventLine(event: SessionEvent): string { } /** - * Compute the byte offset of the END of the last complete `turn/end` line in a - * JSONL log buffer (the header line is index 0). Returns the offset to which a - * crash tail should be truncated, and the contiguous events up to and including - * that `turn/end`. A parse error or a `seq` gap in the MIDDLE (at or before the - * last `turn/end`) makes the session unloadable and throws; trailing garbage - * AFTER the last `turn/end` is the tolerated crash tail and is excluded. + * Parse a JSONL log buffer into its preserved event prefix (the header is line + * 0). Returns the longest prefix of complete, seq-contiguous events plus the + * byte offset of the end of the last preserved line (`committedBytes`). + * + * A crash can leave a durable log whose final turn never closed: real, + * fully-written events sit after the last `turn/end`. Those are PRESERVED (a + * single turn can be huge in a long-horizon task — truncating it would destroy + * real work); the backend closes the orphaned open turn with a synthetic + * `turn/end {kind:'interrupted'}` on reload (ADR 0018). Only a TORN trailing + * fragment — a final line never fully flushed (no newline, unparseable, or a + * seq gap) — is excluded; it bounds the preserved region. A parse error or seq + * gap AT OR BEFORE the last committed `turn/end` is committed-data corruption + * and makes the session unloadable (throws). * * This relies on the session-log invariant that every event lives inside a turn - * (`Session.append` enforces it): the last `turn/end` is therefore the last - * durable boundary, and nothing committed can sit outside a completed turn. + * (`Session.append` enforces it): only the final turn can be open, so the + * preserved tail is at most one unclosed turn. */ export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEvent[]; committedBytes: number } { const text = buffer.toString('utf8') @@ -187,38 +194,46 @@ export function scanLog(buffer: Buffer): { meta: SessionMeta; events: SessionEve } }) - // The last index (into eventEntries) that is a valid `turn/end`. + // The last index (into eventEntries) that is a valid `turn/end` — the last + // fully-committed boundary (the loop flushes only at turn/end). let lastTurnEnd = -1 for (let i = parsed.length - 1; i >= 0; i--) { const p = parsed[i] if (p?.ok && p.event?.type === 'turn/end') { lastTurnEnd = i; break } } - // No committed turn/end anywhere: nothing is committed. The whole event - // region is an uncommitted (first-turn) tail — committedBytes is the header. - if (lastTurnEnd < 0) { - const meta = metaFrom(headerLine) - return { meta, events: [], committedBytes: headerEntry.endByte } - } - - // Pass 2: the committed prefix [0..lastTurnEnd] must be fully intact and - // contiguous (line i is a parsed event with seq === i). A hole or seq gap in - // the committed region means committed data was damaged → unloadable. - const committed: SessionEvent[] = [] - for (let i = 0; i <= lastTurnEnd; i++) { + // Walk the longest PREFIX of complete, seq-contiguous, parseable event lines + // (line i is a parsed event with seq === i). This is the preservable region: + // it includes any fully-written events of an interrupted final turn AFTER the + // last turn/end — those are real, durably-written work and must NOT be + // truncated (a single turn can be huge in a long-horizon task; the orphaned + // open turn is closed with a synthetic turn/end on reload, not discarded — + // ADR 0018). The walk stops at the first hole (unparseable line or seq gap): + // - if that hole is AT OR BEFORE the last committed turn/end, committed data + // was damaged → the session is unloadable (throw); + // - if it is AFTER (or there is no committed turn/end yet), it is the + // tolerated crash boundary — a torn final line never fully flushed — and + // it simply bounds the preserved tail. + const preserved: SessionEvent[] = [] + for (let i = 0; i < parsed.length; i++) { const p = parsed[i] if (!p?.ok || p.event === undefined) { - throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`) + if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`) + break // torn tail fragment after the last turn/end — stop, tolerate } if (p.event.seq !== i) { - throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`) + if (i <= lastTurnEnd) throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${i}, got ${p.event.seq})`) + break // gap after the last turn/end — torn tail, stop } - committed.push(p.event) + preserved.push(p.event) } - const lastEntry = parsed[lastTurnEnd] - /* v8 ignore next -- lastTurnEnd indexes a parsed entry by construction */ - const committedBytes = lastEntry ? lastEntry.endByte : headerEntry.endByte - return { meta: metaFrom(headerLine), events: committed, committedBytes } + + // committedBytes = end of the last PRESERVED line (header if none): the next + // append truncates any torn bytes past this point before writing the + // synthetic closers + new events. + const lastPreserved = parsed[preserved.length - 1] + const committedBytes = preserved.length > 0 && lastPreserved ? lastPreserved.endByte : headerEntry.endByte + return { meta: metaFrom(headerLine), events: preserved, committedBytes } } /** Build the load-time {@link SessionMeta} from a header line (summary overlaid later). */ diff --git a/packages/session-persistence-jsonl/src/index.ts b/packages/session-persistence-jsonl/src/index.ts index 7f8b80a215..3e5320c3a9 100644 --- a/packages/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence-jsonl/src/index.ts @@ -27,7 +27,7 @@ import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node import { resolve } from 'node:path' import { randomBytes } from 'node:crypto' import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' -import { isJsonValue } from '@deepseek-ai/dsh-session' +import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session' import { encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, sidecarPath, toHeaderLine, @@ -58,11 +58,6 @@ interface SessionState { * new session's events to be dropped against the old cursor). */ owner?: Session - /** - * If a load truncation-repair is pending, the byte offset to truncate the - * file to before the next append (discards the never-committed crash tail). - */ - repairTo?: number } /** @@ -238,13 +233,6 @@ export class SessionPersistenceJsonl extends SessionPersistence { let state = this.states.get(id) if (state === undefined) state = await this.adopt(id) // calls loadCore, not load - // Truncation-repair: on the first append after a load that found a crash - // tail, physically discard the orphaned bytes before writing. - if (state.repairTo !== undefined) { - await this.repair(state, state.repairTo) - delete state.repairTo - } - // Contiguity contract: each event's seq must continue the stored log. for (const [i, event] of events.entries()) { if (event.seq !== state.cursor + i) { @@ -280,19 +268,42 @@ export class SessionPersistenceJsonl extends SessionPersistence { const summary = await this.readSidecar(id, meta.cwd) const fullMeta: SessionMeta = { ...meta, ...summary } - // Record the state so the next append repairs the crash tail (if any) and - // continues at the committed length. The state keeps its OWN copy of the - // meta; the value returned to the caller is a SEPARATE copy so a consumer - // mutating `loaded.meta` (e.g. `cwd`) cannot corrupt the backend's pathing - // metadata and send later reads/writes to the wrong log. - const needsRepair = committedBytes < buffer.byteLength - this.states.set(id, { + // Crash-recovery: if the log ended mid-turn (an open turn with real, + // preserved events but no closing turn/end), close it durably DURING load so + // disk, the returned log, and the cursor all agree — both append routes then + // continue with no special-casing. Synthesize the boundary events (a + // step/end if a step was open, then a turn/end {kind:'interrupted'}); the + // interrupted turn's real events are preserved, never truncated (a turn can + // be huge — ADR 0018). + const closers = interruptedTurnClosers(events) + const balanced = [...events, ...closers] + + // Set state BEFORE the repair writes so they can resolve the log path. + const needsTorn = committedBytes < buffer.byteLength + const state: SessionState = { meta: { ...fullMeta }, cursor: events.length, materialized: true, - ...needsRepair ? { repairTo: committedBytes } : {}, - }) - return { meta: fullMeta, events } + } + this.states.set(id, state) + + if (needsTorn) { + // Discard the torn trailing fragment (a final line never fully flushed) + // before writing the closers, so the closers land at a clean EOF. + await this.repair(state, committedBytes) + } + if (closers.length > 0) { + // Durably append the synthetic closers, then advance the cursor to the + // balanced length. After this, disk == balanced and the next append (live + // or direct) continues cleanly. No sidecar touch here: load is not a + // summary-changing op (the closers carry no new title/firstPrompt), and + // the next real append bumps `updatedAt` — keeping the summary write off + // the recovery path avoids a second best-effort failure mode. + await this.appendLines(state, closers) + state.cursor = balanced.length + } + + return { meta: fullMeta, events: balanced } } async list(): Promise { diff --git a/packages/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence-jsonl/tests/jsonl.spec.ts index a91e6f9138..563dd8c627 100644 --- a/packages/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence-jsonl/tests/jsonl.spec.ts @@ -107,36 +107,41 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => { expect(loaded.events).toEqual(log) // chunks preserved, contiguous seqs }) - it('crash tolerance: load truncates an uncommitted final turn back to the last turn/end', async () => { + it('crash recovery: load preserves the interrupted turn and closes it with a synthetic turn/end {interrupted}', async () => { const m = meta('crash', '/proj') await ctx.sessionPersistence.create(m) await ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5, turn/end at 5 // Simulate a crash mid-second-turn: append raw lines that are NOT closed by - // a turn/end (and a final partial line with no newline). + // a turn/end (turn/start + step/start are fully written), plus a final + // partial line with no newline (a torn fragment never fully flushed). const path = logPath(root, '/proj', m.id) - const tail = [ + await writeFile(path, [ JSON.stringify({ type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }), JSON.stringify({ type: 'step/start', seq: 7, time: 9, data: { turn: 2, step: 1 } }), - '{"type":"assistant/chunk","seq":8,"ti', // truncated partial line - ].join('\n') - await writeFile(path, tail, { flag: 'a' }) + '{"type":"assistant/chunk","seq":8,"ti', // truncated partial line (no newline) + ].join('\n'), { flag: 'a' }) - // load returns only the committed first turn. + // load PRESERVES the interrupted turn's real events (turn/start 6, step/start + // 7) — a turn can be huge, so they must not be truncated — and durably closes + // the orphaned turn with synthetic step/end (8) + turn/end {interrupted} (9). const loaded = await ctx.sessionPersistence.load(m.id) - expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5]) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) + const last = loaded.events.at(-1)! + expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' }) + const stepEnd = loaded.events[8]! + expect(stepEnd.type).toBe('step/end') + // the torn seq-8 chunk fragment did not survive + expect(loaded.events.some(e => e.type === 'assistant/chunk' && e.seq === 8)).toBe(false) - // The next append repairs the file (discarding the crash tail) and resumes - // at seq 6. - const turn2 = [ - { type: 'turn/start', seq: 6, time: 10, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, - { type: 'turn/end', seq: 7, time: 11, data: { turn: 2, reason: { kind: 'completed' } } }, + // The next append continues at seq 10 (the balanced length). + const turn3 = [ + { type: 'turn/start', seq: 10, time: 11, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 11, time: 12, data: { turn: 3, reason: { kind: 'completed' } } }, ] as SessionEvent[] - await ctx.sessionPersistence.append(m.id, turn2) + await ctx.sessionPersistence.append(m.id, turn3) const reloaded = await ctx.sessionPersistence.load(m.id) - expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7]) - // and no orphaned seq-8 chunk survived - expect(reloaded.events.some(e => e.seq === 8)).toBe(false) + expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]) }) it('committed events are never rewritten: only the crash tail is repaired', async () => { @@ -490,15 +495,17 @@ describe('SessionPersistenceJsonl: scanLog unit', () => { expect(() => scanLog(Buffer.from('{"type":"event"}\n'))).toThrow(/session header/) }) - it('a seq gap with NO committed turn/end yields zero committed events (uncommitted tail)', () => { + it('a seq gap after the last turn/end bounds the preserved tail (torn fragment tolerated)', () => { const log = [ JSON.stringify({ type: 'session', version: 1, id: 'g', createdAt: 1 }), JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }), JSON.stringify({ type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }), // gap: missing seq 1 ].join('\n') + '\n' - // Nothing reached a turn/end, so nothing is committed — the whole region is - // an uncommitted (crash) tail. Safe to load as empty, NOT a corruption. - expect(scanLog(Buffer.from(log)).events).toEqual([]) + // No committed turn/end, so the gap is a tolerated crash boundary: scanLog + // PRESERVES the contiguous prefix (turn/start seq 0) — real interrupted-turn + // work, not discarded — and stops at the gap. The orphaned open turn is + // closed by loadCore's synthetic turn/end, not here. + expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0]) }) it('rejects a seq gap BEFORE a later committed turn/end (committed data damaged)', () => { @@ -522,13 +529,23 @@ describe('SessionPersistenceJsonl: scanLog unit', () => { expect(() => scanLog(Buffer.from(log))).toThrow(/unparsable committed event/) }) - it('a corrupt line with NO committed turn/end yields zero committed events', () => { + it('a header-only log (no event lines at all) preserves nothing — committedBytes is the header', () => { + const log = JSON.stringify({ type: 'session', version: 1, id: 'h0', createdAt: 1 }) + '\n' + const scanned = scanLog(Buffer.from(log)) + expect(scanned.events).toEqual([]) + // committedBytes falls back to the header line's end (no preserved events). + expect(scanned.committedBytes).toBe(Buffer.byteLength(log, 'utf8')) + }) + + it('a corrupt line after the last turn/end bounds the preserved tail', () => { const log = [ JSON.stringify({ type: 'session', version: 1, id: 'c2', createdAt: 1 }), JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }), '{not json', // corrupt crash fragment, no turn/end committed ].join('\n') + '\n' - expect(scanLog(Buffer.from(log)).events).toEqual([]) + // The contiguous prefix (turn/start seq 0) is preserved; the corrupt + // fragment after it is the tolerated crash boundary. + expect(scanLog(Buffer.from(log)).events.map(e => e.seq)).toEqual([0]) }) it('tolerates a seq gap AFTER a turn/end (uncommitted tail)', () => { @@ -1066,13 +1083,19 @@ describe('SessionPersistenceJsonl: edge cases', () => { await ctx2.fiber.dispose() }) - it('a header-only log (no turn/end) loads as zero committed events', () => { - const log = [ - JSON.stringify({ type: 'session', version: 1, id: 'open', createdAt: 1 }), - JSON.stringify({ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }), - ].join('\n') + '\n' - const { events } = scanLog(Buffer.from(log)) - expect(events).toEqual([]) // nothing committed (no turn/end) + it('a header-only log (open turn, no turn/end) preserves the open turn on load and closes it', async () => { + // A session whose only durable content is an unclosed first turn. scanLog + // preserves the turn/start; loadCore closes it with a synthetic + // turn/end {interrupted} so the returned log is balanced. + const m = meta('open-turn', '/h') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, [ + { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } }, + ] as SessionEvent[]) + const { events } = await ctx.sessionPersistence.load(m.id) + expect(events.map(e => e.type)).toEqual(['turn/start', 'turn/end']) + const end = events[1]! + expect(end.type === 'turn/end' && end.data.reason).toEqual({ kind: 'interrupted' }) }) it('initFor is idempotent: a re-seeded existing session is not re-initialized', async () => { diff --git a/packages/session-persistence/README.md b/packages/session-persistence/README.md index 77ee9a4e2d..e61e988c0e 100644 --- a/packages/session-persistence/README.md +++ b/packages/session-persistence/README.md @@ -10,14 +10,14 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l |---|---| | `create(meta): Promise` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). | | `append(id, events): Promise` | Durably persist a batch (from the `session/flush` drain). Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. | -| `load(id): Promise<{ meta; events }>` | Reload meta + log up to the last complete `turn/end`; events contiguous (`events[i].seq === i`); rejects a mid-log gap/parse error or unknown `version`. | +| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. | | `list(): Promise` | Lightweight listing from metadata, no full-log parse. | | `has(id)` / `delete(id)` | Existence / removal. A zero-event lazily-materialized session is absent from `has`/`list`. | | `update(id, summary): Promise` | Update mutable `SessionSummary` fields without touching the append-only log. | ## Invariants every backend must honor -- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. The only exception is the one-time truncation-repair of a never-committed crash tail on the first `append` after a `load`. +- **Append-only; a crashed turn is closed, not truncated.** Committed events (at or below a flushed `turn/end`) are never rewritten. A crash can leave an unclosed final turn whose events are real and possibly large; `load` preserves them and durably appends synthetic closers (`step/end?`+`turn/end {interrupted}`) to balance the log. Only a never-fully-written torn tail fragment is discarded. - **Contiguous seq.** `load` rejects a `seq` gap/parse error in the MIDDLE of the log; `append`'s first `seq` must equal the stored next-seq. - **JSON-serializable data.** `append` rejects non-serializable `event.data`; backends snapshot each event when buffering (the live `session.events` object is mutable). - **Durability.** `append` returns only once the batch is durable. diff --git a/packages/session-persistence/src/index.ts b/packages/session-persistence/src/index.ts index f08ac2e1f0..f6c98b725e 100644 --- a/packages/session-persistence/src/index.ts +++ b/packages/session-persistence/src/index.ts @@ -42,14 +42,16 @@ declare module 'cordis' { * Contracts every implementation MUST honor (a DB backend asserts them inside * a transaction; a file backend appends at EOF): * - * - **Append-only.** Committed events — those at or below a flushed `turn/end` - * — are never rewritten. The ONLY exception is the one-time truncation-repair - * of a never-committed crash tail on the first {@link append} after a - * {@link load} (see {@link load}). + * - **Append-only; a crashed turn is closed, not truncated.** Committed events + * — those at or below a flushed `turn/end` — are never rewritten. A crash can + * leave an unclosed final turn whose events are real (and possibly large); + * {@link load} preserves them and closes the orphaned turn with synthetic + * boundary events (see {@link load}). Only a never-fully-written torn tail + * fragment is discarded. * - **Contiguous seq.** A persisted log is contiguous: `events[i].seq === i`. - * {@link load} rejects a parse error or a `seq` gap in the MIDDLE of the log + * {@link load} rejects a parse error or a `seq` gap in the COMMITTED region * (unloadable); {@link append}'s first event `seq` MUST equal the backend's - * stored next-seq after any repair. + * stored next-seq (after `load` has balanced any interrupted turn). * - **JSON-serializable data.** `SessionEventMap` is merge-extensible and * `event.data` is typed only as `SessionEventMap[K]`, so {@link append} * REJECTS non-JSON-serializable data with an error naming the offending @@ -75,9 +77,9 @@ export abstract class SessionPersistence extends Service { /** * Durably persist a batch of events (called from the write-behind drain at * the `session/flush` checkpoint). Honors the append-only and contiguous-seq - * contracts: the first event's `seq` MUST equal the stored next-seq after - * any truncation-repair of a crash tail. Rejects non-JSON-serializable - * `event.data` with an error naming the offending event type. + * contracts: the first event's `seq` MUST equal the stored next-seq (after + * `load` has durably closed any interrupted turn). Rejects non-JSON- + * serializable `event.data` with an error naming the offending event type. */ abstract append(id: SessionId, events: readonly SessionEvent[]): Promise @@ -86,13 +88,19 @@ export abstract class SessionPersistence extends Service { * durable checkpoint. Returns `meta` AND `events` so the live session is * reconstructed with its `cwd`/lineage, not just its log. * - * The loop only flushes at `turn/end`, so a crash can leave a half-written - * final turn below the last committed checkpoint. `load` returns events only - * up to the **last complete `turn/end`**; a subsequent {@link append} runs - * the one-time truncation-repair that physically discards the orphaned tail - * before writing. Returned events are contiguous (`events[i].seq === i`); a - * parse error or a `seq` gap in the MIDDLE of the log makes the session - * unloadable (reject). Rejects an unknown format `version`. + * The loop only flushes at `turn/end`, so a crash can leave a durable log + * whose final turn never closed: real, fully-written events sit after the last + * `turn/end`. Those events are PRESERVED — a single turn can be huge in a + * long-horizon task, so truncating it would destroy real work — and `load` + * CLOSES the orphaned turn by durably appending the minimal synthetic boundary + * events (a `step/end` if a step was open, then a `turn/end` carrying the + * `{ kind: 'interrupted' }` reason). The returned `events` therefore end on a + * balanced `turn/end` and are immediately usable as a session seed. Only a + * never-fully-written TORN tail fragment (a half-written final record) is + * discarded. Returned events are contiguous (`events[i].seq === i`); a parse + * error or a `seq` gap in the COMMITTED region (at or before the last real + * `turn/end`) makes the session unloadable (reject). Rejects an unknown format + * `version`. See ADR 0018 for the crash-recovery contract. */ abstract load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> diff --git a/packages/session-persistence/tests/contract.ts b/packages/session-persistence/tests/contract.ts index 3c4e739b60..189a4e10c2 100644 --- a/packages/session-persistence/tests/contract.ts +++ b/packages/session-persistence/tests/contract.ts @@ -64,6 +64,44 @@ export function runPersistenceContract(name: string, make: () => Promise { + const { persistence, dispose } = await make() + try { + const m = meta('interrupted') + await persistence.create(m) + await persistence.append(m.id, oneTurnLog()) // turn 1, committed (seqs 0..5) + // A second turn that crashed mid-flight: turn/start + step/start were + // durably written, but no step/end / turn/end ever arrived. + await persistence.append(m.id, [ + { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } }, + ]) + + // load PRESERVES the interrupted turn's events (a turn can be huge — they + // must not be truncated) and closes the orphaned turn with synthetic + // boundary events: step/end (the step was open) then turn/end {interrupted}. + const loaded = await persistence.load(m.id) + expect(loaded.events.map(e => e.type)).toEqual([ + 'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1 + 'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers + ]) + expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9]) + const last = loaded.events.at(-1)! + expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' }) + + // The closed log is durable and continuable: a fresh append continues at + // the balanced length (seq 10), and a reload round-trips identically. + await persistence.append(m.id, [ + { type: 'turn/start', seq: 10, time: 9, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } }, + { type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } }, + ]) + const reloaded = await persistence.load(m.id) + expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11]) + } finally { + await dispose() + } + }) + it('has()/list() exclude a created-but-never-appended (zero-event) session', async () => { const { persistence, dispose } = await make() try { diff --git a/packages/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/tests/persistence.spec.ts index 9d211fff52..ea55f75d9c 100644 --- a/packages/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/tests/persistence.spec.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' -import { SessionId, isJsonValue } from '@deepseek-ai/dsh-session' +import { SessionId, isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session' import { SessionPersistence } from '../src/index.ts' import { runPersistenceContract, meta, oneTurnLog } from './contract.ts' @@ -46,6 +46,11 @@ class MemoryPersistence extends SessionPersistence { async load(id: SessionId): Promise<{ meta: SessionMeta; events: SessionEvent[] }> { const entry = this.store.get(id) if (!entry) throw new Error(`session "${id}" not found`) + // Honor the crash-recovery contract: if the stored log ends mid-turn, close + // the orphaned turn durably with synthetic boundary events and continue from + // the balanced length. + const closers = interruptedTurnClosers(entry.events) + if (closers.length > 0) entry.events.push(...structuredClone(closers)) return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) } } diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 8fe54cb97b..34eefb24a3 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -15,6 +15,7 @@ import { isJsonValue } from './json.ts' export * from './types.ts' export { isJsonValue } from './json.ts' +export { interruptedTurnClosers } from './repair.ts' declare module 'cordis' { interface Context { diff --git a/packages/session/src/repair.ts b/packages/session/src/repair.ts new file mode 100644 index 0000000000..c09d15eee6 --- /dev/null +++ b/packages/session/src/repair.ts @@ -0,0 +1,79 @@ +/** + * Crash-recovery repair for an interrupted session log. + * + * A persistence backend flushes only at `turn/end`, so a crash can leave a + * durable log whose final turn never closed: real, fully-written events sit + * after the last `turn/end` with no closing boundary. A single turn can be huge + * in a long-horizon task (many steps, large tool output), so those events MUST + * be preserved — truncating the turn would silently destroy real work. Instead, + * on reload the backend CLOSES the orphaned turn by appending the minimal + * synthetic boundary events (a `step/end` if a step was still open, then a + * `turn/end` carrying the merge-extensible `{ kind: 'interrupted' }` reason). + * The marker records that the turn was cut short by a crash, not completed by + * the model. See ADR 0018. + * + * This module computes those synthetic closers from an event list; the backend + * returns them inline from `load` (so the reconstructed session is balanced and + * immediately usable) and persists them on the first post-load `append`. + * + * @module @deepseek-ai/dsh-session/repair + */ + +import type { SessionEvent } from './types.ts' + +/** + * Scan `events` for an open turn/step at the tail and return the synthetic + * boundary events that close them, with `seq` continuing the log and `time` + * copied from the last real event (the closers stand in for the crash moment; + * reusing the last timestamp keeps them deterministic and never invents a + * "future" time). Returns an empty array when the log is already balanced + * (ends on a `turn/end`, or is empty) — the common, non-crash case. + * + * Only the LAST turn can be open: the invariants plugin guarantees a `turn/end` + * before any later `turn/start`, so an interior open turn is impossible in a + * valid committed log. Likewise at most one step is open within that turn. + */ +export function interruptedTurnClosers(events: readonly SessionEvent[]): SessionEvent[] { + let openTurn: number | null = null + let openStep: number | null = null + for (const event of events) { + switch (event.type) { + case 'turn/start': + openTurn = event.data.turn + break + case 'turn/end': + openTurn = null + openStep = null + break + case 'step/start': + openStep = event.data.step + break + case 'step/end': + openStep = null + break + // Other event types do not move the turn/step boundary cursor. + default: + break + } + } + + // Balanced log (no crash mid-turn): nothing to close. An open turn implies + // `events` is non-empty (its turn/start was logged), so `last` exists. + const last = events.at(-1) + if (openTurn === null || last === undefined) return [] + + // The last real event supplies the seq base and the timestamp for the + // synthetic closers (reusing the last timestamp keeps them deterministic and + // never invents a "future" time). + let seq = last.seq + 1 + const time = last.time + const closers: SessionEvent[] = [] + + // Close an open step first — a turn/end while a step is open is an invariant + // violation, so the step's boundary must be synthesized before the turn's. + if (openStep !== null) { + closers.push({ type: 'step/end', seq: seq++, time, data: { turn: openTurn, step: openStep } }) + } + closers.push({ type: 'turn/end', seq: seq++, time, data: { turn: openTurn, reason: { kind: 'interrupted' } } }) + return closers +} diff --git a/packages/session/src/types.ts b/packages/session/src/types.ts index 197e251b59..d8ffaf9dc8 100644 --- a/packages/session/src/types.ts +++ b/packages/session/src/types.ts @@ -99,6 +99,17 @@ export interface TurnEndReasonMap { aborted: { kind: 'aborted'; reason?: string } error: { kind: 'error'; message: string; code?: string } disposed: { kind: 'disposed' } + /** + * The turn never ended on its own: the process crashed mid-turn and a + * persistence backend later closed the orphaned (open) turn on reload so the + * log stays balanced. SYNTHESIZED by the backend's crash-recovery repair — no + * loop ever emits this. Its events are real (they were durably appended before + * the crash) and are PRESERVED, not discarded: a single turn can be huge in a + * long-horizon task (many steps, large tool output), so truncating it would + * lose real work. The marker records that the turn was cut short, not that the + * model completed it. See ADR 0018. + */ + interrupted: { kind: 'interrupted' } } export type TurnEndReason = TurnEndReasonMap[keyof TurnEndReasonMap]