test: close session preparation CI gaps

packages/core/agent-loop/tests/resume.spec.ts

@@ -1,5 +1,5 @@
 import { createUserMessage } from '@deepseek-ai/dsh-llm'
-import { afterEach, describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
 import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
@@ -747,6 +747,24 @@ describe('creation and resume cancellation edges', () => {
     await ctx.fiber.dispose()
   })
 
+  it('rejects when setup synchronously aborts its caller signal', async () => {
+    const { ctx } = await persistentHarness(new MockAdapter([]))
+    const controller = new AbortController()
+
+    const creating = ctx.agents.create({
+      sessionId: SessionId('setup-synchronous-abort'),
+      agentOptions: { provider: 'mock', model: 'mock' },
+      signal: controller.signal,
+      setup() {
+        controller.abort(new Error('setup synchronously cancelled'))
+      },
+    })
+
+    await expect(promptly(creating)).rejects.toThrow('setup synchronously cancelled')
+    expect(ctx.agents.get(SessionId('setup-synchronous-abort'))).toBeUndefined()
+    await ctx.fiber.dispose()
+  })
+
   it('resume with a pre-aborted caller signal rejects out of the load race', async () => {
     const sessionId = SessionId('resume-pre-aborted')
     const root = await persistSession(sessionId)
@@ -760,6 +778,31 @@ describe('creation and resume cancellation edges', () => {
       signal: controller.signal,
     }))).rejects.toThrow('resume abandoned')
 
+    const stringReason = new AbortController()
+    stringReason.abort('resume string reason')
+    await expect(promptly(ctx.agents.resume({
+      resumeSessionId: sessionId,
+      agentOptions: { provider: 'mock', model: 'mock' },
+      signal: stringReason.signal,
+    }))).rejects.toThrow(/creation aborted/)
+
+    expect(ctx.agents.get(sessionId)).toBeUndefined()
+    await ctx.fiber.dispose()
+  })
+
+  it('releases a restored preparation if the loop becomes inactive before setup', async () => {
+    const sessionId = SessionId('resume-loop-inactive-after-prepare')
+    const root = await persistSession(sessionId)
+    const ctx = await mountPersistentHarness(root, new MockAdapter([]))
+    const loop = ctx.agentLoop as unknown as {
+      ownership: { isActive: () => boolean }
+    }
+    vi.spyOn(loop.ownership, 'isActive').mockReturnValueOnce(false)
+
+    await expect(ctx.agents.resume({
+      resumeSessionId: sessionId,
+      agentOptions: { provider: 'mock', model: 'mock' },
+    })).rejects.toThrow('agent loop is not active')
     expect(ctx.agents.get(sessionId)).toBeUndefined()
     await ctx.fiber.dispose()
   })

packages/core/session/tests/session.spec.ts

