fix(server): finished outbox rows and old PR cache files are pruned

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

@@ -16,13 +16,17 @@ import {
 import * as Context from "effect/Context";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
+import * as Schedule from "effect/Schedule";
 import * as Schema from "effect/Schema";
 import * as SqlClient from "effect/sql/SqlClient";
 
+import { forkParked } from "../serverActivation.ts";
+
 export const OrchestrationEffectRequestV2 = Schema.Union([
   Schema.Struct({
     type: Schema.Literal("provider-runtime.continue"),
@@ -132,6 +136,21 @@ export const OrchestrationEffectStatusV2 = Schema.Literals([
 ]);
 export type OrchestrationEffectStatusV2 = typeof OrchestrationEffectStatusV2.Type;
 
+/**
+ * How long a succeeded or cancelled effect row is kept after it completes.
+ * Claiming, recovery, the orchestrator and storage cleanup only ask whether a
+ * row is pending, running or failed, so a missing succeeded or cancelled row
+ * reads the same as a present one. Its one other use is deduplication: `enqueue`
+ * ignores an id that already exists, and ids are deterministic per command or
+ * run. Command retries are deduplicated by their receipts first, and a run's
+ * effects are re-enqueued only around one server restart, so a week is far
+ * longer than either and leaves recent history for debugging.
+ * Failed rows are kept: storage cleanup keeps a deleted thread's worktree while
+ * any of its effects failed.
+ */
+export const SETTLED_EFFECT_RETENTION = Duration.days(7);
+const PRUNE_BATCH_SIZE = 500;
+
 export interface OrchestrationEffectV2 {
   readonly id: string;
   readonly commandId: CommandId;
@@ -217,6 +236,8 @@ export interface EffectOutboxV2Shape {
     readonly workerId: string;
     readonly error: string;
   }) => Effect.Effect<boolean, EffectOutboxError>;
+  /** Deletes succeeded and cancelled rows older than `SETTLED_EFFECT_RETENTION`. Returns the count. */
+  readonly pruneSettled: Effect.Effect<number, EffectOutboxError>;
 }
 
 export class EffectOutboxV2 extends Context.Service<EffectOutboxV2, EffectOutboxV2Shape>()(
@@ -629,8 +650,48 @@ export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = La
         }).pipe(
           Effect.mapError((cause) => new EffectOutboxError({ operation: "fail", effectId, cause })),
         ),
+      pruneSettled: Effect.gen(function* () {
+        const cutoff = DateTime.formatIso(
+          DateTime.subtractDuration(yield* DateTime.now, SETTLED_EFFECT_RETENTION),
+        );
+        // Each batch is its own statement, so a large backlog never holds the
+        // write lock for long and writers can commit between batches.
+        const deleteBatch = sql<{ readonly effect_id: string }>`
+          DELETE FROM orchestration_v2_effect_outbox
+          WHERE rowid IN (
+            SELECT rowid
+            FROM orchestration_v2_effect_outbox
+            WHERE status IN ('succeeded', 'cancelled')
+              AND completed_at < ${cutoff}
+            LIMIT ${PRUNE_BATCH_SIZE}
+          )
+          RETURNING effect_id
+        `;
+        let pruned = 0;
+        while (true) {
+          const deleted = (yield* deleteBatch).length;
+          pruned += deleted;
+          if (deleted < PRUNE_BATCH_SIZE) return pruned;
+          yield* Effect.yieldNow;
+        }
+      }).pipe(
+        Effect.mapError((cause) => new EffectOutboxError({ operation: "prune-settled", cause })),
+      ),
     };
 
     return service;
   }),
 );
+
+/** Prunes settled effect rows once the server is active, then every hour. */
+export const pruneWorkerLive = Layer.effectDiscard(
+  Effect.gen(function* () {
+    const outbox = yield* EffectOutboxV2;
+    yield* forkParked(
+      outbox.pruneSettled.pipe(
+        Effect.catch((cause) => Effect.logWarning("Failed to prune settled effects", { cause })),
+        Effect.repeat(Schedule.spaced(Duration.hours(1))),
+      ),
+    );
+  }),
+);

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

@@ -27,10 +27,12 @@ import {
 } from "@t3tools/contracts";
 import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
