🤖 feat: share workspace memory notes across a sub-agent task tree

src/cli/debug/refinements.test.ts

@@ -2,6 +2,7 @@ import { afterEach, describe, expect, it, spyOn } from "bun:test";
 
 import * as fsPromises from "node:fs/promises";
 import * as path from "node:path";
+import { Config } from "@/node/config";
 import { appendRefinementEvent } from "@/node/services/refinement/refinementJournal";
 import { TestTempDir } from "@/node/services/tools/testHelpers";
 import { refinementsCommand } from "./refinements";
@@ -11,8 +12,14 @@ import { refinementsCommand } from "./refinements";
  * created, inside a `<root>/sessions/<ws>` layout so the confinement roots
  * resolve like a real mux home.
  */
-async function seedFixture(root: string): Promise<{ sessionDir: string; skillFile: string }> {
+async function seedFixture(
+  root: string
+): Promise<{ sessionDir: string; skillFile: string; config: Config }> {
   const sessionDir = path.join(root, "sessions", "ws-cli");
+  // Rollback resolves shared-memory ownership from a config.json that must
+  // exist (an absent one reads as mid-rewrite): persist the empty default.
+  const config = new Config(root);
+  await config.editConfig((cfg) => cfg);
   const skillFile = path.join(root, "checkout", ".mux", "skills", "cli-skill", "SKILL.md");
   await fsPromises.mkdir(path.dirname(skillFile), { recursive: true });
   await fsPromises.writeFile(skillFile, "---\nname: cli-skill\n---\n", "utf-8");
@@ -24,7 +31,7 @@ async function seedFixture(root: string): Promise<{ sessionDir: string; skillFil
     inverse: { op: "delete-files", paths: [skillFile] },
     evidence: { toolName: "agent_skill_write" },
   });
-  return { sessionDir, skillFile };
+  return { sessionDir, skillFile, config };
 }
 
 describe("debug refinements command", () => {
@@ -37,7 +44,7 @@ describe("debug refinements command", () => {
 
   it("lists rows and performs a rollback with lineage output", async () => {
     using tempDir = new TestTempDir("test-debug-refinements");
-    const { sessionDir, skillFile } = await seedFixture(tempDir.path);
+    const { sessionDir, skillFile, config } = await seedFixture(tempDir.path);
     const lines: string[] = [];
     const logSpy = spyOn(console, "log").mockImplementation((line: string) => {
       lines.push(line);
@@ -50,7 +57,7 @@ describe("debug refinements command", () => {
       const rowId = lines[0].split("  ")[0];
 
       lines.length = 0;
-      await refinementsCommand("ws-cli", { sessionDir, rollback: rowId });
+      await refinementsCommand("ws-cli", { sessionDir, config, rollback: rowId });
       // Earlier test files in the same process may have reset exitCode to 0,
       // so assert "not failing" rather than "never touched".
       expect(process.exitCode ?? 0).toBe(0);
@@ -74,19 +81,49 @@ describe("debug refinements command", () => {
 
   it("reports refusals on stderr and sets a failing exit code", async () => {
     using tempDir = new TestTempDir("test-debug-refinements-refuse");
-    const { sessionDir } = await seedFixture(tempDir.path);
+    const { sessionDir, config } = await seedFixture(tempDir.path);
     const logSpy = spyOn(console, "log").mockImplementation(() => undefined);
     const errors: string[] = [];
     const errorSpy = spyOn(console, "error").mockImplementation((line: string) => {
       errors.push(line);
     });
     try {
-      await refinementsCommand("ws-cli", { sessionDir, rollback: "missing-id" });
+      await refinementsCommand("ws-cli", { sessionDir, config, rollback: "missing-id" });
       expect(process.exitCode).toBe(1);
       expect(errors.join("\n")).toContain("No refinement row");
     } finally {
       logSpy.mockRestore();
       errorSpy.mockRestore();
     }
   });
+
+  it("refuses a rollback while config.json is absent instead of assuming self-ownership", async () => {
+    using tempDir = new TestTempDir("test-debug-refinements-noconfig");
+    const { sessionDir, skillFile, config } = await seedFixture(tempDir.path);
+    const lines: string[] = [];
+    const logSpy = spyOn(console, "log").mockImplementation((line: string) => {
+      lines.push(line);
+    });
+    const errors: string[] = [];
+    const errorSpy = spyOn(console, "error").mockImplementation((line: string) => {
+      errors.push(line);
+    });
+    try {
+      await refinementsCommand("ws-cli", { sessionDir });
+      const rowId = lines[0].split("  ")[0];
+      await fsPromises.rm(path.join(tempDir.path, "config.json"));
+      await refinementsCommand("ws-cli", { sessionDir, config, rollback: rowId });
+      expect(process.exitCode).toBe(1);
+      expect(errors.join("\n")).toContain("shared-memory ownership could not be resolved");
+      expect(
+        await fsPromises.access(skillFile).then(
+          () => true,
+          () => false
+        )
+      ).toBe(true);
+    } finally {
+      logSpy.mockRestore();
+      errorSpy.mockRestore();
+    }
+  });
 });

src/cli/debug/refinements.ts

@@ -1,5 +1,10 @@
 import * as path from "path";
-import { defaultConfig } from "@/node/config";
+import { defaultConfig, type Config } from "@/node/config";
+import {
+  resolveSharedWorkspaceMemoryTopology,
+  type SharedWorkspaceMemoryTopology,
+} from "@/node/services/memoryWorkspaceOwner";
+import { getErrorMessage } from "@/common/utils/errors";
 import {
   MemoryRefinementActionSchema,
   RollbackRefinementActionSchema,
@@ -37,6 +42,8 @@ export interface RefinementsCommandOptions {
   force?: boolean;
   /** Test seam: bypass ~/.mux session resolution for fixture sessions. */
   sessionDir?: string;
+  /** Test seam: the config whose task tree resolves shared-memory ownership. */
+  config?: Pick<Config, "loadExistingConfigOrThrow" | "sessionsDir">;
 }
 
 /**
@@ -50,8 +57,32 @@ export async function refinementsCommand(
   const sessionDir = opts.sessionDir ?? path.join(defaultConfig.sessionsDir, workspaceId);
 
   if (opts.rollback !== undefined) {
+    // Sub-agents journal workspace-scope rows that point into the owner's
+    // session dir; admit that root the same way the in-app tool does, from a
+    // config that must EXIST and read: a tolerant (or fresh-install) view
+    // would resolve a sub-agent to ITSELF, and the rollback would then mutate
+    // its hidden legacy notebook (no owner root, no adoption remap) and
+    // report success. Throws → the command fails before touching anything.
+    const config = opts.config ?? defaultConfig;
+    let topology: SharedWorkspaceMemoryTopology;
+    try {
+      topology = resolveSharedWorkspaceMemoryTopology(config, workspaceId);
+    } catch (error) {
+      console.error(
+        `Refusing rollback of '${opts.rollback}': shared-memory ownership could not be resolved (${getErrorMessage(error)})`
+      );
+      process.exitCode = 1;
+      return;
+    }
     const result = await rollbackRefinement({
       sessionDir,
+      sharedWorkspaceMemorySessionDir: topology.ownerSessionDir,
+      // Reloaded per check (plan-time and in-lock), not from the snapshot
+      // above: a live backend may register a new tree member while this
+      // process waits for the shared-store lock, and its rows must count.
+      // Same existence-requiring load: an unproven tree refuses the rollback.
+      listSharedWorkspaceMemoryPeerSessionDirs: () =>
+        resolveSharedWorkspaceMemoryTopology(config, workspaceId).peerSessionDirs,
       id: opts.rollback,
       force: opts.force,
       evidence: { toolName: "debug-cli", actor: "user" },

src/common/constants/memory.ts

@@ -9,7 +9,8 @@
  *                per-project notes; never committed to the repo, survives
  *                workspaces; carried by the settings backup only when the
  *                user opts into the project bundle)
- * - workspace -> <sessionDir>/memory/ (host-local, deleted with the workspace)
+ * - workspace -> <sessionDir>/memory/ of the task-tree OWNER (host-local, deleted
+ *                with that workspace; sub-agents share their parent's store)
  */
 
 /** Virtual root prefix all memory paths are expressed under. */
@@ -18,6 +19,14 @@ export const MEMORY_VIRTUAL_ROOT = "/memories";
 export const MEMORY_SCOPES = ["global", "project", "workspace"] as const;
 export type MemoryScope = (typeof MEMORY_SCOPES)[number];
 
+/**
+ * `<sessionsDir>/<owner>/memory.revision`: opaque token rewritten on every
+ * mutation of that owner's shared `/memories/workspace` store. Sessions and
+ * Memory tabs in OTHER backend processes (multi-instance) compare it before
+ * reusing a cached index/hot set — in-process consumers get change events.
+ */
+export const WORKSPACE_MEMORY_REVISION_FILE_NAME = "memory.revision";
+
 export type MemoryAccessLevel = "read" | "readwrite";
 
 /**

src/common/orpc/schemas/memory.ts

@@ -96,10 +96,12 @@ export type MemoryConsolidationRecordPayload = z.infer<typeof MemoryConsolidatio
 
 export const CompactionCompletionMetadataSchema = z.object({
   workspaceId: z.string(),
+  workspaceMemoryWritable: z.boolean().optional(),
   summaryMessageId: z.string(),
   summaryHistorySequence: z.number(),
   compactionEpoch: z.number(),
   previousBoundaryHistorySequence: z.number().optional(),
+  closingPolicyEpoch: z.number().optional(),
   compactionRequestMessageId: z.string(),
   // RLM keep-recent floor: preserved-tail copies appended after the boundary.
   preservedTailMessageCount: z.number().optional(),
@@ -115,6 +117,12 @@ export const MemoryHarvestRecordSchema = z.object({
   acceptedCandidates: z.number(),
   skippedCandidates: z.number(),
   error: z.string().optional(),
+  /**
+   * Terminal refusal (policy unknown/read-only, or a turn of the epoch never
+   * recorded its policy): unlike an exhausted failure, the epoch's owner
+   * notebook must not be swept on its behalf either.
+   */
+  refused: z.boolean().optional(),
   usage: z.object({ inputTokens: z.number(), outputTokens: z.number() }).optional(),
   completionMetadata: CompactionCompletionMetadataSchema.optional(),
 });

src/common/schemas/project.ts

@@ -105,6 +105,14 @@ 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).",
   }),
+  workspaceMemoryWritableByEpoch: z.record(z.string(), z.boolean()).optional().meta({
+    description:
+      "Whether this workspace's agent may write /memories/workspace, accumulated (fail-closed AND over its normal turns) per compaction epoch — keyed by the history sequence of the durable context boundary that opened the epoch (-1 before any boundary); only the newest few epochs are kept. Persisted so a post-compaction memory harvest that resumes in a fresh session (restart, recovery) or observes a closing epoch while another backend already records the next one still knows the policy; harvest fails closed when unknown.",
+  }),
+  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/common/types/compaction.ts

@@ -1,9 +1,25 @@
 export interface CompactionCompletionMetadata {
   workspaceId: string;
+  /**
+   * Whether the workspace's agent may write `/memories/workspace` (its memory
+   * access policy on the last non-compaction turn). Post-compaction harvest
+   * writes candidates into that store — for a sub-agent, the OWNER's shared
+   * notebook — so a read-only (explore-like) agent must not harvest. Absent on
+   * legacy records; only an explicit `false` refuses.
+   */
+  workspaceMemoryWritable?: boolean;
   summaryMessageId: string;
   summaryHistorySequence: number;
   compactionEpoch: number;
   previousBoundaryHistorySequence?: number;
+  /**
+   * The workspace-memory policy epoch the compacted rows belonged to
+   * (workspaceMemoryPolicyEpochOf over the compacted history): the key their
+   * turns recorded under and the one the harvest and the policy carry read.
+   * Absent on records persisted by builds before the history segment stamp
+   * (compactionClosingPolicyEpoch derives their legacy identity).
+   */
+  closingPolicyEpoch?: number;
   compactionRequestMessageId: string;
   /**
    * RLM keep-recent floor: number of preserved-tail copies appended after the

src/common/types/durableEvent.ts

@@ -87,6 +87,45 @@ export const RefinementDataSchema = z.object({
   evidence: JsonValueSchema.optional(),
   /** Envelope `id` of the entry this one rolls back. */
   rollbackOf: z.string().optional(),
+  /**
+   * Stable source identity (`<workspaceId>:<row id>`) when this row was
+   * copied from a removed sub-agent's journal into its memory owner's
+   * (sharedMemoryRowMigration.ts); lets a retried migration skip it.
+   */
+  migratedFrom: z.string().optional(),
+  /**
+   * Where a migrated row was ORIGINALLY appended: the journal (workspace id)
+   * whose append sequence positions it, and its `seq` there. Two rows of one
+   * origin were serialized by that store's mutation lock (clock + append run
+   * inside it), so their origin sequence IS their mutation order — even when
+   * neither carries a usable clock value (pre-sharing history, a failed clock
+   * write). Copies are appended to the owner journal later than they happened
+   * and in migration order, so their own `seq` is no order evidence; a
+   * copy-of-copy keeps the first origin. Absent on a copy (older builds,
+   * corruption) = order unknown, never the copying journal's position.
+   * Native rows need no fields: their origin is (`workspaceId`, `seq`).
+   */
+  originJournal: z.string().optional(),
+  originSeq: z.number().optional(),
+  /**
+   * Cross-session order key for rollback conflict detection (`sourceTs ?? ts`):
+   * a shared workspace store's monotonic clock, advanced under the store's
+   * mutation lock by every mutation (workspaceMemoryRevision.ts), so rows in
+   * an owner's and its sub-agents' journals — whose `ts`/`seq` are not
+   * comparable — still order totally; migrated rows keep their source value.
+   * Persisted rows are raw JSON: only a value isValidSourceClock accepts is
+   * order evidence; a present value outside that domain reads as order
+   * unknown (refinementRollback.ts), never as "earlier than everything".
+   */
+  sourceTs: z.number().optional(),
+  /**
+   * The mutation landed but the shared store's clock write failed, so this
+   * row has NO defensible position relative to other rows (its `ts`/`seq`
+   * are journal-local). Rollback conflict detection treats such a row as
+   * conflicting with every overlapping row in either direction (force
+   * overrides), instead of ordering it by an incomparable timestamp.
+   */
+  orderUnknown: z.literal(true).optional(),
   /** Expected post-action file hashes (RefinementPostStateSchema in refinement.ts). */
   postState: JsonValueSchema.optional(),
   /**
@@ -176,3 +215,52 @@ type DistributiveOmit<T, K extends PropertyKey> = T extends unknown ? Omit<T, K>
 export type DurableEventDraft = DistributiveOmit<DurableEvent, "v" | "seq" | "id" | "ts"> & {
   id?: string;
 };
+
+/**
+ * A usable shared-store clock value: the clock is `max(Date.now(), prev + 1)`
+ * (workspaceMemoryRevision.ts), so a genuine value is a positive safe integer.
+ * Zero, negatives, fractions, unsafe integers and non-numbers are corruption.
+ */
+export function isValidSourceClock(value: unknown): value is number {
+  return typeof value === "number" && Number.isSafeInteger(value) && value > 0;
+}
+
+/** Where a refinement row was originally appended (see `originJournal`). */
+export interface RefinementRowOrigin {
+  journal: string;
+  seq: number;
+}
+
+/**
+ * The journal position that orders a refinement row against rows of the same
+ * origin (RefinementDataSchema.originJournal). A native row is positioned by
+ * the journal it was READ from (`journalWorkspaceId`, the session's workspace
+ * id — the caller's knowledge, never the row's own `workspaceId`, which is
+ * persisted data a corrupt row could carry equal to another journal's; r79)
+ * and only while the row agrees with it. A migrated copy is positioned only
+ * by a carried, well-formed origin. Anything else is no order evidence (null).
+ */
+export function refinementRowOrigin(
+  row: {
+    workspaceId: string;
+    seq: number;
+    data: { migratedFrom?: string; originJournal?: unknown; originSeq?: unknown };
+  },
+  journalWorkspaceId: string | undefined
+): RefinementRowOrigin | null {
+  if (row.data.migratedFrom === undefined) {
+    if (journalWorkspaceId === undefined || row.workspaceId !== journalWorkspaceId) return null;
+    return { journal: journalWorkspaceId, seq: row.seq };
+  }
+  const { originJournal, originSeq } = row.data;
+  if (
+    typeof originJournal !== "string" ||
+    originJournal === "" ||
+    typeof originSeq !== "number" ||
+    !Number.isSafeInteger(originSeq) ||
+    originSeq < 0
+  ) {
+    return null;
+  }
+  return { journal: originJournal, seq: originSeq };
+}

src/common/types/message.ts

@@ -963,7 +963,31 @@ export interface ContextBudgetRejectedMessage {
 export interface MuxMetadata {
   /** Highest persisted history sequence included in the provider request that produced this assistant. */
   requestHistorySequence?: number;
+  /**
+   * The compaction epoch (opening boundary's history sequence, -1 before any
+   * boundary) under which the turn that produced this assistant row recorded
+   * its workspace-memory write policy, before the row was appended
+   * (TurnRequestBuilder start()). The post-compaction harvest accepts a turn
+   * only when this matches the epoch being harvested: a turn started before
+   * a destructive reset but appended after the new boundary carries the old
+   * epoch, whose deny the reset discarded. Builds that do not maintain the
+   * policy (older ones, after a downgrade) leave it unset, so their turns'
+   * user rows are never taken as accounted for (memoryConsolidationService.ts).
+   */
+  workspaceMemoryPolicyEpoch?: number;
   historySequence?: number; // Assigned by backend for global message ordering (required when writing to history)
+  /**
+   * Start sequence of the history segment this row was appended in
+   * (HistoryService, `history-segment.json`). A full clear opens a new
+   * segment whose sequences continue above every sequence the cleared
+   * history ever used, so a boundary's history sequence — and the
+   * boundary-less epoch identity `-(segmentStart + 1)` derived from this
+   * stamp (workspaceMemoryPolicyEpochOf) — never recurs across destructive
+   * clears: a turn started before a clear cannot be mistaken for one of the
+   * segment that replaced it. Omitted in the first segment (start 0), which
+   * legacy rows without the stamp belong to as well.
+   */
+  historySegment?: number;
   /** Provider step boundaries in parts, persisted so continuous compaction can keep complete steps. */
   stepStartPartIndices?: number[];
   duration?: number;
@@ -1081,6 +1105,18 @@ export interface MuxMetadata {
    */
   rlmPreservedTailCopy?: boolean;
 
+  /**
+   * Compaction epoch (opening boundary's history sequence, -1 before any) the
+   * copied row was ORIGINALLY produced under — carried unchanged through
+   * repeated copies, so a copy of a copy still names the first epoch. The
+   * workspace-memory write policy of that epoch applies to the copy
+   * (TurnRequestBuilder → WorkspaceService.recordWorkspaceMemoryWritable):
+   * derived from history it would be wrong whenever the active-epoch read
+   * holds only the newest boundary. Absent on copies written before the
+   * field existed, whose source epochs never had a policy record either.
+   */
+  rlmPreservedTailSourcePolicyEpoch?: number;
+
   /**
    * @file mention snapshot token(s) this message provides content for.
    * Marks send-time materialized snapshot rows (the only @mention expansion