refactor(daemon): move singletons into per-daemon layers

src/events.ts

@@ -144,6 +144,8 @@ export const RunEventPayload = Schema.TaggedUnion({
     /** The run currently holding `key` (non-terminal — it is holding). */
     holderRunId: Schema.String,
     childWorkflowId: Schema.String,
+    /** Human-readable reason, naming the held key and its holder. */
+    error: Schema.optional(Schema.String),
   },
 
   // ---- ctx.agent(name, prompt, opts?) --------------------------------

src/lib/dedupe.ts

@@ -58,6 +58,3 @@ export function createDedupeRegistry(): DedupeRegistry {
     },
   };
 }
-
-/** The daemon's shared registry — every run start and dispatch claim in this process goes through it. */
-export const dedupeRegistry: DedupeRegistry = createDedupeRegistry();

src/lib/workspace.test.ts

@@ -3,7 +3,7 @@ import { tmpdir } from "node:os";
 import { join } from "node:path";
 import { describe, expect, test } from "bun:test";
 import type { ExecResult } from "./exec";
-import { allocateWorkspace, evictOldWorkspaces } from "./workspace";
+import { allocateWorkspace, createRefreshGates, evictOldWorkspaces } from "./workspace";
 import { writeBack } from "./writeback";
 import type { GitIdentity } from "./clone";
 import { hostExec } from "./exec";
@@ -29,6 +29,8 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-a",
       workspaceRoot,
       sshUrl: seed,
@@ -53,6 +55,8 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-one",
       workspaceRoot,
       sshUrl: seed,
