fix(server): deleted threads release their stored history

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

@@ -5676,6 +5676,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
       );
     // Read under the source's lock; validate the modes copied here, not an earlier read.
     yield* refuseAboveDispatchModeLimit(command, command.sourceThreadId, sourceProjection.thread);
+    // Deleted threads lose their content once purged, so nothing may start depending on one.
     if (sourceProjection.thread.deletedAt !== null) {
       return yield* new OrchestratorDispatchError({
         commandId: command.commandId,

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,214 @@ 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 counts: 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_turn_item_counts WHERE thread_id = ${threadId}) AS counts,
+        (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.isAbove(beforeOld?.counts ?? 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,
+      counts: 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",
+      // Kept exact by triggers on projection_turn_items.
+      "orchestration_v2_projection_turn_item_counts",
+    ];
+    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 deleted threads a live thread still inherits or merges back from", () =>
+  Effect.gen(function* () {
+    const { maintenance, now, seed, remove } = yield* purgeFixture;
+    const sql = yield* SqlClient.SqlClient;
+    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");
+    const mergedBack = yield* seed("merged-back");
+    yield* sql`
+      INSERT INTO orchestration_v2_projection_context_transfers
+        (context_transfer_id, source_thread_id, target_thread_id, type, status, updated_at, payload_json)
+      VALUES ('transfer:purge', ${mergedBack.id}, ${fork.id},
+        'merge_back', 'pending', ${DateTime.formatIso(now)}, '{}')
+    `;
+    yield* remove(mergedBack, yield* DateTime.now);
+    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);
+    assert.equal((yield* countContent(mergedBack.id))?.items, 1);
+  }).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");
+
+    const result = yield* maintenance.pruneEventStore();
+    assert.isAtLeast(result.purgedProjectionRowCount, 1200);
+    assert.equal((yield* countContent(big.id))?.positions, 0);
+  }).pipe(Effect.provide(layerTest)),
+);