// @effect-diagnostics nodeBuiltinImport:off - test fixtures build real SQLite files on disk.import { assert, describe, it } from "@effect/vitest";import * as NodeFS from "node:fs";import * as NodeOS from "node:os";import * as NodePath from "node:path";import * as NodeSqlite from "node:sqlite"; import { DevSeedError, MAX_SEED_THREAD_LIMIT, seedDevDatabase } from "./dev-seed.ts"; const SEEDED_AT = "2026-07-26T00:00:00.000Z";/** * A completion timestamp left on a still-running turn by a settle that a later * checkpoint reverted. Far enough in the future to be unmistakable if it leaks * into a seeded row. */const STALE_COMPLETED_AT = "2099-01-01T00:00:00.000Z"; /** * The subset of the real schema the seeder touches. `monitor_json` is included * only in the source, to stand in for the migration drift between an installed * app and a worktree that is a migration behind. */function createSchema( database: NodeSqlite.DatabaseSync, options: { readonly withMonitor: boolean; readonly legacyThreads?: boolean },) { database.exec(`CREATE TABLE projection_projects ( project_id TEXT PRIMARY KEY, title TEXT NOT NULL, workspace_root TEXT NOT NULL, scripts_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, deleted_at TEXT)`); // A pre-017/023 source has neither archived_at nor latest_user_message_at. database.exec(`CREATE TABLE projection_threads ( thread_id TEXT PRIMARY KEY, project_id TEXT NOT NULL, title TEXT NOT NULL, latest_turn_id TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, deleted_at TEXT${options.legacyThreads ? "" : ", archived_at TEXT, latest_user_message_at TEXT"}, pending_approval_count INTEGER NOT NULL DEFAULT 0, pending_user_input_count INTEGER NOT NULL DEFAULT 0 ${options.withMonitor ? ", monitor_json TEXT" : ""})`); database.exec(`CREATE TABLE projection_thread_messages ( message_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, role TEXT NOT NULL, text TEXT NOT NULL, is_streaming INTEGER NOT NULL, attachments_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)`); // `sequence` (migration 008) is what the app orders activities by, ahead of // created_at, so the copy's per-thread cap has to see it. database.exec(`CREATE TABLE projection_thread_activities ( activity_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, tone TEXT NOT NULL, kind TEXT NOT NULL, summary TEXT NOT NULL, payload_json TEXT NOT NULL, sequence INTEGER, created_at TEXT NOT NULL)`); database.exec(`CREATE TABLE projection_thread_sessions ( thread_id TEXT PRIMARY KEY, status TEXT NOT NULL, provider_name TEXT, active_turn_id TEXT, last_error TEXT, updated_at TEXT NOT NULL)`); // Mirrors migration 005: `checkpoint_turn_count` matters because // dropPendingTurnStarts keys its guard on it, and a fixture without the // column only ever exercises the degraded path. database.exec(`CREATE TABLE projection_turns ( row_id INTEGER PRIMARY KEY AUTOINCREMENT, thread_id TEXT NOT NULL, turn_id TEXT, state TEXT NOT NULL, requested_at TEXT NOT NULL, completed_at TEXT, checkpoint_turn_count INTEGER, checkpoint_files_json TEXT NOT NULL, UNIQUE (thread_id, turn_id))`); database.exec(`CREATE TABLE projection_thread_proposed_plans ( plan_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, plan_markdown TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)`); database.exec(`CREATE TABLE projection_pending_approvals ( request_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL)`); database.exec(`CREATE TABLE projection_state ( projector TEXT PRIMARY KEY, last_applied_sequence INTEGER NOT NULL, updated_at TEXT NOT NULL)`); database.exec(`CREATE TABLE orchestration_events ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL UNIQUE, stream_id TEXT NOT NULL, event_type TEXT NOT NULL, payload_json TEXT NOT NULL)`); database.exec(`CREATE TABLE provider_session_runtime ( thread_id TEXT PRIMARY KEY, provider_name TEXT NOT NULL, adapter_key TEXT NOT NULL, status TEXT NOT NULL, last_seen_at TEXT NOT NULL)`); // Cleared alongside the event log: a receipt records a command whose events // the seed just deleted, so a retry of that command would be treated as // already applied. database.exec(`CREATE TABLE orchestration_command_receipts ( command_id TEXT PRIMARY KEY, aggregate_kind TEXT NOT NULL, aggregate_id TEXT NOT NULL, accepted_at TEXT NOT NULL, result_sequence INTEGER NOT NULL, status TEXT NOT NULL, error TEXT)`);} /** Source DB with `threadCount` threads, oldest first so recency ordering is testable. */function makeSource( path: string, threadCount: number, activitiesPerThread = 3, options?: { readonly legacyThreads?: boolean; readonly minimalRows?: boolean },) { const legacyThreads = options?.legacyThreads ?? false; // Thread rows only, for the scale test: building sessions and messages for // 32k threads costs half a minute and proves nothing the thread-keyed // statements don't already exercise. const minimalRows = options?.minimalRows ?? false; const database = new NodeSqlite.DatabaseSync(path); createSchema(database, { withMonitor: true, legacyThreads }); database .prepare( `INSERT INTO projection_projects VALUES ('p1','Project','/repo','[]','${SEEDED_AT}','${SEEDED_AT}',NULL)`, ) .run(); // One transaction rather than one per insert: at 32k threads the implicit // per-statement commits dominate the fixture's runtime. database.exec("BEGIN"); for (let index = 0; index < threadCount; index += 1) { const threadId = `t${String(index)}`; // Later index → later timestamp → more recent. const at = `2026-07-${String(10 + index).padStart(2, "0")}T00:00:00.000Z`; if (legacyThreads) { database .prepare( `INSERT INTO projection_threads (thread_id, project_id, title, latest_turn_id, created_at, updated_at, pending_approval_count, pending_user_input_count, monitor_json) VALUES (?,?,?,?,?,?,?,?,?)`, ) .run(threadId, "p1", `Thread ${String(index)}`, `turn-${threadId}`, at, at, 4, 2, "{}"); } else { database .prepare( `INSERT INTO projection_threads (thread_id, project_id, title, latest_turn_id, created_at, updated_at, latest_user_message_at, pending_approval_count, pending_user_input_count, monitor_json) VALUES (?,?,?,?,?,?,?,?,?,?)`, ) .run(threadId, "p1", `Thread ${String(index)}`, `turn-${threadId}`, at, at, at, 4, 2, "{}"); } // The newest thread is left mid-turn, as a copy taken while an agent is // working would be: a running turn with no completion, and a message still // streaming into it. The one before it is left "pending" — a user message // accepted but whose provider turn never started. const midFlight = index === threadCount - 1; const queued = index === threadCount - 2; database .prepare( `INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, completed_at, checkpoint_files_json) VALUES (?,?,?,?,?,'[]')`, ) .run( threadId, // A pending start has no turn id — that is what identifies it. queued ? null : `turn-${threadId}`, midFlight ? "running" : queued ? "pending" : "completed", at, // The running turn carries a stale completion from an earlier settle // that a later checkpoint reverted, which must not survive as the // interruption time. midFlight ? STALE_COMPLETED_AT : queued ? null : at, ); if (minimalRows) { continue; } database .prepare(`INSERT INTO projection_thread_sessions VALUES (?,'running','claude',?,'boom',?)`) .run(threadId, `turn-${threadId}`, at); database .prepare(`INSERT INTO projection_thread_messages VALUES (?,?, 'user','hello',?,?,?,?)`) .run( `m-${threadId}`, threadId, midFlight ? 1 : 0, JSON.stringify([{ type: "image", id: `${threadId}-image`, name: "shot.png" }]), at, at, ); for (let a = 0; a < activitiesPerThread; a += 1) { database .prepare( `INSERT INTO projection_thread_activities VALUES (?,?,'neutral','tool','ran',?,?,?)`, ) .run( `a-${threadId}-${String(a)}`, threadId, "{}", a, `2026-07-10T00:00:0${String(a)}.000Z`, ); } } database.exec("COMMIT"); database.close();} function makeTarget(path: string, options?: { readonly dropProposedPlans?: boolean }) { const database = new NodeSqlite.DatabaseSync(path); // No monitor_json: the target is a migration behind, as a worktree often is. createSchema(database, { withMonitor: false }); database .prepare( `INSERT INTO projection_projects VALUES ('stale','Stale','/old','[]','${SEEDED_AT}','${SEEDED_AT}',NULL)`, ) .run(); // History from the target's own past life, which must not survive the seed. database .prepare(`INSERT INTO orchestration_events (event_id, stream_id, event_type, payload_json) VALUES ('old-event','stale-thread','thread.created','{}')`) .run(); // A binding for a thread the seed is about to delete. database .prepare(`INSERT INTO provider_session_runtime VALUES ('stale-thread','claude','claude-code','running','${SEEDED_AT}')`) .run(); // A receipt for a command whose events the seed is about to delete. database .prepare(`INSERT INTO orchestration_command_receipts VALUES ('old-command','thread','stale-thread','${SEEDED_AT}',1,'accepted',NULL)`) .run(); if (options?.dropProposedPlans) { // A target far enough behind that a whole table is missing (migration 013). database.exec("DROP TABLE projection_thread_proposed_plans"); } database.close();} const withDatabases = <A>( run: (paths: { readonly source: string; readonly target: string }) => A, options?: { readonly threads?: number; readonly activities?: number; readonly dropProposedPlans?: boolean; readonly legacySource?: boolean; readonly minimalRows?: boolean; },): A => { const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-dev-seed-")); const source = NodePath.join(directory, "source.sqlite"); const target = NodePath.join(directory, "target.sqlite"); makeSource(source, options?.threads ?? 5, options?.activities ?? 3, { legacyThreads: options?.legacySource ?? false, minimalRows: options?.minimalRows ?? false, }); makeTarget(target, { dropProposedPlans: options?.dropProposedPlans ?? false }); try { return run({ source, target }); } finally { NodeFS.rmSync(directory, { recursive: true, force: true }); }}; const query = <T>(path: string, sql: string): Array<T> => { const database = new NodeSqlite.DatabaseSync(path, { readOnly: true }); try { return database.prepare(sql).all() as Array<T>; } finally { database.close(); }}; describe("seedDevDatabase", () => { it("copies the most recent threads and their project", () => { withDatabases(({ source, target }) => { const summary = seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 2, activityLimit: 10, seededAt: SEEDED_AT, }); assert.equal(summary.threads, 2); assert.equal(summary.projects, 1); const titles = query<{ title: string }>( target, "SELECT title FROM projection_threads ORDER BY latest_user_message_at DESC", ).map((row) => row.title); // Threads 4 and 3 are the newest of the five. assert.deepStrictEqual(titles, ["Thread 4", "Thread 3"]); }); }); it("replaces whatever the target held before", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 10, seededAt: SEEDED_AT, }); const projects = query<{ project_id: string }>( target, "SELECT project_id FROM projection_projects", ).map((row) => row.project_id); assert.deepStrictEqual(projects, ["p1"]); }); }); // The target can be a migration behind the source; copying its columns // blindly would fail on the first schema change. it("skips columns the target does not have", () => { withDatabases(({ source, target }) => { const summary = seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 10, seededAt: SEEDED_AT, }); assert.deepStrictEqual(summary.skippedColumns, ["projection_threads.monitor_json"]); assert.equal(summary.threads, 1); }); }); // A copied "running" session has no agent behind it, so the thread would spin // forever and the session reaper would skip it. it("neutralizes live session state", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const sessions = query<{ status: string; active_turn_id: string | null }>( target, "SELECT status, active_turn_id FROM projection_thread_sessions", ); assert.isAbove(sessions.length, 0); for (const session of sessions) { assert.equal(session.status, "stopped"); assert.isNull(session.active_turn_id); } const [threadsWithLatestTurn] = query<{ count: number }>( target, "SELECT COUNT(*) count FROM projection_threads WHERE latest_turn_id IS NOT NULL", ); assert.equal(threadsWithLatestTurn?.count, 0); }); }); // Approvals are not copied, so the badge counts must not survive. it("clears pending counts that have no rows behind them", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const [counts] = query<{ approvals: number; inputs: number }>( target, "SELECT SUM(pending_approval_count) approvals, SUM(pending_user_input_count) inputs FROM projection_threads", ); assert.equal(counts?.approvals, 0); assert.equal(counts?.inputs, 0); }); }); // A copied running turn has no agent left to finish it. The session override // alone does not cover it: the turn is read independently, and an unsettled // one keeps the thread spinning and its timeline unfoldable forever. it("settles turns copied while still running", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const turns = query<{ state: string; completed_at: string | null }>( target, "SELECT state, completed_at FROM projection_turns", ); assert.isAbove(turns.length, 0); for (const turn of turns) { assert.notEqual(turn.state, "running"); assert.isNotNull(turn.completed_at); // A settle that a later checkpoint reverted can leave a completion on a // running turn; carrying it over would date the interruption to then. assert.notEqual(turn.completed_at, STALE_COMPLETED_AT); } }); }); // The server deletes pending starts rather than settling them // (clearPendingProjectionTurnsByThread). They also cannot be settled // coherently: turn_id is NULL, so an "interrupted" one is a terminal turn // with no id — a shape no real turn has. it("drops pending turn starts instead of settling them", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const [pending] = query<{ count: number }>( target, "SELECT COUNT(*) count FROM projection_turns WHERE state = 'pending' OR turn_id IS NULL", ); assert.equal(pending?.count, 0); }); }); // A burst of tool activity lands many rows on the same timestamp. Capping on // created_at alone then cuts arbitrarily among them, and can keep exactly the // rows the timeline treats as oldest. it("caps activities by the order the app displays them", () => { withDatabases(({ source, target }) => { const sourceDatabase = new NodeSqlite.DatabaseSync(source); sourceDatabase.exec("DELETE FROM projection_thread_activities"); const insert = sourceDatabase.prepare( `INSERT INTO projection_thread_activities VALUES (?, 't4', 'neutral', 'tool', 'ran', '{}', ?, ?)`, ); const tied = "2026-07-10T00:00:00.000Z"; // Timestamps tie within a burst... for (const sequence of [1, 2, 3]) { insert.run(`tie-${String(sequence)}`, sequence, tied); } // ...and can disagree with sequence outright under clock skew. The app // ranks by sequence first, so this row outranks every tied one despite // its older timestamp. insert.run("skewed-4", 4, "2026-07-09T00:00:00.000Z"); sourceDatabase.close(); seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 2, seededAt: SEEDED_AT, }); const kept = query<{ activity_id: string }>( target, "SELECT activity_id FROM projection_thread_activities ORDER BY sequence DESC", ).map((row) => row.activity_id); // The two highest sequences — not the two newest timestamps, and not // whichever two the storage order yielded. assert.deepStrictEqual(kept, ["skewed-4", "tie-3"]); }); }); it("uses activity id as the final cap tiebreaker", () => { withDatabases(({ source, target }) => { const sourceDatabase = new NodeSqlite.DatabaseSync(source); sourceDatabase.exec("DELETE FROM projection_thread_activities"); const insert = sourceDatabase.prepare( `INSERT INTO projection_thread_activities VALUES (?, 't4', 'neutral', 'tool', 'ran', '{}', 7, ?)`, ); const tied = "2026-07-10T00:00:00.000Z"; insert.run("tie-a", tied); insert.run("tie-z", tied); sourceDatabase.close(); seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 1, seededAt: SEEDED_AT, }); const kept = query<{ activity_id: string }>( target, "SELECT activity_id FROM projection_thread_activities", ); assert.deepStrictEqual(kept, [{ activity_id: "tie-z" }]); }); }); // The delete mirrors the server's predicate, which spares rows carrying a // checkpoint. Without the `checkpoint_turn_count IS NULL` clause a checkpoint // row that happened to be pending would be destroyed along with the starts. it("spares a pending row that carries a checkpoint", () => { withDatabases(({ source, target }) => { // A checkpoint row on a thread inside the copied range. const sourceDatabase = new NodeSqlite.DatabaseSync(source); sourceDatabase .prepare( `INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, checkpoint_turn_count, checkpoint_files_json) VALUES ('t4', NULL, 'pending', ?, 7, '[]')`, ) .run(SEEDED_AT); sourceDatabase.close(); seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const checkpoints = query<{ checkpoint_turn_count: number }>( target, "SELECT checkpoint_turn_count FROM projection_turns WHERE checkpoint_turn_count IS NOT NULL", );Lines 1–500 of 971. Load more
// @effect-diagnostics nodeBuiltinImport:off - test fixtures build real SQLite files on disk. import { assert, describe, it } from "@effect/vitest"; import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeSqlite from "node:sqlite"; import { DevSeedError, MAX_SEED_THREAD_LIMIT, seedDevDatabase } from "./dev-seed.ts"; const SEEDED_AT = "2026-07-26T00:00:00.000Z"; /** * A completion timestamp left on a still-running turn by a settle that a later * checkpoint reverted. Far enough in the future to be unmistakable if it leaks * into a seeded row. */ const STALE_COMPLETED_AT = "2099-01-01T00:00:00.000Z"; /** * The subset of the real schema the seeder touches. `monitor_json` is included * only in the source, to stand in for the migration drift between an installed * app and a worktree that is a migration behind. */ function createSchema( database: NodeSqlite.DatabaseSync, options: { readonly withMonitor: boolean; readonly legacyThreads?: boolean }, ) { database.exec(`CREATE TABLE projection_projects ( project_id TEXT PRIMARY KEY, title TEXT NOT NULL, workspace_root TEXT NOT NULL, scripts_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, deleted_at TEXT)`); // A pre-017/023 source has neither archived_at nor latest_user_message_at. database.exec(`CREATE TABLE projection_threads ( thread_id TEXT PRIMARY KEY, project_id TEXT NOT NULL, title TEXT NOT NULL, latest_turn_id TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, deleted_at TEXT${options.legacyThreads ? "" : ", archived_at TEXT, latest_user_message_at TEXT"}, pending_approval_count INTEGER NOT NULL DEFAULT 0, pending_user_input_count INTEGER NOT NULL DEFAULT 0 ${options.withMonitor ? ", monitor_json TEXT" : ""})`); database.exec(`CREATE TABLE projection_thread_messages ( message_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, role TEXT NOT NULL, text TEXT NOT NULL, is_streaming INTEGER NOT NULL, attachments_json TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)`); // `sequence` (migration 008) is what the app orders activities by, ahead of // created_at, so the copy's per-thread cap has to see it. database.exec(`CREATE TABLE projection_thread_activities ( activity_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, tone TEXT NOT NULL, kind TEXT NOT NULL, summary TEXT NOT NULL, payload_json TEXT NOT NULL, sequence INTEGER, created_at TEXT NOT NULL)`); database.exec(`CREATE TABLE projection_thread_sessions ( thread_id TEXT PRIMARY KEY, status TEXT NOT NULL, provider_name TEXT, active_turn_id TEXT, last_error TEXT, updated_at TEXT NOT NULL)`); // Mirrors migration 005: `checkpoint_turn_count` matters because // dropPendingTurnStarts keys its guard on it, and a fixture without the // column only ever exercises the degraded path. database.exec(`CREATE TABLE projection_turns ( row_id INTEGER PRIMARY KEY AUTOINCREMENT, thread_id TEXT NOT NULL, turn_id TEXT, state TEXT NOT NULL, requested_at TEXT NOT NULL, completed_at TEXT, checkpoint_turn_count INTEGER, checkpoint_files_json TEXT NOT NULL, UNIQUE (thread_id, turn_id))`); database.exec(`CREATE TABLE projection_thread_proposed_plans ( plan_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, plan_markdown TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)`); database.exec(`CREATE TABLE projection_pending_approvals ( request_id TEXT PRIMARY KEY, thread_id TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL)`); database.exec(`CREATE TABLE projection_state ( projector TEXT PRIMARY KEY, last_applied_sequence INTEGER NOT NULL, updated_at TEXT NOT NULL)`); database.exec(`CREATE TABLE orchestration_events ( sequence INTEGER PRIMARY KEY AUTOINCREMENT, event_id TEXT NOT NULL UNIQUE, stream_id TEXT NOT NULL, event_type TEXT NOT NULL, payload_json TEXT NOT NULL)`); database.exec(`CREATE TABLE provider_session_runtime ( thread_id TEXT PRIMARY KEY, provider_name TEXT NOT NULL, adapter_key TEXT NOT NULL, status TEXT NOT NULL, last_seen_at TEXT NOT NULL)`); // Cleared alongside the event log: a receipt records a command whose events // the seed just deleted, so a retry of that command would be treated as // already applied. database.exec(`CREATE TABLE orchestration_command_receipts ( command_id TEXT PRIMARY KEY, aggregate_kind TEXT NOT NULL, aggregate_id TEXT NOT NULL, accepted_at TEXT NOT NULL, result_sequence INTEGER NOT NULL, status TEXT NOT NULL, error TEXT)`); } /** Source DB with `threadCount` threads, oldest first so recency ordering is testable. */ function makeSource( path: string, threadCount: number, activitiesPerThread = 3, options?: { readonly legacyThreads?: boolean; readonly minimalRows?: boolean }, ) { const legacyThreads = options?.legacyThreads ?? false; // Thread rows only, for the scale test: building sessions and messages for // 32k threads costs half a minute and proves nothing the thread-keyed // statements don't already exercise. const minimalRows = options?.minimalRows ?? false; const database = new NodeSqlite.DatabaseSync(path); createSchema(database, { withMonitor: true, legacyThreads }); database .prepare( `INSERT INTO projection_projects VALUES ('p1','Project','/repo','[]','${SEEDED_AT}','${SEEDED_AT}',NULL)`, ) .run(); // One transaction rather than one per insert: at 32k threads the implicit // per-statement commits dominate the fixture's runtime. database.exec("BEGIN"); for (let index = 0; index < threadCount; index += 1) { const threadId = `t${String(index)}`; // Later index → later timestamp → more recent. const at = `2026-07-${String(10 + index).padStart(2, "0")}T00:00:00.000Z`; if (legacyThreads) { database .prepare( `INSERT INTO projection_threads (thread_id, project_id, title, latest_turn_id, created_at, updated_at, pending_approval_count, pending_user_input_count, monitor_json) VALUES (?,?,?,?,?,?,?,?,?)`, ) .run(threadId, "p1", `Thread ${String(index)}`, `turn-${threadId}`, at, at, 4, 2, "{}"); } else { database .prepare( `INSERT INTO projection_threads (thread_id, project_id, title, latest_turn_id, created_at, updated_at, latest_user_message_at, pending_approval_count, pending_user_input_count, monitor_json) VALUES (?,?,?,?,?,?,?,?,?,?)`, ) .run(threadId, "p1", `Thread ${String(index)}`, `turn-${threadId}`, at, at, at, 4, 2, "{}"); } // The newest thread is left mid-turn, as a copy taken while an agent is // working would be: a running turn with no completion, and a message still // streaming into it. The one before it is left "pending" — a user message // accepted but whose provider turn never started. const midFlight = index === threadCount - 1; const queued = index === threadCount - 2; database .prepare( `INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, completed_at, checkpoint_files_json) VALUES (?,?,?,?,?,'[]')`, ) .run( threadId, // A pending start has no turn id — that is what identifies it. queued ? null : `turn-${threadId}`, midFlight ? "running" : queued ? "pending" : "completed", at, // The running turn carries a stale completion from an earlier settle // that a later checkpoint reverted, which must not survive as the // interruption time. midFlight ? STALE_COMPLETED_AT : queued ? null : at, ); if (minimalRows) { continue; } database .prepare(`INSERT INTO projection_thread_sessions VALUES (?,'running','claude',?,'boom',?)`) .run(threadId, `turn-${threadId}`, at); database .prepare(`INSERT INTO projection_thread_messages VALUES (?,?, 'user','hello',?,?,?,?)`) .run( `m-${threadId}`, threadId, midFlight ? 1 : 0, JSON.stringify([{ type: "image", id: `${threadId}-image`, name: "shot.png" }]), at, at, ); for (let a = 0; a < activitiesPerThread; a += 1) { database .prepare( `INSERT INTO projection_thread_activities VALUES (?,?,'neutral','tool','ran',?,?,?)`, ) .run( `a-${threadId}-${String(a)}`, threadId, "{}", a, `2026-07-10T00:00:0${String(a)}.000Z`, ); } } database.exec("COMMIT"); database.close(); } function makeTarget(path: string, options?: { readonly dropProposedPlans?: boolean }) { const database = new NodeSqlite.DatabaseSync(path); // No monitor_json: the target is a migration behind, as a worktree often is. createSchema(database, { withMonitor: false }); database .prepare( `INSERT INTO projection_projects VALUES ('stale','Stale','/old','[]','${SEEDED_AT}','${SEEDED_AT}',NULL)`, ) .run(); // History from the target's own past life, which must not survive the seed. database .prepare(`INSERT INTO orchestration_events (event_id, stream_id, event_type, payload_json) VALUES ('old-event','stale-thread','thread.created','{}')`) .run(); // A binding for a thread the seed is about to delete. database .prepare(`INSERT INTO provider_session_runtime VALUES ('stale-thread','claude','claude-code','running','${SEEDED_AT}')`) .run(); // A receipt for a command whose events the seed is about to delete. database .prepare(`INSERT INTO orchestration_command_receipts VALUES ('old-command','thread','stale-thread','${SEEDED_AT}',1,'accepted',NULL)`) .run(); if (options?.dropProposedPlans) { // A target far enough behind that a whole table is missing (migration 013). database.exec("DROP TABLE projection_thread_proposed_plans"); } database.close(); } const withDatabases = <A>( run: (paths: { readonly source: string; readonly target: string }) => A, options?: { readonly threads?: number; readonly activities?: number; readonly dropProposedPlans?: boolean; readonly legacySource?: boolean; readonly minimalRows?: boolean; }, ): A => { const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-dev-seed-")); const source = NodePath.join(directory, "source.sqlite"); const target = NodePath.join(directory, "target.sqlite"); makeSource(source, options?.threads ?? 5, options?.activities ?? 3, { legacyThreads: options?.legacySource ?? false, minimalRows: options?.minimalRows ?? false, }); makeTarget(target, { dropProposedPlans: options?.dropProposedPlans ?? false }); try { return run({ source, target }); } finally { NodeFS.rmSync(directory, { recursive: true, force: true }); } }; const query = <T>(path: string, sql: string): Array<T> => { const database = new NodeSqlite.DatabaseSync(path, { readOnly: true }); try { return database.prepare(sql).all() as Array<T>; } finally { database.close(); } }; describe("seedDevDatabase", () => { it("copies the most recent threads and their project", () => { withDatabases(({ source, target }) => { const summary = seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 2, activityLimit: 10, seededAt: SEEDED_AT, }); assert.equal(summary.threads, 2); assert.equal(summary.projects, 1); const titles = query<{ title: string }>( target, "SELECT title FROM projection_threads ORDER BY latest_user_message_at DESC", ).map((row) => row.title); // Threads 4 and 3 are the newest of the five. assert.deepStrictEqual(titles, ["Thread 4", "Thread 3"]); }); }); it("replaces whatever the target held before", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 10, seededAt: SEEDED_AT, }); const projects = query<{ project_id: string }>( target, "SELECT project_id FROM projection_projects", ).map((row) => row.project_id); assert.deepStrictEqual(projects, ["p1"]); }); }); // The target can be a migration behind the source; copying its columns // blindly would fail on the first schema change. it("skips columns the target does not have", () => { withDatabases(({ source, target }) => { const summary = seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 10, seededAt: SEEDED_AT, }); assert.deepStrictEqual(summary.skippedColumns, ["projection_threads.monitor_json"]); assert.equal(summary.threads, 1); }); }); // A copied "running" session has no agent behind it, so the thread would spin // forever and the session reaper would skip it. it("neutralizes live session state", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const sessions = query<{ status: string; active_turn_id: string | null }>( target, "SELECT status, active_turn_id FROM projection_thread_sessions", ); assert.isAbove(sessions.length, 0); for (const session of sessions) { assert.equal(session.status, "stopped"); assert.isNull(session.active_turn_id); } const [threadsWithLatestTurn] = query<{ count: number }>( target, "SELECT COUNT(*) count FROM projection_threads WHERE latest_turn_id IS NOT NULL", ); assert.equal(threadsWithLatestTurn?.count, 0); }); }); // Approvals are not copied, so the badge counts must not survive. it("clears pending counts that have no rows behind them", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const [counts] = query<{ approvals: number; inputs: number }>( target, "SELECT SUM(pending_approval_count) approvals, SUM(pending_user_input_count) inputs FROM projection_threads", ); assert.equal(counts?.approvals, 0); assert.equal(counts?.inputs, 0); }); }); // A copied running turn has no agent left to finish it. The session override // alone does not cover it: the turn is read independently, and an unsettled // one keeps the thread spinning and its timeline unfoldable forever. it("settles turns copied while still running", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const turns = query<{ state: string; completed_at: string | null }>( target, "SELECT state, completed_at FROM projection_turns", ); assert.isAbove(turns.length, 0); for (const turn of turns) { assert.notEqual(turn.state, "running"); assert.isNotNull(turn.completed_at); // A settle that a later checkpoint reverted can leave a completion on a // running turn; carrying it over would date the interruption to then. assert.notEqual(turn.completed_at, STALE_COMPLETED_AT); } }); }); // The server deletes pending starts rather than settling them // (clearPendingProjectionTurnsByThread). They also cannot be settled // coherently: turn_id is NULL, so an "interrupted" one is a terminal turn // with no id — a shape no real turn has. it("drops pending turn starts instead of settling them", () => { withDatabases(({ source, target }) => { seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const [pending] = query<{ count: number }>( target, "SELECT COUNT(*) count FROM projection_turns WHERE state = 'pending' OR turn_id IS NULL", ); assert.equal(pending?.count, 0); }); }); // A burst of tool activity lands many rows on the same timestamp. Capping on // created_at alone then cuts arbitrarily among them, and can keep exactly the // rows the timeline treats as oldest. it("caps activities by the order the app displays them", () => { withDatabases(({ source, target }) => { const sourceDatabase = new NodeSqlite.DatabaseSync(source); sourceDatabase.exec("DELETE FROM projection_thread_activities"); const insert = sourceDatabase.prepare( `INSERT INTO projection_thread_activities VALUES (?, 't4', 'neutral', 'tool', 'ran', '{}', ?, ?)`, ); const tied = "2026-07-10T00:00:00.000Z"; // Timestamps tie within a burst... for (const sequence of [1, 2, 3]) { insert.run(`tie-${String(sequence)}`, sequence, tied); } // ...and can disagree with sequence outright under clock skew. The app // ranks by sequence first, so this row outranks every tied one despite // its older timestamp. insert.run("skewed-4", 4, "2026-07-09T00:00:00.000Z"); sourceDatabase.close(); seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 2, seededAt: SEEDED_AT, }); const kept = query<{ activity_id: string }>( target, "SELECT activity_id FROM projection_thread_activities ORDER BY sequence DESC", ).map((row) => row.activity_id); // The two highest sequences — not the two newest timestamps, and not // whichever two the storage order yielded. assert.deepStrictEqual(kept, ["skewed-4", "tie-3"]); }); }); it("uses activity id as the final cap tiebreaker", () => { withDatabases(({ source, target }) => { const sourceDatabase = new NodeSqlite.DatabaseSync(source); sourceDatabase.exec("DELETE FROM projection_thread_activities"); const insert = sourceDatabase.prepare( `INSERT INTO projection_thread_activities VALUES (?, 't4', 'neutral', 'tool', 'ran', '{}', 7, ?)`, ); const tied = "2026-07-10T00:00:00.000Z"; insert.run("tie-a", tied); insert.run("tie-z", tied); sourceDatabase.close(); seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 1, activityLimit: 1, seededAt: SEEDED_AT, }); const kept = query<{ activity_id: string }>( target, "SELECT activity_id FROM projection_thread_activities", ); assert.deepStrictEqual(kept, [{ activity_id: "tie-z" }]); }); }); // The delete mirrors the server's predicate, which spares rows carrying a // checkpoint. Without the `checkpoint_turn_count IS NULL` clause a checkpoint // row that happened to be pending would be destroyed along with the starts. it("spares a pending row that carries a checkpoint", () => { withDatabases(({ source, target }) => { // A checkpoint row on a thread inside the copied range. const sourceDatabase = new NodeSqlite.DatabaseSync(source); sourceDatabase .prepare( `INSERT INTO projection_turns (thread_id, turn_id, state, requested_at, checkpoint_turn_count, checkpoint_files_json) VALUES ('t4', NULL, 'pending', ?, 7, '[]')`, ) .run(SEEDED_AT); sourceDatabase.close(); seedDevDatabase({ sourceDbPath: source, targetDbPath: target, threadLimit: 3, activityLimit: 10, seededAt: SEEDED_AT, }); const checkpoints = query<{ checkpoint_turn_count: number }>( target, "SELECT checkpoint_turn_count FROM projection_turns WHERE checkpoint_turn_count IS NOT NULL", );