feat(orchestrator): archive and delete owned subagent subtrees

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

@@ -0,0 +1,82 @@
+import { assert, it } from "@effect/vitest";
+import {
+  CheckpointId,
+  CheckpointRef,
+  CheckpointScopeId,
+  ThreadId,
+  type OrchestrationV2Checkpoint,
+  type OrchestrationV2CheckpointScope,
+  type OrchestrationV2ThreadProjection,
+} from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import { CheckpointCleanupServiceV2, layer } from "./CheckpointCleanupService.ts";
+import { CheckpointServiceV2 } from "./CheckpointService.ts";
+import { ProjectionStoreV2 } from "./ProjectionStore.ts";
+
+it.effect("cleans every checkpoint scope through the latest known ordinal", () => {
+  const threadId = ThreadId.make("thread_checkpoint_cleanup");
+  const firstScope = {
+    id: CheckpointScopeId.make("scope_first"),
+    threadId,
+  } as OrchestrationV2CheckpointScope;
+  const secondScope = {
+    id: CheckpointScopeId.make("scope_second"),
+    threadId,
+  } as OrchestrationV2CheckpointScope;
+  const checkpoint = {
+    id: CheckpointId.make("checkpoint_first"),
+    scopeId: firstScope.id,
+    ordinalWithinScope: 2,
+    ref: CheckpointRef.make("refs/t3/test/checkpoint-first"),
+  } as OrchestrationV2Checkpoint;
+  const projection = {
+    thread: { id: threadId },
+    runs: [{ ordinal: 4 }],
+    checkpointScopes: [firstScope, secondScope],
+    checkpoints: [checkpoint],
+  } as unknown as OrchestrationV2ThreadProjection;
+  const calls: Array<{
+    readonly scopeId: CheckpointScopeId;
+    readonly checkpointIds: ReadonlyArray<CheckpointId>;
+    readonly throughOrdinal: number;
+  }> = [];
+  const testLayer = layer.pipe(
+    Layer.provide(
+      Layer.merge(
+        Layer.mock(ProjectionStoreV2)({
+          getThreadProjection: () => Effect.succeed(projection),
+        }),
+        Layer.mock(CheckpointServiceV2)({
+          deleteScopeRefs: (input) => {
+            calls.push({
+              scopeId: input.scope.id,
+              checkpointIds: input.checkpoints.map((item) => item.id),
+              throughOrdinal: input.throughOrdinal,
+            });
+            return Effect.void;
+          },
+        }),
+      ),
+    ),
+  );
+
+  return Effect.gen(function* () {
+    const cleanup = yield* CheckpointCleanupServiceV2;
+    yield* cleanup.cleanup(threadId);
+
+    assert.deepEqual(calls, [
+      {
+        scopeId: firstScope.id,
+        checkpointIds: [checkpoint.id],
+        throughOrdinal: 4,
+      },
+      {
+        scopeId: secondScope.id,
+        checkpointIds: [],
+        throughOrdinal: 4,
+      },
+    ]);
+  }).pipe(Effect.provide(testLayer));
+});

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

