diff --git a/packages/agent-loop/src/agent.ts b/packages/agent-loop/src/agent.ts index 7705f8c3ed..64576186c3 100644 --- a/packages/agent-loop/src/agent.ts +++ b/packages/agent-loop/src/agent.ts @@ -93,29 +93,51 @@ export class LoopAgent implements Agent { // open injection turn that would corrupt later turns/replay. (If the // turn/start append throws BEFORE pushing — non-serializable trigger, which // can't happen for our fixed trigger — no turn was opened and none is owed.) - let turnRecorded = false try { this.session.append('turn/start', { turn, trigger: { kind: 'injection', source } }) this.session.append('context/message', { content, source }) } finally { - // A turn was recorded iff turn/start made it into the log. Close it and - // mark it for the durability checkpoint below — which must run even when - // an append's listener threw (the turn is balanced and in memory, so it - // still needs a flush or a crash before the next turn/dispose loses it). + // Close the turn if turn/start made it into the log. Contain a throwing + // turn/end listener: Session.append pushes before notifying, so a throw + // here still leaves turn/end in the log (the turn is balanced) — swallow + // it so it neither replaces the original exception nor skips the flush + // decision below. (It surfaces through the flush path is not needed; the + // turn-balance contract is what matters and it holds.) if (isTurnOpen(this.session)) { - this.session.append('turn/end', { turn, reason: { kind: 'completed' } }) - turnRecorded = true + try { + this.session.append('turn/end', { turn, reason: { kind: 'completed' } }) + } catch { + // turn/end is already in the log (pushed before the listener threw), + // so the turn is balanced; the throw is the listener's bug. + } } + // Decide the durability checkpoint from the LOG, not a flag: a turn was + // recorded iff this turn's turn/start is logged (it may have been closed + // by a throwing-listener turn/end above, which still counts). A + // `turnRecorded` boolean set after append('turn/end') would be skipped by + // a throwing turn/end listener, losing the flush for a balanced in-memory + // turn (crash before the next turn/dispose would drop the idle injection). + const turnRecorded = this.session.events.some(e => e.type === 'turn/start' && e.data.turn === turn) // Checkpoint the one-shot turn for durability, exactly as the loop does at // every turn/end. The loop is NOT running (we are idle), so nothing else // will flush this turn. Fire-and-forget with error containment: inject() // is synchronous, and a persistence backend failing must not throw into // the caller (e.g. a tool-bash task-done callback). Disposal still drains - // independently, so a slow flush is safe. In the finally so it also runs - // when an append's listener threw (the turn is still balanced + durable). + // independently, so a slow flush is safe. A flush failure is reported via + // agent/error (step 0 — the idle-injection convention, there is no real + // step) AND the logger, mirroring the loop's post-turn/end flush path so + // plugins monitoring agent/error see idle-injection persistence failures + // too. A throwing agent/error listener is contained. if (turnRecorded) { void Promise.resolve(this.ctx.parallel('session/flush', this.session)).catch((error: unknown) => { - this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${String(error)}`) + const err = error instanceof Error ? error : new Error(String(error)) + this.ctx.logger.warn(`agent "${this.id}": flush after idle injection failed: ${err.message}`) + try { + this.ctx.emit('agent/error', this, turn, 0, err) + } catch { + // contained: the failure is already logged; a throwing agent/error + // listener must not escape this fire-and-forget catch. + } }) } } diff --git a/packages/agent-loop/src/index.ts b/packages/agent-loop/src/index.ts index 355c610989..a1efc78d56 100644 --- a/packages/agent-loop/src/index.ts +++ b/packages/agent-loop/src/index.ts @@ -114,7 +114,7 @@ export class AgentLoop extends Service implements AgentFactory { * continue), and starts a fresh agent on it. The live session id is the * resumed id, NOT `${agentId}-session`. * - * Requires `ctx.sessionPersistence`; throws a typed error if it is not + * Requires `ctx.sessionPersistence`; rejects with a clear error if it is not * configured. NOT hard-injected (that would make non-persistent demos pend * forever) — callers that need resume (ACP) inject `sessionPersistence`, so * by the time this runs the service exists. diff --git a/packages/agent-loop/src/loop.ts b/packages/agent-loop/src/loop.ts index 71b78d2e58..3497892c1c 100644 --- a/packages/agent-loop/src/loop.ts +++ b/packages/agent-loop/src/loop.ts @@ -170,7 +170,6 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: let reason: TurnEndReason = { kind: 'completed' } let step = 0 - let turnStarted = false let turnEnded = false let stepOpen = false let errorReported = false @@ -236,9 +235,10 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: try { // --- Turn boundary. Once turn/start is appended, a turn/end is owed no - // matter what throws below; the catch + closeTurn guarantee it. + // matter what throws below; the catch + closeTurn guarantee it (the catch + // decides "owed" from the log via isTurnOpen, so even a throwing turn/start + // listener — append pushes before notifying — still gets its turn/end). session.append('turn/start', { turn, trigger }) - turnStarted = true // Record the queued user messages INSIDE the turn (after turn/start), so // every event in the log is turn-enclosed. turn/end is now owed, so a throw // while appending these is caught below and the turn is still closed. @@ -320,12 +320,20 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: // Normal / inline-error loop exit: close the turn and notify. closeTurn(true) } catch (error: unknown) { - // A pre-turn throw (turn/start append) is owed no turn/end — rethrow to - // the backstop. Otherwise a boundary emit (turn-start, step-start, the - // normal-path turn-end) or other unhandled throw escaped: close any open - // step, choose the reason (disposal wins only if no error was reported), - // record the error, and close the turn WITHOUT re-emitting agent/turn-end. - if (!turnStarted) throw error + // Decide whether this turn was ever opened from the LOG, not a flag. + // Session.append pushes the event BEFORE notifying session/event listeners, + // so a throwing listener on the `turn/start` append leaves turn/start in the + // log even though execution never reached the lines after that append. + // Gating on a "turn started" boolean would skip turn/end and leave a + // permanently OPEN turn that poisons the next turn/replay (ADR 0017). We + // check the log for THIS turn's turn/start: present means a turn/end is owed + // (or was already appended — closeTurn/failTurn are idempotent, so running + // them again is a safe no-op that still preserves the disposed/error reason + // chosen below). Absent means the turn/start append threw BEFORE its push (a + // non-serializable trigger — impossible for our fixed trigger); nothing was + // opened, so rethrow to the runLoop backstop. + const turnStartLogged = session.events.some(e => e.type === 'turn/start' && e.data.turn === turn) + if (!turnStartLogged) throw error closeStep() // Choose the close reason. Disposal wins only if no error was already // reported: a turn disposed mid-step sets reason=disposed in the step-error diff --git a/packages/agent-loop/tests/agent.spec.ts b/packages/agent-loop/tests/agent.spec.ts index 32b28be74a..6ec8d7d8f5 100644 --- a/packages/agent-loop/tests/agent.spec.ts +++ b/packages/agent-loop/tests/agent.spec.ts @@ -141,6 +141,50 @@ describe('LoopAgent', () => { expect(flushes).toBe(1) // checkpoint fired despite the throw }) + it('idle inject() still checkpoints when a listener throws on the synthetic turn/end', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + let flushes = 0 + ctx.on('session/flush', () => { flushes += 1 }) + // A session/event listener that throws on the synthetic turn/end. Append + // pushes before notifying, so turn/end is in the log (turn balanced) but the + // throw must NOT skip the durability checkpoint — the flush decision is made + // from the log, not a flag set after the (throwing) append. + let threw = false + ctx.on('session/event', (_s, event) => { + if (!threw && event.type === 'turn/end') { threw = true; throw new Error('boom turn/end') } + }) + + expect(() => { agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) }).not.toThrow() + const types = agent.session.events.map(e => e.type) + expect(types).toEqual(['turn/start', 'context/message', 'turn/end']) // balanced + await new Promise(r => setTimeout(r, 10)) + expect(flushes).toBe(1) // checkpoint fired despite the throwing turn/end listener + }) + + it('idle inject() reports a failing flush via agent/error (step 0) AND the logger', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + // A non-Error rejection exercises the String() normalization branch. + ctx.on('session/flush', () => { throw 'disk gone' }) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + const errors: { turn: number; step: number; message: string }[] = [] + ctx.on('agent/error', (_a, turn, step, error) => void errors.push({ turn, step, message: error.message })) + + agent.inject([{ type: 'text', text: 'notice' }], { source: { kind: 'plugin', plugin: 'p' } }) + await new Promise(r => setTimeout(r, 20)) // let the contained flush settle + + // Reported via agent/error (step 0 — the idle-injection convention) so + // plugins monitoring agent/error see idle-injection persistence failures, + // mirroring the loop's post-turn/end flush path. A non-Error throw is + // normalized to an Error. + expect(errors).toEqual([{ turn: 1, step: 0, message: 'disk gone' }]) + expect(warn).toHaveBeenCalledWith(expect.stringContaining('flush after idle injection failed')) + warn.mockRestore() + }) + it('idle inject() with a non-serializable source opens no turn (nothing to close)', async () => { const adapter = new MockAdapter([textResponse('ok')]) const ctx = await harness(adapter) diff --git a/packages/agent-loop/tests/coverage-edges.spec.ts b/packages/agent-loop/tests/coverage-edges.spec.ts index 260c521de1..90b87d02aa 100644 --- a/packages/agent-loop/tests/coverage-edges.spec.ts +++ b/packages/agent-loop/tests/coverage-edges.spec.ts @@ -35,9 +35,11 @@ function send(agent: LoopAgent, text: string) { agent.send([{ type: 'text', text }]) } -describe('loop backstop catch', () => { - it('a throwing turn-start listener is caught by the backstop and loop survives', async () => { - // The first turn will abort before the model call (turn-start throw). +describe('turn boundary listener throws (handled in-turn, loop survives)', () => { + it('a throwing agent/turn-start listener surfaces via agent/error and the loop survives', async () => { + // The agent/turn-start emit happens AFTER turn/start is appended to the log, + // so a throwing listener is handled inside runTurn (the turn is balanced and + // closed via failTurn → agent/error), NOT rethrown to the runLoop backstop. // The second turn should proceed normally and consume the first script entry. const adapter = new MockAdapter([textResponse('turn 2')]) const ctx = await harness(adapter) @@ -57,6 +59,9 @@ describe('loop backstop catch', () => { send(agent, 'first') await waitForIdle(ctx, agent) expect(errors.map(e => e.message)).toEqual(['broken turn-start listener']) + // The turn is balanced: its turn/start was logged, so a turn/end was owed + // and appended (decided from the log, not a flag). + expect(agent.session.events.at(-1)?.type).toBe('turn/end') // loop survives: second turn works fine and makes the model call send(agent, 'second') @@ -65,7 +70,7 @@ describe('loop backstop catch', () => { expect(adapter.requests[0]!.messages.some(m => m.content.some(b => 'text' in b && b.text === 'second'))).toBe(true) }) - it('a throwing turn-end listener is caught by the backstop and loop survives', async () => { + it('a throwing agent/turn-end listener surfaces via agent/error and the loop survives', async () => { const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create('a1', { model: 'mock' }) @@ -84,7 +89,8 @@ describe('loop backstop catch', () => { send(agent, 'first') await waitForIdle(ctx, agent) // The turn-end throw happens after the model call is complete, so turn 1's - // request is consumed. The error is surfaced by the backstop. + // request is consumed. turn/end is already in the log (append pushes before + // notifying), so the turn is balanced; the error is surfaced via agent/error. expect(errors.map(e => e.message)).toEqual(['broken turn-end listener']) // loop survives: second turn works fine @@ -92,6 +98,35 @@ describe('loop backstop catch', () => { await waitForIdle(ctx, agent) expect(adapter.requests).toHaveLength(2) }) + + it('a pre-push turn/start failure (non-serializable source) is rethrown to the runLoop backstop', async () => { + // A non-serializable message source makes the turn/start append throw BEFORE + // the event is pushed (Session.append validates before push), so turn/start + // never enters the log. runTurn sees no logged turn/start and rethrows; the + // runLoop backstop reports via agent/error (step 0) + the logger and the + // driver survives. This is the ONLY path that reaches the backstop. + const adapter = new MockAdapter([textResponse('turn 2')]) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + + const errors: { turn: number; step: number; message: string }[] = [] + ctx.on('agent/error', (_a, turn, step, error) => void errors.push({ turn, step, message: error.message })) + + // A non-serializable source (BigInt) on the queued message. + agent.send([{ type: 'text', text: 'first' }], { source: { kind: 'plugin', plugin: 'p', bad: 1n } as never }) + await waitForIdle(ctx, agent) + + expect(errors).toHaveLength(1) + expect(errors[0]!.step).toBe(0) + expect(errors[0]!.message).toMatch(/non-JSON-serializable/) + // No turn boundary was written (the turn/start append threw before push). + expect(agent.session.events.some(e => e.type === 'turn/start')).toBe(false) + + // loop survives: a well-formed second turn runs normally. + send(agent, 'second') + await waitForIdle(ctx, agent) + expect(adapter.requests).toHaveLength(1) + }) }) describe('tool JSON parse', () => { @@ -157,7 +192,7 @@ describe('tool JSON parse', () => { }) describe('toError normalization', () => { - it('normalizes non-Error throws from turn-start listeners via toError in the backstop', async () => { + it('normalizes non-Error throws from turn-start listeners via toError', async () => { const adapter = new MockAdapter([textResponse('ok')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create('a1', { model: 'mock' }) @@ -166,7 +201,7 @@ describe('toError normalization', () => { ctx.on('agent/turn-start', () => { if (!threwOnce) { threwOnce = true - throw 'naked string error' // non-Error throw, goes through backstop's toError + throw 'naked string error' // non-Error throw, normalized via toError } }) diff --git a/packages/agent-loop/tests/review-fixes.spec.ts b/packages/agent-loop/tests/review-fixes.spec.ts index d6fa3870b9..438e246009 100644 --- a/packages/agent-loop/tests/review-fixes.spec.ts +++ b/packages/agent-loop/tests/review-fixes.spec.ts @@ -822,13 +822,15 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar expect(errorEmits).toHaveLength(0) }) - it('a throw at the turn/start append (before turnStarted) is rethrown to the runLoop backstop', async () => { - // A session/event listener that throws specifically on the turn/start - // event makes session.append('turn/start') throw while turnStarted is - // still false. runTurn must NOT try to close a turn it never opened — it - // rethrows, and the runLoop backstop records the error and survives. - // (Uses the plain harness — NOT the invariants oracle — because the - // throwing listener is itself a session/event subscriber.) + it('a throwing session/event listener on the turn/start append still balances the turn', async () => { + // Session.append pushes the event BEFORE notifying session/event listeners, + // so a listener throwing on turn/start leaves turn/start IN THE LOG. The + // loop must therefore still owe (and append) a turn/end — deciding "owed" + // from the log via isTurnOpen, not a "turn started" flag that the throw + // skipped. Otherwise the turn stays permanently open and poisons the next + // turn/replay (ADR 0017). (Uses the plain harness — NOT the invariants + // oracle — because the throwing listener is itself a session/event + // subscriber.) const adapter = new MockAdapter([textResponse('turn 2')]) const ctx = await harness(adapter) const agent = ctx.agentLoop.create('a-preturn', { model: 'mock' }) @@ -843,10 +845,18 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar send(agent, 'go') await waitForIdle(ctx, agent) - // The backstop logged exactly one error for the failed pre-turn append. + // The error was surfaced exactly once via agent/error. expect(errors.map(e => e.message)).toEqual(['boom turn/start append']) - // No turn/end was appended (none is owed — the turn never opened). - expect([...agent.session.events].some(e => e.type === 'turn/end')).toBe(false) + // The turn is BALANCED: turn/start is in the log (it was pushed before the + // listener threw), so a turn/end was owed and appended — no open turn. The + // last turn-boundary event being turn/end is exactly the loop's isTurnOpen + // check (no open turn remains). + const types = [...agent.session.events].map(e => e.type) + expect(types.filter(t => t === 'turn/start')).toHaveLength(1) + expect(types.filter(t => t === 'turn/end')).toHaveLength(1) + const lastBoundary = [...agent.session.events].reverse().find(e => e.type === 'turn/start' || e.type === 'turn/end') + expect(lastBoundary?.type).toBe('turn/end') + expect(agent.session.events.at(-1)?.type).toBe('turn/end') // loop survives: a second turn runs normally. send(agent, 'second') diff --git a/packages/agent/README.md b/packages/agent/README.md index 5bffc4681b..73914ea05b 100644 --- a/packages/agent/README.md +++ b/packages/agent/README.md @@ -53,7 +53,7 @@ The handle every plugin programs against: - `agent.send(content, options?)` — queue a message; starts a turn when idle - `agent.steer(content, options?)` — steer a running turn (inject between steps); behaves like `send` when idle -- `agent.inject(content, options?)` — inject in-session context without triggering a turn (context/message event); next request sees it +- `agent.inject(content, options?)` — inject in-session context (context/message event); the next request sees it. Does not run the model. While a turn is open it joins that turn; while idle it is wrapped in a one-shot `injection` turn so every event stays turn-enclosed (ADR 0017) - `agent.abort(reason?)` — abort the in-flight step - `agent.session`, `agent.status`, `agent.options`, `agent.id` diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index d6adc4760e..cee1e02666 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -68,7 +68,9 @@ export interface Agent { * an inject while idle wraps its `context/message` in a one-shot `injection` * turn (`turn/start` → `context/message` → `turn/end`) and checkpoints it for * durability, so every event stays inside a turn and a persistence backend - * never loses a between-turn notice. + * never loses a between-turn notice. The idle checkpoint is fire-and-forget + * (inject is synchronous): a failing flush is reported via `agent/error` + * (step `0`) and the logger, never thrown into the caller. * * TODO(review): exact envelope/rendering rules live in dsh-session and need * review once a real adapter exists. diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts index 13f908b451..b96d172c22 100644 --- a/packages/invariants/tests/invariants.spec.ts +++ b/packages/invariants/tests/invariants.spec.ts @@ -277,12 +277,15 @@ describe('dev-freeze', () => { // mutable. deepFreeze must descend into the already-frozen object and // freeze the descendant, not short-circuit on the frozen container — // otherwise dev-mode misses exactly the history mutation ADR 0012 catches. + // `append` snapshots `data`, so the freeze applies to the LOGGED clone, not + // the caller's input — read the event back and assert on its data. const innerContent: { type: 'text'; text: string }[] = [{ type: 'text', text: 'inner' }] const block = Object.freeze({ type: 'tool-result' as const, toolCallId: CallId('c1'), content: innerContent, isError: false }) - session.append('user/message', { content: [block], source: { kind: 'user' } }) - expect(Object.isFrozen(block.content)).toBe(true) - expect(Object.isFrozen(block.content[0])).toBe(true) - expect(() => { block.content.push({ type: 'text', text: 'mutation' }) }).toThrow() + const event = session.append('user/message', { content: [block], source: { kind: 'user' } }) + const logged = event.data.content[0] as { content: { type: 'text'; text: string }[] } + expect(Object.isFrozen(logged.content)).toBe(true) + expect(Object.isFrozen(logged.content[0])).toBe(true) + expect(() => { logged.content.push({ type: 'text', text: 'mutation' }) }).toThrow() }) it('terminates on a cyclic event datum (WeakSet guard)', async () => { diff --git a/packages/session-persistence-jsonl/src/index.ts b/packages/session-persistence-jsonl/src/index.ts index 8f45dccd6a..47d593d3c0 100644 --- a/packages/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence-jsonl/src/index.ts @@ -23,7 +23,7 @@ import { Context } from 'cordis' import z from 'schemastery' -import { open, mkdir, readFile, readdir, rename, link, rm, writeFile, truncate } from 'node:fs/promises' +import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node:fs/promises' import { resolve } from 'node:path' import { randomBytes } from 'node:crypto' import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' @@ -103,6 +103,19 @@ function assertSerializable(events: readonly SessionEvent[]): void { } } +/** + * Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY + * filesystem error that legitimately means "this session/root is absent" for a + * durable backend. Any OTHER error (`EACCES`, `ENOTDIR`, transient I/O) must + * surface rather than be silently reported as absence: masking it would let + * `list()` report no sessions, `load()` report "not found", and collision + * checks proceed under a false absence assumption — all unsafe for durable + * persistence. (A NodeJS filesystem rejection carries a string `code`.) + */ +function isENOENT(error: unknown): boolean { + return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT' +} + /** * The JSONL persistence backend. Load as a plugin; it registers as * `ctx.sessionPersistence` and installs the write-path listeners. @@ -504,7 +517,17 @@ export class SessionPersistenceJsonl extends SessionPersistence { ...meta.firstPrompt !== undefined ? { firstPrompt: meta.firstPrompt } : {}, } const tmp = `${path}.${randomBytes(6).toString('hex')}.tmp` - await writeFile(tmp, JSON.stringify(summary), { mode: 0o600 }) + // Exclusive owner-only create ('wx', 0o600), matching the log-materialization + // temp write: the sidecar can carry user data (title/firstPrompt), so a + // predictable/pre-existing temp path must never be silently truncated and + // followed (symlink race / disclosure). The random suffix already makes a + // collision unlikely; 'wx' makes reuse an error rather than a clobber. + const handle = await open(tmp, 'wx', 0o600) + try { + await handle.writeFile(JSON.stringify(summary)) + } finally { + await handle.close() + } await rename(tmp, path) } @@ -550,8 +573,13 @@ export class SessionPersistenceJsonl extends SessionPersistence { try { const entries = await readdir(this.root, { withFileTypes: true }) return entries.filter(e => e.isDirectory()).map(e => `${this.root}/${e.name}`) - } catch { - return [] // root does not exist yet → no sessions + } catch (error) { + // ENOENT = the root has not been created yet → genuinely no sessions. + // Any other error (EACCES, ENOTDIR, transient I/O) must NOT be reported + // as "no sessions" — a durable backend cannot silently pretend persisted + // state is absent on a storage fault. + if (isENOENT(error)) return [] + throw error } } @@ -565,8 +593,12 @@ export class SessionPersistenceJsonl extends SessionPersistence { const handle = await open(path, 'r') await handle.close() return true - } catch { - return false + } catch (error) { + // Only ENOENT means absent. A permission/I/O error must surface, not be + // collapsed to `false` — otherwise load() reports "not found" and + // collision checks proceed under a false absence assumption. + if (isENOENT(error)) return false + throw error } } diff --git a/packages/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence-jsonl/tests/jsonl.spec.ts index c63d05226d..a91e6f9138 100644 --- a/packages/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence-jsonl/tests/jsonl.spec.ts @@ -1014,6 +1014,36 @@ describe('SessionPersistenceJsonl: edge cases', () => { await ctx2.fiber.dispose() }) + it('list surfaces a non-ENOENT root error (ENOTDIR) instead of reporting no sessions', async () => { + // A durable backend must NOT collapse a storage fault to "no sessions". Point + // the root at a regular FILE: readdir then fails with ENOTDIR, which must + // propagate rather than be swallowed as an empty listing. + const filePath = join(root, 'not-a-dir') + await writeFile(filePath, 'x') + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + await ctx2.plugin(SessionPersistenceJsonl, { root: filePath }) + await expect(ctx2.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/) + await ctx2.fiber.dispose() + }) + + it('exists() surfaces a non-ENOENT lookup error (ENOTDIR) instead of reporting absent', async () => { + // Same contract on the existence path: a non-ENOENT error from the per-id + // open() must surface, not be collapsed to "not found" (which would let a + // collision check proceed under a false absence assumption). A LAZY session + // (created, never appended) keeps its cwd in state, so has() reaches + // findLog(id, cwd) → exists(logPath). Make that cwd's bucket DIRECTORY a + // regular file: open()ing `bucket/.jsonl` under it then fails ENOTDIR. + const cwd = '/x' + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + await ctx2.sessionPersistence.create(meta('exists-fault', cwd)) // lazy: no bucket yet + await writeFile(sessionDir(root, cwd), 'x') // bucket path is now a FILE + await expect(ctx2.sessionPersistence.has(SessionId('exists-fault'))).rejects.toThrow(/ENOTDIR/) + await ctx2.fiber.dispose() + }) + it('append() to a disk-only session adopts it and repairs a crash tail', async () => { // Persist a session, then corrupt its tail, all through ONE backend. const m = meta('disk-append', '/d') diff --git a/packages/session-persistence/tests/contract.ts b/packages/session-persistence/tests/contract.ts index 94f2ab9e4e..3c4e739b60 100644 --- a/packages/session-persistence/tests/contract.ts +++ b/packages/session-persistence/tests/contract.ts @@ -120,13 +120,31 @@ export function runPersistenceContract(name: string, make: () => Promise { const { persistence, dispose } = await make() try { - const m = meta('s5') - await persistence.create(m) - // A plugin-added event carrying a BigInt (not JSON-serializable). - const bad = [ - { type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: 1n } }, - ] as unknown as SessionEvent[] - await expect(persistence.append(m.id, bad)).rejects.toThrow(/user\/message/) + // Every value `isJsonValue` rejects must be rejected by the backend, not + // just BigInt — otherwise a backend could pass this contract while still + // accepting values that corrupt the durable round-trip. Each is a + // plugin-added `extra` field on a single user/message (seq 0). + const cyclic: Record = { type: 'text', text: 'x' } + cyclic['self'] = cyclic + const badValues: unknown[] = [ + 1n, // BigInt + undefined, // dropped by JSON.stringify + Infinity, // → null + () => 0, // function + Symbol('s'), // symbol + new Map(), // exotic object + cyclic, // circular ref + ] + for (const [i, bad] of badValues.entries()) { + // A fresh session per value isolates each rejection (a rejected append + // must leave no state behind, but isolating keeps the assertion clean). + const mi = meta(`s5-${i}`) + await persistence.create(mi) + const events = [ + { type: 'user/message', seq: 0, time: 1, data: { content: [{ type: 'text', text: 'x' }], source: { kind: 'user' }, extra: bad } }, + ] as unknown as SessionEvent[] + await expect(persistence.append(mi.id, events)).rejects.toThrow(/user\/message/) + } } finally { await dispose() } diff --git a/packages/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/tests/persistence.spec.ts index 0ce720cd51..9d211fff52 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 } from '@deepseek-ai/dsh-session' +import { SessionId, isJsonValue } 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' @@ -30,7 +30,7 @@ class MemoryPersistence extends SessionPersistence { for (let i = 0; i < events.length; i++) { const e = events[i]! if (e.seq !== nextSeq + i) throw new Error(`non-contiguous seq in batch for "${id}" at index ${i}`) - if (containsNonSerializable(e.data)) { + if (!isJsonValue(e.data)) { throw new Error(`event "${e.type}" carries non-JSON-serializable data`) } } @@ -68,15 +68,6 @@ class MemoryPersistence extends SessionPersistence { } } -/** Detect BigInt (and other JSON-hostile values) in event data. */ -function containsNonSerializable(value: unknown): boolean { - if (typeof value === 'bigint' || typeof value === 'function' || typeof value === 'symbol') return true - if (value && typeof value === 'object') { - return Object.values(value).some(containsNonSerializable) - } - return false -} - // Run the shared contract against the in-memory backend. runPersistenceContract('memory', async () => { const ctx = new Context() diff --git a/packages/session/README.md b/packages/session/README.md index 705e3d5b39..7cf443fa71 100644 --- a/packages/session/README.md +++ b/packages/session/README.md @@ -8,7 +8,7 @@ Creates and holds event-sourced `Session` instances. Persistence is intentionall ### Public API -- `ctx.sessions.create(id?: string, options?: { seed?: SessionEvent[]; meta?: { cwd?: string; parentSession?: SessionId } }): Session` — Create a session. `options.seed` replays/forks an existing event log; `options.meta` attaches creation metadata (validated absolute `cwd`, `parentSession` lineage) as the immutable `SessionHeader` (the store fills `version`/`id`/`createdAt`). Disposed with the calling fiber. +- `ctx.sessions.create(id?: string, options?: { seed?: SessionEvent[]; meta?: { cwd?: string; parentSession?: SessionId; createdAt?: number } }): Session` — Create a session. `options.seed` replays/forks an existing event log; `options.meta` attaches creation metadata (validated absolute `cwd`, `parentSession` lineage) as the immutable `SessionHeader`. The store fills `version`/`id` and defaults `createdAt` to now; a caller reconstructing a persisted session passes the original `createdAt` to preserve it. Disposed with the calling fiber. - `ctx.sessions.get(id: string): Session | undefined` - `ctx.sessions.list(): Session[]` diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 62cce89c7b..8fe54cb97b 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -90,7 +90,15 @@ export class Session { throw new Error(`seed event "${event.type}" (seq ${event.seq}) carries non-JSON-serializable data`) } }) - this.log = [...seed] + // Deep-clone each seed event, NOT just the array: the seed events and + // their `data` are still owned by the caller (or the source session of a + // fork), so keeping the references would let a post-create mutation of the + // original rewrite this session's durable log — or reintroduce a + // non-JSON-serializable value AFTER the validation above. Snapshotting at + // the boundary makes `session.events` independent and keeps it equal to + // what was validated. Serializability is guaranteed by the check above, so + // structuredClone can never hit a non-cloneable value here. + this.log = seed.map(event => structuredClone(event)) } this.header = header ?? { version: 1, id, createdAt: Date.now() } } @@ -120,7 +128,16 @@ export class Session { if (!isJsonValue(data)) { throw new Error(`session event "${type}" carries non-JSON-serializable data`) } - const event = { type, seq: this.log.length, time: Date.now(), data } as SessionEvent + // Snapshot `data` into the log, NOT the caller's reference: the validation + // above proves it is JSON-serializable AT THIS MOMENT, but the caller still + // owns the object and could mutate it afterwards (before a persistence + // flush, or permanently in the in-memory history) — making `session.events` + // diverge from the value that passed validation, or reintroducing a + // non-serializable value. Cloning here keeps the log equal to what was + // validated. structuredClone is safe because serializability was just + // checked. The returned event carries the SAME snapshot, so a caller reading + // back `event.data` sees the logged value, not its own mutable input. + const event = { type, seq: this.log.length, time: Date.now(), data: structuredClone(data) } as SessionEvent this.log.push(event) this.onAppend?.(event) return event diff --git a/packages/session/src/json.ts b/packages/session/src/json.ts index ebb0c00c64..99fe80ea80 100644 --- a/packages/session/src/json.ts +++ b/packages/session/src/json.ts @@ -22,6 +22,14 @@ * convert lossily. Sparse arrays are rejected too: a hole serializes to `null`, * so `[1, , 3]` would not round-trip. Detects circular references (which would * throw) and reports them as non-serializable rather than propagating the throw. + * + * Scope — matches `JSON.stringify` exactly: only an object's OWN ENUMERABLE + * STRING-keyed properties are inspected (`Object.values`). Symbol-keyed and + * non-enumerable properties are NOT examined, because `JSON.stringify` likewise + * drops them — they never reach the durable form, so a non-serializable value + * hiding under a symbol/non-enumerable key cannot make the round-trip lossy. + * Getters are invoked during the check (again as `JSON.stringify` would), so the + * contract is for plain data records, not objects with side-effecting accessors. */ export function isJsonValue(value: unknown, seen: Set = new Set()): boolean { if (value === null) return true diff --git a/packages/session/tests/session.spec.ts b/packages/session/tests/session.spec.ts index 39d29872ee..50dc9d5907 100644 --- a/packages/session/tests/session.spec.ts +++ b/packages/session/tests/session.spec.ts @@ -139,6 +139,39 @@ describe('Session', () => { const session = new Session(SessionId('seed-ok'), goodSeed) expect(session.events).toHaveLength(3) }) + + it('snapshots the seed: mutating the original after construction does not affect session.events', () => { + const seed = [ + { type: 'turn/start' as const, seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message' as const, source: { kind: 'user' as const } } } }, + { type: 'user/message' as const, seq: 1, time: 2, data: { content: [{ type: 'text' as const, text: 'original' }], source: { kind: 'user' as const } } }, + { type: 'turn/end' as const, seq: 2, time: 3, data: { turn: 1, reason: { kind: 'completed' as const } } }, + ] as SessionEvent[] + const session = new Session(SessionId('seed-snapshot'), seed) + // Mutate the ORIGINAL seed objects after construction: a shared reference + // would let this rewrite the forked log (or reintroduce non-serializable + // data past validation). The snapshot must shield session.events. + const um = seed[1]! + ;(um.data as { content: { type: 'text'; text: string }[] }).content[0]!.text = 'HACKED' + ;(um.data as Record)['injected'] = 1n // would have failed validation + const logged = session.events[1]! + expect(logged.type === 'user/message' && (logged.data.content[0] as { text: string }).text).toBe('original') + expect((logged.data as Record)['injected']).toBeUndefined() + }) + + it('snapshots append data: mutating the passed object after append does not affect session.events', () => { + const session = new Session(SessionId('append-snapshot')) + const data = { content: [{ type: 'text' as const, text: 'original' }], source: { kind: 'user' as const } } + const event = session.append('user/message', data) + // Mutate the caller's object after append returns. A shared reference would + // make session.events diverge from the value that passed validation. + data.content[0]!.text = 'HACKED' + ;(data as Record)['injected'] = 1n + const logged = session.events[0]! + expect(logged.type === 'user/message' && (logged.data.content[0] as { text: string }).text).toBe('original') + expect((logged.data as Record)['injected']).toBeUndefined() + // The returned event carries the same snapshot, not the caller's input. + expect((event.data.content[0] as { text: string }).text).toBe('original') + }) })