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

src/cli-commands.ts

@@ -83,8 +83,12 @@ export const serveCommand = Command.make("serve", serveConfig, (config) =>
           : "scheduler: none (no schedules in config)",
       ),
     );
-    // The daemon's shutdown path: stop the scheduler and server and dispose
-    // the agent runtime before exiting, instead of dying with them live.
+    // The daemon's shutdown path: stop the scheduler, cancel every active run
+    // and wait (bounded) for each to record `RunCancelled` and kill its
+    // `ctx.exec` children, then stop the server and dispose the runtime —
+    // instead of dying with them live and leaving orphaned processes and
+    // "interrupted" runs behind. `once`: a second signal falls through to the
+    // default handler and kills the process outright.
     yield* Effect.sync(() => {
       const shutdown = (signal: NodeJS.Signals) => {
         void handle.stop().finally(() => process.exit(signal === "SIGINT" ? 130 : 143));

src/lib/dedupe.ts

@@ -16,9 +16,15 @@
  * again — a dead run must not hold a key forever. Release happens the moment
  * the holding run reaches any terminal state (including cancelled/failed);
  * see `startTrackedRun` in server/runs.ts.
+ *
+ * #38: "per-process" means per daemon. The registry is a `Context.Service`
+ * whose layer builds a fresh map each time a daemon runtime is built
+ * (`server/daemon-runtime.ts`), so two daemons in one process never share
+ * keys. `createDedupeRegistry` stays exported for unit tests of the plain
+ * map logic.
  */
 
-import { Schema } from "effect";
+import { Context, Layer, Schema } from "effect";
 
 export class DedupeKeyError extends Schema.TaggedError<DedupeKeyError>()("DedupeKeyError", {
   key: Schema.String,
@@ -44,7 +50,7 @@ export class DedupeKeyError extends Schema.TaggedError<DedupeKeyError>()("Dedupe
   }
 }
 
-export interface DedupeRegistry {
+export interface DedupeRegistryShape {
   /** The run id currently holding `key`, or `undefined`. */
   holderOf(key: string): string | undefined;
   /**
@@ -57,7 +63,7 @@ export interface DedupeRegistry {
   release(key: string, runId: string): void;
 }
 
-export function createDedupeRegistry(): DedupeRegistry {
+export function createDedupeRegistry(): DedupeRegistryShape {
   const held = new Map<string, string>();
   return {
     holderOf: (key) => held.get(key),
@@ -73,5 +79,16 @@ 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();
+/**
+ * The dedupe registry as a daemon service (#38, ADR 0009 §5). Resolve it from
+ * the daemon runtime's context; provide it with `DedupeRegistryLayer`.
+ */
+export class DedupeRegistry extends Context.Service<DedupeRegistry, DedupeRegistryShape>()(
+  "DedupeRegistry",
+) {}
+
+/** A fresh, empty registry per layer build — i.e. per daemon runtime. */
+export const DedupeRegistryLayer: Layer.Layer<DedupeRegistry> = Layer.sync(
+  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,7 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-a",
       workspaceRoot,
       sshUrl: seed,
@@ -53,6 +54,7 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-one",
       workspaceRoot,
       sshUrl: seed,
@@ -65,6 +67,7 @@ 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 +86,7 @@ describe("allocateWorkspace (D28)", () => {
 
     for (const runId of ["run-1", "run-2", "run-3"]) {
       await allocateWorkspace({
+        refreshGates: createRefreshGates(),
         runId,
         workspaceRoot,
         sshUrl: seed,
@@ -103,20 +107,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 +139,7 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-a",
       workspaceRoot,
       sshUrl: seed,
@@ -159,6 +167,7 @@ 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 +212,7 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-1",
       workspaceRoot,
       sshUrl: seed,
@@ -211,6 +221,7 @@ describe("allocateWorkspace (D28)", () => {
     });
     await Bun.sleep(5);
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-2",
       workspaceRoot,
       sshUrl: seed,
@@ -220,6 +231,7 @@ describe("allocateWorkspace (D28)", () => {
     await Bun.sleep(5);
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-3",
       workspaceRoot,
       sshUrl: seed,
@@ -241,6 +253,7 @@ describe("allocateWorkspace (D28)", () => {
     await seedRepo(seed, "one");
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-1",
       workspaceRoot,
       sshUrl: seed,
@@ -249,6 +262,7 @@ describe("allocateWorkspace (D28)", () => {
     });
     await Bun.sleep(5);
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-2",
       workspaceRoot,
       sshUrl: seed,
@@ -258,6 +272,7 @@ describe("allocateWorkspace (D28)", () => {
     await Bun.sleep(5);
 
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-3",
       workspaceRoot,
       sshUrl: seed,
@@ -280,6 +295,7 @@ describe("scratch workspaces (issue #13)", () => {
     await seedRepo(seed, "one");
 
     const dir = await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "run-scratch",
       workspaceRoot,
       sshUrl: seed,
@@ -304,6 +320,7 @@ 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 +330,7 @@ describe("scratch workspaces (issue #13)", () => {
       });
     }
     await allocateWorkspace({
+      refreshGates: createRefreshGates(),
       runId: "clone-1",
       workspaceRoot,
       sshUrl: seed,
@@ -367,6 +385,7 @@ 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

@@ -22,6 +22,7 @@
  * non-host workspace is plumbing or a rewrite.
  */
 
+import { Context, Layer } from "effect";
 import { existsSync, readdirSync, statSync } from "node:fs";
 import { mkdir, rm } from "node:fs/promises";
 import { join } from "node:path";
@@ -31,6 +32,40 @@ import type { WorkspaceKind } from "../workflow";
 
 const MIRROR_DIR = ".mirror.git";
 
+/**
+ * The mirror-refresh promise queue, keyed by mirror path (see the
+ * concurrency note above). #38: per daemon, not per module — it is the
+ * `RefreshGates` service below, and `allocateWorkspace` takes the resolved
+ * value as a plain argument.
+ */
+export interface RefreshGatesShape {
+  get(mirrorPath: string): Promise<void> | undefined;
+  set(mirrorPath: string, promise: Promise<void>): void;
+}
+
+export function createRefreshGates(): RefreshGatesShape {
+  const gates = new Map<string, Promise<void>>();
+  return {
+    get: (mirrorPath) => gates.get(mirrorPath),
+    set: (mirrorPath, promise) => {
+      gates.set(mirrorPath, promise);
+    },
+  };
+}
+
+/**
+ * The refresh gates as a daemon service (#38, ADR 0009 §5); provided by
+ * `RefreshGatesLayer`, one fresh queue per daemon runtime.
+ */
+export class RefreshGates extends Context.Service<RefreshGates, RefreshGatesShape>()(
+  "RefreshGates",
+) {}
+
+export const RefreshGatesLayer: Layer.Layer<RefreshGates> = Layer.sync(
+  RefreshGates,
+  createRefreshGates,
+);
+
 export interface WorkspaceAllocationInput {
   readonly runId: string;
   readonly workspaceRoot: string;
@@ -56,11 +91,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: RefreshGatesShape;
 }
 
-const refreshGates = new Map<string, Promise<void>>();
-
-function enqueueRefresh(mirrorPath: string, task: () => Promise<void>): Promise<void> {
+function enqueueRefresh(
+  refreshGates: RefreshGatesShape,
+  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 +149,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/agent-step.test.ts

@@ -2,7 +2,7 @@ import { Effect, ManagedRuntime } from "effect";
 import { describe, expect, test } from "bun:test";
 import type { AgentAdapterYield, AgentSignal } from "./agent-adapter";
 import { AgentRuntimeLayer } from "./agent-runtime";
-import { buildAgentStepEffect } from "./agent-step";
+import { abortableIterable, buildAgentStepEffect } from "./agent-step";
 
 function makeYield(chunk: unknown, signal?: AgentSignal): AgentAdapterYield {
   return signal !== undefined ? { chunk, signal } : { chunk };
@@ -169,3 +169,46 @@ describe("buildAgentStepEffect signal extraction", () => {
     expect(outcome.chunkCount).toBe(1);
   });
 });
+
+describe("abortableIterable (#38: abandoning a step is immediate)", () => {
+  test("a stream that ends on its own is passed through and never aborted", async () => {
+    const controller = new AbortController();
+    async function* source() {
+      yield 1;
+      yield 2;
+    }
+    const { iterable, teardown } = abortableIterable(source(), controller);
+    const seen: Array<number> = [];
+    for await (const n of iterable) seen.push(n);
+    await teardown();
+    expect(seen).toEqual([1, 2]);
+    expect(controller.signal.aborted).toBe(false);
+  });
+
+  test("return() during a pending next() resolves at once, aborts, and tears down in the background", async () => {
+    const controller = new AbortController();
+    let finalized = false;
+    async function* source() {
+      try {
+        yield 1;
+        // Waiting on the model: only the abort lets this generator move on.
+        await new Promise<void>((resolve) =>
+          controller.signal.addEventListener("abort", () => setTimeout(resolve, 50)),
+        );
+      } finally {
+        finalized = true;
+      }
+    }
+    const { iterable, teardown } = abortableIterable(source(), controller);
+    const iterator = iterable[Symbol.asyncIterator]();
+    expect(await iterator.next()).toEqual({ done: false, value: 1 });
+    const pending = iterator.next();
+
+    expect(await iterator.return!()).toEqual({ done: true, value: undefined });
+    expect(controller.signal.aborted).toBe(true);
+    expect(await pending).toEqual({ done: true, value: undefined });
+    expect(finalized).toBe(false);
+    await teardown();
+    expect(finalized).toBe(true);
+  });
+});