@@ -65,6 +69,8 @@ describe("allocateWorkspace (D28)", () => {
     await Bun.$`git -C ${seed} -c user.name=seed -c user.email=seed@seed.local commit -q -m second`.quiet();
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-two",
       workspaceRoot,
       sshUrl: seed,
@@ -83,6 +89,8 @@ describe("allocateWorkspace (D28)", () => {
 
     for (const runId of ["run-1", "run-2", "run-3"]) {
       await allocateWorkspace({
+        refreshGates: createRefreshGates(),
+
         runId,
         workspaceRoot,
         sshUrl: seed,
@@ -103,20 +111,23 @@ describe("allocateWorkspace (D28)", () => {
     const seed = join(root, "seed-repo");
     await seedRepo(seed, "one");
 
+    const sharedGates = createRefreshGates();
     const [a, b] = await Promise.all([
       allocateWorkspace({
         runId: "run-x",
         workspaceRoot,
         sshUrl: seed,
         identity: IDENTITY,
         retainedWorkspaces: 10,
+        refreshGates: sharedGates,
       }),
       allocateWorkspace({
         runId: "run-y",
         workspaceRoot,
         sshUrl: seed,
         identity: IDENTITY,
         retainedWorkspaces: 10,
+        refreshGates: sharedGates,
       }),
     ]);
     expect(existsSync(join(a, "seed.txt"))).toBe(true);
@@ -132,6 +143,8 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-a",
       workspaceRoot,
       sshUrl: seed,
@@ -159,6 +172,8 @@ describe("allocateWorkspace (D28)", () => {
     await Bun.$`git -C ${seed} push -q ${remote} main`.quiet();
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-a",
       workspaceRoot,
       sshUrl: remote,
@@ -203,6 +218,8 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-1",
       workspaceRoot,
       sshUrl: seed,
@@ -211,6 +228,8 @@ describe("allocateWorkspace (D28)", () => {
     });
     await Bun.sleep(5);
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-2",
       workspaceRoot,
       sshUrl: seed,
@@ -220,6 +239,8 @@ describe("allocateWorkspace (D28)", () => {
     await Bun.sleep(5);
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-3",
       workspaceRoot,
       sshUrl: seed,
@@ -241,6 +262,8 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-1",
       workspaceRoot,
       sshUrl: seed,
@@ -249,6 +272,8 @@ describe("allocateWorkspace (D28)", () => {
     });
     await Bun.sleep(5);
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-2",
       workspaceRoot,
       sshUrl: seed,
@@ -258,6 +283,8 @@ describe("allocateWorkspace (D28)", () => {
     await Bun.sleep(5);
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-3",
       workspaceRoot,
       sshUrl: seed,
@@ -280,6 +307,8 @@ describe("scratch workspaces (issue #13)", () => {
     await seedRepo(seed, "one");
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-scratch",
       workspaceRoot,
       sshUrl: seed,
@@ -304,6 +333,8 @@ describe("scratch workspaces (issue #13)", () => {
     // Two failed-run scratch leftovers, plus one clone.
     for (const runId of ["scratch-1", "scratch-2"]) {
       await allocateWorkspace({
+        refreshGates: createRefreshGates(),
+
         runId,
         workspaceRoot,
         sshUrl: seed,
@@ -313,6 +344,8 @@ describe("scratch workspaces (issue #13)", () => {
       });
     }
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "clone-1",
       workspaceRoot,
       sshUrl: seed,
@@ -367,6 +400,8 @@ describe("host exec injection (issue #13)", () => {
     // `hostExec` directly, `git clone` would really run and contradict it.
     const exit0: ExecResult = { command: "fake", exitCode: 0, stdout: "", stderr: "" };
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
+
       runId: "run-fake",
       workspaceRoot: workspaceRoot2,
       sshUrl: seed,

src/lib/workspace.ts

@@ -31,6 +31,21 @@ import type { WorkspaceKind } from "../workflow";
 
 const MIRROR_DIR = ".mirror.git";
 
+export interface RefreshGates {
+  get(mirrorPath: string): Promise<void> | undefined;
+  set(mirrorPath: string, promise: Promise<void>): void;
+}
+
+export function createRefreshGates(): RefreshGates {
+  const gates = new Map<string, Promise<void>>();
+  return {
+    get: (mirrorPath) => gates.get(mirrorPath),
+    set: (mirrorPath, promise) => {
+      gates.set(mirrorPath, promise);
+    },
+  };
+}
+
 export interface WorkspaceAllocationInput {
   readonly runId: string;
   readonly workspaceRoot: string;
@@ -56,11 +71,15 @@ export interface WorkspaceAllocationInput {
   readonly protectedEntries?: ReadonlyArray<string>;
   /** Injectable host-exec seam; defaults to `hostExec` (no behaviour change). */
   readonly exec?: ExecFn;
+  /** Per-daemon refresh gates for serializing mirror maintenance. */
+  readonly refreshGates: RefreshGates;
 }
 
-const refreshGates = new Map<string, Promise<void>>();
-
-function enqueueRefresh(mirrorPath: string, task: () => Promise<void>): Promise<void> {
+function enqueueRefresh(
+  refreshGates: RefreshGates,
+  mirrorPath: string,
+  task: () => Promise<void>,
+): Promise<void> {
   const prior = refreshGates.get(mirrorPath) ?? Promise.resolve();
   const next = prior.then(task, task);
   refreshGates.set(mirrorPath, next);
@@ -110,7 +129,9 @@ export async function allocateWorkspace(input: WorkspaceAllocationInput): Promis
 
   const mirrorPath = join(workspaceRoot, MIRROR_DIR);
 
-  await enqueueRefresh(mirrorPath, () => refreshMirror(mirrorPath, sshUrl, exec));
+  await enqueueRefresh(input.refreshGates, mirrorPath, () =>
+    refreshMirror(mirrorPath, sshUrl, exec),
+  );
 
   if (existsSync(dir)) {
     await rm(dir, { recursive: true, force: true });

src/runtime/run.ts

@@ -361,6 +361,7 @@ export async function startRun<I, O>(
           key: err.key,
           holderRunId: err.holderRunId,
           childWorkflowId: child.id,
+          error: domainErrorMessage(err),
         });
       }
       throw err;

src/server/concurrency.test.ts

@@ -22,6 +22,10 @@ import { createSlowFakeAdapter } from "../replay/adapter";
 import { makeAgentRuntime } from "../runtime/agent-runtime";
 import { admitRun } from "./admission";
 import { serve } from "./http";
+import { createRunRegistry } from "./runs";
+import { createPubSub } from "./pubsub";
+import { createDedupeRegistry } from "../lib/dedupe";
+import { createRefreshGates } from "../lib/workspace";
 
 const TREE_WORKFLOW = `${import.meta.dir}/../../test/fixtures/tree-workflow.ts`;
 
@@ -71,7 +75,14 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => {
 
     const workspaceRoot = join(root, "workspaces");
     const db = openStore(join(root, "factory.db"));
+    const services = {
+      registry: createRunRegistry(),
+      pubsub: createPubSub(),
+      dedupeRegistry: createDedupeRegistry(),
+      refreshGates: createRefreshGates(),
+    };
     const server = serve({
+      services,
       db,
       runtime: makeAgentRuntime(
         createSlowFakeAdapter(
@@ -160,7 +171,14 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => {
     await Bun.$`git -C ${seed} -c user.name=seed -c user.email=seed@seed.local commit -q --allow-empty -m seed`.quiet();
 
     const db = openStore(join(root, "factory.db"));
+    const services = {
+      registry: createRunRegistry(),
+      pubsub: createPubSub(),
+      dedupeRegistry: createDedupeRegistry(),
+      refreshGates: createRefreshGates(),
+    };
     const server = serve({
+      services,
       db,
       runtime: makeAgentRuntime(
         createSlowFakeAdapter(
@@ -222,7 +240,14 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => {
 
     const workspaceRoot = join(root, "workspaces");
     const db = openStore(join(root, "factory.db"));
+    const services = {
+      registry: createRunRegistry(),
+      pubsub: createPubSub(),
+      dedupeRegistry: createDedupeRegistry(),
+      refreshGates: createRefreshGates(),
+    };
     const server = serve({
+      services,
       db,
       runtime: makeAgentRuntime(
         createSlowFakeAdapter(