From 5299e43bed538d28e856720dadedb8af1711870a Mon Sep 17 00:00:00 2001
From: Tianyi Cui <53024+tianyicui@users.noreply.github.com>
Date: Mon, 15 Jun 2026 23:33:00 +0800
Subject: [PATCH 1/4] fix(session): snapshot seed + appended data at the
 boundary (review #31)

The source-level JSON-serializability invariant was only a preflight: the
Session constructor copied the seed array but shared every event/data
object with the caller, and append() stored the caller's `data` reference
verbatim. A post-create/post-append mutation could rewrite the durable
log or reintroduce a non-JSON-serializable value AFTER validation, so
session.events could diverge from what was validated / what a backend
can persist.

- ctor deep-clones each seed event after validation (not just the array).
- append() stores structuredClone(data) (serializability already checked,
  so the clone is safe); the returned event carries the same snapshot.

Regression tests: mutating the original seed / the passed append object
after the call leaves session.events unchanged. Adapted the dev-freeze
invariants test to assert on the logged clone (append no longer freezes
the caller's input). Documented isJsonValue's exact scope (own enumerable
string keys, matching JSON.stringify) and synced the README create()
signature with meta.createdAt.
---
 packages/invariants/tests/invariants.spec.ts | 11 ++++---
 packages/session/README.md                   |  2 +-
 packages/session/src/index.ts                | 21 +++++++++++--
 packages/session/src/json.ts                 |  8 +++++
 packages/session/tests/session.spec.ts       | 33 ++++++++++++++++++++
 5 files changed, 68 insertions(+), 7 deletions(-)

diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts
index bbf93f44dc..a2e205457f 100644
--- a/packages/invariants/tests/invariants.spec.ts
+++ b/packages/invariants/tests/invariants.spec.ts
@@ -241,12 +241,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/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<T>
+    // 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<T>
     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<object> = 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<string, unknown>)['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<string, unknown>)['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<string, unknown>)['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<string, unknown>)['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')
+  })
 })
 
 

From 4535bfab755db8a799b76295dd6da7cda7d600bd Mon Sep 17 00:00:00 2001
From: Tianyi Cui <53024+tianyicui@users.noreply.github.com>
Date: Mon, 15 Jun 2026 23:44:54 +0800
Subject: [PATCH 2/4] fix(agent-loop): decide turn balance + idle-injection
 flush from the log (review #32)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Session.append pushes the event BEFORE notifying session/event listeners,
so a throwing listener leaves the event in the log while the line after
the append (a boolean flag) never runs. Both turn-balance decisions were
gated on such flags, so a throwing listener could strand an open turn or
skip a durability checkpoint.

- loop.ts: the outer catch decided "turn/end owed" from `turnStarted`.
  A throwing listener on the turn/start append left turn/start logged but
  the flag false → catch rethrew and skipped turn/end → permanently open
  turn (violating ADR 0017). Now decided from the log (this turn's
  turn/start present), so the turn is always balanced; only a genuine
  pre-push failure (non-serializable trigger — turn/start never logged) is
  rethrown to the runLoop backstop. Removed the now-dead `turnStarted`.

- agent.ts inject(): the idle one-shot-turn flush was gated on a
  `turnRecorded` flag set after append('turn/end'); a throwing turn/end
  listener skipped the flush, losing the balanced in-memory injection turn
  on crash. Now the flush decision is read from the log, the synthetic
  turn/end append contains a throwing listener (turn stays balanced), and
  a failing idle flush is reported via agent/error (step 0 convention) AND
  the logger — mirroring the loop's post-turn/end flush path — with a
  throwing agent/error listener contained.

Rewrote the test that encoded the old (buggy) "turn/start listener throw
is rethrown, no turn/end" semantics to assert the balanced-turn contract,
and added regressions for the throwing-turn/end-listener flush and the
agent/error report. Updated Agent.inject JSDoc.
---
 packages/agent-loop/src/agent.ts              | 42 ++++++++++++----
 packages/agent-loop/src/loop.ts               | 26 ++++++----
 packages/agent-loop/tests/agent.spec.ts       | 44 +++++++++++++++++
 .../agent-loop/tests/coverage-edges.spec.ts   | 49 ++++++++++++++++---
 .../agent-loop/tests/review-fixes.spec.ts     | 30 ++++++++----
 packages/agent/src/types.ts                   |  4 +-
 6 files changed, 158 insertions(+), 37 deletions(-)

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/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/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.

From 01621d38b68218087ce4af4af003818de37c5520 Mon Sep 17 00:00:00 2001
From: Tianyi Cui <53024+tianyicui@users.noreply.github.com>
Date: Mon, 15 Jun 2026 23:53:03 +0800
Subject: [PATCH 3/4] fix(session-persistence-jsonl): surface non-ENOENT
 storage errors; harden sidecar; broaden contract (review #33)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A durable persistence backend must not treat a storage fault as absence.
listCwdDirs() and exists() swallowed EVERY error and reported "no
sessions" / "not found", so EACCES/ENOTDIR/transient I/O could make
list() return nothing, load() report not-found, and collision checks
proceed under a false absence assumption.

- Add an isENOENT() helper; listCwdDirs() and exists() now return the
  empty/absent result ONLY for ENOENT and rethrow every other error.
  Regression tests drive ENOTDIR through both paths.

TODO-level hardening also addressed:
- writeSidecar() now uses an exclusive owner-only temp open ('wx', 0o600)
  like the log-materialization path, instead of a truncating writeFile —
  the sidecar can carry user data (title/firstPrompt), so a predictable/
  pre-existing temp path must never be silently followed.
- The shared runPersistenceContract serializability case now exercises
  EVERY value isJsonValue rejects (BigInt, undefined, Infinity, function,
  symbol, Map, circular), not just BigInt, so a backend cannot pass the
  contract while accepting values that corrupt the round-trip. The mock
  MemoryPersistence now validates via the canonical isJsonValue.
---
 .../session-persistence-jsonl/src/index.ts    | 44 ++++++++++++++++---
 .../tests/jsonl.spec.ts                       | 30 +++++++++++++
 .../session-persistence/tests/contract.ts     | 32 +++++++++++---
 .../tests/persistence.spec.ts                 | 13 +-----
 4 files changed, 95 insertions(+), 24 deletions(-)

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/<id>.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<Contrac
     it('append rejects non-JSON-serializable event data, naming the event type', async () => {
       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<string, unknown> = { 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()

From 5284ed4806c30cb4661b7242905075a372b7b4e0 Mon Sep 17 00:00:00 2001
From: Tianyi Cui <53024+tianyicui@users.noreply.github.com>
Date: Mon, 15 Jun 2026 23:55:40 +0800
Subject: [PATCH 4/4] docs(agent): sync inject wording + resume error wording
 with code (review #34)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- agent/README: the inject() line said "without triggering a turn",
  regressing the #32 turn-enclosure model. Restored the running-vs-idle
  wording (idle inject wraps a one-shot injection turn; ADR 0017) to match
  the interface JSDoc.
- agent-loop resume() JSDoc said "throws a typed error" but the code
  throws a plain Error (consistent with the sibling assertAgentIdFree
  throw). Softened to "rejects with a clear error" — no behavior change;
  plain Error is intentional (no consumer needs a structured code here).
---
 packages/agent-loop/src/index.ts | 2 +-
 packages/agent/README.md         | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

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/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`