[Tournament p2] SQL run context for runGraph

.github/workflows/sql-run-context-p2.yml

@@ -0,0 +1,47 @@
+name: SQL run context p2 conformance
+on:
+  pull_request:
+    branches: [main]
+    paths:
+      - 'src/**'
+      - 'tests/**'
+      - '.github/workflows/sql-run-context-p2.yml'
+permissions:
+  contents: read
+jobs:
+  conformance:
+    if: github.head_ref == 'pro/sql-run-context/p2' && github.event.pull_request.head.repo.full_name == github.repository
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+          persist-credentials: false
+      - name: Capture exact source for offline reproduction
+        run: tar --exclude=.git --exclude=node_modules -czf /tmp/sql-p2-source.tgz .
+      - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
+        with:
+          name: sql-p2-source
+          path: /tmp/sql-p2-source.tgz
+          retention-days: 1
+      - uses: pnpm/action-setup@0977fd99725f1db4007ccb2928dbb4e90d06cc86
+      - uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020
+        with:
+          node-version: 22
+          cache: pnpm
+      - run: pnpm install --frozen-lockfile
+      - name: Capture locked dependencies (no repository or credential files)
+        run: |
+          npm pack pnpm@11.17.0 --pack-destination /tmp
+          tar -czf /tmp/sql-p2-deps.tgz node_modules bench/node_modules
+      - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02
+        with:
+          name: sql-p2-deps
+          path: |
+            /tmp/sql-p2-deps.tgz
+            /tmp/pnpm-11.17.0.tgz
+          retention-days: 1
+          compression-level: 0
+      - name: SQL kill and resume conformance
+        run: pnpm exec vitest run tests/durability/sql-run-context.test.ts tests/durability/sql-graph-kill-resume.test.ts

tests/durability/sql-graph-kill-resume.test.ts

@@ -0,0 +1,78 @@
+/** The existing conformance graph, moved between independent hosts' local directories via SQL. */
+import { spawn } from 'node:child_process'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { describe, expect, it } from 'vitest'
+import type { SpawnEvent } from '../../src/runtime/supervise/types'
+import { ARTIFACT_KEY, EXPECTED_OUT, RUN_ID, WORKER_NODES } from '../helpers/durability/conformance-graph'
+
+const script = new URL('../helpers/durability/sql-graph-child.ts', import.meta.url).pathname
+interface Report {
+  kind: string
+  out: unknown
+  labels: string[]
+  events: SpawnEvent[]
+  effects: Array<{ effect_key: string }>
+  escalations: unknown[]
+}
+async function phase(file: string, label?: string) {
+  return new Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }>((resolve, reject) => {
+    const child = spawn(process.execPath, ['--import', 'tsx', script, file, ...(label ? [label] : [])], { stdio: ['ignore', 'pipe', 'pipe'] })
+    let stdout = ''
+    let stderr = ''
+    const timer = setTimeout(() => child.kill('SIGKILL'), 60_000)
+    child.stdout.on('data', (chunk) => { stdout += String(chunk) })
+    child.stderr.on('data', (chunk) => { stderr += String(chunk) })
+    child.once('error', (error) => { clearTimeout(timer); reject(error) })
+    child.once('close', (code, signal) => { clearTimeout(timer); resolve({ code, signal, stdout, stderr }) })
+  })
+}
+function report(stdout: string): Report {
+  const line = stdout.split('\n').filter((value) => value.startsWith('{')).at(-1)
+  if (!line) throw new Error(`missing child report: ${stdout}`)
+  return JSON.parse(line) as Report
+}
+function assertConformance(value: Report) {
+  expect(value.kind).toBe('winner')
+  expect(value.out).toEqual(EXPECTED_OUT)
+  expect(value.effects).toEqual([{ effect_key: ARTIFACT_KEY }])
+  // Unlike the old file suite, SQL must recover the ORIGINAL spawn, not escalate to a new key.
+  expect(value.escalations).toEqual([])
+  expect(value.events.filter((e) => e.kind === 'spawned' && e.parent === undefined)).toHaveLength(1)
+  expect(value.events.filter((e) => e.kind === 'materialized' && e.id === RUN_ID).length).toBeLessThanOrEqual(1)
+  const cursors = value.events.flatMap((e) => e.kind === 'settled' || e.kind === 'cancelled' ? [e.seq] : [])
+  expect(new Set(cursors).size).toBe(cursors.length)
+  for (const label of WORKER_NODES) {
+    const spawns = value.events.filter((e): e is Extract<SpawnEvent, { kind: 'spawned' }> => e.kind === 'spawned' && e.label === label)
+    expect(spawns, `one original spawn for ${label}`).toHaveLength(1)
+    expect(value.events.filter((e) => e.kind === 'settled' && e.status === 'done' && e.id === spawns[0]?.id)).toHaveLength(1)
+  }
+}
+
+describe('runGraph SQL kill-and-resume conformance', () => {
+  it('SIGKILL at every observed boundary resumes without losing/duplicating a step or an in-doubt spawn', { timeout: 900_000 }, async () => {
+    const dir = await mkdtemp(join(tmpdir(), 'sql-graph-matrix-'))
+    try {
+      const reference = await phase(join(dir, 'reference.sqlite'))
+      expect(reference.code, reference.stderr).toBe(0)
+      const baseline = report(reference.stdout)
+      assertConformance(baseline)
+      expect(baseline.labels).toContain('tool:commit:after-effect')
+      expect(baseline.labels.some((label) => label.startsWith('journal:after:execution-result:'))).toBe(true)
+      for (const [index, label] of baseline.labels.entries()) {
+        const file = join(dir, `kill-${index}.sqlite`)
+        const killed = await phase(file, label)
+        expect(killed.signal, `${label}: ${killed.stderr}`).toBe('SIGKILL')
+        // Database clock has one-second resolution on SQLite; use a conservative expiry margin.
+        await new Promise<void>((resolve) => setTimeout(resolve, 2200))
+        const resumed = await phase(file)
+        expect(resumed.code, `${label}: ${resumed.stderr}`).toBe(0)
+        assertConformance(report(resumed.stdout))
+      }
+      console.log(`SQL graph kill matrix: ${baseline.labels.length} boundaries passed`)
+    } finally {
+      await rm(dir, { recursive: true, force: true })
+    }
+  })
+})

