Merge branch 'split/agent-factory' into split/session-persistence-sqlite

docs/adr/0016-session-persistence.md

@@ -21,7 +21,7 @@ Key choices recorded here because they are durable, contested, and surprising:
 - **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.
 - **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 typed error when it is absent.
+- **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.
 
 Format versioning: the header carries a `version`; `load` rejects an unknown version (no v1 migration). Stated honestly: append-only + flush is robust to partial trailing writes (tolerated on load) but not to fsync-less power loss mid-line; a DB/WAL backend is the stronger option later.
 

packages/agent-loop/README.md

@@ -13,7 +13,7 @@ This is the only package in the harness that contains concrete loop logic. Every
 `AgentLoop` also implements the `AgentFactory` seam and registers itself via `ctx.agents.setFactory(this)`, so plugins create/resume agents through `ctx.agents` (the interface):
 
 - `ctx.agents.create({ agentId, sessionId, meta?, agentOptions? })` — programmatic create on a caller-supplied `sessionId` (e.g. an ACP-generated id), NOT `${id}-session`.
-- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions? })` — load a persisted session via `ctx.sessionPersistence` (RFC 009) and resume an agent on it. The live session id is the resumed id; turn numbering and derived history continue from the loaded log. Requires a session-persistence backend (NOT hard-injected — non-persistent demos still work; `resume` rejects with a typed error when persistence is absent).
+- `ctx.agents.resume({ agentId, resumeSessionId, agentOptions? })` — load a persisted session via `ctx.sessionPersistence` (RFC 009) and resume an agent on it. The live session id is the resumed id; turn numbering and derived history continue from the loaded log. Requires a session-persistence backend (NOT hard-injected — non-persistent demos still work; `resume` rejects with a clear error when persistence is absent).
 
 ### Injected services
 

packages/agent-loop/src/loop.ts

@@ -181,16 +181,27 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
   const closeStep = (): void => {
     if (!stepOpen) return
     stepOpen = false
-    session.append('step/end', { turn, step })
+    // Session.append pushes step/end BEFORE notifying session/event listeners,
+    // so a throwing listener leaves step/end in the log (balance holds) but
+    // would otherwise abort finalization. Contain it and surface it as a turn
+    // error below — the same outcome as a throwing agent/step-end listener.
+    let failure: unknown
+    try {
+      session.append('step/end', { turn, step })
+    } catch (error: unknown) {
+      failure = error
+    }
     try {
       ctx.emit('agent/step-end', agent, turn, step)
     } catch (error: unknown) {
-      // step/end is already recorded so balance holds; surface the throwing
-      // listener as a turn error via failTurn (idempotent). This prevents a
-      // throwing step-end listener from producing a silent "completed" turn
-      // when the step itself succeeded (the normal-path closeStep call).
-      failTurn(toError(error))
+      failure ??= error
     }
+    // A throwing step/end session-event listener OR a throwing agent/step-end
+    // listener surfaces as a turn error via failTurn (idempotent). This prevents
+    // a throwing listener from producing a silent "completed" turn when the step
+    // itself succeeded, AND keeps finalization going when closeStep runs from
+    // the outer catch.
+    if (failure !== undefined) failTurn(toError(failure))
   }
 
   // Record a step/turn failure exactly once: append the single `error` event
@@ -208,8 +219,18 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
     // backend treats it as a crash tail and drops it on resume (ADR 0017). In
     // that case report via agent/error + the logger only; the turn is balanced.
     if (!turnEnded) {
-      session.append('error', { turn, step, ...errorData(err) })
+      // Set `reason` BEFORE the append: Session.append pushes the error event
+      // before notifying session/event listeners, so a throwing listener would
+      // otherwise leave `reason` unset (and closeTurn would record the wrong
+      // reason / the outer catch would skip closeTurn). The append is contained
+      // — the error event is already in the log either way; a throwing listener
+      // must not abort finalization.
       reason = { kind: 'error', ...errorData(err) }
+      try {
+        session.append('error', { turn, step, ...errorData(err) })
+      } catch (appendError: unknown) {
+        ctx.logger.warn(`agent "${agent.id}": session/event listener threw on the error event at turn ${turn}: ${toError(appendError).message}`)
+      }
     } else {
       ctx.logger.warn(`agent "${agent.id}": agent/turn-end listener threw after turn ${turn} closed: ${err.message}`)
     }
@@ -229,7 +250,19 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn:
   const closeTurn = (emit: boolean): void => {
     if (turnEnded) return
     turnEnded = true
-    session.append('turn/end', { turn, reason })
+    // Session.append pushes turn/end BEFORE notifying session/event listeners,
+    // so a throwing listener leaves turn/end in the log (the turn is balanced)
+    // but would otherwise escape — from the outer catch's closeTurn(false) it
+    // would propagate to the runLoop backstop, and from the normal-path
+    // closeTurn(true) it would skip the agent/turn-end emit. Contain it: the
+    // boundary is durable either way, and finalization must not abort on a bad
+    // listener. (On the normal path the outer catch also re-runs closeTurn,
+    // which is an idempotent no-op once turnEnded is set.)
+    try {
+      session.append('turn/end', { turn, reason })
+    } catch (error: unknown) {
+      ctx.logger.warn(`agent "${agent.id}": session/event listener threw on turn/end at turn ${turn}: ${toError(error).message}`)
+    }
     if (emit) ctx.emit('agent/turn-end', agent, turn, reason)
   }
 

packages/agent-loop/tests/review-fixes.spec.ts

@@ -969,6 +969,110 @@ describe('P1-5: a started turn (and any open step) is always closed on a boundar
     await waitForIdle(ctx, agent)
     expect(boundaryCounts(agent).turnEnd).toBe(2)
   })
+
+  it('a throwing session/event listener on the error event still closes the turn (finalizer containment)', async () => {
+    // failTurn appends the `error` event; Session.append pushes it BEFORE
+    // notifying session/event listeners, so a throwing listener leaves `error`
+    // in the log but must NOT abort finalization — `reason` is set before the
+    // append and the throw is contained, so closeTurn(false) still runs and
+    // turn/end is appended (the turn is balanced, not left open).
+    // Plain harness (no invariants oracle): the throwing listener is itself a
+    // session/event subscriber. A finish-error drives the boundary-error path.
+    const errorStream: StreamChunk[] = [{ type: 'finish', reason: { kind: 'error', message: 'provider down' } }]
+    const adapter = new MockAdapter([errorStream, textResponse('turn 2 ok')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a-errthrow', { model: 'mock' })
+
+    let threw = false
+    ctx.on('session/event', (_s, event) => {
+      if (!threw && event.type === 'error') { threw = true; throw new Error('boom error-event listener') }
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    const e = [...agent.session.events]
+    // The error event is in the log (pushed before the listener threw)…
+    expect(e.some(x => x.type === 'error')).toBe(true)
+    // …and the turn was still closed with the error reason (finalization did not
+    // abort): the last event is turn/end carrying the error reason.
+    const last = e.at(-1)
+    expect(last?.type).toBe('turn/end')
+    expect(last?.type === 'turn/end' && last.data.reason).toMatchObject({ kind: 'error', message: 'provider down' })
+
+    // loop survives: a second turn runs normally.
+    send(agent, 'again')
+    await waitForIdle(ctx, agent)
+    expect(adapter.requests).toHaveLength(2)
+  })
+
+  it('a throwing session/event listener on step/end during finalization still appends turn/end', async () => {
+    // A throwing agent/step-start listener drives the outer catch, which calls
+    // closeStep() during finalization. closeStep appends step/end; a
+    // session/event listener throwing on THAT must not abort the catch before
+    // closeTurn(false) — step/end is already logged (balance holds) and the
+    // throw is contained + surfaced via failTurn, so turn/end is still appended.
+    const adapter = new MockAdapter([textResponse('never reached')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a-stependthrow', { model: 'mock' })
+
+    // Open a step, then make the agent/step-start emit throw (boundary throw →
+    // outer catch → closeStep during finalization).
+    ctx.on('agent/step-start', () => { throw new Error('boom step-start') })
+    let threw = false
+    ctx.on('session/event', (_s, event) => {
+      if (!threw && event.type === 'step/end') { threw = true; throw new Error('boom step/end listener') }
+    })
+    const errors: Error[] = []
+    ctx.on('agent/error', (_a, _t, _s, error) => void errors.push(error))
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+
+    const e = [...agent.session.events]
+    // Both step/end and turn/end are present — finalization ran to completion.
+    expect(e.some(x => x.type === 'step/end')).toBe(true)
+    expect(e.some(x => x.type === 'turn/end')).toBe(true)
+    expect(e.at(-1)?.type).toBe('turn/end')
+    expect(errors.length).toBeGreaterThanOrEqual(1) // surfaced via agent/error
+
+    // loop survives.
+    send(agent, 'again')
+    await waitForIdle(ctx, agent)
+    expect(e.filter(x => x.type === 'turn/start').length).toBeGreaterThanOrEqual(1)
+  })
+
+  it('a throwing session/event listener on turn/end is contained (turn still balanced, loop survives)', async () => {
+    // closeTurn appends turn/end; Session.append pushes it BEFORE notifying
+    // session/event listeners, so a throwing listener leaves turn/end in the log
+    // (the turn is balanced) but must not escape — from the normal-path
+    // closeTurn(true) it would otherwise propagate; the append is contained so
+    // the turn/end emit + loop continue. (A throwing agent/turn-end LISTENER is
+    // a separate, already-tested path; here the session/event append notify is
+    // what throws.)
+    const adapter = new MockAdapter([textResponse('turn 1'), textResponse('turn 2')])
+    const ctx = await harness(adapter)
+    const agent = ctx.agentLoop.create('a-turnendappend', { model: 'mock' })
+
+    let threw = false
+    ctx.on('session/event', (_s, event) => {
+      if (!threw && event.type === 'turn/end') { threw = true; throw new Error('boom turn/end listener') }
+    })
+
+    send(agent, 'go')
+    await waitForIdle(ctx, agent)
+    // turn 1 is balanced despite the throwing turn/end listener.
+    const e1 = [...agent.session.events]
+    expect(e1.filter(x => x.type === 'turn/start')).toHaveLength(1)
+    expect(e1.filter(x => x.type === 'turn/end')).toHaveLength(1)
+    expect(e1.at(-1)?.type).toBe('turn/end')
+
+    // loop survives: a second turn runs to completion.
+    send(agent, 'again')
+    await waitForIdle(ctx, agent)
+    expect(adapter.requests).toHaveLength(2)
+    expect([...agent.session.events].filter(x => x.type === 'turn/end')).toHaveLength(2)
+  })
 })
 
 describe('P1-7: tool/result is logged under the originating call.id, not result.callId', () => {

packages/session-persistence-jsonl/src/index.ts

@@ -424,20 +424,38 @@ export class SessionPersistenceJsonl extends SessionPersistence {
     // final path already exists, so two processes materializing the same id
     // concurrently cannot clobber each other (both could pass the exists() check
     // above, but only one link() wins). rename() would silently overwrite the
-    // log the other process just committed. The temp link is always removed,
-    // whether link() succeeds or throws (EEXIST on a race, or any I/O error).
+    // log the other process just committed.
+    let linked = false
     try {
       await link(tmp, finalPath)
+      linked = true
     } finally {
-      await rm(tmp, { force: true })
+      // If link FAILED (EEXIST on a race, or any I/O error), the temp is the
+      // only reference and must be removed before the original error propagates.
+      // If link SUCCEEDED, the temp cleanup is deferred to AFTER the publish is
+      // durable (below) so a temp-rm failure can never reject a session whose
+      // log already published — that would leave state.materialized false and
+      // wedge every retry on the exists() backstop above.
+      /* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
+      if (!linked) await rm(tmp, { force: true })
     }
-    // fsync the directory so the new entry survives a power loss: on POSIX
-    // filesystems the new link is not crash-durable until the parent directory's
-    // metadata is synced. The seam contract is "append returns once durable",
-    // and materialize is the first append's write — so the directory entry must
-    // be durable before we return.
+    // link() succeeded — the log is published. fsync the directory so the new
+    // entry survives a power loss: on POSIX filesystems the new link is not
+    // crash-durable until the parent directory's metadata is synced. The seam
+    // contract is "append returns once durable", and materialize is the first
+    // append's write — so the directory entry must be durable before we return.
     await this.syncDir(dir)
     state.materialized = true
+    // Best-effort temp cleanup: the log is already published and durable, so a
+    // failure to remove the (now-redundant) temp hard link must NOT reject the
+    // append. A leftover `*.tmp` is harmless — it is never read, and the next
+    // materialize of this id is guarded by exists()/link(). Swallow only the
+    // rm failure; nothing else of consequence runs in the try.
+    try {
+      await rm(tmp, { force: true })
+    } catch {
+      /* v8 ignore next -- redundant temp link; publish already durable, rm failure is an unreachable IO edge */
+    }
   }
 
   /** fsync a directory so a just-created/renamed entry inside it is crash-durable. */