From d5cd54b148482a51622c777f3576664783c1a7a6 Mon Sep 17 00:00:00 2001 From: tangletools Date: Thu, 24 Sep 2026 15:44:15 -0600 Subject: [PATCH 1/2] test(durable): specify SQL run context kill/resume and two-process fencing (red first) --- .github/workflows/sql-run-context-p2.yml | 35 +++++ .../durability/sql-graph-kill-resume.test.ts | 78 +++++++++++ tests/durability/sql-run-context.test.ts | 130 ++++++++++++++++++ tests/helpers/durability/sql-context-child.ts | 36 +++++ tests/helpers/durability/sql-graph-child.ts | 66 +++++++++ tests/helpers/durability/sqlite-adapter.ts | 19 +++ 6 files changed, 364 insertions(+) create mode 100644 .github/workflows/sql-run-context-p2.yml create mode 100644 tests/durability/sql-graph-kill-resume.test.ts create mode 100644 tests/durability/sql-run-context.test.ts create mode 100644 tests/helpers/durability/sql-context-child.ts create mode 100644 tests/helpers/durability/sql-graph-child.ts create mode 100644 tests/helpers/durability/sqlite-adapter.ts diff --git a/.github/workflows/sql-run-context-p2.yml b/.github/workflows/sql-run-context-p2.yml new file mode 100644 index 000000000..772399700 --- /dev/null +++ b/.github/workflows/sql-run-context-p2.yml @@ -0,0 +1,35 @@ +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' + 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@v4 + 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: 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 diff --git a/tests/durability/sql-graph-kill-resume.test.ts b/tests/durability/sql-graph-kill-resume.test.ts new file mode 100644 index 000000000..f58419d4b --- /dev/null +++ b/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 => 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((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 }) + } + }) +}) diff --git a/tests/durability/sql-run-context.test.ts b/tests/durability/sql-run-context.test.ts new file mode 100644 index 000000000..122bcf677 --- /dev/null +++ b/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((resolve) => setTimeout(resolve, ms)) + +describe('SQL run context: atomic records, fencing and cross-process takeover', () => { + let dir: string + const children = new Set() + const cleanups: Array<() => Promise | 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((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((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) + }) +}) diff --git a/tests/helpers/durability/sql-context-child.ts b/tests/helpers/durability/sql-context-child.ts new file mode 100644 index 000000000..f1ba2a4b7 --- /dev/null +++ b/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 ') +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?.() +} diff --git a/tests/helpers/durability/sql-graph-child.ts b/tests/helpers/durability/sql-graph-child.ts new file mode 100644 index 000000000..fdeea05bf --- /dev/null +++ b/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 [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() +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() + 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>) + 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 }) +} diff --git a/tests/helpers/durability/sqlite-adapter.ts b/tests/helpers/durability/sqlite-adapter.ts new file mode 100644 index 000000000..057960268 --- /dev/null +++ b/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(sql: string, params: readonly unknown[] = []): Promise { + return connection.prepare(sql).all(...params as never[]) as TRow[] + }, + }, + close: () => connection.close(), + } +} From ffd6ddd0c86e5a5827d03cd823f9c40868b72d4c Mon Sep 17 00:00:00 2001 From: tangletools Date: Thu, 24 Sep 2026 15:55:12 -0600 Subject: [PATCH 2/2] test(durable): capture locked test dependencies for offline reproduction --- .github/workflows/sql-run-context-p2.yml | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/.github/workflows/sql-run-context-p2.yml b/.github/workflows/sql-run-context-p2.yml index 772399700..aac414e1b 100644 --- a/.github/workflows/sql-run-context-p2.yml +++ b/.github/workflows/sql-run-context-p2.yml @@ -10,7 +10,7 @@ permissions: contents: read jobs: conformance: - if: github.head_ref == 'pro/sql-run-context/p2' + 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: @@ -20,7 +20,7 @@ jobs: 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@v4 + - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 with: name: sql-p2-source path: /tmp/sql-p2-source.tgz @@ -31,5 +31,17 @@ jobs: 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