tests/durability/sql-run-context.test.ts

@@ -0,0 +1,130 @@
+/** Tests committed before implementation. Run against real SQLite, including competing OS processes. */
+import { spawn, type ChildProcess } from 'node:child_process'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { contentAddress } from '../../src/durable/content-address'
+import { createSqlRunContext } from '../../src/runtime/supervise/sql-run-context'
+import type { SpawnEvent } from '../../src/runtime/supervise/types'
+import { openSqlite } from '../helpers/durability/sqlite-adapter'
+
+const childScript = new URL('../helpers/durability/sql-context-child.ts', import.meta.url).pathname
+const at = '2026-09-24T00:00:00.000Z'
+const root = { kind: 'spawned', id: 'run', label: 'root', runtime: 'router', budget: { maxIterations: 20 }, at } as SpawnEvent
+const delay = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms))
+
+describe('SQL run context: atomic records, fencing and cross-process takeover', () => {
+  let dir: string
+  const children = new Set<ChildProcess>()
+  const cleanups: Array<() => Promise<void> | void> = []
+  beforeEach(async () => { dir = await mkdtemp(join(tmpdir(), 'sql-run-context-')) })
+  afterEach(async () => {
+    for (const child of children) {
+      if (child.exitCode === null && child.signalCode === null) {
+        child.kill('SIGCONT')
+        child.kill('SIGKILL')
+        await new Promise<void>((resolve) => child.once('close', () => resolve()))
+      }
+    }
+    children.clear()
+    for (const cleanup of cleanups.reverse()) await cleanup()
+    cleanups.length = 0
+    await rm(dir, { recursive: true, force: true })
+  })
+
+  function connect() {
+    const connection = openSqlite(join(dir, 'context.sqlite'))
+    cleanups.push(connection.close)
+    return connection.db
+  }
+
+  function processFor(mode: string) {
+    const child = spawn(process.execPath, ['--import', 'tsx', childScript, join(dir, 'context.sqlite'), mode], {
+      stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
+    })
+    children.add(child)
+    let stderr = ''
+    child.stderr?.on('data', (chunk) => { stderr += String(chunk) })
+    const messages: unknown[] = []
+    const waiters: Array<(value: unknown) => void> = []
+    child.on('message', (message) => {
+      const waiter = waiters.shift()
+      if (waiter) waiter(message)
+      else messages.push(message)
+    })
+    const exit = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
+      child.once('error', reject)
+      child.once('close', (code, signal) => resolve({ code, signal }))
+    })
+    return {
+      child, exit, stderr: () => stderr,
+      next: () => messages.length ? Promise.resolve(messages.shift()) : new Promise<unknown>((resolve, reject) => {
+        const timer = setTimeout(() => reject(new Error(`child IPC timeout: ${stderr}`)), 15_000)
+        waiters.push((message) => { clearTimeout(timer); resolve(message) })
+      }),
+    }
+  }
+
+  it('reopens all acknowledged events and content-addressed blobs on an independent connection', async () => {
+    const db = connect()
+    const first = await createSqlRunContext(db, { runId: 'run' })
+    await first.journal.beginTree('run', at)
+    await first.journal.appendEvent('run', root)
+    const artifact = { answer: 42 }
+    const ref = contentAddress(artifact)
+    await first.blobs.put(ref, artifact)
+    await first.close()
+    const second = await createSqlRunContext(connect(), { runId: 'run' })
+    cleanups.push(() => second.close())
+    expect(await second.journal.loadTree('run')).toEqual([root])
+    expect(await second.blobs.get(ref)).toEqual(artifact)
+    expect(second.resume).toBe(true)
+  })
+
+  it('retries an acknowledged append without duplicating it, but refuses conflicting records', async () => {
+    const context = await createSqlRunContext(connect(), { runId: 'run' })
+    cleanups.push(() => context.close())
+    await context.journal.beginTree('run', at)
+    await context.journal.appendEvent('run', root)
+    await context.journal.appendEvent('run', structuredClone(root))
+    expect(await context.journal.loadTree('run')).toEqual([root])
+    await expect(context.journal.appendEvent('run', { ...root, label: 'conflict' } as SpawnEvent)).rejects.toThrow()
+    await expect(context.journal.beginTree('run', '2026-09-25T00:00:00.000Z')).rejects.toThrow()
+  })
+
+  it('rejects a forged blob and an append before beginTree', async () => {
+    const context = await createSqlRunContext(connect(), { runId: 'run' })
+    cleanups.push(() => context.close())
+    await expect(context.blobs.put(contentAddress({ answer: 41 }), { answer: 42 })).rejects.toThrow()
+    await expect(context.journal.appendEvent('missing', root)).rejects.toThrow()
+  })
+
+  it('two live processes cannot both own a run; SIGKILL permits takeover', { timeout: 30_000 }, async () => {
+    const owner = processFor('hold')
+    expect(await owner.next()).toEqual({ kind: 'ready' })
+    const contender = processFor('attempt')
+    expect(await contender.next()).toMatchObject({ kind: 'busy' })
+    expect((await contender.exit).code, contender.stderr()).toBe(0)
+    owner.child.kill('SIGKILL')
+    expect((await owner.exit).signal).toBe('SIGKILL')
+    await delay(2200)
+    const successor = processFor('attempt')
+    expect(await successor.next()).toEqual({ kind: 'acquired' })
+    expect((await successor.exit).code, successor.stderr()).toBe(0)
+  })
+
+  it('fences a stopped owner after takeover, even when that process later wakes', { timeout: 30_000 }, async () => {
+    const owner = processFor('hold')
+    expect(await owner.next()).toEqual({ kind: 'ready' })
+    owner.child.kill('SIGSTOP')
+    await delay(2200)
+    const successor = processFor('hold')
+    expect(await successor.next()).toEqual({ kind: 'ready' })
+    owner.child.kill('SIGCONT')
+    owner.child.send({ kind: 'append' })
+    expect(await owner.next()).toMatchObject({ kind: 'fenced' })
+    successor.child.send({ kind: 'close' })
+    expect((await successor.exit).code, successor.stderr()).toBe(0)
+  })
+})

