perf(server): events stop maintaining the stream sequence index

apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts

@@ -1,3 +1,5 @@
+import * as NodeSqlite from "node:sqlite";
+
 import { assert, it } from "@effect/vitest";
 import {
   EventId,
@@ -29,6 +31,13 @@ import * as ProjectionMaintenance from "./ProjectionMaintenance.ts";
 import * as ProjectionStore from "./ProjectionStore.ts";
 import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts";
 
+declare module "node:sqlite" {
+  interface StatementSync {
+    // Node 26 has this; @types/node 24 does not declare it yet.
+    stat(counter: "vmStep"): number;
+  }
+}
+
 const database = SqlitePersistence.layerMemory;
 const eventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(database));
 const projectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(database));
@@ -383,3 +392,113 @@ it.effect("keeps an anchor that a row appended during the pass borrows from", ()
     }),
   ).pipe(Effect.provide(layerTest)),
 );
+
+it.effect(
+  "finds each stream's newest event in steps that do not grow with retained V1 history",
+  () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      const sink = yield* EventSink.EventSinkV2;
+      const events = yield* OrchestrationEventStore.OrchestrationEventStore;
+      const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
+      const now = yield* DateTime.now;
+      const occurredAt = DateTime.formatIso(now);
+      const projectId = ProjectId.make("project:prune-legacy-history");
+      const threadId = ThreadId.make("thread:prune-legacy-history");
+      // Pruning keeps V1 rows, so a stream written before V2 keeps its whole V1 history.
+      const v1History = 1000;
+      for (const [kind, streamId] of [
+        ["project", projectId],
+        ["thread", threadId],
+      ] as const) {
+        yield* sql`
+        WITH RECURSIVE version(n) AS (SELECT 0 UNION ALL SELECT n + 1 FROM version WHERE n + 1 < ${v1History})
+        INSERT INTO orchestration_events (
+          event_id, aggregate_kind, stream_id, stream_version, event_type,
+          occurred_at, actor_kind, payload_json, metadata_json
+        )
+        SELECT 'event:v1:' || ${kind} || ':' || n, ${kind}, ${streamId}, n, 'legacy.updated',
+          ${occurredAt}, 'server', '{}', '{}'
+        FROM version
+      `;
+      }
+      const [v1Head] = yield* sql<{
+        readonly sequence: number;
+      }>`SELECT MAX(sequence) AS sequence FROM orchestration_events`;
+      const v1Through = v1Head!.sequence;
+      const updates = 25;
+      for (let index = 0; index < updates; index++) {
+        yield* events.appendProjectEvent({
+          type: "project.meta-updated",
+          eventId: EventId.make(`event:prune-legacy-history:project:${index}`),
+          aggregateKind: "project",
+          aggregateId: projectId,
+          occurredAt,
+          commandId: null,
+          causationEventId: null,
+          correlationId: null,
+          metadata: {},
+          payload: { projectId, title: `Project ${index}`, updatedAt: occurredAt },
+        });
+      }
+      yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] });
+      for (let revision = 0; revision < updates - 1; revision++) {
+        yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], revision)] });
+      }
+      yield* maintenance.pruneEventStore();
+      yield* TestClock.adjust("1 day");
+
+      // One copy per planner statistics mode; the server deletes sqlite_stat1 at startup.
+      using none = new NodeSqlite.DatabaseSync("file:/prune-head-steps-none?vfs=memdb");
+      using analyzed = new NodeSqlite.DatabaseSync("file:/prune-head-steps-analyze?vfs=memdb");
+      using optimized = new NodeSqlite.DatabaseSync("file:/prune-head-steps-optimize?vfs=memdb");
+      const copies = [
+        ["none", none, ""],
+        ["analyze", analyzed, "ANALYZE"],
+        ["optimize", optimized, "PRAGMA optimize = 0x10002"],
+      ] as const;
+      for (const [mode] of copies)
+        yield* sql`VACUUM INTO ${`file:/prune-head-steps-${mode}?vfs=memdb`}`;
+      const pages: Array<ReturnType<Statement.Statement<unknown>["compile"]>> = [];
+      const observe: Statement.Transformer = (statement) => {
+        const compiled = statement.compile();
+        if (compiled[0].includes("END AS prunable")) pages.push(compiled);
+        return Effect.succeed(statement);
+      };
+      // Each stream keeps its newest row, and the thread also keeps thread.created.
+      assert.equal(
+        (yield* maintenance
+          .pruneEventStore()
+          .pipe(Effect.provideService(Statement.CurrentTransformer, observe))).deletedEventCount,
+        2 * updates - 3,
+      );
+      const [query, [, throughSequence, pageSize]] = pages[0]!;
+
+      // VM steps are deterministic; looking up the head by MAX(sequence) through the stream_version
+      // index read the whole stream for each V2 row: 7.5 times the steps of 100 V1 rows here.
+      for (const [mode, copy, analyze] of copies) {
+        if (analyze !== "") copy.exec(analyze);
+        const page = () => {
+          const statement = copy.prepare(query);
+          const rows = statement.all(
+            v1Through,
+            throughSequence as NodeSqlite.SQLInputValue,
+            pageSize as NodeSqlite.SQLInputValue,
+          );
+          return { rows, steps: statement.stat("vmStep") };
+        };
+        const full = page();
+        assert.lengthOf(full.rows, 2 * updates);
+        assert.deepEqual(
+          full.rows.flatMap((row, position) => (row.prunable === 0 ? [position] : [])),
+          [updates - 1, updates, 2 * updates - 1],
+        );
+        copy.exec(
+          `DELETE FROM orchestration_events WHERE stream_version BETWEEN 100 AND ${v1History - 1}`,
+        );
+        const trimmed = page();
+        assert.deepEqual(trimmed.rows, full.rows);
+        assert.isAtMost(full.steps, trimmed.steps * 1.1, mode);
+      }
+    }).pipe(Effect.provide(layerTest)),
+);

