🤖 feat: keep shared sub-agent notebooks safe across workspace removal

src/common/schemas/project.ts

@@ -105,6 +105,10 @@ export const WorkspaceConfigSchema = z.object({
     description:
       "If set, this workspace is a child workspace spawned from the parent workspaceId (enables nesting in UI and backend orchestration).",
   }),
+  memoryOwnerWorkspaceId: z.string().optional().meta({
+    description:
+      "Memory owner pinned when an intermediate ancestor was removed while this descendant stayed alive: the parentWorkspaceId chain no longer reaches the task-tree root, so this keeps /memories/workspace bound to the root's store (memoryWorkspaceOwner.ts). Set only by workspace removal.",
+  }),
   agentType: z.string().optional().meta({
     description: 'If set, selects an agent preset for this workspace (e.g., "explore" or "exec").',
   }),

src/node/services/memoryConsolidationService.test.ts

@@ -1,4 +1,5 @@
 import { describe, expect, it } from "bun:test";
+import { Effect } from "effect";
 
 import * as fsPromises from "node:fs/promises";
 import * as path from "node:path";
@@ -7,7 +8,10 @@ import type { LanguageModelV3CallOptions, LanguageModelV3StreamPart } from "@ai-
 
 import type { CompactionCompletionMetadata } from "@/common/types/compaction";
 import { createMuxMessage } from "@/common/types/message";
