diff --git a/src/cli-commands.ts b/src/cli-commands.ts index b1d2b95..0e03f67 100644 --- a/src/cli-commands.ts +++ b/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)); diff --git a/src/lib/dedupe.ts b/src/lib/dedupe.ts index b10f7b1..41a35c9 100644 --- a/src/lib/dedupe.ts +++ b/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", { key: Schema.String, @@ -44,7 +50,7 @@ export class DedupeKeyError extends Schema.TaggedError()("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(); 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", +) {} + +/** A fresh, empty registry per layer build — i.e. per daemon runtime. */ +export const DedupeRegistryLayer: Layer.Layer = Layer.sync( + DedupeRegistry, + createDedupeRegistry, +); diff --git a/src/lib/workspace.test.ts b/src/lib/workspace.test.ts index dfa8305..2a107bf 100644 --- a/src/lib/workspace.test.ts +++ b/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,6 +107,7 @@ 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", @@ -110,6 +115,7 @@ describe("allocateWorkspace (D28)", () => { sshUrl: seed, identity: IDENTITY, retainedWorkspaces: 10, + refreshGates: sharedGates, }), allocateWorkspace({ runId: "run-y", @@ -117,6 +123,7 @@ describe("allocateWorkspace (D28)", () => { 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, diff --git a/src/lib/workspace.ts b/src/lib/workspace.ts index 94008f9..d2868df 100644 --- a/src/lib/workspace.ts +++ b/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 | undefined; + set(mirrorPath: string, promise: Promise): void; +} + +export function createRefreshGates(): RefreshGatesShape { + const gates = new Map>(); + 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", +) {} + +export const RefreshGatesLayer: Layer.Layer = Layer.sync( + RefreshGates, + createRefreshGates, +); + export interface WorkspaceAllocationInput { readonly runId: string; readonly workspaceRoot: string; @@ -56,11 +91,15 @@ export interface WorkspaceAllocationInput { readonly protectedEntries?: ReadonlyArray; /** 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>(); - -function enqueueRefresh(mirrorPath: string, task: () => Promise): Promise { +function enqueueRefresh( + refreshGates: RefreshGatesShape, + mirrorPath: string, + task: () => Promise, +): Promise { 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 }); diff --git a/src/runtime/agent-step.test.ts b/src/runtime/agent-step.test.ts index 9afe22e..eac7f0c 100644 --- a/src/runtime/agent-step.test.ts +++ b/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 = []; + 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((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); + }); +}); diff --git a/src/runtime/agent-step.ts b/src/runtime/agent-step.ts index 5c9dfb1..4df2d64 100644 --- a/src/runtime/agent-step.ts +++ b/src/runtime/agent-step.ts @@ -19,6 +19,21 @@ * wiring even though `Stream.fromAsyncIterable`'s implicit `.return()` on * scope closure was sufficient in the tested case — cheap, never harmful, * covers the untested non-cooperative-abort case. + * + * WHY THE ADAPTER STREAM IS WRAPPED (`abortableIterable`, #38 shutdown): the + * implicit `.return()` above is a scope finalizer, and an async generator's + * `return()` queues behind its pending `next()`. Interrupting a step that was + * waiting on the model therefore blocked until opencode's *next* chunk — and + * the `onInterrupt` abort, which runs after that finalizer, never got the + * chance to hurry it along. A cancel took tens of seconds, and a shutdown + * outlived its budget. The wrapper makes the stream end the moment the step + * is abandoned: `next()` races the abort signal, and `return()` aborts first + * (an early `return()` *is* the consumer giving up), fires the adapter's own + * `return()` without awaiting it, and resolves at once. The adapter's + * teardown still runs, in the background, and `AgentStepHandle.teardown` + * lets the caller wait for it. (For opencode the process itself dies on the + * abort: the local-process sandbox kills its process group on the spawn's + * abort signal, independently of the generator's `finally`.) */ import { Effect, Schema, Stream } from "effect"; @@ -96,14 +111,84 @@ export interface AgentStepHandle { readonly effect: Effect.Effect; readonly abortController: AbortController; readonly partial: AgentStepPartial; + /** + * Settles once the adapter has torn its stream down after the step was + * abandoned (e.g. the opencode process is gone) — at once when it never + * was. The step itself does not wait for this; shutdown does. + */ + readonly teardown: () => Promise; +} + +/** + * Wrap an adapter's stream so abandoning it is immediate (see the module + * comment): `next()` resolves `done` as soon as `abortController` aborts, and + * `return()` aborts (unless the stream already finished), starts the inner + * `return()` without waiting for it, and resolves at once. `teardown()` + * settles when that inner `return()` does — i.e. when the adapter has + * actually cleaned up — or at once if the stream was never abandoned. + */ +export function abortableIterable( + iterable: AsyncIterable, + abortController: AbortController, +): { readonly iterable: AsyncIterable; readonly teardown: () => Promise } { + let teardown: Promise = Promise.resolve(); + const done: IteratorReturnResult = { done: true, value: undefined }; + const wrapped: AsyncIterable = { + [Symbol.asyncIterator](): AsyncIterator { + const inner = iterable[Symbol.asyncIterator](); + const signal = abortController.signal; + const aborted = new Promise>((resolve) => { + if (signal.aborted) resolve(done); + else signal.addEventListener("abort", () => resolve(done), { once: true }); + }); + // `innerDone`: the adapter's own stream ended — nothing to tear down. + // `released`: we abandoned it, and its `return()` is already started. + let innerDone = false; + let released = false; + const release = (): void => { + if (innerDone || released) return; + released = true; + abortController.abort(); + if (inner.return !== undefined) { + teardown = Promise.resolve(inner.return()).then( + () => undefined, + () => undefined, + ); + } + }; + return { + next: async () => { + if (innerDone || released || signal.aborted) { + release(); + return done; + } + const result = await Promise.race([ + inner.next().then((r) => { + if (r.done === true) innerDone = true; + return r; + }), + aborted, + ]); + if (result.done === true) release(); + return result; + }, + return: async () => { + release(); + return done; + }, + }; + }, + }; + return { iterable: wrapped, teardown: () => teardown }; } export function buildAgentStepEffect(options: AgentStepEffectOptions): AgentStepHandle { const abortController = new AbortController(); + let teardown: () => Promise = () => Promise.resolve(); const rawStream = Stream.unwrap( - Effect.map(Effect.service(AgentRuntime), ({ adapter }) => - Stream.fromAsyncIterable( + Effect.map(Effect.service(AgentRuntime), ({ adapter }) => { + const wrapped = abortableIterable( adapter.stream({ threadId: options.threadId, dir: options.dir, @@ -112,9 +197,14 @@ export function buildAgentStepEffect(options: AgentStepEffectOptions): AgentStep outputSchema: options.outputSchema, abortController, }), + abortController, + ); + teardown = wrapped.teardown; + return Stream.fromAsyncIterable( + wrapped.iterable, (cause) => new AgentStepChunkError({ cause }), - ), - ), + ); + }), ); const partial: AgentStepPartial = { @@ -198,5 +288,5 @@ export function buildAgentStepEffect(options: AgentStepEffectOptions): AgentStep durationMs: Date.now() - startedAt, })); - return { effect, abortController, partial }; + return { effect, abortController, partial, teardown: () => teardown() }; } diff --git a/src/runtime/run.ts b/src/runtime/run.ts index 8387371..ad5f5eb 100644 --- a/src/runtime/run.ts +++ b/src/runtime/run.ts @@ -111,9 +111,38 @@ export type RunOutcome = export interface RunHandle { readonly result: Promise>; + /** + * Resolves after `result`, once every agent step the run abandoned has + * finished tearing its adapter down — each bounded by + * `AGENT_TEARDOWN_GRACE_MS`. + * A cancelled run records `RunCancelled` without waiting for that; daemon + * shutdown waits for this so it does not exit under a live agent process. + */ + readonly settled: Promise; cancel(): Promise; } +/** + * How long a run's `settled` waits for one abandoned agent step's adapter + * teardown. Bounded because that teardown is third-party code that need not + * finish: measured live, `@tanstack/ai`'s chat engine can sit on an aborted + * opencode stream indefinitely, even though the abort itself already killed + * the opencode process (the local-process sandbox kills the process group on + * the spawn's abort signal). The grace covers a cooperative adapter's + * cleanup without letting an uncooperative one hold shutdown to its budget. + */ +export const AGENT_TEARDOWN_GRACE_MS = 1_000; + +function boundedTeardown(teardown: Promise): Promise { + return new Promise((resolve) => { + const timer = setTimeout(resolve, AGENT_TEARDOWN_GRACE_MS); + void teardown.then(() => { + clearTimeout(timer); + resolve(); + }); + }); +} + function makeIdCounter(prefix: string): () => string { let n = 0; return () => `${prefix}-${n++}`; @@ -142,6 +171,18 @@ function resolveOutput( } } +/** + * What `startRun` needs from a runtime: forking and running effects that + * require `AgentRuntime`. Any `ManagedRuntime` that provides at least the + * agent runtime satisfies it — the CLI's agent-only runtime and the daemon's + * runtime (#38), which provides more. (`ManagedRuntime` itself is invariant + * in its services, so naming the two methods is what lets both through.) + */ +export type AgentRunner = Pick< + ManagedRuntime.ManagedRuntime, + "runFork" | "runPromise" +>; + /** * `runtime` is the composition root's `ManagedRuntime` (issue #36): agent * steps resolve their adapter from its `AgentRuntime` service and their @@ -150,7 +191,7 @@ function resolveOutput( */ export function startRun( workflow: WorkflowDefinition, - runtime: ManagedRuntime.ManagedRuntime, + runtime: AgentRunner, options: StartRunOptions, ): RunHandle { let seq = 0; @@ -161,6 +202,7 @@ export function startRun( const runController = new AbortController(); let cancelled = false; let activeAgentFiber: Fiber.Fiber | null = null; + const agentTeardowns: Array> = []; runController.signal.addEventListener("abort", () => { cancelled = true; @@ -245,6 +287,7 @@ export function startRun( activeAgentFiber = fiber; const exit = await Effect.runPromise(Fiber.await(fiber)); activeAgentFiber = null; + agentTeardowns.push(boundedTeardown(handle.teardown())); const durationMs = Date.now() - stepStartedAt; if (Exit.isFailure(exit)) { @@ -526,6 +569,7 @@ export function startRun( return { result: resultPromise, + settled: resultPromise.then(() => Promise.all(agentTeardowns)).then(() => undefined), cancel: async () => { runController.abort(); await resultPromise; diff --git a/src/server/concurrency.test.ts b/src/server/concurrency.test.ts index 051774e..f2691a0 100644 --- a/src/server/concurrency.test.ts +++ b/src/server/concurrency.test.ts @@ -19,9 +19,9 @@ import { describe, expect, test } from "bun:test"; import { defineConfig } from "../config"; import { getRunEvents, openStore } from "../persistence/store"; import { createSlowFakeAdapter } from "../replay/adapter"; -import { makeAgentRuntime } from "../runtime/agent-runtime"; import { admitRun } from "./admission"; import { serve } from "./http"; +import { makeDaemonRuntime } from "./daemon-runtime"; const TREE_WORKFLOW = `${import.meta.dir}/../../test/fixtures/tree-workflow.ts`; @@ -73,7 +73,7 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => { const db = openStore(join(root, "factory.db")); const server = serve({ db, - runtime: makeAgentRuntime( + runtime: makeDaemonRuntime( createSlowFakeAdapter( [ { type: "TEXT_MESSAGE_START" }, @@ -162,7 +162,7 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => { const db = openStore(join(root, "factory.db")); const server = serve({ db, - runtime: makeAgentRuntime( + runtime: makeDaemonRuntime( createSlowFakeAdapter( [ { type: "TEXT_MESSAGE_START" }, @@ -224,7 +224,7 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => { const db = openStore(join(root, "factory.db")); const server = serve({ db, - runtime: makeAgentRuntime( + runtime: makeDaemonRuntime( createSlowFakeAdapter( [ { type: "TEXT_MESSAGE_START" }, diff --git a/src/server/daemon-runtime.ts b/src/server/daemon-runtime.ts new file mode 100644 index 0000000..f1fde25 --- /dev/null +++ b/src/server/daemon-runtime.ts @@ -0,0 +1,58 @@ +/** + * The daemon's composition root as a value (#38, ADR 0009 §5, following + * #36's `AgentRuntime`): every per-daemon singleton is a `Context.Service`, + * and `DaemonLayer` composes their layers with `AgentRuntimeLayer` into one + * `ManagedRuntime`. Building the runtime builds the state — a fresh run + * registry, pubsub, dedupe registry and refresh-gate queue — so two daemons + * are two runtimes, and a test that wants clean state builds a new one. + * + * The boundaries (HTTP handlers, the scheduler's fire function, `ctx.dispatch`) + * stay plain async per ADR 0009 §5 and resolve services from the runtime with + * `serviceOf`. Every layer here is synchronous, so `serviceOf` is a + * synchronous lookup — which is what lets `startTrackedRun` resolve its + * services and still reserve its slot with no `await` between check and set. + */ + +import { Effect, Layer, ManagedRuntime, type Context } from "effect"; +import { DedupeRegistry, DedupeRegistryLayer } from "../lib/dedupe"; +import { RefreshGates, RefreshGatesLayer } from "../lib/workspace"; +import type { AgentAdapter } from "../runtime/agent-adapter"; +import { AgentRuntime, AgentRuntimeLayer } from "../runtime/agent-runtime"; +import { RunPubSub, RunPubSubLayer } from "./pubsub"; +import { RunRegistry, RunRegistryLayer } from "./runs"; + +/** Every service a daemon runtime provides. */ +export type DaemonServices = AgentRuntime | RunRegistry | RunPubSub | DedupeRegistry | RefreshGates; + +/** The daemon's `ManagedRuntime`: the agent runtime plus its per-daemon state. */ +export type DaemonRuntime = ManagedRuntime.ManagedRuntime; + +/** The daemon's layer: `adapter` (or the opencode default) plus fresh per-daemon state. */ +export const DaemonLayer = (adapter?: AgentAdapter): Layer.Layer => + Layer.mergeAll( + AgentRuntimeLayer(adapter), + RunRegistryLayer, + RunPubSubLayer, + DedupeRegistryLayer, + RefreshGatesLayer, + ); + +/** + * Build a daemon runtime. The caller owns it and must `dispose()` it; tests + * build one per case for isolated state. + */ +export const makeDaemonRuntime = (adapter?: AgentAdapter): DaemonRuntime => + ManagedRuntime.make(DaemonLayer(adapter)); + +/** + * Resolve one of the daemon's services from its runtime, synchronously — the + * plain-async boundary's way into the context. Owners build the layers eagerly + * (`startDaemon` awaits `runtime.context()`; `createTestDaemon` resolves at + * construction), so this reads an already-built context. + */ +export function serviceOf( + runtime: DaemonRuntime, + service: Context.Key, +): S { + return runtime.runSync(Effect.service(service)); +} diff --git a/src/server/daemon.ts b/src/server/daemon.ts index 8c56f06..31fb20b 100644 --- a/src/server/daemon.ts +++ b/src/server/daemon.ts @@ -10,6 +10,22 @@ * scheduler; the adapter is resolved from that runtime's context inside * `runtime/agent-step.ts` instead of being threaded through every options * type. `DaemonHandle.stop` disposes it. + * + * #38: that runtime is the daemon runtime (`server/daemon-runtime.ts`): the + * agent runtime layer composed with the per-daemon state layers (run + * registry, pubsub, dedupe registry, refresh gates). Building it builds fresh + * state, so two daemons are two runtimes and coexist in one process without + * sharing anything. The scheduler loop is forked on it and resolves its + * dedupe registry and clock from its context. + * + * Shutdown order (`DaemonHandle.stop`, which `factory serve`'s SIGINT/SIGTERM + * handler calls): interrupt the scheduler so nothing new fires; shut the run + * registry down — refuse new starts, cancel every active run and reserved + * slot, and wait (bounded) for them to persist `RunCancelled` and kill their + * `ctx.exec` children; only then stop the HTTP server (so live SSE tails see + * the cancellations) and dispose the runtime; if the shutdown wait timed + * out, wait a further bounded moment for the cancelled runs to record their + * end, so the caller's `process.exit` does not cut off a `RunCancelled`. */ import { mkdir } from "node:fs/promises"; @@ -19,15 +35,14 @@ type AnyFiber = Fiber.Fiber; import type { FactoryConfig } from "../config"; import { openStore } from "../persistence/store"; import { serve } from "./http"; -import { type DispatchEnv, type WorkspaceSpec } from "./runs"; import { - createSchedulerState, - makeScheduleFire, - runSchedulerLoop, - toRuntimeSchedules, - type SchedulerDeps, -} from "./scheduler"; -import { makeAgentRuntime } from "../runtime/agent-runtime"; + DEFAULT_SHUTDOWN_TIMEOUT_MS, + RunRegistry, + type DispatchEnv, + type WorkspaceSpec, +} from "./runs"; +import { makeScheduleFire, runSchedulerLoop, toRuntimeSchedules } from "./scheduler"; +import { makeDaemonRuntime, serviceOf, type DaemonRuntime } from "./daemon-runtime"; export interface DaemonOptions { readonly dbPath: string; @@ -41,6 +56,11 @@ export interface DaemonOptions { * Absent, the legacy per-request `{dir, clone}` behaviour is kept. */ readonly config?: FactoryConfig; + /** + * How long `stop()` waits for cancelled runs to settle before abandoning + * them, over `DEFAULT_SHUTDOWN_TIMEOUT_MS`. + */ + readonly shutdownTimeoutMs?: number; } export interface DaemonHandle { @@ -48,12 +68,23 @@ export interface DaemonHandle { /** Issue #16: the loop that fires the config's schedules, when it has any. */ readonly schedulerFiber: AnyFiber | undefined; /** - * Shut the daemon down: interrupt the scheduler, stop the HTTP server, and - * dispose the agent runtime (issue #36). + * Issue #38: this daemon's runtime — its agent runtime plus its own + * registry, pubsub, dedupe registry and refresh gates, resolvable with + * `serviceOf`. + */ + readonly runtime: DaemonRuntime; + /** + * Shut the daemon down: interrupt the scheduler, cancel every active run + * and wait (bounded) for them to settle, stop the HTTP server, and dispose + * the daemon runtime (issue #36). Idempotent: every call returns the first + * call's promise. */ readonly stop: () => Promise; } +/** After disposal, how long `stop()` still waits for cancelled runs to record their end. */ +const SETTLE_AFTER_DISPOSE_MS = 2_000; + /** Issue #16: how often the scheduler's due window check runs. */ export const DEFAULT_SCHEDULER_INTERVAL_MS = 30_000; @@ -61,7 +92,10 @@ export async function startDaemon(options: DaemonOptions): Promise await mkdir(dirname(options.dbPath), { recursive: true }); const db = openStore(options.dbPath); - const runtime = makeAgentRuntime(options.config?.agent.adapter); + const runtime = makeDaemonRuntime(options.config?.agent.adapter); + // Build the layers now: a failing (or, one day, asynchronous) layer + // surfaces at startup, and every later `serviceOf` reads the built context. + await runtime.context(); const server = serve({ db, @@ -89,33 +123,37 @@ export async function startDaemon(options: DaemonOptions): Promise maxDispatchDepth: config.maxDispatchDepth, maxChildrenPerRun: config.maxChildrenPerRun, }; - const deps: SchedulerDeps = { - schedules: toRuntimeSchedules(config), - fire: makeScheduleFire({ - db, - runtime, - maxConcurrentRuns, - workspace, - repo, - dispatchEnv, - }), - }; - const state = createSchedulerState(deps); - return Effect.runFork( + const fire = makeScheduleFire({ + db, + runtime, + maxConcurrentRuns, + workspace, + repo, + dispatchEnv, + }); + return runtime.runFork( runSchedulerLoop( - deps, - state, + toRuntimeSchedules(config), + fire, options.schedulerIntervalMs ?? DEFAULT_SCHEDULER_INTERVAL_MS, ), ); })() : undefined; - const stop = async (): Promise => { - if (schedulerFiber !== undefined) await Effect.runPromise(Fiber.interrupt(schedulerFiber)); - await server.stop(true); - await runtime.dispose(); - }; + let stopping: Promise | undefined; + const stop = (): Promise => + (stopping ??= (async () => { + if (schedulerFiber !== undefined) await Effect.runPromise(Fiber.interrupt(schedulerFiber)); + const registry = serviceOf(runtime, RunRegistry); + await registry.shutdown(options.shutdownTimeoutMs ?? DEFAULT_SHUTDOWN_TIMEOUT_MS); + await server.stop(true); + await runtime.dispose(); + // If shutdown timed out, disposing interrupted whatever still held the + // runs; give them a short, bounded moment to persist `RunCancelled` + // before the caller (`factory serve`) exits the process. + await registry.awaitCancelled(SETTLE_AFTER_DISPOSE_MS); + })()); - return { server, schedulerFiber, stop }; + return { server, schedulerFiber, runtime, stop }; } diff --git a/src/server/dedupe.test.ts b/src/server/dedupe.test.ts index 845d8ac..934e0c4 100644 --- a/src/server/dedupe.test.ts +++ b/src/server/dedupe.test.ts @@ -4,6 +4,10 @@ * releases it when the run reaches any terminal state. Starting a second run * with a key a non-terminal run already holds throws a `DedupeKeyError` * naming the key and the holding run. + * + * #38: each test builds its own daemon runtime (`createTestDaemon`) instead + * of sharing module-level singletons. A holder is kept in its reserved-but-not-started window by + * pre-seeding the daemon's refresh gate for its workspace mirror. */ import { mkdtempSync, mkdirSync, rmSync } from "node:fs"; @@ -12,16 +16,16 @@ import { join } from "node:path"; import { describe, expect, test } from "bun:test"; import { openStore, getRunEvents } from "../persistence/store"; import { createSlowFakeAdapter } from "../replay/adapter"; -import { makeAgentRuntime } from "../runtime/agent-runtime"; import { defineWorkflow, Schema } from "../workflow"; -import { createDedupeRegistry, DedupeKeyError } from "../lib/dedupe"; -import { activeRunIds, getActiveHandle, isActive, startTrackedRun } from "./runs"; +import { DedupeKeyError } from "../lib/dedupe"; +import { startTrackedRun } from "./runs"; +import { createTestDaemon, type TestDaemon } from "./test-daemon"; import { serve } from "./http"; import { defineConfig } from "../config"; /** Fire-and-forget children keep filling the registry; drain before closing the store. */ -async function drain(timeoutMs = 10_000): Promise { - await waitFor(() => activeRunIds().length === 0, timeoutMs); +async function drain(daemon: TestDaemon, timeoutMs = 10_000): Promise { + await waitFor(() => daemon.registry.activeRunIds().length === 0, timeoutMs); } const SLOW_ADAPTER = createSlowFakeAdapter( @@ -33,8 +37,6 @@ const SLOW_ADAPTER = createSlowFakeAdapter( 25, ); -const runtime = makeAgentRuntime(SLOW_ADAPTER); - const echoWorkflow = defineWorkflow("echo-wf", { input: Schema.Struct({}), run: async (ctx) => { @@ -44,7 +46,7 @@ const echoWorkflow = defineWorkflow("echo-wf", { }); interface Gate { - readonly wait: () => Promise; + readonly promise: Promise; readonly release: () => void; } @@ -53,7 +55,7 @@ function makeGate(): Gate { const promise = new Promise((resolve) => { release = resolve; }); - return { wait: () => promise, release }; + return { promise, release }; } async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise { @@ -69,23 +71,48 @@ interface StartOpts { readonly runId?: string; readonly dir: string; readonly dedupeKey?: string; - readonly beforeStart?: () => Promise; - /** Shared holder registry when one test needs two starts to agree. */ - readonly registry?: ReturnType; + /** The daemon daemon both starts share, so they agree on holders. */ + readonly daemon: TestDaemon; + /** + * Holds the run in its reserved-but-not-started window until released: the + * run allocates a clone workspace under `dir` whose mirror refresh is + * queued behind the gate. + */ + readonly gate?: Gate; } -function start(db: ReturnType, opts: StartOpts): Promise { - return startTrackedRun(runtime, db, echoWorkflow, { - dir: opts.dir, +async function start(db: ReturnType, opts: StartOpts): Promise { + const location = + opts.gate === undefined + ? { dir: opts.dir } + : { workspace: await gatedWorkspace(opts.dir, opts.daemon, opts.gate) }; + return startTrackedRun(opts.daemon.runtime, db, echoWorkflow, { + ...location, input: {}, - dedupeRegistry: opts.registry ?? createDedupeRegistry(), maxConcurrentRuns: 4, ...(opts.runId !== undefined ? { runId: opts.runId } : {}), ...(opts.dedupeKey !== undefined ? { dedupeKey: opts.dedupeKey } : {}), - ...(opts.beforeStart !== undefined ? { beforeStart: opts.beforeStart } : {}), }); } +async function gatedWorkspace( + root: string, + daemon: TestDaemon, + gate: Gate, +): Promise { + const seed = join(root, "seed"); + await Bun.$`git init -b main -q ${seed}`.quiet(); + await Bun.$`git -C ${seed} -c user.name=seed -c user.email=seed@seed.local commit -q --allow-empty -m seed`.quiet(); + const workspaceRoot = join(root, "workspaces"); + daemon.refreshGates.set(join(workspaceRoot, ".mirror.git"), gate.promise); + return { + workspaceRoot, + sshUrl: seed, + identity: { name: "Test Bot", email: "test@factory.local" }, + retainedWorkspaces: 10, + }; +} + /** The `DedupeKeyError` a rejected start produced — else a placeholder string. */ async function errorOf(promise: Promise): Promise { return promise.then( @@ -130,6 +157,7 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { test("a ctx.dispatch collision throws into the parent and is recorded as DispatchCollision", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-dispatch-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); mkdirSync(join(root, "workspaces"), { recursive: true }); const parent = defineWorkflow("parent-wf", { @@ -147,14 +175,14 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { }, }); - const parentRunId = await startTrackedRun(runtime, db, parent, { + const parentRunId = await startTrackedRun(daemon.runtime, db, parent, { input: { wait: 3 }, workspace: workspaces(join(root)), maxConcurrentRuns: 10, dispatchEnv: withWorkspaces(join(root)), }); - await waitFor(() => !isActive(parentRunId)); + await waitFor(() => !daemon.registry.isActive(parentRunId)); const events = getRunEvents(db, parentRunId); const collided = events.find((event) => event.payload._tag === "DispatchCollision"); @@ -170,7 +198,7 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { failed !== undefined && failed.payload._tag === "RunFailed" ? failed.payload.message : "", ).toMatch(/dedupe key held: "item:7"/); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -178,15 +206,16 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { test("the key is released when the holding child run finishes, so a later dispatch succeeds", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-dispatch-release-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); mkdirSync(join(root, "workspaces"), { recursive: true }); - let release: () => void = () => undefined; const gate = makeGate(); + const parentGate = makeGate(); const heldWorkflow = defineWorkflow("held-wf", { input: Schema.Struct({}), workspace: { kind: "scratch" }, run: async () => { - await gate.wait; + await gate.promise; return {}; }, }); @@ -197,16 +226,14 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { run: async (ctx) => { const ids = []; ids.push(await ctx.dispatch(heldWorkflow, {}, { dedupeKey: "item:8" })); - await new Promise((r) => { - release = r; - }); + await parentGate.promise; // The first child is terminal by now; the key must be free again. ids.push(await ctx.dispatch(heldWorkflow, {}, { dedupeKey: "item:8" })); return { ids }; }, }); - const parentRunId = await startTrackedRun(runtime, db, parent, { + const parentRunId = await startTrackedRun(daemon.runtime, db, parent, { input: {}, workspace: workspaces(join(root)), maxConcurrentRuns: 10, @@ -218,16 +245,26 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { ); await Bun.sleep(150); // The first child dispatched, still holding while the gate is shut — one - // more dispatch inside the parent would collide. Then let the child end. + // more dispatch inside the parent would collide. Then let the child end, + // and only once it is terminal let the parent dispatch again. + const firstChild = getRunEvents(db, parentRunId).find( + (event) => event.payload._tag === "RunDispatched", + ); + const firstChildRunId = + firstChild !== undefined && firstChild.payload._tag === "RunDispatched" + ? firstChild.payload.childRunId + : ""; + expect(daemon.dedupeRegistry.holderOf("item:8")).toBe(firstChildRunId); gate.release(); - if (typeof release === "function") release(); + await waitFor(() => !daemon.registry.isActive(firstChildRunId)); + parentGate.release(); await waitFor( () => getRunEvents(db, parentRunId).filter((event) => event.payload._tag === "RunDispatched") .length >= 2, 10_000, ); - await waitFor(() => !isActive(parentRunId)); + await waitFor(() => !daemon.registry.isActive(parentRunId)); const dispatchedCount = getRunEvents(db, parentRunId).filter( (event) => event.payload._tag === "RunDispatched", @@ -237,7 +274,7 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { getRunEvents(db, parentRunId).some((event) => event.payload._tag === "DispatchCollision"), ).toBe(false); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -248,19 +285,19 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-hold-")); const db = openStore(join(root, "factory.db")); const gate = makeGate(); - const registry = createDedupeRegistry(); + const daemon = createTestDaemon(SLOW_ADAPTER); const first = start(db, { runId: "run-holder", dir: join(root, "dir-1"), dedupeKey: "item:41", - beforeStart: gate.wait, - registry, + gate, + daemon, }); - await waitFor(() => isActive("run-holder")); + await waitFor(() => daemon.registry.isActive("run-holder")); const err = (await errorOf( - start(db, { runId: "run-collide", dir: join(root, "dir-2"), dedupeKey: "item:41", registry }), + start(db, { runId: "run-collide", dir: join(root, "dir-2"), dedupeKey: "item:41", daemon }), )) as DedupeKeyError | string; expect(err).toBeInstanceOf(DedupeKeyError); if (err instanceof DedupeKeyError) { @@ -271,9 +308,9 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { gate.release(); await first; - await waitFor(() => !isActive("run-holder")); + await waitFor(() => !daemon.registry.isActive("run-holder")); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -282,24 +319,24 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-log-")); const db = openStore(join(root, "factory.db")); const gate = makeGate(); - const registry = createDedupeRegistry(); + const daemon = createTestDaemon(SLOW_ADAPTER); const first = start(db, { runId: "run-holder", dir: join(root, "dir-1"), dedupeKey: "item:42", - beforeStart: gate.wait, - registry, + gate, + daemon, }); - await waitFor(() => isActive("run-holder")); + await waitFor(() => daemon.registry.isActive("run-holder")); await errorOf( - start(db, { runId: "run-collide", dir: join(root, "dir-2"), dedupeKey: "item:42", registry }), + start(db, { runId: "run-collide", dir: join(root, "dir-2"), dedupeKey: "item:42", daemon }), ); gate.release(); await first; - await waitFor(() => !isActive("run-holder")); + await waitFor(() => !daemon.registry.isActive("run-holder")); const holderStarted = getRunEvents(db, "run-holder").find( (e) => e.payload._tag === "RunStarted", @@ -314,7 +351,7 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { // A dropped start never started: the collision run's log has no rows. expect(getRunEvents(db, "run-collide")).toHaveLength(0); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -322,16 +359,25 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { test("the key is released when the holding run reaches its terminal state", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-finish-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); - const firstRunId = await start(db, { dir: join(root, "dir-1"), dedupeKey: "item:43" }); + const firstRunId = await start(db, { + dir: join(root, "dir-1"), + dedupeKey: "item:43", + daemon, + }); expect(firstRunId).toBeTypeOf("string"); - await waitFor(() => !isActive(firstRunId)); + await waitFor(() => !daemon.registry.isActive(firstRunId)); - const retryRunId = await start(db, { dir: join(root, "dir-2"), dedupeKey: "item:43" }); - await waitFor(() => !isActive(retryRunId)); + const retryRunId = await start(db, { + dir: join(root, "dir-2"), + dedupeKey: "item:43", + daemon, + }); + await waitFor(() => !daemon.registry.isActive(retryRunId)); expect(getRunEvents(db, retryRunId).some((e) => e.payload._tag === "RunStarted")).toBe(true); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -339,21 +385,25 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { test("a failed start releases its claim so a corrected retry can start", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-fail-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); // allocation of an explicitly undefined dir fails after the claim await expect( - startTrackedRun(runtime, db, echoWorkflow, { + startTrackedRun(daemon.runtime, db, echoWorkflow, { input: {}, dedupeKey: "item:44", - dedupeRegistry: createDedupeRegistry(), }), ).rejects.toThrow(/needs `dir` or `workspace`/); - const retryRunId = await start(db, { dir: join(root, "dir-1"), dedupeKey: "item:44" }); - await waitFor(() => !isActive(retryRunId)); + const retryRunId = await start(db, { + dir: join(root, "dir-1"), + dedupeKey: "item:44", + daemon, + }); + await waitFor(() => !daemon.registry.isActive(retryRunId)); expect(getRunEvents(db, retryRunId).some((e) => e.payload._tag === "RunStarted")).toBe(true); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -361,31 +411,29 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { test("a cancelled run releases its key (RunCancelled is a terminal state)", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-cancel-")); const db = openStore(join(root, "factory.db")); - const registry = createDedupeRegistry(); + const daemon = createTestDaemon(SLOW_ADAPTER); - const holderRunId = await startTrackedRun(runtime, db, echoWorkflow, { + const holderRunId = await startTrackedRun(daemon.runtime, db, echoWorkflow, { runId: "run-cancelled", dir: join(root, "dir-1"), input: {}, dedupeKey: "item:45", - dedupeRegistry: registry, }); - await waitFor(() => !isActive(holderRunId) || true); - const handle = getActiveHandle("run-cancelled"); + await waitFor(() => !daemon.registry.isActive(holderRunId) || true); + const handle = daemon.registry.getActiveHandle("run-cancelled"); expect(handle).toBeDefined(); if (handle !== undefined) await handle.cancel(); - await waitFor(() => !isActive(holderRunId)); + await waitFor(() => !daemon.registry.isActive(holderRunId)); const started = getRunEvents(db, holderRunId).some((e) => e.payload._tag === "RunStarted"); - const retryRunId = await startTrackedRun(runtime, db, echoWorkflow, { + const retryRunId = await startTrackedRun(daemon.runtime, db, echoWorkflow, { dir: join(root, "dir-2"), input: {}, dedupeKey: "item:45", - dedupeRegistry: registry, }); expect(started).toBe(true); expect(retryRunId).toBeTypeOf("string"); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); @@ -394,14 +442,17 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { test("runs started without a key behave exactly as before", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-none-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); - const firstRunId = await start(db, { dir: join(root, "dir-1") }); - const secondRunId = await start(db, { dir: join(root, "dir-2") }); + const firstRunId = await start(db, { dir: join(root, "dir-1"), daemon }); + const secondRunId = await start(db, { dir: join(root, "dir-2"), daemon }); expect(firstRunId).toBeTypeOf("string"); expect(secondRunId).toBeTypeOf("string"); - await waitFor(() => !isActive(firstRunId) && !isActive(secondRunId)); + await waitFor( + () => !daemon.registry.isActive(firstRunId) && !daemon.registry.isActive(secondRunId), + ); - await drain(); + await drain(daemon); db.close(); rmSync(root, { recursive: true, force: true }); }); @@ -425,6 +476,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { test("a key held by a running run surfaces as a 409 conflict naming the key and the holder", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-http-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); const slowStartWorkflow = defineWorkflow("dedupe-http-slow", { input: Schema.Struct({}), @@ -438,7 +490,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { const server = serve({ db, - runtime: makeAgentRuntime(SLOW_ADAPTER), + runtime: daemon.runtime, port: 0, config: defineConfig({ repo: { @@ -494,7 +546,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { const { runId: retryRunId } = (await retryRes.json()) as { runId: string }; await waitForTerminalStatus(db, retryRunId); } finally { - await drain(); + await drain(daemon); await server.stop(true); db.close(); rmSync(root, { recursive: true, force: true }); @@ -504,6 +556,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { test("a non-string dedupeKey is a 400, and omitting one behaves as today", async () => { const root = mkdtempSync(join(tmpdir(), "factory-dedupe-http-shape-")); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); const plainWorkflow = defineWorkflow("dedupe-http-plain", { input: Schema.Struct({}), workspace: { kind: "scratch" }, @@ -512,7 +565,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { const server = serve({ db, - runtime: makeAgentRuntime(SLOW_ADAPTER), + runtime: daemon.runtime, port: 0, config: defineConfig({ repo: { @@ -543,7 +596,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { const { runId } = (await plain.json()) as { runId: string }; await waitForTerminalStatus(db, runId); } finally { - await drain(); + await drain(daemon); await server.stop(true); db.close(); rmSync(root, { recursive: true, force: true }); diff --git a/src/server/http.test.ts b/src/server/http.test.ts index 30e7676..780950e 100644 --- a/src/server/http.test.ts +++ b/src/server/http.test.ts @@ -6,12 +6,14 @@ import { Schema } from "effect"; import { defineWorkflow, type WorkflowDefinition } from "../workflow"; import { createSlowFakeAdapter } from "../replay/adapter"; import type { AgentAdapter } from "../runtime/agent-adapter"; -import { makeAgentRuntime } from "../runtime/agent-runtime"; import { getRunEvents, openStore } from "../persistence/store"; import { defineConfig, loadFactoryConfig } from "../config"; import registryWorkflow from "../../test/fixtures/registry-workflow"; import registryScratchWorkflow from "../../test/fixtures/registry-scratch-workflow"; import { serve } from "./http"; +// Issue #38: each server under test gets its own daemon runtime, so its own +// registry, pubsub, dedupe registry and refresh gates. +import { makeDaemonRuntime } from "./daemon-runtime"; const ECHO_WORKFLOW = `${import.meta.dir}/../../test/fixtures/echo-workflow.ts`; const QUIET_GAP_WORKFLOW = `${import.meta.dir}/../../test/fixtures/quiet-gap-workflow.ts`; @@ -91,7 +93,12 @@ describe("GET /api/workflows (D30)", () => { const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([], 1); const config = await loadFactoryConfig(FIXTURE_CONFIG); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -116,7 +123,11 @@ describe("GET /api/workflows (D30)", () => { const dir = mkdtempSync(join(tmpdir(), "factory-workflows-empty-test-")); const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([], 1); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -136,7 +147,11 @@ describe("GET /api/workflows (D30)", () => { const dir = mkdtempSync(join(tmpdir(), "factory-workflows-empty-test-")); const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([], 1); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -156,7 +171,11 @@ describe("phase 4 SPA serving", () => { const dir = mkdtempSync(join(tmpdir(), "factory-spa-test-")); const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([{ type: "TEXT_MESSAGE_START" }], 1); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -239,7 +258,12 @@ describe("SSE keepalive", () => { const dir = mkdtempSync(join(tmpdir(), "factory-sse-keepalive-test-")); const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([], 1); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, sseKeepaliveMs: 25 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + sseKeepaliveMs: 25, + }); const base = `http://localhost:${server.port}`; try { @@ -275,7 +299,12 @@ describe("SSE keepalive", () => { const dir = mkdtempSync(join(tmpdir(), "factory-sse-keepalive-stop-test-")); const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([], 1); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, sseKeepaliveMs: 10 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + sseKeepaliveMs: 10, + }); const base = `http://localhost:${server.port}`; try { @@ -315,7 +344,11 @@ describe("phase 3 HTTP API + SSE", () => { ], 1, ); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -364,7 +397,11 @@ describe("phase 3 HTTP API + SSE", () => { ], 500, ); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -400,7 +437,11 @@ describe("phase 3 HTTP API + SSE", () => { ], 60, ); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -439,7 +480,11 @@ describe("phase 3 HTTP API + SSE", () => { ], 400, ); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -477,7 +522,11 @@ describe("phase 3 HTTP API + SSE", () => { ], 1, ); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -523,7 +572,12 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { 1, ); const config = await loadFactoryConfig(FIXTURE_CONFIG); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -549,7 +603,12 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { const db = openStore(join(dir, "factory.db")); const adapter = createSlowFakeAdapter([], 1); const config = await loadFactoryConfig(FIXTURE_CONFIG); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -595,7 +654,12 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { maxConcurrentRuns: 1, retainedWorkspaces: 10, }); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -658,7 +722,12 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { maxConcurrentRuns: 3, retainedWorkspaces: 10, }); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -725,7 +794,12 @@ describe("a scratch workflow through POST /api/runs (issue #13)", () => { maxConcurrentRuns: 3, retainedWorkspaces: 10, }); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -798,7 +872,7 @@ describe("ctx.dispatch through POST /api/runs (issue #14)", () => { const server = serve({ db, - runtime: makeAgentRuntime(adapter), + runtime: makeDaemonRuntime(adapter), port: 0, config: defineConfig({ repo: { @@ -896,7 +970,7 @@ describe("GET /api/schedules (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config: schedulesConfig(root, workflow), }); @@ -965,7 +1039,7 @@ describe("GET /api/schedules (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config, }); @@ -989,7 +1063,11 @@ describe("GET /api/schedules (issue #17)", () => { test("serves an empty list on a legacy no-config server", async () => { const root = mkdtempSync(join(tmpdir(), "factory-schedules-empty-test-")); const db = openStore(join(root, "factory.db")); - const server = serve({ db, runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), + port: 0, + }); const base = `http://localhost:${server.port}`; try { @@ -1033,7 +1111,7 @@ describe("GET /api/schedules (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config, }); @@ -1134,7 +1212,7 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config, }); @@ -1188,7 +1266,7 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config, }); @@ -1231,7 +1309,7 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config, }); @@ -1296,7 +1374,7 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { }); const server = serve({ db, - runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + runtime: makeDaemonRuntime(createSlowFakeAdapter([], 1)), port: 0, config, }); @@ -1345,7 +1423,11 @@ describe("adapter.prepareWorkspace through the legacy POST /api/runs (#37)", () const seed = join(root, "seed"); await Bun.$`git init -b main -q ${seed}`.quiet(); const adapter = trackingAdapter(); - const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); + const server = serve({ + db, + runtime: makeDaemonRuntime(adapter), + port: 0, + }); const base = `http://localhost:${server.port}`; try { diff --git a/src/server/http.ts b/src/server/http.ts index 755a1fb..c41829f 100644 --- a/src/server/http.ts +++ b/src/server/http.ts @@ -25,27 +25,32 @@ * between, then drains the buffer de-duplicated by `seq` once the persisted * read completes. Without this ordering a live event emitted between the * subscribe and the read could be lost. + * + * #38: the handler resolves the daemon's registry and pubsub from the daemon + * runtime it is given (`server/daemon-runtime.ts`) rather than reading + * module-level singletons. The handlers themselves stay plain functions + * (ADR 0009 §5) — the runtime is only the way into the context. */ import type { Database } from "bun:sqlite"; import { isTerminal, type RunEvent } from "../events"; import type { FactoryConfig } from "../config"; -import { Schema, SchemaParser, type ManagedRuntime } from "effect"; +import { Schema, SchemaParser } from "effect"; import { resetClone, type GitIdentity } from "../lib/clone"; import { loadWorkflow } from "../lib/load-workflow"; import { getRunEvents, listRuns, type RunSummary } from "../persistence/store"; -import type { AgentRuntime } from "../runtime/agent-runtime"; import index from "../web/index.html"; import { admitRun } from "./admission"; -import { subscribe } from "./pubsub"; import { DedupeKeyError } from "../lib/dedupe"; +import { serviceOf, type DaemonRuntime } from "./daemon-runtime"; +import { RunPubSub, type PubSub } from "./pubsub"; import { ConcurrencyLimitError, - activeRunIds, - cancelRegisteredRun, - isActive, + DaemonShuttingDownError, + RunRegistry, startTrackedRun, type DispatchEnv, + type RunRegistryShape, type StartTrackedRunOptions, } from "./runs"; import { nextFireAt, toRuntimeSchedules } from "./scheduler"; @@ -77,9 +82,11 @@ export interface ServerOptions { /** * Issue #36: the Effect managed runtime that provides the agent runtime * service. The adapter is resolved from its context inside the agent step - * rather than threaded through `ServerOptions`. + * rather than threaded through `ServerOptions`. #38: the same runtime + * provides this daemon's run registry, pubsub, dedupe registry and refresh + * gates. */ - readonly runtime: ManagedRuntime.ManagedRuntime; + readonly runtime: DaemonRuntime; /** * The loaded `factory.config.ts` (D27). Absent = the phase 3 path-based API * behaves exactly as before (no limit, explicit dir+clone, `/api/workflows` @@ -125,8 +132,11 @@ export type RunSummaryResponse = RunSummary & { readonly active: boolean }; * that bit to group live rows, so it is derived here from the in-memory * registry (`server/runs.ts`) rather than stored (D24: active is process state). */ -function listSummaries(db: Database): ReadonlyArray { - return listRuns(db).map((run) => ({ ...run, active: isActive(run.runId) })); +function listSummaries( + db: Database, + registry: RunRegistryShape, +): ReadonlyArray { + return listRuns(db).map((run) => ({ ...run, active: registry.isActive(run.runId) })); } /** @@ -216,6 +226,8 @@ function parseLastEventId(req: Request): number | undefined { function sseStream( db: Database, + registry: RunRegistryShape, + pubsub: PubSub, runId: string, lastEventId?: number, keepaliveMs: number = DEFAULT_SSE_KEEPALIVE_MS, @@ -267,7 +279,7 @@ function sseStream( controller.close(); }; - unsubscribe = subscribe(runId, (event) => { + unsubscribe = pubsub.subscribe(runId, (event) => { if (!draining) { buffered.push(event); return; @@ -290,7 +302,7 @@ function sseStream( if (isTerminal(event.payload)) sawTerminal = true; } - if (sawTerminal || !isActive(runId)) { + if (sawTerminal || !registry.isActive(runId)) { finish(); return; } @@ -326,11 +338,13 @@ function sseStream( } export function createHandler(options: ServerOptions): (req: Request) => Promise { + const registry = serviceOf(options.runtime, RunRegistry); + const pubsub = serviceOf(options.runtime, RunPubSub); return async (req: Request): Promise => { const url = new URL(req.url); if (req.method === "GET" && url.pathname === "/api/runs") { - return json(listSummaries(options.db)); + return json(listSummaries(options.db, registry)); } if (req.method === "GET" && url.pathname === "/api/workflows") { @@ -349,7 +363,10 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise // path alike. 409 over the limit; the limit is only consulted when a // config is present (the no-config legacy path kept its old behaviour). const maxConcurrentRuns = options.config?.maxConcurrentRuns; - if (maxConcurrentRuns !== undefined && !admitRun(maxConcurrentRuns, activeRunIds().length)) { + if ( + maxConcurrentRuns !== undefined && + !admitRun(maxConcurrentRuns, registry.activeRunIds().length) + ) { return json( { error: ConcurrencyLimitError.of({ maxConcurrentRuns }).message }, { status: 409 }, @@ -403,6 +420,9 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise if (err instanceof ConcurrencyLimitError) { return json({ error: err.message }, { status: 409 }); } + if (err instanceof DaemonShuttingDownError) { + return json({ error: err.message }, { status: 503 }); + } // Issue #15: the key collision surfaces as a conflict that names both // the key and the run holding it, so a client can see exactly whom it // raced with. @@ -472,6 +492,9 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise if (err instanceof ConcurrencyLimitError) { return json({ error: err.message }, { status: 409 }); } + if (err instanceof DaemonShuttingDownError) { + return json({ error: err.message }, { status: 503 }); + } if (err instanceof DedupeKeyError) { return json( { error: err.message, dedupeKey: err.key, holderRunId: err.holderRunId }, @@ -546,7 +569,7 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise // D29 admission, checked here exactly like POST /api/runs so a manual // trigger over the limit is a visible 409 rather than a surprise. const maxConcurrentRuns = options.config.maxConcurrentRuns; - if (!admitRun(maxConcurrentRuns, activeRunIds().length)) { + if (!admitRun(maxConcurrentRuns, registry.activeRunIds().length)) { return json( { error: ConcurrencyLimitError.of({ maxConcurrentRuns }).message }, { status: 409 }, @@ -572,6 +595,9 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise if (err instanceof ConcurrencyLimitError) { return json({ error: err.message }, { status: 409 }); } + if (err instanceof DaemonShuttingDownError) { + return json({ error: err.message }, { status: 503 }); + } if (err instanceof DedupeKeyError) { return json( { error: err.message, dedupeKey: err.key, holderRunId: err.holderRunId }, @@ -586,7 +612,7 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise const cancelMatch = /^\/api\/runs\/([^/]+)\/cancel$/.exec(url.pathname); if (req.method === "POST" && cancelMatch) { const runId = cancelMatch[1] as string; - const target = cancelRegisteredRun(runId); + const target = registry.cancelRegisteredRun(runId); if (target === undefined) return json({ error: "run not active" }, { status: 409 }); if (target.kind === "handle") await target.handle.cancel(); return json({ runId, cancelled: true }); @@ -595,10 +621,17 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise const eventsMatch = /^\/api\/runs\/([^/]+)\/events$/.exec(url.pathname); if (req.method === "GET" && eventsMatch) { const runId = eventsMatch[1] as string; - const exists = listSummaries(options.db).some((r) => r.runId === runId); + const exists = listSummaries(options.db, registry).some((r) => r.runId === runId); if (!exists) return json({ error: "not found" }, { status: 404 }); return new Response( - sseStream(options.db, runId, parseLastEventId(req), options.sseKeepaliveMs), + sseStream( + options.db, + registry, + pubsub, + runId, + parseLastEventId(req), + options.sseKeepaliveMs, + ), { headers: { "content-type": "text/event-stream", @@ -612,7 +645,7 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise const runMatch = /^\/api\/runs\/([^/]+)$/.exec(url.pathname); if (req.method === "GET" && runMatch) { const runId = runMatch[1] as string; - const run = listSummaries(options.db).find((r) => r.runId === runId); + const run = listSummaries(options.db, registry).find((r) => r.runId === runId); if (run === undefined) return json({ error: "not found" }, { status: 404 }); return json(run); } diff --git a/src/server/nested-runs.test.ts b/src/server/nested-runs.test.ts index 7c9e8ac..4fe57fa 100644 --- a/src/server/nested-runs.test.ts +++ b/src/server/nested-runs.test.ts @@ -8,6 +8,8 @@ * carries the message). A validated child is then started fire-and-forget * with the parent's run id recorded on its `RunStarted.parentId`, and the * parent never exposes an awaitable child (D-epic 19). + * + * #38: each test builds its own daemon runtime (`createTestDaemon`). */ import { mkdirSync, mkdtempSync, rmSync } from "node:fs"; @@ -18,9 +20,9 @@ import { describe, expect, test } from "bun:test"; import { Schema } from "effect"; import { getRunEvents, listRuns, openStore } from "../persistence/store"; import { createSlowFakeAdapter } from "../replay/adapter"; -import { activeRunIds, isActive, startTrackedRun } from "./runs"; +import { startTrackedRun } from "./runs"; +import { createTestDaemon } from "./test-daemon"; import { defineWorkflow } from "../workflow"; -import { makeAgentRuntime } from "../runtime/agent-runtime"; const SLOW_ADAPTER = createSlowFakeAdapter( [ @@ -31,8 +33,6 @@ const SLOW_ADAPTER = createSlowFakeAdapter( 25, ); -const runtime = makeAgentRuntime(SLOW_ADAPTER); - // The child does a real ctx.exec so its outcome is observable in its log. // Scratch (issue #13): no mirror refresh, so no git fixture is needed. const numberedChild = defineWorkflow("child-wf", { @@ -82,6 +82,7 @@ describe("nested runs through the daemon (issue #14)", () => { const root = mkdtempSync(join(tmpdir(), "factory-nested-happy-")); const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "workspaces"), { recursive: true }); + const daemon = createTestDaemon(SLOW_ADAPTER); const parent = defineWorkflow("parent-wf", { input: Schema.Struct({}), @@ -92,7 +93,7 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - const parentRunId = await startTrackedRun(runtime, db, parent, { + const parentRunId = await startTrackedRun(daemon.runtime, db, parent, { workspace: { workspaceRoot: join(root, "workspaces"), sshUrl: join(root, "seed-not-used"), @@ -110,7 +111,9 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - await waitFor(() => !isActive(parentRunId) && activeRunIds().length === 0); + await waitFor( + () => !daemon.registry.isActive(parentRunId) && daemon.registry.activeRunIds().length === 0, + ); const parentEvents = getRunEvents(db, parentRunId); const dispatched = parentEvents.find((e) => e.payload._tag === "RunDispatched"); @@ -145,8 +148,9 @@ describe("nested runs through the daemon (issue #14)", () => { const root = mkdtempSync(join(tmpdir(), "factory-nested-depth-")); const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "workspaces"), { recursive: true }); + const daemon = createTestDaemon(SLOW_ADAPTER); - await startTrackedRun(runtime, db, selfDispatching, { + await startTrackedRun(daemon.runtime, db, selfDispatching, { input: {}, workspace: { workspaceRoot: join(root, "workspaces"), @@ -166,7 +170,7 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - await waitFor(() => activeRunIds().length === 0); + await waitFor(() => daemon.registry.activeRunIds().length === 0); // The chain built to the cap, no farther: root + 5 children max. const selfRuns = listRuns(db).filter((run) => run.workflowId === "self-dispatch-wf"); @@ -196,8 +200,9 @@ describe("nested runs through the daemon (issue #14)", () => { const root = mkdtempSync(join(tmpdir(), "factory-nested-count-")); const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "workspaces"), { recursive: true }); + const daemon = createTestDaemon(SLOW_ADAPTER); - const parentRunId = await startTrackedRun(runtime, db, manyChildren, { + const parentRunId = await startTrackedRun(daemon.runtime, db, manyChildren, { input: {}, workspace: { workspaceRoot: join(root, "workspaces"), @@ -218,7 +223,7 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - await waitFor(() => activeRunIds().length === 0); + await waitFor(() => daemon.registry.activeRunIds().length === 0); expect(tagsOf(db, parentRunId).filter((tag) => tag === "RunDispatched").length).toBe(5); const parentEvents = getRunEvents(db, parentRunId); @@ -237,6 +242,7 @@ describe("nested runs through the daemon (issue #14)", () => { const root = mkdtempSync(join(tmpdir(), "factory-nested-wip-")); const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "workspaces"), { recursive: true }); + const daemon = createTestDaemon(SLOW_ADAPTER); const parent = defineWorkflow("parent-wip-wf", { input: Schema.Struct({}), @@ -247,7 +253,7 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - const parentRunId = await startTrackedRun(runtime, db, parent, { + const parentRunId = await startTrackedRun(daemon.runtime, db, parent, { input: {}, workspace: { workspaceRoot: join(root, "workspaces"), @@ -267,7 +273,7 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - await waitFor(() => !isActive(parentRunId)); + await waitFor(() => !daemon.registry.isActive(parentRunId)); const events = getRunEvents(db, parentRunId); expect(events.map((event) => event.payload._tag)).not.toContain("RunDispatched"); diff --git a/src/server/pubsub.ts b/src/server/pubsub.ts index 93e00e3..f837789 100644 --- a/src/server/pubsub.ts +++ b/src/server/pubsub.ts @@ -4,27 +4,47 @@ * in flight — persisted history always comes from `getRunEvents` * (`src/persistence/store.ts`); this is the gap between "last flushed to * sqlite" and "just happened", not a second source of truth. + * + * #38: the subscriber map is per daemon, not per module. It is the + * `RunPubSub` service below; its layer builds a fresh map each time a daemon + * runtime is built, so two daemons in one process do not share subscribers. */ +import { Context, Layer } from "effect"; import type { RunEvent } from "../events"; type Listener = (event: RunEvent) => void; -const subscribers = new Map>(); - -export function publish(runId: string, event: RunEvent): void { - for (const listener of subscribers.get(runId) ?? []) listener(event); +export interface PubSub { + publish(runId: string, event: RunEvent): void; + subscribe(runId: string, listener: Listener): () => void; } -export function subscribe(runId: string, listener: Listener): () => void { - let set = subscribers.get(runId); - if (set === undefined) { - set = new Set(); - subscribers.set(runId, set); - } - set.add(listener); - return () => { - set.delete(listener); - if (set.size === 0) subscribers.delete(runId); +export function createPubSub(): PubSub { + const subscribers = new Map>(); + return { + publish(runId: string, event: RunEvent): void { + for (const listener of subscribers.get(runId) ?? []) listener(event); + }, + subscribe(runId: string, listener: Listener): () => void { + let set = subscribers.get(runId); + if (set === undefined) { + set = new Set(); + subscribers.set(runId, set); + } + set.add(listener); + return () => { + set!.delete(listener); + if (set!.size === 0) subscribers.delete(runId); + }; + }, }; } + +/** + * The live run-event fan-out as a daemon service (#38, ADR 0009 §5). Named + * `RunPubSub` rather than `PubSub` so it never reads as Effect's own module. + */ +export class RunPubSub extends Context.Service()("RunPubSub") {} + +export const RunPubSubLayer: Layer.Layer = Layer.sync(RunPubSub, createPubSub); diff --git a/src/server/runs.test.ts b/src/server/runs.test.ts index 7e050bc..a0b870f 100644 --- a/src/server/runs.test.ts +++ b/src/server/runs.test.ts @@ -7,6 +7,11 @@ * - a `cancel` arriving while a run is only a reserved slot is deferred into * run start (L1): the run starts, is cancelled immediately, and ends as a * clean RunCancelled with no orphaned slot. + * + * #38: each test builds its own daemon runtime (`createTestDaemon`) instead + * of sharing module-level singletons. The reserved-but-not-started window is held open by pre-seeding + * the daemon's refresh gate for the workspace mirror, so allocation blocks + * until the test releases it. */ import { existsSync, mkdirSync, mkdtempSync, readdirSync, rmSync } from "node:fs"; @@ -18,17 +23,14 @@ import type { AgentAdapter } from "../runtime/agent-adapter"; import echoWorkflow from "../../test/fixtures/echo-workflow"; import { appendEvent, openStore, getRunEvents } from "../persistence/store"; import { createSlowFakeAdapter } from "../replay/adapter"; -import { makeAgentRuntime } from "../runtime/agent-runtime"; import { defineWorkflow, Schema } from "../workflow"; import { ConcurrencyLimitError, DispatchCapError, - activeRunIds, - cancelRegisteredRun, - getActiveHandle, - isActive, startTrackedRun, + type WorkspaceSpec, } from "./runs"; +import { createTestDaemon, type TestDaemon } from "./test-daemon"; const SLOW_ADAPTER = createSlowFakeAdapter( [ @@ -39,8 +41,6 @@ const SLOW_ADAPTER = createSlowFakeAdapter( 25, ); -const runtime = makeAgentRuntime(SLOW_ADAPTER); - const sleepWorkflow = defineWorkflow("sleep-test", { input: Schema.Struct({}), run: async (ctx) => { @@ -55,7 +55,7 @@ function tmpRoot(): { root: string; finish: () => void } { } interface Gate { - readonly wait: () => Promise; + readonly promise: Promise; readonly release: () => void; } @@ -64,7 +64,30 @@ function makeGate(): Gate { const promise = new Promise((resolve) => { release = resolve; }); - return { wait: () => promise, release }; + return { promise, release }; +} + +/** + * A clone workspace whose allocation blocks until `gate` is released: the + * gate is pre-seeded as the mirror's in-flight refresh, so `allocateWorkspace` + * queues behind it — the run holds a reserved slot and nothing more. + */ +async function gatedWorkspace( + root: string, + daemon: TestDaemon, + gate: Gate, +): Promise { + const seed = join(root, "seed"); + await Bun.$`git init -b main -q ${seed}`.quiet(); + await Bun.$`git -C ${seed} -c user.name=seed -c user.email=seed@seed.local commit -q --allow-empty -m seed`.quiet(); + const workspaceRoot = join(root, "workspaces"); + daemon.refreshGates.set(join(workspaceRoot, ".mirror.git"), gate.promise); + return { + workspaceRoot, + sshUrl: seed, + identity: { name: "T", email: "t@t.test" }, + retainedWorkspaces: 10, + }; } async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise { @@ -102,21 +125,21 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" test("a second start while the first is still reserving is refused atomically, and the slot frees after", async () => { const { root, finish } = tmpRoot(); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); const gate = makeGate(); - const first = startTrackedRun(runtime, db, echoWorkflow, { + const first = startTrackedRun(daemon.runtime, db, echoWorkflow, { runId: "run-first", - dir: join(root, "first-dir"), + workspace: await gatedWorkspace(root, daemon, gate), input: {}, maxConcurrentRuns: 1, - beforeStart: gate.wait, }); - await waitFor(() => activeRunIds().length === 1); - expect(activeRunIds()).toEqual(["run-first"]); - expect(getActiveHandle("run-first")).toBeUndefined(); + await waitFor(() => daemon.registry.activeRunIds().length === 1); + expect(daemon.registry.activeRunIds()).toEqual(["run-first"]); + expect(daemon.registry.getActiveHandle("run-first")).toBeUndefined(); - const second = startTrackedRun(runtime, db, echoWorkflow, { + const second = startTrackedRun(daemon.runtime, db, echoWorkflow, { runId: "run-second", dir: join(root, "second-dir"), input: {}, @@ -133,8 +156,104 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" gate.release(); const firstRunId = await first; expect(firstRunId).toBe("run-first"); - await waitFor(() => !isActive(firstRunId)); - expect(activeRunIds()).toEqual([]); + await waitFor(() => !daemon.registry.isActive(firstRunId)); + expect(daemon.registry.activeRunIds()).toEqual([]); + + db.close(); + finish(); + }); + + test("a second start issued synchronously, before the first is awaited, is refused", async () => { + const { root, finish } = tmpRoot(); + const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); + const gate = makeGate(); + const workspace = await gatedWorkspace(root, daemon, gate); + + // No await between the two calls: the first must already hold its slot. + const first = startTrackedRun(daemon.runtime, db, echoWorkflow, { + runId: "run-sync-first", + workspace, + input: {}, + maxConcurrentRuns: 1, + }); + const second = startTrackedRun(daemon.runtime, db, echoWorkflow, { + runId: "run-sync-second", + workspace, + input: {}, + maxConcurrentRuns: 1, + }); + + expect(await second.catch((err: unknown) => err)).toBeInstanceOf(ConcurrencyLimitError); + expect(daemon.registry.activeRunIds()).toEqual(["run-sync-first"]); + + gate.release(); + const firstRunId = await first; + await waitFor(() => !daemon.registry.isActive(firstRunId)); + + db.close(); + finish(); + }); + + test("a second start with the same runId while the first is still reserving is refused", async () => { + const { root, finish } = tmpRoot(); + const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); + const gate = makeGate(); + const workspace = await gatedWorkspace(root, daemon, gate); + + // No limit, so admission cannot be what refuses it: the reserved slot is. + const first = startTrackedRun(daemon.runtime, db, echoWorkflow, { + runId: "run-same-id", + workspace, + input: {}, + }); + const second = startTrackedRun(daemon.runtime, db, echoWorkflow, { + runId: "run-same-id", + workspace, + input: {}, + }); + + const err = await second.catch((e: unknown) => e); + expect(err).toBeInstanceOf(Error); + expect((err as Error).message).toContain("already active"); + expect(daemon.registry.activeRunIds()).toEqual(["run-same-id"]); + + gate.release(); + await first; + await waitFor(() => !daemon.registry.isActive("run-same-id")); + + db.close(); + finish(); + }); + + test("a start refused at the limit does not keep its dedupe key", async () => { + const { root, finish } = tmpRoot(); + const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); + const gate = makeGate(); + + const first = startTrackedRun(daemon.runtime, db, echoWorkflow, { + runId: "run-holding-slot", + workspace: await gatedWorkspace(root, daemon, gate), + input: {}, + maxConcurrentRuns: 1, + }); + + // e.g. a scheduled fire with `overlap: "skip"` while the daemon is full. + const refused = startTrackedRun(daemon.runtime, db, echoWorkflow, { + runId: "run-refused", + dir: join(root, "refused-dir"), + input: {}, + maxConcurrentRuns: 1, + dedupeKey: "schedule:nightly", + }); + expect(await refused.catch((err: unknown) => err)).toBeInstanceOf(ConcurrencyLimitError); + expect(daemon.dedupeRegistry.holderOf("schedule:nightly")).toBeUndefined(); + + gate.release(); + const firstRunId = await first; + await waitFor(() => !daemon.registry.isActive(firstRunId)); db.close(); finish(); @@ -143,9 +262,10 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" test("a failed allocation releases the reserved slot", async () => { const { root, finish } = tmpRoot(); const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); await expect( - startTrackedRun(runtime, db, echoWorkflow, { + startTrackedRun(daemon.runtime, db, echoWorkflow, { runId: "run-doomed", input: {}, maxConcurrentRuns: 1, @@ -158,15 +278,15 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" }), ).rejects.toThrow(/workspace/); - expect(activeRunIds()).toEqual([]); + expect(daemon.registry.activeRunIds()).toEqual([]); - const runId = await startTrackedRun(runtime, db, echoWorkflow, { + const runId = await startTrackedRun(daemon.runtime, db, echoWorkflow, { dir: join(root, "dir"), input: {}, maxConcurrentRuns: 1, }); - expect(activeRunIds()).toEqual([runId]); - await waitFor(() => !isActive(runId)); + expect(daemon.registry.activeRunIds()).toEqual([runId]); + await waitFor(() => !daemon.registry.isActive(runId)); db.close(); finish(); @@ -177,35 +297,34 @@ describe("cancel of a reserved-but-not-started run (L1)", () => { test("cancelling during the allocation window defers into run start; the run ends cancelled with no leak", async () => { const { root, finish } = tmpRoot(); const db = openStore(join(root, "factory.db")); - mkdirSync(join(root, "dir"), { recursive: true }); + const daemon = createTestDaemon(SLOW_ADAPTER); const gate = makeGate(); - const starting = startTrackedRun(runtime, db, sleepWorkflow, { + const starting = startTrackedRun(daemon.runtime, db, sleepWorkflow, { runId: "run-gated", - dir: join(root, "dir"), + workspace: await gatedWorkspace(root, daemon, gate), input: {}, maxConcurrentRuns: 1, - beforeStart: gate.wait, }); - await waitFor(() => activeRunIds().length === 1); - expect(getActiveHandle("run-gated")).toBeUndefined(); + await waitFor(() => daemon.registry.activeRunIds().length === 1); + expect(daemon.registry.getActiveHandle("run-gated")).toBeUndefined(); - const cancelled = cancelRegisteredRun("run-gated"); + const cancelled = daemon.registry.cancelRegisteredRun("run-gated"); expect(cancelled).toEqual({ kind: "reserved" }); gate.release(); const runId = await starting; expect(runId).toBe("run-gated"); - await waitFor(() => activeRunIds().length === 0); + await waitFor(() => daemon.registry.activeRunIds().length === 0); const events = getRunEvents(db, runId).map((e) => e.payload._tag); expect(events).toContain("RunStarted"); expect(events).toContain("RunCancelled"); - expect(cancelRegisteredRun(runId)).toBeUndefined(); + expect(daemon.registry.cancelRegisteredRun(runId)).toBeUndefined(); await Bun.sleep(50); - expect(activeRunIds()).toEqual([]); + expect(daemon.registry.activeRunIds()).toEqual([]); db.close(); finish(); @@ -247,14 +366,15 @@ describe("scratch workspaces through startTrackedRun (issue #13)", () => { const { root, finish } = tmpRoot(); _tmp = root; const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); mkdirSync(join(root, "seed"), { recursive: true }); - const runId = await startTrackedRun(runtime, db, failingScratchWorkflow, { + const runId = await startTrackedRun(daemon.runtime, db, failingScratchWorkflow, { runId: "run-scratch-empty", workspace: { ...workspaceSpec(), sshUrl: join(root, "no-such-remote") }, input: {}, }); - await waitFor(() => !isActive(runId)); + await waitFor(() => !daemon.registry.isActive(runId)); const dir = join(root, "workspaces", "run-scratch-empty"); expect(existsSync(dir)).toBe(true); @@ -268,23 +388,24 @@ describe("scratch workspaces through startTrackedRun (issue #13)", () => { const { root, finish } = tmpRoot(); _tmp = root; const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); mkdirSync(join(root, "seed"), { recursive: true }); - const okId = await startTrackedRun(runtime, db, scratchWorkflow, { + const okId = await startTrackedRun(daemon.runtime, db, scratchWorkflow, { runId: "run-scratch-ok", workspace: workspaceSpec(), input: {}, }); - await waitFor(() => !isActive(okId)); + await waitFor(() => !daemon.registry.isActive(okId)); await new Promise((resolve) => setTimeout(resolve, 50)); // reap lands async expect(existsSync(join(root, "workspaces", "run-scratch-ok"))).toBe(false); - const badId = await startTrackedRun(runtime, db, failingScratchWorkflow, { + const badId = await startTrackedRun(daemon.runtime, db, failingScratchWorkflow, { runId: "run-scratch-bad", workspace: workspaceSpec(), input: {}, }); - await waitFor(() => !isActive(badId)); + await waitFor(() => !daemon.registry.isActive(badId)); await new Promise((resolve) => setTimeout(resolve, 50)); expect(existsSync(join(root, "workspaces", "run-scratch-bad"))).toBe(true); finish(); @@ -294,6 +415,7 @@ describe("scratch workspaces through startTrackedRun (issue #13)", () => { const { root, finish } = tmpRoot(); _tmp = root; const db = openStore(join(root, "factory.db")); + const daemon = createTestDaemon(SLOW_ADAPTER); const seed = join(root, "seed"); await Bun.$`git init -b main -q ${seed}`.quiet(); @@ -314,12 +436,12 @@ describe("scratch workspaces through startTrackedRun (issue #13)", () => { // retention = 1: with the scratch dir correctly excluded, the only clone // survives: the leftover must not count toward retention. - await startTrackedRun(runtime, db, echoWorkflow, { + await startTrackedRun(daemon.runtime, db, echoWorkflow, { runId: "run-clone", workspace: { ...workspaceSpec(), sshUrl: seed, retainedWorkspaces: 1 }, input: {}, }); - await waitFor(() => !isActive("run-clone")); + await waitFor(() => !daemon.registry.isActive("run-clone")); expect(existsSync(join(root, "workspaces", "run-clone"))).toBe(true); expect(existsSync(join(root, "workspaces", "run-scratch-kept"))).toBe(true); @@ -354,11 +476,12 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { test("an allocated clone workspace is prepared on the allocated dir", async () => { const { root, finish } = tmpRoot(); const db = openStore(join(root, "factory.db")); + const adapter = trackingAdapter(); const seed = join(root, "seed"); await Bun.$`git init -b main -q ${seed}`.quiet(); - const adapter = trackingAdapter(); + const daemon = createTestDaemon(adapter); - const runId = await startTrackedRun(makeAgentRuntime(adapter), db, echoWorkflow, { + const runId = await startTrackedRun(daemon.runtime, db, echoWorkflow, { runId: "run-prep-clone", workspace: { workspaceRoot: join(root, "workspaces"), @@ -368,7 +491,7 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { }, input: {}, }); - await waitFor(() => !isActive(runId)); + await waitFor(() => !daemon.registry.isActive(runId)); expect(adapter.prepared).toEqual([join(root, "workspaces", "run-prep-clone")]); finish(); @@ -378,8 +501,9 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { const { root, finish } = tmpRoot(); const db = openStore(join(root, "factory.db")); const adapter = trackingAdapter(); + const daemon = createTestDaemon(adapter); - const runId = await startTrackedRun(makeAgentRuntime(adapter), db, scratchWorkflow, { + const runId = await startTrackedRun(daemon.runtime, db, scratchWorkflow, { runId: "run-prep-scratch", workspace: { workspaceRoot: join(root, "workspaces"), @@ -389,7 +513,7 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { }, input: {}, }); - await waitFor(() => !isActive(runId)); + await waitFor(() => !daemon.registry.isActive(runId)); expect(adapter.prepared).toEqual([]); finish(); diff --git a/src/server/runs.ts b/src/server/runs.ts index 3630e7a..935f1c6 100644 --- a/src/server/runs.ts +++ b/src/server/runs.ts @@ -11,9 +11,28 @@ * workspace. A reserved slot is released if allocation/startup fails, and a * `cancel` that arrives while a run is still reserving is deferred into run * start rather than dropped. + * + * Shutdown cancels what the registry holds: `RunRegistryShape.shutdown` + * refuses further starts, cancels every running handle, marks every reserved + * slot cancelled (so the run is cancelled the moment it starts, exactly like + * a deferred API cancel), and waits — bounded — for all of them to settle. + * Settled means each run persisted its `RunCancelled` and its `ctx.exec` + * children were killed, rather than being orphaned by process exit and read + * back as "interrupted" after a restart. `DaemonHandle.stop` runs it while + * the daemon runtime is still live; the registry layer's finalizer runs it + * again as a safety net when the runtime is disposed without `stop`. + * + * #38 (ADR 0009 §5): the registry is a `Context.Service` (`RunRegistry`) + * whose layer builds a fresh map per daemon runtime, alongside the pubsub, + * dedupe registry and refresh gates. Nothing here holds module-level state: + * `startTrackedRun` resolves the services from the daemon runtime it is + * handed (`server/daemon-runtime.ts`), so two daemons can coexist in one + * process without sharing state. Resolution happens up front, synchronously, + * so the check-then-set below is still one uninterrupted stretch of plain + * code over plain maps. */ -import { Schema, type ManagedRuntime } from "effect"; +import { Context, Effect, Layer, Schema } from "effect"; import type { Database } from "bun:sqlite"; import { rm } from "node:fs/promises"; import { admitRun } from "./admission"; @@ -21,18 +40,16 @@ import { appendEvent, getRunEvents, listRuns } from "../persistence/store"; import type { RunRepo } from "../runtime/run"; import { startRun, type RunHandle } from "../runtime/run"; import type { GitIdentity } from "../lib/clone"; -import { allocateWorkspace } from "../lib/workspace"; -import { dedupeRegistry, type DedupeRegistry } from "../lib/dedupe"; -import type { AgentRuntime } from "../runtime/agent-runtime"; +import { allocateWorkspace, RefreshGates } from "../lib/workspace"; +import { DedupeRegistry } from "../lib/dedupe"; import type { DispatchChildFn, WorkflowDefinition, WorkspaceKind } from "../workflow"; -import { publish } from "./pubsub"; +import { serviceOf, type DaemonRuntime } from "./daemon-runtime"; +import { RunPubSub } from "./pubsub"; interface ReservedSlot { cancelled: boolean; } -const active = new Map | ReservedSlot>(); - /** * Issue #14: how deep a parent → child → grandchild chain may nest — the cap * that keeps a workflow which dispatches itself from filling the daemon. @@ -50,6 +67,30 @@ function isReserved(entry: RunHandle | ReservedSlot | undefined): boole return entry !== undefined && !("result" in entry) && "cancelled" in entry; } +/** + * A start refused because the daemon is shutting down — its registry stopped + * admitting runs. HTTP maps it to 503; a `ctx.dispatch` surfaces it to the + * parent, which is itself being cancelled. + */ +export class DaemonShuttingDownError extends Schema.TaggedError()( + "DaemonShuttingDownError", + { message: Schema.String }, +) { + // Same `.stack` header fix as `ConcurrencyLimitError` below. + static { + this.prototype.name = "DaemonShuttingDownError"; + } + + static of(): DaemonShuttingDownError { + return new DaemonShuttingDownError({ + message: "the daemon is shutting down and no longer starts runs", + }); + } +} + +/** How long shutdown waits for cancelled runs to settle before giving up on them. */ +export const DEFAULT_SHUTDOWN_TIMEOUT_MS = 5_000; + export class ConcurrencyLimitError extends Schema.TaggedError()( "ConcurrencyLimitError", { maxConcurrentRuns: Schema.Number, message: Schema.String }, @@ -72,39 +113,158 @@ export class ConcurrencyLimitError extends Schema.TaggedError; + getActiveHandle(runId: string): RunHandle | undefined; + /** + * Cancellation for a run this registry holds — whether it is already + * running (returns its handle), still reserving/allocation-bound (marks the + * slot so the run is cancelled the moment it starts), or unknown + * (`undefined`). + */ + cancelRegisteredRun( + runId: string, + ): + | { readonly kind: "handle"; readonly handle: RunHandle } + | { readonly kind: "reserved" } + | undefined; + reserve(runId: string): void; + /** True once `shutdown` has begun: starts are refused from then on. */ + readonly closed: boolean; + /** + * Close admission, cancel every run held — running handles directly, + * reserved slots by marking them for cancel-on-start — and resolve once the + * registry is empty and every cancelled run has `settled` (its agent + * processes torn down), or `timeoutMs` has passed (runs still pending then + * are logged). Idempotent: every call returns the first call's promise. + */ + shutdown(timeoutMs: number): Promise; + /** + * After a timed-out `shutdown`: wait up to `timeoutMs` more for the runs it + * cancelled to settle — e.g. once disposing the runtime has interrupted + * what was still holding them — so their final `RunCancelled` lands before + * the process exits. Resolves at once when nothing is pending. + */ + awaitCancelled(timeoutMs: number): Promise; + setHandle(runId: string, handle: RunHandle): void; + delete(runId: string): void; + get(runId: string): RunHandle | ReservedSlot | undefined; + size: number; } -export function activeRunIds(): ReadonlyArray { - return [...active.keys()]; +export function createRunRegistry(): RunRegistryShape { + const active = new Map | ReservedSlot>(); + let closed = false; + let shutdownPromise: Promise | undefined; + const onEmpty: Array<() => void> = []; + // The runs shutdown cancelled (or that started after it began), so it can + // wait for their `settled`, not merely for the registry to empty. + const cancelled: Array> = []; + const allSettled = (): Promise => + new Promise((resolve) => { + onEmpty.push(resolve); + notifyIfEmpty(); + }) + .then(() => Promise.allSettled(cancelled.map((handle) => handle.settled))) + .then(() => undefined); + const within = async (promise: Promise, timeoutMs: number): Promise => { + let timer: ReturnType | undefined; + const timedOut = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), timeoutMs); + }); + const outcome = await Promise.race([promise.then(() => true as const), timedOut]); + clearTimeout(timer); + return outcome; + }; + const notifyIfEmpty = (): void => { + if (active.size === 0) for (const resolve of onEmpty.splice(0)) resolve(); + }; + return { + isActive: (runId) => active.has(runId), + activeRunIds: () => [...active.keys()], + getActiveHandle(runId) { + const entry = active.get(runId); + if (entry === undefined || isReserved(entry)) return undefined; + return entry as RunHandle; + }, + cancelRegisteredRun(runId) { + const entry = active.get(runId); + if (entry === undefined) return undefined; + if (isReserved(entry)) { + (entry as ReservedSlot).cancelled = true; + return { kind: "reserved" }; + } + return { kind: "handle", handle: entry as RunHandle }; + }, + reserve(runId) { + active.set(runId, { cancelled: false }); + }, + setHandle(runId, handle) { + active.set(runId, handle); + if (closed) cancelled.push(handle); + }, + delete(runId) { + active.delete(runId); + notifyIfEmpty(); + }, + get closed() { + return closed; + }, + shutdown(timeoutMs) { + if (shutdownPromise !== undefined) return shutdownPromise; + closed = true; + for (const entry of active.values()) { + if (isReserved(entry)) { + (entry as ReservedSlot).cancelled = true; + } else { + const handle = entry as RunHandle; + cancelled.push(handle); + void handle.cancel(); + } + } + shutdownPromise = within(allSettled(), timeoutMs).then((settled) => { + if (!settled) { + console.error( + `[daemon] shutdown: run(s) did not settle within ${timeoutMs}ms: ` + + `${[...active.keys()].join(", ") || "(agent teardown pending)"}`, + ); + } + }); + return shutdownPromise; + }, + async awaitCancelled(timeoutMs) { + await within(allSettled(), timeoutMs); + }, + get(runId) { + return active.get(runId); + }, + get size() { + return active.size; + }, + }; } -export function getActiveHandle(runId: string): RunHandle | undefined { - const entry = active.get(runId); - if (entry === undefined || isReserved(entry)) return undefined; - return entry as RunHandle; -} +/** + * The active-run registry as a daemon service (#38, ADR 0009 §5). Resolve it + * from the daemon runtime's context; provide it with `RunRegistryLayer`. + */ +export class RunRegistry extends Context.Service()("RunRegistry") {} /** - * Cancellation for a run this process holds — whether it is already running - * (returns its handle), still reserving/allocation-bound (marks the slot so - * the run is cancelled the moment it starts), or unknown (`undefined`). + * A fresh, empty registry per layer build — i.e. per daemon runtime. Its + * finalizer shuts the registry down, so disposing a daemon runtime never + * orphans a run. It is the safety net, not the main path: by the time it + * runs the runtime is already disposed, so a reserved run that starts during + * it cannot resolve its adapter (`DaemonHandle.stop` shuts the registry down + * first, while the runtime is live, and this is then a no-op). */ -export function cancelRegisteredRun(runId: string): - | { readonly kind: "handle"; readonly handle: RunHandle } - | { - readonly kind: "reserved"; - } - | undefined { - const entry = active.get(runId); - if (entry === undefined) return undefined; - if (isReserved(entry)) { - (entry as ReservedSlot).cancelled = true; - return { kind: "reserved" }; - } - return { kind: "handle", handle: entry as RunHandle }; -} +export const RunRegistryLayer: Layer.Layer = Layer.effect( + RunRegistry, + Effect.acquireRelease(Effect.sync(createRunRegistry), (registry) => + Effect.promise(() => registry.shutdown(DEFAULT_SHUTDOWN_TIMEOUT_MS)), + ), +); export interface WorkspaceSpec { readonly workspaceRoot: string; @@ -127,8 +287,6 @@ export interface StartTrackedRunOptions { * start requests cannot both slip past it. Absent, no limit applies. */ readonly maxConcurrentRuns?: number; - /** Injectable for tests: holds the reserved-but-not-started window open. */ - readonly beforeStart?: () => Promise; /** * Issue #14: the environment a child run of this run starts with, so * `ctx.dispatch` can allocate a workspace, admission-limit and repo for it. @@ -156,11 +314,6 @@ export interface StartTrackedRunOptions { * re-asserted here. Release still happens here, keyed to this run id. */ readonly dedupeKeyClaimed?: boolean; - /** - * Issue #15: injectable holder registry, for tests. Absent, the daemon's - * shared process-wide registry (`lib/dedupe.ts`) is used. - */ - readonly dedupeRegistry?: DedupeRegistry; /** * Issue #16: the schedule that started this run, when any did - passed * through to `RunStarted.scheduleId`. @@ -194,8 +347,6 @@ export interface DispatchEnv { /** Issue #14: dispatch depth / per-run child caps, over the defaults. */ readonly maxDispatchDepth?: number; readonly maxChildrenPerRun?: number; - /** Issue #15: injectable holder registry, over the daemon's shared one. */ - readonly dedupeRegistry?: DedupeRegistry; } /** @@ -234,7 +385,7 @@ function dispatchDepth(db: Database, runId: string): number { * parent's. */ async function dispatchChildRun( - runtime: ManagedRuntime.ManagedRuntime, + runtime: DaemonRuntime, db: Database, env: DispatchEnv, parentRunId: string, @@ -244,11 +395,13 @@ async function dispatchChildRun( ): Promise { const maxDepth = env.maxDispatchDepth ?? DEFAULT_MAX_DISPATCH_DEPTH; const maxChildren = env.maxChildrenPerRun ?? DEFAULT_MAX_CHILDREN_PER_RUN; - const registry = env.dedupeRegistry ?? dedupeRegistry; + const registry = serviceOf(runtime, RunRegistry); + const dedupeRegistry = serviceOf(runtime, DedupeRegistry); + if (registry.closed) throw DaemonShuttingDownError.of(); if ( env.maxConcurrentRuns !== undefined && - !admitRun(env.maxConcurrentRuns, activeRunIds().length) + !admitRun(env.maxConcurrentRuns, registry.activeRunIds().length) ) { throw ConcurrencyLimitError.of({ maxConcurrentRuns: env.maxConcurrentRuns }); } @@ -272,7 +425,7 @@ async function dispatchChildRun( // and *before* the fire-and-forget start, so a collision throws into the // parent here instead of being swallowed by the un-awaited start's catch. const childRunId = `run-${crypto.randomUUID()}`; - if (opts?.dedupeKey !== undefined) registry.claim(opts.dedupeKey, childRunId); + if (opts?.dedupeKey !== undefined) dedupeRegistry.claim(opts.dedupeKey, childRunId); void (async () => { await startTrackedRun(runtime, db, child, { @@ -285,10 +438,9 @@ async function dispatchChildRun( dispatchEnv: env, ...(opts?.dedupeKey !== undefined ? { dedupeKey: opts.dedupeKey } : {}), ...(opts?.dedupeKey !== undefined ? { dedupeKeyClaimed: true } : {}), - ...(env.dedupeRegistry !== undefined ? { dedupeRegistry: env.dedupeRegistry } : {}), }).catch((err: unknown) => { if (opts?.dedupeKey !== undefined) { - (env.dedupeRegistry ?? dedupeRegistry).release(opts.dedupeKey, childRunId); + dedupeRegistry.release(opts.dedupeKey, childRunId); } console.error( `nested run start failed (parent ${parentRunId}, child ${childRunId}):` + @@ -304,9 +456,16 @@ function countDispatchedChildren(db: Database, runId: string): number { return getRunEvents(db, runId).filter((event) => event.payload._tag === "RunDispatched").length; } -/** Starts a run, persists+publishes every event, and tracks it until terminal. */ +/** + * Starts a run, persists+publishes every event, and tracks it until terminal. + * + * `runtime` is the daemon's composition root (#36/#38): the run's agent steps + * resolve their adapter from it, and this function resolves the registry, + * pubsub, dedupe registry and refresh gates from it — synchronously, before + * the admission check, so resolution never opens a gap inside it. + */ export async function startTrackedRun( - runtime: ManagedRuntime.ManagedRuntime, + runtime: DaemonRuntime, db: Database, workflow: WorkflowDefinition, options: StartTrackedRunOptions, @@ -317,30 +476,41 @@ export async function startTrackedRun( // unrelated runs — one run's SSE watcher can then see the other's terminal event and close // its own db while its real run is still writing to it. const runId = options.runId ?? `run-${crypto.randomUUID()}`; - - const existing = active.get(runId); - if (existing !== undefined && !isReserved(existing)) { + const registry = serviceOf(runtime, RunRegistry); + const dedupeRegistry = serviceOf(runtime, DedupeRegistry); + const pubsub = serviceOf(runtime, RunPubSub); + const refreshGates = serviceOf(runtime, RefreshGates); + + // Shutdown closed admission: nothing new starts, so nothing escapes the + // cancel-and-wait. + if (registry.closed) throw DaemonShuttingDownError.of(); + + // Running or still reserving alike: a run id is started once. (A reserved + // entry here can only be a concurrent start with the same explicit runId, + // which must not slip past admission and start a duplicate.) + if (registry.get(runId) !== undefined) { throw new Error(`run ${runId} is already active`); } + // D29 admission is decided before the dedupe claim, so a refused start + // throws having claimed nothing — otherwise its key would stay held by a + // run that never started (and a `"skip"` schedule would skip forever). + const limit = options.maxConcurrentRuns; + if (limit !== undefined && !admitRun(limit, registry.size)) { + throw ConcurrencyLimitError.of({ maxConcurrentRuns: limit }); + } // Issue #15: claim the dedupe key synchronously — check-then-claim with no // `await` in between, the same atomicity the registry slot reservation has — // so two near-simultaneous starts on the same key cannot both slip past. A // collision throws before anything is started, leaving no trace. - const registry = options.dedupeRegistry ?? dedupeRegistry; if (options.dedupeKey !== undefined && options.dedupeKeyClaimed !== true) { - registry.claim(options.dedupeKey, runId); - } - if (existing === undefined && options.maxConcurrentRuns !== undefined) { - if (!admitRun(options.maxConcurrentRuns, active.size)) { - throw ConcurrencyLimitError.of({ maxConcurrentRuns: options.maxConcurrentRuns }); - } - active.set(runId, { cancelled: false }); + dedupeRegistry.claim(options.dedupeKey, runId); } + // Every run reserves — not only limited ones — so shutdown sees runs that + // are still allocating whether or not an admission limit applies. + registry.reserve(runId); try { - if (options.beforeStart !== undefined) await options.beforeStart(); - // Issue #13: a scratch workspace takes its kind from the workflow and // never evicts clone workspaces — leftover scratch dirs (kept failures) // are excluded from retention via the run log's recorded kinds. @@ -363,7 +533,8 @@ export async function startTrackedRun( ...options.workspace, kind, ...(scratchEntries !== undefined ? { scratchEntries } : {}), - protectedEntries: [runId, ...activeRunIds()], + protectedEntries: [runId, ...registry.activeRunIds()], + refreshGates, })); if (dir === undefined) throw new Error("startTrackedRun needs `dir` or `workspace`"); @@ -396,27 +567,27 @@ export async function startTrackedRun( input: options.input, onEvent: (event) => { appendEvent(db, event); - publish(runId, event); + pubsub.publish(runId, event); }, }); // A cancel that arrived while this run was only a reserved slot is // deferred into run start (L1): the run starts, is cancelled immediately, // and ends as a clean RunCancelled instead of orphaning the slot. - const beforeStartEntry = active.get(runId); + const beforeStartEntry = registry.get(runId); const reservedSlot = isReserved(beforeStartEntry) ? (beforeStartEntry as ReservedSlot) : undefined; const cancelRequested = reservedSlot?.cancelled === true; - active.set(runId, handle); + registry.setHandle(runId, handle); void handle.result.finally(() => { - if (active.get(runId) === handle) active.delete(runId); + if (registry.get(runId) === handle) registry.delete(runId); // Issue #15: any terminal state — completed, failed, cancelled — // releases the run's key. (An interrupted run — process death — is - // covered by the registry being per-process: the new process holds + // covered by the registry being per-daemon: a new daemon holds // nothing.) - if (options.dedupeKey !== undefined) registry.release(options.dedupeKey, runId); + if (options.dedupeKey !== undefined) dedupeRegistry.release(options.dedupeKey, runId); }); // Issue #13: a scratch dir is reaped when the run succeeds — there is no @@ -433,9 +604,9 @@ export async function startTrackedRun( return runId; } catch (err) { - if (options.dedupeKey !== undefined) registry.release(options.dedupeKey, runId); - const current = active.get(runId); - if (current === undefined || isReserved(current)) active.delete(runId); + if (options.dedupeKey !== undefined) dedupeRegistry.release(options.dedupeKey, runId); + const current = registry.get(runId); + if (current === undefined || isReserved(current)) registry.delete(runId); throw err; } } diff --git a/src/server/scheduler.test.ts b/src/server/scheduler.test.ts index 4e8f5c0..f9c2e23 100644 --- a/src/server/scheduler.test.ts +++ b/src/server/scheduler.test.ts @@ -4,13 +4,16 @@ * skipped by default via the schedule's own id as a dedupe key, and no * replay of windows the daemon missed while it was down. * - * Driven entirely by injected wall-clock (`deps.now`): the "due" decision is + * Driven by an injected Clock (#38): the "due" decision is * `Cron.next(cron, lastTick) <= now`, so the cron itself never needs the - * real clock in a test. + * real clock in a test. The clock comes from Effect's real `TestClock` + * (`effect/testing`), not a hand-rolled fake — see `createTestClock` below + * for how it is bridged into `tickOnce`'s plain-async world. */ import { describe, expect, test } from "bun:test"; -import { Cron } from "effect"; +import { Clock, Cron, Effect } from "effect"; +import { TestClock } from "effect/testing"; import { createDedupeRegistry } from "../lib/dedupe"; import type { WorkflowDefinition } from "../workflow"; import { ConcurrencyLimitError, DispatchCapError } from "./runs"; @@ -45,6 +48,30 @@ function schedule(overrides: Partial = {}): RuntimeSchedule { }; } +/** + * `tickOnce` is deliberately a plain async function (ADR 0009 §5), and it + * only ever reads the clock synchronously via `currentTimeMillisUnsafe()` — + * it never suspends on `Effect.sleep`. That means none of `TestClock`'s + * fiber-coordination machinery (scheduled sleeps, the "hung test" warning + * fiber) is exercised here; what's needed from it is just a `Clock.Clock` + * whose time we can move. A real `TestClock` still satisfies that cleanly: + * `TestClock.make()` is built once per fixture via `Effect.runPromise`, and + * `setTime` bridges back into the plain-async test body the same way + * `ManagedRuntime` would for any other Effect service consumed from + * imperative code. + */ +async function createTestClock(initialTime: number): Promise<{ + clock: Clock.Clock; + setTime: (t: number) => Promise; +}> { + const testClock = await Effect.runPromise(Effect.scoped(TestClock.make())); + await Effect.runPromise(testClock.setTime(initialTime)); + return { + clock: testClock, + setTime: (t: number) => Effect.runPromise(testClock.setTime(t)), + }; +} + interface Fixture { deps: () => SchedulerDeps; readonly fired: Array; @@ -52,26 +79,27 @@ interface Fixture { /** Per-schedule injected fire failures. */ readonly fireError: Map; /** The fixture's injectable clock. */ - readonly setTime: (t: number) => void; + readonly setTime: (t: number) => Promise; } -function fixture(schedules: ReadonlyArray): Fixture { +async function fixture(schedules: ReadonlyArray): Promise { const registry = createDedupeRegistry(); const fired: Array = []; // Per-schedule injected fire failures. const fireError: Map = new Map(); - let now = 0; + const { clock, setTime } = await createTestClock(0); + const deps: SchedulerDeps = { schedules, - registry, + dedupeRegistry: registry, + clock, fire: async (s) => { fired.push(s.id); const error = fireError.get(s.id); if (error !== undefined) throw error; return `run-for-${s.id}`; }, - now: () => now, }; return { @@ -79,7 +107,7 @@ function fixture(schedules: ReadonlyArray): Fixture { fired, registry, fireError, - setTime: (t: number) => (now = t), + setTime, }; } @@ -88,38 +116,38 @@ const DAY01_0400 = Date.parse("2026-01-01T04:00:00Z"); describe("scheduler tick (issue #16)", () => { test("fires when its cron window fell between lastTick and now", async () => { - const fx = fixture([schedule()]); + const fx = await fixture([schedule()]); // state created before the window (03:05), tick after it (04:00) - fx.setTime(DAY01_0259); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); const results = await tickOnce(fx.deps(), state); expect(results).toEqual([{ scheduleId: "nightly", action: "fired", runId: "run-for-nightly" }]); }); test("does not fire twice for the same window across ticks", async () => { - const fx = fixture([schedule()]); - fx.setTime(DAY01_0259); + const fx = await fixture([schedule()]); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); await tickOnce(fx.deps(), state); - fx.setTime(DAY01_0400 + 1); + await fx.setTime(DAY01_0400 + 1); await tickOnce(fx.deps(), state); expect(fx.fired).toEqual(["nightly"]); }); test("skips the window while its previous run still holds the dedupe key (overlap)", async () => { - const fx = fixture([schedule()]); - fx.setTime(DAY01_0259); + const fx = await fixture([schedule()]); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); // A previous run of this schedule is still going. fx.registry.claim("schedule:nightly", "run-previous"); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); const results = await tickOnce(fx.deps(), state); expect(results).toEqual([{ scheduleId: "nightly", action: "skipped-overlap" }]); @@ -127,24 +155,24 @@ describe("scheduler tick (issue #16)", () => { }); test("overlap: 'stack' fires regardless of the previous run", async () => { - const fx = fixture([schedule({ overlap: "stack" })]); - fx.setTime(DAY01_0259); + const fx = await fixture([schedule({ overlap: "stack" })]); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); fx.registry.claim("schedule:nightly", "run-previous"); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); const results = await tickOnce(fx.deps(), state); expect(results).toEqual([{ scheduleId: "nightly", action: "fired", runId: "run-for-nightly" }]); }); test("a rejected fire (e.g. concurrency limit) skips the window instead of dying", async () => { - const fx = fixture([schedule()]); - fx.setTime(DAY01_0259); + const fx = await fixture([schedule()]); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); fx.fireError.set("nightly", ConcurrencyLimitError.of({ maxConcurrentRuns: 1 })); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); const results = await tickOnce(fx.deps(), state); expect(results).toEqual([{ scheduleId: "nightly", action: "skipped-concurrency" }]); @@ -158,14 +186,14 @@ describe("scheduler tick (issue #16)", () => { // Every-minute cron, three windows pass between two ticks after the // daemon "came up": one fire, not three — and the pre-start windows // (12:00:00–12:00:30 before the state existed) never fire at all. - const fx = fixture([schedule({ cron: Cron.parseUnsafe("* * * * *", "UTC") })]); + const fx = await fixture([schedule({ cron: Cron.parseUnsafe("* * * * *", "UTC") })]); const start = Date.parse("2026-01-01T12:00:00Z"); - fx.setTime(start); + await fx.setTime(start); const state = createSchedulerState(fx.deps()); - fx.setTime(start + 30_000); + await fx.setTime(start + 30_000); await tickOnce(fx.deps(), state); - fx.setTime(start + 3 * 60_000); + await fx.setTime(start + 3 * 60_000); const results = await tickOnce(fx.deps(), state); expect(fx.fired).toEqual(["nightly"]); @@ -173,7 +201,7 @@ describe("scheduler tick (issue #16)", () => { }); test("runOnStart fires once at scheduler start, before any cron window", async () => { - const fx = fixture([schedule({ runOnStart: true })]); + const fx = await fixture([schedule({ runOnStart: true })]); const state = createSchedulerState(fx.deps()); expect(state.runOnStartPending).toEqual(new Set(["nightly"])); @@ -188,12 +216,12 @@ describe("scheduler tick (issue #16)", () => { }); test("an independent schedule is unaffected by another's failure", async () => { - const fx = fixture([schedule({ id: "a" }), schedule({ id: "b" })]); - fx.setTime(DAY01_0259); + const fx = await fixture([schedule({ id: "a" }), schedule({ id: "b" })]); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); fx.fireError.set("a", new Error("boom")); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); const results = await tickOnce(fx.deps(), state); expect(results.filter((r) => r.scheduleId === "a")).toEqual([ @@ -205,32 +233,32 @@ describe("scheduler tick (issue #16)", () => { }); test("each domain error type is explicitly classified (exhaustive match, #34)", async () => { - const concurrencyFx = fixture([schedule({ id: "conc" })]); - concurrencyFx.setTime(DAY01_0259); + const concurrencyFx = await fixture([schedule({ id: "conc" })]); + await concurrencyFx.setTime(DAY01_0259); const concurrencyState = createSchedulerState(concurrencyFx.deps()); concurrencyFx.fireError.set("conc", ConcurrencyLimitError.of({ maxConcurrentRuns: 1 })); - concurrencyFx.setTime(DAY01_0400); + await concurrencyFx.setTime(DAY01_0400); expect(await tickOnce(concurrencyFx.deps(), concurrencyState)).toEqual([ { scheduleId: "conc", action: "skipped-concurrency" }, ]); - const dedupeFx = fixture([schedule({ id: "dedupe" })]); - dedupeFx.setTime(DAY01_0259); + const dedupeFx = await fixture([schedule({ id: "dedupe" })]); + await dedupeFx.setTime(DAY01_0259); const dedupeState = createSchedulerState(dedupeFx.deps()); dedupeFx.fireError.set( "dedupe", DedupeKeyError.of({ key: "schedule:dedupe", holderRunId: "run-x" }), ); - dedupeFx.setTime(DAY01_0400); + await dedupeFx.setTime(DAY01_0400); expect(await tickOnce(dedupeFx.deps(), dedupeState)).toEqual([ { scheduleId: "dedupe", action: "fire-failed" }, ]); - const capFx = fixture([schedule({ id: "cap" })]); - capFx.setTime(DAY01_0259); + const capFx = await fixture([schedule({ id: "cap" })]); + await capFx.setTime(DAY01_0259); const capState = createSchedulerState(capFx.deps()); capFx.fireError.set("cap", new DispatchCapError({ message: "depth exceeded" })); - capFx.setTime(DAY01_0400); + await capFx.setTime(DAY01_0400); expect(await tickOnce(capFx.deps(), capState)).toEqual([ { scheduleId: "cap", action: "fire-failed" }, ]); @@ -245,12 +273,12 @@ describe("scheduler tick (issue #16)", () => { class UnknownTaggedError extends Error { readonly _tag = "SomeFutureDomainError"; } - const fx = fixture([schedule()]); - fx.setTime(DAY01_0259); + const fx = await fixture([schedule()]); + await fx.setTime(DAY01_0259); const state = createSchedulerState(fx.deps()); fx.fireError.set("nightly", new UnknownTaggedError("mystery failure")); - fx.setTime(DAY01_0400); + await fx.setTime(DAY01_0400); const results = await tickOnce(fx.deps(), state); expect(results).toEqual([{ scheduleId: "nightly", action: "fire-failed" }]); diff --git a/src/server/scheduler.ts b/src/server/scheduler.ts index 6a97d67..79bd710 100644 --- a/src/server/scheduler.ts +++ b/src/server/scheduler.ts @@ -22,12 +22,18 @@ * non-terminal run holds it. The fired run itself is started with the same * key by the daemon's fire function, which makes the skip observable exactly * like any other dedupe collision; `"stack"` fires regardless. + * + * #38: the clock comes from Effect's Clock service. Tests use TestClock or + * provide a custom Clock instance rather than an injected `now` function. + * In the daemon, `runSchedulerLoop` resolves both the clock and the dedupe + * registry from the context of the daemon runtime it is forked on; the plain + * `tickOnce` receives them as ordinary arguments in `SchedulerDeps`. */ -import { Cron, Effect, Schedule, Schema, type ManagedRuntime } from "effect"; +import { Clock, Cron, Effect, Schedule, Schema } from "effect"; import type { Database } from "bun:sqlite"; import type { FactoryConfig } from "../config"; -import { DedupeKeyError, dedupeRegistry, type DedupeRegistry } from "../lib/dedupe"; +import { DedupeKeyError, DedupeRegistry, type DedupeRegistryShape } from "../lib/dedupe"; import type { WorkflowDefinition } from "../workflow"; import { ConcurrencyLimitError, @@ -37,7 +43,7 @@ import { type WorkspaceSpec, } from "./runs"; import { RunCancelledSignal, type RunRepo } from "../runtime/run"; -import type { AgentRuntime } from "../runtime/agent-runtime"; +import type { DaemonRuntime } from "./daemon-runtime"; export class SchedulerError extends Schema.TaggedError()("SchedulerError", { cause: Schema.Defect(), @@ -95,10 +101,10 @@ export interface SchedulerDeps { readonly schedules: ReadonlyArray; /** Starts the scheduled run; the daemon wires it to `startTrackedRun`. */ readonly fire: (schedule: RuntimeSchedule) => Promise; - /** Injectable for tests; defaults to the daemon's shared registry. */ - readonly registry?: DedupeRegistry; - /** Injectable for tests; defaults to `Date.now`. */ - readonly now?: () => number; + /** Issue #38: the daemon's own dedupe registry — where schedule keys are held. */ + readonly dedupeRegistry: DedupeRegistryShape; + /** Issue #38: the wall clock windows are judged against; tests pass a `TestClock`. */ + readonly clock: Clock.Clock; } export interface SchedulerState { @@ -114,7 +120,7 @@ export interface SchedulerState { * once per session. */ export function createSchedulerState(deps: SchedulerDeps): SchedulerState { - const now = deps.now?.() ?? Date.now(); + const now = deps.clock.currentTimeMillisUnsafe(); const schedules = deps.schedules; return { lastTick: new Map(schedules.map((s) => [s.id, now])), @@ -173,8 +179,8 @@ export async function tickOnce( deps: SchedulerDeps, state: SchedulerState, ): Promise> { - const now = deps.now?.() ?? Date.now(); - const registry = deps.registry ?? dedupeRegistry; + const now = deps.clock.currentTimeMillisUnsafe(); + const registry = deps.dedupeRegistry; const results: Array = []; for (const schedule of deps.schedules) { @@ -217,8 +223,31 @@ export async function tickOnce( return results; } -/** The daemon's scheduler loop — `Effect.repeat` around the plain `tickOnce`. */ +/** + * The daemon's scheduler loop — `Effect.repeat` around the plain `tickOnce`. + * The dedupe registry and the clock come from the context it runs in (#38): + * forked on the daemon runtime, that is the daemon's own registry and + * Effect's wall clock. The session state (`lastTick` = now) is taken when the + * loop starts. + */ export function runSchedulerLoop( + schedules: ReadonlyArray, + fire: SchedulerDeps["fire"], + intervalMs: number, +): Effect.Effect { + return Effect.gen(function* () { + const deps: SchedulerDeps = { + schedules, + fire, + dedupeRegistry: yield* DedupeRegistry, + clock: yield* Clock.Clock, + }; + const state = createSchedulerState(deps); + return yield* schedulerTicks(deps, state, intervalMs); + }); +} + +function schedulerTicks( deps: SchedulerDeps, state: SchedulerState, intervalMs: number, @@ -252,7 +281,7 @@ export function runSchedulerLoop( */ export function makeScheduleFire(options: { readonly db: Database; - readonly runtime: ManagedRuntime.ManagedRuntime; + readonly runtime: DaemonRuntime; readonly maxConcurrentRuns: number; readonly workspace: WorkspaceSpec; readonly repo: RunRepo; diff --git a/src/server/shutdown.test.ts b/src/server/shutdown.test.ts new file mode 100644 index 0000000..48da70c --- /dev/null +++ b/src/server/shutdown.test.ts @@ -0,0 +1,359 @@ +/** + * Shutdown cancels active runs (#38): `DaemonHandle.stop` — the path + * `factory serve`'s SIGINT/SIGTERM handler takes — cancels every run the + * daemon holds and waits for it to settle before the runtime goes away, so a + * run ends as a recorded `RunCancelled` with its `ctx.exec` child killed, + * rather than as an orphaned process and an "interrupted" run after restart. + */ + +import { existsSync, mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "bun:test"; +import { Schema } from "effect"; +import { defineConfig } from "../config"; +import { RefreshGates } from "../lib/workspace"; +import { getRunEvents, openStore } from "../persistence/store"; +import { createSlowFakeAdapter } from "../replay/adapter"; +import type { AgentAdapter, AgentAdapterYield } from "../runtime/agent-adapter"; +import { makeAgentRuntime } from "../runtime/agent-runtime"; +import { startRun } from "../runtime/run"; +import type { RunEvent } from "../events"; +import { defineWorkflow } from "../workflow"; +import { startDaemon, type DaemonHandle } from "./daemon"; +import { makeDaemonRuntime, serviceOf } from "./daemon-runtime"; +import { RunRegistry, startTrackedRun } from "./runs"; + +/** Writes its shell's pid, then becomes a 300s sleep — a child that outlives any test. */ +const longExec = defineWorkflow("shutdown-long-exec", { + input: Schema.Struct({ pidFile: Schema.String }), + workspace: { kind: "scratch" }, + run: async (ctx, input) => { + await ctx.exec(["sh", "-c", `echo $$ > ${input.pidFile}; exec sleep 300`]); + return {}; + }, +}); + +/** The same, on a clone workspace, so its allocation can be held open. */ +const cloneLongExec = defineWorkflow("shutdown-clone-long-exec", { + input: Schema.Struct({}), + run: async (ctx) => { + await ctx.exec(["sleep", "300"]); + return {}; + }, +}); + +/** One agent step and nothing else: the run is mid-step until cancelled. */ +const agentStep = defineWorkflow("shutdown-agent-step", { + input: Schema.Struct({}), + workspace: { kind: "scratch" }, + run: async (ctx) => { + await ctx.agent("think", "waiting on the model"); + return {}; + }, +}); + +/** + * An adapter waiting on the model, like opencode between chunks: its + * generator's `next()` is pending until the step's abort reaches it, then + * takes `abortLatencyMs` to wind down (the HTTP `session.abort()` round trip) + * before its `finally` — the process kill — runs. Before #38's fix the + * stream's implicit `return()` queued behind that `next()`, and the abort that + * would release it only fired after `return()` finished: a deadlock until the + * next chunk. `cooperative: false` never ends at all, abort or not. + */ +function waitingAdapter(options: { + readonly cooperative: boolean; + readonly abortLatencyMs?: number; +}): AgentAdapter & { + readonly tornDown: () => boolean; +} { + let tornDown = false; + return { + tornDown: () => tornDown, + async prepareWorkspace(): Promise {}, + async *stream({ abortController }): AsyncIterable { + try { + yield { chunk: { type: "RUN_STARTED" } }; + await new Promise((resolve) => { + if (!options.cooperative) return; + abortController.signal.addEventListener("abort", () => + setTimeout(resolve, options.abortLatencyMs ?? 0), + ); + }); + } finally { + tornDown = true; + } + }, + }; +} + +function isAlive(pid: number): boolean { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + +async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (predicate()) return; + await Bun.sleep(20); + } + throw new Error("condition not met within timeout"); +} + +function tagsOf(dbPath: string, runId: string): Array { + const db = openStore(dbPath); + try { + return getRunEvents(db, runId).map((event) => event.payload._tag); + } finally { + db.close(); + } +} + +async function makeDaemon(root: string, sshUrl = join(root, "no-remote")): Promise { + return startDaemon({ + dbPath: join(root, "factory.db"), + port: 0, + config: defineConfig({ + agent: { adapter: createSlowFakeAdapter([], 1) }, + repo: { + sshUrl, + identity: { name: "Factory", email: "factory@factory.test" }, + baseBranch: "main", + slug: "acme/widgets", + }, + workflows: [longExec, cloneLongExec], + workspaceRoot: join(root, "workspaces"), + retainedWorkspaces: 10, + }), + }); +} + +async function post(handle: DaemonHandle, body: unknown): Promise { + return fetch(`http://localhost:${handle.server.port}/api/runs`, { + method: "POST", + body: JSON.stringify(body), + }); +} + +describe("daemon shutdown cancels active runs (#38)", () => { + test("stop() cancels a running exec: RunCancelled is recorded and the child is gone", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-shutdown-exec-")); + const pidFile = join(root, "child.pid"); + const handle = await makeDaemon(root); + try { + const res = await post(handle, { workflowId: "shutdown-long-exec", input: { pidFile } }); + expect(res.status).toBe(201); + const { runId } = (await res.json()) as { runId: string }; + + await waitFor(() => existsSync(pidFile) && readFileSync(pidFile, "utf8").trim() !== ""); + const pid = Number(readFileSync(pidFile, "utf8").trim()); + expect(isAlive(pid)).toBe(true); + + await handle.stop(); + + expect(tagsOf(join(root, "factory.db"), runId).at(-1)).toBe("RunCancelled"); + expect(isAlive(pid)).toBe(false); + } finally { + await handle.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + test("stop() cancels a reserved, allocation-bound run cleanly and refuses new starts meanwhile", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-shutdown-reserved-")); + const seed = join(root, "seed"); + await Bun.$`git init -b main -q ${seed}`.quiet(); + await Bun.$`git -C ${seed} -c user.name=seed -c user.email=seed@seed.local commit -q --allow-empty -m seed`.quiet(); + const handle = await makeDaemon(root, seed); + + // Hold allocation open: the mirror's refresh queues behind this gate, so + // the run is a reserved slot and nothing more until it is released. + let release: () => void = () => undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + serviceOf(handle.runtime, RefreshGates).set(join(root, "workspaces", ".mirror.git"), gate); + const registry = serviceOf(handle.runtime, RunRegistry); + + try { + // The POST itself is held by the allocation, so it is not awaited here. + const starting = post(handle, { workflowId: "shutdown-clone-long-exec", input: {} }); + await waitFor(() => registry.activeRunIds().length === 1); + const runId = registry.activeRunIds()[0]!; + expect(registry.getActiveHandle(runId)).toBeUndefined(); + + let stopped = false; + const stopping = handle.stop().then(() => { + stopped = true; + }); + + // Shutdown is waiting on the reserved run, and admits nothing new. + await Bun.sleep(50); + expect(stopped).toBe(false); + const refused = await post(handle, { workflowId: "shutdown-clone-long-exec", input: {} }); + expect(refused.status).toBe(503); + + release(); + await stopping; + expect((await starting).status).toBe(201); + + expect(registry.activeRunIds()).toEqual([]); + const tags = tagsOf(join(root, "factory.db"), runId); + expect(tags[0]).toBe("RunStarted"); + expect(tags.at(-1)).toBe("RunCancelled"); + expect(tags).not.toContain("ExecStarted"); + } finally { + release(); + await handle.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + test("stop() is idempotent", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-shutdown-idem-")); + const pidFile = join(root, "child.pid"); + const handle = await makeDaemon(root); + try { + const res = await post(handle, { workflowId: "shutdown-long-exec", input: { pidFile } }); + const { runId } = (await res.json()) as { runId: string }; + await waitFor(() => existsSync(pidFile) && readFileSync(pidFile, "utf8").trim() !== ""); + + const first = handle.stop(); + const second = handle.stop(); + expect(second).toBe(first); + await Promise.all([first, second]); + await handle.stop(); + + const tags = tagsOf(join(root, "factory.db"), runId); + expect(tags.filter((tag) => tag === "RunCancelled")).toHaveLength(1); + } finally { + rmSync(root, { recursive: true, force: true }); + } + }); + + test("disposing a daemon runtime without stop() still cancels its runs (layer finalizer)", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-shutdown-dispose-")); + const pidFile = join(root, "child.pid"); + const dbPath = join(root, "factory.db"); + const db = openStore(dbPath); + const runtime = makeDaemonRuntime(createSlowFakeAdapter([], 1)); + try { + const runId = await startTrackedRun(runtime, db, longExec, { + workspace: { + workspaceRoot: join(root, "workspaces"), + sshUrl: join(root, "no-remote"), + identity: { name: "T", email: "t@t.test" }, + retainedWorkspaces: 10, + }, + input: { pidFile }, + }); + await waitFor(() => existsSync(pidFile) && readFileSync(pidFile, "utf8").trim() !== ""); + const pid = Number(readFileSync(pidFile, "utf8").trim()); + + await runtime.dispose(); + + expect(getRunEvents(db, runId).at(-1)?.payload._tag).toBe("RunCancelled"); + expect(isAlive(pid)).toBe(false); + } finally { + db.close(); + rmSync(root, { recursive: true, force: true }); + } + }); + + test("a cancel mid-agent-step records RunCancelled promptly even if the adapter never yields again", async () => { + const adapter = waitingAdapter({ cooperative: false }); + const events: Array = []; + const handle = startRun(agentStep, makeAgentRuntime(adapter), { + runId: "run-blocked-agent", + dir: tmpdir(), + input: {}, + onEvent: (event) => events.push(event), + }); + await waitFor(() => events.some((e) => e.payload._tag === "AgentChunk")); + + const started = Date.now(); + await handle.cancel(); + expect(Date.now() - started).toBeLessThan(500); + expect(events.at(-1)?.payload._tag).toBe("RunCancelled"); + expect( + events.find((e) => e.payload._tag === "AgentStepFinished")?.payload as { outcome?: string }, + ).toMatchObject({ outcome: "cancelled" }); + }); + + test("stop() mid-agent-step finishes well within budget and the adapter is torn down", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-shutdown-agent-")); + const adapter = waitingAdapter({ cooperative: true, abortLatencyMs: 100 }); + const handle = await startDaemon({ + dbPath: join(root, "factory.db"), + port: 0, + config: defineConfig({ + agent: { adapter }, + repo: { + sshUrl: join(root, "no-remote"), + identity: { name: "Factory", email: "factory@factory.test" }, + baseBranch: "main", + slug: "acme/widgets", + }, + workflows: [agentStep], + workspaceRoot: join(root, "workspaces"), + retainedWorkspaces: 10, + }), + }); + try { + const res = await post(handle, { workflowId: "shutdown-agent-step", input: {} }); + const { runId } = (await res.json()) as { runId: string }; + await waitFor(() => tagsOf(join(root, "factory.db"), runId).includes("AgentChunk")); + + const started = Date.now(); + await handle.stop(); + expect(Date.now() - started).toBeLessThan(1_000); + expect(tagsOf(join(root, "factory.db"), runId).at(-1)).toBe("RunCancelled"); + // stop() waited for the adapter's own teardown, not just the run's end. + expect(adapter.tornDown()).toBe(true); + } finally { + await handle.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); + + test("stop() is bounded even when an adapter never tears down", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-shutdown-stuck-")); + const handle = await startDaemon({ + dbPath: join(root, "factory.db"), + port: 0, + shutdownTimeoutMs: 200, + config: defineConfig({ + agent: { adapter: waitingAdapter({ cooperative: false }) }, + repo: { + sshUrl: join(root, "no-remote"), + identity: { name: "Factory", email: "factory@factory.test" }, + baseBranch: "main", + slug: "acme/widgets", + }, + workflows: [agentStep], + workspaceRoot: join(root, "workspaces"), + retainedWorkspaces: 10, + }), + }); + try { + const res = await post(handle, { workflowId: "shutdown-agent-step", input: {} }); + const { runId } = (await res.json()) as { runId: string }; + await waitFor(() => tagsOf(join(root, "factory.db"), runId).includes("AgentChunk")); + + const started = Date.now(); + await handle.stop(); + // 200ms shutdown budget + the bounded post-dispose settle wait. + expect(Date.now() - started).toBeLessThan(3_000); + expect(tagsOf(join(root, "factory.db"), runId).at(-1)).toBe("RunCancelled"); + } finally { + await handle.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); +}); diff --git a/src/server/test-daemon.ts b/src/server/test-daemon.ts new file mode 100644 index 0000000..695d994 --- /dev/null +++ b/src/server/test-daemon.ts @@ -0,0 +1,35 @@ +/** + * Test support (#38): a fresh daemon runtime per test, with its per-daemon + * services already resolved for assertions. A new runtime is new state — the + * tests' isolation comes from building one each, not from resetting + * singletons. Resolving the services below builds the runtime's layers + * eagerly and synchronously, so a layer that cannot build that way fails here, + * at construction, not inside a test's first `serviceOf`. + */ + +import type { DedupeRegistryShape } from "../lib/dedupe"; +import { DedupeRegistry } from "../lib/dedupe"; +import { RefreshGates, type RefreshGatesShape } from "../lib/workspace"; +import type { AgentAdapter } from "../runtime/agent-adapter"; +import { makeDaemonRuntime, serviceOf, type DaemonRuntime } from "./daemon-runtime"; +import { RunPubSub, type PubSub } from "./pubsub"; +import { RunRegistry, type RunRegistryShape } from "./runs"; + +export interface TestDaemon { + readonly runtime: DaemonRuntime; + readonly registry: RunRegistryShape; + readonly pubsub: PubSub; + readonly dedupeRegistry: DedupeRegistryShape; + readonly refreshGates: RefreshGatesShape; +} + +export function createTestDaemon(adapter?: AgentAdapter): TestDaemon { + const runtime = makeDaemonRuntime(adapter); + return { + runtime, + registry: serviceOf(runtime, RunRegistry), + pubsub: serviceOf(runtime, RunPubSub), + dedupeRegistry: serviceOf(runtime, DedupeRegistry), + refreshGates: serviceOf(runtime, RefreshGates), + }; +} diff --git a/src/server/two-daemons.test.ts b/src/server/two-daemons.test.ts new file mode 100644 index 0000000..82ad6a8 --- /dev/null +++ b/src/server/two-daemons.test.ts @@ -0,0 +1,238 @@ +/** + * #38 acceptance criterion: two daemons can run in one process without + * sharing run state. This test starts two daemons on different ports, starts + * a run on each, and verifies that each daemon's registry, pubsub, and dedupe + * state is independent. Runs are held open on a gate so the cross-daemon + * checks happen while both are still live. + */ + +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, test } from "bun:test"; +import { Schema } from "effect"; +import { defineWorkflow } from "../workflow"; +import { defineConfig } from "../config"; +import { createSlowFakeAdapter } from "../replay/adapter"; +import { startDaemon } from "./daemon"; +import { isTerminal, type RunEvent } from "../events"; +import type { DaemonHandle } from "./daemon"; +import { serviceOf } from "./daemon-runtime"; +import { RunRegistry } from "./runs"; +import { RunPubSub } from "./pubsub"; +import { DedupeRegistry } from "../lib/dedupe"; +import { RefreshGates } from "../lib/workspace"; + +/** Each daemon's own services, resolved from its own runtime's context. */ +const registryOf = (daemon: DaemonHandle) => serviceOf(daemon.runtime, RunRegistry); +const pubsubOf = (daemon: DaemonHandle) => serviceOf(daemon.runtime, RunPubSub); +const dedupeOf = (daemon: DaemonHandle) => serviceOf(daemon.runtime, DedupeRegistry); + +let releaseRuns: () => void = () => undefined; +let runsGate = Promise.resolve(); + +function closeGate(): void { + runsGate = new Promise((resolve) => { + releaseRuns = resolve; + }); +} + +const scratchWorkflow = defineWorkflow("two-daemon-test", { + input: Schema.Struct({ marker: Schema.String }), + workspace: { kind: "scratch" }, + run: async (ctx, input) => { + await runsGate; + await ctx.exec(["sh", "-c", "sleep 0.1"]); + return { marker: input.marker }; + }, +}); + +async function waitForTerminal(port: number, runId: string, timeoutMs = 10_000): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const res = await fetch(`http://localhost:${port}/api/runs/${runId}`); + const run = (await res.json()) as { status: string }; + if (["RunFinished", "RunFailed", "RunCancelled"].includes(run.status)) return; + await Bun.sleep(25); + } + throw new Error(`run ${runId} did not reach terminal state within ${timeoutMs}ms`); +} + +/** + * Subscribes `runId` on both daemons' pubsubs: the owner's events land in + * `own`, anything the other daemon publishes for it lands in `foreign`. + */ +function watch( + owner: DaemonHandle, + other: DaemonHandle, + runId: string, +): { readonly own: Array; readonly foreign: Array; readonly stop: () => void } { + const own: Array = []; + const foreign: Array = []; + const unsubOwn = pubsubOf(owner).subscribe(runId, (event) => own.push(event)); + const unsubOther = pubsubOf(other).subscribe(runId, (event) => foreign.push(event)); + return { + own, + foreign, + stop: () => { + unsubOwn(); + unsubOther(); + }, + }; +} + +describe("two daemons in one process (#38)", () => { + test("two daemons have independent run registries, pubsub, and dedupe state", async () => { + const root = mkdtempSync(join(tmpdir(), "factory-two-daemons-")); + + const adapter = createSlowFakeAdapter( + [ + { type: "TEXT_MESSAGE_START" }, + { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, + { type: "TEXT_MESSAGE_END" }, + ], + 10, + ); + + const daemon1 = await startDaemon({ + dbPath: join(root, "daemon1.db"), + port: 0, + config: defineConfig({ + agent: { adapter }, + repo: { + sshUrl: join(root, "seed-not-used"), + identity: { name: "Factory", email: "factory@factory.test" }, + baseBranch: "main", + slug: "acme/widgets", + }, + workflows: [scratchWorkflow], + workspaceRoot: join(root, "workspaces1"), + retainedWorkspaces: 10, + }), + }); + + const daemon2 = await startDaemon({ + dbPath: join(root, "daemon2.db"), + port: 0, + config: defineConfig({ + agent: { adapter }, + repo: { + sshUrl: join(root, "seed-not-used"), + identity: { name: "Factory", email: "factory@factory.test" }, + baseBranch: "main", + slug: "acme/widgets", + }, + workflows: [scratchWorkflow], + workspaceRoot: join(root, "workspaces2"), + retainedWorkspaces: 10, + }), + }); + + try { + // Two runtimes, two builds of every per-daemon layer: no service + // instance is shared. + expect(registryOf(daemon1)).not.toBe(registryOf(daemon2)); + expect(pubsubOf(daemon1)).not.toBe(pubsubOf(daemon2)); + expect(dedupeOf(daemon1)).not.toBe(dedupeOf(daemon2)); + expect(serviceOf(daemon1.runtime, RefreshGates)).not.toBe( + serviceOf(daemon2.runtime, RefreshGates), + ); + + closeGate(); + const port1 = daemon1.server.port; + const port2 = daemon2.server.port; + if (port1 === undefined || port2 === undefined) { + throw new Error("daemons did not bind to ports"); + } + + const res1 = await fetch(`http://localhost:${port1}/api/runs`, { + method: "POST", + body: JSON.stringify({ workflowId: "two-daemon-test", input: { marker: "daemon1" } }), + }); + expect(res1.status).toBe(201); + const { runId: runId1 } = (await res1.json()) as { runId: string }; + + const res2 = await fetch(`http://localhost:${port2}/api/runs`, { + method: "POST", + body: JSON.stringify({ workflowId: "two-daemon-test", input: { marker: "daemon2" } }), + }); + expect(res2.status).toBe(201); + const { runId: runId2 } = (await res2.json()) as { runId: string }; + + expect(runId1).not.toBe(runId2); + + const watch1 = watch(daemon1, daemon2, runId1); + const watch2 = watch(daemon2, daemon1, runId2); + + expect(registryOf(daemon1).activeRunIds()).toEqual([runId1]); + expect(registryOf(daemon2).activeRunIds()).toEqual([runId2]); + + const list1 = (await fetch(`http://localhost:${port1}/api/runs`).then((r) => + r.json(), + )) as Array<{ runId: string }>; + const list2 = (await fetch(`http://localhost:${port2}/api/runs`).then((r) => + r.json(), + )) as Array<{ runId: string }>; + + expect(list1.map((r) => r.runId)).toEqual([runId1]); + expect(list2.map((r) => r.runId)).toEqual([runId2]); + + const detail1FromDaemon2 = await fetch(`http://localhost:${port2}/api/runs/${runId1}`); + expect(detail1FromDaemon2.status).toBe(404); + + const detail2FromDaemon1 = await fetch(`http://localhost:${port1}/api/runs/${runId2}`); + expect(detail2FromDaemon1.status).toBe(404); + + releaseRuns(); + await waitForTerminal(port1, runId1); + await waitForTerminal(port2, runId2); + + // Each run's live events reached its own daemon's pubsub, never the other's. + expect(watch1.own.some((event) => isTerminal(event.payload))).toBe(true); + expect(watch2.own.some((event) => isTerminal(event.payload))).toBe(true); + expect(watch1.own.every((event) => event.runId === runId1)).toBe(true); + expect(watch2.own.every((event) => event.runId === runId2)).toBe(true); + expect(watch1.foreign).toEqual([]); + expect(watch2.foreign).toEqual([]); + watch1.stop(); + watch2.stop(); + + expect(registryOf(daemon1).activeRunIds()).toEqual([]); + expect(registryOf(daemon2).activeRunIds()).toEqual([]); + + const dedupeKey = "shared-key"; + closeGate(); + const res3 = await fetch(`http://localhost:${port1}/api/runs`, { + method: "POST", + body: JSON.stringify({ workflowId: "two-daemon-test", input: { marker: "d1" }, dedupeKey }), + }); + expect(res3.status).toBe(201); + const { runId: runId3 } = (await res3.json()) as { runId: string }; + + // Daemon 1 still holds the key (run 3 is gated) when daemon 2 is asked + // for the same key — and daemon 2 knows nothing of it. + expect(dedupeOf(daemon1).holderOf(dedupeKey)).toBe(runId3); + expect(dedupeOf(daemon2).holderOf(dedupeKey)).toBeUndefined(); + + const res4 = await fetch(`http://localhost:${port2}/api/runs`, { + method: "POST", + body: JSON.stringify({ workflowId: "two-daemon-test", input: { marker: "d2" }, dedupeKey }), + }); + expect(res4.status).toBe(201); + const { runId: runId4 } = (await res4.json()) as { runId: string }; + + expect(runId3).not.toBe(runId4); + expect(dedupeOf(daemon1).holderOf(dedupeKey)).toBe(runId3); + expect(dedupeOf(daemon2).holderOf(dedupeKey)).toBe(runId4); + + releaseRuns(); + await waitForTerminal(port1, runId3); + await waitForTerminal(port2, runId4); + } finally { + releaseRuns(); + await daemon1.stop(); + await daemon2.stop(); + rmSync(root, { recursive: true, force: true }); + } + }); +});