fix(session): add semantic crash checkpoints

packages/session-persistence/session-checkpoint-policy/tests/crash-recovery.e2e.ts

@@ -0,0 +1,104 @@
+import { spawn } from 'node:child_process'
+import { access, mkdtemp, readFile, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { Context } from 'cordis'
+import { afterEach, describe, expect, it } from 'vitest'
+import SessionStore, {
+  SessionId, TOOL_OUTCOME_UNKNOWN,
+  type SessionEvent,
+} from '@deepseek-ai/dsh-session'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+
+const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
+const childScript = fileURLToPath(new URL('./fixtures/crash-child.ts', import.meta.url))
+const tsxLoader = fileURLToPath(import.meta.resolve('tsx'))
+const sessionId = SessionId('semantic-checkpoint-crash')
+const roots: string[] = []
+
+async function waitForFile(path: string): Promise<void> {
+  for (let attempt = 0; attempt < 500; attempt += 1) {
+    try {
+      await access(path)
+      return
+    } catch (error: unknown) {
+      if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error
+    }
+    await new Promise(resolve => setTimeout(resolve, 10))
+  }
+  throw new Error(`crash child did not reach failpoint ${path}`)
+}
+
+async function crashAt(mode: 'request' | 'tool'): Promise<{ root: string; markerText: string }> {
+  const root = await mkdtemp(join(tmpdir(), `dsh-semantic-${mode}-`))
+  roots.push(root)
+  const marker = join(root, 'failpoint')
+  const child = spawn(process.execPath, ['--import', tsxLoader, childScript, mode, root, marker], {
+    cwd: repoRoot,
+    env: { ...process.env, TSX_TSCONFIG_PATH: join(repoRoot, 'tsconfig.json') },
+    stdio: ['ignore', 'ignore', 'pipe'],
+  })
+  let stderr = ''
+  child.stderr.setEncoding('utf8')
+  child.stderr.on('data', (chunk: string) => { stderr += chunk })
+  try {
+    await waitForFile(marker)
+    const markerText = await readFile(marker, 'utf8')
+    const closed = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => {
+      child.once('close', (code, signal) => { resolve({ code, signal }) })
+    })
+    child.kill('SIGKILL')
+    const exit = await closed
+    expect(exit).toEqual({ code: null, signal: 'SIGKILL' })
+    return { root, markerText }
+  } catch (error: unknown) {
+    if (child.exitCode === null && child.signalCode === null) child.kill('SIGKILL')
+    throw new Error(`crash child failed: ${stderr}`, { cause: error })
+  }
+}
+
+async function load(root: string): Promise<SessionEvent[]> {
+  const ctx = new Context()
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(SessionPersistenceJsonl, { root, compression: 'none' })
+  try {
+    return (await ctx.sessionPersistence.load(sessionId)).events
+  } finally {
+    await ctx.fiber.dispose()
+  }
+}
+
+afterEach(async () => {
+  await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true })))
+})
+
+describe.skipIf(process.platform === 'win32')('semantic checkpoint hard-crash recovery', () => {
+  it('persists the complete request before model dispatch', async () => {
+    const crashed = await crashAt('request')
+    expect(crashed.markerText).toBe('request-dispatched')
+    const events = await load(crashed.root)
+    expect(events.map(event => event.type)).toEqual([
+      'turn/start', 'user/message', 'step/start', 'request/header', 'step/end', 'turn/end',
+    ])
+    expect(events.at(-1)).toMatchObject({
+      type: 'turn/end', data: { reason: { kind: 'interrupted' } },
+    })
+  })
+
+  it('persists tool intent before a side effect and repairs its missing result as unknown', async () => {
+    const crashed = await crashAt('tool')
+    expect(crashed.markerText).toBe('tool-side-effect')
+    const events = await load(crashed.root)
+    expect(events.some(event => event.type === 'assistant/message')).toBe(true)
+    expect(events.some(event => event.type === 'tool/call')).toBe(true)
+    const result = events.find(event => event.type === 'tool/result')
+    expect(result?.type === 'tool/result' && result.data.error).toEqual({
+      name: 'ToolOutcomeUnknownError', code: TOOL_OUTCOME_UNKNOWN,
+    })
+    if (result?.type !== 'tool/result' || result.data.content[0]?.type !== 'text') {
+      throw new Error('expected a text tool result')
+    }
+    expect(result.data.content[0].text).toContain('Do not retry blindly.')
+  })
+})

packages/session-persistence/session-checkpoint-policy/tests/fixtures/crash-child.ts