+import * as Queue from "effect/Queue";
 import * as Ref from "effect/Ref";
 import * as Scheduler from "effect/Scheduler";
 import * as Schema from "effect/Schema";
@@ -65,6 +67,8 @@ const storesProvided = Layer.mergeAll(databaseLayer, eventStoreProvided, project
 const eventSinkProvided = EventSink.layer.pipe(Layer.provide(storesProvided));
 const effectOutboxProvided = EffectOutbox.layer.pipe(Layer.provide(databaseLayer));
 const commandReceiptStoreProvided = CommandReceiptStore.layer.pipe(Layer.provide(databaseLayer));
+// Its own database, for tests that act on every row in the outbox table.
+const isolatedOutboxLayer = Layer.fresh(EffectOutbox.layer.pipe(Layer.provideMerge(databaseLayer)));
 const projectionMaintenanceProvided = ProjectionMaintenance.layer.pipe(
   Layer.provide(storesProvided),
 );
@@ -2223,6 +2227,95 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => {
     }),
   );
 
+  it.effect("prunes settled effects past retention in batches and keeps the rest", () =>
+    Effect.gen(function* () {
+      const outbox = yield* EffectOutbox.EffectOutboxV2;
+      const sql = yield* SqlClient.SqlClient;
+      const now = yield* DateTime.now;
+      const insert = (
+        prefix: string,
+        count: number,
+        status: EffectOutbox.OrchestrationEffectStatusV2,
+        completedAgo: Duration.Duration | null,
+      ) => {
+        const completedAt =
+          completedAgo === null
+            ? null
+            : DateTime.formatIso(DateTime.subtractDuration(now, completedAgo));
+        const createdAt = DateTime.formatIso(now);
+        return sql`
+          WITH RECURSIVE n(i) AS (SELECT 1 UNION ALL SELECT i + 1 FROM n WHERE i < ${count})
+          INSERT INTO orchestration_v2_effect_outbox (
+            effect_id, command_id, thread_id, effect_type, payload_json, status,
+            available_at, created_at, updated_at, completed_at
+          )
+          SELECT ${prefix} || i, 'command:prune', 'thread:prune', 'terminal.cleanup',
+            '{"type":"terminal.cleanup"}', ${status}, ${createdAt}, ${createdAt}, ${createdAt},
+            ${completedAt}
+          FROM n
+        `;
+      };
+      const old = Duration.sum(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1));
+      const recent = Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(1));
+      // More expired rows than one delete batch.
+      yield* insert("succeeded-old:", 1_201, "succeeded", old);
+      yield* insert("cancelled-old:", 2, "cancelled", old);
+      yield* insert("succeeded-recent:", 1, "succeeded", recent);
+      yield* insert("failed-old:", 1, "failed", old);
+      yield* insert("pending:", 1, "pending", null);
+      yield* insert("running:", 1, "running", null);
+
+      assert.equal(yield* outbox.pruneSettled, 1_203);
+
+      const remaining = yield* sql<{ readonly effect_id: string }>`
+        SELECT effect_id FROM orchestration_v2_effect_outbox ORDER BY effect_id
+      `;
+      assert.deepEqual(
+        remaining.map((row) => row.effect_id),
+        ["failed-old:1", "pending:1", "running:1", "succeeded-recent:1"],
+      );
+      assert.equal(yield* outbox.pruneSettled, 0);
+    }).pipe(Effect.provide(isolatedOutboxLayer)),
+  );
+
+  it.effect("prunes settled effects hourly from the layer-owned worker", () =>
+    Effect.gen(function* () {
+      const outbox = yield* EffectOutbox.EffectOutboxV2;
+      const commandId = CommandId.make("command:foundation-prune-worker");
+      const threadId = ThreadId.make("thread:foundation-prune-worker");
+      const workerId = "prune-worker";
+      const request = { type: "terminal.cleanup" } as const;
+      yield* outbox.enqueue([{ id: "effect:prune-worker:done", commandId, threadId, request }]);
+      yield* outbox.claimNext({ workerId, leaseDurationMs: 30_000 });
+      assert.isTrue(yield* outbox.succeed({ effectId: "effect:prune-worker:done", workerId }));
+      yield* outbox.enqueue([{ id: "effect:prune-worker:pending", commandId, threadId, request }]);
+      // Observe each run so the test waits for it instead of racing the clock.
+      const runs = yield* Queue.unbounded<number>();
+      const observed = EffectOutbox.EffectOutboxV2.of({
+        ...outbox,
+        pruneSettled: outbox.pruneSettled.pipe(Effect.tap((pruned) => Queue.offer(runs, pruned))),
+      });
+      const ids = Effect.map(outbox.listByCommandId(commandId), (rows) =>
+        rows.map((row) => row.id).toSorted(),
+      );
+
+      yield* TestClock.adjust(
+        Duration.subtract(EffectOutbox.SETTLED_EFFECT_RETENTION, Duration.minutes(30)),
+      );
+      yield* Layer.build(
+        EffectOutbox.pruneWorkerLive.pipe(
+          Layer.provide(Layer.succeed(EffectOutbox.EffectOutboxV2, observed)),
+        ),
+      );
+      assert.equal(yield* Queue.take(runs), 0);
+      assert.deepEqual(yield* ids, ["effect:prune-worker:done", "effect:prune-worker:pending"]);
+
+      yield* TestClock.adjust("1 hour");
+      assert.equal(yield* Queue.take(runs), 1);
+      assert.deepEqual(yield* ids, ["effect:prune-worker:pending"]);
+    }).pipe(Effect.scoped, Effect.provide(isolatedOutboxLayer)),
+  );
+
   it.effect("does not emit a SQL span for an empty safety claim", () =>
     Effect.gen(function* () {
       const outbox = yield* EffectOutbox.EffectOutboxV2;

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

@@ -17,7 +17,10 @@ import { layer as checkpointRollbackServiceLayer } from "./CheckpointRollbackSer
 import { layer as commandPolicyLayer } from "./CommandPolicy.ts";
 import { layerFromApplicationReceipts as commandReceiptStoreLayer } from "./CommandReceiptStore.ts";
 import { layer as contextHandoffServiceLayer } from "./ContextHandoffService.ts";
-import { layer as effectOutboxLayer } from "./EffectOutbox.ts";
+import {
+  layer as effectOutboxLayer,
+  pruneWorkerLive as effectOutboxPruneWorkerLive,
+} from "./EffectOutbox.ts";
 import {
   executorLayer as effectExecutorLayer,
   layer as effectWorkerLayer,
@@ -319,6 +322,7 @@ export const OrchestrationV2ProductionLayerLive = Layer.mergeAll(
   ),
   providerContinuationWorkerProvided,
   agentSessionImporterProvided,
+  effectOutboxPruneWorkerLive.pipe(Layer.provide(effectOutboxLayer)),
 ).pipe(
   Layer.provide(Scheduler.layer),
   Layer.provideMerge(OrchestrationEventInfrastructureLayerLive),

apps/server/src/pullRequest/PullRequestReadCache.test.ts

@@ -1,18 +1,107 @@
 import { assert, it } from "@effect/vitest";
 import * as NodeServices from "@effect/platform-node/NodeServices";
 import { PullRequestOperationError } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
 import * as Deferred from "effect/Deferred";
+import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Fiber from "effect/Fiber";
 import * as FileSystem from "effect/FileSystem";
+import * as Layer from "effect/Layer";
+import * as Path from "effect/Path";
+import * as Queue from "effect/Queue";
 import * as TestClock from "effect/testing/TestClock";
 import * as KeyValueStore from "effect/persistence/KeyValueStore";
+import * as ServerConfig from "../config.ts";
 import * as PullRequestReadCache from "./PullRequestReadCache.ts";
 
 const cacheLayer = (directory: string) =>
   PullRequestReadCache.make.pipe(Effect.provide(KeyValueStore.layerFileSystem(directory)));
 
+/** Sets every file's mtime to `age` before the current test clock time. */
+const ageFiles = (directory: string, names: ReadonlyArray<string>, age: Duration.Duration) =>
+  Effect.gen(function* () {
+    const fs = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const mtime = DateTime.toDateUtc(DateTime.subtractDuration(yield* DateTime.now, age));
+    for (const name of names) yield* fs.utimes(path.join(directory, name), mtime, mtime);
+  });
+
 it.layer(NodeServices.layer)("PR filesystem cache", (it) => {
+  it.effect("prunes entry files written before the max age and keeps fresh ones", () =>
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem;
+      const directory = yield* fs.makeTempDirectoryScoped({ prefix: "t3-pr-cache-" });
+      yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z")));
+      const cache = yield* cacheLayer(directory);
+      yield* cache.get("old", Effect.succeed("old"), ["pr"]);
+      yield* cache.invalidate("pr");
+      const [stale, revisions] = (yield* fs.readDirectory(directory)).toSorted();
+      assert.strictEqual(revisions, "revisions");
+      yield* cache.get("fresh", Effect.succeed("fresh"));
+      const fresh = (yield* fs.readDirectory(directory)).find(
+        (name) => name !== stale && name !== revisions,
+      );
+      const unrelated = "unrelated.json";
+      yield* fs.writeFileString(`${directory}/${unrelated}`, "{}");
+      const justExpired = Duration.sum(
+        PullRequestReadCache.ENTRY_FILE_MAX_AGE,
+        Duration.seconds(1),
+      );
+      yield* ageFiles(directory, [stale!, revisions!, unrelated], justExpired);
+      yield* ageFiles(directory, [fresh!], PullRequestReadCache.ENTRY_FILE_MAX_AGE);
+
+      yield* PullRequestReadCache.pruneExpiredEntryFiles(directory);
+
+      assert.deepStrictEqual(
+        (yield* fs.readDirectory(directory)).toSorted(),
+        [fresh!, revisions!, unrelated].toSorted(),
+      );
+    }),
+  );
+
+  it.effect("sweeps the cache directory every hour once the layer is built", () =>
+    Effect.gen(function* () {
+      const fs = yield* FileSystem.FileSystem;
+      const { providerStatusCacheDir } = yield* ServerConfig.ServerConfig;
+      const directory = `${providerStatusCacheDir}/pull-requests`;
+      yield* TestClock.setTime(DateTime.toEpochMillis(DateTime.makeUnsafe("2026-10-01T00:00:00Z")));
+      yield* (yield* cacheLayer(directory)).get("summary", Effect.succeed("cached"));
+      const [entry] = yield* fs.readDirectory(directory);
+      // Within the max age at the first two sweeps, past it at the third.
+      yield* ageFiles(
+        directory,
+        [entry!],
+        Duration.subtract(PullRequestReadCache.ENTRY_FILE_MAX_AGE, Duration.minutes(90)),
+      );
+      // A sweep's last file operation is the entry's stat when it keeps the
+      // entry, and its removal otherwise. Waiting for it means the sweep has
+      // finished before the clock moves.
+      const operations = yield* Queue.unbounded<string>();
+      const observedFs = FileSystem.FileSystem.of({
+        ...fs,
+        stat: (path) => fs.stat(path).pipe(Effect.tap(() => Queue.offer(operations, "stat"))),
+        remove: (path, options) =>
+          fs.remove(path, options).pipe(Effect.tap(() => Queue.offer(operations, "remove"))),
+      });
+      yield* Layer.build(
+        PullRequestReadCache.layer.pipe(
+          Layer.provide(Layer.succeed(FileSystem.FileSystem, observedFs)),
+        ),
+      );
+      assert.strictEqual(yield* Queue.take(operations), "stat");
+      yield* TestClock.adjust("1 hour");
+      assert.strictEqual(yield* Queue.take(operations), "stat");
+      assert.deepStrictEqual(yield* fs.readDirectory(directory), [entry]);
+      yield* TestClock.adjust("1 hour");
+      assert.deepStrictEqual(yield* Queue.takeN(operations, 2), ["stat", "remove"]);
+      assert.deepStrictEqual(yield* fs.readDirectory(directory), []);
+    }).pipe(
+      Effect.scoped,
+      Effect.provide(ServerConfig.layerTest(process.cwd(), { prefix: "t3-pr-cache-layer-" })),
+    ),
+  );
+
   it.effect("reuses files after restart and respects the original expiry", () =>
     Effect.gen(function* () {
       const fs = yield* FileSystem.FileSystem;