tests/helpers/durability/sql-context-child.ts

@@ -0,0 +1,36 @@
+import { createSqlRunContext } from '../../../src/runtime/supervise/sql-run-context'
+import { openSqlite } from './sqlite-adapter'
+
+const [file, mode] = process.argv.slice(2)
+if (!file || !mode) throw new Error('usage: sql-context-child.ts <database> <hold|attempt>')
+const connection = openSqlite(file)
+try {
+  const context = await createSqlRunContext(connection.db, { runId: 'run', leaseMs: 1000 })
+  if (mode === 'attempt') {
+    process.send?.({ kind: 'acquired' })
+    await context.close()
+    connection.close()
+    process.disconnect?.()
+  } else {
+    process.send?.({ kind: 'ready' })
+    process.on('message', async (message: { kind: string }) => {
+      if (message.kind === 'append') {
+        try {
+          await context.journal.beginTree('late-owner', new Date().toISOString())
+          process.send?.({ kind: 'unsafe-append' })
+        } catch {
+          process.send?.({ kind: 'fenced' })
+        }
+      } else if (message.kind === 'close') {
+        await context.close()
+        connection.close()
+        process.disconnect?.()
+      }
+    })
+  }
+} catch (error) {
+  if (mode !== 'attempt' || !(error instanceof Error) || !/owned|busy|lease/i.test(error.message)) throw error
+  process.send?.({ kind: 'busy' })
+  connection.close()
+  process.disconnect?.()
+}