@@ -0,0 +1,59 @@
+import { writeFile } from 'node:fs/promises'
+import { Context } from 'cordis'
+import AgentLoop from '@deepseek-ai/dsh-agent-loop'
+import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
+import { CallId, type GenerateOptions, LlmAdapter, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as checkpointPolicy from '../../src/index.ts'
+
+function waitForCrash(): Promise<never> {
+  return new Promise(() => { setInterval(() => {}, 60_000) })
+}
+
+const [mode, root, marker] = process.argv.slice(2)
+if ((mode !== 'request' && mode !== 'tool') || root === undefined || marker === undefined) {
+  throw new Error('usage: crash-child.ts <request|tool> <persistence-root> <marker>')
+}
+const persistenceRoot = root
+const failpoint = marker
+
+class CrashAdapter extends LlmAdapter {
+  async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    if (mode === 'request') {
+      await writeFile(failpoint, 'request-dispatched')
+      await waitForCrash()
+      return
+    }
+    yield { type: 'block-start', index: 0, blockType: 'tool-call' }
+    yield {
+      type: 'block-end',
+      index: 0,
+      block: { type: 'tool-call', id: CallId('crash-call'), name: 'crash_tool', arguments: '{}' },
+    }
+    yield { type: 'finish', reason: { kind: 'tool-calls' } }
+  }
+}
+
+const ctx = new Context()
+await mountAgentLoopTestDependencies(ctx)
+await ctx.plugin(AgentLoop, { agents: [] })
+await ctx.plugin(SessionPersistenceJsonl, { root: persistenceRoot, compression: 'none' })
+await ctx.plugin(checkpointPolicy)
+ctx.llm.registerAdapter(['crash'], new CrashAdapter())
+ctx.tools.register({
+  name: 'crash_tool',
+  description: 'records an external effect and never returns',
+  parameters: {},
+  async execute() {
+    await writeFile(failpoint, 'tool-side-effect')
+    return waitForCrash()
+  },
+})
+
+const handle = await ctx.agents.create({
+  sessionId: SessionId('semantic-checkpoint-crash'),
+  agentOptions: { provider: 'crash', model: 'crash' },
+})
+handle.agent.send([{ type: 'text', text: 'exercise the crash boundary' }])
+await waitForCrash()

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

