diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 5dc42df07674f..6a4c51a7a4278 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -3956,6 +3956,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }); const afterUpdate = yield* projectionStore.getThreadProjection(threadId); assert.equal(afterUpdate.turnItems[0]?.ordinal, ordinals[0]); + assert.equal((yield* projectionStore.getThreadShell(threadId))?.itemCount, 151); assert.equal((yield* maintenance.verify).valid, true); yield* sql` @@ -3970,6 +3971,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { const rebuilt = yield* maintenance.rebuild; assert.isTrue(rebuilt.valid); assert.lengthOf((yield* projectionStore.getThreadProjection(threadId)).turnItems, 151); + assert.equal((yield* projectionStore.getThreadShell(threadId))?.itemCount, 151); }), ); }); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 53cd77a88f852..856db5249885d 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -5359,7 +5359,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }), ); - it.effect("looks up shell stop requests through the interrupt request index", () => + it.effect("reads shell runs and item counts without walking a thread's history", () => Effect.gen(function* () { const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const sql = yield* SqlClient.SqlClient; @@ -5368,20 +5368,44 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { yield* projectionStore .getShellSnapshot({ projectId, location: "active" }) .pipe(Effect.withTracer(tracer)); + const planOf = (statement: string) => + sql + .unsafe<{ readonly detail: string }>(`EXPLAIN QUERY PLAN ${statement}`) + .pipe(Effect.map((rows) => rows.map((row) => row.detail))); const threadQuery = statements.find((statement) => statement.includes("AS latest_run_stop_requested"), ); assert.isDefined(threadQuery); - const stopRequestLookups = (yield* sql.unsafe<{ readonly detail: string }>( - `EXPLAIN QUERY PLAN ${threadQuery}`, - )) - .map((row) => row.detail) - .filter((detail) => detail.startsWith("SEARCH stop_request ")); + const plan = yield* planOf(threadQuery); + const stopRequestLookups = plan.filter((detail) => detail.startsWith("SEARCH stop_request ")); assert.lengthOf(stopRequestLookups, 2); for (const lookup of stopRequestLookups) { assert.include(lookup, "orchestration_v2_projection_turn_items_stop_request_idx"); } + // The active, activity, and pending-secret lookups; walking the ordinal index instead + // reads every run of an idle thread. + const activeRunLookups = plan.filter( + (detail) => detail.startsWith("SEARCH r ") && detail.includes("(thread_id=?"), + ); + assert.lengthOf(activeRunLookups, 4); + for (const lookup of activeRunLookups) { + assert.include(lookup, "orchestration_v2_projection_runs_thread_status_idx"); + } + const lastExecutedIndex = plan.findIndex((detail) => + detail.includes("orchestration_v2_projection_runs_last_executed_idx"), + ); + assert.notStrictEqual(lastExecutedIndex, -1); + // The lookup that follows the search is the sort it no longer needs. + assert.notInclude(plan[lastExecutedIndex + 1] ?? "", "USE TEMP B-TREE"); + + const countQueries = statements.filter((statement) => statement.includes("item_count")); + assert.isNotEmpty(countQueries); + for (const countQuery of countQueries) { + for (const detail of yield* planOf(countQuery)) { + assert.notInclude(detail, "orchestration_v2_projection_turn_items"); + } + } }), ); @@ -5636,6 +5660,7 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }, }); } + let runWrites = 0; const applyRun = ( threadId: ThreadId, suffix: string, @@ -5644,9 +5669,10 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ) => { const runId = RunId.make(`run:shell-item-counts:${suffix}`); const rootNodeId = NodeId.make(`node:shell-item-counts:${suffix}`); + runWrites += 1; return projectionStore .apply({ - id: EventId.make(`event:shell-item-counts:${suffix}:run`), + id: EventId.make(`event:shell-item-counts:${suffix}:run:${runWrites}`), type: "run.created", threadId, runId, @@ -5673,11 +5699,14 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { }) .pipe(Effect.as(runId)); }; - let ordinal = 0; - const applyItem = (threadId: ThreadId, runId: RunId | null) => { - ordinal += 1; + let items = 0; + let writes = 0; + // Passing `rewrite` writes that earlier item again, possibly under another run. + const applyItem = (threadId: ThreadId, runId: RunId | null, rewrite?: number) => { + const ordinal = rewrite ?? (items += 1); + writes += 1; return projectionStore.apply({ - id: EventId.make(`event:shell-item-counts:item:${ordinal}`), + id: EventId.make(`event:shell-item-counts:write:${writes}`), type: "turn-item.updated", threadId, ...(runId === null ? {} : { runId }), @@ -5730,6 +5759,28 @@ it.layer(layerTest)("ProjectionStoreV2", (it) => { ); assert.equal(shell?.itemCount, itemCount); } + + // Writing an item again keeps its count; moving it to another run moves the count. + const itemCountOf = (threadId: ThreadId) => + projectionStore.getThreadShell(threadId).pipe(Effect.map((shell) => shell?.itemCount)); + yield* applyItem(busyThreadId, completedRunId, 3); + assert.equal(yield* itemCountOf(busyThreadId), 4); + yield* applyItem(busyThreadId, rolledBackRunId, 3); + assert.equal(yield* itemCountOf(busyThreadId), 3); + yield* applyItem(busyThreadId, null, 3); + assert.equal(yield* itemCountOf(busyThreadId), 4); + + // A run's rollback uncounts its items, and writing the run again counts them back. + yield* applyRun(busyThreadId, "busy-completed", 1, true); + assert.equal(yield* itemCountOf(busyThreadId), 3); + yield* applyRun(busyThreadId, "busy-completed", 1, false); + assert.equal(yield* itemCountOf(busyThreadId), 4); + // Items written before their run exist count once it does. + const lateRunId = RunId.make("run:shell-item-counts:late"); + yield* applyItem(otherThreadId, lateRunId); + assert.equal(yield* itemCountOf(otherThreadId), 1); + yield* applyRun(otherThreadId, "late", 2, false); + assert.equal(yield* itemCountOf(otherThreadId), 2); }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 292aa61ecdd6f..a4487831c450c 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6727,7 +6727,10 @@ export const layer: Layer.Layer = active.status AS active_run_status, ( SELECT r.status + -- Read active runs by status. Walking the ordinal index to skip this sort + -- reads every run of a thread with none active. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6742,6 +6745,7 @@ export const layer: Layer.Layer = r.requested_at ) FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') AND NOT ${isSilentShellRun("r")} @@ -6804,6 +6808,7 @@ export const layer: Layer.Layer = SELECT secret.payload_json -- Keep runs outermost so completed history is never scanned for a pending secret. FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx CROSS JOIN orchestration_v2_projection_turn_items secret INDEXED BY orchestration_v2_projection_turn_items_thread_run_idx WHERE r.thread_id = t.thread_id @@ -6852,6 +6857,7 @@ export const layer: Layer.Layer = LEFT JOIN orchestration_v2_projection_runs active ON active.run_id = ( SELECT r.run_id FROM orchestration_v2_projection_runs r + INDEXED BY orchestration_v2_projection_runs_thread_status_idx WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running') AND NOT ${isSilentShellRun("r")} @@ -6923,27 +6929,21 @@ export const layer: Layer.Layer = const selectShellRunItemCounts = (threadIds: ReadonlyArray) => sql` - SELECT thread_id, run_id, COUNT(*) AS item_count - FROM orchestration_v2_projection_turn_items - WHERE run_id IS NOT NULL AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) - GROUP BY thread_id, run_id + SELECT thread_id, run_key AS run_id, item_count + FROM orchestration_v2_projection_turn_item_counts + WHERE run_key <> '' AND thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) `; - // Count all selected threads in one index walk; fork sources still need per-run counts. const selectShellItemCounts = (threadIds: ReadonlyArray) => sql` SELECT counts.thread_id, - SUM(CASE WHEN counts.run_id IS NULL OR r.status <> 'rolled_back' + SUM(CASE WHEN counts.run_key = '' OR r.status <> 'rolled_back' THEN counts.item_count ELSE 0 END) AS item_count, - SUM(CASE WHEN counts.run_id IS NULL THEN counts.item_count ELSE 0 END) + SUM(CASE WHEN counts.run_key = '' THEN counts.item_count ELSE 0 END) AS runless_item_count - FROM ( - SELECT thread_id, run_id, COUNT(*) AS item_count - FROM orchestration_v2_projection_turn_items - WHERE thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) - GROUP BY thread_id, run_id - ) counts - LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_id + FROM orchestration_v2_projection_turn_item_counts counts + LEFT JOIN orchestration_v2_projection_runs r ON r.run_id = counts.run_key + WHERE counts.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) GROUP BY counts.thread_id `; diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 422d2122d03bd..37b74d5c6db01 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -86,6 +86,8 @@ import Migration0080 from "./Migrations/080_ScheduledTaskThreadMode.ts"; import Migration0081 from "./Migrations/081_ProjectionMessageSearchText.ts"; import Migration0082 from "./Migrations/082_TurnItemConversationIndex.ts"; import Migration0083 from "./Migrations/083_CommandReadIndexes.ts"; +import Migration0084 from "./Migrations/084_LastExecutedRunIndex.ts"; +import Migration0085 from "./Migrations/085_TurnItemCounts.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -172,6 +174,8 @@ const migrationEntries = [ [81, "ProjectionMessageSearchText", Migration0081], [82, "TurnItemConversationIndex", Migration0082], [83, "CommandReadIndexes", Migration0083], + [84, "LastExecutedRunIndex", Migration0084], + [85, "TurnItemCounts", Migration0085], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts b/apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts new file mode 100644 index 0000000000000..2c36d62634873 --- /dev/null +++ b/apps/server/src/persistence/Migrations/084_LastExecutedRunIndex.ts @@ -0,0 +1,18 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // The shell's last executed run lookup sorts every run of the thread. This index carries + // that order and the same predicate the lookup filters on. A database that already has + // the index keeps it, so a hand-applied one does not fail the migration. + yield* sql` + CREATE INDEX IF NOT EXISTS orchestration_v2_projection_runs_last_executed_idx + ON orchestration_v2_projection_runs( + thread_id, (completed_at IS NULL) DESC, completed_at DESC, ordinal DESC, run_id DESC + ) + WHERE status <> 'queued' + AND NOT (status = 'cancelled' AND json_extract(payload_json, '$.startedAt') IS NULL) + `; +}); diff --git a/apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts b/apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts new file mode 100644 index 0000000000000..cb73f5a8fd8d0 --- /dev/null +++ b/apps/server/src/persistence/Migrations/085_TurnItemCounts.test.ts @@ -0,0 +1,60 @@ +import { assert, it } from "@effect/vitest"; +import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; + +const layer = it.layer(NodeSqliteClient.layer({ filename: ":memory:" })); + +layer("085_TurnItemCounts", (it) => { + it.effect("backfills item counts, then follows upserts, moves, and deletes", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 84 }); + const upsert = (id: string, threadId: string, runId: string | null) => sql` + INSERT INTO orchestration_v2_projection_turn_items ( + turn_item_id, thread_id, run_id, ordinal, type, status, updated_at, payload_json + ) + VALUES ( + ${id}, ${threadId}, ${runId}, 1, 'assistant_message', 'completed', + '2026-10-29T00:00:00.000Z', '{}' + ) + ON CONFLICT(turn_item_id) DO UPDATE SET + thread_id = excluded.thread_id, + run_id = excluded.run_id, + status = excluded.status + `; + yield* upsert("runless-1", "thread-1", null); + yield* upsert("runless-2", "thread-1", null); + yield* upsert("run-1", "thread-1", "run-1"); + + yield* runMigrations(); + + const counts = () => + sql<{ readonly thread_id: string; readonly run_key: string; readonly item_count: number }>` + SELECT thread_id, run_key, item_count + FROM orchestration_v2_projection_turn_item_counts + ORDER BY thread_id, run_key + `; + assert.deepEqual(yield* counts(), [ + { thread_id: "thread-1", run_key: "", item_count: 2 }, + { thread_id: "thread-1", run_key: "run-1", item_count: 1 }, + ]); + + yield* upsert("run-1", "thread-1", "run-1"); + yield* upsert("run-2", "thread-2", "run-2"); + yield* upsert("runless-2", "thread-1", "run-1"); + assert.deepEqual(yield* counts(), [ + { thread_id: "thread-1", run_key: "", item_count: 1 }, + { thread_id: "thread-1", run_key: "run-1", item_count: 2 }, + { thread_id: "thread-2", run_key: "run-2", item_count: 1 }, + ]); + + yield* sql`DELETE FROM orchestration_v2_projection_turn_items WHERE thread_id = 'thread-1'`; + assert.deepEqual(yield* counts(), [ + { thread_id: "thread-2", run_key: "run-2", item_count: 1 }, + ]); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/085_TurnItemCounts.ts b/apps/server/src/persistence/Migrations/085_TurnItemCounts.ts new file mode 100644 index 0000000000000..a432046e6361a --- /dev/null +++ b/apps/server/src/persistence/Migrations/085_TurnItemCounts.ts @@ -0,0 +1,60 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +/** + * Shell reads count items per (thread, run) here instead of walking every turn item. Triggers + * keep it exact for every writer, including projection rebuilds and dev-database pruning. + */ +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // run_key is '' for items without a run; NULLs never conflict in a primary key. + yield* sql` + CREATE TABLE orchestration_v2_projection_turn_item_counts ( + thread_id TEXT NOT NULL, + run_key TEXT NOT NULL, + item_count INTEGER NOT NULL, + PRIMARY KEY (thread_id, run_key) + ) WITHOUT ROWID + `; + yield* sql` + INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count) + SELECT thread_id, COALESCE(run_id, ''), COUNT(*) + FROM orchestration_v2_projection_turn_items + GROUP BY thread_id, run_id + `; + // An upsert that hits an existing item fires only the update trigger. + yield* sql` + CREATE TRIGGER orchestration_v2_turn_item_counts_insert + AFTER INSERT ON orchestration_v2_projection_turn_items + BEGIN + INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count) + VALUES (new.thread_id, COALESCE(new.run_id, ''), 1) + ON CONFLICT (thread_id, run_key) DO UPDATE SET item_count = item_count + 1; + END + `; + yield* sql` + CREATE TRIGGER orchestration_v2_turn_item_counts_update + AFTER UPDATE OF thread_id, run_id ON orchestration_v2_projection_turn_items + WHEN old.thread_id IS NOT new.thread_id OR old.run_id IS NOT new.run_id + BEGIN + UPDATE orchestration_v2_projection_turn_item_counts SET item_count = item_count - 1 + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, ''); + DELETE FROM orchestration_v2_projection_turn_item_counts + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '') AND item_count <= 0; + INSERT INTO orchestration_v2_projection_turn_item_counts (thread_id, run_key, item_count) + VALUES (new.thread_id, COALESCE(new.run_id, ''), 1) + ON CONFLICT (thread_id, run_key) DO UPDATE SET item_count = item_count + 1; + END + `; + yield* sql` + CREATE TRIGGER orchestration_v2_turn_item_counts_delete + AFTER DELETE ON orchestration_v2_projection_turn_items + BEGIN + UPDATE orchestration_v2_projection_turn_item_counts SET item_count = item_count - 1 + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, ''); + DELETE FROM orchestration_v2_projection_turn_item_counts + WHERE thread_id = old.thread_id AND run_key = COALESCE(old.run_id, '') AND item_count <= 0; + END + `; +}); diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 05f09f8e16850..3c2cb075cfab4 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -63,6 +63,8 @@ describe("V2 preview upgrade", () => { [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], [83, "CommandReadIndexes"], + [84, "LastExecutedRunIndex"], + [85, "TurnItemCounts"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -169,6 +171,8 @@ describe("V2 preview upgrade", () => { [81, "ProjectionMessageSearchText"], [82, "TurnItemConversationIndex"], [83, "CommandReadIndexes"], + [84, "LastExecutedRunIndex"], + [85, "TurnItemCounts"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/docs/internals/performance-regressions.md b/docs/internals/performance-regressions.md index 24edebc5823db..a7d5be74fd175 100644 --- a/docs/internals/performance-regressions.md +++ b/docs/internals/performance-regressions.md @@ -111,9 +111,9 @@ sessions. The orchestrator still validates the current thread before applying se Read receipts use `ProjectionStore.getThread` to load only canonical thread metadata. A visit advances `lastVisitedAt` monotonically without changing `updatedAt` or decoding history. Shell -queries use a covering thread/run index for item counts and a partial index for nonterminal -background items, including idle items. Latest user-message reads use an index ordered by update -time. +queries read item counts from a per-thread, per-run table that turn-item triggers keep exact, and +use a partial index for nonterminal background items, including idle items. Latest user-message +reads use an index ordered by update time. Command palette results keep live VCS and linked pull-request queries leased to the visible viewport, including a small overscan region. Cached badges remain available while leases are