diff --git a/apps/daemon/AGENTS.md b/apps/daemon/AGENTS.md index 9b7ac11b1..fe1116f00 100644 --- a/apps/daemon/AGENTS.md +++ b/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) diff --git a/apps/daemon/src/__tests__/conversation-store.test.ts b/apps/daemon/src/__tests__/conversation-store.test.ts index 2343fd898..9c100c489 100644 --- a/apps/daemon/src/__tests__/conversation-store.test.ts +++ b/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(); + +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 { +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 { }), ); } - 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,7 +335,8 @@ 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([]); @@ -323,15 +344,13 @@ describe('SQLite conversation store', () => { 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' }, diff --git a/apps/daemon/src/__tests__/resource-store.test.ts b/apps/daemon/src/__tests__/resource-store.test.ts index bfdd674cc..2da6ec343 100644 --- a/apps/daemon/src/__tests__/resource-store.test.ts +++ b/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(); 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]), ); diff --git a/apps/daemon/src/__tests__/session-store.test.ts b/apps/daemon/src/__tests__/session-store.test.ts index 0850f3e47..13df38cb8 100644 --- a/apps/daemon/src/__tests__/session-store.test.ts +++ b/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(); + +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'); }); }); diff --git a/apps/daemon/src/conversation-store.ts b/apps/daemon/src/conversation-store.ts index 00461834f..73103d1a5 100644 --- a/apps/daemon/src/conversation-store.ts +++ b/apps/daemon/src/conversation-store.ts @@ -1,5 +1,3 @@ -import { mkdirSync } from 'node:fs'; -import { dirname } from 'node:path'; import type { ConversationStore, ConversationTurnIntent } from '@linkcode/engine'; import { ConversationSessionBusyError } from '@linkcode/engine'; import type { @@ -18,9 +16,8 @@ import { PromptRecordSchema, ProviderTurnBindingSchema, } from '@linkcode/schema'; -import Sqlite from 'better-sqlite3'; import { and, asc, count, eq, inArray, isNotNull, isNull, notInArray } from 'drizzle-orm'; -import { drizzle } from 'drizzle-orm/better-sqlite3'; +import type { DaemonDatabaseClient } from './db/database'; import { conversationOperations, conversationTurns, @@ -33,22 +30,14 @@ type TurnRow = typeof conversationTurns.$inferSelect; type PromptRow = typeof prompts.$inferSelect; type OperationRow = typeof conversationOperations.$inferSelect; -type Db = ReturnType; +type Db = DaemonDatabaseClient; type DbOrTx = Db | Parameters[0]>[0]; /** - * SQLite-backed `ConversationStore` on ONE dedicated connection — the multi-table methods run in - * `db.transaction`, which the submit saga's atomicity guarantees hang on. Rows are validated back - * through the zod schemas on load. Migrations are owned by the session store, which must be - * constructed first. + * SQLite-backed `ConversationStore` on the daemon's shared graph/session connection. Multi-table + * methods run in `db.transaction`; rows are validated back through the zod schemas on load. */ -export function createConversationStore(dbPath: string): ConversationStore { - if (dbPath !== ':memory:') mkdirSync(dirname(dbPath), { recursive: true }); - const sqlite = new Sqlite(dbPath); - sqlite.pragma('journal_mode = WAL'); - sqlite.pragma('foreign_keys = ON'); - const db = drizzle(sqlite); - +export function createConversationStore(db: DaemonDatabaseClient): ConversationStore { function upsertTurn(tx: DbOrTx, turn: ConversationTurn): void { const row = toTurnRow(turn); tx.insert(conversationTurns) diff --git a/apps/daemon/src/database-migrations.ts b/apps/daemon/src/database-migrations.ts new file mode 100644 index 000000000..92cba76b8 --- /dev/null +++ b/apps/daemon/src/database-migrations.ts @@ -0,0 +1,3 @@ +import { fileURLToPath } from 'node:url'; + +export const daemonMigrationsFolder = fileURLToPath(new URL('../drizzle', import.meta.url)); diff --git a/apps/daemon/src/db/database.ts b/apps/daemon/src/db/database.ts new file mode 100644 index 000000000..f48837f61 --- /dev/null +++ b/apps/daemon/src/db/database.ts @@ -0,0 +1,48 @@ +import { mkdirSync } from 'node:fs'; +import { dirname } from 'node:path'; +import Sqlite from 'better-sqlite3'; +import { drizzle } from 'drizzle-orm/better-sqlite3'; +import { migrate } from 'drizzle-orm/better-sqlite3/migrator'; +import { readMigrationFiles } from 'drizzle-orm/migrator'; +import { daemonMigrationsFolder } from '../database-migrations'; + +export type DaemonDatabaseClient = ReturnType; + +export interface DaemonDatabase { + readonly client: DaemonDatabaseClient; + readonly close: () => void; +} + +/** Drizzle keys applied migrations by journal time, so realign known hashes before migrating. */ +function reconcileMigrationLedger(sqlite: Sqlite.Database, migrationsFolder: string): void { + const hasLedger = sqlite + .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '__drizzle_migrations'") + .get(); + if (!hasLedger) return; + const realign = sqlite.prepare( + 'UPDATE __drizzle_migrations SET created_at = ? WHERE hash = ? AND created_at <> ?', + ); + sqlite.transaction(() => { + const migrations = readMigrationFiles({ migrationsFolder }); + for (let i = 0, len = migrations.length; i < len; i++) { + const migration = migrations[i]; + realign.run(migration.folderMillis, migration.hash, migration.folderMillis); + } + })(); +} + +export function openDaemonDatabase(dbPath: string): DaemonDatabase { + if (dbPath !== ':memory:') mkdirSync(dirname(dbPath), { recursive: true }); + const sqlite = new Sqlite(dbPath); + try { + sqlite.pragma('journal_mode = WAL'); + sqlite.pragma('foreign_keys = ON'); + const client = drizzle(sqlite); + reconcileMigrationLedger(sqlite, daemonMigrationsFolder); + migrate(client, { migrationsFolder: daemonMigrationsFolder }); + return { client, close: () => sqlite.close() }; + } catch (error) { + sqlite.close(); + throw error; + } +} diff --git a/apps/daemon/src/db/schema.ts b/apps/daemon/src/db/schema.ts index 658ee79af..9211f69cb 100644 --- a/apps/daemon/src/db/schema.ts +++ b/apps/daemon/src/db/schema.ts @@ -32,8 +32,8 @@ export const sessions = sqliteTable( /** Automation that created this session (`SessionRecord.automation`); null for user sessions. */ automationKind: text('automation_kind', { enum: ['loop', 'schedule'] }), automationId: text('automation_id'), - /** Deliberately no FK to `conversation_turns`: the turn tree is written on the conversation - * store's own connection, and the two tables would otherwise cycle. */ + /** Deliberately no FK to `conversation_turns`: turns already reference sessions, and the two + * tables would otherwise cycle. */ activeLeafTurnId: text('active_leaf_turn_id'), graphRevision: integer('graph_revision').notNull().default(0), createdAt: integer('created_at').notNull(), @@ -103,9 +103,9 @@ export const sessionResources = sqliteTable( /** * Conversation turn-tree tables. These mirror the `Conversation*` schemas from `@linkcode/schema` - * and are written ONLY by the conversation store's dedicated connection (../conversation-store.ts): - * the submit saga's transactions are multi-table, and atomicity across the per-store connections - * does not exist. Prompt content is user-authored and must never enter logs/telemetry. + * and are written by the conversation store (../conversation-store.ts) on the shared connection + * owned by ./database.ts — the submit saga's multi-table transactions need that one connection. + * Prompt content is user-authored and must never enter logs/telemetry. */ export const prompts = sqliteTable('prompts', { promptId: text('prompt_id').primaryKey(), diff --git a/apps/daemon/src/index.ts b/apps/daemon/src/index.ts index 66c426ed9..df9e5098e 100644 --- a/apps/daemon/src/index.ts +++ b/apps/daemon/src/index.ts @@ -42,6 +42,7 @@ import { worktreeRoot, } from './config'; import { createConversationStore } from './conversation-store'; +import { openDaemonDatabase } from './db/database'; import { DaemonLoggerLive, logger } from './logger'; import { createLoopStore } from './loop-store'; import type { ManagedAgentKind } from './managed-agent-refresh'; @@ -267,18 +268,20 @@ async function main(): Promise { if (simulatorMcp) { yield* Effect.addFinalizer(() => finalize(() => simulatorMcp.close())); } + const database = yield* Effect.acquireRelease( + Effect.sync(() => openDaemonDatabase(databasePath())), + (owned) => finalize(owned.close), + ); const EngineInfrastructureLive = makeEngineInfrastructureLayer(hub, { providerStore: store, ptyBackend: new SidecarPtyBackend(resolveSidecarPath()), simulators, simulatorMcp, simulatorConsent, - sessionStore: createSessionStore(databasePath()), - // After sessionStore, whose constructor applies the migrations these tables come from. - conversationStore: createConversationStore(databasePath()), + sessionStore: createSessionStore(database.client), + conversationStore: createConversationStore(database.client), resourceStore: createResourceStore(databasePath()), stateDir: daemonStateDir(), - // After sessionStore so its migration-ledger reconcile runs before this store migrates. scheduleStore: createScheduleStore(databasePath()), loopStore: createLoopStore(databasePath()), workspaceStore: createWorkspaceStore(databasePath()), diff --git a/apps/daemon/src/session-store.ts b/apps/daemon/src/session-store.ts index 0ff006ecf..2f8f9c89f 100644 --- a/apps/daemon/src/session-store.ts +++ b/apps/daemon/src/session-store.ts @@ -1,58 +1,19 @@ -import { mkdirSync } from 'node:fs'; -import { dirname } from 'node:path'; -import { fileURLToPath } from 'node:url'; import type { SessionStore } from '@linkcode/engine'; import type { SessionRecord } from '@linkcode/schema'; import { SessionRecordSchema } from '@linkcode/schema'; -import Sqlite from 'better-sqlite3'; import { asc, eq } from 'drizzle-orm'; -import { drizzle } from 'drizzle-orm/better-sqlite3'; -import { migrate } from 'drizzle-orm/better-sqlite3/migrator'; -import { readMigrationFiles } from 'drizzle-orm/migrator'; import { nullthrow } from 'foxts/guard'; +import type { DaemonDatabaseClient } from './db/database'; import { sessionRuns, sessions } from './db/schema'; type SessionRow = typeof sessions.$inferSelect; type RunRow = typeof sessionRuns.$inferSelect; /** - * drizzle's migrator keys "already applied?" on the journal `when` vs the recorded `created_at`, - * never the content hash — a regenerated migration gets a fresh `when`, re-runs its non-idempotent - * DDL, and crashes the daemon at boot. A recorded hash (sha256 of the SQL) proves that migration - * already ran, so realign its `created_at` to the current journal before migrating; genuinely new - * migrations (unrecorded hash) still run and still fail loudly. + * SQLite-backed `SessionStore` borrowing the shared graph/session connection. Rows are validated + * back through `SessionRecordSchema` on load — the zod schema stays the contract. */ -function reconcileMigrationLedger(sqlite: Sqlite.Database, migrationsFolder: string): void { - const hasLedger = sqlite - .prepare("SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '__drizzle_migrations'") - .get(); - if (!hasLedger) return; // First boot — nothing has been applied yet. - const realign = sqlite.prepare( - 'UPDATE __drizzle_migrations SET created_at = ? WHERE hash = ? AND created_at <> ?', - ); - sqlite.transaction(() => { - const migrations = readMigrationFiles({ migrationsFolder }); - for (let i = 0, len = migrations.length; i < len; i++) { - const migration = migrations[i]; - realign.run(migration.folderMillis, migration.hash, migration.folderMillis); - } - })(); -} - -/** - * SQLite-backed `SessionStore` (drizzle over better-sqlite3) at `~/.linkcode/daemon.db`. Rows are - * validated back through `SessionRecordSchema` on load — the zod schema stays the contract. - */ -export function createSessionStore(dbPath: string): SessionStore { - if (dbPath !== ':memory:') mkdirSync(dirname(dbPath), { recursive: true }); - const sqlite = new Sqlite(dbPath); - sqlite.pragma('journal_mode = WAL'); - sqlite.pragma('foreign_keys = ON'); - const db = drizzle(sqlite); - const migrationsFolder = fileURLToPath(new URL('../drizzle', import.meta.url)); - reconcileMigrationLedger(sqlite, migrationsFolder); - migrate(db, { migrationsFolder }); - +export function createSessionStore(db: DaemonDatabaseClient): SessionStore { return { load(): Promise { const sessionRows = db.select().from(sessions).all(); diff --git a/apps/daemon/tests/integration/session-store.test.ts b/apps/daemon/tests/integration/session-store.test.ts index 89627eaa9..244c93bb9 100644 --- a/apps/daemon/tests/integration/session-store.test.ts +++ b/apps/daemon/tests/integration/session-store.test.ts @@ -5,6 +5,8 @@ import type { SessionRecord } from '@linkcode/schema'; import { SessionRecordSchema } from '@linkcode/schema'; import Sqlite from 'better-sqlite3'; import { afterEach, describe, expect, it } from 'vitest'; +import type { DaemonDatabase } from '../../src/db/database'; +import { openDaemonDatabase } from '../../src/db/database'; import { createSessionStore } from '../../src/session-store'; const collator = new Intl.Collator(); @@ -24,12 +26,24 @@ function makeRecord(value: Record): SessionRecord { describe('daemon sqlite session store', () => { const tmpDirs: string[] = []; + const openDatabases = new Set(); + const openDatabase = (path: string): DaemonDatabase => { + const database = openDaemonDatabase(path); + openDatabases.add(database); + return database; + }; + const closeDatabase = (database: DaemonDatabase): void => { + database.close(); + openDatabases.delete(database); + }; afterEach(() => { + for (const database of openDatabases) database.close(); + openDatabases.clear(); while (tmpDirs.length > 0) rmSync(tmpDirs.pop()!, { recursive: true, force: true }); }); it('round-trips created and imported records', async () => { - const store = createSessionStore(':memory:'); + const store = createSessionStore(openDatabase(':memory:').client); const created = makeRecord({ runs: [ { runId: 'run-a', startedAt: 1 }, @@ -51,7 +65,7 @@ describe('daemon sqlite session store', () => { }); it('saves as a whole-record upsert, rewriting runs', async () => { - const store = createSessionStore(':memory:'); + const store = createSessionStore(openDatabase(':memory:').client); await store.save(makeRecord({ runs: [{ runId: 'run-a', startedAt: 1 }] })); const next = makeRecord({ title: 'Renamed', @@ -72,9 +86,11 @@ describe('daemon sqlite session store', () => { tmpDirs.push(dir); const dbPath = join(dir, 'daemon.db'); - const first = createSessionStore(dbPath); + const firstDatabase = openDatabase(dbPath); + const first = createSessionStore(firstDatabase.client); const record = makeRecord({ runs: [{ runId: 'run-a', startedAt: 1 }] }); await first.save(record); + closeDatabase(firstDatabase); // Simulate a dev DB migrated under an older journal: the newest migration's created_at predates // the journal's `when`, which without reconciliation re-runs it and crashes on the duplicate column. @@ -87,12 +103,12 @@ describe('daemon sqlite session store', () => { .run(); raw.close(); - const second = createSessionStore(dbPath); + const second = createSessionStore(openDatabase(dbPath).client); expect(await second.load()).toEqual([record]); }); it('deletes a record together with its runs', async () => { - const store = createSessionStore(':memory:'); + const store = createSessionStore(openDatabase(':memory:').client); const record = makeRecord({ runs: [{ runId: 'run-a', startedAt: 1 }] }); await store.save(record); await store.delete(record.sessionId); diff --git a/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts b/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts index fcb4fc496..cd798fa37 100644 --- a/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts +++ b/packages/host/engine/src/__tests__/engine-conversation-stubs.test.ts @@ -3,25 +3,18 @@ import { describe, expect, it } from 'vitest'; import { createSessionHarness } from './fixtures/session-harness'; const requests: WirePayload[] = [ - { - kind: 'turn.submit', - clientReqId: 'r-submit', - sessionId: 'session-1', - operationId: 'op-1', - input: { type: 'prompt', blocks: [{ type: 'text', text: 'hello' }] }, - }, { kind: 'conversation.graph.get', clientReqId: 'r-graph', sessionId: 'session-1' }, { kind: 'conversation.read', clientReqId: 'r-read', sessionId: 'session-1' }, ] as WirePayload[]; describe('conversation request stubs', () => { - it('refuses every conversation request loudly instead of dropping it', async () => { + it('refuses unimplemented conversation reads loudly instead of dropping them', async () => { const h = createSessionHarness(); await h.engine.start(); await Promise.all(requests.map((request) => h.inject(request))); - const replyIds = ['r-submit', 'r-graph', 'r-read']; + const replyIds = ['r-graph', 'r-read']; for (let i = 0, len = replyIds.length; i < len; i++) { expect(h.sent).toContainEqual( expect.objectContaining({ diff --git a/packages/host/engine/src/__tests__/engine-schedule.test.ts b/packages/host/engine/src/__tests__/engine-schedule.test.ts index 8ef55c337..c388c649f 100644 --- a/packages/host/engine/src/__tests__/engine-schedule.test.ts +++ b/packages/host/engine/src/__tests__/engine-schedule.test.ts @@ -12,13 +12,15 @@ import type { ValidatedWireMessage, WirePayload, } from '@linkcode/schema'; -import { textBlock } from '@linkcode/schema'; +import { SessionResourceIdSchema, textBlock } from '@linkcode/schema'; import type { Transport } from '@linkcode/transport'; import { createWireMessage } from '@linkcode/transport'; import { nullthrow } from 'foxts/guard'; import { noop } from 'foxts/noop'; import { wait } from 'foxts/wait'; import { describe, expect, it } from 'vitest'; +import { InMemoryConversationStore } from '../conversation/conversation-store'; +import { InMemoryResourceStore } from '../resource/resource-store'; import { createTestEngine } from './fixtures/test-engine'; /** Adapter that answers a prompt turn by emitting one assistant chunk and a stop. */ @@ -32,6 +34,8 @@ class ScheduleFakeAdapter implements AgentAdapter { }; private readonly listeners = new Set<(e: AgentEvent) => void>(); + constructor(private readonly promptInputs: AgentInput[]) {} + start(): Promise { return Promise.resolve(); } @@ -42,6 +46,7 @@ class ScheduleFakeAdapter implements AgentAdapter { send(input: AgentInput): Promise { if (input.type === 'prompt') { + this.promptInputs.push(structuredClone(input)); this.emit({ type: 'agent-message-chunk', messageId: 'm1' as MessageId, @@ -92,6 +97,7 @@ function pick( function harness() { const sent: WirePayload[] = []; + const promptInputs: AgentInput[] = []; let handler: ((msg: ValidatedWireMessage) => void) | null = null; const transport: Transport = { connect: () => Promise.resolve(), @@ -105,8 +111,10 @@ function harness() { onClose: () => noop, close: noop, }; - const factory: AdapterFactory = () => new ScheduleFakeAdapter(); - const engine = createTestEngine(transport, { factory }); + const conversationStore = new InMemoryConversationStore(); + const resourceStore = new InMemoryResourceStore(); + const factory: AdapterFactory = () => new ScheduleFakeAdapter(promptInputs); + const engine = createTestEngine(transport, { factory, conversationStore, resourceStore }); function inject(payload: WirePayload): void { nullthrow(handler, 'engine not started')(createWireMessage(payload)); @@ -116,7 +124,7 @@ function harness() { await wait(0); } } - return { engine, sent, inject, settle }; + return { engine, sent, promptInputs, conversationStore, resourceStore, inject, settle }; } const SPEC = { @@ -167,6 +175,65 @@ describe('engine schedule wiring', () => { expect(automationSession?.status).toBe('stopped'); }); + it('records an existing-session schedule prompt as a completed turn', async () => { + const h = harness(); + await h.engine.start(); + h.inject({ + kind: 'session.start', + clientReqId: 'session', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + await h.settle(); + const sessionId = pick(h.sent, 'session.started', 'session').sessionId; + const sourceUrl = 'https://example.com/reference'; + await h.resourceStore.save( + { + resourceId: SessionResourceIdSchema.parse('resource-source'), + sessionId, + direction: 'source', + name: 'Reference', + kind: 'link', + status: 'ready', + locator: { type: 'url', url: sourceUrl }, + createdAt: 1, + updatedAt: 1, + }, + sourceUrl, + ); + + h.inject({ + kind: 'schedule.create', + clientReqId: 'create', + spec: { ...SPEC, target: { type: 'session', sessionId } }, + }); + await h.settle(); + const scheduleId = pick(h.sent, 'schedule.created', 'create').schedule.scheduleId; + h.inject({ kind: 'schedule.run-once', clientReqId: 'run', scheduleId }); + await h.settle(); + + const [turn] = await h.conversationStore.listTurns(sessionId); + expect(turn).toMatchObject({ + parentTurnId: null, + siblingOrdinal: 1, + state: 'completed', + input: { type: 'prompt' }, + }); + if (turn.input.type !== 'prompt') throw new Error('expected a prompt turn'); + expect((await h.conversationStore.getPrompt(nullthrow(turn.input.promptId)))?.blocks).toEqual([ + { type: 'text', text: SPEC.prompt }, + ]); + expect(await h.conversationStore.listOpenOperations(sessionId)).toHaveLength(0); + expect(h.promptInputs).toEqual([ + { type: 'prompt', content: [{ type: 'text', text: SPEC.prompt }] }, + ]); + expect(h.sent).toContainEqual({ + kind: 'conversation.graph.changed', + sessionId, + graphRevision: 1, + activeLeafTurnId: turn.turnId, + }); + }); + it('reports an unknown schedule as not found', async () => { const h = harness(); await h.engine.start(); diff --git a/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts b/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts index 8a5906330..6d018a4dd 100644 --- a/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts +++ b/packages/host/engine/src/__tests__/engine-session-lifecycle.test.ts @@ -1,4 +1,4 @@ -import type { AgentInput, SessionId, StartOptions } from '@linkcode/schema'; +import type { AgentInput, RunId, SessionId, StartOptions } from '@linkcode/schema'; import { ConversationOperationSchema, ConversationTurnSchema, @@ -371,6 +371,7 @@ describe('engine session lifecycle', () => { const session = new LiveSession( new FakeAdapter(), 'sess-interrupt' as SessionId, + 'run-interrupt' as RunId, scope, closed, ); diff --git a/packages/host/engine/src/__tests__/engine-turn-submit.test.ts b/packages/host/engine/src/__tests__/engine-turn-submit.test.ts new file mode 100644 index 000000000..6b0762cf1 --- /dev/null +++ b/packages/host/engine/src/__tests__/engine-turn-submit.test.ts @@ -0,0 +1,598 @@ +import { setImmediate as nextLoopTurn } from 'node:timers/promises'; +import { asHistoryId } from '@linkcode/agent-adapter'; +import type { AgentHistoryResumeOptions, AgentInput, TurnId, WirePayload } from '@linkcode/schema'; +import { + AttachmentIdSchema, + OperationIdSchema, + RunIdSchema, + SessionIdSchema, + TurnIdSchema, +} from '@linkcode/schema'; +import { nullthrow } from 'foxts/guard'; +import { noop } from 'foxts/noop'; +import { describe, expect, it, vi } from 'vitest'; +import { InMemoryConversationStore } from '../conversation/conversation-store'; +import { InMemorySessionStore } from '../session/session-store'; +import { + FakeAdapter, + createSessionHarness as harness, + settleEngineTasks, + startedSessionId as startedId, +} from './fixtures/session-harness'; + +class RejectingTurnAdapter extends FakeAdapter { + override send(input: AgentInput): Promise { + this.sentInputs.push(input); + return Promise.reject(new Error('provider rejected input')); + } +} + +class RejectOnceAdapter extends FakeAdapter { + private rejected = false; + + override send(input: AgentInput): Promise { + if (!this.rejected) { + this.rejected = true; + return Promise.reject(new Error('provider rejected input')); + } + return super.send(input); + } +} + +class HangingResumeAdapter extends FakeAdapter { + override resumeHistory(): Promise { + return new Promise(noop); + } +} + +class GatedResumeAdapter extends FakeAdapter { + releaseResume: () => void = noop; + + override resumeHistory(opts: AgentHistoryResumeOptions): Promise { + this.resumedFrom = opts.historyId; + return new Promise((resolve) => { + this.releaseResume = resolve; + }); + } +} + +/** send() spans the whole turn (pi-style): the adapter emits `running` and resolves only later. */ +class WholeTurnSendAdapter extends FakeAdapter { + override send(input: AgentInput): Promise { + this.sentInputs.push(input); + this.emit({ type: 'status', status: 'running' }); + return new Promise(noop); + } +} + +class SilentHangingSendAdapter extends FakeAdapter { + override send(input: AgentInput): Promise { + this.sentInputs.push(input); + return new Promise(noop); + } +} + +/** First start is a normal adapter; the first relaunch is `make()`; later ones are normal. */ +function secondAdapter(make: () => FakeAdapter): () => FakeAdapter { + let index = 0; + return () => { + index += 1; + return index === 2 ? make() : new FakeAdapter(); + }; +} + +function submittedTurnId(sent: WirePayload[], replyTo: string): TurnId { + const reply = sent.find( + (payload) => payload.kind === 'turn.submitted' && payload.replyTo === replyTo, + ); + if (reply?.kind !== 'turn.submitted') throw new Error(`no turn.submitted for ${replyTo}`); + return reply.turnId; +} + +function failure(sent: WirePayload[], replyTo: string) { + const reply = sent.find( + (payload) => payload.kind === 'request.failed' && payload.replyTo === replyTo, + ); + if (reply?.kind !== 'request.failed') throw new Error(`no request.failed for ${replyTo}`); + return reply; +} + +async function startedHarness(makeAdapter: () => FakeAdapter = () => new FakeAdapter()) { + const conversationStore = new InMemoryConversationStore(); + const h = harness( + new InMemorySessionStore(), + makeAdapter, + undefined, + undefined, + undefined, + undefined, + { conversationStore }, + ); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const sessionId = startedId(h.sent, 'r1'); + return { ...h, conversationStore, sessionId, adapter: nullthrow(h.adapters[0]) }; +} + +function submitPrompt( + h: Awaited>, + clientReqId: string, + text: string, + extra: Partial<{ parentTurnId: TurnId | null; expectedGraphRevision: number }> = {}, +) { + return h.inject({ + kind: 'turn.submit', + clientReqId, + sessionId: h.sessionId, + operationId: OperationIdSchema.parse(`op-${clientReqId}`), + input: { type: 'prompt', blocks: [{ type: 'text', text }] }, + ...extra, + }); +} + +describe('turn.submit saga', () => { + it('submits a plain send onto a live session and commits the turn', async () => { + const h = await startedHarness(); + + await submitPrompt(h, 's1', 'hello'); + + const turnId = submittedTurnId(h.sent, 's1'); + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn).toMatchObject({ + turnId, + parentTurnId: null, + siblingOrdinal: 1, + state: 'running', + }); + expect(h.adapter.sentInputs).toEqual([ + { type: 'prompt', content: [{ type: 'text', text: 'hello' }] }, + ]); + expect(h.sent).toContainEqual( + expect.objectContaining({ + kind: 'conversation.graph.changed', + sessionId: h.sessionId, + graphRevision: 1, + activeLeafTurnId: turnId, + }), + ); + expect( + h.sent.some( + (payload) => payload.kind === 'agent.event' && payload.event.type === 'user-message', + ), + ).toBe(true); + }); + + it('replays a lost reply verbatim instead of duplicating a sibling', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'hello'); + const firstTurnId = submittedTurnId(h.sent, 's1'); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await h.inject({ + kind: 'turn.submit', + clientReqId: 's1-retry', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-s1'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'hello' }] }, + }); + + expect(submittedTurnId(h.sent, 's1-retry')).toBe(firstTurnId); + expect(await h.conversationStore.listTurns(h.sessionId)).toHaveLength(1); + expect(h.adapter.sentInputs).toHaveLength(1); + }); + + it('refuses a submit while a turn is running', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'hello'); + h.adapter.emit({ type: 'status', status: 'running' }); + + await submitPrompt(h, 's2', 'racing'); + + expect(failure(h.sent, 's2').code).toBe('busy'); + expect(await h.conversationStore.listTurns(h.sessionId)).toHaveLength(1); + }); + + it('refuses a submit while another operation is open', async () => { + const h = await startedHarness(); + await h.conversationStore.persistTurnIntent({ + turn: { + turnId: TurnIdSchema.parse('turn-open'), + sessionId: h.sessionId, + parentTurnId: null, + input: { type: 'shell-command', command: 'sleep 1' }, + runId: RunIdSchema.parse('run-elsewhere'), + state: 'preparing', + createdAt: Date.now(), + }, + operation: { + operationId: OperationIdSchema.parse('op-open'), + sessionId: h.sessionId, + kind: 'turn.submit', + state: 'open', + createdAt: Date.now(), + }, + }); + + await submitPrompt(h, 's1', 'hello'); + + expect(failure(h.sent, 's1').code).toBe('busy'); + }); + + it('tip-continues the active leaf with the revision guard, and conflicts when stale', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'first'); + const firstTurnId = submittedTurnId(h.sent, 's1'); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await submitPrompt(h, 's2', 'stale', { parentTurnId: firstTurnId, expectedGraphRevision: 0 }); + expect(failure(h.sent, 's2').code).toBe('conflict'); + + await submitPrompt(h, 's3', 'continue', { + parentTurnId: firstTurnId, + expectedGraphRevision: 1, + }); + const secondTurnId = submittedTurnId(h.sent, 's3'); + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns.find((turn) => turn.turnId === secondTurnId)).toMatchObject({ + parentTurnId: firstTurnId, + siblingOrdinal: 1, + state: 'running', + }); + }); + + it('plain sends carry no revision guard even after the graph moved', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'first'); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await submitPrompt(h, 's2', 'second'); + + const secondTurnId = submittedTurnId(h.sent, 's2'); + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns.find((turn) => turn.turnId === secondTurnId)?.parentTurnId).toBe( + submittedTurnId(h.sent, 's1'), + ); + }); + + it('refuses an interior fork while no provider checkpoint exists', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'first'); + const firstTurnId = submittedTurnId(h.sent, 's1'); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + await submitPrompt(h, 's2', 'second'); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await submitPrompt(h, 's3', 'fork attempt', { + parentTurnId: firstTurnId, + expectedGraphRevision: 2, + }); + + expect(failure(h.sent, 's3').code).toBe('unsupported'); + expect(await h.conversationStore.listTurns(h.sessionId)).toHaveLength(2); + }); + + it('persists the intent before dispatch and replays the stored failure', async () => { + const h = await startedHarness(() => new RejectingTurnAdapter()); + + await submitPrompt(h, 's1', 'doomed'); + + const stored = failure(h.sent, 's1'); + expect(stored.code).toBe('operation_failed'); + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn.state).toBe('failed'); + // The dispatcher broadcast the rejection into the conversation, and the reply says so — the + // client must not raise the failure a second time. + expect( + h.sent.some( + (p) => p.kind === 'agent.event' && p.sessionId === h.sessionId && p.event.type === 'error', + ), + ).toBe(true); + expect(stored.reportedInConversation).toBe(true); + + // Same operationId as s1 replays the stored error without touching the adapter again. A replay + // has no live event behind it, so it does not claim the conversation reported it. + await h.inject({ + kind: 'turn.submit', + clientReqId: 's1-replay', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-s1'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'doomed' }] }, + }); + const replayed = failure(h.sent, 's1-replay'); + expect(replayed.code).toBe(stored.code); + expect(replayed.message).toBe(stored.message); + expect(replayed.reportedInConversation).toBeUndefined(); + expect(h.adapter.sentInputs).toHaveLength(1); + }); + + it('refuses a replay whose operation id belongs to another session', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'hello'); + await h.inject({ + kind: 'session.start', + clientReqId: 'r2', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + const otherSessionId = startedId(h.sent, 'r2'); + + await h.inject({ + kind: 'turn.submit', + clientReqId: 's1-elsewhere', + sessionId: otherSessionId, + operationId: OperationIdSchema.parse('op-s1'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'hello' }] }, + }); + + expect(failure(h.sent, 's1-elsewhere')).toMatchObject({ code: 'invalid_request' }); + expect(await h.conversationStore.listTurns(otherSessionId)).toHaveLength(0); + expect( + (await h.conversationStore.getOperation(OperationIdSchema.parse('op-s1')))?.sessionId, + ).toBe(h.sessionId); + }); + + it('keeps ordinals stable across failed siblings', async () => { + const h = await startedHarness(() => new RejectOnceAdapter()); + + await submitPrompt(h, 's1', 'first try'); + await submitPrompt(h, 's2', 'second try'); + + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns).toHaveLength(2); + expect(turns.map((turn) => [turn.siblingOrdinal, turn.state]).sort()).toEqual([ + [1, 'failed'], + [2, 'running'], + ]); + expect(new Set(turns.map((turn) => turn.parentTurnId))).toEqual(new Set([null])); + }); + + it('starts a fresh provider session for a null-parent submit', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'first'); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await submitPrompt(h, 's2', 'new root', { parentTurnId: null, expectedGraphRevision: 1 }); + await vi.waitFor(() => submittedTurnId(h.sent, 's2')); + + const rootTurnId = submittedTurnId(h.sent, 's2'); + expect(h.adapter.stopped).toBe(true); + const replacement = nullthrow(h.adapters[1]); + expect(replacement.startedWith).not.toBeNull(); + expect(replacement.resumedFrom).toBeNull(); + expect(replacement.sentInputs).toEqual([ + { type: 'prompt', content: [{ type: 'text', text: 'new root' }] }, + ]); + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns.find((turn) => turn.turnId === rootTurnId)).toMatchObject({ + parentTurnId: null, + siblingOrdinal: 2, + state: 'running', + }); + }); + + it('resumes a cold session for a plain send with an addressable new run', async () => { + const h = await startedHarness(); + await submitPrompt(h, 's1', 'first'); + const firstTurnId = submittedTurnId(h.sent, 's1'); + h.adapter.emit({ type: 'session-ref', historyId: asHistoryId('native-1') }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + + await submitPrompt(h, 's2', 'wake up'); + await vi.waitFor(() => submittedTurnId(h.sent, 's2')); + + const resumed = nullthrow(h.adapters[1]); + expect(resumed.resumedFrom).toBe('native-1'); + const secondTurnId = submittedTurnId(h.sent, 's2'); + const turns = await h.conversationStore.listTurns(h.sessionId); + const second = nullthrow(turns.find((turn) => turn.turnId === secondTurnId)); + expect(second).toMatchObject({ parentTurnId: firstTurnId, state: 'running' }); + const [record] = await h.store.load(); + const run = record.runs.at(-1); + expect(run?.runId).toBe(second.runId); + expect(run?.baseTurnId).toBe(firstTurnId); + }); + + it('refuses unknown sessions, unknown parents, and attachment blocks', async () => { + const h = await startedHarness(); + + await h.inject({ + kind: 'turn.submit', + clientReqId: 's-nosession', + sessionId: SessionIdSchema.parse('sess-missing'), + operationId: OperationIdSchema.parse('op-nosession'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'x' }] }, + }); + expect(failure(h.sent, 's-nosession').code).toBe('not_found'); + + await submitPrompt(h, 's-noparent', 'x', { + parentTurnId: TurnIdSchema.parse('turn-missing'), + expectedGraphRevision: 0, + }); + expect(failure(h.sent, 's-noparent').code).toBe('not_found'); + + await h.inject({ + kind: 'turn.submit', + clientReqId: 's-attachment', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-attachment'), + input: { + type: 'prompt', + blocks: [{ type: 'attachment_ref', attachmentId: AttachmentIdSchema.parse('att-1') }], + }, + }); + expect(failure(h.sent, 's-attachment').code).toBe('unsupported'); + expect(await h.conversationStore.listTurns(h.sessionId)).toHaveLength(0); + }); + + it('resolves the operation when the session stops mid-dispatch, and the next submit is not busy', async () => { + const h = await startedHarness(secondAdapter(() => new HangingResumeAdapter())); + await submitPrompt(h, 's1', 'first'); + h.adapter.emit({ type: 'session-ref', historyId: asHistoryId('native-1') }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + await h.inject({ kind: 'session.stop', clientReqId: 'stop-1', sessionId: h.sessionId }); + + // The submit relaunches into an adapter hanging in resume; stopping the session interrupts it. + await submitPrompt(h, 's2', 'wake up'); + await vi.waitFor(() => expect(h.adapters).toHaveLength(2)); + await h.inject({ kind: 'session.stop', clientReqId: 'stop-2', sessionId: h.sessionId }); + await vi.waitFor(() => + expect(h.sent).toContainEqual({ kind: 'request.succeeded', replyTo: 'stop-2' }), + ); + + // The interrupted dispatch left no open operation; a retry replays the stored failure. + await vi.waitFor(async () => { + const operation = await h.conversationStore.getOperation(OperationIdSchema.parse('op-s2')); + expect(operation?.state).toBe('failed'); + }); + await h.inject({ + kind: 'turn.submit', + clientReqId: 's2-retry', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-s2'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'wake up' }] }, + }); + expect(failure(h.sent, 's2-retry').code).toBe('cancelled'); + + // A fresh submit is admitted and relaunches instead of replying busy. + await submitPrompt(h, 's3', 'again'); + await vi.waitFor(() => submittedTurnId(h.sent, 's3')); + expect(nullthrow(h.adapters[2]).resumedFrom).toBe('native-1'); + }); + + it('discards a start interrupted by the launch timeout so the next submit relaunches', async () => { + const h = await startedHarness(secondAdapter(() => new HangingResumeAdapter())); + await submitPrompt(h, 's1', 'first'); + h.adapter.emit({ type: 'session-ref', historyId: asHistoryId('native-1') }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + await h.inject({ kind: 'session.stop', clientReqId: 'stop-1', sessionId: h.sessionId }); + + // Fake only timers: the Effect clock sleeps on setTimeout, fibers schedule on setImmediate. + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + try { + const pending = h.inject({ + kind: 'turn.submit', + clientReqId: 's2', + sessionId: h.sessionId, + operationId: OperationIdSchema.parse('op-s2'), + input: { type: 'prompt', blocks: [{ type: 'text', text: 'wake up' }] }, + }); + // Let the dispatch reach the hung resume, then fire the launch timeout as it stands today. + while (h.adapters.length < 2) await nextLoopTurn(); + await vi.advanceTimersByTimeAsync(360_000); + await pending; + } finally { + vi.useRealTimers(); + } + + await vi.waitFor(() => + expect(failure(h.sent, 's2')).toMatchObject({ + code: 'timeout', + message: 'The provider did not start in time', + }), + ); + // The interrupted start was discarded — no registered zombie holding an unstarted adapter. + await vi.waitFor(() => expect(nullthrow(h.adapters[1]).stopped).toBe(true)); + + await submitPrompt(h, 's3', 'again'); + await vi.waitFor(() => submittedTurnId(h.sent, 's3')); + expect(nullthrow(h.adapters[2]).resumedFrom).toBe('native-1'); + }); + + it('tolerates a launch slower than the dispatch timer but within the launch budget', async () => { + const h = await startedHarness(secondAdapter(() => new GatedResumeAdapter())); + await submitPrompt(h, 's1', 'first'); + h.adapter.emit({ type: 'session-ref', historyId: asHistoryId('native-1') }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + await h.inject({ kind: 'session.stop', clientReqId: 'stop-1', sessionId: h.sessionId }); + + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + try { + const pending = submitPrompt(h, 's2', 'wake up'); + while (h.adapters.length < 2) await nextLoopTurn(); + // Past the retired flat 60s budget, well under the launch budget: the claude peak cold-start. + await vi.advanceTimersByTimeAsync(120_000); + const gated = h.adapters[1]; + if (!(gated instanceof GatedResumeAdapter)) throw new Error('expected the gated adapter'); + gated.releaseResume(); + await pending; + } finally { + vi.useRealTimers(); + } + + await vi.waitFor(() => submittedTurnId(h.sent, 's2')); + const resumed = nullthrow(h.adapters[1]); + expect(resumed.resumedFrom).toBe('native-1'); + expect(resumed.sentInputs).toEqual([ + { type: 'prompt', content: [{ type: 'text', text: 'wake up' }] }, + ]); + const operation = await h.conversationStore.getOperation(OperationIdSchema.parse('op-s2')); + expect(operation?.state).toBe('succeeded'); + }); + + it('commits the turn when the dispatch timer fires while the adapter is visibly running', async () => { + const h = await startedHarness(() => new WholeTurnSendAdapter()); + + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + try { + const pending = submitPrompt(h, 's1', 'long turn'); + while (h.adapter.sentInputs.length === 0) await nextLoopTurn(); + await vi.advanceTimersByTimeAsync(60_000); + await pending; + } finally { + vi.useRealTimers(); + } + + await vi.waitFor(() => submittedTurnId(h.sent, 's1')); + const turnId = submittedTurnId(h.sent, 's1'); + expect((await h.conversationStore.listTurns(h.sessionId))[0]).toMatchObject({ + turnId, + state: 'running', + }); + // Exactly one commit: the rescue's graph move, no second one from any surviving continuation. + expect(h.sent.filter((p) => p.kind === 'conversation.graph.changed')).toHaveLength(1); + + // The rescued turn settles through the adapter's own stop frame. + h.adapter.emit({ type: 'stop', stopReason: 'end_turn' }); + await settleEngineTasks(); + expect((await h.conversationStore.listTurns(h.sessionId))[0].state).toBe('completed'); + const operation = await h.conversationStore.getOperation(OperationIdSchema.parse('op-s1')); + expect(operation?.state).toBe('succeeded'); + }); + + it('fails the dispatch timer when the adapter never reported running', async () => { + const h = await startedHarness(() => new SilentHangingSendAdapter()); + + vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] }); + try { + const pending = submitPrompt(h, 's1', 'doomed'); + while (h.adapter.sentInputs.length === 0) await nextLoopTurn(); + await vi.advanceTimersByTimeAsync(60_000); + await pending; + } finally { + vi.useRealTimers(); + } + + await vi.waitFor(() => + expect(failure(h.sent, 's1')).toMatchObject({ + code: 'timeout', + message: 'The provider did not accept the turn in time', + }), + ); + const operation = await h.conversationStore.getOperation(OperationIdSchema.parse('op-s1')); + expect(operation?.state).toBe('failed'); + }); +}); diff --git a/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts b/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts new file mode 100644 index 000000000..ddff0e181 --- /dev/null +++ b/packages/host/engine/src/__tests__/engine-turn-tracking.test.ts @@ -0,0 +1,500 @@ +import { asHistoryId } from '@linkcode/agent-adapter'; +import type { + AgentHistoryCapabilities, + AgentInput, + ValidatedWireMessage, + WirePayload, +} from '@linkcode/schema'; +import { + MessageIdSchema, + OperationIdSchema, + RunIdSchema, + SessionIdSchema, + TurnIdSchema, + textBlock, +} from '@linkcode/schema'; +import type { Transport } from '@linkcode/transport'; +import { Effect } from 'effect'; +import { nullthrow } from 'foxts/guard'; +import { noop } from 'foxts/noop'; +import { describe, expect, it, vi } from 'vitest'; +import { InMemoryConversationStore } from '../conversation/conversation-store'; +import { ConversationTurnService } from '../conversation/turn-service'; +import { SessionRecordRegistry } from '../session/session-record-registry'; +import { InMemorySessionStore } from '../session/session-store'; +import { + FakeAdapter, + createSessionHarness as harness, + settleEngineTasks, + startedSessionId as startedId, +} from './fixtures/session-harness'; + +class RejectingTurnAdapter extends FakeAdapter { + override send(input: AgentInput): Promise { + this.sentInputs.push(input); + return Promise.reject(new Error('provider rejected input')); + } +} + +class HangingSendAdapter extends FakeAdapter { + override send(input: AgentInput): Promise { + this.sentInputs.push(input); + return new Promise(noop); + } +} + +class BranchingAdapter extends FakeAdapter { + override readonly historyCapabilities: AgentHistoryCapabilities = { + list: false, + read: true, + resume: true, + branch: true, + }; + + branchHistory(): Promise { + this.emit({ type: 'session-ref', historyId: asHistoryId('native-child') }); + return Promise.resolve(); + } +} + +async function startedHarness(makeAdapter: () => FakeAdapter = () => new FakeAdapter()) { + const conversationStore = new InMemoryConversationStore(); + const h = harness( + new InMemorySessionStore(), + makeAdapter, + undefined, + undefined, + undefined, + undefined, + { + conversationStore, + }, + ); + await h.engine.start(); + await h.inject({ + kind: 'session.start', + clientReqId: 'r1', + opts: { kind: 'claude-code', cwd: '/repo' }, + }); + return { + ...h, + conversationStore, + sessionId: startedId(h.sent, 'r1'), + adapter: nullthrow(h.adapters[0]), + }; +} + +describe('legacy input turn tracking', () => { + it('persists a turn for a legacy prompt and completes it on idle', async () => { + const h = await startedHarness(); + + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('hello')] }, + }); + + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn).toMatchObject({ + parentTurnId: null, + siblingOrdinal: 1, + state: 'running', + input: { type: 'prompt' }, + }); + if (turn.input.type !== 'prompt') throw new Error('expected a prompt turn'); + const prompt = await h.conversationStore.getPrompt(nullthrow(turn.input.promptId)); + expect(prompt?.blocks).toEqual([{ type: 'text', text: 'hello' }]); + expect(await h.conversationStore.listOpenOperations(h.sessionId)).toHaveLength(0); + const [record] = await h.store.load(); + expect(record.activeLeafTurnId).toBe(turn.turnId); + expect(record.graphRevision).toBe(1); + expect(h.sent).toContainEqual( + expect.objectContaining({ + kind: 'conversation.graph.changed', + sessionId: h.sessionId, + graphRevision: 1, + activeLeafTurnId: turn.turnId, + }), + ); + + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + const [settled] = await h.conversationStore.listTurns(h.sessionId); + expect(settled.state).toBe('completed'); + }); + + it('flips a cancelled turn to cancelled on the stop frame', async () => { + const h = await startedHarness(); + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('work')] }, + }); + + h.adapter.emit({ type: 'stop', stopReason: 'cancelled' }); + await settleEngineTasks(); + + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn.state).toBe('cancelled'); + }); + + it('marks an error-terminated turn failed on the stop-less idle settle', async () => { + const h = await startedHarness(); + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('work')] }, + }); + + h.adapter.emit({ type: 'error', message: 'provider exploded', recoverable: true }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn.state).toBe('failed'); + }); + + it('records command and shell inputs as turns and closes an unsettled predecessor', async () => { + const h = await startedHarness(); + h.adapter.emit({ + type: 'capabilities-update', + capabilities: { slashCommands: true, shellCommand: true }, + }); + h.adapter.emit({ type: 'available-commands-update', commands: [{ name: 'review' }] }); + + await h.inject({ + kind: 'agent.input', + clientReqId: 'r-cmd', + sessionId: h.sessionId, + input: { type: 'command', name: 'review' }, + }); + await h.inject({ + kind: 'agent.input', + clientReqId: 'r-sh', + sessionId: h.sessionId, + input: { type: 'shell-command', command: 'git status' }, + }); + await settleEngineTasks(); + + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns).toHaveLength(2); + const command = turns.find((turn) => turn.input.type === 'command'); + const shell = turns.find((turn) => turn.input.type === 'shell-command'); + expect(command).toMatchObject({ parentTurnId: null, state: 'completed' }); + expect(shell).toMatchObject({ parentTurnId: command?.turnId, state: 'running' }); + }); + + it('stores the failure when the adapter rejects a turn input', async () => { + const h = await startedHarness(() => new RejectingTurnAdapter()); + + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('hello')] }, + }); + + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn.state).toBe('failed'); + const operations = await h.conversationStore.listOpenOperations(h.sessionId); + expect(operations).toHaveLength(0); + const [record] = await h.store.load(); + expect(record.activeLeafTurnId).toBeUndefined(); + expect(record.graphRevision).toBe(0); + }); + + it('resolves the persisted turn when the session stops while its dispatch hangs', async () => { + const h = await startedHarness(() => new HangingSendAdapter()); + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('work')] }, + }); + // The intent is persisted and the adapter never acknowledges; stopping interrupts the dispatch. + expect(await h.conversationStore.listOpenOperations(h.sessionId)).toHaveLength(1); + + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + + await vi.waitFor(async () => { + expect(await h.conversationStore.listOpenOperations(h.sessionId)).toHaveLength(0); + }); + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn.state).toBe('failed'); + }); + + it('closes an unsettled predecessor as failed when its run saw an adapter error', async () => { + const h = await startedHarness(); + await h.inject({ + kind: 'agent.input', + clientReqId: 'first', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('one')] }, + }); + h.adapter.emit({ type: 'error', message: 'provider exploded', recoverable: true }); + await settleEngineTasks(); + + // No idle/stop settle arrived; admitting the next turn closes the predecessor out honestly. + await h.inject({ + kind: 'agent.input', + clientReqId: 'second', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('two')] }, + }); + await settleEngineTasks(); + + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns.map((turn) => turn.state)).toEqual(['failed', 'running']); + }); + + it('cancels the running turn when the session is stopped mid-turn', async () => { + const h = await startedHarness(); + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('work')] }, + }); + h.adapter.emit({ type: 'status', status: 'running' }); + + await h.inject({ kind: 'session.stop', clientReqId: 'stop', sessionId: h.sessionId }); + await settleEngineTasks(); + + const [turn] = await h.conversationStore.listTurns(h.sessionId); + expect(turn.state).toBe('cancelled'); + }); + + it('records a legacy rewrite as a sibling turn and moves the active leaf', async () => { + const h = await startedHarness(() => new BranchingAdapter()); + await h.inject({ + kind: 'agent.input', + clientReqId: 'original', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('original prompt')] }, + }); + h.adapter.emit({ type: 'session-ref', historyId: asHistoryId('native-source') }); + h.adapter.emit({ type: 'status', status: 'idle' }); + await settleEngineTasks(); + + await h.inject({ + kind: 'history.branch', + clientReqId: 'rewrite', + sourceSessionId: h.sessionId, + sourceMessageId: MessageIdSchema.parse('source-message'), + branchCursor: 'opaque-cursor', + content: [textBlock('edited prompt')], + }); + await vi.waitFor(() => + expect(h.sent).toContainEqual( + expect.objectContaining({ kind: 'session.started', replyTo: 'rewrite' }), + ), + ); + + const turns = await h.conversationStore.listTurns(h.sessionId); + expect(turns).toHaveLength(2); + const original = nullthrow(turns.find((turn) => turn.siblingOrdinal === 1)); + const replacement = nullthrow(turns.find((turn) => turn.siblingOrdinal === 2)); + expect(original.state).toBe('completed'); + // The replacement is a sibling under the rewritten turn's parent, never a child of the leaf. + expect(replacement).toMatchObject({ parentTurnId: null, state: 'running' }); + if (replacement.input.type !== 'prompt') throw new Error('expected a prompt turn'); + const prompt = await h.conversationStore.getPrompt(nullthrow(replacement.input.promptId)); + expect(prompt?.blocks).toEqual([{ type: 'text', text: 'edited prompt' }]); + const [record] = await h.store.load(); + expect(record.activeLeafTurnId).toBe(replacement.turnId); + expect(record.runs.at(-1)?.runId).toBe(replacement.runId); + }); + + it('resolves open operations and dead turns at boot, and replays the stored error', async () => { + const conversationStore = new InMemoryConversationStore(); + const sessionId = SessionIdSchema.parse('sess-recover'); + await conversationStore.persistTurnIntent({ + turn: { + turnId: TurnIdSchema.parse('turn-preparing'), + sessionId, + parentTurnId: null, + input: { type: 'shell-command', command: 'sleep 1' }, + runId: RunIdSchema.parse('run-dead'), + state: 'preparing', + createdAt: Date.now(), + }, + operation: { + operationId: OperationIdSchema.parse('op-interrupted'), + sessionId, + kind: 'turn.submit', + state: 'open', + createdAt: Date.now(), + }, + }); + await conversationStore.saveTurn({ + turnId: TurnIdSchema.parse('turn-running'), + sessionId, + parentTurnId: null, + siblingOrdinal: 2, + input: { type: 'shell-command', command: 'sleep 2' }, + runId: RunIdSchema.parse('run-dead'), + state: 'running', + createdAt: Date.now(), + }); + + const h = harness( + new InMemorySessionStore(), + () => new FakeAdapter(), + undefined, + undefined, + undefined, + undefined, + { conversationStore }, + ); + await h.engine.start(); + + const turns = await conversationStore.listTurns(sessionId); + expect(turns.map((turn) => turn.state)).toEqual(['failed', 'failed']); + expect(await conversationStore.listOpenOperations()).toHaveLength(0); + + // Replay happens before any validation, so even an unknown session replays the result. + await h.inject({ + kind: 'turn.submit', + clientReqId: 'replayed', + sessionId, + operationId: OperationIdSchema.parse('op-interrupted'), + input: { type: 'shell-command', command: 'sleep 1' }, + }); + expect(h.sent).toContainEqual( + expect.objectContaining({ + kind: 'request.failed', + replyTo: 'replayed', + code: 'operation_failed', + message: 'The daemon restarted before the turn was dispatched', + }), + ); + }); + + it('refuses a legacy turn input while an operation is open', async () => { + const h = await startedHarness(); + await h.conversationStore.persistTurnIntent({ + turn: { + turnId: TurnIdSchema.parse('turn-open'), + sessionId: h.sessionId, + parentTurnId: null, + input: { type: 'shell-command', command: 'sleep 1' }, + runId: RunIdSchema.parse('run-elsewhere'), + state: 'preparing', + createdAt: Date.now(), + }, + operation: { + operationId: OperationIdSchema.parse('op-open'), + sessionId: h.sessionId, + kind: 'turn.submit', + state: 'open', + createdAt: Date.now(), + }, + }); + + await h.inject({ + kind: 'agent.input', + clientReqId: 'input', + sessionId: h.sessionId, + input: { type: 'prompt', content: [textBlock('hello')] }, + }); + + expect(h.sent).toContainEqual({ + kind: 'request.failed', + replyTo: 'input', + code: 'busy', + message: `Session is busy: ${h.sessionId}`, + reportedInConversation: true, + }); + expect(h.adapter.sentInputs).toHaveLength(0); + }); +}); + +describe('commitRunning idempotence', () => { + const sessionId = SessionIdSchema.parse('sess-commit'); + + async function turnServiceFixture() { + const sent: WirePayload[] = []; + const transport: Transport = { + connect: () => Promise.resolve(), + send(message: ValidatedWireMessage) { + sent.push(message.payload); + }, + onMessage: () => noop, + onClose: () => noop, + close: noop, + }; + const registry = new SessionRecordRegistry(new InMemorySessionStore(), noop); + await Effect.runPromise( + registry.start((effect) => { + void Effect.runPromise(effect); + }), + ); + registry.register({ + sessionId, + kind: 'claude-code', + cwd: '/repo', + origin: { type: 'created' }, + createdAt: 1, + updatedAt: 1, + runs: [], + graphRevision: 0, + }); + const store = new InMemoryConversationStore(); + const turns = new ConversationTurnService(store, registry, transport, (effect) => { + void Effect.runPromise(effect); + }); + const intent = await Effect.runPromise( + turns.persistIntent({ + sessionId, + operationId: OperationIdSchema.parse('op-1'), + runId: RunIdSchema.parse('run-1'), + parentTurnId: null, + input: { type: 'shell-command', command: 'git status' }, + }), + ); + return { sent, registry, store, turns, intent }; + } + + it('a second commit for a resolved operation is a no-op: no error, no second graph move', async () => { + const { sent, registry, store, turns, intent } = await turnServiceFixture(); + + await Effect.runPromise(turns.commitRunning(intent)); + await Effect.runPromise(turns.commitRunning(intent)); + + expect((await store.getOperation(OperationIdSchema.parse('op-1')))?.state).toBe('succeeded'); + expect(registry.get(sessionId)?.graphRevision).toBe(1); + expect(sent.filter((payload) => payload.kind === 'conversation.graph.changed')).toHaveLength(1); + }); + + it('a commit that lost the resolve race runs no side effects at all', async () => { + const { sent, registry, store, turns, intent } = await turnServiceFixture(); + await Effect.runPromise(turns.resolveFailed(intent, { code: 'timeout', message: 'too slow' })); + + await Effect.runPromise(turns.commitRunning(intent)); + + const operation = await store.getOperation(OperationIdSchema.parse('op-1')); + expect(operation).toMatchObject({ state: 'failed', error: { code: 'timeout' } }); + expect(registry.get(sessionId)?.graphRevision).toBe(0); + expect(sent.filter((payload) => payload.kind === 'conversation.graph.changed')).toHaveLength(0); + // Nor tracking: a settle for this run must find nothing to flip. + turns.settleStop(sessionId, RunIdSchema.parse('run-1'), 'end_turn'); + await settleEngineTasks(); + expect((await store.listTurns(sessionId))[0].state).toBe('failed'); + }); + + it('a resolveFailed that lost the race returns the stored terminal result', async () => { + const { store, turns, intent } = await turnServiceFixture(); + await Effect.runPromise(turns.commitRunning(intent)); + + const result = await Effect.runPromise( + turns.resolveFailed(intent, { code: 'busy', message: 'late loser' }), + ); + + const stored = await store.getOperation(OperationIdSchema.parse('op-1')); + expect(result).toEqual(stored); + expect(result.state).toBe('succeeded'); + }); +}); diff --git a/packages/host/engine/src/__tests__/session-record-registry.test.ts b/packages/host/engine/src/__tests__/session-record-registry.test.ts new file mode 100644 index 000000000..837a74ef5 --- /dev/null +++ b/packages/host/engine/src/__tests__/session-record-registry.test.ts @@ -0,0 +1,77 @@ +import { asHistoryId } from '@linkcode/agent-adapter'; +import type { RunId, SessionId, SessionRecord } from '@linkcode/schema'; +import { Effect } from 'effect'; +import { noop } from 'foxts/noop'; +import { describe, expect, it } from 'vitest'; +import { SessionRecordRegistry } from '../session/session-record-registry'; +import { InMemorySessionStore } from '../session/session-store'; + +const sessionId = 'sess-registry' as SessionId; + +function makeRecord(): SessionRecord { + return { + sessionId, + kind: 'claude-code', + cwd: '/repo', + origin: { type: 'created' }, + createdAt: 1, + updatedAt: 1, + runs: [], + graphRevision: 0, + }; +} + +async function startedRegistry() { + const registry = new SessionRecordRegistry(new InMemorySessionStore(), noop); + await Effect.runPromise( + registry.start((effect) => { + void Effect.runPromise(effect); + }), + ); + registry.register(makeRecord()); + return registry; +} + +describe('session record registry run addressing', () => { + it('seals the addressed run, not the newest one', async () => { + const registry = await startedRegistry(); + const first = registry.beginRun(sessionId); + const second = registry.beginRun(sessionId); + + registry.sealRun(sessionId, first); + + const runs = registry.get(sessionId)?.runs ?? []; + expect(runs.find((run) => run.runId === first)?.endedAt).toBeTypeOf('number'); + expect(runs.find((run) => run.runId === second)?.endedAt).toBeUndefined(); + }); + + it('binds a history id to the addressed run while a newer run exists', async () => { + const registry = await startedRegistry(); + const first = registry.beginRun(sessionId); + const second = registry.beginRun(sessionId); + + registry.bindHistoryId(sessionId, first, asHistoryId('native-old')); + + const runs = registry.get(sessionId)?.runs ?? []; + expect(runs.find((run) => run.runId === first)?.historyId).toBe('native-old'); + expect(runs.find((run) => run.runId === second)?.historyId).toBeUndefined(); + }); + + it('reports only the newest run as current', async () => { + const registry = await startedRegistry(); + const first = registry.beginRun(sessionId); + const second = registry.beginRun(sessionId); + + expect(registry.isCurrentRun(sessionId, first)).toBe(false); + expect(registry.isCurrentRun(sessionId, second)).toBe(true); + expect(registry.isCurrentRun(sessionId, 'run-unknown' as RunId)).toBe(false); + }); + + it('adopts a caller-minted run id', async () => { + const registry = await startedRegistry(); + const minted = 'run-preminted' as RunId; + + expect(registry.beginRun(sessionId, { runId: minted })).toBe(minted); + expect(registry.get(sessionId)?.runs.at(-1)?.runId).toBe(minted); + }); +}); diff --git a/packages/host/engine/src/automation/turn-watcher.ts b/packages/host/engine/src/automation/turn-watcher.ts index 77ed80ab0..d2d1f15c6 100644 --- a/packages/host/engine/src/automation/turn-watcher.ts +++ b/packages/host/engine/src/automation/turn-watcher.ts @@ -11,6 +11,12 @@ export interface TurnResult { text: string; } +interface WatchTurnOptions { + readonly timeoutMs?: number; + /** Durable acknowledgement that must run once after dispatch acceptance or a terminal event. */ + readonly onDispatchAccepted?: Effect.Effect; +} + function joinSegments(segments: Map): string { return Array.from(segments.values()) .filter((text) => text.length > 0) @@ -28,14 +34,18 @@ function joinSegments(segments: Map): string { * a `send` rejection, or `opts.timeoutMs` elapsing. On every reject it best-effort cancels the turn so * the underlying session returns to idle. Interruption remains interruption and also cancels the turn. */ -export function watchTurn( +export function watchTurn( adapter: Pick, send: () => Promise, - opts: { timeoutMs?: number } = {}, -): Effect.Effect { + opts: WatchTurnOptions = {}, +): Effect.Effect { return Effect.gen(function* () { const segments = new Map(); const outcome = yield* Deferred.make(); + const acceptDispatch = + opts.onDispatchAccepted === undefined + ? Effect.void + : yield* Effect.cached(Effect.uninterruptible(opts.onDispatchAccepted)); const cancel = Effect.tryPromise({ try: () => adapter.send({ type: 'cancel' }), catch: (cause) => cause, @@ -114,11 +124,12 @@ export function watchTurn( ), () => Deferred.await(outcome).pipe( + Effect.tap(() => acceptDispatch), Effect.raceFirst( Effect.tryPromise({ try: () => send(), catch: (cause) => new AutomationDispatchFailure({ cause }), - }).pipe(Effect.andThen(Effect.never)), + }).pipe(Effect.andThen(acceptDispatch), Effect.andThen(Effect.never)), ), ), (unsubscribe) => Effect.sync(unsubscribe), diff --git a/packages/host/engine/src/conversation/request-handler.ts b/packages/host/engine/src/conversation/request-handler.ts index 426f1c2c3..02b45b496 100644 --- a/packages/host/engine/src/conversation/request-handler.ts +++ b/packages/host/engine/src/conversation/request-handler.ts @@ -1,32 +1,64 @@ import type { WirePayload } from '@linkcode/schema'; +import type { Transport } from '@linkcode/transport'; +import { createWireMessage } from '@linkcode/transport'; import { Effect } from 'effect'; import { RequestError } from '../failure'; +import type { SessionLifecycleService } from '../session/lifecycle-service'; import type { WireResponder } from '../wire/responder'; -import type { ConversationStore } from './conversation-store'; type ConversationRequest = Extract< WirePayload, { kind: 'turn.submit' | 'conversation.graph.get' | 'conversation.read' } >; -/** Declared wire surface for the turn tree. A known kind must fail loudly, never be silently - * ignored — every request is refused with a typed error until the submit saga and the projection - * land on top of {@link ConversationStore}. */ +/** Wire surface for the turn tree. `turn.submit` runs the submit saga; the read/projection kinds + * keep failing loudly — never silently ignored — until the projection lands. */ export class ConversationRequestHandler { constructor( - /** Held for the submit saga and projection reads that build on this handler. */ - readonly store: ConversationStore, + private readonly transport: Transport, + private readonly lifecycle: SessionLifecycleService, private readonly responder: WireResponder, ) {} handle(payload: ConversationRequest): Effect.Effect { + if (payload.kind !== 'turn.submit') { + return this.responder.reply( + payload.clientReqId, + Effect.fail( + new RequestError({ + code: 'unsupported', + message: `${payload.kind} is not implemented yet`, + }), + ), + ); + } return this.responder.reply( payload.clientReqId, - Effect.fail( - new RequestError({ - code: 'unsupported', - message: `${payload.kind} is not implemented yet`, - }), + this.lifecycle.submitTurn(payload).pipe( + Effect.flatMap((operation) => + Effect.sync(() => { + // A stored failure replays verbatim: its code/message ARE the terminal result. + this.transport.send( + createWireMessage( + operation.state === 'succeeded' + ? { + kind: 'turn.submitted', + replyTo: payload.clientReqId, + turnId: operation.turnId, + } + : { + kind: 'request.failed', + replyTo: payload.clientReqId, + code: operation.error.code, + message: operation.error.message, + ...(operation.error.reportedInConversation && { + reportedInConversation: true, + }), + }, + ), + ); + }), + ), ), ); } diff --git a/packages/host/engine/src/conversation/turn-service.ts b/packages/host/engine/src/conversation/turn-service.ts new file mode 100644 index 000000000..3afae3801 --- /dev/null +++ b/packages/host/engine/src/conversation/turn-service.ts @@ -0,0 +1,399 @@ +import { randomUUID } from 'node:crypto'; +import type { + ContentBlock, + ConversationOperation, + ConversationTurn, + ConversationTurnState, + OperationId, + PromptBlock, + PromptId, + PromptRecord, + ProviderTurnBinding, + RunId, + SessionId, + StopReason, + TurnId, + TurnInput, +} from '@linkcode/schema'; +import type { Transport } from '@linkcode/transport'; +import { createWireMessage } from '@linkcode/transport'; +import { Effect } from 'effect'; +import { OperationError, RequestError } from '../failure'; +import type { SessionRecordRegistry } from '../session/session-record-registry'; +import type { ConversationStore } from './conversation-store'; +import { ConversationSessionBusyError } from './conversation-store'; + +export function mintOperationId(): OperationId { + return `op-${randomUUID()}` as OperationId; +} + +function mintTurnId(): TurnId { + return `turn-${randomUUID()}` as TurnId; +} + +function mintPromptId(): PromptId { + return `prompt-${randomUUID()}` as PromptId; +} + +/** Durable prompt blocks from legacy prompt content: text only for now — binary attachments + * become `attachment_ref`s once the attachment store lands. */ +export function promptBlocksFromContent(content: ContentBlock[]): PromptBlock[] { + return content.flatMap((block) => + block.type === 'text' ? [{ type: 'text' as const, text: block.text }] : [], + ); +} + +/** What a submit wants persisted, before ids and ordinals exist. */ +export interface TurnIntentSpec { + readonly sessionId: SessionId; + readonly operationId: OperationId; + readonly runId: RunId; + readonly parentTurnId: TurnId | null; + readonly input: + | { readonly type: 'prompt'; readonly blocks: PromptBlock[] } + | { readonly type: 'command'; readonly name: string; readonly arguments?: string } + | { readonly type: 'shell-command'; readonly command: string }; +} + +export interface PersistedTurnIntent { + readonly turn: ConversationTurn; + readonly operation: Extract; +} + +/** A dispatch failure as the saga reports it. The stored row keeps `code`/`message` only; the flag + * says the rejection was also broadcast into the conversation live, which only the immediate reply + * can lean on — a replay after a disconnect has no live event behind it. */ +export interface TurnFailure { + readonly code: string; + readonly message: string; + readonly reportedInConversation?: true; +} + +export type TerminalOperation = + | Extract + | (Omit, 'error'> & { + readonly error: TurnFailure; + }); + +const TERMINAL_TURN_STATES = new Set(['completed', 'failed', 'cancelled']); + +interface RunningTurn { + readonly turn: ConversationTurn; + sawError: boolean; +} + +/** + * The durable side of turn execution: persists submit intents into the {@link ConversationStore}, + * commits/fails their operations, and tracks the running turn per session so adapter lifecycle + * events settle it. Every turn-starting input — new `turn.submit` or legacy `agent.input` — flows + * through here, so the graph never misses a turn. + */ +export class ConversationTurnService { + /** The running turn per session; settles are addressed by the turn's own runId. */ + private readonly running = new Map(); + + constructor( + private readonly store: ConversationStore, + private readonly records: SessionRecordRegistry, + private readonly transport: Transport, + private readonly runTask: (effect: Effect.Effect) => void, + ) {} + + getOperation( + operationId: OperationId, + ): Effect.Effect { + return storeOperation('conversation.operation.get', () => this.store.getOperation(operationId)); + } + + hasOpenOperation(sessionId: SessionId): Effect.Effect { + return storeOperation('conversation.operations.list', () => + this.store.listOpenOperations(sessionId), + ).pipe(Effect.map((operations) => operations.length > 0)); + } + + listTurns(sessionId: SessionId): Effect.Effect { + return storeOperation('conversation.turns.list', () => this.store.listTurns(sessionId)); + } + + listBindings(turnId: TurnId): Effect.Effect { + return storeOperation('conversation.bindings.list', () => this.store.listBindings(turnId)); + } + + deleteSession(sessionId: SessionId): Effect.Effect { + return storeOperation('conversation.delete-session', () => + this.store.deleteSession(sessionId), + ).pipe( + Effect.tap(() => + Effect.sync(() => { + this.running.delete(sessionId); + }), + ), + ); + } + + /** The durable commit point: turn (`preparing`, ordinal store-assigned inside the transaction), + * prompt, and open operation persist in one transaction, before any irreversible provider work. + * The store's own admission guard turns a racing intent into a typed `busy`. */ + persistIntent( + spec: TurnIntentSpec, + ): Effect.Effect { + return Effect.gen({ self: this }, function* () { + const now = Date.now(); + let prompt: PromptRecord | undefined; + let input: TurnInput; + if (spec.input.type === 'prompt') { + prompt = { + promptId: mintPromptId(), + blocks: spec.input.blocks, + contextAttachmentIds: [], + createdAt: now, + }; + input = { type: 'prompt', promptId: prompt.promptId }; + } else { + input = spec.input; + } + const turn: Omit = { + turnId: mintTurnId(), + sessionId: spec.sessionId, + parentTurnId: spec.parentTurnId, + input, + runId: spec.runId, + state: 'preparing', + createdAt: now, + }; + const operation = { + operationId: spec.operationId, + sessionId: spec.sessionId, + kind: 'turn.submit' as const, + state: 'open' as const, + createdAt: now, + }; + const persisted = yield* storeOperation('conversation.intent.persist', () => + this.store.persistTurnIntent({ turn, prompt, operation }), + ).pipe( + Effect.catch((error) => + Effect.fail( + error.cause instanceof ConversationSessionBusyError + ? new RequestError({ + code: 'busy', + message: 'Another operation is open on this session', + }) + : error, + ), + ), + ); + return { turn: persisted, operation }; + }); + } + + /** The provider accepted the dispatch: one transaction stores the success and flips the turn to + * `running`, and ONLY the call that transitioned the row runs the side effects — a concurrent + * commit (dispatch-timer rescue vs the send continuation) must move the graph exactly once. + * Uninterruptible: an interrupt between the store write and the graph move would strand a + * succeeded operation behind a stale active leaf; the whole chain is a few sync-SQLite hops. */ + commitRunning(intent: PersistedTurnIntent): Effect.Effect { + const turn: ConversationTurn = { ...intent.turn, state: 'running' }; + const operation: ConversationOperation = { + ...intent.operation, + state: 'succeeded', + turnId: turn.turnId, + resolvedAt: Date.now(), + }; + return storeOperation('conversation.operation.resolve', () => + this.store.resolveOperation(operation, turn), + ).pipe( + Effect.flatMap((transitioned) => + transitioned + ? Effect.sync(() => { + this.trackRunning(turn); + const graphRevision = this.records.commitGraphMove(turn.sessionId, turn.turnId); + if (graphRevision !== undefined) { + this.transport.send( + createWireMessage({ + kind: 'conversation.graph.changed', + sessionId: turn.sessionId, + graphRevision, + activeLeafTurnId: turn.turnId, + }), + ); + } + }) + : Effect.void, + ), + Effect.uninterruptible, + ); + } + + /** Store the typed failure — unless a concurrent resolver already stored a terminal result, in + * which case the STORED result is returned: the reply must never differ from what a retry of the + * operationId will replay. */ + resolveFailed( + intent: PersistedTurnIntent, + error: TurnFailure, + ): Effect.Effect { + return Effect.gen({ self: this }, function* () { + const operation = { + ...intent.operation, + state: 'failed' as const, + error: { code: error.code, message: error.message }, + resolvedAt: Date.now(), + }; + const transitioned = yield* storeOperation('conversation.operation.resolve', () => + this.store.resolveOperation(operation, { ...intent.turn, state: 'failed' }), + ); + if (!transitioned) { + const stored = yield* this.getOperation(intent.operation.operationId); + if (stored === undefined || stored.state === 'open') { + return yield* Effect.fail( + new OperationError({ + subsystem: 'store', + operation: 'conversation.operation.resolve', + publicMessage: 'The operation resolution was lost', + cause: undefined, + }), + ); + } + return stored; + } + const running = this.running.get(intent.turn.sessionId); + if (running?.turn.turnId === intent.turn.turnId) this.running.delete(intent.turn.sessionId); + return { ...operation, error }; + }); + } + + /** {@link resolveFailed} for exit paths inside a session-scoped fiber: enqueued on the engine + * task runner so an interrupting teardown never waits behind the store write. */ + resolveFailedDetached(intent: PersistedTurnIntent, error: TurnFailure): void { + this.runTask( + this.resolveFailed(intent, error).pipe( + Effect.catch((resolveError) => + Effect.logError( + 'Failed to record the rejected turn', + { sessionId: intent.turn.sessionId }, + resolveError.cause, + ), + ), + Effect.asVoid, + ), + ); + } + + /** Boot recovery: no adapter survives a restart, so every open operation and every non-terminal + * turn is dead. Resolve them as failed — a retry then replays a typed error instead of hanging, + * and the graph shows the attempt as `failed`, never absent. */ + recover(sessionIds: Iterable): Effect.Effect { + return Effect.gen({ self: this }, function* () { + const open = yield* storeOperation('conversation.operations.list', () => + this.store.listOpenOperations(), + ); + // Open operations can outlive their session record; sweep their sessions too. + const sweep = new Set(sessionIds); + for (let i = 0, len = open.length; i < len; i++) sweep.add(open[i].sessionId); + for (const sessionId of sweep) { + const turns = yield* this.listTurns(sessionId); + for (let i = 0, len = turns.length; i < len; i++) { + const turn = turns[i]; + if (TERMINAL_TURN_STATES.has(turn.state)) continue; + yield* storeOperation('conversation.turn.save', () => + this.store.saveTurn({ ...turn, state: 'failed' }), + ); + } + } + const resolvedAt = Date.now(); + for (let i = 0, len = open.length; i < len; i++) { + const operation = open[i]; + yield* storeOperation('conversation.operation.resolve', () => + this.store.resolveOperation({ + ...operation, + state: 'failed', + error: { + code: 'operation_failed', + message: 'The daemon restarted before the turn was dispatched', + }, + resolvedAt, + }), + ); + } + }); + } + + /** An adapter `error` while the run's turn is live; decides `failed` on a stop-less settle. */ + noteError(sessionId: SessionId, runId: RunId): void { + const entry = this.runningFor(sessionId, runId); + if (entry) entry.sawError = true; + } + + /** `stop` is the turn's own settle signal; `cancelled` is the only non-complete reason. */ + settleStop(sessionId: SessionId, runId: RunId, stopReason: StopReason): void { + this.settle(sessionId, runId, stopReason === 'cancelled' ? 'cancelled' : 'completed'); + } + + /** Fallback settle for turns that end without a `stop` frame: a failed turn's idle settle, or an + * adapter stopped/torn down mid-turn. */ + settleStatus(sessionId: SessionId, runId: RunId, status: 'idle' | 'stopped'): void { + const entry = this.runningFor(sessionId, runId); + if (!entry) return; + this.settle( + sessionId, + runId, + entry.sawError ? 'failed' : status === 'idle' ? 'completed' : 'cancelled', + ); + } + + private settle(sessionId: SessionId, runId: RunId, state: ConversationTurnState): void { + const entry = this.runningFor(sessionId, runId); + if (!entry) return; + this.running.delete(sessionId); + this.persistTurnState(entry.turn, state); + } + + /** The session's running turn, only when it belongs to `runId` — a replaced run's stragglers + * must not settle its successor's turn. */ + private runningFor(sessionId: SessionId, runId: RunId): RunningTurn | undefined { + const entry = this.running.get(sessionId); + if (entry === undefined) return undefined; + return entry.turn.runId === runId ? entry : undefined; + } + + private trackRunning(turn: ConversationTurn): void { + const stale = this.running.get(turn.sessionId); + // A new dispatch was admitted, so an unsettled predecessor demonstrably ended; close it out — + // as failed when an adapter error was seen during its run, never a guessed 'completed'. + if (stale && stale.turn.turnId !== turn.turnId) { + this.persistTurnState(stale.turn, stale.sawError ? 'failed' : 'completed'); + } + this.running.set(turn.sessionId, { turn, sawError: false }); + } + + /** Settles run off synchronous adapter callbacks, so persistence is enqueued best-effort. */ + private persistTurnState(turn: ConversationTurn, state: ConversationTurnState): void { + if (TERMINAL_TURN_STATES.has(turn.state)) return; + this.runTask( + storeOperation('conversation.turn.save', () => this.store.saveTurn({ ...turn, state })).pipe( + Effect.catch((error) => + Effect.logError( + error.publicMessage, + { operation: error.operation, sessionId: turn.sessionId }, + error.cause, + ), + ), + ), + ); + } +} + +function storeOperation( + operation: string, + run: () => Promise, +): Effect.Effect { + return Effect.tryPromise({ + try: run, + catch: (cause) => + new OperationError({ + subsystem: 'store', + operation, + publicMessage: 'Conversation store operation failed', + cause, + }), + }); +} diff --git a/packages/host/engine/src/engine.ts b/packages/host/engine/src/engine.ts index 1df1b13e2..8d8987e37 100644 --- a/packages/host/engine/src/engine.ts +++ b/packages/host/engine/src/engine.ts @@ -24,6 +24,7 @@ import { BrowserReplHost } from './browser/repl-host'; import { BrowserRequestHandler } from './browser/request-handler'; import { InMemoryConversationStore } from './conversation/conversation-store'; import { ConversationRequestHandler } from './conversation/request-handler'; +import { ConversationTurnService } from './conversation/turn-service'; import type { EngineDeps } from './deps'; import type { EngineFailure, OperationSubsystem } from './failure'; import { toOperationFailure } from './failure'; @@ -103,7 +104,6 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( deps.stateDir, fileHost, ); - const conversations = deps.conversationStore ?? new InMemoryConversationStore(); const plugins = new PluginService(deps.pluginFactory ?? createPluginProviderAdapter); const translator = deps.translator; const startOptions = new SessionStartOptionsResolver( @@ -146,6 +146,13 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( // predicate that gates claims on a live session. const simulators = deps.simulators; const browserBroker = new BrowserBrokerService(transport); + const conversationStore = deps.conversationStore ?? new InMemoryConversationStore(); + const conversationTurns = new ConversationTurnService( + conversationStore, + records, + transport, + runTask, + ); const sessions = new SessionOrchestrator( transport, factory, @@ -159,7 +166,7 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( deps.simulatorMcp?.release(sessionId); }, resources, - conversations, + conversationTurns, deps.browserToolsEnabled ? () => new BrowserReplHost((op, args) => browserBroker.dispatch(op, args)) : undefined, @@ -207,6 +214,7 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( startOptions, workspaces, worktrees, + conversationTurns, ); const sessionRequests = new SessionRequestHandler( transport, @@ -220,7 +228,11 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( sessionLifecycle, responder, ); - const conversationRequests = new ConversationRequestHandler(conversations, responder); + const conversationRequests = new ConversationRequestHandler( + transport, + sessionLifecycle, + responder, + ); const scheduler = new ScheduleService( transport, deps.scheduleStore ?? new InMemoryScheduleStore(), @@ -287,6 +299,9 @@ export const createEngineRuntime = Effect.fn('Engine.create')(function* ( yield* records.start((effect) => { runTask(effect); }); + // Before requests are accepted: open operations and non-terminal turns cannot outlive the + // adapters that ran them, and a retried operation must replay a terminal result. + yield* conversationTurns.recover(Array.from(records.values(), ({ sessionId }) => sessionId)); yield* worktrees.start(new Set(Array.from(records.values(), ({ sessionId }) => sessionId))); yield* tryOperation('store', 'workspaces.load', 'Failed to load workspaces', () => workspaces.start(), diff --git a/packages/host/engine/src/failure.ts b/packages/host/engine/src/failure.ts index 5f066afee..0a522b1db 100644 --- a/packages/host/engine/src/failure.ts +++ b/packages/host/engine/src/failure.ts @@ -1,4 +1,4 @@ -import { Data } from 'effect'; +import { Cause, Data } from 'effect'; interface FailureReporting { /** True only when an emitted conversation event already owns presentation of this failure. */ @@ -9,6 +9,8 @@ export type RequestErrorCode = | 'invalid_request' | 'not_found' | 'conflict' + /** A turn is running or another operation is open on the session; retry once it settles. */ + | 'busy' /** The request was understood and refused — the user withheld consent, not a broken call. */ | 'forbidden' | 'unsupported' @@ -104,6 +106,15 @@ export function toRequestFailure(error: unknown): RequestFailure { return { code: 'internal_error', message: 'Internal engine error' }; } +/** A storable failure for exits that bypass the typed error channel: interruption (teardown or a + * timeout race) and defects. Typed failures inside the cause keep their precise mapping. */ +export function causeToRequestFailure(cause: Cause.Cause): RequestFailure { + if (Cause.hasInterruptsOnly(cause)) { + return { code: 'cancelled', message: 'The turn was interrupted before dispatch completed' }; + } + return toRequestFailure(Cause.squash(cause)); +} + function withFailureReporting(error: FailureReporting, failure: RequestFailure): RequestFailure { return error.reportedInConversation === true ? { ...failure, reportedInConversation: true } diff --git a/packages/host/engine/src/session/lifecycle-service.ts b/packages/host/engine/src/session/lifecycle-service.ts index f3513db0e..9e60f3f26 100644 --- a/packages/host/engine/src/session/lifecycle-service.ts +++ b/packages/host/engine/src/session/lifecycle-service.ts @@ -5,19 +5,36 @@ import type { AgentKind, ContentBlock, MessageId, + OperationId, + RunId, SessionAutomation, SessionId, SessionRecord, StartOptions, + TurnId, + TurnSubmitInput, WorkspaceId, WorkspaceRecord, WorktreeRecord, } from '@linkcode/schema'; -import { Effect, Semaphore } from 'effect'; +import { Effect, Exit, Semaphore } from 'effect'; import { nullthrow } from 'foxts/guard'; import type { SessionDriver } from '../automation'; +import type { + ConversationTurnService, + PersistedTurnIntent, + TerminalOperation, +} from '../conversation/turn-service'; +import { mintOperationId, promptBlocksFromContent } from '../conversation/turn-service'; import type { EngineFailure } from '../failure'; -import { RequestError, toOperationFailure } from '../failure'; +import { + causeToRequestFailure, + OperationError, + OperationTimeout, + RequestError, + toOperationFailure, + toRequestFailure, +} from '../failure'; import type { WorkspaceRegistry } from '../workspace/workspace-registry'; import type { WorktreeService } from '../worktree/worktree-service'; import type { HistoryService } from './history-service'; @@ -33,6 +50,36 @@ import type { ResolvedStartOptions, SessionStartOptionsResolver } from './start- type RunEffect = (effect: Effect.Effect, options?: Effect.RunOptions) => Promise; +/** A wedged provider dispatch must fail the operation, never the session forever. */ +const TURN_SUBMIT_TIMEOUT_MS = 60_000; +/** Launch budget: the claude CLI can legitimately take ~3 minutes to cold-start at peak hours; + * the other harnesses bound their own startup well under this. */ +const LAUNCH_TIMEOUT_MS = 300_000; + +export interface TurnSubmitRequest { + readonly sessionId: SessionId; + readonly operationId: OperationId; + readonly input: TurnSubmitInput; + /** Absent = plain send onto the active leaf; `null` = new root lineage; a turn id = tip-continue + * or (once checkpoints exist) fork. */ + readonly parentTurnId?: TurnId | null; + readonly expectedGraphRevision?: number; +} + +/** Provider work a submit needs: none (live adapter continues), a cold resume, or a fresh session. */ +type TurnLaunch = 'continue' | 'resume' | 'fresh'; + +function toAgentInput(input: TurnSubmitInput): AgentInput { + if (input.type !== 'prompt') return input; + // attachment_ref blocks are refused at admit until the attachment store lands. + return { + type: 'prompt', + content: input.blocks.flatMap((block) => + block.type === 'text' ? [{ type: 'text' as const, text: block.text }] : [], + ), + }; +} + export class SessionLifecycleService { readonly driver: SessionDriver; private readonly importSemaphores = new Map(); @@ -47,6 +94,7 @@ export class SessionLifecycleService { private readonly startOptions: SessionStartOptionsResolver, private readonly workspaces: WorkspaceRegistry, private readonly worktrees: WorktreeService, + private readonly turns: ConversationTurnService, ) { this.driver = { createSession: ({ signal, ...options }) => @@ -104,6 +152,7 @@ export class SessionLifecycleService { if (worktree) yield* workspaceRegisterWorktree(workspaces, worktree, parent.workspaceId); } const now = Date.now(); + const runId = mintRunId(); const record: SessionRecord = { sessionId, kind: resolved.kind, @@ -112,12 +161,13 @@ export class SessionLifecycleService { createdVia: resolved.createdVia, createdAt: now, updatedAt: now, - runs: [{ runId: mintRunId(), startedAt: now, ...runOf(resolved, accountId) }], + runs: [{ runId, startedAt: now, ...runOf(resolved, accountId) }], graphRevision: 0, }; yield* sessions.startLive( replyTo, record, + runId, (adapter) => sessions.startAdapter(adapter, resolved), warnings, ); @@ -183,6 +233,7 @@ export class SessionLifecycleService { if (worktree) yield* workspaceRegisterWorktree(workspaces, worktree, parent.workspaceId); } const now = Date.now(); + const runId = mintRunId(); const record: SessionRecord = { sessionId, kind, @@ -190,14 +241,13 @@ export class SessionLifecycleService { origin: { type: 'imported', historyId, importedAt: now }, createdAt: now, updatedAt: now, - runs: [ - { runId: mintRunId(), historyId, startedAt: now, ...runOf(startOptions, accountId) }, - ], + runs: [{ runId, historyId, startedAt: now, ...runOf(startOptions, accountId) }], graphRevision: 0, }; yield* sessions.startLive( replyTo, record, + runId, (adapter) => history.resume(adapter, historyId, startOptions), warnings, ); @@ -252,47 +302,351 @@ export class SessionLifecycleService { ); } - const { history, sessions } = this; + const { history, sessions, turns } = this; const resolveForRecord = this.resolveForRecord.bind(this); const launchRun = this.launchRun.bind(this); return Effect.gen(function* () { + if (yield* turns.hasOpenOperation(sourceSessionId)) { + return yield* Effect.fail( + new RequestError({ + code: 'busy', + message: 'Another operation is open on this session', + }), + ); + } const resolved = yield* resolveForRecord(source); - yield* sessions.stopForReplacement(sourceSessionId); - const resolvedBranchCursor = - liveCursor.type === 'live' - ? yield* history.resolveLiveBranchCursor( - source.kind, - sourceHistoryId, - source.cwd, - liveCursor.offsetFromEnd, - liveCursor.contentFingerprint, - ) - : branchCursor; + // The runtime rewrite stays destructive for old clients, but the tree records the + // replacement non-destructively. Live-echo message ids are never persisted, so + // `sourceMessageId` cannot name a graph turn; best-effort, the replacement lands as a + // sibling of the active leaf. Nothing is guessed destructively. + const runId = mintRunId(); + const existingTurns = yield* turns.listTurns(sourceSessionId); + const activeLeaf = existingTurns.find((turn) => turn.turnId === source.activeLeafTurnId); + const intent = yield* turns.persistIntent({ + sessionId: sourceSessionId, + operationId: mintOperationId(), + runId, + parentTurnId: activeLeaf?.parentTurnId ?? null, + input: { type: 'prompt', blocks: promptBlocksFromContent(content) }, + }); + yield* Effect.gen(function* () { + yield* sessions.stopForReplacement(sourceSessionId); + const resolvedBranchCursor = + liveCursor.type === 'live' + ? yield* history.resolveLiveBranchCursor( + source.kind, + sourceHistoryId, + source.cwd, + liveCursor.offsetFromEnd, + liveCursor.contentFingerprint, + ) + : branchCursor; + yield* launchRun( + replyTo, + source, + resolved, + (adapter) => + history.branch( + adapter, + { historyId: sourceHistoryId, cursor: resolvedBranchCursor }, + resolved.options, + ), + { + initialInput: { type: 'prompt', content }, + preparedTurn: intent, + registerRecord: false, + rewindMessageId: sourceMessageId, + runId, + }, + ); + }).pipe( + // The dispatcher does not resolve saga-prepared intents; every failure exit — stop, + // branch, or dispatch failures, interrupts, defects — resolves here. + Effect.onExit((exit) => + Exit.isFailure(exit) + ? turns.resolveFailed(intent, causeToRequestFailure(exit.cause)).pipe( + Effect.catch(() => Effect.void), + Effect.asVoid, + ) + : Effect.void, + ), + ); + }); + }), + ); + } + + /** + * The `turn.submit` saga — idempotent by `operationId`, atomic from the client's view: + * replay → admit (short critical section) → persist intent (the durable commit point) → + * provider work + dispatch outside the semaphore under phase-scoped hard timeouts. Every + * post-persist outcome is committed-or-failed, never absent; the returned terminal operation + * is the reply. + */ + submitTurn(request: TurnSubmitRequest): Effect.Effect { + const { sessions, turns } = this; + const admitSubmit = this.admitSubmit.bind(this); + const relaunchFresh = this.relaunchFresh.bind(this); + const resumeSession = this.resumeSession.bind(this); + return Effect.gen(function* () { + // Replay before any validation: a reply lost to a disconnect must not duplicate a sibling. + const existing = yield* turns.getOperation(request.operationId); + if (existing !== undefined) { + // An operation id names one submit on one session; the same id from another session is a + // client defect, never a replay — answering would hand it that session's turn. + if (existing.sessionId !== request.sessionId) { + return yield* Effect.fail( + new RequestError({ + code: 'invalid_request', + message: 'The operation id belongs to another session', + }), + ); + } + if (existing.state !== 'open') return existing; + return yield* Effect.fail( + new RequestError({ code: 'busy', message: 'The operation is still in flight' }), + ); + } + const { intent, launch } = yield* admitSubmit(request); + const dispatch = Effect.gen(function* () { + if (launch !== 'continue') { + const launchSession = + launch === 'fresh' + ? relaunchFresh(request.sessionId, intent.turn.runId) + : resumeSession(undefined, request.sessionId, { + runId: intent.turn.runId, + baseTurnId: intent.turn.parentTurnId ?? undefined, + }); + yield* launchSession.pipe( + Effect.timeoutOrElse({ + duration: LAUNCH_TIMEOUT_MS, + orElse: () => + Effect.fail( + new OperationTimeout({ + operation: 'turn.submit.launch', + duration: LAUNCH_TIMEOUT_MS, + publicMessage: 'The provider did not start in time', + }), + ), + }), + ); + } + // The adapter contract emits `running` at dispatch, so a send outliving the timer while + // the turn is visibly running is committed, not failed — pi-style send() spans the whole + // turn. commitRunning completes before the race interrupts the losing send fiber, so its + // exit backstop then sees an already-resolved operation and stands down. + yield* sessions.sendInput(request.sessionId, toAgentInput(request.input), intent).pipe( + Effect.timeoutOrElse({ + duration: TURN_SUBMIT_TIMEOUT_MS, + orElse: (): Effect.Effect => + sessions.isTurnRunning(request.sessionId) + ? turns.commitRunning(intent) + : Effect.fail( + new OperationTimeout({ + operation: 'turn.submit', + duration: TURN_SUBMIT_TIMEOUT_MS, + publicMessage: 'The provider did not accept the turn in time', + }), + ), + }), + ); + }); + return yield* dispatch.pipe( + Effect.matchEffect({ + onSuccess: () => + turns.getOperation(request.operationId).pipe( + Effect.flatMap((operation) => + operation === undefined || operation.state === 'open' + ? Effect.fail( + new OperationError({ + subsystem: 'store', + operation: 'turn.submit.commit', + publicMessage: 'The dispatched turn was not committed', + cause: undefined, + }), + ) + : Effect.succeed(operation), + ), + ), + // Any post-persist failure resolves the operation; a retry replays this stored error. + onFailure: (error) => turns.resolveFailed(intent, toRequestFailure(error)), + }), + // Interrupts and defects bypass the typed match; the open operation must still resolve, + // or the session wedges `busy` until the daemon restarts. + Effect.onExit((exit) => + Exit.isFailure(exit) + ? turns.resolveFailed(intent, causeToRequestFailure(exit.cause)).pipe( + Effect.catch((error) => + Effect.logError( + 'Failed to resolve the interrupted turn', + { sessionId: request.sessionId }, + error.cause, + ), + ), + Effect.asVoid, + ) + : Effect.void, + ), + ); + }); + } + + /** + * Steps 1–2 of the submit saga under the per-session critical section: typed `busy` while a + * turn runs or another operation is open, parent/revision validation for explicit-parent + * submits, then the durable intent persist. At most one operation can be open per session and + * this section is serialized, so the revision check here is decisive. + */ + private admitSubmit( + request: TurnSubmitRequest, + ): Effect.Effect<{ intent: PersistedTurnIntent; launch: TurnLaunch }, EngineFailure> { + return this.sessionSemaphore(request.sessionId).withPermit( + Effect.suspend(() => { + const record = this.records.get(request.sessionId); + if (!record) { + return Effect.fail( + new RequestError({ + code: 'not_found', + message: `Unknown session: ${request.sessionId}`, + }), + ); + } + if (this.sessions.isBusy(request.sessionId)) { + return Effect.fail( + new RequestError({ code: 'busy', message: `Session is busy: ${request.sessionId}` }), + ); + } + if ( + request.input.type === 'prompt' && + request.input.blocks.some((block) => block.type === 'attachment_ref') + ) { + // Seam: attachment existence/readiness/capability validation lands with the store. + return Effect.fail( + new RequestError({ + code: 'unsupported', + message: 'Prompt attachments are not supported yet', + }), + ); + } + // Seam: the worktree co-leaseholder busy gate joins this critical section later. + const { sessions, turns } = this; + return Effect.gen(function* () { + if (yield* turns.hasOpenOperation(request.sessionId)) { + return yield* Effect.fail( + new RequestError({ + code: 'busy', + message: 'Another operation is open on this session', + }), + ); + } + let parentTurnId: TurnId | null; + let launch: TurnLaunch; + if (request.parentTurnId === undefined) { + // Plain send: no guards — targets the current active leaf under the busy rules alone. + parentTurnId = record.activeLeafTurnId ?? null; + launch = 'continue'; + } else if (request.parentTurnId === null) { + if (request.expectedGraphRevision !== record.graphRevision) { + return yield* Effect.fail( + new RequestError({ code: 'conflict', message: 'The conversation graph has moved' }), + ); + } + parentTurnId = null; + launch = 'fresh'; + } else { + const existingTurns = yield* turns.listTurns(request.sessionId); + const parent = existingTurns.find((turn) => turn.turnId === request.parentTurnId); + if (!parent) { + return yield* Effect.fail( + new RequestError({ + code: 'not_found', + message: `Unknown turn: ${request.parentTurnId}`, + }), + ); + } + if (parent.state !== 'completed') { + return yield* Effect.fail( + new RequestError({ + code: 'conflict', + message: 'The parent turn has not completed', + }), + ); + } + if (request.expectedGraphRevision !== record.graphRevision) { + return yield* Effect.fail( + new RequestError({ code: 'conflict', message: 'The conversation graph has moved' }), + ); + } + if (request.parentTurnId === record.activeLeafTurnId) { + // Tip-continue on the active lineage: the provider history head IS this leaf. + parentTurnId = request.parentTurnId; + launch = 'continue'; + } else { + // Fork seam: per-turn provider checkpoints are not captured yet, so this read + // always finds none and every interior/edit fork is refused loudly. + const bindings = yield* turns.listBindings(request.parentTurnId); + return yield* Effect.fail( + new RequestError({ + code: 'unsupported', + message: + bindings.length === 0 + ? 'This turn has no provider checkpoint to fork from' + : 'Forking from an earlier turn is not supported yet', + }), + ); + } + } + const liveRunId = + launch === 'continue' ? sessions.liveRunId(request.sessionId) : undefined; + if (launch === 'continue' && liveRunId === undefined) launch = 'resume'; + const intent = yield* turns.persistIntent({ + sessionId: request.sessionId, + operationId: request.operationId, + runId: liveRunId ?? mintRunId(), + parentTurnId, + input: request.input, + }); + return { intent, launch }; + }); + }), + ); + } + + /** Replace the session's adapter with a fresh provider session under the same LinkCode id — + * the `parentTurnId: null` (new root lineage) submit path. */ + private relaunchFresh(sessionId: SessionId, runId: RunId): Effect.Effect { + return this.sessionSemaphore(sessionId).withPermit( + Effect.suspend(() => { + const record = this.records.get(sessionId); + if (!record) { + return Effect.fail( + new RequestError({ code: 'not_found', message: `Unknown session: ${sessionId}` }), + ); + } + const { sessions } = this; + const resolveForRecord = this.resolveForRecord.bind(this); + const launchRun = this.launchRun.bind(this); + return Effect.gen(function* () { + const resolved = yield* resolveForRecord(record); + yield* sessions.stopForReplacement(sessionId); yield* launchRun( - replyTo, - source, + undefined, + record, resolved, - (adapter) => - history.branch( - adapter, - { historyId: sourceHistoryId, cursor: resolvedBranchCursor }, - resolved.options, - ), - { - initialInput: { type: 'prompt', content }, - registerRecord: false, - rewindMessageId: sourceMessageId, - }, + (adapter) => sessions.startAdapter(adapter, resolved.options), + { registerRecord: false, runId }, ); }); }), ); } - /** Wake a cold session in place under the same LinkCode id. */ + /** Wake a cold session in place under the same LinkCode id. `run` lets a submit pre-mint the + * relaunch's run identity so the persisted turn references it. */ resumeSession( replyTo: string | undefined, sessionId: SessionId, + run: { runId?: RunId; baseTurnId?: TurnId } = {}, ): Effect.Effect { return this.sessionSemaphore(sessionId).withPermit( Effect.suspend(() => { @@ -331,6 +685,7 @@ export class SessionLifecycleService { } yield* launchRun(replyTo, record, resolved, resumeStrategy(historyId, resolved.options), { historyId, + ...run, }); }); }), @@ -470,7 +825,8 @@ export class SessionLifecycleService { } /** Record the run this launch begins, then bind the record to a fresh adapter. Every relaunch of - * an existing record goes through here, so `runs` has exactly one writer. */ + * an existing record goes through here, so `runs` has exactly one writer. `runId`/`baseTurnId` + * let a submit pre-mint the run its persisted turn references. */ private launchRun( replyTo: string | undefined, record: SessionRecord, @@ -478,20 +834,26 @@ export class SessionLifecycleService { startAdapter: (adapter: AgentAdapter) => Effect.Effect, options: { historyId?: AgentHistoryId; + runId?: RunId; + baseTurnId?: TurnId; initialInput?: AgentInput; + preparedTurn?: PersistedTurnIntent; registerRecord?: boolean; rewindMessageId?: MessageId; } = {}, ): Effect.Effect { - const { historyId, ...startOptions } = options; + const { baseTurnId, historyId, runId, ...startOptions } = options; return Effect.suspend(() => { - this.records.beginRun(record.sessionId, { + const launchedRunId = this.records.beginRun(record.sessionId, { ...runOf(resolved.options, resolved.accountId), historyId, + runId, + baseTurnId, }); return this.sessions.startLive( replyTo, record, + launchedRunId, startAdapter, resolved.warnings, startOptions, @@ -526,6 +888,7 @@ export class SessionLifecycleService { sessionId, ); const now = Date.now(); + const runId = mintRunId(); const record: SessionRecord = { sessionId, kind: startOptions.kind, @@ -535,11 +898,11 @@ export class SessionLifecycleService { automation: options.automation, createdAt: now, updatedAt: now, - runs: [{ runId: mintRunId(), startedAt: now, ...runOf(startOptions, accountId) }], + runs: [{ runId, startedAt: now, ...runOf(startOptions, accountId) }], graphRevision: 0, }; if (startOptions.cwd) yield* workspaceTouch(workspaces, startOptions.cwd); - yield* sessions.startLive(undefined, record, (adapter) => + yield* sessions.startLive(undefined, record, runId, (adapter) => sessions.startAdapter(adapter, startOptions), ); return record.sessionId; diff --git a/packages/host/engine/src/session/live-session.ts b/packages/host/engine/src/session/live-session.ts index d2251c759..26f32bf37 100644 --- a/packages/host/engine/src/session/live-session.ts +++ b/packages/host/engine/src/session/live-session.ts @@ -11,6 +11,7 @@ import type { ContentBlock, EffortLevel, MessageId, + RunId, SessionId, SessionInfo, } from '@linkcode/schema'; @@ -58,6 +59,8 @@ export class LiveSession { constructor( readonly adapter: AgentAdapter, sessionId: SessionId, + /** The `SessionRun` this adapter serves — run bookkeeping addresses runs by this id. */ + readonly runId: RunId, readonly scope: Scope.Closeable, readonly closed: Deferred.Deferred, ) { diff --git a/packages/host/engine/src/session/orchestrator.ts b/packages/host/engine/src/session/orchestrator.ts index a640f7c28..ef3659f5a 100644 --- a/packages/host/engine/src/session/orchestrator.ts +++ b/packages/host/engine/src/session/orchestrator.ts @@ -7,6 +7,7 @@ import type { ContentBlock, McpWarning, MessageId, + RunId, SessionId, SessionInfo, SessionRecord, @@ -17,7 +18,8 @@ import { Cause, Deferred, Effect, Exit, Scope } from 'effect'; import type { AgentRuntimeService } from '../agent/runtime-service'; import type { TurnResult } from '../automation/turn-watcher'; import { watchTurn } from '../automation/turn-watcher'; -import type { ConversationStore } from '../conversation/conversation-store'; +import type { ConversationTurnService, PersistedTurnIntent } from '../conversation/turn-service'; +import { mintOperationId, promptBlocksFromContent } from '../conversation/turn-service'; import type { EngineFailure } from '../failure'; import { OperationError, RequestError, toOperationFailure } from '../failure'; import { observeOperation, recordLiveSessions } from '../observability'; @@ -41,11 +43,18 @@ export class SessionOrchestrator { reportFailure: (effect: Effect.Effect) => void, private readonly onStopped: (sessionId: SessionId) => void, private readonly resources: ResourceService, - private readonly conversations: ConversationStore, + private readonly turns: ConversationTurnService, private readonly browserTools?: BrowserToolsetFactory, ) { - this.events = new SessionEventProcessor(transport, records, runtimes, reportFailure, resources); - this.inputs = new SessionInputDispatcher(records, this.events, resources); + this.events = new SessionEventProcessor( + transport, + records, + runtimes, + reportFailure, + resources, + turns, + ); + this.inputs = new SessionInputDispatcher(records, this.events, resources, turns); } private get(sessionId: SessionId): LiveSession | undefined { @@ -76,6 +85,17 @@ export class SessionOrchestrator { return session !== undefined && (session.turnInputActive || session.status === 'running'); } + /** The adapter has visibly emitted `running` — deliberately narrower than {@link isBusy}: + * `turnInputActive` is set by the dispatch itself and proves nothing about acceptance. */ + isTurnRunning(sessionId: SessionId): boolean { + return this.sessions.get(sessionId)?.status === 'running'; + } + + /** The run the live adapter serves; `undefined` doubles as the cold-session signal. */ + liveRunId(sessionId: SessionId): RunId | undefined { + return this.sessions.get(sessionId)?.runId; + } + /** The running adapter's history capabilities — asked of the live instance rather than a fresh * one, so a caller about to tear it down learns what *this* session can do. */ historyCapabilities(sessionId: SessionId): AgentHistoryCapabilities | undefined { @@ -87,10 +107,16 @@ export class SessionOrchestrator { if (session) this.events.broadcast(sessionId, session.replay()); } - sendInput(sessionId: SessionId, input: AgentInput): Effect.Effect { + sendInput( + sessionId: SessionId, + input: AgentInput, + prepared?: PersistedTurnIntent, + ): Effect.Effect { return Effect.suspend(() => { const session = this.requireSession(sessionId); - return session.run(Effect.suspend(() => this.inputs.send(sessionId, session, input))); + return session.run( + Effect.suspend(() => this.inputs.send(sessionId, session, input, prepared)), + ); }); } @@ -109,22 +135,14 @@ export class SessionOrchestrator { } delete(sessionId: SessionId): Effect.Effect { - const { conversations, resources } = this; + const { resources } = this; return Effect.gen({ self: this }, function* () { const session = this.sessions.get(sessionId); if (session) { yield* this.teardown(sessionId, session, 'session.delete'); } yield* resources.deleteSession(sessionId); - yield* Effect.tryPromise({ - try: () => conversations.deleteSession(sessionId), - catch: (cause) => - toOperationFailure(cause, { - subsystem: 'store', - operation: 'conversation.delete-session', - publicMessage: 'Failed to delete conversation graph', - }), - }); + yield* this.turns.deleteSession(sessionId); yield* this.records.delete(sessionId); }); } @@ -167,20 +185,56 @@ export class SessionOrchestrator { } session.turnInputActive = true; const content: ContentBlock[] = [{ type: 'text', text }]; + const { records, turns } = this; return session.run( - Effect.sync(() => { - this.events.broadcast(sessionId, [ - { type: 'user-message', messageId: nextMessageId(), content }, - ]); - this.records.setTitleFromContent(sessionId, content); - }).pipe( - Effect.andThen( - watchTurn( - session.adapter, - () => session.adapter.send({ type: 'prompt', content }), - opts, + Effect.gen({ self: this }, function* () { + if (yield* turns.hasOpenOperation(sessionId)) { + return yield* Effect.fail( + new RequestError({ code: 'busy', message: `Session is busy: ${sessionId}` }), + ); + } + const intent = yield* turns.persistIntent({ + sessionId, + operationId: mintOperationId(), + runId: session.runId, + parentTurnId: records.get(sessionId)?.activeLeafTurnId ?? null, + input: { type: 'prompt', blocks: promptBlocksFromContent(content) }, + }); + const result = yield* Effect.sync(() => { + this.events.broadcast(sessionId, [ + { type: 'user-message', messageId: nextMessageId(), content }, + ]); + records.setTitleFromContent(sessionId, content); + }).pipe( + Effect.andThen( + watchTurn(session.adapter, () => session.adapter.send({ type: 'prompt', content }), { + ...opts, + onDispatchAccepted: turns.commitRunning(intent), + }), ), - ), + Effect.onExit((exit) => + Exit.isFailure(exit) + ? turns + .resolveFailed(intent, { + code: 'operation_failed', + message: 'Automation prompt failed', + }) + .pipe( + Effect.catch((error) => + Effect.logError( + 'Failed to record the rejected automation turn', + { sessionId }, + error.cause, + ), + ), + ) + : Effect.void, + ), + ); + turns.settleStop(sessionId, session.runId, result.stopReason); + if (session.status !== 'running') session.turnInputActive = false; + return result; + }).pipe( Effect.onExit((exit) => Exit.isFailure(exit) ? Effect.sync(() => { @@ -194,14 +248,17 @@ export class SessionOrchestrator { }); } - /** Bind a record to a live adapter. The record's current run must already be last in `runs`. */ + /** Bind a record to a live adapter serving `runId` — the run the caller just recorded. */ startLive( replyTo: string | undefined, record: SessionRecord, + runId: RunId, startAdapter: (adapter: AgentAdapter) => Effect.Effect, mcpWarnings: readonly McpWarning[] = [], options: { initialInput?: AgentInput; + /** Turn intent already persisted for `initialInput`; its dispatch commits or fails it. */ + preparedTurn?: PersistedTurnIntent; registerRecord?: boolean; rewindMessageId?: MessageId; } = {}, @@ -219,7 +276,7 @@ export class SessionOrchestrator { const { browserTools } = this; const discardFailedStart = (session: LiveSession): Effect.Effect => this.discardFailedStart(record.sessionId, session); - const { initialInput, registerRecord = true, rewindMessageId } = options; + const { initialInput, preparedTurn, registerRecord = true, rewindMessageId } = options; return observeOperation( Effect.gen(function* () { const sessionId = record.sessionId; @@ -227,7 +284,7 @@ export class SessionOrchestrator { if (browserTools) adapter.attachBrowserTools?.(browserTools); const scope = yield* Scope.fork(parentScope); const closed = yield* Deferred.make(); - const session = new LiveSession(adapter, sessionId, scope, closed); + const session = new LiveSession(adapter, sessionId, runId, scope, closed); const startupEvents: AgentEvent[] = []; let bufferEvents = rewindMessageId !== undefined; session.listen((event) => { @@ -255,13 +312,20 @@ export class SessionOrchestrator { yield* startAdapter(adapter); if (sessions.get(sessionId) !== session) return yield* Effect.interrupt; }); + // Exit-based, not tapError: an interrupted start (submit timeout, teardown racing the + // launch) must also unregister the session, or it stays a zombie in `'starting'` whose + // next submit would 'continue' into an adapter that never started. yield* session .run(startAdapterSession) .pipe( - Effect.tapError(() => - discardFailedStart(session).pipe( - Effect.catch((error) => Effect.logError('Failed to discard session record', error)), - ), + Effect.onExit((exit) => + Exit.isFailure(exit) + ? discardFailedStart(session).pipe( + Effect.catch((error) => + Effect.logError('Failed to discard session record', error), + ), + ) + : Effect.void, ), ); if (rewindMessageId !== undefined) { @@ -276,7 +340,7 @@ export class SessionOrchestrator { } if (initialInput !== undefined) { yield* session - .run(Effect.suspend(() => inputs.send(sessionId, session, initialInput))) + .run(Effect.suspend(() => inputs.send(sessionId, session, initialInput, preparedTurn))) .pipe( Effect.mapError((cause) => toOperationFailure(cause, { @@ -359,7 +423,9 @@ export class SessionOrchestrator { Effect.suspend(() => { if (!this.remove(sessionId, session)) return Effect.void; if (releaseSession) this.onStopped(sessionId); - this.records.sealCurrentRun(sessionId); + // Teardown mid-turn kills the turn without a stop frame; settle it here. + this.turns.settleStatus(sessionId, session.runId, 'stopped'); + this.records.sealRun(sessionId, session.runId); return recordLiveSessions(this.sessions.size); }), ), @@ -384,7 +450,7 @@ export class SessionOrchestrator { Effect.andThen( Effect.sync(() => { session.stopListening(); - this.records.sealCurrentRun(sessionId); + this.records.sealRun(sessionId, session.runId); }), ), Effect.andThen(stopBestEffort(session.adapter)), diff --git a/packages/host/engine/src/session/session-event-processor.ts b/packages/host/engine/src/session/session-event-processor.ts index 590798c45..0d90833be 100644 --- a/packages/host/engine/src/session/session-event-processor.ts +++ b/packages/host/engine/src/session/session-event-processor.ts @@ -4,12 +4,26 @@ import type { Transport } from '@linkcode/transport'; import { createWireMessage } from '@linkcode/transport'; import { Effect } from 'effect'; import type { AgentRuntimeService } from '../agent/runtime-service'; +import type { ConversationTurnService } from '../conversation/turn-service'; import type { ResourceService } from '../resource/service'; import type { LiveSession } from './live-session'; import type { SessionRecordRegistry } from './session-record-registry'; const SOURCE_TOOL_KINDS = new Set(['fetch', 'read', 'search']); +/** Events describing the SESSION rather than a turn — status and catalogs. A replaced adapter's + * stragglers of these kinds must not paint the session with a dead run's state; turn-scoped + * events keep their old attribution and pass through. */ +const SESSION_SCOPED_EVENT_TYPES = new Set([ + 'status', + 'approval-policy-update', + 'model-update', + 'effort-update', + 'available-commands-update', + 'available-models-update', + 'capabilities-update', +]); + /** Applies adapter events to live state, durable records, and wire projections. */ export class SessionEventProcessor { constructor( @@ -18,6 +32,7 @@ export class SessionEventProcessor { private readonly runtimes: AgentRuntimeService, private readonly reportFailure: (effect: Effect.Effect) => void, private readonly resources: ResourceService, + private readonly turns: ConversationTurnService, ) {} broadcast(sessionId: SessionId, events: Iterable): void { @@ -95,20 +110,33 @@ export class SessionEventProcessor { // Adapter callbacks are synchronous; contain failures to this session instead of throwing into // the SDK operation that emitted the event. try { + if ( + SESSION_SCOPED_EVENT_TYPES.has(event.type) && + !this.records.isCurrentRun(sessionId, session.runId) + ) { + return; + } this.broadcast(sessionId, session.apply(event)); this.registerResources(sessionId, event); switch (event.type) { case 'status': - if (event.status === 'stopped') this.records.sealCurrentRun(sessionId); + if (event.status === 'stopped') this.records.sealRun(sessionId, session.runId); + if (event.status === 'idle' || event.status === 'stopped') { + this.turns.settleStatus(sessionId, session.runId, event.status); + } + break; + case 'stop': + this.turns.settleStop(sessionId, session.runId, event.stopReason); break; case 'session-ref': - this.records.bindHistoryId(sessionId, event.historyId); + this.records.bindHistoryId(sessionId, session.runId, event.historyId); break; case 'title-update': this.records.setProviderTitle(sessionId, event.title); break; case 'error': if (event.code === AUTH_FAILED_ERROR_CODE) this.runtimes.refresh(); + this.turns.noteError(sessionId, session.runId); break; default: break; diff --git a/packages/host/engine/src/session/session-input-dispatcher.ts b/packages/host/engine/src/session/session-input-dispatcher.ts index 371a7e2d1..0096f3f3e 100644 --- a/packages/host/engine/src/session/session-input-dispatcher.ts +++ b/packages/host/engine/src/session/session-input-dispatcher.ts @@ -1,8 +1,10 @@ import { nextMessageId } from '@linkcode/agent-adapter'; import type { AgentInput, SessionId } from '@linkcode/schema'; import { agentCommandMatches } from '@linkcode/schema'; -import { Effect } from 'effect'; -import { OperationError, RequestError } from '../failure'; +import { Cause, Effect, Exit } from 'effect'; +import type { ConversationTurnService, PersistedTurnIntent } from '../conversation/turn-service'; +import { mintOperationId, promptBlocksFromContent } from '../conversation/turn-service'; +import { causeToRequestFailure, OperationError, RequestError } from '../failure'; import type { ResourceService } from '../resource/service'; import { RESOURCE_CONTEXT_SENTINEL } from '../resource/service'; import { assertAttachmentContentAllowed } from './attachment-guard'; @@ -16,12 +18,16 @@ export class SessionInputDispatcher { private readonly records: SessionRecordRegistry, private readonly events: SessionEventProcessor, private readonly resources: ResourceService, + private readonly turns: ConversationTurnService, ) {} + /** `prepared` is a submit-saga intent already persisted for this dispatch; without one, a + * turn-starting legacy input persists its own plain-send intent — the graph misses no turns. */ send( sessionId: SessionId, session: LiveSession, input: AgentInput, + prepared?: PersistedTurnIntent, ): Effect.Effect { const startsTurn = input.type === 'prompt' || input.type === 'command' || input.type === 'shell-command'; @@ -56,9 +62,22 @@ export class SessionInputDispatcher { this.events.rejectInput(sessionId, error.message); return Effect.fail(error); } - const { events, records, resources } = this; + const { events, records, resources, turns } = this; const promptMessageId = input.type === 'prompt' ? nextMessageId() : undefined; + // Set synchronously, before the first await, so a same-tick second turn input cannot slip + // past the gate above while this one is still validating; every failure exit releases it. + if (startsTurn) session.turnInputActive = true; return Effect.gen(function* () { + // A submit operation in flight owns the session; legacy inputs respect the same admit gate. + if (startsTurn && prepared === undefined && (yield* turns.hasOpenOperation(sessionId))) { + const error = new RequestError({ + code: 'busy', + message: `Session is busy: ${sessionId}`, + reportedInConversation: true, + }); + events.rejectInput(sessionId, error.message); + return yield* Effect.fail(error); + } let adapterInput: AgentInput = input; if (input.type === 'prompt') { yield* Effect.try({ @@ -82,73 +101,131 @@ export class SessionInputDispatcher { ), ); } - if (startsTurn) session.turnInputActive = true; - // Echo before awaiting send: provider events can outrun the dispatch acknowledgement. - if (promptMessageId !== undefined && input.type === 'prompt') { - events.broadcast(sessionId, session.trackPrompt(promptMessageId, input.content)); - records.setTitleFromContent(sessionId, input.content); - } else if (input.type === 'command' || input.type === 'shell-command') { - const text = - input.type === 'command' - ? `/${input.name}${input.arguments ? ` ${input.arguments}` : ''}` - : `$ ${input.command}`; - events.broadcast(sessionId, [ - { - type: 'user-message', - messageId: nextMessageId(), - content: [{ type: 'text', text }], - }, - ]); + // The durable commit point precedes the irreversible dispatch: kill or failure past here + // leaves a `failed` turn, never an absent one. + let intent = prepared; + if (startsTurn && intent === undefined) { + intent = yield* turns.persistIntent({ + sessionId, + operationId: mintOperationId(), + runId: session.runId, + parentTurnId: records.get(sessionId)?.activeLeafTurnId ?? null, + input: + input.type === 'prompt' + ? { type: 'prompt', blocks: promptBlocksFromContent(input.content) } + : input, + }); } - const responseInput = - input.type === 'permission-response' || input.type === 'question-response' - ? input + const persisted = intent; + const dispatch = Effect.gen(function* () { + // Echo before awaiting send: provider events can outrun the dispatch acknowledgement. + if (promptMessageId !== undefined && input.type === 'prompt') { + events.broadcast(sessionId, session.trackPrompt(promptMessageId, input.content)); + records.setTitleFromContent(sessionId, input.content); + } else if (input.type === 'command' || input.type === 'shell-command') { + const text = + input.type === 'command' + ? `/${input.name}${input.arguments ? ` ${input.arguments}` : ''}` + : `$ ${input.command}`; + events.broadcast(sessionId, [ + { + type: 'user-message', + messageId: nextMessageId(), + content: [{ type: 'text', text }], + }, + ]); + } + const responseInput = + input.type === 'permission-response' || input.type === 'question-response' + ? input + : undefined; + const respondingAsk = responseInput + ? session.interactions.beginResponse(responseInput) : undefined; - const respondingAsk = responseInput - ? session.interactions.beginResponse(responseInput) - : undefined; - if (responseInput && respondingAsk) { - events.broadcast(sessionId, [ - { - type: 'prompt-response-status', - requestId: responseInput.requestId, - status: 'responding', - }, - ]); - } - yield* Effect.tryPromise({ - try: () => session.adapter.send(adapterInput), - catch: (cause) => - new OperationError({ - subsystem: 'agent', - operation: 'session.input', - publicMessage: 'Agent input was rejected', - cause, - ...(startsTurn && { reportedInConversation: true }), - }), - }).pipe( - Effect.catch((error) => - Effect.sync(() => { - if (responseInput && respondingAsk) { - events.broadcast( - sessionId, - session.interactions.restoreResponse(responseInput.requestId, respondingAsk), - ); - } - if (promptMessageId !== undefined) { - events.broadcast(sessionId, session.untrackPrompt(promptMessageId)); - } - if (startsTurn && session.status !== 'running') session.turnInputActive = false; - if (startsTurn) events.rejectInput(sessionId, error.publicMessage); - }).pipe(Effect.andThen(Effect.fail(error))), - ), + if (responseInput && respondingAsk) { + events.broadcast(sessionId, [ + { + type: 'prompt-response-status', + requestId: responseInput.requestId, + status: 'responding', + }, + ]); + } + yield* Effect.tryPromise({ + try: () => session.adapter.send(adapterInput), + catch: (cause) => + new OperationError({ + subsystem: 'agent', + operation: 'session.input', + publicMessage: 'Agent input was rejected', + cause, + ...(startsTurn && { reportedInConversation: true }), + }), + }).pipe( + Effect.tapError((error) => + Effect.sync(() => { + if (responseInput && respondingAsk) { + events.broadcast( + sessionId, + session.interactions.restoreResponse(responseInput.requestId, respondingAsk), + ); + } + if (promptMessageId !== undefined) { + events.broadcast(sessionId, session.untrackPrompt(promptMessageId)); + } + if (startsTurn) events.rejectInput(sessionId, error.publicMessage); + }), + ), + ); + if (responseInput && respondingAsk) { + const resolution = session.interactions.resolveResponse(responseInput, respondingAsk); + if (resolution) events.broadcast(sessionId, [resolution]); + } + // The provider accepted the dispatch: the turn flips to running and the default leaf moves. + if (persisted !== undefined) yield* turns.commitRunning(persisted); + // Synchronous controls may not produce lifecycle events; only a running turn keeps the gate. + if (startsTurn && session.status !== 'running') session.turnInputActive = false; + }); + // A saga-prepared intent is resolved by its saga's own exit backstop in the request fiber; + // this fiber resolves only the intents it minted, so the saga's precise error (e.g. the + // dispatch timeout) can never lose the store race to this fiber's interrupt exit. + if (persisted === undefined || prepared !== undefined) return yield* dispatch; + // Every non-success exit past the durable commit point — dispatch rejection, commit failure, + // interrupt, defect — must resolve the operation, or the session wedges `busy`. + return yield* dispatch.pipe( + Effect.onExit((exit) => { + if (!Exit.isFailure(exit)) return Effect.void; + const failure = causeToRequestFailure(exit.cause); + // An interrupt exit means the session scope is tearing down, and awaiting store hops in + // this finalizer would block Scope.close — that one path stays detached. + if (Cause.hasInterruptsOnly(exit.cause)) { + return Effect.sync(() => { + turns.resolveFailedDetached(persisted, failure); + }); + } + // Typed failures await, so the failure reply can never beat the stored resolution and + // hand an instant retry a spurious `busy`. + return turns.resolveFailed(persisted, failure).pipe( + Effect.catch((resolveError) => + Effect.logError( + 'Failed to record the rejected turn', + { sessionId }, + resolveError.cause, + ), + ), + Effect.asVoid, + ); + }), ); - if (responseInput && respondingAsk) { - const resolution = session.interactions.resolveResponse(responseInput, respondingAsk); - if (resolution) events.broadcast(sessionId, [resolution]); - } - // Synchronous controls may not produce lifecycle events; only a running turn keeps the gate. - if (startsTurn && session.status !== 'running') session.turnInputActive = false; - }); + }).pipe( + Effect.onExit((exit) => + startsTurn && Exit.isFailure(exit) + ? Effect.sync(() => { + // A failed or interrupted dispatch can exit before a lifecycle event releases it. + if (session.status !== 'running') session.turnInputActive = false; + }) + : Effect.void, + ), + ); } } diff --git a/packages/host/engine/src/session/session-record-registry.ts b/packages/host/engine/src/session/session-record-registry.ts index 5580c8a45..616be6187 100644 --- a/packages/host/engine/src/session/session-record-registry.ts +++ b/packages/host/engine/src/session/session-record-registry.ts @@ -10,6 +10,7 @@ import type { SessionRecord, SessionRun, StartOptions, + TurnId, } from '@linkcode/schema'; import { Effect } from 'effect'; import { nullthrow } from 'foxts/guard'; @@ -145,9 +146,11 @@ export class SessionRecordRegistry { ); } - bindHistoryId(sessionId: SessionId, historyId: AgentHistoryId): void { + /** Bind a provider history to the run that reported it — by id, never "the newest row": a + * replacement adapter's late `session-ref` must not rebind whatever run launched after it. */ + bindHistoryId(sessionId: SessionId, runId: RunId, historyId: AgentHistoryId): void { const record = this.records.get(sessionId); - const run = record?.runs.at(-1); + const run = record?.runs.find((candidate) => candidate.runId === runId); if (!record || !run || run.historyId === historyId) return; run.historyId = historyId; this.persist(record); @@ -177,28 +180,49 @@ export class SessionRecordRegistry { this.persist(record); } - sealCurrentRun(sessionId: SessionId): void { + /** Seal the run the ending adapter served — by id, so a stale adapter's death cannot stamp + * `endedAt` onto a replacement run that is still live. */ + sealRun(sessionId: SessionId, runId: RunId): void { const record = this.records.get(sessionId); - const run = record?.runs.at(-1); + const run = record?.runs.find((candidate) => candidate.runId === runId); if (!record || !run || run.endedAt !== undefined) return; run.endedAt = Date.now(); this.persist(record); } + /** Whether `runId` is the session's current (newest) run — the source-side gate that drops a + * replaced adapter's session-scoped events. */ + isCurrentRun(sessionId: SessionId, runId: RunId): boolean { + return this.records.get(sessionId)?.runs.at(-1)?.runId === runId; + } + + /** A submit committed: the graph gained a running turn — bump the revision and move the host + * default leaf. Not an identity change (`SessionInfo` projects neither field), so it notifies + * nobody; clients follow `conversation.graph.changed`. Returns the new revision. */ + commitGraphMove(sessionId: SessionId, leafTurnId: TurnId): number | undefined { + const record = this.records.get(sessionId); + if (!record) return undefined; + record.graphRevision += 1; + record.activeLeafTurnId = leafTurnId; + this.persist(record); + return record.graphRevision; + } + /** The single writer for a relaunch's run entry. `historyId` is known up front only when the - * relaunch resumes a transcript; a fresh one gets it later via {@link bindHistoryId}. */ - beginRun( - sessionId: SessionId, - run: Omit = {}, - ): void { + * relaunch resumes a transcript; a fresh one gets it later via {@link bindHistoryId}. Returns + * the run's identity (caller-supplied or minted here) even when the record is gone, so a + * launch already in flight keeps an addressable run. */ + beginRun(sessionId: SessionId, run: Omit = {}): RunId { + const runId = run.runId ?? mintRunId(); const record = this.records.get(sessionId); - if (!record) return; - record.runs.push({ runId: mintRunId(), startedAt: Date.now(), ...definedFields(run) }); + if (!record) return runId; + record.runs.push({ startedAt: Date.now(), ...definedFields(run), runId }); this.persist(record); // A new run re-points the identity `list()` projects — `accountId`, `historyId` — so clients // must revalidate. Nothing else announces a relaunch: it sends no `session.started`, and a // resumed run already carries the historyId that would otherwise notify via `bindHistoryId`. this.onChanged(sessionId, 'updated'); + return runId; } setTitleFromContent(sessionId: SessionId, content: ContentBlock[]): void {