diff --git a/e2e/implement-issue.test.ts b/e2e/implement-issue.test.ts index 4527ed0..cad95f6 100644 --- a/e2e/implement-issue.test.ts +++ b/e2e/implement-issue.test.ts @@ -26,6 +26,7 @@ import { join } from "node:path"; import { describe, expect, test } from "bun:test"; import { hostExec } from "../src/lib/exec"; import { createCorpusReplayAdapter } from "../src/replay/adapter"; +import { makeAgentRuntime } from "../src/runtime/agent-runtime"; import { startRun } from "../src/runtime/run"; import implementIssue from "./implement-issue"; @@ -118,16 +119,19 @@ describe("implement-issue workflow, replayed against the recorded round-trip cor process.env.PATH = `${binDir}:${originalPath}`; const events: Array = []; - const handle = startRun(implementIssue, { - runId: "test-run-corpus-replay", - dir: workDir, - input: { - issueNumber: 1, + const handle = startRun( + implementIssue, + makeAgentRuntime(createCorpusReplayAdapter(FULL_ROUND_TRIP_CORPUS)), + { + runId: "test-run-corpus-replay", + dir: workDir, + input: { + issueNumber: 1, + }, + repo: { slug: "local/fixture", baseBranch: "main" }, + onEvent: (event) => events.push(event), }, - repo: { slug: "local/fixture", baseBranch: "main" }, - adapter: createCorpusReplayAdapter(FULL_ROUND_TRIP_CORPUS), - onEvent: (event) => events.push(event), - }); + ); const outcome = await handle.result; diff --git a/e2e/server.ts b/e2e/server.ts index 99be761..0a86181 100644 --- a/e2e/server.ts +++ b/e2e/server.ts @@ -22,6 +22,7 @@ import { defineConfig } from "../src/config"; import { hostExec } from "../src/lib/exec"; import { appendEvent, openStore } from "../src/persistence/store"; import { createCorpusReplayAdapter, createSlowFakeAdapter } from "../src/replay/adapter"; +import { makeAgentRuntime } from "../src/runtime/agent-runtime"; import { startRun } from "../src/runtime/run"; import { startDaemon } from "../src/server/daemon"; import { defineWorkflow, Schema } from "../src/workflow"; @@ -137,12 +138,11 @@ async function seedCorpusRuns(root: string, db: ReturnType): P process.env.PATH = `${binDir}:${originalPath}`; await awaitRun( - startRun(implementIssue, { + startRun(implementIssue, makeAgentRuntime(createCorpusReplayAdapter(CORPUS_ROUND_TRIP)), { runId: "run-static-corpus", dir: workDir, repo: { slug: "local/fixture", baseBranch: "main" }, input: { issueNumber: 1 }, - adapter: createCorpusReplayAdapter(CORPUS_ROUND_TRIP), onEvent: (event) => appendEvent(db, event), }), ); @@ -151,11 +151,10 @@ async function seedCorpusRuns(root: string, db: ReturnType): P // A second, cheap run so list ordering has more than one data point. await awaitRun( - startRun(echoWorkflow, { + startRun(echoWorkflow, makeAgentRuntime(createCorpusReplayAdapter(CORPUS_ONE_STEP)), { runId: "run-static-echo", dir: workDir, input: {}, - adapter: createCorpusReplayAdapter(CORPUS_ONE_STEP), onEvent: (event) => appendEvent(db, event), }), ); @@ -253,7 +252,6 @@ async function main(): Promise { db.close(); } - const adapter = createSlowFakeAdapter(SLOW_CHUNKS, 1_000); const config = defineConfig({ repo: { sshUrl: remoteDir, @@ -278,8 +276,9 @@ async function main(): Promise { timezone: "UTC", }, ], + agent: { adapter: createSlowFakeAdapter(SLOW_CHUNKS, 1_000) }, }); - const { server } = await startDaemon({ dbPath, port, adapter, config }); + const { server } = await startDaemon({ dbPath, port, config }); console.log(`factory e2e: listening on http://localhost:${server.port}`); } diff --git a/src/cli-argv.test.ts b/src/cli-argv.test.ts index 572c942..f57742d 100644 --- a/src/cli-argv.test.ts +++ b/src/cli-argv.test.ts @@ -419,3 +419,57 @@ describe("factory binary: process exit codes", () => { }); }); }); + +/** + * PR #56 merged this branch onto `33-parse-cli-with-effect-cli` with a bad + * conflict resolution: `src/cli.ts`'s `import.meta.main` block regressed to + * the pre-#52 hand-rolled `USAGE`/`parseFlags`/`usageError` parser while + * `factoryCommand` above kept passing — the tests here only ever drove + * `factoryCommand` directly, never the actual binary entrypoint, so CI stayed + * green while the shipped CLI silently lost `effect/unstable/cli` (generated + * help, typed flag validation, `--wizard`/`--completions`, …). These tests + * exercise `src/cli.ts` itself — via `import.meta.main`, the same path + * `bin/factory.js` runs in production — so that regression can't recur + * unnoticed. + */ +describe("cli.ts entrypoint wiring", () => { + test("cli.ts contains no hand-rolled argv parser", async () => { + const source = await Bun.file(CLI).text(); + expect(source).toContain('import { factoryCommand } from "./cli-commands"'); + expect(source).toMatch(/Command\.run\(factoryCommand/); + expect(source).not.toMatch(/\bconst USAGE\b/); + expect(source).not.toMatch(/\bfunction usageError\b/); + expect(source).not.toMatch(/\bfunction parseFlags\b/); + expect(source).not.toMatch(/\bfunction parseArgs\b/); + expect(source).not.toMatch(/\bfunction parseStartArgs\b/); + expect(source).not.toMatch(/\bfunction parseServeArgs\b/); + }); + + test("--help at the real entrypoint is generated by effect/unstable/cli, not a hand-rolled banner", async () => { + const proc = Bun.spawn(["bun", CLI, "--help"], { stdout: "pipe", stderr: "pipe" }); + const [exitCode, stdout] = await Promise.all([proc.exited, new Response(proc.stdout).text()]); + expect(exitCode).toBe(0); + // effect/unstable/cli's generated help renders these section headings; + // the hand-rolled USAGE banner (a lowercase "usage:" line) never did. + expect(stdout).toContain("SUBCOMMANDS"); + expect(stdout).toContain("GLOBAL FLAGS"); + expect(stdout).not.toContain("usage:\n"); + }); + + test("factory serve --port at the real entrypoint rejects a non-numeric value before starting the daemon", async () => { + // The hand-rolled parser did `Number(portRaw)` with no validation at all — + // a bad --port silently became NaN. effect/unstable/cli's typed Int flag + // rejects it up front. + const proc = Bun.spawn(["bun", CLI, "serve", "--port", "abc"], { + stdout: "pipe", + stderr: "pipe", + }); + const [exitCode, stdout, stderr] = await Promise.all([ + proc.exited, + new Response(proc.stdout).text(), + new Response(proc.stderr).text(), + ]); + expect(exitCode).not.toBe(0); + expect(stdout + stderr).toContain("Invalid value for flag --port"); + }); +}); diff --git a/src/cli-commands.ts b/src/cli-commands.ts index 651217e..b1d2b95 100644 --- a/src/cli-commands.ts +++ b/src/cli-commands.ts @@ -2,7 +2,6 @@ import { Effect, Option } from "effect"; import { Argument, Command, Flag } from "effect/unstable/cli"; import { findFactoryConfig, loadFactoryConfig } from "./config"; import { initCli } from "./init"; -import { opencodeAdapter } from "./runtime/opencode-adapter"; import { resolve } from "node:path"; import { runCli, @@ -84,6 +83,15 @@ 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. + yield* Effect.sync(() => { + const shutdown = (signal: NodeJS.Signals) => { + void handle.stop().finally(() => process.exit(signal === "SIGINT" ? 130 : 143)); + }; + process.once("SIGINT", shutdown); + process.once("SIGTERM", shutdown); + }); }), ).pipe(Command.withDescription("Run the daemon: HTTP API, live event stream, and the web UI")); @@ -216,7 +224,6 @@ export const runCommand = Command.make("run", runConfig, (config) => clone, outPath: Option.getOrElse(config.out, () => `.factory/runs/run-${Date.now()}/events.ndjson`), dbPath: config.db, - adapter: opencodeAdapter, }; const exitCode = yield* Effect.promise(() => runCli(options)); diff --git a/src/cli.start.test.ts b/src/cli.start.test.ts index 4757dbf..9f7851c 100644 --- a/src/cli.start.test.ts +++ b/src/cli.start.test.ts @@ -25,15 +25,17 @@ async function startTestDaemon(delayMs: number, root: string): Promise { - await daemon.server.stop(true); - }, + stop: daemon.stop, }; } diff --git a/src/cli.ts b/src/cli.ts index 8203647..ed4b0c7 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -28,6 +28,7 @@ import { loadWorkflow } from "./lib/load-workflow"; import { streamSse } from "./lib/sse-client"; import { appendEvent, getRunEvents, listRuns, openStore } from "./persistence/store"; import type { AgentAdapter } from "./runtime/agent-adapter"; +import { makeAgentRuntime } from "./runtime/agent-runtime"; import { startRun } from "./runtime/run"; import { factoryCommand } from "./cli-commands"; @@ -38,7 +39,11 @@ export interface CliOptions { readonly clone?: { readonly sshUrl: string; readonly identity: GitIdentity }; readonly outPath: string; readonly dbPath: string; - readonly adapter: AgentAdapter; + /** + * Injectable agent adapter for tests. When omitted, `runCli` falls back to + * `factory.config.ts`'s `agent.adapter` (or the live opencode adapter). + */ + readonly adapter?: AgentAdapter; } function formatEvent(event: RunEvent): string { @@ -61,17 +66,21 @@ export async function runCli(options: CliOptions): Promise { const runId = `run-${Date.now()}`; let repo: RunRepo | undefined; + let adapter = options.adapter; try { const config = await loadFactoryConfig(); repo = { slug: config.repo.slug, baseBranch: config.repo.baseBranch }; + adapter ??= config.agent.adapter; } catch { repo = undefined; } - const handle = startRun(workflow, { + // The direct-run path's composition root (issue #36): one runtime for this + // one run, disposed once the run settles. + const runtime = makeAgentRuntime(adapter); + const handle = startRun(workflow, runtime, { runId, dir: options.dir, input: options.input, - adapter: options.adapter, prepareWorkspace: options.clone !== undefined, ...(repo !== undefined ? { repo } : {}), onEvent: (event) => { @@ -89,6 +98,7 @@ export async function runCli(options: CliOptions): Promise { const outcome = await handle.result; process.off("SIGINT", onSigint); + await runtime.dispose(); await sink.end(); db.close(); diff --git a/src/config.ts b/src/config.ts index adaf83d..10fdd20 100644 --- a/src/config.ts +++ b/src/config.ts @@ -10,6 +10,7 @@ import { Cron, Result, SchemaParser } from "effect"; import type { GitIdentity } from "./lib/clone"; import type { WorkflowDefinition } from "./workflow"; +import type { AgentAdapter } from "./runtime/agent-adapter"; export const DEFAULT_WORKSPACE_ROOT = ".factory/workspaces"; export const DEFAULT_MAX_CONCURRENT_RUNS = 3; @@ -69,6 +70,12 @@ export interface FactoryConfig { */ readonly maxDispatchDepth: number; readonly maxChildrenPerRun: number; + /** + * Issue #36: the agent runtime. `adapter` unset means the runtime's default + * (the live opencode adapter, `runtime/agent-runtime.ts`), so existing + * configs keep working unchanged. + */ + readonly agent: { readonly adapter?: AgentAdapter }; } /** @@ -169,6 +176,11 @@ export interface FactoryConfigInput { readonly maxDispatchDepth?: number; readonly maxChildrenPerRun?: number; readonly schedules?: ReadonlyArray>; + /** + * Issue #36: the agent runtime. Optional — when omitted the runtime uses + * the live opencode adapter. + */ + readonly agent?: { readonly adapter?: AgentAdapter }; } export function defineConfig(config: FactoryConfigInput): FactoryConfig { @@ -216,6 +228,7 @@ export function defineConfig(config: FactoryConfigInput): FactoryConfig { retainedWorkspaces, maxDispatchDepth, maxChildrenPerRun, + agent: config.agent?.adapter !== undefined ? { adapter: config.agent.adapter } : {}, }; } diff --git a/src/replay/adapter.test.ts b/src/replay/adapter.test.ts index 460a2a5..1f9295d 100644 --- a/src/replay/adapter.test.ts +++ b/src/replay/adapter.test.ts @@ -1,5 +1,6 @@ import { Effect } from "effect"; import { describe, expect, test } from "bun:test"; +import { AgentRuntimeLayer } from "../runtime/agent-runtime"; import { buildAgentStepEffect } from "../runtime/agent-step"; import type { AgentAdapterYield } from "../runtime/agent-adapter"; import { createCorpusReplayAdapter, createSlowFakeAdapter, loadCorpusBlocks } from "./adapter"; @@ -54,11 +55,12 @@ describe("createCorpusReplayAdapter", () => { dir: "/tmp", model: "opencode-go/deepseek-v4.1-flash", prompt: "irrelevant, replay ignores it", - adapter, onChunk: (chunk) => chunks.push(chunk), }); - const outcome = await Effect.runPromise(handle.effect); + const outcome = await Effect.runPromise( + Effect.provide(handle.effect, AgentRuntimeLayer(adapter)), + ); expect(outcome.chunkCount).toBe(39); expect(chunks.length).toBe(39); @@ -107,11 +109,12 @@ describe("createSlowFakeAdapter", () => { dir: "/tmp", model: "m", prompt: "p", - adapter, onChunk: () => {}, }); - const outcome = await Effect.runPromise(handle.effect); + const outcome = await Effect.runPromise( + Effect.provide(handle.effect, AgentRuntimeLayer(adapter)), + ); expect(outcome.chunkCount).toBe(3); expect(outcome.finalText).toBe("hi"); }); diff --git a/src/replay/adapter.ts b/src/replay/adapter.ts index 5130bcc..b3713ae 100644 --- a/src/replay/adapter.ts +++ b/src/replay/adapter.ts @@ -23,6 +23,7 @@ import type { AgentAdapter, AgentAdapterOptions, AgentAdapterYield, + AgentSignal, } from "../runtime/agent-adapter"; import { extractOpencodeSignal } from "../runtime/opencode-adapter"; @@ -94,17 +95,34 @@ export function createCorpusReplayAdapter(path: string): AgentAdapter { * chunk arrives and reliably interrupt mid-stream. Used by the cancellation * regression test in place of a corpus (no recorded trace can be paused on * demand; a corpus is a fixed sequence, not a controllable one). + * + * Signals can be attached declaratively by index, so a test can exercise the + * signal path without imitating any vendor chunk shape (ADR 0012 §2). */ -export function createSlowFakeAdapter(chunks: ReadonlyArray, delayMs = 20): AgentAdapter { +export interface AttachedSignal { + /** The zero-based position of the chunk this signal attaches to. */ + readonly index: number; + readonly signal: AgentSignal; +} + +export function createSlowFakeAdapter( + chunks: ReadonlyArray, + delayMs = 20, + signals: ReadonlyArray = [], +): AgentAdapter { + const byIndex = new Map(signals.map((s) => [s.index, s.signal])); return { async prepareWorkspace(_dir: string): Promise {}, stream(_options: AgentAdapterOptions): AsyncIterable { return { async *[Symbol.asyncIterator]() { - for (const chunk of chunks) { + for (let index = 0; index < chunks.length; index++) { await new Promise((resolve) => setTimeout(resolve, delayMs)); - yield { chunk }; + const signal = byIndex.get(index); + yield signal === undefined + ? { chunk: chunks[index] } + : { chunk: chunks[index], signal }; } }, }; diff --git a/src/runtime/agent-runtime.test.ts b/src/runtime/agent-runtime.test.ts new file mode 100644 index 0000000..1d968f2 --- /dev/null +++ b/src/runtime/agent-runtime.test.ts @@ -0,0 +1,38 @@ +/** + * The agent runtime as a context service (ADR 0012 §4). + */ + +import { describe, expect, test } from "bun:test"; +import { Effect, ManagedRuntime } from "effect"; +import { createSlowFakeAdapter } from "../replay/adapter"; +import { opencodeAdapter } from "./opencode-adapter"; +import { AgentRuntime, AgentRuntimeLayer } from "./agent-runtime"; + +describe("AgentRuntime service", () => { + test("without an adapter, the layer provides the live opencode adapter", async () => { + const runtime = ManagedRuntime.make(AgentRuntimeLayer()); + const { adapter } = await runtime.runPromise(AgentRuntime); + expect(adapter).toBe(opencodeAdapter); + await runtime.dispose(); + }); + + test("AgentRuntimeLayer swaps the adapter via context", async () => { + const fake = createSlowFakeAdapter([], 1); + const runtime = ManagedRuntime.make(AgentRuntimeLayer(fake)); + const { adapter } = await runtime.runPromise(AgentRuntime); + expect(adapter).toBe(fake); + await runtime.dispose(); + }); + + test("an effect that requires the service resolves it from context", async () => { + const fake = createSlowFakeAdapter([], 1); + const program = Effect.gen(function* () { + const runtime = yield* AgentRuntime; + return runtime.adapter; + }); + const runtime = ManagedRuntime.make(AgentRuntimeLayer(fake)); + const adapter = await runtime.runPromise(program); + expect(adapter).toBe(fake); + await runtime.dispose(); + }); +}); diff --git a/src/runtime/agent-runtime.ts b/src/runtime/agent-runtime.ts new file mode 100644 index 0000000..09cfaee --- /dev/null +++ b/src/runtime/agent-runtime.ts @@ -0,0 +1,47 @@ +/** + * The agent runtime as an Effect context service (ADR 0012 §4). + * + * The runtime is the daemon's first service to move into Effect's dependency + * injection: it stops being a value threaded through every interface between + * the CLI and the step runner, and becomes a service resolved from context at + * the point of use (`runtime/agent-step.ts`). + * + * The service shape is deliberately small: it currently exposes only the + * adapter, which is the part issue #36 needs to make selectable from config. + */ + +import { Context, Layer, ManagedRuntime } from "effect"; +import type { AgentAdapter } from "./agent-adapter"; +import { opencodeAdapter } from "./opencode-adapter"; + +export interface AgentRuntimeShape { + /** The adapter that produces agent chunk streams (ADR 0012 §2). */ + readonly adapter: AgentAdapter; +} + +/** + * The agent-runtime service. Yield it inside an Effect to read the adapter; + * provide it with `AgentRuntimeLayer`. + */ +export class AgentRuntime extends Context.Service()( + "AgentRuntime", +) {} + +/** + * A layer that provides `adapter`, or the live opencode adapter when none is + * given. This is the single place the opencode default lives: config leaves + * `agent.adapter` unset unless a project chooses one. + */ +export const AgentRuntimeLayer = ( + adapter: AgentAdapter = opencodeAdapter, +): Layer.Layer => Layer.succeed(AgentRuntime, { adapter }); + +/** + * Build a `ManagedRuntime` providing the agent runtime — the composition root + * of the daemon and the CLI's direct-run path, and a convenience for tests. + * The caller owns it and must `dispose()` it once its runs are done. + */ +export const makeAgentRuntime = ( + adapter?: AgentAdapter, +): ManagedRuntime.ManagedRuntime => + ManagedRuntime.make(AgentRuntimeLayer(adapter)); diff --git a/src/runtime/agent-step.test.ts b/src/runtime/agent-step.test.ts index c03c34a..9afe22e 100644 --- a/src/runtime/agent-step.test.ts +++ b/src/runtime/agent-step.test.ts @@ -1,6 +1,7 @@ -import { Effect } from "effect"; +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"; function makeYield(chunk: unknown, signal?: AgentSignal): AgentAdapterYield { @@ -38,11 +39,12 @@ describe("buildAgentStepEffect signal extraction", () => { dir: "/tmp", model: "m", prompt: "p", - adapter, onChunk: () => {}, }); - const outcome = await Effect.runPromise(handle.effect); + const outcome = await Effect.runPromise( + Effect.provide(handle.effect, AgentRuntimeLayer(adapter)), + ); expect(outcome.sessionId).toBe("ses_abc"); expect(outcome.finalText).toBe("hello"); expect(outcome.chunkCount).toBe(4); @@ -65,11 +67,12 @@ describe("buildAgentStepEffect signal extraction", () => { dir: "/tmp", model: "m", prompt: "p", - adapter, onChunk: () => {}, }); - const outcome = await Effect.runPromise(handle.effect); + const outcome = await Effect.runPromise( + Effect.provide(handle.effect, AgentRuntimeLayer(adapter)), + ); expect(outcome.structuredOutput).toEqual(outputObject); }); @@ -87,11 +90,12 @@ describe("buildAgentStepEffect signal extraction", () => { dir: "/tmp", model: "m", prompt: "p", - adapter, onChunk: () => {}, }); - const outcome = await Effect.runPromise(handle.effect); + const outcome = await Effect.runPromise( + Effect.provide(handle.effect, AgentRuntimeLayer(adapter)), + ); expect(outcome.runError).toBe("something broke"); }); @@ -105,11 +109,10 @@ describe("buildAgentStepEffect signal extraction", () => { dir: "/tmp", model: "m", prompt: "p", - adapter, onChunk: (chunk) => chunks.push(chunk), }); - await Effect.runPromise(handle.effect); + await Effect.runPromise(Effect.provide(handle.effect, AgentRuntimeLayer(adapter))); expect(chunks).toHaveLength(1); expect(chunks[0]).toBe(rawChunk); }); @@ -126,15 +129,43 @@ describe("buildAgentStepEffect signal extraction", () => { dir: "/tmp", model: "m", prompt: "p", - adapter, onChunk: () => {}, }); - const outcome = await Effect.runPromise(handle.effect); + const outcome = await Effect.runPromise( + Effect.provide(handle.effect, AgentRuntimeLayer(adapter)), + ); expect(outcome.chunkCount).toBe(3); expect(outcome.finalText).toBe("no signals here"); expect(outcome.sessionId).toBeUndefined(); expect(outcome.structuredOutput).toBeUndefined(); expect(outcome.runError).toBeUndefined(); }); + + test("the adapter is resolved from the AgentRuntime service when the step runs (#36)", async () => { + let streamed = 0; + const base = signalAdapter([makeYield({ type: "TEXT_MESSAGE_START" })]); + const adapter = { + ...base, + stream: () => { + streamed += 1; + return base.stream(); + }, + }; + + const handle = buildAgentStepEffect({ + threadId: "t", + dir: "/tmp", + model: "m", + prompt: "p", + onChunk: () => {}, + }); + expect(streamed).toBe(0); + + const runtime = ManagedRuntime.make(AgentRuntimeLayer(adapter)); + const outcome = await runtime.runPromise(handle.effect); + await runtime.dispose(); + expect(streamed).toBe(1); + expect(outcome.chunkCount).toBe(1); + }); }); diff --git a/src/runtime/agent-step.ts b/src/runtime/agent-step.ts index e597aba..5c9dfb1 100644 --- a/src/runtime/agent-step.ts +++ b/src/runtime/agent-step.ts @@ -23,7 +23,8 @@ import { Effect, Schema, Stream } from "effect"; import type { AgentStepUsage } from "../events"; -import type { AgentAdapter, AgentAdapterYield } from "./agent-adapter"; +import type { AgentAdapterYield } from "./agent-adapter"; +import { AgentRuntime } from "./agent-runtime"; /** * Pull the four token counts out of a `RUN_FINISHED.usage` object. @@ -60,7 +61,6 @@ export interface AgentStepEffectOptions { readonly model: string; readonly prompt: string; readonly outputSchema?: unknown; - readonly adapter: AgentAdapter; /** Fired synchronously per chunk, before any bookkeeping — the runtime's `AgentChunk` emission point. */ readonly onChunk: (chunk: unknown) => void; } @@ -88,8 +88,12 @@ export interface AgentStepPartial { } export interface AgentStepHandle { - /** Run this with `Effect.runFork` to get an interruptible `Fiber`. */ - readonly effect: Effect.Effect; + /** + * Run this with `runtime.runFork` to get an interruptible `Fiber`. The + * adapter is resolved from the `AgentRuntime` service when the effect runs + * (issue #36), not threaded through the options. + */ + readonly effect: Effect.Effect; readonly abortController: AbortController; readonly partial: AgentStepPartial; } @@ -97,18 +101,20 @@ export interface AgentStepHandle { export function buildAgentStepEffect(options: AgentStepEffectOptions): AgentStepHandle { const abortController = new AbortController(); - const iterable = options.adapter.stream({ - threadId: options.threadId, - dir: options.dir, - model: options.model, - prompt: options.prompt, - outputSchema: options.outputSchema, - abortController, - }); - - const rawStream = Stream.fromAsyncIterable( - iterable, - (cause) => new AgentStepChunkError({ cause }), + const rawStream = Stream.unwrap( + Effect.map(Effect.service(AgentRuntime), ({ adapter }) => + Stream.fromAsyncIterable( + adapter.stream({ + threadId: options.threadId, + dir: options.dir, + model: options.model, + prompt: options.prompt, + outputSchema: options.outputSchema, + abortController, + }), + (cause) => new AgentStepChunkError({ cause }), + ), + ), ); const partial: AgentStepPartial = { diff --git a/src/runtime/agent-step.usage.test.ts b/src/runtime/agent-step.usage.test.ts index 0031220..67dffe8 100644 --- a/src/runtime/agent-step.usage.test.ts +++ b/src/runtime/agent-step.usage.test.ts @@ -15,29 +15,30 @@ import { describe, expect, test } from "bun:test"; import { Effect } from "effect"; import { agentStepContextTokens } from "../events"; -import type { AgentAdapterYield } from "./agent-adapter"; +import type { AgentAdapter, AgentAdapterYield } from "./agent-adapter"; +import { AgentRuntimeLayer } from "./agent-runtime"; import { loadCorpusBlocks } from "../replay/adapter"; import { buildAgentStepEffect } from "./agent-step"; const CORPUS = "test/corpus/run-1789308170212.ndjson"; async function runBlock(chunks: ReadonlyArray) { + const adapter: AgentAdapter = { + async prepareWorkspace(_dir: string): Promise {}, + async *stream(): AsyncGenerator { + for (const chunk of chunks) { + yield { chunk }; + } + }, + }; const handle = buildAgentStepEffect({ threadId: "thread", dir: ".", model: "model", prompt: "prompt", - adapter: { - async prepareWorkspace(_dir: string): Promise {}, - async *stream(): AsyncGenerator { - for (const chunk of chunks) { - yield { chunk }; - } - }, - }, onChunk: () => {}, }); - return await Effect.runPromise(handle.effect); + return await Effect.runPromise(Effect.provide(handle.effect, AgentRuntimeLayer(adapter))); } describe("agent step usage", () => { diff --git a/src/runtime/run-signals.test.ts b/src/runtime/run-signals.test.ts new file mode 100644 index 0000000..88f5034 --- /dev/null +++ b/src/runtime/run-signals.test.ts @@ -0,0 +1,164 @@ +/** + * The run-level contract behind issue #35's acceptance criteria: the adapter + * supplies normalized signals (ADR 0012 §2), and `AgentStepFinished`'s + * `sessionId`, `output` and `error` fields are populated from them exactly as + * before — including the two-tier structured-output resolution (signal value + * first, `finalText` re-parse as tier-2 fallback) and the RUN_ERROR-shaped + * failure outcome. + */ + +import { describe, expect, test } from "bun:test"; +import type { RunEvent } from "../events"; +import { defineWorkflow, Schema } from "../workflow"; +import { createSlowFakeAdapter } from "../replay/adapter"; +import { makeAgentRuntime } from "./agent-runtime"; +import { startRun } from "./run"; + +const TEXT_CHUNKS = [ + { type: "TEXT_MESSAGE_START", messageId: "m1" }, + { type: "TEXT_MESSAGE_CONTENT", delta: '{"where":"from-final-text"}' }, + { type: "TEXT_MESSAGE_END", messageId: "m1" }, +]; + +const SIGNAL_SCHEMA = Schema.Struct({ where: Schema.String }); + +function agentWorkflow(outputSchema?: Schema.Codec) { + return defineWorkflow("signal-e2e", { + input: Schema.Struct({}), + run: async (ctx) => { + const result = await ctx.agent("step", "irrelevant, replay ignores it", { + ...(outputSchema !== undefined ? { output: outputSchema } : {}), + }); + return { sessionId: result.sessionId, error: result.error, output: result.output }; + }, + }); +} + +async function runWith( + chunks: ReadonlyArray, + signals: Parameters[2], +) { + const events: Array = []; + const runtime = makeAgentRuntime(createSlowFakeAdapter(chunks, 1, signals)); + const handle = startRun(agentWorkflow(), runtime, { + runId: "run-signals", + dir: "/tmp", + input: {}, + onEvent: (event) => events.push(event), + }); + const outcome = await handle.result; + return { outcome, events }; +} + +function stepFinished(events: ReadonlyArray) { + const event = events.find((e) => e.payload._tag === "AgentStepFinished"); + if (event === undefined || event.payload._tag !== "AgentStepFinished") { + throw new Error("no AgentStepFinished event"); + } + return event.payload; +} + +describe("signals populate AgentStepFinished as before (issue #35)", () => { + test("a session signal lands on AgentStepFinished.sessionId", async () => { + const { events } = await runWith(TEXT_CHUNKS, [ + { index: 0, signal: { _tag: "sessionId", value: "ses_e2e" } }, + ]); + expect(stepFinished(events).sessionId).toBe("ses_e2e"); + }); + + test("a structured-output signal is decoded into AgentStepFinished.output (tier 1)", async () => { + const events: Array = []; + const runtime = makeAgentRuntime( + createSlowFakeAdapter([...TEXT_CHUNKS, { type: "CUSTOM", name: "anything-at-all" }], 1, [ + { index: 3, signal: { _tag: "structuredOutput", value: { where: "from-signal" } } }, + ]), + ); + const handle = startRun(agentWorkflow(SIGNAL_SCHEMA), runtime, { + runId: "run-signals-output", + dir: "/tmp", + input: {}, + onEvent: (event) => events.push(event), + }); + const outcome = await handle.result; + expect(outcome.outcome).toBe("completed"); + + expect(stepFinished(events).output).toEqual({ where: "from-signal" }); + }); + + test("without a signal, tier 2 re-parses the final text — the fallback is unchanged", async () => { + const events: Array = []; + const runtime = makeAgentRuntime(createSlowFakeAdapter(TEXT_CHUNKS, 1)); + const handle = startRun(agentWorkflow(SIGNAL_SCHEMA), runtime, { + runId: "run-signals-tier2", + dir: "/tmp", + input: {}, + onEvent: (event) => events.push(event), + }); + const outcome = await handle.result; + expect(outcome.outcome).toBe("completed"); + + expect(stepFinished(events).output).toEqual({ where: "from-final-text" }); + }); + + test("an error signal fails the step and lands on AgentStepFinished.error", async () => { + const { outcome, events } = await runWith( + [...TEXT_CHUNKS, { type: "RUN_FINISHED" }], + [{ index: 3, signal: { _tag: "runError", value: "sandbox vanished" } }], + ); + expect(outcome.outcome).toBe("failed"); + + const finished = stepFinished(events); + expect(finished.outcome).toBe("failed"); + expect(finished.error).toBe("sandbox vanished"); + }); +}); + +describe("token usage on AgentStepFinished", () => { + const USAGE_CHUNK = { + type: "RUN_FINISHED", + usage: { + promptTokens: 120, + completionTokens: 30, + promptTokensDetails: { cachedTokens: 900 }, + completionTokensDetails: { reasoningTokens: 7 }, + }, + }; + const EXPECTED_USAGE = { + inputTokens: 120, + outputTokens: 30, + cachedInputTokens: 900, + reasoningTokens: 7, + }; + + test("a completed step carries the usage its RUN_FINISHED chunk reported", async () => { + const { outcome, events } = await runWith([...TEXT_CHUNKS, USAGE_CHUNK], []); + expect(outcome.outcome).toBe("completed"); + + const finished = stepFinished(events); + expect(finished.outcome).toBe("completed"); + expect(finished.usage).toEqual(EXPECTED_USAGE); + }); + + test("a cancelled step still carries the usage seen before cancellation", async () => { + const events: Array = []; + let handle: ReturnType | undefined; + // The step is cancelled right after its usage chunk lands, while the next + // (slow) chunk is still pending. + const runtime = makeAgentRuntime(createSlowFakeAdapter([USAGE_CHUNK, ...TEXT_CHUNKS], 50)); + handle = startRun(agentWorkflow(), runtime, { + runId: "run-usage-cancelled", + dir: "/tmp", + input: {}, + onEvent: (event) => { + events.push(event); + if (event.payload._tag === "AgentChunk") void handle?.cancel(); + }, + }); + const outcome = await handle.result; + expect(outcome.outcome).toBe("cancelled"); + + const finished = stepFinished(events); + expect(finished.outcome).toBe("cancelled"); + expect(finished.usage).toEqual(EXPECTED_USAGE); + }); +}); diff --git a/src/runtime/run.test.ts b/src/runtime/run.test.ts index 1f3c4fd..01c8f7d 100644 --- a/src/runtime/run.test.ts +++ b/src/runtime/run.test.ts @@ -10,6 +10,7 @@ import { describe, expect, test } from "bun:test"; import type { RunEvent } from "../events"; import { defineWorkflow, Schema } from "../workflow"; import { createSlowFakeAdapter } from "../replay/adapter"; +import { makeAgentRuntime } from "./agent-runtime"; import { startRun, RunCancelledSignal } from "./run"; import type { AgentAdapter } from "./agent-adapter"; @@ -40,11 +41,10 @@ describe("startRun cancellation", () => { }, }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter(SLOW_CHUNKS, 20)), { runId: "run-cancel-test", dir: "/tmp", input: {}, - adapter: createSlowFakeAdapter(SLOW_CHUNKS, 20), onEvent: (event) => events.push(event), }); @@ -75,11 +75,10 @@ describe("startRun cancellation", () => { }, }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter(SLOW_CHUNKS, 1)), { runId: "run-no-cancel-test", dir: "/tmp", input: {}, - adapter: createSlowFakeAdapter(SLOW_CHUNKS, 1), onEvent: () => {}, }); @@ -106,11 +105,10 @@ describe("startRun workspace kind (issue #13)", () => { }, }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter([])), { runId: "run-scratch-wb", dir: "/tmp/nothing", input: {}, - adapter: createSlowFakeAdapter([]), workspaceKind: "scratch", repo: { slug: "owner/repo", baseBranch: "main" }, onEvent: (event) => events.push(event), @@ -141,11 +139,10 @@ describe("startRun workspace kind (issue #13)", () => { }), }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter([])), { runId: "run-clone-wb", dir: "/tmp/nothing", input: {}, - adapter: createSlowFakeAdapter([]), repo: { slug: "owner/repo", baseBranch: "main" }, onEvent: (event) => events.push(event), }); @@ -167,20 +164,18 @@ describe("startRun workspace kind (issue #13)", () => { }); const scratchEvents: Array = []; - await startRun(workflow, { + await startRun(workflow, makeAgentRuntime(createSlowFakeAdapter([])), { runId: "run-kind-scratch", dir: "/tmp/s", input: {}, - adapter: createSlowFakeAdapter([]), workspaceKind: "scratch", onEvent: (event) => scratchEvents.push(event), }).result; const cloneEvents: Array = []; - await startRun(workflow, { + await startRun(workflow, makeAgentRuntime(createSlowFakeAdapter([])), { runId: "run-kind-clone", dir: "/tmp/c", input: {}, - adapter: createSlowFakeAdapter([]), onEvent: (event) => cloneEvents.push(event), }).result; @@ -203,11 +198,10 @@ describe("startRun schedule trigger (issue #16)", () => { return { finalText: result.finalText }; }, }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter(SLOW_CHUNKS, 5)), { runId: "run-by-schedule", dir: "/tmp", input: { issueNumber: 7 }, - adapter: createSlowFakeAdapter(SLOW_CHUNKS, 5), scheduleId: "nightly", onEvent: (event) => events.push(event), }); @@ -222,11 +216,10 @@ describe("startRun schedule trigger (issue #16)", () => { input: Schema.Struct({}), run: async () => ({}), }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter([])), { runId: "run-manual", dir: "/tmp", input: {}, - adapter: createSlowFakeAdapter([]), onEvent: (event) => events.push(event), }); await handle.result; @@ -249,11 +242,10 @@ describe("startRun model precedence (issue #16)", () => { return {}; }, }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter(SLOW_CHUNKS, 5)), { runId: "run-models", dir: "/tmp", input: {}, - adapter: createSlowFakeAdapter(SLOW_CHUNKS, 5), agentOverrides: { model: "schedule-override" }, onEvent: (event) => events.push(event), }); @@ -277,11 +269,10 @@ describe("startRun model precedence (issue #16)", () => { return {}; }, }); - const handle = startRun(workflow, { + const handle = startRun(workflow, makeAgentRuntime(createSlowFakeAdapter(SLOW_CHUNKS, 5)), { runId: "run-models-fallback", dir: "/tmp", input: {}, - adapter: createSlowFakeAdapter(SLOW_CHUNKS, 5), onEvent: (event) => events.push(event), }); const outcome = await handle.result; @@ -314,11 +305,10 @@ describe("startRun prepareWorkspace (ADR 0012 §3, #37)", () => { run: async () => ({}), }); - await startRun(workflow, { + await startRun(workflow, makeAgentRuntime(adapter), { runId: "run-prep-clone", dir: "/tmp/clone-dir", input: {}, - adapter, prepareWorkspace: true, onEvent: (event) => events.push(event), }).result; @@ -333,11 +323,10 @@ describe("startRun prepareWorkspace (ADR 0012 §3, #37)", () => { run: async () => ({}), }); - await startRun(workflow, { + await startRun(workflow, makeAgentRuntime(adapter), { runId: "run-prep-scratch", dir: "/tmp/scratch-dir", input: {}, - adapter, onEvent: () => {}, }).result; diff --git a/src/runtime/run.ts b/src/runtime/run.ts index d211425..8387371 100644 --- a/src/runtime/run.ts +++ b/src/runtime/run.ts @@ -13,7 +13,7 @@ * else, and turns into `RunCancelled` rather than `RunFailed`. */ -import { Cause, Effect, Exit, Fiber, Schema, SchemaParser } from "effect"; +import { Cause, Effect, Exit, Fiber, Schema, SchemaParser, type ManagedRuntime } from "effect"; import type { RunEvent, RunEventPayload } from "../events"; import { hostExec, type ExecResult } from "../lib/exec"; import { writeBack as writeBackLib, type WriteBackResult } from "../lib/writeback"; @@ -29,7 +29,7 @@ import type { WriteBackCallOptions, } from "../workflow"; import { DedupeKeyError } from "../lib/dedupe"; -import type { AgentAdapter } from "./agent-adapter"; +import { AgentRuntime } from "./agent-runtime"; import { buildAgentStepEffect } from "./agent-step"; export class RunCancelledSignal extends Schema.TaggedError()( @@ -58,7 +58,6 @@ export interface StartRunOptions { readonly runId: string; readonly dir: string; readonly input: unknown; - readonly adapter: AgentAdapter; /** Write-back environment (D32). Absent, a workflow's `ctx.writeBack` fails. */ readonly repo?: RunRepo; /** @@ -143,8 +142,15 @@ function resolveOutput( } } +/** + * `runtime` is the composition root's `ManagedRuntime` (issue #36): agent + * steps resolve their adapter from its `AgentRuntime` service and their + * fibers are forked on it. `startRun` stays synchronous: nothing is resolved + * from the runtime until a step (or workspace preparation) actually runs. + */ export function startRun( workflow: WorkflowDefinition, + runtime: ManagedRuntime.ManagedRuntime, options: StartRunOptions, ): RunHandle { let seq = 0; @@ -224,7 +230,6 @@ export function startRun( model, prompt, outputSchema, - adapter: options.adapter, onChunk: (chunk) => { const record = chunk as { type?: unknown }; emit({ @@ -236,7 +241,7 @@ export function startRun( }, }); - const fiber = Effect.runFork(handle.effect); + const fiber = runtime.runFork(handle.effect); activeAgentFiber = fiber; const exit = await Effect.runPromise(Fiber.await(fiber)); activeAgentFiber = null; @@ -482,7 +487,8 @@ export function startRun( try { if (options.prepareWorkspace === true) { - await options.adapter.prepareWorkspace(options.dir); + const { adapter } = await runtime.runPromise(Effect.service(AgentRuntime)); + await adapter.prepareWorkspace(options.dir); } const output = await workflow.run(ctx, decodedInput); diff --git a/src/server/concurrency.test.ts b/src/server/concurrency.test.ts index b0a3e18..051774e 100644 --- a/src/server/concurrency.test.ts +++ b/src/server/concurrency.test.ts @@ -19,6 +19,7 @@ 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"; @@ -72,13 +73,15 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => { const db = openStore(join(root, "factory.db")); const server = serve({ db, - adapter: createSlowFakeAdapter( - [ - { type: "TEXT_MESSAGE_START" }, - { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, - { type: "TEXT_MESSAGE_END" }, - ], - 30, + runtime: makeAgentRuntime( + createSlowFakeAdapter( + [ + { type: "TEXT_MESSAGE_START" }, + { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, + { type: "TEXT_MESSAGE_END" }, + ], + 30, + ), ), port: 0, config: defineConfig({ @@ -159,13 +162,15 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => { const db = openStore(join(root, "factory.db")); const server = serve({ db, - adapter: createSlowFakeAdapter( - [ - { type: "TEXT_MESSAGE_START" }, - { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, - { type: "TEXT_MESSAGE_END" }, - ], - 300, + runtime: makeAgentRuntime( + createSlowFakeAdapter( + [ + { type: "TEXT_MESSAGE_START" }, + { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, + { type: "TEXT_MESSAGE_END" }, + ], + 300, + ), ), port: 0, config: defineConfig({ @@ -219,13 +224,15 @@ describe("phase 5 P1: per-run working trees (D28) over the real server", () => { const db = openStore(join(root, "factory.db")); const server = serve({ db, - adapter: createSlowFakeAdapter( - [ - { type: "TEXT_MESSAGE_START" }, - { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, - { type: "TEXT_MESSAGE_END" }, - ], - 1_000, + runtime: makeAgentRuntime( + createSlowFakeAdapter( + [ + { type: "TEXT_MESSAGE_START" }, + { type: "TEXT_MESSAGE_CONTENT", delta: "hi" }, + { type: "TEXT_MESSAGE_END" }, + ], + 1_000, + ), ), port: 0, config: defineConfig({ diff --git a/src/server/daemon.schedules.test.ts b/src/server/daemon.schedules.test.ts index 61fd1d5..a5e78a5 100644 --- a/src/server/daemon.schedules.test.ts +++ b/src/server/daemon.schedules.test.ts @@ -5,7 +5,6 @@ */ import { afterAll, describe, expect, test } from "bun:test"; -import { Effect, Fiber } from "effect"; import { rmSync } from "node:fs"; import { mkdtempSync } from "node:fs"; import { tmpdir } from "node:os"; @@ -77,10 +76,7 @@ describe("daemon schedules (issue #16)", () => { expect(scheduled[0]?.scheduleId).toBe("hourly-echo"); expect(scheduled[0]?.status).toBe("RunFinished"); } finally { - if (handle.schedulerFiber !== undefined) { - Effect.runFork(Fiber.interrupt(handle.schedulerFiber)); - } - handle.server.stop(true); + await handle.stop(); } }); }); diff --git a/src/server/daemon.ts b/src/server/daemon.ts index 01ef0fd..8c56f06 100644 --- a/src/server/daemon.ts +++ b/src/server/daemon.ts @@ -4,16 +4,20 @@ * third column). Automatic dispatch is not daemon logic since epic #19: it * is project policy living in scheduled wrapper workflows — e.g. the sample * project's Ready sweep — fired by the scheduler loop on their cron. + * + * Issue #36: this is the daemon's Effect composition root. It builds the + * agent runtime's `ManagedRuntime` once and hands it to `serve()` and the + * 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. */ import { mkdir } from "node:fs/promises"; import { dirname } from "node:path"; -import { Effect, type Fiber } from "effect"; +import { Effect, Fiber } from "effect"; type AnyFiber = Fiber.Fiber; import type { FactoryConfig } from "../config"; import { openStore } from "../persistence/store"; -import type { AgentAdapter } from "../runtime/agent-adapter"; -import { opencodeAdapter } from "../runtime/opencode-adapter"; import { serve } from "./http"; import { type DispatchEnv, type WorkspaceSpec } from "./runs"; import { @@ -23,11 +27,11 @@ import { toRuntimeSchedules, type SchedulerDeps, } from "./scheduler"; +import { makeAgentRuntime } from "../runtime/agent-runtime"; export interface DaemonOptions { readonly dbPath: string; readonly port?: number; - readonly adapter?: AgentAdapter; /** Issue #16: the tick cadence of the config schedules, over the default. */ readonly schedulerIntervalMs?: number; /** @@ -43,6 +47,11 @@ export interface DaemonHandle { readonly server: ReturnType; /** 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). + */ + readonly stop: () => Promise; } /** Issue #16: how often the scheduler's due window check runs. */ @@ -51,11 +60,12 @@ export const DEFAULT_SCHEDULER_INTERVAL_MS = 30_000; export async function startDaemon(options: DaemonOptions): Promise { await mkdir(dirname(options.dbPath), { recursive: true }); const db = openStore(options.dbPath); - const adapter = options.adapter ?? opencodeAdapter; + + const runtime = makeAgentRuntime(options.config?.agent.adapter); const server = serve({ db, - adapter, + runtime, port: options.port, ...(options.config !== undefined ? { config: options.config } : {}), }); @@ -76,7 +86,6 @@ export async function startDaemon(options: DaemonOptions): Promise workspace, repo, maxConcurrentRuns, - adapter, maxDispatchDepth: config.maxDispatchDepth, maxChildrenPerRun: config.maxChildrenPerRun, }; @@ -84,7 +93,7 @@ export async function startDaemon(options: DaemonOptions): Promise schedules: toRuntimeSchedules(config), fire: makeScheduleFire({ db, - adapter, + runtime, maxConcurrentRuns, workspace, repo, @@ -102,5 +111,11 @@ export async function startDaemon(options: DaemonOptions): Promise })() : undefined; - return { server, schedulerFiber }; + const stop = async (): Promise => { + if (schedulerFiber !== undefined) await Effect.runPromise(Fiber.interrupt(schedulerFiber)); + await server.stop(true); + await runtime.dispose(); + }; + + return { server, schedulerFiber, stop }; } diff --git a/src/server/dedupe.test.ts b/src/server/dedupe.test.ts index 85059bd..845d8ac 100644 --- a/src/server/dedupe.test.ts +++ b/src/server/dedupe.test.ts @@ -12,6 +12,7 @@ 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"; @@ -32,6 +33,8 @@ const SLOW_ADAPTER = createSlowFakeAdapter( 25, ); +const runtime = makeAgentRuntime(SLOW_ADAPTER); + const echoWorkflow = defineWorkflow("echo-wf", { input: Schema.Struct({}), run: async (ctx) => { @@ -72,10 +75,9 @@ interface StartOpts { } function start(db: ReturnType, opts: StartOpts): Promise { - return startTrackedRun(db, echoWorkflow, { + return startTrackedRun(runtime, db, echoWorkflow, { dir: opts.dir, input: {}, - adapter: SLOW_ADAPTER, dedupeRegistry: opts.registry ?? createDedupeRegistry(), maxConcurrentRuns: 4, ...(opts.runId !== undefined ? { runId: opts.runId } : {}), @@ -111,13 +113,11 @@ function workspaces(root: string): WorkspaceSpec { function withWorkspaces(root: string): Record { return { workspace: workspaces(root), - adapter: SLOW_ADAPTER, maxConcurrentRuns: 10, }; } describe("dedupe keys through ctx.dispatch (issue #15)", () => { - const driftAdapter = SLOW_ADAPTER; const busyWorkflow = defineWorkflow("busy-wf", { input: Schema.Struct({ n: Schema.Number }), workspace: { kind: "scratch" }, @@ -147,12 +147,11 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { }, }); - const parentRunId = await startTrackedRun(db, parent, { + const parentRunId = await startTrackedRun(runtime, db, parent, { input: { wait: 3 }, - adapter: driftAdapter, workspace: workspaces(join(root)), maxConcurrentRuns: 10, - dispatchEnv: { ...withWorkspaces(join(root)), adapter: driftAdapter } as never, + dispatchEnv: withWorkspaces(join(root)), }); await waitFor(() => !isActive(parentRunId)); @@ -207,13 +206,12 @@ describe("dedupe keys through ctx.dispatch (issue #15)", () => { }, }); - const parentRunId = (await startTrackedRun(db, parent, { + const parentRunId = await startTrackedRun(runtime, db, parent, { input: {}, - adapter: driftAdapter, workspace: workspaces(join(root)), maxConcurrentRuns: 10, - dispatchEnv: { ...withWorkspaces(join(root)), adapter: driftAdapter } as never, - })) as string; + dispatchEnv: withWorkspaces(join(root)), + }); await waitFor(() => getRunEvents(db, parentRunId).some((event) => event.payload._tag === "RunDispatched"), @@ -344,9 +342,8 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { // allocation of an explicitly undefined dir fails after the claim await expect( - startTrackedRun(db, echoWorkflow, { + startTrackedRun(runtime, db, echoWorkflow, { input: {}, - adapter: SLOW_ADAPTER, dedupeKey: "item:44", dedupeRegistry: createDedupeRegistry(), }), @@ -366,11 +363,10 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { const db = openStore(join(root, "factory.db")); const registry = createDedupeRegistry(); - const holderRunId = await startTrackedRun(db, echoWorkflow, { + const holderRunId = await startTrackedRun(runtime, db, echoWorkflow, { runId: "run-cancelled", dir: join(root, "dir-1"), input: {}, - adapter: SLOW_ADAPTER, dedupeKey: "item:45", dedupeRegistry: registry, }); @@ -381,10 +377,9 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => { await waitFor(() => !isActive(holderRunId)); const started = getRunEvents(db, holderRunId).some((e) => e.payload._tag === "RunStarted"); - const retryRunId = await startTrackedRun(db, echoWorkflow, { + const retryRunId = await startTrackedRun(runtime, db, echoWorkflow, { dir: join(root, "dir-2"), input: {}, - adapter: SLOW_ADAPTER, dedupeKey: "item:45", dedupeRegistry: registry, }); @@ -443,7 +438,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { const server = serve({ db, - adapter: SLOW_ADAPTER, + runtime: makeAgentRuntime(SLOW_ADAPTER), port: 0, config: defineConfig({ repo: { @@ -517,7 +512,7 @@ describe("dedupe keys over POST /api/runs (issue #15)", () => { const server = serve({ db, - adapter: SLOW_ADAPTER, + runtime: makeAgentRuntime(SLOW_ADAPTER), port: 0, config: defineConfig({ repo: { diff --git a/src/server/http.test.ts b/src/server/http.test.ts index 1ca9983..30e7676 100644 --- a/src/server/http.test.ts +++ b/src/server/http.test.ts @@ -6,6 +6,7 @@ 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"; @@ -90,7 +91,7 @@ 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, adapter, port: 0, config }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); const base = `http://localhost:${server.port}`; try { @@ -115,7 +116,7 @@ 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, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -135,7 +136,7 @@ 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, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -155,7 +156,7 @@ 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, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -238,7 +239,7 @@ 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, adapter, port: 0, sseKeepaliveMs: 25 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, sseKeepaliveMs: 25 }); const base = `http://localhost:${server.port}`; try { @@ -274,7 +275,7 @@ 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, adapter, port: 0, sseKeepaliveMs: 10 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, sseKeepaliveMs: 10 }); const base = `http://localhost:${server.port}`; try { @@ -314,7 +315,7 @@ describe("phase 3 HTTP API + SSE", () => { ], 1, ); - const server = serve({ db, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -363,7 +364,7 @@ describe("phase 3 HTTP API + SSE", () => { ], 500, ); - const server = serve({ db, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -399,7 +400,7 @@ describe("phase 3 HTTP API + SSE", () => { ], 60, ); - const server = serve({ db, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -438,7 +439,7 @@ describe("phase 3 HTTP API + SSE", () => { ], 400, ); - const server = serve({ db, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -476,7 +477,7 @@ describe("phase 3 HTTP API + SSE", () => { ], 1, ); - const server = serve({ db, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -522,7 +523,7 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { 1, ); const config = await loadFactoryConfig(FIXTURE_CONFIG); - const server = serve({ db, adapter, port: 0, config }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); const base = `http://localhost:${server.port}`; try { @@ -548,7 +549,7 @@ 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, adapter, port: 0, config }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); const base = `http://localhost:${server.port}`; try { @@ -594,7 +595,7 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { maxConcurrentRuns: 1, retainedWorkspaces: 10, }); - const server = serve({ db, adapter, port: 0, config }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); const base = `http://localhost:${server.port}`; try { @@ -657,7 +658,7 @@ describe("POST /api/runs {workflowId, input} (D31)", () => { maxConcurrentRuns: 3, retainedWorkspaces: 10, }); - const server = serve({ db, adapter, port: 0, config }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); const base = `http://localhost:${server.port}`; try { @@ -724,7 +725,7 @@ describe("a scratch workflow through POST /api/runs (issue #13)", () => { maxConcurrentRuns: 3, retainedWorkspaces: 10, }); - const server = serve({ db, adapter, port: 0, config }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0, config }); const base = `http://localhost:${server.port}`; try { @@ -797,7 +798,7 @@ describe("ctx.dispatch through POST /api/runs (issue #14)", () => { const server = serve({ db, - adapter, + runtime: makeAgentRuntime(adapter), port: 0, config: defineConfig({ repo: { @@ -895,7 +896,7 @@ describe("GET /api/schedules (issue #17)", () => { }); const server = serve({ db, - adapter: createSlowFakeAdapter([], 1), + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), port: 0, config: schedulesConfig(root, workflow), }); @@ -962,7 +963,12 @@ describe("GET /api/schedules (issue #17)", () => { }, ], }); - const server = serve({ db, adapter: createSlowFakeAdapter([], 1), port: 0, config }); + const server = serve({ + db, + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; const { Cron } = await import("effect"); @@ -983,7 +989,7 @@ 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, adapter: createSlowFakeAdapter([], 1), port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), port: 0 }); const base = `http://localhost:${server.port}`; try { @@ -1025,7 +1031,12 @@ describe("GET /api/schedules (issue #17)", () => { }, ], }); - const server = serve({ db, adapter: createSlowFakeAdapter([], 1), port: 0, config }); + const server = serve({ + db, + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -1121,7 +1132,12 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { }, ], }); - const server = serve({ db, adapter: createSlowFakeAdapter([], 1), port: 0, config }); + const server = serve({ + db, + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -1170,7 +1186,12 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { { id: "nightly-run", workflow: workflow.id, input: {}, cron: "0 3 * * *", timezone: "UTC" }, ], }); - const server = serve({ db, adapter: createSlowFakeAdapter([], 1), port: 0, config }); + const server = serve({ + db, + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -1208,7 +1229,12 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { { id: "slow-skip", workflow: workflow.id, input: {}, cron: "0 3 * * *", timezone: "UTC" }, ], }); - const server = serve({ db, adapter: createSlowFakeAdapter([], 1), port: 0, config }); + const server = serve({ + db, + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -1268,7 +1294,12 @@ describe("POST /api/schedules/:id/run (issue #17)", () => { }, ], }); - const server = serve({ db, adapter: createSlowFakeAdapter([], 1), port: 0, config }); + const server = serve({ + db, + runtime: makeAgentRuntime(createSlowFakeAdapter([], 1)), + port: 0, + config, + }); const base = `http://localhost:${server.port}`; try { @@ -1314,7 +1345,7 @@ 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, adapter, port: 0 }); + const server = serve({ db, runtime: makeAgentRuntime(adapter), port: 0 }); const base = `http://localhost:${server.port}`; try { diff --git a/src/server/http.ts b/src/server/http.ts index 1c5cc6b..755a1fb 100644 --- a/src/server/http.ts +++ b/src/server/http.ts @@ -30,11 +30,11 @@ import type { Database } from "bun:sqlite"; import { isTerminal, type RunEvent } from "../events"; import type { FactoryConfig } from "../config"; -import { Schema, SchemaParser } from "effect"; +import { Schema, SchemaParser, type ManagedRuntime } 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 { AgentAdapter } from "../runtime/agent-adapter"; +import type { AgentRuntime } from "../runtime/agent-runtime"; import index from "../web/index.html"; import { admitRun } from "./admission"; import { subscribe } from "./pubsub"; @@ -74,7 +74,12 @@ export interface ScheduleSummary { export interface ServerOptions { readonly db: Database; - readonly adapter: AgentAdapter; + /** + * 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`. + */ + readonly runtime: ManagedRuntime.ManagedRuntime; /** * The loaded `factory.config.ts` (D27). Absent = the phase 3 path-based API * behaves exactly as before (no limit, explicit dir+clone, `/api/workflows` @@ -131,7 +136,7 @@ function listSummaries(db: Database): ReadonlyArray { * it `ctx.dispatch` throws, because in-process/path-based legacy runs have * no registry to start children from. */ -function dispatchEnvFor(config: FactoryConfig, adapter: AgentAdapter): DispatchEnv { +function dispatchEnvFor(config: FactoryConfig): DispatchEnv { return { workspace: { workspaceRoot: config.workspaceRoot, @@ -141,7 +146,6 @@ function dispatchEnvFor(config: FactoryConfig, adapter: AgentAdapter): DispatchE }, repo: { slug: config.repo.slug, baseBranch: config.repo.baseBranch }, maxConcurrentRuns: config.maxConcurrentRuns, - adapter, maxDispatchDepth: config.maxDispatchDepth, maxChildrenPerRun: config.maxChildrenPerRun, }; @@ -155,14 +159,13 @@ function dispatchEnvFor(config: FactoryConfig, adapter: AgentAdapter): DispatchE */ function configRunOptions( config: FactoryConfig, - adapter: AgentAdapter, maxConcurrentRuns: number | undefined, extra: { readonly scheduleId?: string; readonly dedupeKey?: string; readonly agentOverrides?: { readonly model?: string }; } = {}, -): Omit { +): Omit { return { workspace: { workspaceRoot: config.workspaceRoot, @@ -172,7 +175,7 @@ function configRunOptions( }, repo: { slug: config.repo.slug, baseBranch: config.repo.baseBranch }, maxConcurrentRuns, - dispatchEnv: dispatchEnvFor(config, adapter), + dispatchEnv: dispatchEnvFor(config), ...(extra.scheduleId !== undefined ? { scheduleId: extra.scheduleId } : {}), ...(extra.dedupeKey !== undefined ? { dedupeKey: extra.dedupeKey } : {}), ...(extra.agentOverrides !== undefined ? { agentOverrides: extra.agentOverrides } : {}), @@ -388,15 +391,14 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise } const startOptions: StartTrackedRunOptions = { - ...configRunOptions(runEnv, options.adapter, maxConcurrentRuns, { + ...configRunOptions(runEnv, maxConcurrentRuns, { dedupeKey: typeof body.dedupeKey === "string" ? body.dedupeKey : undefined, }), input: decodedInput, - adapter: options.adapter, }; let runId: string; try { - runId = await startTrackedRun(options.db, workflow, startOptions); + runId = await startTrackedRun(options.runtime, options.db, workflow, startOptions); } catch (err) { if (err instanceof ConcurrencyLimitError) { return json({ error: err.message }, { status: 409 }); @@ -456,9 +458,8 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise ? undefined : { slug: runEnv.repo.slug, baseBranch: runEnv.repo.baseBranch }, ...(runEnv !== undefined ? { maxConcurrentRuns } : {}), - ...(runEnv !== undefined ? { dispatchEnv: dispatchEnvFor(runEnv, options.adapter) } : {}), + ...(runEnv !== undefined ? { dispatchEnv: dispatchEnvFor(runEnv) } : {}), input: body.input, - adapter: options.adapter, ...(typeof body.dedupeKey === "string" ? { dedupeKey: body.dedupeKey } : {}), ...(body.clone !== undefined && typeof body.dir === "string" ? { prepareWorkspace: true } @@ -466,7 +467,7 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise }; let runId: string; try { - runId = await startTrackedRun(options.db, workflow, startOptions); + runId = await startTrackedRun(options.runtime, options.db, workflow, startOptions); } catch (err) { if (err instanceof ConcurrencyLimitError) { return json({ error: err.message }, { status: 409 }); @@ -559,14 +560,13 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise // it fires regardless. let runId: string; try { - runId = await startTrackedRun(options.db, workflow, { - ...configRunOptions(options.config, options.adapter, maxConcurrentRuns, { + runId = await startTrackedRun(options.runtime, options.db, workflow, { + ...configRunOptions(options.config, maxConcurrentRuns, { scheduleId: schedule.id, ...(schedule.overlap === "skip" ? { dedupeKey: `schedule:${schedule.id}` } : {}), ...(schedule.agent !== undefined ? { agentOverrides: schedule.agent } : {}), }), input: schedule.input, - adapter: options.adapter, }); } catch (err) { if (err instanceof ConcurrencyLimitError) { diff --git a/src/server/nested-runs.test.ts b/src/server/nested-runs.test.ts index cf9c45c..7c9e8ac 100644 --- a/src/server/nested-runs.test.ts +++ b/src/server/nested-runs.test.ts @@ -20,6 +20,7 @@ import { getRunEvents, listRuns, openStore } from "../persistence/store"; import { createSlowFakeAdapter } from "../replay/adapter"; import { activeRunIds, isActive, startTrackedRun } from "./runs"; import { defineWorkflow } from "../workflow"; +import { makeAgentRuntime } from "../runtime/agent-runtime"; const SLOW_ADAPTER = createSlowFakeAdapter( [ @@ -30,6 +31,8 @@ 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", { @@ -89,7 +92,7 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - const parentRunId = await startTrackedRun(db, parent, { + const parentRunId = await startTrackedRun(runtime, db, parent, { workspace: { workspaceRoot: join(root, "workspaces"), sshUrl: join(root, "seed-not-used"), @@ -97,7 +100,6 @@ describe("nested runs through the daemon (issue #14)", () => { retainedWorkspaces: 10, }, input: {}, - adapter: SLOW_ADAPTER, dispatchEnv: { workspace: { workspaceRoot: join(root, "workspaces"), @@ -105,7 +107,6 @@ describe("nested runs through the daemon (issue #14)", () => { identity: { name: "Test Bot", email: "test@factory.local" }, retainedWorkspaces: 10, }, - adapter: SLOW_ADAPTER, }, }); @@ -145,9 +146,8 @@ describe("nested runs through the daemon (issue #14)", () => { const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "workspaces"), { recursive: true }); - await startTrackedRun(db, selfDispatching, { + await startTrackedRun(runtime, db, selfDispatching, { input: {}, - adapter: SLOW_ADAPTER, workspace: { workspaceRoot: join(root, "workspaces"), sshUrl: join(root, "seed-not-used"), @@ -162,7 +162,6 @@ describe("nested runs through the daemon (issue #14)", () => { identity: { name: "Test Bot", email: "test@factory.local" }, retainedWorkspaces: 20, }, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 20, }, }); @@ -198,9 +197,8 @@ describe("nested runs through the daemon (issue #14)", () => { const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "workspaces"), { recursive: true }); - const parentRunId = await startTrackedRun(db, manyChildren, { + const parentRunId = await startTrackedRun(runtime, db, manyChildren, { input: {}, - adapter: SLOW_ADAPTER, workspace: { workspaceRoot: join(root, "workspaces"), sshUrl: join(root, "seed-not-used"), @@ -215,7 +213,6 @@ describe("nested runs through the daemon (issue #14)", () => { identity: { name: "Test Bot", email: "test@factory.local" }, retainedWorkspaces: 30, }, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 40, maxChildrenPerRun: 5, }, @@ -250,9 +247,8 @@ describe("nested runs through the daemon (issue #14)", () => { }, }); - const parentRunId = await startTrackedRun(db, parent, { + const parentRunId = await startTrackedRun(runtime, db, parent, { input: {}, - adapter: SLOW_ADAPTER, workspace: { workspaceRoot: join(root, "workspaces"), sshUrl: join(root, "seed-not-used"), @@ -267,7 +263,6 @@ describe("nested runs through the daemon (issue #14)", () => { identity: { name: "Test Bot", email: "test@factory.local" }, retainedWorkspaces: 5, }, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 1, }, }); diff --git a/src/server/runs.test.ts b/src/server/runs.test.ts index 68ba552..7e050bc 100644 --- a/src/server/runs.test.ts +++ b/src/server/runs.test.ts @@ -18,6 +18,7 @@ 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, @@ -38,6 +39,8 @@ const SLOW_ADAPTER = createSlowFakeAdapter( 25, ); +const runtime = makeAgentRuntime(SLOW_ADAPTER); + const sleepWorkflow = defineWorkflow("sleep-test", { input: Schema.Struct({}), run: async (ctx) => { @@ -101,11 +104,10 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" const db = openStore(join(root, "factory.db")); const gate = makeGate(); - const first = startTrackedRun(db, echoWorkflow, { + const first = startTrackedRun(runtime, db, echoWorkflow, { runId: "run-first", dir: join(root, "first-dir"), input: {}, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 1, beforeStart: gate.wait, }); @@ -114,11 +116,10 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" expect(activeRunIds()).toEqual(["run-first"]); expect(getActiveHandle("run-first")).toBeUndefined(); - const second = startTrackedRun(db, echoWorkflow, { + const second = startTrackedRun(runtime, db, echoWorkflow, { runId: "run-second", dir: join(root, "second-dir"), input: {}, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 1, }); @@ -144,10 +145,9 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" const db = openStore(join(root, "factory.db")); await expect( - startTrackedRun(db, echoWorkflow, { + startTrackedRun(runtime, db, echoWorkflow, { runId: "run-doomed", input: {}, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 1, workspace: { workspaceRoot: join(root, "workspaces"), @@ -160,10 +160,9 @@ describe("startTrackedRun admission (M1: the slot is reserved before any await)" expect(activeRunIds()).toEqual([]); - const runId = await startTrackedRun(db, echoWorkflow, { + const runId = await startTrackedRun(runtime, db, echoWorkflow, { dir: join(root, "dir"), input: {}, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 1, }); expect(activeRunIds()).toEqual([runId]); @@ -181,11 +180,10 @@ describe("cancel of a reserved-but-not-started run (L1)", () => { mkdirSync(join(root, "dir"), { recursive: true }); const gate = makeGate(); - const starting = startTrackedRun(db, sleepWorkflow, { + const starting = startTrackedRun(runtime, db, sleepWorkflow, { runId: "run-gated", dir: join(root, "dir"), input: {}, - adapter: SLOW_ADAPTER, maxConcurrentRuns: 1, beforeStart: gate.wait, }); @@ -251,11 +249,10 @@ describe("scratch workspaces through startTrackedRun (issue #13)", () => { const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "seed"), { recursive: true }); - const runId = await startTrackedRun(db, failingScratchWorkflow, { + const runId = await startTrackedRun(runtime, db, failingScratchWorkflow, { runId: "run-scratch-empty", workspace: { ...workspaceSpec(), sshUrl: join(root, "no-such-remote") }, input: {}, - adapter: SLOW_ADAPTER, }); await waitFor(() => !isActive(runId)); @@ -273,21 +270,19 @@ describe("scratch workspaces through startTrackedRun (issue #13)", () => { const db = openStore(join(root, "factory.db")); mkdirSync(join(root, "seed"), { recursive: true }); - const okId = await startTrackedRun(db, scratchWorkflow, { + const okId = await startTrackedRun(runtime, db, scratchWorkflow, { runId: "run-scratch-ok", workspace: workspaceSpec(), input: {}, - adapter: SLOW_ADAPTER, }); await waitFor(() => !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(db, failingScratchWorkflow, { + const badId = await startTrackedRun(runtime, db, failingScratchWorkflow, { runId: "run-scratch-bad", workspace: workspaceSpec(), input: {}, - adapter: SLOW_ADAPTER, }); await waitFor(() => !isActive(badId)); await new Promise((resolve) => setTimeout(resolve, 50)); @@ -319,11 +314,10 @@ 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(db, echoWorkflow, { + await startTrackedRun(runtime, db, echoWorkflow, { runId: "run-clone", workspace: { ...workspaceSpec(), sshUrl: seed, retainedWorkspaces: 1 }, input: {}, - adapter: SLOW_ADAPTER, }); await waitFor(() => !isActive("run-clone")); @@ -364,7 +358,7 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { await Bun.$`git init -b main -q ${seed}`.quiet(); const adapter = trackingAdapter(); - const runId = await startTrackedRun(db, echoWorkflow, { + const runId = await startTrackedRun(makeAgentRuntime(adapter), db, echoWorkflow, { runId: "run-prep-clone", workspace: { workspaceRoot: join(root, "workspaces"), @@ -373,7 +367,6 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { retainedWorkspaces: 10, }, input: {}, - adapter, }); await waitFor(() => !isActive(runId)); @@ -386,7 +379,7 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { const db = openStore(join(root, "factory.db")); const adapter = trackingAdapter(); - const runId = await startTrackedRun(db, scratchWorkflow, { + const runId = await startTrackedRun(makeAgentRuntime(adapter), db, scratchWorkflow, { runId: "run-prep-scratch", workspace: { workspaceRoot: join(root, "workspaces"), @@ -395,7 +388,6 @@ describe("adapter.prepareWorkspace through startTrackedRun (#37)", () => { retainedWorkspaces: 10, }, input: {}, - adapter, }); await waitFor(() => !isActive(runId)); diff --git a/src/server/runs.ts b/src/server/runs.ts index 464f7c8..3630e7a 100644 --- a/src/server/runs.ts +++ b/src/server/runs.ts @@ -13,7 +13,7 @@ * start rather than dropped. */ -import { Schema } from "effect"; +import { Schema, type ManagedRuntime } from "effect"; import type { Database } from "bun:sqlite"; import { rm } from "node:fs/promises"; import { admitRun } from "./admission"; @@ -23,7 +23,7 @@ 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 { AgentAdapter } from "../runtime/agent-adapter"; +import type { AgentRuntime } from "../runtime/agent-runtime"; import type { DispatchChildFn, WorkflowDefinition, WorkspaceKind } from "../workflow"; import { publish } from "./pubsub"; @@ -120,7 +120,6 @@ export interface StartTrackedRunOptions { /** Write-back environment for the run (D32): came from config, not the caller. */ readonly repo?: RunRepo; readonly input: unknown; - readonly adapter: AgentAdapter; readonly runId?: string; /** * D29's ceiling, enforced atomically at reservation time — the reservation @@ -183,7 +182,7 @@ export interface StartTrackedRunOptions { /** * The environment `ctx.dispatch`'s children share with their parent (issue - * #14): workspace provisioning, write-back environment, adapter and the + * #14): workspace provisioning, write-back environment and the * concurrency ceiling — the same wiring a config-backed server wraps every * run in, simply reused for child runs (children inherit it, so a child can * dispatch grandchildren). @@ -192,7 +191,6 @@ export interface DispatchEnv { readonly workspace?: WorkspaceSpec; readonly repo?: RunRepo; readonly maxConcurrentRuns?: number; - readonly adapter: AgentAdapter; /** Issue #14: dispatch depth / per-run child caps, over the defaults. */ readonly maxDispatchDepth?: number; readonly maxChildrenPerRun?: number; @@ -236,6 +234,7 @@ function dispatchDepth(db: Database, runId: string): number { * parent's. */ async function dispatchChildRun( + runtime: ManagedRuntime.ManagedRuntime, db: Database, env: DispatchEnv, parentRunId: string, @@ -276,13 +275,12 @@ async function dispatchChildRun( if (opts?.dedupeKey !== undefined) registry.claim(opts.dedupeKey, childRunId); void (async () => { - await startTrackedRun(db, child, { + await startTrackedRun(runtime, db, child, { runId: childRunId, ...(env.workspace !== undefined ? { workspace: env.workspace } : {}), ...(env.repo !== undefined ? { repo: env.repo } : {}), ...(env.maxConcurrentRuns !== undefined ? { maxConcurrentRuns: env.maxConcurrentRuns } : {}), input, - adapter: env.adapter, parentRunId, dispatchEnv: env, ...(opts?.dedupeKey !== undefined ? { dedupeKey: opts.dedupeKey } : {}), @@ -308,6 +306,7 @@ function countDispatchedChildren(db: Database, runId: string): number { /** Starts a run, persists+publishes every event, and tracks it until terminal. */ export async function startTrackedRun( + runtime: ManagedRuntime.ManagedRuntime, db: Database, workflow: WorkflowDefinition, options: StartTrackedRunOptions, @@ -380,9 +379,9 @@ export async function startTrackedRun( options.dispatchEnv === undefined ? undefined : (child, input, opts) => - dispatchChildRun(db, options.dispatchEnv!, runId, child, input, opts); + dispatchChildRun(runtime, db, options.dispatchEnv!, runId, child, input, opts); - const handle = startRun(workflow, { + const handle = startRun(workflow, runtime, { runId, dir, ...(options.repo !== undefined ? { repo: options.repo } : {}), @@ -395,7 +394,6 @@ export async function startTrackedRun( ...(options.scheduleId !== undefined ? { scheduleId: options.scheduleId } : {}), ...(options.agentOverrides !== undefined ? { agentOverrides: options.agentOverrides } : {}), input: options.input, - adapter: options.adapter, onEvent: (event) => { appendEvent(db, event); publish(runId, event); diff --git a/src/server/scheduler.ts b/src/server/scheduler.ts index 01bcabd..6a97d67 100644 --- a/src/server/scheduler.ts +++ b/src/server/scheduler.ts @@ -24,7 +24,7 @@ * like any other dedupe collision; `"stack"` fires regardless. */ -import { Cron, Effect, Schedule, Schema } from "effect"; +import { Cron, Effect, Schedule, Schema, type ManagedRuntime } from "effect"; import type { Database } from "bun:sqlite"; import type { FactoryConfig } from "../config"; import { DedupeKeyError, dedupeRegistry, type DedupeRegistry } from "../lib/dedupe"; @@ -37,7 +37,7 @@ import { type WorkspaceSpec, } from "./runs"; import { RunCancelledSignal, type RunRepo } from "../runtime/run"; -import type { AgentAdapter } from "../runtime/agent-adapter"; +import type { AgentRuntime } from "../runtime/agent-runtime"; export class SchedulerError extends Schema.TaggedError()("SchedulerError", { cause: Schema.Defect(), @@ -252,7 +252,7 @@ export function runSchedulerLoop( */ export function makeScheduleFire(options: { readonly db: Database; - readonly adapter: AgentAdapter; + readonly runtime: ManagedRuntime.ManagedRuntime; readonly maxConcurrentRuns: number; readonly workspace: WorkspaceSpec; readonly repo: RunRepo; @@ -260,10 +260,9 @@ export function makeScheduleFire(options: { }): (schedule: RuntimeSchedule) => Promise { const env = options; return async (schedule: RuntimeSchedule): Promise => { - return startTrackedRun(env.db, schedule.workflow, { + return startTrackedRun(env.runtime, env.db, schedule.workflow, { input: schedule.input, repo: env.repo, - adapter: options.adapter, maxConcurrentRuns: options.maxConcurrentRuns, workspace: env.workspace, dispatchEnv: env.dispatchEnv, diff --git a/src/workflow.dispatch.test.ts b/src/workflow.dispatch.test.ts index 0f0e4f0..3619842 100644 --- a/src/workflow.dispatch.test.ts +++ b/src/workflow.dispatch.test.ts @@ -19,6 +19,7 @@ import { describe, expect, test } from "bun:test"; import { Schema } from "effect"; import type { RunEvent } from "./events"; import { createSlowFakeAdapter } from "./replay/adapter"; +import { makeAgentRuntime } from "./runtime/agent-runtime"; import { startRun } from "./runtime/run"; import { defineWorkflow } from "./workflow"; @@ -56,11 +57,10 @@ describe("ctx.dispatch (issue #14)", () => { }, }); - const handle = startRun(parent, { + const handle = startRun(parent, makeAgentRuntime(SLOW_ADAPTER), { runId: "run-parent-1", dir: root, input: {}, - adapter: SLOW_ADAPTER, dispatch: (_child, _input): Promise => { void _child; void _input; @@ -93,19 +93,17 @@ describe("ctx.dispatch (issue #14)", () => { let childResult: Promise<{ outcome: string }> | undefined; - const handle = startRun(parent, { + const handle = startRun(parent, makeAgentRuntime(SLOW_ADAPTER), { runId: "run-parent-x", dir: root, input: {}, - adapter: SLOW_ADAPTER, dispatch: (childWorkflow, input) => { void childWorkflow.id; childEvents = []; - const childRun = startRun(childWorkflow as never, { + const childRun = startRun(childWorkflow as never, makeAgentRuntime(SLOW_ADAPTER), { runId: "run-child-x", dir: root, input, - adapter: SLOW_ADAPTER, parentRunId: "run-parent-x", onEvent: (event) => { childEvents.push(event); @@ -147,11 +145,10 @@ describe("ctx.dispatch (issue #14)", () => { run: async (ctx) => ctx.dispatch(numberedChild, { n: 1 }), }); - const handle = startRun(parent, { + const handle = startRun(parent, makeAgentRuntime(SLOW_ADAPTER), { runId: "run-no-daemon", dir: root, input: {}, - adapter: SLOW_ADAPTER, onEvent: () => undefined, });