@@ -0,0 +1,212 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { Context } from 'cordis'
+import Loader from '@cordisjs/plugin-loader'
+import { agentEvents, type Agent } from '@deepseek-ai/dsh-agent'
+import LlmService, { CallId, type GenerateOptions, LlmAdapter, type StreamChunk } from '@deepseek-ai/dsh-llm'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
+import SessionPersistence from '@deepseek-ai/dsh-session-persistence'
+import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
+import ToolRegistry from '@deepseek-ai/dsh-tools'
+import * as checkpointPolicy from '../src/index.ts'
+
+const contexts: Context[] = []
+
+class TestPersistence extends SessionPersistence {
+  locate(_meta: SessionHeader): undefined { return undefined }
+  create(_meta: SessionHeader): Promise<void> { return Promise.resolve() }
+  append(_id: SessionId, _events: readonly SessionEvent[]): Promise<void> { return Promise.resolve() }
+  load(_id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return Promise.reject(new Error('not used'))
+  }
+  list(): Promise<SessionHeader[]> { return Promise.resolve([]) }
+}
+
+class RecordingAdapter extends LlmAdapter {
+  constructor(private readonly order: string[]) { super() }
+  async * stream(_options: GenerateOptions): AsyncIterable<StreamChunk> {
+    this.order.push('adapter')
+    yield { type: 'finish', reason: { kind: 'stop' } }
+  }
+}
+
+async function setup(): Promise<Context> {
+  const ctx = new Context()
+  contexts.push(ctx)
+  await ctx.plugin(SessionStore)
+  await ctx.plugin(LlmService)
+  await ctx.plugin(SystemPrompt)
+  await ctx.plugin(ToolRegistry)
+  await ctx.plugin(TestPersistence)
+  await ctx.plugin(checkpointPolicy)
+  return ctx
+}
+
+async function drain(stream: AsyncIterable<StreamChunk>): Promise<void> {
+  for await (const _chunk of stream) { /* drain */ }
+}
+
+afterEach(async () => {
+  await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose()))
+})
+
+describe('session-checkpoint-policy request boundary', () => {
+  it('awaits the live session checkpoint before constructing the downstream model stream', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('request-checkpoint'))
+    session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+    const gate = Promise.withResolvers<undefined>()
+    const order: string[] = []
+    ctx.on('session/flush', async () => {
+      order.push('flush:start')
+      await gate.promise
+      order.push('flush:end')
+    })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+
+    const pending = drain(ctx.llm.stream({
+      provider: 'mock', model: 'mock', messages: [], sessionId: session.id,
+    }))
+    await Promise.resolve()
+    expect(order).toEqual(['flush:start'])
+    gate.resolve(undefined)
+    await pending
+    expect(order).toEqual(['flush:start', 'flush:end', 'adapter'])
+  })
+
+  it('delegates a request without a live session without checkpointing', async () => {
+    const ctx = await setup()
+    const order: string[] = []
+    ctx.on('session/flush', () => { order.push('flush') })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+    await drain(ctx.llm.stream({ provider: 'mock', model: 'mock', messages: [] }))
+    expect(order).toEqual(['adapter'])
+  })
+
+  it('delegates an already-detached session id without checkpointing', async () => {
+    const ctx = await setup()
+    const order: string[] = []
+    ctx.on('session/flush', () => { order.push('flush') })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+    await drain(ctx.llm.stream({
+      provider: 'mock', model: 'mock', messages: [], sessionId: SessionId('detached'),
+    }))
+    expect(order).toEqual(['adapter'])
+  })
+
+  it('does not dispatch the adapter when the checkpoint rejects', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('request-failure'))
+    const order: string[] = []
+    ctx.on('session/flush', () => Promise.reject(new Error('disk unavailable')))
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter(order))
+    await expect(drain(ctx.llm.stream({
+      provider: 'mock', model: 'mock', messages: [], sessionId: session.id,
+    }))).rejects.toThrow('disk unavailable')
+    expect(order).toEqual([])
+  })
+})
+
+describe('session-checkpoint-policy tool and step boundaries', () => {
+  it('awaits the checkpoint before a top-level tool body', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('tool-checkpoint'))
+    const agent = { session } as Agent
+    const gate = Promise.withResolvers<undefined>()
+    const order: string[] = []
+    ctx.on('session/flush', async () => {
+      order.push('flush:start')
+      await gate.promise
+      order.push('flush:end')
+    })
+    ctx.tools.register({
+      name: 'write', description: 'side effect', parameters: {},
+      execute: async () => { order.push('tool'); return [] },
+    })
+
+    const pending = ctx.tools.execute({
+      callId: CallId('write-1'), name: 'write', arguments: {}, agent,
+    })
+    await Promise.resolve()
+    expect(order).toEqual(['flush:start'])
+    gate.resolve(undefined)
+    await expect(pending).resolves.toMatchObject({ isError: false })
+    expect(order).toEqual(['flush:start', 'flush:end', 'tool'])
+  })
+
+  it('turns a rejected checkpoint into an error result without running the tool body', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('tool-failure'))
+    const agent = { session } as Agent
+    let ran = false
+    ctx.on('session/flush', () => Promise.reject(new Error('disk unavailable')))
+    ctx.tools.register({
+      name: 'write', description: 'side effect', parameters: {},
+      execute: async () => { ran = true; return [] },
+    })
+    const result = await ctx.tools.execute({
+      callId: CallId('write-2'), name: 'write', arguments: {}, agent,
+    })
+    expect(result.isError).toBe(true)
+    expect(result.content).toEqual([{ type: 'text', text: 'Error: disk unavailable' }])
+    expect(ran).toBe(false)
+  })
+
+  it('reuses the outer checkpoint for a nested tool dispatch', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('nested-tool'))
+    const agent = { session } as Agent
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+    ctx.tools.register({ name: 'nested', description: 'nested', parameters: {}, execute: async () => [] })
+    await ctx.tools.execute({
+      callId: CallId('nested-1'), name: 'nested', arguments: {}, agent,
+      parent: Symbol('outer') as never,
+    })
+    expect(flushes).toBe(0)
+  })
+
+  it('checkpoints the complete recorded step at agent/post-step', async () => {
+    const ctx = await setup()
+    const session = ctx.sessions.create(SessionId('post-step'))
+    const agent = { session } as Agent
+    const flushed: string[] = []
+    ctx.on('session/flush', (current) => { flushed.push(current.id) })
+    await agentEvents(ctx, agent).serial(
+      'agent/post-step', 1, 1, new AbortController().signal,
+    )
+    expect(flushed).toEqual([session.id])
+  })
+})
+
+describe('session-checkpoint-policy lifecycle', () => {
+  it('removes its wrappers when the owning fiber is disposed', async () => {
+    const ctx = new Context()
+    contexts.push(ctx)
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(LlmService)
+    await ctx.plugin(SystemPrompt)
+    await ctx.plugin(ToolRegistry)
+    await ctx.plugin(TestPersistence)
+    const session = ctx.sessions.create(SessionId('disposed-policy'))
+    let flushes = 0
+    ctx.on('session/flush', () => { flushes += 1 })
+    ctx.llm.registerAdapter(['mock'], new RecordingAdapter([]))
+    const fiber = await ctx.plugin(checkpointPolicy)
+    await drain(ctx.llm.stream({ provider: 'mock', model: 'mock', messages: [], sessionId: session.id }))
+    expect(flushes).toBe(1)
+    await fiber.dispose()
+    await drain(ctx.llm.stream({ provider: 'mock', model: 'mock', messages: [], sessionId: session.id }))
+    expect(flushes).toBe(1)
+  })
+
+  it('keeps the Loader-safe namespace plugin shape', () => {
+    expect('default' in checkpointPolicy).toBe(false)
+    const loader = Object.create(Loader.prototype) as Loader
+    const unwrapped = loader.unwrapExports(checkpointPolicy) as Record<string, unknown>
+    expect(unwrapped).toBe(checkpointPolicy)
+    expect(unwrapped.name).toBe('session-checkpoint-policy')
+    expect(unwrapped.inject).toEqual(['llm', 'sessionPersistence', 'sessions', 'tools'])
+    expect(typeof unwrapped.apply).toBe('function')
+  })
+})