@@ -0,0 +1,70 @@
+import { ThreadId } from "@t3tools/contracts";
+import * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Schema from "effect/Schema";
+
+import { CheckpointServiceV2 } from "./CheckpointService.ts";
+import { ProjectionStoreV2 } from "./ProjectionStore.ts";
+
+export class CheckpointCleanupError extends Schema.TaggedErrorClass<CheckpointCleanupError>()(
+  "CheckpointCleanupError",
+  {
+    threadId: ThreadId,
+    cause: Schema.Defect(),
+  },
+) {
+  override get message(): string {
+    return `Failed to clean checkpoint refs for thread ${this.threadId}.`;
+  }
+}
+
+export class CheckpointCleanupServiceV2 extends Context.Service<
+  CheckpointCleanupServiceV2,
+  {
+    readonly cleanup: (threadId: ThreadId) => Effect.Effect<void, CheckpointCleanupError>;
+  }
+>()("t3/orchestration-v2/CheckpointCleanupService/CheckpointCleanupServiceV2") {}
+
+export const layer: Layer.Layer<
+  CheckpointCleanupServiceV2,
+  never,
+  CheckpointServiceV2 | ProjectionStoreV2
+> = Layer.effect(
+  CheckpointCleanupServiceV2,
+  Effect.gen(function* () {
+    const checkpoints = yield* CheckpointServiceV2;
+    const projections = yield* ProjectionStoreV2;
+
+    const cleanup = Effect.fn("orchestrationV2.checkpointCleanup.cleanup")(function* (
+      threadId: ThreadId,
+    ) {
+      const projection = yield* projections.getThreadProjection(threadId);
+      const throughOrdinal = Math.max(
+        0,
+        ...projection.runs.map((run) => run.ordinal),
+        ...projection.checkpoints.map((checkpoint) => checkpoint.ordinalWithinScope),
+      );
+
+      yield* Effect.forEach(
+        projection.checkpointScopes,
+        (scope) =>
+          checkpoints.deleteScopeRefs({
+            scope,
+            checkpoints: projection.checkpoints.filter(
+              (checkpoint) => checkpoint.scopeId === scope.id,
+            ),
+            throughOrdinal,
+          }),
+        { concurrency: 1, discard: true },
+      );
+    });
+
+    return CheckpointCleanupServiceV2.of({
+      cleanup: (threadId) =>
+        cleanup(threadId).pipe(
+          Effect.mapError((cause) => new CheckpointCleanupError({ threadId, cause })),
+        ),
+    });
+  }),
+);

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

@@ -103,13 +103,26 @@ export class CheckpointDeleteStaleRefsError extends Schema.TaggedErrorClass<Chec
   }
 }
 
+export class CheckpointDeleteScopeRefsError extends Schema.TaggedErrorClass<CheckpointDeleteScopeRefsError>()(
+  "CheckpointDeleteScopeRefsError",
+  {
+    scopeId: CheckpointScopeId,
+    cause: Schema.Defect(),
+  },
+) {
+  override get message(): string {
+    return `Failed to delete checkpoint refs for scope ${this.scopeId}.`;
+  }
+}
+
 export const CheckpointServiceV2Error = Schema.Union([
   CheckpointRootScopePrepareError,
   CheckpointScopeEnsureError,
   CheckpointBaselineCaptureError,
   CheckpointCaptureError,
   CheckpointRestoreError,
   CheckpointDeleteStaleRefsError,
+  CheckpointDeleteScopeRefsError,
 ]);
 export type CheckpointServiceV2Error = typeof CheckpointServiceV2Error.Type;
 
@@ -150,6 +163,16 @@ export interface CheckpointServiceV2Shape {
     readonly scope: OrchestrationV2CheckpointScope;
     readonly checkpoints: ReadonlyArray<OrchestrationV2Checkpoint>;
   }) => Effect.Effect<void, CheckpointServiceV2Error>;
+  readonly deleteScopeRefs: (input: {
+    readonly scope: OrchestrationV2CheckpointScope;
+    readonly checkpoints: ReadonlyArray<OrchestrationV2Checkpoint>;
+    /**
+     * Baselines can be captured before their projection event is committed.
+     * Include every deterministic ordinal through this value so deletion also
+     * cleans refs left by an interrupted capture.
+     */
+    readonly throughOrdinal: number;
+  }) => Effect.Effect<void, CheckpointServiceV2Error>;
 }
 
 export class CheckpointServiceV2 extends Context.Service<
@@ -541,6 +564,40 @@ export const layer: Layer.Layer<
         ),
       );
 
+    const deleteScopeRefs: CheckpointServiceV2Shape["deleteScopeRefs"] = (input) => {
+      const checkpointRefs = new Set<CheckpointRef>(
+        input.checkpoints.map((checkpoint) => checkpoint.ref),
+      );
+      for (
+        let ordinalWithinScope = 0;
+        ordinalWithinScope <= input.throughOrdinal;
+        ordinalWithinScope += 1
+      ) {
+        checkpointRefs.add(
+          checkpointRefForScopeOrdinal({
+            scopeId: input.scope.id,
+            ordinalWithinScope,
+          }),
+        );
+      }
+
+      return withWorkspaceLock(
+        input.scope.cwd,
+        checkpointStore.deleteCheckpointRefs({
+          cwd: input.scope.cwd,
+          checkpointRefs: Array.from(checkpointRefs),
+        }),
+      ).pipe(
+        Effect.mapError(
+          (cause) =>
+            new CheckpointDeleteScopeRefsError({
+              scopeId: input.scope.id,
+              cause,
+            }),
+        ),
+      );
+    };
+
     return CheckpointServiceV2.of({
       prepareRootRunScope: (input) =>
         makeRootRunScope({ ...input, idAllocator }).pipe(
@@ -559,6 +616,7 @@ export const layer: Layer.Layer<
       capture,
       restore,
       deleteStaleRefs,
+      deleteScopeRefs,
     } satisfies CheckpointServiceV2Shape);
   }),
 );

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

