feat(engine): outbox-ordered turn.submit saga with an operation journal

apps/daemon/AGENTS.md

@@ -82,10 +82,11 @@ Runs via `tsx` in dev (`pnpm -F @linkcode/daemon dev`) and a `tsup` bundle in pr
     reach `ensureDeviceKey`.
   - A missing `@napi-rs/keyring` native binding is a **packaging** defect, not a host property, and it
     silently downgrades every credential to plaintext. `verify-artifacts.mts` fails the release on it.
-- **`daemon.db`** — better-sqlite3 session/workspace registry (`session-store.ts` / `workspace-store.ts`,
-  tables in `src/db/schema.ts`). The zod `SessionRecordSchema` is the contract: rows are re-validated
-  through it on load; the table is just storage. After editing `src/db/schema.ts`, run
-  `pnpm -F @linkcode/daemon exec drizzle-kit generate` and commit `drizzle/` — migrations run at boot.
+- **`daemon.db`** — better-sqlite3 persistence (tables in `src/db/schema.ts`). `src/db/database.ts`
+  owns and migrates the shared graph/session connection; its stores borrow that client, while the
+  other stores retain their own connections. The zod `SessionRecordSchema` is the contract: rows
+  are re-validated through it on load; the table is just storage. After editing `src/db/schema.ts`,
+  run `pnpm -F @linkcode/daemon exec drizzle-kit generate` and commit `drizzle/` — migrations run at boot.
   - **A record field with no column is dropped in silence.** The store enumerates columns on write and
     rebuilds the record on read, so an `.optional()` field added to the schema alone survives until the
     next boot and then parses cleanly as `undefined`. Adding one is three edits (column, write, read)

apps/daemon/src/__tests__/conversation-store.test.ts

@@ -17,22 +17,40 @@ import {
 } from '@linkcode/schema';
 import { afterEach, describe, expect, it } from 'vitest';
 import { createConversationStore } from '../conversation-store';
+import type { DaemonDatabase } from '../db/database';
+import { openDaemonDatabase } from '../db/database';
 import { createSessionStore } from '../session-store';
 
 const temporaryDirectories: string[] = [];
+const openDatabases = new Set<DaemonDatabase>();
+
+function openDatabase(path: string): DaemonDatabase {
+  const database = openDaemonDatabase(path);
+  openDatabases.add(database);
+  return database;
+}
+
+function closeDatabase(database: DaemonDatabase): void {
+  database.close();
+  openDatabases.delete(database);
+}
 
 afterEach(async () => {
+  for (const database of openDatabases) database.close();
+  openDatabases.clear();
   await Promise.all(
     temporaryDirectories.splice(0).map((path) => rm(path, { recursive: true, force: true })),
   );
 });
 