@@ -998,6 +998,15 @@ describe('Session', () => {
 
     expect(() => Session.create(SessionId('header-invalid'), undefined, new ExoticHeader()))
       .toThrow(/not losslessly JSON-serializable/)
+    expect(() => Session.fromRestore(SessionId('header-invalid'), [], new ExoticHeader()))
+      .toThrow(/not a plain JSON record/)
+    for (const header of [null, 1, []]) {
+      expect(() => Session.fromRestore(
+        SessionId('header-invalid'),
+        [],
+        header as unknown as SessionHeader,
+      )).toThrow(/not a plain JSON record/)
+    }
     expect(() => Session.create(SessionId('header-invalid'), undefined, {
       version: SESSION_FORMAT_VERSION,
       id: SessionId('header-invalid'),

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

@@ -109,7 +109,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     // every storage hook awaits the same readiness promise.
     this.ready = this.openDb(config.path, (config as Required<Config>).journalMode)
     this.coordinator = new PersistenceCoordinator<number>(this.ctx, this, {
-      preparedSessionCacheSize: config.preparedSessionCacheSize ?? DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+      preparedSessionCacheSize: (config as Required<Config>).preparedSessionCacheSize,
     })
   }
 

packages/session-persistence/session-persistence-sqlite/tests/sqlite.spec.ts

@@ -641,6 +641,23 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
 })
 
 describe('SessionPersistenceSqlite: edge cases', () => {
+  it('uses the configured preparation cache through the public service', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(SessionPersistenceSqlite, {
+      path: ':memory:',
+      preparedSessionCacheSize: 1,
+    })
+    const m = meta('sqlite-preparation-cache')
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, oneTurnLog())
+
+    const preparation = await ctx.sessionPersistence.prepare(m.id)
+    expect(preparation.session.header).toEqual(m)
+    preparation[Symbol.dispose]()
+    await fiber.dispose()
+  })
+
   it('rejects and closes a current-schema database with an invalid store identity', async () => {
     const path = await freshDbPath()
     const db = openDatabase(path, 'wal')

packages/session-persistence/session-persistence/tests/persistence.spec.ts

@@ -1,7 +1,7 @@
 import { describe, expect, it, vi } from 'vitest'
 import { Context } from 'cordis'
-import SessionStore, { SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
+import SessionStore, { Session, SessionId, isJsonValue } from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
   SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
   type PersistenceBackend, type SessionPersistenceSnapshot, type StoredPrefix,
@@ -371,6 +371,162 @@ describe('PersistenceCoordinator stored identity', () => {
 })
 
 describe('PersistenceCoordinator session preparations', () => {
+  it.each([0, 1.5])('rejects invalid preparation cache capacity %s', (capacity) => {
+    const ctx = new Context()
+    const backend = new ControlledBackend()
+
+    expect(() => new PersistenceCoordinator(ctx, backend, {
+      preparedSessionCacheSize: capacity,
+    })).toThrow(/positive safe integer/)
+  })
+
+  it('retries invalidated prepare and load reservations', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const prepareId = SessionId('prepare-reservation-retry')
+    const loadId = SessionId('load-reservation-retry')
+    backend.store.set(prepareId, { meta: meta(prepareId), events: oneTurnLog() })
+    backend.store.set(loadId, { meta: meta(loadId), events: oneTurnLog() })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const preparations = (coordinator as unknown as {
+      preparations: { reserve: (...args: unknown[]) => Promise<unknown> }
+    }).preparations
+    const reserve = vi.spyOn(preparations, 'reserve')
+
+    try {
+      reserve.mockResolvedValueOnce(undefined)
+      const preparation = await coordinator.prepare(prepareId)
+      preparation[Symbol.dispose]()
+
+      reserve.mockResolvedValueOnce(undefined)
+      await expect(coordinator.load(loadId)).resolves.toMatchObject({ meta: { id: loadId } })
+    } finally {
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('prefers a session that becomes live across preparation reads', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const prepareId = SessionId('prepare-became-live')
+    const loadId = SessionId('load-became-live')
+    const inspectId = SessionId('inspect-became-live')
+    const failedInspectId = SessionId('failed-inspect-became-live')
+    for (const id of [prepareId, loadId, inspectId]) {
+      backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
+    }
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      const prepareLive = Session.create(prepareId, oneTurnLog(), meta(prepareId))
+      const prepareGet = vi.spyOn(ctx.sessions, 'get')
+        .mockReturnValueOnce(undefined)
+        .mockReturnValueOnce(prepareLive)
+      await expect(coordinator.prepare(prepareId)).rejects.toThrow(/while it is live/)
+      prepareGet.mockRestore()
+
+      const loadLive = Session.create(loadId, oneTurnLog(), meta(loadId))
+      const loadGet = vi.spyOn(ctx.sessions, 'get')
+        .mockReturnValueOnce(undefined)
+        .mockReturnValueOnce(loadLive)
+      await expect(coordinator.load(loadId)).resolves.toMatchObject({ meta: { id: loadId } })
+      loadGet.mockRestore()
+
+      const inspectLive = Session.create(inspectId, oneTurnLog(), meta(inspectId))
+      const inspectGet = vi.spyOn(ctx.sessions, 'get')
+        .mockReturnValueOnce(undefined)
+        .mockReturnValueOnce(inspectLive)
+      await expect(coordinator.inspect(inspectId)).resolves.toMatchObject({ meta: { id: inspectId } })
+      inspectGet.mockRestore()
+
+      const failedInspectLive = Session.create(failedInspectId, oneTurnLog(), meta(failedInspectId))
+      backend.beforeLoadStored = () => Promise.reject(new Error('load failed'))
+      const failedInspectGet = vi.spyOn(ctx.sessions, 'get')
+        .mockReturnValueOnce(undefined)
+        .mockReturnValueOnce(failedInspectLive)
+      await expect(coordinator.inspect(failedInspectId))
+        .resolves.toMatchObject({ meta: { id: failedInspectId } })
+      failedInspectGet.mockRestore()
+    } finally {
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('rejects a prepared commit when durable state already has a live owner', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('prepared-commit-live-owner')
+    backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const owner = Session.create(id, oneTurnLog(), meta(id))
+    const states = (coordinator as unknown as {
+      states: Map<SessionId, {
+        meta: SessionHeader
+        cursor: number
+        materialized: boolean
+        owner?: Session
+      }>
+    }).states
+    states.set(id, {
+      meta: owner.header,
+      cursor: oneTurnLog().length,
+      materialized: true,
+      owner,
+    })
+
+    try {
+      await expect(coordinator.prepare(id)).rejects.toThrow(/live persistence owner/)
+    } finally {
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('rejects publication after a preparation state no longer matches', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const id = SessionId('prepared-publication-mismatch')
+    backend.store.set(id, { meta: meta(id), events: oneTurnLog() })
+    let coordinator!: PersistenceCoordinator<never>
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      coordinator = new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+    const preparation = await coordinator.prepare(id)
+    const preparations = (coordinator as unknown as {
+      preparations: {
+        reservationFor: (session: Session) => { state: { cursor: number } } | undefined
+      }
+    }).preparations
+    const reservation = preparations.reservationFor(preparation.session)
+    if (reservation === undefined) throw new Error('test preparation must stay reserved')
+    reservation.state.cursor += 1
+    const detach = ctx.sessions.enter(preparation.session)
+
+    try {
+      expect(() => { ctx.sessions.announce(preparation.session) }).toThrow(/no longer matches/)
+    } finally {
+      detach()
+      preparation[Symbol.dispose]()
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
   it('reuses the exact Session from inspect through repeated unpublished prepare calls', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
@@ -1112,6 +1268,50 @@ describe('PersistenceCoordinator retirement', () => {
 })
 
 describe('SessionPersistence service registration', () => {
+  it('provides a cancellation-aware default preparation for simple backends', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(MemoryPersistence)
+    const m = meta('default-preparation')
+    await ctx.sessionPersistence.create(m)
+    await ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const defaultPrepare = SessionPersistence.prototype.prepare.bind(ctx.sessionPersistence)
+
+    const preparation = await defaultPrepare(m.id)
+    expect(preparation.session.header).toEqual(m)
+    preparation[Symbol.dispose]()
+
+    const preAborted = new AbortController()
+    const preAbortReason = new Error('pre-aborted preparation')
+    preAborted.abort(preAbortReason)
+    await expect(defaultPrepare(m.id, preAborted.signal))
+      .rejects.toBe(preAbortReason)
+
+    const postAborted = new AbortController()
+    const postAbortReason = new Error('post-load preparation abort')
+    const originalLoad = ctx.sessionPersistence.load.bind(ctx.sessionPersistence)
+    ctx.sessionPersistence.load = async (id) => {
+      const loaded = await originalLoad(id)
+      postAborted.abort(postAbortReason)
+      return loaded
+    }
+    await expect(defaultPrepare(m.id, postAborted.signal))
+      .rejects.toBe(postAbortReason)
+
+    await fiber.dispose()
+  })
+
+  it('requires SessionStore for the default preparation', async () => {
+    const id = SessionId('default-preparation-without-store')
+    const persistence = {
+      ctx: new Context(),
+      load: () => Promise.resolve({ meta: meta(id), events: oneTurnLog() }),
+    } as unknown as SessionPersistence
+
+    await expect(SessionPersistence.prototype.prepare.call(persistence, id))
+      .rejects.toThrow(/SessionStore is not configured/)
+  })
+
   it('registers as ctx.sessionPersistence and is removed on fiber dispose (HMR safety)', async () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)

packages/session-persistence/session-persistence/tests/preparations.spec.ts

@@ -98,7 +98,7 @@ describe('SessionPreparations reservation', () => {
     expect(first).toBeDefined()
     expect(preparations.reservationFor(source.session)).toBe(first)
     expect(() => preparations.reservationFor(Session.create(id))).toThrow(/cannot publish/)
-    expect(() => preparations.assertWritable(id)).toThrow(/is reserved/)
+    expect(() => { preparations.assertWritable(id) }).toThrow(/is reserved/)
 
     let secondSettled = false
     const secondPromise = preparations.reserve(id, () => Promise.resolve(prepared('unused')), committed)
@@ -116,10 +116,10 @@ describe('SessionPreparations reservation', () => {
     expect(second?.source).toBe(source)
     preparations.attach(second!)
     expect(preparations.reservationFor(source.session)).toBeUndefined()
-    expect(() => preparations.attach(second!)).toThrow(/no longer reserved/)
+    expect(() => { preparations.attach(second!) }).toThrow(/no longer reserved/)
     preparations.discard(second!)
     preparations.release(second!, true)
-    expect(() => preparations.assertWritable(id)).not.toThrow()
+    expect(() => { preparations.assertWritable(id) }).not.toThrow()
   })
 
   it('supports abortable reservation waits without cancelling the held reservation', async () => {
@@ -152,7 +152,7 @@ describe('SessionPreparations reservation', () => {
       return commitGate.promise
     })
     await commitStarted.promise
-    expect(() => preparations.assertWritable(id)).toThrow(/is reserved/)
+    expect(() => { preparations.assertWritable(id) }).toThrow(/is reserved/)
     const second = preparations.reserve(id, () => Promise.resolve(prepared('unused')), committed)
 
     commitGate.reject(failure)
@@ -168,7 +168,7 @@ describe('SessionPreparations reservation', () => {
     const controller = new AbortController()
     const reason = new Error('cancel after commit')
 
-    await expect(preparations.reserve(id, () => Promise.resolve(source), async value => {
+    await expect(preparations.reserve(id, () => Promise.resolve(source), async (value) => {
       controller.abort(reason)
       return { source: value, state: value.label }
     }, controller.signal)).rejects.toBe(reason)
@@ -185,7 +185,7 @@ describe('SessionPreparations reservation', () => {
     const commitGate = Promise.withResolvers<undefined>()
     const controller = new AbortController()
     const reason = new Error('cancel invalidated commit')
-    const reservation = preparations.reserve(id, () => Promise.resolve(source), async value => {
+    const reservation = preparations.reserve(id, () => Promise.resolve(source), async (value) => {
       commitStarted.resolve(undefined)
       await commitGate.promise
       return { source: value, state: value.label }
@@ -235,7 +235,9 @@ describe('observeQueuedAbort', () => {
     const signal = new AbortController().signal
     await expect(observeQueuedAbort(Promise.resolve('value'), signal)).resolves.toBe('value')
     const failure = { kind: 'failed' }
-    await expect(observeQueuedAbort(Promise.reject(failure), signal)).rejects.toBe(failure)
+    const rejected = Promise.withResolvers<never>()
+    rejected.reject(failure)
+    await expect(observeQueuedAbort(rejected.promise, signal)).rejects.toBe(failure)
   })
 
   it('rejects promptly with an exact abort reason and ignores later settlement', async () => {

packages/subagent/subagent/tests/continuation.spec.ts

@@ -568,6 +568,17 @@ describe('SubagentService.followup residency routing', () => {
       .rejects.toMatchObject({ code: 'NOT_RESUMABLE' })
   })
 
+  it('preserves a SubagentError raised while cold-materializing a child', async () => {
+    const { ctx, parent } = await setup([textResponse('first')])
+    const started = await ctx.subagents.startContinuable(startSpec(parent))
+    await waitNoActivation(ctx, started.childId)
+    const failure = new SubagentError('materialization denied', 'UNAUTHORIZED')
+    ctx.agents.resume = () => Promise.reject(failure)
+
+    await expect(followup(ctx, parent, started.childId, message('continue')))
+      .rejects.toBe(failure)
+  })
+
   it('cold-resumes a delivery that lost the race with final disposal', async () => {
     const { ctx, parent } = await setup([textResponse('first'), textResponse('after the race')])
     const started = await ctx.subagents.startContinuable(startSpec(parent))

packages/subagent/subagent/tests/list-children.spec.ts

@@ -363,6 +363,17 @@ describe('SubagentService.listChildren', () => {
     expect(entries).toEqual([{ kind: 'diagnostic', id: childId, reason: 'unavailable' }])
   })
 
+  it('maps an invalid child surface to corrupt', async () => {
+    const { ctx, parent } = await setup([textResponse('done')])
+    const childId = await startChild(ctx, parent, 'invalid surface')
+    const query = ctx.get('sessionQuery')!
+    query.listEvents = () =>
+      Promise.reject(new SessionQueryError('invalid surface', 'SESSION_QUERY_INVALID_SURFACE'))
+
+    const entries = await ctx.subagents.listChildren(parent.id)
+    expect(entries).toEqual([{ kind: 'diagnostic', id: childId, reason: 'corrupt' }])
+  })
+
   it('diagnoses a read whose header no longer names this parent as corrupt', async () => {
     const { ctx, parent } = await setup([textResponse('done')])
     const childId = await startChild(ctx, parent, 'reparented child')