-import type { MemoryConsolidationStatusChangeEventPayload } from "@/common/orpc/schemas/memory";
+import type {
+  MemoryConsolidationStatusChangeEventPayload,
+  MemoryHarvestRecordPayload,
+} from "@/common/orpc/schemas/memory";
 import { MULTI_PROJECT_CONFIG_KEY } from "@/common/constants/multiProject";
 import { EXPERIMENT_IDS } from "@/common/constants/experiments";
 import {
@@ -17,6 +21,7 @@ import {
 import { Ok } from "@/common/types/result";
 import { Config } from "@/node/config";
 import {
+  HARVEST_MAX_ATTEMPTS,
   MemoryConsolidationService,
   resolveDreamAgentBody,
   resolveDreamModelString,
@@ -1503,6 +1508,102 @@ describe("MemoryConsolidationService", () => {
     expect(fixture.modelCalls).toHaveLength(3);
   });
 
+  it("releases the teardown gate when a removal aborts before its point of no return", async () => {
+    using fixture = await createFixture();
+    // The removal drain marks the workspace; every trigger is refused...
+    await fixture.service.cancelInFlightConsolidation("ws-dream");
+    const refused = await fixture.service.maybeRun("ws-dream", "manual");
+    expect(refused.success).toBe(false);
+    if (!refused.success) expect(refused.error).toContain("being removed");
+    expect(fixture.modelCalls).toHaveLength(0);
+    // ...until the aborted removal (no tombstone, workspace intact) lifts it.
+    fixture.service.releaseRemovalCancellation("ws-dream");
+    expect((await fixture.service.maybeRun("ws-dream", "manual")).success).toBe(true);
+    expect(fixture.modelCalls).toHaveLength(1);
+  });
+
+  it("finalizes a removed workspace's retryable harvest records so they are never retried", async () => {
+    using fixture = await createFixture({ modelFactory: harvestCandidateModel });
+    await fixture.addWorkspace("ws-sub", { parentWorkspaceId: "ws-dream" });
+    const metadata = await seedCompactionEpoch(fixture, "ws-sub");
+    await fsPromises.writeFile(
+      path.join(fixture.xumHome, "memory-consolidation.json"),
+      JSON.stringify({
+        workspaces: {},
+        harvestsByWorkspace: {
+          "ws-sub": {
+            [metadata.summaryMessageId]: {
+              status: "failed",
+              startedAt: Date.now() - 10_000,
+              completedAt: Date.now() - 9_000,
+              attemptCount: 1,
+              boundaryKey: metadata.summaryMessageId,
+              compactionEpoch: metadata.compactionEpoch,
+              acceptedCandidates: 0,
+              skippedCandidates: 0,
+              error: "crashed mid-harvest",
+              completionMetadata: metadata,
+            },
+          },
+        },
+      })
+    );
+    await fixture.service.finalizeHarvestsForRemoval("ws-sub");
+    const record = (await fixture.service.getStatus("ws-sub")).latestHarvestRecord;
+    expect(record?.status).toBe("failed");
+    expect(record?.attemptCount).toBe(HARVEST_MAX_ATTEMPTS);
+    // The owner's run no longer sees a retryable child bucket.
+    expect((await fixture.service.maybeRun("ws-dream", "manual")).success).toBe(true);
+    expect((await fixture.service.getStatus("ws-sub")).latestHarvestRecord?.attemptCount).toBe(
+      HARVEST_MAX_ATTEMPTS
+    );
+  });
+
+  it("keeps a removal-finalized harvest record terminal against residual retryable writes", async () => {
+    using fixture = await createFixture({ modelFactory: harvestCandidateModel });
+    await fixture.addWorkspace("ws-sub", { parentWorkspaceId: "ws-dream" });
+    const metadata = await seedCompactionEpoch(fixture, "ws-sub");
+    const boundaryKey = metadata.summaryMessageId;
+    const base = {
+      startedAt: Date.now() - 10_000,
+      attemptCount: 1,
+      boundaryKey,
+      compactionEpoch: metadata.compactionEpoch,
+      acceptedCandidates: 0,
+      skippedCandidates: 0,
+      completionMetadata: metadata,
+    };
+    // Residual runs of the bounded cancellation drain record through the same
+    // path as the live harvest; reach it directly to interleave with finalization.
+    const save = (record: MemoryHarvestRecordPayload) =>
+      Effect.runPromise(
+        (
+          fixture.service as unknown as {
+            saveHarvestRecordEffect: (
+              workspaceId: string,
+              boundaryKey: string,
+              record: MemoryHarvestRecordPayload,
+              projectPath: string
+            ) => Effect.Effect<void>;
+          }
+        ).saveHarvestRecordEffect("ws-sub", boundaryKey, record, "")
+      );
+    await save({ ...base, status: "pending" });
+    await fixture.service.finalizeHarvestsForRemoval("ws-sub");
+    const latest = async () => (await fixture.service.getStatus("ws-sub")).latestHarvestRecord;
+    expect((await latest())?.attemptCount).toBe(HARVEST_MAX_ATTEMPTS);
+
+    // A residual retryable failure landing after finalization must not reopen the bucket...
+    await save({ ...base, status: "failed", completedAt: Date.now(), error: "residual failure" });
+    expect((await latest())?.attemptCount).toBe(HARVEST_MAX_ATTEMPTS);
+    expect((await latest())?.error).toContain("workspace removed");
+    // ...while a residual completion (its writes really landed) is kept as the truth,
+    // and finalization never demotes a completed record.
+    await save({ ...base, status: "completed", completedAt: Date.now(), acceptedCandidates: 1 });
+    await fixture.service.finalizeHarvestsForRemoval("ws-sub");
+    expect((await latest())?.status).toBe("completed");
+  });
+
   it("launch sweep skips archived workspaces and caps runs per launch", async () => {
     using fixture = await createFixture();
     const dayAgo = Date.now() - 25 * 60 * 60 * 1000;

src/node/services/memoryConsolidationService.ts

@@ -291,7 +291,26 @@ function pruneHarvestRecords(records: Record<string, MemoryHarvestRecord>): void
   }
 }
 
-const HARVEST_MAX_ATTEMPTS = 3;
+export const HARVEST_MAX_ATTEMPTS = 3;
+
+/** Completed, or failed with retries exhausted: nothing may retry it. */
+function isTerminalHarvestRecord(record: MemoryHarvestRecord): boolean {
+  return (
+    record.status === "completed" ||
+    (record.status === "failed" && record.attemptCount >= HARVEST_MAX_ATTEMPTS)
+  );
+}
+
+/** Terminal marker for a bucket whose transcript is being deleted (see finalizeHarvestsForRemoval). */
+function finalizeHarvestRecordForRemoval(record: MemoryHarvestRecord): MemoryHarvestRecord {
+  return {
+    ...record,
+    status: "failed",
+    completedAt: record.completedAt ?? Date.now(),
+    attemptCount: HARVEST_MAX_ATTEMPTS,
+    error: "workspace removed before the harvest could be retried; transcript no longer available",
+  };
+}
 
 export class MemoryConsolidationService extends EventEmitter {
   private readonly sidecarPath: string;
@@ -324,10 +343,9 @@ export class MemoryConsolidationService extends EventEmitter {
    * post-harvest sweep, and a cancelled run still starts retryable-harvest
    * recovery, each with a fresh un-aborted signal. Entry points refuse and
    * new controllers start pre-aborted while a workspace is in this set.
-   * Entries are never cleared: removal is terminal, and if a force=false
-   * removal fails after the drain, losing background consolidation for the
-   * surviving workspace (until restart) matches the documented drained-
-   * producers tradeoff in WorkspaceService.removeWorkspace. Cross-PROCESS
+   * Entries are cleared only when removal aborts before its point of no
+   * return (releaseRemovalCancellation); once the tombstone is published,
+   * removal is terminal. Cross-PROCESS
    * teardown is covered by the durable removal tombstone instead (see
    * workspaceRemoval.ts), checked at memory mutation commit points.
    */
@@ -485,16 +503,27 @@ export class MemoryConsolidationService extends EventEmitter {
     const self = this;
     return Effect.uninterruptible(
       Effect.gen(function* () {
-        yield* Effect.promise(() =>
+        const saved = yield* Effect.promise(() =>
           self.locks.withLock(self.sidecarPath, async () => {
             const file = await self.load();
             file.harvestsByWorkspace[workspaceId] ??= {};
+            const existing = file.harvestsByWorkspace[workspaceId][boundaryKey];
+            // A terminal record is never reopened: removal finalization
+            // (finalizeHarvestsForRemoval) races the bounded cancellation
+            // drain's residual harvest runs on this file, and a residual
+            // pending/retryable-failure write landing afterwards would turn
+            // a bucket whose transcript is gone back into a retry candidate.
+            // Only a genuine completion may replace it (the writes happened).
+            if (existing !== undefined && isTerminalHarvestRecord(existing)) {
+              if (record.status !== "completed") return false;
+            }
             file.harvestsByWorkspace[workspaceId][boundaryKey] = record;
             pruneHarvestRecords(file.harvestsByWorkspace[workspaceId]);
             await writeFileAtomic(self.sidecarPath, JSON.stringify(file, null, 2));
+            return true;
           })
         );
-        self.emitStatusChange(workspaceId, projectPath);
+        if (saved) self.emitStatusChange(workspaceId, projectPath);
       })
     );
   }
@@ -626,6 +655,51 @@ export class MemoryConsolidationService extends EventEmitter {
     return Effect.runPromise(this.cancelInFlightConsolidationEffect(workspaceId));
   }
 
+  /**
+   * Removal aborted BEFORE its point of no return (no tombstone published, the
+   * workspace stays registered and intact — e.g. a non-forced removal whose
+   * checkout deletion was refused): lift the teardown gate again, or the
+   * surviving workspace would refuse every Dream run and post-compaction
+   * harvest until restart. The drained in-flight runs are gone regardless
+   * (retryable harvests recover on the next trigger).
+   */
+  releaseRemovalCancellation(workspaceId: string): void {
+    this.removalCancelled.delete(workspaceId);
+  }
+
+  /**
+   * Removal teardown for harvest state: the workspace's transcript is about
+   * to be deleted, so its failed/stale-pending harvest records can never be
+   * retried (recovery needs the compaction epoch's messages) — and once the
+   * config entry is gone they could not even be associated with the memory
+   * owner. Mark them terminal now so nothing lingers as "retryable".
+   */
+  async finalizeHarvestsForRemoval(workspaceId: string): Promise<void> {
+    // One read-check-write under the sidecar lock: residual harvest runs
+    // (cancelInFlightConsolidation's drain is bounded) may still be recording
+    // outcomes, and a completion landing between an unlocked read and this
+    // write must not be overwritten with a failure.
+    const finalized = await this.locks.withLock(this.sidecarPath, async () => {
+      const file = await this.load();
+      const records = file.harvestsByWorkspace[workspaceId];
+      if (records === undefined) return false;
+      let changed = false;
+      for (const [boundaryKey, record] of Object.entries(records)) {
+        if (isTerminalHarvestRecord(record)) continue;
+        records[boundaryKey] = finalizeHarvestRecordForRemoval(record);
+        changed = true;
+      }
+      if (changed) await writeFileAtomic(this.sidecarPath, JSON.stringify(file, null, 2));
+      return changed;
+    });
+    if (!finalized) return;
+    const workspace = this.config.findWorkspace(workspaceId);
+    this.emitStatusChange(
+      workspaceId,
+      workspace == null ? "" : resolveConsolidationProjectPath(workspace)
+    );
+  }
+
   /**
    * Teardown pipeline: uninterruptible end-to-end so the r61 mark, the abort
    * loop, and the residual-run handoff can never be separated, with the
@@ -846,11 +920,18 @@ export class MemoryConsolidationService extends EventEmitter {
       }
 
       const projectPath = resolveConsolidationProjectPath(workspace);
+      // A child's redirected run sweeps under the owner's identity; the
+      // child's own removal tombstone still refuses every read and commit of
+      // the run (MemoryScopeContext.guardedWorkspaceId) — a remover in another
+      // backend cannot abort this controller.
       const ctx: MemoryScopeContext = {
         runtime: null,
         checkoutCwd: "",
         workspaceId,
         projectPath,
+        ...(options.actingWorkspaceId !== undefined && options.actingWorkspaceId !== workspaceId
+          ? { guardedWorkspaceId: options.actingWorkspaceId }
+          : {}),
       };
 
       const result = yield* Effect.promise(async () =>