apps/server/src/orchestration-v2/ProjectionMaintenance.ts

@@ -129,8 +129,10 @@ export const layer: Layer.Layer<ProjectionMaintenanceV2, never, SqlClient.SqlCli
               WHEN event.event_type = 'run.updated'
                 AND json_extract(event.payload_json, '$.status') = 'running' THEN 0
               -- The newest row keeps each stream's high water, stream_version, and snapshot sequence.
-              WHEN event.sequence = (
-                SELECT MAX(newer.sequence) FROM orchestration_events AS newer
+              -- Appends take the stream's highest stream_version plus one, so it marks the newest
+              -- row, and the stream_version index seeks it where MAX(sequence) reads the stream.
+              WHEN event.stream_version = (
+                SELECT MAX(newer.stream_version) FROM orchestration_events AS newer
                 WHERE newer.aggregate_kind = event.aggregate_kind
                   AND newer.stream_id = event.stream_id
               ) THEN 0

apps/server/src/orchestration-v2/ThreadFind.ts

@@ -281,7 +281,7 @@ export const makeThreadFind = Effect.fn("makeThreadFind")(function* (
       SELECT COALESCE(MAX(sequence), 0) AS sequence FROM orchestration_events
       -- The planner otherwise walks every V2 event backwards from the newest,
       -- which takes seconds for a thread that has been idle a long time.
-      INDEXED BY idx_orch_events_stream_sequence
+      INDEXED BY idx_orchestration_events_agent_stream_sequence
       WHERE application_event_version = 2 AND aggregate_kind = 'thread'
         AND stream_id IN (SELECT thread_id FROM source_threads)
     `;

apps/server/src/persistence/Migrations.test.ts

@@ -101,9 +101,9 @@ describe("runMigrations", () => {
       `;
       const indexNames = indexes.map((row) => row.name);
       assert.notInclude(indexNames, "idx_orch_events_correlation_id");
+      assert.notInclude(indexNames, "idx_orch_events_stream_sequence");
       for (const kept of [
         "idx_orch_events_stream_version",
-        "idx_orch_events_stream_sequence",
         "idx_orch_events_command_id",
         "idx_orchestration_events_agent_stream_sequence",
       ]) {

apps/server/src/persistence/Migrations.ts

@@ -99,6 +99,7 @@ import Migration0088 from "./Migrations/088_TurnItemTypeStatusIndex.ts";
 import Migration0089 from "./Migrations/089_ShellThreadReadIndexes.ts";
 import Migration0090 from "./Migrations/090_AuthConnectDevices.ts";
 import Migration0091 from "./Migrations/091_AuthConnectDeniedRequests.ts";
+import Migration0092 from "./Migrations/092_DropOrchestrationEventStreamSequenceIndex.ts";
 
 const migrationEntries = [
   [1, "OrchestrationEvents", Migration0001],
@@ -193,6 +194,7 @@ const migrationEntries = [
   [89, "ShellThreadReadIndexes", Migration0089],
   [90, "AuthConnectDevices", Migration0090],
   [91, "AuthConnectDeniedRequests", Migration0091],
+  [92, "DropOrchestrationEventStreamSequenceIndex", Migration0092],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);

apps/server/src/persistence/Migrations/092_DropOrchestrationEventStreamSequenceIndex.ts

@@ -0,0 +1,10 @@
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/sql/SqlClient";
+
+export default Effect.gen(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  // idx_orchestration_events_agent_stream_sequence serves its V2 thread reads and
+  // idx_orch_events_stream_version its whole-stream reads as well, so it only adds write cost.
+  yield* sql`DROP INDEX IF EXISTS idx_orch_events_stream_sequence`;
+});

apps/server/src/persistence/OrchestrationEventStore.sequence.test.ts

@@ -7,6 +7,7 @@ import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 import * as SqlClient from "effect/sql/SqlClient";
+import * as Statement from "effect/sql/Statement";
 
 import * as OrchestrationEventStore from "./OrchestrationEventStore.ts";
 import * as SqlitePersistence from "./Sqlite.ts";
@@ -205,3 +206,49 @@ it.effect(
       assert.isAbove(liveSequence, highWater);
     }).pipe(Effect.provide(Layer.fresh(layerEventStore))),
 );
+
+it.effect("reads one thread's events through the thread index, not the V2 sequence range", () =>
+  Effect.gen(function* () {
+    const store = yield* OrchestrationEventStore.OrchestrationEventStore;
+    const sql = yield* SqlClient.SqlClient;
+    const rows = yield* seedEvents(
+      Arr.makeBy(40, (index) => eventRow(index, "thread", 2, index % 4 === 0 ? "target" : "other")),
+    );
+    const target = rows.filter((row) => row.stream_id === "target").map((row) => row.sequence);
+    const threadId = ThreadId.make("target");
+    const statements: Array<ReturnType<Statement.Statement<unknown>["compile"]>> = [];
+    const observe: Statement.Transformer = (statement) => {
+      statements.push(statement.compile());
+      return Effect.succeed(statement);
+    };
+    const byThread = yield* store
+      .readAgentEvents({ threadId })
+      .pipe(Stream.runCollect, Effect.provideService(Statement.CurrentTransformer, observe));
+    const byThreadAndType = yield* store
+      .readAgentEvents({ threadId, eventType: "provider-session.detached" })
+      .pipe(Stream.runCollect, Effect.provideService(Statement.CurrentTransformer, observe));
+    const stats = yield* store
+      .getAgentReplayStats({ threadId, afterSequence: 0, throughSequence: 1_000, maxEvents: 128 })
+      .pipe(Effect.provideService(Statement.CurrentTransformer, observe));
+    assert.deepEqual(
+      byThread.map((event) => event.sequence),
+      target,
+    );
+    assert.deepEqual(
+      byThreadAndType.map((event) => event.sequence),
+      target,
+    );
+    assert.equal(stats.eventCount, target.length);
+    assert.equal(statements.length, 3);
+    for (const [query, params] of statements) {
+      const plan = yield* sql.unsafe<{ readonly detail: string }>(
+        `EXPLAIN QUERY PLAN ${query}`,
+        params,
+      );
+      assert.include(
+        plan.map((row) => row.detail),
+        "SEARCH orchestration_events USING INDEX idx_orchestration_events_agent_stream_sequence (stream_id=? AND sequence>? AND sequence<?)",
+      );
+    }
+  }).pipe(Effect.provide(layerEventStore)),
+);

apps/server/src/persistence/OrchestrationEventStore.ts

@@ -416,12 +416,14 @@ const makeEventStore = Effect.gen(function* () {
         correlation_id
       FROM orchestration_events
       ${
-        // Otherwise the sequence index wins and command lookups scan every V2 event.
+        // Otherwise the sequence index wins and command or thread lookups scan every V2 event.
         input.commandId !== undefined
           ? sql`INDEXED BY idx_orch_events_command_id`
-          : input.onlyAgentEvents === true
-            ? sql``
-            : sql`INDEXED BY idx_orchestration_events_application_high_water`
+          : input.onlyAgentEvents !== true
+            ? sql`INDEXED BY idx_orchestration_events_application_high_water`
+            : input.threadId !== undefined
+              ? sql`INDEXED BY idx_orchestration_events_agent_stream_sequence`
+              : sql``
       }
       WHERE sequence > ${input.afterSequence}
         AND sequence <= ${input.throughSequence ?? Number.MAX_SAFE_INTEGER}
@@ -604,7 +606,8 @@ const makeEventStore = Effect.gen(function* () {
         COALESCE(MAX(event_type = 'thread.created'), 0) AS "hasCreateEvent"
       FROM (
         SELECT ${restoredPayloadJson} AS payload_json, event_type
-        FROM orchestration_events
+        -- The planner otherwise ranges over every V2 event after the cursor.
+        FROM orchestration_events INDEXED BY idx_orchestration_events_agent_stream_sequence
         WHERE aggregate_kind = 'thread'
           AND stream_id = ${input.threadId}
           AND application_event_version = 2

apps/server/src/persistence/reconcileV2PreviewMigration.test.ts

@@ -71,6 +71,7 @@ describe("V2 preview upgrade", () => {
         [89, "ShellThreadReadIndexes"],
         [90, "AuthConnectDevices"],
         [91, "AuthConnectDeniedRequests"],
+        [92, "DropOrchestrationEventStreamSequenceIndex"],
       ]);
       assert.deepStrictEqual(yield* runMigrations(), []);
       assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports);
@@ -185,6 +186,7 @@ describe("V2 preview upgrade", () => {
         [89, "ShellThreadReadIndexes"],
         [90, "AuthConnectDevices"],
         [91, "AuthConnectDeniedRequests"],
+        [92, "DropOrchestrationEventStreamSequenceIndex"],
       ]);
     }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))),
   );

apps/server/src/usage/CodexUsageHistory.ts

@@ -62,7 +62,9 @@ const make = Effect.gen(function* () {
         json_extract(turn.payload_json, '$.nativeTurnRef.nativeId') AS native_turn_id,
         (
           SELECT json_extract(event.payload_json, '$.modelSelection')
+          -- The planner otherwise ranges over every V2 event for each turn.
           FROM orchestration_events AS event
+            INDEXED BY idx_orchestration_events_agent_stream_sequence
           WHERE event.aggregate_kind = 'thread'
             AND event.application_event_version = 2
             AND event.stream_id = attempt.thread_id