@@ -81,6 +81,9 @@ export const OrchestrationEffectRequestV2 = Schema.Union([
     runId: RunId,
     scopeId: CheckpointScopeId,
   }),
+  Schema.Struct({
+    type: Schema.Literal("checkpoint.cleanup"),
+  }),
   Schema.Struct({
     type: Schema.Literal("terminal.cleanup"),
   }),
@@ -95,6 +98,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
   "provider-session.detach",
   "provider-thread.rollback",
   "checkpoint.capture",
+  "checkpoint.cleanup",
   "terminal.cleanup",
   "attachment.cleanup",
 ] as const satisfies ReadonlyArray<OrchestrationEffectRequestV2["type"]>;

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

@@ -15,6 +15,7 @@ import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Ref from "effect/Ref";
 
+import { CheckpointCleanupServiceV2 } from "./CheckpointCleanupService.ts";
 import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
 import type { OrchestrationEffectV2 } from "./EffectOutbox.ts";
 import { executorLayer, OrchestrationEffectExecutorV2 } from "./EffectWorker.ts";
@@ -125,6 +126,10 @@ function makeExecutorLayer(input: {
       CheckpointRollbackServiceV2,
       CheckpointRollbackServiceV2.of({ execute: () => Effect.void }),
     ),
+    Layer.succeed(
+      CheckpointCleanupServiceV2,
+      CheckpointCleanupServiceV2.of({ cleanup: () => Effect.void }),
+    ),
     Layer.succeed(
       RuntimeRequestServiceV2,
       RuntimeRequestServiceV2.of({ respond: () => Effect.void }),

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

@@ -7,10 +7,11 @@ import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 
+import { CheckpointCleanupServiceV2 } from "./CheckpointCleanupService.ts";
+import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
+import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts";
 import { RunFinalizationService } from "./RunFinalizationService.ts";
 import { ResourceCleanupService } from "./ResourceCleanupService.ts";
-import { EffectOutboxV2, type OrchestrationEffectV2 } from "./EffectOutbox.ts";
-import { CheckpointRollbackServiceV2 } from "./CheckpointRollbackService.ts";
 import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
 import { ProviderTurnControlServiceV2 } from "./ProviderTurnControlService.ts";
 import { ProviderTurnStartServiceV2 } from "./ProviderTurnStartService.ts";
@@ -41,6 +42,7 @@ export const executorLayer: Layer.Layer<
   never,
   | ProviderSessionManagerV2
   | RunFinalizationService
+  | CheckpointCleanupServiceV2
   | CheckpointRollbackServiceV2
   | ProviderTurnControlServiceV2
   | ProviderTurnStartServiceV2
@@ -50,6 +52,7 @@ export const executorLayer: Layer.Layer<
   Effect.gen(function* () {
     const runFinalization = yield* RunFinalizationService;
     const resourceCleanup = yield* ResourceCleanupService;
+    const checkpointCleanup = yield* CheckpointCleanupServiceV2;
     const checkpointRollback = yield* CheckpointRollbackServiceV2;
     const providerSessions = yield* ProviderSessionManagerV2;
     const providerTurnControl = yield* ProviderTurnControlServiceV2;
@@ -229,6 +232,17 @@ export const executorLayer: Layer.Layer<
                     }),
                 ),
               );
+          case "checkpoint.cleanup":
+            return checkpointCleanup.cleanup(effect.threadId).pipe(
+              Effect.mapError(
+                (cause) =>
+                  new OrchestrationEffectExecutionError({
+                    effectId: effect.id,
+                    effectType: effect.request.type,
+                    cause,
+                  }),
+              ),
+            );
           case "terminal.cleanup":
             return resourceCleanup.cleanupTerminals(effect.threadId).pipe(
               Effect.mapError(