fix(server): deleted threads release their stored history

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

@@ -6,7 +6,9 @@ import {
   type OrchestrationV2AppThread,
   type OrchestrationV2DomainEvent,
   type OrchestrationV2TurnItem,
+  MessageId,
   ProjectId,
+  RunId,
   ProviderDriverKind,
   ProviderInstanceId,
   ThreadId,
@@ -94,8 +96,8 @@ function itemEvent(
   spec: (typeof itemSpecs)[number],
   revision: number,
   eventId = `event:turn-item-compaction:${spec.key}:${revision}`,
+  id = TurnItemId.make(`turn-item:compaction:${spec.key}`),
 ): OrchestrationV2DomainEvent {
-  const id = TurnItemId.make(`turn-item:compaction:${spec.key}`);
   const growing = `${spec.key}:${revision}:` + "x".repeat((revision + 1) * 192);
   const completed = revision === 69;
   const base = {
@@ -383,3 +385,198 @@ it.effect("keeps an anchor that a row appended during the pass borrows from", ()
     }),
   ).pipe(Effect.provide(layerTest)),
 );
+
+const countContent = (threadId: ThreadId) =>
+  Effect.flatMap(
+    SqlClient.SqlClient,
+    (sql) => sql<{
+      readonly items: number;
+      readonly positions: number;
+      readonly messages: number;
+      readonly found: number;
+      readonly threads: number;
+    }>`
+      SELECT
+        (SELECT COUNT(*) FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId}) AS items,
+        (SELECT COUNT(*) FROM orchestration_v2_turn_item_positions WHERE thread_id = ${threadId}) AS positions,
+        (SELECT COUNT(*) FROM orchestration_v2_projection_messages WHERE thread_id = ${threadId}) AS messages,
+        (SELECT COUNT(*) FROM orchestration_v2_message_search WHERE orchestration_v2_message_search MATCH ${`purgeable${threadId.replace(/\W/g, "")}`}) AS found,
+        (SELECT COUNT(*) FROM orchestration_v2_projection_threads WHERE thread_id = ${threadId}) AS threads
+    `,
+  ).pipe(Effect.map((rows) => rows[0]));
+
+const purgeFixture = Effect.gen(function* () {
+  const sink = yield* EventSink.EventSinkV2;
+  const projections = yield* ProjectionStore.ProjectionStoreV2;
+  const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2;
+  const now = yield* DateTime.now;
+  const seed = Effect.fnUntraced(function* (
+    name: string,
+    forkedFrom: OrchestrationV2AppThread["forkedFrom"] = null,
+  ) {
+    const threadId = ThreadId.make(`thread:purge-${name}`);
+    const thread = { ...makeThread(threadId, now), forkedFrom };
+    yield* sink.write({
+      events: [{ ...threadCreated(thread, now), id: EventId.make(`event:purge:${name}:thread`) }],
+    });
+    yield* sink.write({
+      events: [
+        itemEvent(
+          threadId,
+          now,
+          itemSpecs[0],
+          0,
+          `event:purge:${name}:item`,
+          TurnItemId.make(`turn-item:purge:${name}`),
+        ),
+      ],
+    });
+    yield* sink.write({
+      events: [
+        {
+          id: EventId.make(`event:purge:${name}:message`),
+          type: "message.updated",
+          threadId,
+          occurredAt: now,
+          payload: {
+            id: MessageId.make(`message:purge:${name}`),
+            threadId,
+            runId: null,
+            nodeId: null,
+            createdBy: "agent",
+            creationSource: "provider",
+            role: "assistant",
+            text: `purgeable${threadId.replace(/\W/g, "")}`,
+            streaming: false,
+            attachments: [],
+            createdAt: now,
+            updatedAt: now,
+          },
+        },
+      ],
+    });
+    return thread;
+  });
+  const remove = (thread: OrchestrationV2AppThread, at: DateTime.Utc) =>
+    sink.write({
+      events: [
+        {
+          id: EventId.make(`event:purge:${thread.id}:deleted`),
+          type: "thread.deleted",
+          threadId: thread.id,
+          occurredAt: at,
+          payload: { ...thread, deletedAt: at, updatedAt: at },
+        },
+      ],
+    });
+  return { projections, maintenance, now, seed, remove };
+});
+
+it.effect("purges content of threads deleted beyond the retained history", () =>
+  Effect.gen(function* () {
+    const { projections, maintenance, now, seed, remove } = yield* purgeFixture;
+    const old = yield* seed("old");
+    const live = yield* seed("live");
+    yield* remove(old, now);
+    yield* TestClock.adjust("1 day");
+    const recent = yield* seed("recent");
+    const recentDeletedAt = yield* DateTime.now;
+    yield* remove(recent, recentDeletedAt);
+    const liveBefore = yield* projections.getThreadProjection(live.id);
+    const beforeOld = yield* countContent(old.id);
+    assert.isAbove(beforeOld?.items ?? 0, 0);
+    assert.isAbove(beforeOld?.positions ?? 0, 0);
+    assert.equal(beforeOld?.found, 1);
+
+    const first = yield* maintenance.pruneEventStore();
+    assert.isAbove(first.purgedProjectionRowCount, 0);
+    const afterOld = yield* countContent(old.id);
+    assert.deepEqual(afterOld, { items: 0, positions: 0, messages: 0, found: 0, threads: 1 });
+    assert.equal((yield* countContent(recent.id))?.items, 1);
+    assert.equal((yield* countContent(recent.id))?.found, 1);
+    assert.deepEqual(yield* projections.getThreadProjection(live.id), liveBefore);
+
+    assert.equal((yield* maintenance.pruneEventStore()).purgedProjectionRowCount, 0);
+    yield* TestClock.adjust("1 day");
+    assert.isAbove((yield* maintenance.pruneEventStore()).purgedProjectionRowCount, 0);
+    assert.equal((yield* countContent(recent.id))?.items, 0);
+    assert.equal((yield* countContent(live.id))?.items, 1);
+    assert.equal((yield* countContent(live.id))?.found, 1);
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("lists every thread-scoped projection table or names why it is kept", () =>
+  Effect.gen(function* () {
+    const sql = yield* SqlClient.SqlClient;
+    const scoped = yield* sql<{ readonly name: string }>`
+      SELECT m.name FROM sqlite_master AS m, pragma_table_info(m.name) AS c
+      WHERE m.type = 'table' AND c.name = 'thread_id'
+        AND (m.name LIKE 'orchestration_v2_projection_%' OR m.name LIKE 'orchestration_v2_turn_item_%')
+    `;
+    const kept = [
+      "orchestration_v2_projection_threads",
+      "orchestration_v2_projection_provider_sessions",
+      "orchestration_v2_projection_provider_session_bindings",
+    ];
+    assert.deepEqual(
+      scoped
+        .map((row) => row.name)
+        .filter((name) => !kept.includes(name))
+        .toSorted(),
+      [...ProjectionMaintenance.THREAD_CONTENT_TABLES].toSorted(),
+    );
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("keeps a deleted thread that a live run-fork still inherits history from", () =>
+  Effect.gen(function* () {
+    const { maintenance, seed, remove } = yield* purgeFixture;
+    const root = yield* seed("root");
+    const middle = yield* seed("middle", {
+      type: "run",
+      threadId: root.id,
+      runId: RunId.make("run:purge:root"),
+    });
+    const fork = yield* seed("fork", {
+      type: "run",
+      threadId: middle.id,
+      runId: RunId.make("run:purge:middle"),
+    });
+    const unrelated = yield* seed("unrelated");
+    yield* remove(root, yield* DateTime.now);
+    yield* remove(middle, yield* DateTime.now);
+    yield* remove(unrelated, yield* DateTime.now);
+    yield* TestClock.adjust("2 days");
+
+    yield* maintenance.pruneEventStore();
+    assert.equal((yield* countContent(root.id))?.items, 1);
+    assert.equal((yield* countContent(middle.id))?.items, 1);
+    assert.equal((yield* countContent(fork.id))?.items, 1);
+    assert.equal((yield* countContent(unrelated.id))?.items, 0);
+  }).pipe(Effect.provide(layerTest)),
+);
+
+it.effect("purges a very large thread in bounded batches", () =>
+  Effect.gen(function* () {
+    const { maintenance, seed, remove } = yield* purgeFixture;
+    const sql = yield* SqlClient.SqlClient;
+    const big = yield* seed("big");
+    yield* sql`
+      INSERT INTO orchestration_v2_turn_item_positions (thread_id, turn_item_id, ordinal)
+      WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < 1200)
+      SELECT ${big.id}, 'turn-item:purge:filler:' || i, i + 100 FROM n
+    `;
+    yield* remove(big, yield* DateTime.now);
+    yield* TestClock.adjust("2 days");
+
+    let batches = 0;
+    const result = yield* maintenance.pruneEventStore(
+      Effect.sync(() => {
+        batches += 1;
+      }),
+    );
+    assert.isAtLeast(result.purgedProjectionRowCount, 1200);
+    assert.isAtLeast(batches, 3);
+    assert.equal((yield* countContent(big.id))?.positions, 0);
+  }).pipe(Effect.provide(layerTest)),
+);

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

@@ -5,6 +5,7 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Schema from "effect/Schema";
+import type * as SqlError from "effect/sql/SqlError";
 import * as SqlClient from "effect/sql/SqlClient";
 
 import {
@@ -22,11 +23,15 @@ export class ProjectionMaintenanceError extends Schema.TaggedError<ProjectionMai
 ) {}
 
 export interface ProjectionMaintenanceV2Shape {
-  /** Deletes events the projections already cover once they are older than the retained history. */
+  /**
+   * Deletes events the projections already cover once they are older than the retained history,
+   * and the projection content of threads deleted for longer than that.
+   */
   readonly pruneEventStore: (betweenPages?: Effect.Effect<void>) => Effect.Effect<
     {
       readonly deletedEventCount: number;
       readonly deletedReceiptCount: number;
+      readonly purgedProjectionRowCount: number;
       readonly reclaimableBytes: number;
     },
     ProjectionMaintenanceError
@@ -42,8 +47,30 @@ export class ProjectionMaintenanceV2 extends Context.Service<
 const RETAINED_EVENT_HISTORY = Duration.days(1);
 // Small pages bound work on the reader worker and synchronous fallback.
 const PAGE_SIZE = 50;
+// Bounds the rows one purge statement deletes, FTS trigger work included.
+const ROW_BATCH_SIZE = 500;
 const REFERENCE_SCAN_WINDOW = 1000;
 
+// Thread-scoped projection tables, children first. The thread row stays as the tombstone.
+// provider_sessions is handled separately because a session can serve several threads, and
+// context_transfers is skipped because a live thread can be its other endpoint.
+export const THREAD_CONTENT_TABLES = [
+  "orchestration_v2_turn_item_positions",
+  "orchestration_v2_projection_turn_items",
+  "orchestration_v2_projection_nodes",
+  "orchestration_v2_projection_messages",
+  "orchestration_v2_projection_checkpoints",
+  "orchestration_v2_projection_checkpoint_scopes",
+  "orchestration_v2_projection_run_attempts",
+  "orchestration_v2_projection_runs",
+  "orchestration_v2_projection_provider_turns",
+  "orchestration_v2_projection_provider_threads",
+  "orchestration_v2_projection_subagents",
+  "orchestration_v2_projection_context_handoffs",
+  "orchestration_v2_projection_runtime_requests",
+  "orchestration_v2_projection_plans",
+] as const;
+
 type MetadataRow = {
   readonly projection_name: string;
   readonly last_sequence: number;
@@ -252,6 +279,97 @@ export const layer: Layer.Layer<ProjectionMaintenanceV2, never, SqlClient.SqlCli
       return { deletedEventCount, deletedReceiptCount };
     });
 
+    // A deleted thread keeps only its tombstone row once the event history it could be replayed
+    // from has been pruned. A thread that a surviving run-fork inherits history from is kept
+    // whole. Idempotent: threads without content match nothing.
+    const purgeDeletedThreadContent = Effect.fnUntraced(function* (
+      now: DateTime.Utc,
+      betweenPages: Effect.Effect<void>,
+    ) {
+      const cutoff = DateTime.formatIso(DateTime.subtractDuration(now, RETAINED_EVENT_HISTORY));
+      let purged = 0;
+      // Each batch is one statement that commits on its own, so no transaction spans a yield.
+      const deleteBatches = Effect.fnUntraced(function* (
+        batch: (limit: number) => Effect.Effect<unknown, SqlError.SqlError>,
+      ) {
+        for (;;) {
+          yield* batch(ROW_BATCH_SIZE);
+          const changes = yield* sql<{ readonly count: number }>`SELECT changes() AS count`;
+          const count = changes[0]?.count ?? 0;
+          purged += count;
+          yield* Effect.yieldNow;
+          yield* betweenPages;
+          if (count < ROW_BATCH_SIZE) return;
+        }
+      });
+      for (let cursor = ""; ;) {
+        const threads = yield* sql<{ readonly thread_id: string }>`
+          WITH RECURSIVE inherited(thread_id) AS (
+            SELECT json_extract(fork.payload_json, '$.forkedFrom.threadId')
+            FROM orchestration_v2_projection_threads AS fork
+            WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run'
+              AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1
+              AND (fork.deleted_at IS NULL OR fork.deleted_at > ${cutoff})
+            UNION
+            SELECT json_extract(fork.payload_json, '$.forkedFrom.threadId')
+            FROM inherited
+            JOIN orchestration_v2_projection_threads AS fork ON fork.thread_id = inherited.thread_id
+            WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run'
+              AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1
+          )
+          SELECT thread_id FROM orchestration_v2_projection_threads
+          WHERE deleted_at IS NOT NULL AND deleted_at <= ${cutoff} AND thread_id > ${cursor}
+            AND thread_id NOT IN (SELECT thread_id FROM inherited WHERE thread_id IS NOT NULL)
+          ORDER BY thread_id ASC
+          LIMIT ${PAGE_SIZE}
+        `.pipe(NodeSqliteClient.readOnly);
+        const threadIds = threads.map((row) => row.thread_id);
+        yield* Effect.yieldNow;
+        if (threadIds.length > 0) {
+          for (const table of THREAD_CONTENT_TABLES) {
+            yield* deleteBatches(
+              (limit) => sql`
+                DELETE FROM ${sql(table)}
+                WHERE rowid IN (
+                  SELECT rowid FROM ${sql(table)}
+                  WHERE thread_id IN ${sql.in(threadIds)}
+                  LIMIT ${limit}
+                )
+              `,
+            );
+          }
+          yield* deleteBatches(
+            (limit) => sql`
+              DELETE FROM orchestration_v2_projection_provider_sessions
+              WHERE rowid IN (
+                SELECT session.rowid FROM orchestration_v2_projection_provider_sessions AS session
+                WHERE session.thread_id IN ${sql.in(threadIds)}
+                  AND NOT EXISTS (
+                    SELECT 1 FROM orchestration_v2_projection_provider_session_bindings AS binding
+                    WHERE binding.provider_session_id = session.provider_session_id
+                      AND binding.thread_id NOT IN ${sql.in(threadIds)}
+                  )
+                LIMIT ${limit}
+              )
+            `,
+          );
+          yield* deleteBatches(
+            (limit) => sql`
+              DELETE FROM orchestration_v2_projection_provider_session_bindings
+              WHERE rowid IN (
+                SELECT rowid FROM orchestration_v2_projection_provider_session_bindings
+                WHERE thread_id IN ${sql.in(threadIds)}
+                LIMIT ${limit}
+              )
+            `,
+          );
+        }
+        if (threadIds.length < PAGE_SIZE) break;
+        cursor = threadIds.at(-1) ?? cursor;
+      }
+      return purged;
+    });
+
     // A pass prunes through the mark the previous pass left once that mark is a day old, so
     // at least RETAINED_EVENT_HISTORY of events always remains. Receipts stay for idempotency.
     const pruneEventStore = Effect.fn("ProjectionMaintenance.pruneEventStore")(
@@ -298,6 +416,8 @@ export const layer: Layer.Layer<ProjectionMaintenanceV2, never, SqlClient.SqlCli
         );
         deletedEventCount += legacy.deletedEventCount;
 
+        const purgedProjectionRowCount = yield* purgeDeletedThreadContent(now, betweenPages);
+
         const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`;
         const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`;
         const reclaimableBytes =
@@ -306,6 +426,7 @@ export const layer: Layer.Layer<ProjectionMaintenanceV2, never, SqlClient.SqlCli
         return {
           deletedEventCount,
           deletedReceiptCount: legacy.deletedReceiptCount,
+          purgedProjectionRowCount,
           reclaimableBytes,
         };
       },