-/** Migrations belong to the session store; turn rows FK-reference its sessions table. */
-async function databaseWithSessions(...sessionIds: string[]): Promise<string> {
+async function databaseWithSessions(
+  ...sessionIds: string[]
+): Promise<{ readonly path: string; readonly database: DaemonDatabase }> {
   const directory = await mkdtemp(join(tmpdir(), 'linkcode-conversation-store-'));
   temporaryDirectories.push(directory);
-  const database = join(directory, 'daemon.db');
-  const sessions = createSessionStore(database);
+  const path = join(directory, 'daemon.db');
+  const database = openDatabase(path);
+  const sessions = createSessionStore(database.client);
   for (let i = 0, len = sessionIds.length; i < len; i++) {
     await sessions.save(
       SessionRecordSchema.parse({
@@ -46,7 +64,7 @@ async function databaseWithSessions(...sessionIds: string[]): Promise<string> {
       }),
     );
   }
-  return database;
+  return { path, database };
 }
 
 function turn(value: {
@@ -112,7 +130,7 @@ describe('SQLite conversation store', () => {
    * catch it (the daemon store trap in apps/daemon/AGENTS.md).
    */
   it('round-trips every turn field, all three input shapes included', async () => {
-    const database = await databaseWithSessions('s-1');
+    const { path, database } = await databaseWithSessions('s-1');
     const migratedTurn = turn({
       turnId: 't-5',
       parentTurnId: TurnIdSchema.parse('t-4'),
@@ -153,7 +171,7 @@ describe('SQLite conversation store', () => {
       }),
       migratedTurn,
     ];
-    const store = createConversationStore(database);
+    const store = createConversationStore(database.client);
     await store.persistTurnIntent({
       turn: turns[0],
       prompt: prompt('p-1'),
@@ -176,23 +194,24 @@ describe('SQLite conversation store', () => {
       operation: openOperation('op-migrated'),
     });
 
-    const reopened = createConversationStore(database);
+    closeDatabase(database);
+    const reopened = createConversationStore(openDatabase(path).client);
     expect(await reopened.listTurns(SessionIdSchema.parse('s-1'))).toEqual(turns);
     expect(await reopened.getPrompt(PromptIdSchema.parse('p-migrated'))).toBeUndefined();
   });
 
   it('round-trips prompts, preserving block and context order', async () => {
-    const database = await databaseWithSessions('s-1');
-    await seedIntent(createConversationStore(database));
+    const { database } = await databaseWithSessions('s-1');
+    await seedIntent(createConversationStore(database.client));
 
-    expect(await createConversationStore(database).getPrompt(PromptIdSchema.parse('p-1'))).toEqual(
-      prompt('p-1'),
-    );
+    expect(
+      await createConversationStore(database.client).getPrompt(PromptIdSchema.parse('p-1')),
+    ).toEqual(prompt('p-1'));
   });
 
   it('round-trips bindings and re-captures by (turn, history)', async () => {
-    const database = await databaseWithSessions('s-1');
-    const store = createConversationStore(database);
+    const { database } = await databaseWithSessions('s-1');
+    const store = createConversationStore(database.client);
     await seedIntent(store);
     const live = ProviderTurnBindingSchema.parse({
       turnId: 't-prompted',
@@ -211,13 +230,13 @@ describe('SQLite conversation store', () => {
     await store.saveBinding(recaptured);
 
     expect(
-      await createConversationStore(database).listBindings(TurnIdSchema.parse('t-prompted')),
+      await createConversationStore(database.client).listBindings(TurnIdSchema.parse('t-prompted')),
     ).toEqual([recaptured, { ...live, historyId: 'native-2', capturedFrom: 'replay' }]);
   });
 
   it('round-trips operations through every state', async () => {
-    const database = await databaseWithSessions('s-1');
-    const store = createConversationStore(database);
+    const { path, database } = await databaseWithSessions('s-1');
+    const store = createConversationStore(database.client);
     await seedIntent(store);
     expect(await store.getOperation(OperationIdSchema.parse('op-1'))).toEqual(
       openOperation('op-1'),
@@ -257,7 +276,8 @@ describe('SQLite conversation store', () => {
     });
     await store.resolveOperation(failed, { ...doomed, state: 'failed' });
 
-    const reopened = createConversationStore(database);
+    closeDatabase(database);
+    const reopened = createConversationStore(openDatabase(path).client);
     expect(await reopened.getOperation(OperationIdSchema.parse('op-1'))).toEqual(succeeded);
     expect(await reopened.getOperation(OperationIdSchema.parse('op-2'))).toEqual(failed);
     expect(await reopened.listOpenOperations()).toEqual([]);
@@ -268,8 +288,8 @@ describe('SQLite conversation store', () => {
   });
 
   it('deleteSession purges turns, bindings, and operations but keeps prompts shared with a fork', async () => {
-    const database = await databaseWithSessions('s-parent', 's-fork');
-    const store = createConversationStore(database);
+    const { path, database } = await databaseWithSessions('s-parent', 's-fork');
+    const store = createConversationStore(database.client);
     await store.persistTurnIntent({
       turn: turn({
         turnId: 't-shared',
@@ -315,23 +335,22 @@ describe('SQLite conversation store', () => {
 
     await store.deleteSession(SessionIdSchema.parse('s-parent'));
 
-    const reopened = createConversationStore(database);
+    closeDatabase(database);
+    const reopened = createConversationStore(openDatabase(path).client);
     expect(await reopened.listTurns(SessionIdSchema.parse('s-parent'))).toEqual([]);
     expect(await reopened.listBindings(TurnIdSchema.parse('t-shared'))).toEqual([]);
     expect(await reopened.listOpenOperations()).toEqual([]);
     expect(await reopened.getPrompt(PromptIdSchema.parse('p-own'))).toBeUndefined();
     expect(await reopened.getPrompt(PromptIdSchema.parse('p-shared'))).toEqual(prompt('p-shared'));
     expect(await reopened.listTurns(SessionIdSchema.parse('s-fork'))).toHaveLength(1);
 
-    await store.deleteSession(SessionIdSchema.parse('s-fork'));
-    expect(
-      await createConversationStore(database).getPrompt(PromptIdSchema.parse('p-shared')),
-    ).toBeUndefined();
+    await reopened.deleteSession(SessionIdSchema.parse('s-fork'));
+    expect(await reopened.getPrompt(PromptIdSchema.parse('p-shared'))).toBeUndefined();
   });
 
   it('refuses a second intent while the session has an open operation', async () => {
-    const database = await databaseWithSessions('s-1', 's-2');
-    const store = createConversationStore(database);
+    const { database } = await databaseWithSessions('s-1', 's-2');
+    const store = createConversationStore(database.client);
     await store.persistTurnIntent({
       turn: turn({ turnId: 't-1' }),
       operation: openOperation('op-1'),
@@ -357,8 +376,8 @@ describe('SQLite conversation store', () => {
   });
 
   it('assigns sibling ordinals in the transaction and the unique index rejects duplicates', async () => {
-    const database = await databaseWithSessions('s-1');
-    const store = createConversationStore(database);
+    const { database } = await databaseWithSessions('s-1');
+    const store = createConversationStore(database.client);
     const first = await store.persistTurnIntent({
       turn: turn({ turnId: 't-1' }),
       operation: openOperation('op-1'),
@@ -390,8 +409,8 @@ describe('SQLite conversation store', () => {
   });
 
   it('refuses a replayed operation id instead of re-opening the terminal row', async () => {
-    const database = await databaseWithSessions('s-1');
-    const store = createConversationStore(database);
+    const { database } = await databaseWithSessions('s-1');
+    const store = createConversationStore(database.client);
     const first = await store.persistTurnIntent({
       turn: turn({ turnId: 't-1' }),
       operation: openOperation('op-1'),
@@ -417,8 +436,8 @@ describe('SQLite conversation store', () => {
   });
 
   it('resolveOperation transitions open rows only — the first terminal result stands', async () => {
-    const database = await databaseWithSessions('s-1');
-    const store = createConversationStore(database);
+    const { path, database } = await databaseWithSessions('s-1');
+    const store = createConversationStore(database.client);
     const first = await store.persistTurnIntent({
       turn: turn({ turnId: 't-1' }),
       operation: openOperation('op-1'),
@@ -447,7 +466,8 @@ describe('SQLite conversation store', () => {
       ),
     ).toBe(false);
 
-    const reopened = createConversationStore(database);
+    closeDatabase(database);
+    const reopened = createConversationStore(openDatabase(path).client);
     expect(await reopened.getOperation(OperationIdSchema.parse('op-1'))).toEqual(failed);
     expect(await reopened.listTurns(SessionIdSchema.parse('s-1'))).toEqual([
       { ...first, state: 'failed' },

apps/daemon/src/__tests__/resource-store.test.ts

@@ -3,12 +3,17 @@ import { tmpdir } from 'node:os';
 import { join } from 'node:path';
 import { SessionRecordSchema, SessionResourceSchema } from '@linkcode/schema';
 import { afterEach, describe, expect, it } from 'vitest';
+import type { DaemonDatabase } from '../db/database';
+import { openDaemonDatabase } from '../db/database';
 import { createResourceStore } from '../resource-store';
 import { createSessionStore } from '../session-store';
 
 const temporaryDirectories: string[] = [];
+const openDatabases = new Set<DaemonDatabase>();
 
 afterEach(async () => {
+  for (const database of openDatabases) database.close();
+  openDatabases.clear();
   await Promise.all(
     temporaryDirectories.splice(0).map((path) => rm(path, { recursive: true, force: true })),
   );
@@ -18,8 +23,10 @@ describe('SQLite resource store', () => {
   it('persists resources across store instances and deduplicates output locators', async () => {
     const directory = await mkdtemp(join(tmpdir(), 'linkcode-resource-store-'));
     temporaryDirectories.push(directory);
-    const database = join(directory, 'daemon.db');
-    const sessions = createSessionStore(database);
+    const databasePath = join(directory, 'daemon.db');
+    const database = openDaemonDatabase(databasePath);
+    openDatabases.add(database);
+    const sessions = createSessionStore(database.client);
     const session = SessionRecordSchema.parse({
       sessionId: 'session-resource-test',
       kind: 'codex',
@@ -43,7 +50,7 @@ describe('SQLite resource store', () => {
       updatedAt: 2,
     });
 
-    const first = createResourceStore(database);
+    const first = createResourceStore(databasePath);
     expect(await first.save(resource, locatorKey)).toBe(true);
     expect(await first.findByLocator(session.sessionId, locatorKey)).toEqual(resource);
     expect(
@@ -75,7 +82,7 @@ describe('SQLite resource store', () => {
     expect(await first.save(promotedOutput, sourceUrl)).toBe(true);
     expect(await first.findByLocator(session.sessionId, sourceUrl)).toEqual(promotedOutput);
 
-    const restarted = createResourceStore(database);
+    const restarted = createResourceStore(databasePath);
     expect(await restarted.list(session.sessionId)).toEqual(
       expect.arrayContaining([resource, promotedOutput]),
     );

apps/daemon/src/__tests__/session-store.test.ts

@@ -4,11 +4,27 @@ import { join } from 'node:path';
 import { SessionRecordSchema, SessionRunSchema } from '@linkcode/schema';
 import Sqlite from 'better-sqlite3';
 import { afterEach, describe, expect, it } from 'vitest';
+import type { DaemonDatabase } from '../db/database';
+import { openDaemonDatabase } from '../db/database';
 import { createSessionStore } from '../session-store';
 
 const temporaryDirectories: string[] = [];
+const openDatabases = new Set<DaemonDatabase>();
+
+function openDatabase(path: string): DaemonDatabase {
+  const database = openDaemonDatabase(path);
+  openDatabases.add(database);
+  return database;
+}
+
+function closeDatabase(database: DaemonDatabase): void {
+  database.close();
+  openDatabases.delete(database);
+}
 
 afterEach(async () => {
+  for (const database of openDatabases) database.close();
+  openDatabases.clear();
   await Promise.all(
     temporaryDirectories.splice(0).map((path) => rm(path, { recursive: true, force: true })),
   );
@@ -51,9 +67,11 @@ describe('SQLite session store', () => {
         },
       ],
     });
-    await createSessionStore(database).save(record);
+    const first = openDatabase(database);
+    await createSessionStore(first.client).save(record);
+    closeDatabase(first);
 
-    expect(await createSessionStore(database).load()).toEqual([record]);
+    expect(await createSessionStore(openDatabase(database).client).load()).toEqual([record]);
   });
 
   it('round-trips additive fork provenance and upgrades an existing forked origin row', async () => {
@@ -72,17 +90,21 @@ describe('SQLite session store', () => {
       updatedAt: 6,
       runs: [],
     });
-    await createSessionStore(database).save(record);
+    const first = openDatabase(database);
+    await createSessionStore(first.client).save(record);
+    closeDatabase(first);
 
-    expect(await createSessionStore(database).load()).toEqual([record]);
+    const second = openDatabase(database);
+    expect(await createSessionStore(second.client).load()).toEqual([record]);
 
+    closeDatabase(second);
     const sqlite = new Sqlite(database);
     expect(sqlite.prepare('SELECT origin_type FROM sessions').pluck().get()).toBe('created');
     sqlite
       .prepare("UPDATE sessions SET origin_type = 'forked' WHERE session_id = ?")
       .run(record.sessionId);
     sqlite.close();
-    expect(await createSessionStore(database).load()).toEqual([record]);
+    expect(await createSessionStore(openDatabase(database).client).load()).toEqual([record]);
   });
 
   it('keeps run order across a reload, since the array position is part of the record', async () => {
@@ -100,7 +122,8 @@ describe('SQLite session store', () => {
         { runId: 'run-3', startedAt: 3, model: 'third' },
       ],
     });
-    const store = createSessionStore(database);
+    const first = openDatabase(database);
+    const store = createSessionStore(first.client);
     await store.save(record);
     // A later save rewrites the whole run list; the newest run is what a relaunch reads back.
     await store.save({
@@ -111,7 +134,8 @@ describe('SQLite session store', () => {
       ],
     });
 
-    const [reloaded] = await createSessionStore(database).load();
+    closeDatabase(first);
+    const [reloaded] = await createSessionStore(openDatabase(database).client).load();
     expect(reloaded.runs.map((run) => run.model)).toEqual(['first', 'second', 'third', 'fourth']);
   });
 
@@ -128,8 +152,7 @@ describe('SQLite session store', () => {
       runs: [{ startedAt: 1 }],
     });
 
-    await expect(async () => createSessionStore(database).save(record)).rejects.toThrow(
-      'without runId',
-    );
+    const store = createSessionStore(openDatabase(database).client);
+    await expect(async () => store.save(record)).rejects.toThrow('without runId');
   });
 });