tests/helpers/durability/sql-graph-child.ts

@@ -0,0 +1,66 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { createSqlRunContext } from '../../../src/runtime/supervise/sql-run-context'
+import { runGraph } from '../../../src/runtime/supervise/graph'
+import { ConductorPlanner, conformanceGraph, instrumentedLeafSeam, RUN_ID } from './conformance-graph'
+import { sideEffectToolSpec, SIDE_EFFECT_TOOL_NAME } from './side-effect'
+import { openSqlite } from './sqlite-adapter'
+
+const [file, killAt = '-'] = process.argv.slice(2)
+if (!file) throw new Error('usage: sql-graph-child.ts <database> [killAt]')
+// Deliberately disjoint local directories: the only inheritance is SQL, not runDir.
+const dir = await mkdtemp(join(tmpdir(), 'sql-graph-host-'))
+const { db, close } = openSqlite(file)
+const labels = new Set<string>()
+const kill = (label: string) => {
+  labels.add(label)
+  if (label === killAt) process.kill(process.pid, 'SIGKILL')
+}
+try {
+  const context = await createSqlRunContext(db, { runId: RUN_ID, leaseMs: 1000 })
+  await db.exec('CREATE TABLE IF NOT EXISTS test_effects (effect_key TEXT PRIMARY KEY, payload TEXT NOT NULL)')
+  const append = context.journal.appendEvent.bind(context.journal)
+  const seen = new Set<string>()
+  context.journal.appendEvent = async (root, event) => {
+    const label = `${event.kind}:${event.id === RUN_ID ? 'root' : 'child'}`
+    const first = !seen.has(label)
+    seen.add(label)
+    if (first) kill(`journal:before:${label}`)
+    await append(root, event)
+    if (first) kill(`journal:after:${label}`)
+  }
+  const planner = new ConductorPlanner()
+  let turns = 0
+  const result = await runGraph(conformanceGraph(), {
+    runId: RUN_ID,
+    runContext: context,
+    workerSlots: 3,
+    maxTurns: 24,
+    perWorker: { maxIterations: 60, maxTokens: 500_000 },
+    brain: async (messages) => {
+      turns += 1
+      kill(`driver:turn:${turns}:before`)
+      const reply = planner.nextTurn(messages as ReadonlyArray<Record<string, unknown>>)
+      kill(`driver:turn:${turns}:after`)
+      return reply
+    },
+    makeLeafAgent: instrumentedLeafSeam({ dir, phase: 'sql', kill }),
+    extraTools: [sideEffectToolSpec()],
+    executeExtraTool: async (name, args) => {
+      if (name !== SIDE_EFFECT_TOOL_NAME) return null
+      kill('tool:commit:before')
+      const key = String(args.idempotencyKey)
+      const result = await db.exec('INSERT INTO test_effects (effect_key, payload) VALUES (?, ?) ON CONFLICT (effect_key) DO NOTHING', [key, JSON.stringify(args.payload)])
+      kill('tool:commit:after-effect')
+      return JSON.stringify({ committed: result.rowsAffected === 1, key })
+    },
+  })
+  const events = await context.journal.loadTree(RUN_ID)
+  const effects = await db.query<{ effect_key: string }>('SELECT effect_key FROM test_effects ORDER BY effect_key')
+  process.stdout.write(`${JSON.stringify({ kind: result.result.kind, out: result.result.kind === 'winner' ? result.result.out : null, labels: [...labels], events, effects, escalations: planner.report().escalations })}\n`)
+  await context.close()
+} finally {
+  close()
+  await rm(dir, { recursive: true, force: true })
+}

tests/helpers/durability/sqlite-adapter.ts

@@ -0,0 +1,19 @@
+import { DatabaseSync } from 'node:sqlite'
+import type { SqlAdapter } from '../../../src/conversation/journal-sql'
+
+/** Real SQL, never a statement-matching mock; separate processes open separate connections. */
+export function openSqlite(path: string): { db: SqlAdapter; close(): void } {
+  const connection = new DatabaseSync(path)
+  connection.exec('PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; PRAGMA busy_timeout=5000')
+  return {
+    db: {
+      async exec(sql, params = []) {
+        return { rowsAffected: Number(connection.prepare(sql).run(...params as never[]).changes) }
+      },
+      async query<TRow>(sql: string, params: readonly unknown[] = []): Promise<TRow[]> {
+        return connection.prepare(sql).all(...params as never[]) as TRow[]
+      },
+    },
+    close: () => connection.close(),
+  }
+}