From 12c4300af494f082de8940eb199a1d9c8177f938 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Mon, 25 May 2026 13:10:33 -0400 Subject: [PATCH 1/3] fix(recommender): run on the daemon's engine, not a standalone second one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The recommender never actually ran from the daemon. dispatchRecommender stood up its OWN HookServer (on the dispatcher port → EADDRINUSE for mm run-recommender while the daemon was up) AND its OWN embedded engine inside the daemon (a second in-process engine that never processed the job → 202 'enqueued' but no workflow row, no tmux, no run). Dispatch was migrated to the daemon's engine long ago; the recommender was left on the Phase-7 standalone path. The autonomous loop's first link was broken end to end. - recommender workflow: per-repo settings via an optional resolveRunSettings(repo) resolver (schemaPath/config/repoConfig/agentTimeoutMs), so ONE registration on the daemon's long-lived engine serves every managed repo — mirroring how the implementation workflow resolves per-repo. Static fields stay as the fallback for the standalone runner (backward-compatible). - main.ts: register the recommender on the daemon's engine; runRecommenderForRepo now engine.start("recommender", …) on that engine (reusing the daemon's HookServer/sessionGate + dispatcherUrl), exactly like startDispatchImpl. - mm run-recommender: now a thin client — auto-starts the daemon (like mm dispatch) and POSTs /trigger/recommender; no more standalone engine/port clash. - integration test: proves the daemon path RUNS the recommender on the engine and creates the recommender workflow row (the row the dead-engine path never made). --- packages/cli/src/commands/run-recommender.ts | 117 +++++++++++++----- packages/cli/test/run-recommender.test.ts | 98 ++++++++++----- packages/dispatcher/src/main.ts | 95 +++++++++++--- packages/dispatcher/src/recommender-run.ts | 2 +- .../dispatcher/src/workflows/recommender.ts | 76 +++++++++--- .../test/recommender-workflow.test.ts | 52 ++++++++ 6 files changed, 345 insertions(+), 95 deletions(-) diff --git a/packages/cli/src/commands/run-recommender.ts b/packages/cli/src/commands/run-recommender.ts index 443a50c1..13bb762b 100644 --- a/packages/cli/src/commands/run-recommender.ts +++ b/packages/cli/src/commands/run-recommender.ts @@ -1,31 +1,70 @@ import { existsSync } from "node:fs"; -import { join } from "node:path"; -import { claudeAdapter } from "@middle/adapter-claude"; -import type { AgentAdapter } from "@middle/core"; +import { join, resolve } from "node:path"; import { loadConfig } from "@middle/core"; -import { - dispatchRecommender, - resolveRecommenderOptions, -} from "@middle/dispatcher/src/recommender-run.ts"; +import { runStart, type StartOptions } from "./start.ts"; export type RunRecommenderOptions = { /** Override the global config path (defaults to `~/.middle/config.toml`). */ configPath?: string; - /** Injected dispatch seam — defaults to the real runner. Tests override it. */ - dispatch?: typeof dispatchRecommender; + /** Override the daemon spawn (defaults to {@link runStart}). Returns its exit code. */ + startDaemon?: (opts: StartOptions) => number; + /** Readiness-poll budget after a spawn before giving up (default 10000ms). */ + healthTimeoutMs?: number; + /** Probe the daemon's `/health` (injectable for tests; defaults to a real fetch). */ + probeHealth?: (base: string) => Promise; + /** POST the recommender trigger (injectable for tests; defaults to a real fetch). */ + trigger?: (base: string, repoPath: string) => Promise<{ status: number; body: string }>; }; -/** Phase 7 adapter registry — only `claude` is implemented. */ -function getAdapter(name: string): AgentAdapter { - if (name !== "claude") throw new Error(`unknown adapter: ${name}`); - return claudeAdapter; +const DEFAULT_HEALTH_TIMEOUT_MS = 10_000; + +/** Probe `GET /health`; true only on `{ ok: true }`. Connection errors are "down", not a throw. */ +async function probeHealthDefault(base: string): Promise { + try { + const res = await fetch(`${base}/health`); + if (!res.ok) return false; + const body = (await res.json().catch(() => null)) as { ok?: unknown } | null; + return body?.ok === true; + } catch { + return false; + } +} + +/** Poll `/health` (via `probe`) until ready or the deadline. */ +async function waitForHealth( + base: string, + timeoutMs: number, + probe: (base: string) => Promise, +): Promise { + const deadline = Date.now() + timeoutMs; + for (;;) { + if (await probe(base)) return true; + if (Date.now() >= deadline) return false; + await Bun.sleep(50); + } +} + +/** POST `/trigger/recommender` with the repo's checkout path; relay status + body. */ +async function triggerDefault( + base: string, + repoPath: string, +): Promise<{ status: number; body: string }> { + const res = await fetch(`${base}/trigger/recommender`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ repoPath }), + }); + return { status: res.status, body: (await res.text().catch(() => "")).trim() }; } /** - * `mm run-recommender ` — trigger a recommender run for the given repo. - * Read-only at this phase: the recommender rewrites the repo's state issue but - * nothing auto-dispatches. Returns a process exit code: 0 when the run - * completes, 1 otherwise. + * `mm run-recommender ` — trigger a recommender run **through the daemon**, + * exactly like `mm dispatch`: a thin client that auto-starts the dispatcher if + * it's down, then POSTs `/trigger/recommender`. The run executes on the daemon's + * long-lived engine (not a standalone second engine that collides with the + * daemon's port). The daemon validates the repo (state issue, schema, adapter) + * and resolves per-repo settings; this command relays its verdict. Returns a + * process exit code: 0 when the run is accepted (202), 1 otherwise. */ export async function runRecommender( repoPath: string, @@ -38,32 +77,48 @@ export async function runRecommender( let config: ReturnType; try { - config = loadConfig({ - globalPath: opts.configPath, - repoPath: join(repoPath, ".middle", "config.toml"), - }); + config = loadConfig({ globalPath: opts.configPath }); } catch (error) { console.error(`mm run-recommender: failed to load config — ${(error as Error).message}`); return 1; } - const resolved = await resolveRecommenderOptions(repoPath, config, getAdapter); - if (!resolved.ok) { - console.error(`mm run-recommender: ${resolved.error}`); - return 1; + const base = `http://127.0.0.1:${config.global.dispatcherPort}`; + + // Ensure the daemon is up — auto-start it if not, same as `mm dispatch`. + const probe = opts.probeHealth ?? probeHealthDefault; + if (!(await probe(base))) { + (opts.startDaemon ?? runStart)({}); + const ready = await waitForHealth( + base, + opts.healthTimeoutMs ?? DEFAULT_HEALTH_TIMEOUT_MS, + probe, + ); + if (!ready) { + console.error(`mm run-recommender: dispatcher did not become ready on ${base}`); + return 1; + } } - const dispatch = opts.dispatch ?? dispatchRecommender; - let result: Awaited>; + let result: { status: number; body: string }; try { - result = await dispatch(resolved.options); + result = await (opts.trigger ?? triggerDefault)(base, resolve(repoPath)); } catch (error) { - console.error(`mm run-recommender: failed — ${(error as Error).message}`); + console.error( + `mm run-recommender: could not reach the dispatcher — ${(error as Error).message}`, + ); + return 1; + } + + if (result.status !== 202) { + console.error( + `mm run-recommender: dispatch rejected (${result.status})${result.body ? ` — ${result.body}` : ""}`, + ); return 1; } console.log( - `mm run-recommender: ${resolved.options.repoSlug} state issue #${resolved.options.stateIssue} → workflow ${result.workflowId} settled — ${result.state}`, + `mm run-recommender: ${resolve(repoPath)} → recommender run started on ${base} — watch it with \`mm status\` or the dashboard`, ); - return result.state === "completed" ? 0 : 1; + return 0; } diff --git a/packages/cli/test/run-recommender.test.ts b/packages/cli/test/run-recommender.test.ts index 0a94cf81..0862434c 100644 --- a/packages/cli/test/run-recommender.test.ts +++ b/packages/cli/test/run-recommender.test.ts @@ -2,7 +2,6 @@ import { afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test"; import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { DispatchRecommenderOptions } from "@middle/dispatcher/src/recommender-run.ts"; import { runRecommender } from "../src/commands/run-recommender.ts"; let dir: string; @@ -71,7 +70,7 @@ afterEach(() => { rmSync(dir, { recursive: true, force: true }); }); -describe("runRecommender — input validation", () => { +describe("runRecommender — local validation", () => { test("rejects a path that is not a git repository", async () => { const restore = silence(); try { @@ -80,64 +79,101 @@ describe("runRecommender — input validation", () => { restore(); } }); +}); + +describe("runRecommender — thin client to the daemon", () => { + const up = async () => true; + const down = async () => false; - test("rejects when no state issue is configured for the repo", async () => { - // A repo with no per-repo config (no state_issue). - const bare = join(dir, "bare"); - mkdirSync(join(bare, "schemas"), { recursive: true }); - await git(bare, ["init"]); - writeFileSync(join(bare, "schemas", "state-issue.v1.md"), "# schema\n"); + test("daemon already up: POSTs /trigger/recommender and returns 0 on 202", async () => { + const posted: Array<{ base: string; repoPath: string }> = []; + let started = 0; const restore = silence(); try { - expect(await runRecommender(bare, { configPath })).toBe(1); + const code = await runRecommender(repoPath, { + configPath, + probeHealth: up, + startDaemon: () => { + started++; + return 0; + }, + trigger: async (base, rp) => { + posted.push({ base, repoPath: rp }); + return { status: 202, body: "recommender run started" }; + }, + }); + expect(code).toBe(0); } finally { restore(); } + expect(started).toBe(0); // already up → not started + expect(posted).toHaveLength(1); + expect(posted[0]!.repoPath).toBe(repoPath); // resolved absolute checkout path + expect(posted[0]!.base).toBe("http://127.0.0.1:4120"); // the configured dispatcher port (default) }); - test("rejects when the state-issue schema is missing", async () => { - rmSync(join(repoPath, "schemas"), { recursive: true, force: true }); + test("daemon down: auto-starts it, waits for health, then triggers", async () => { + let started = 0; + let probes = 0; const restore = silence(); try { - expect(await runRecommender(repoPath, { configPath })).toBe(1); + const code = await runRecommender(repoPath, { + configPath, + probeHealth: async () => probes++ > 0, // down on the first probe, up after start + startDaemon: () => { + started++; + return 0; + }, + trigger: async () => ({ status: 202, body: "recommender run started" }), + }); + expect(code).toBe(0); } finally { restore(); } + expect(started).toBe(1); // it was down → auto-started, like `mm dispatch` }); -}); -describe("runRecommender — enqueues a recommender workflow for the repo", () => { - test("resolves config and dispatches a recommender run with the repo's state issue + adapter", async () => { - const calls: DispatchRecommenderOptions[] = []; + test("relays a daemon rejection (non-202) as exit 1", async () => { const restore = silence(); try { const code = await runRecommender(repoPath, { configPath, - dispatch: async (opts) => { - calls.push(opts); - return { workflowId: "wf-test", state: "completed" }; - }, + probeHealth: up, + startDaemon: () => 0, + trigger: async () => ({ status: 400, body: "no state issue configured for this repo" }), }); - expect(code).toBe(0); + expect(code).toBe(1); + } finally { + restore(); + } + }); + + test("returns 1 when the daemon never becomes ready after an auto-start", async () => { + const restore = silence(); + try { + const code = await runRecommender(repoPath, { + configPath, + probeHealth: down, // never ready + startDaemon: () => 0, + healthTimeoutMs: 30, + trigger: async () => ({ status: 202, body: "x" }), + }); + expect(code).toBe(1); } finally { restore(); } - expect(calls).toHaveLength(1); - const opts = calls[0]!; - expect(opts.stateIssue).toBe(42); // from the repo's config - expect(opts.adapterName).toBe("claude"); - expect(opts.repoPath).toBe(repoPath); - expect(opts.schemaPath).toBe(join(repoPath, "schemas", "state-issue.v1.md")); - // Read-only run-config: autoDispatch defaults off. - expect(opts.runConfig.autoDispatch).toBe(false); }); - test("returns 1 when the dispatched run does not complete", async () => { + test("returns 1 when the dispatcher is unreachable (the POST throws)", async () => { const restore = silence(); try { const code = await runRecommender(repoPath, { configPath, - dispatch: async () => ({ workflowId: "wf-x", state: "failed" }), + probeHealth: up, + startDaemon: () => 0, + trigger: async () => { + throw new Error("connection refused"); + }, }); expect(code).toBe(1); } finally { diff --git a/packages/dispatcher/src/main.ts b/packages/dispatcher/src/main.ts index 7a754727..85ef0ddc 100644 --- a/packages/dispatcher/src/main.ts +++ b/packages/dispatcher/src/main.ts @@ -23,16 +23,17 @@ import { collectMetrics } from "./metrics.ts"; import type { RecommenderTrigger } from "./hook-server.ts"; import { DbHookStore } from "./hook-store.ts"; import { addRateLimitObserver, clearRateLimitObservers, getRateLimitState } from "./rate-limits.ts"; -import { dispatchRecommender, resolveRecommenderOptions } from "./recommender-run.ts"; +import { ghSurfaceProblem, resolveRecommenderOptions } from "./recommender-run.ts"; import { ghPollGateway } from "./poller-gateway.ts"; import { startPoller } from "./poller-cron.ts"; import { startRecommenderCron } from "./recommender-cron.ts"; import { isPaused, listManagedRepos, registerManagedRepo } from "./repo-config.ts"; import { getSlotState, hasFreeSlot } from "./slots.ts"; import { ghStateIssueGateway, readState, type StateIssueGateway } from "./state-issue.ts"; -import { killSession, status } from "./tmux.ts"; +import { killSession, newSession, sendEnter, sendText, status } from "./tmux.ts"; import { startWatchdog } from "./watchdog-cron.ts"; -import { pruneWorktreeAt } from "./worktree.ts"; +import { createWorktree, destroyWorktree, pruneWorktreeAt } from "./worktree.ts"; +import { buildRecommenderContext, createRecommenderWorkflow } from "./workflows/recommender.ts"; import { addWorkflowObserver, clearWorkflowObservers, @@ -333,14 +334,15 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { }); void disposeRateLimitObserver; // daemon clears all observers on shutdown - // Run the recommender for a repo by checkout path: load its merged config, - // resolve the run options, register the repo (durable + in-memory), and fire - // the run on an ephemeral engine/port with the auto-dispatch trigger wired — - // Trigger #1: when the run completes (clean parse + auto_dispatch on), the - // recommender workflow fires `triggerAutoDispatch` back into THIS daemon's - // engine to run the loop. Shared by the `/trigger/recommender` route AND the - // periodic recommender cron (#135), so both behave identically. The run itself - // is fire-and-forget; this returns once it's launched. + // Run the recommender for a repo by checkout path. The recommender runs on the + // daemon's OWN long-lived engine (registered below), exactly like dispatch — + // NOT a second ephemeral engine/HookServer (the old standalone path collided + // with the daemon's port and its in-process second engine never processed the + // job). This resolves the run input (slug, state-issue number, adapter), + // registers the repo, then `engine.start("recommender", …)`. On a clean run the + // workflow's trigger-auto-dispatch step fires `scheduleAutoDispatch` back into + // this same engine (Trigger #1). Shared by the `/trigger/recommender` route and + // the cron, so both behave identically. Returns once enqueued. async function runRecommenderForRepo( repoPath: string, ): Promise<{ status: number; body: string }> { @@ -356,13 +358,15 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { const resolved = await resolveRecommenderOptions(repoPath, repoConfig, getAdapter); if (!resolved.ok) return { status: 400, body: resolved.error }; rememberRepoPath(resolved.options.repoSlug, repoPath); - void dispatchRecommender({ - ...resolved.options, - dispatcherPort: 0, - triggerAutoDispatch: async ({ repo }) => scheduleAutoDispatch(repo), - }).catch((error: unknown) => { - console.error(`[main] recommender run failed: ${(error as Error).message}`); - }); + try { + await engine.start("recommender", { + repo: resolved.options.repoSlug, + stateIssue: resolved.options.stateIssue, + adapter: resolved.options.adapterName, + }); + } catch (error) { + return { status: 500, body: `recommender enqueue failed: ${(error as Error).message}` }; + } return { status: 202, body: "recommender run started" }; } @@ -461,6 +465,61 @@ export async function runDaemon(opts: RunDaemonOptions = {}): Promise { // serviced on a later event-loop tick, by which point the workflow is registered. engine.register(createImplementationWorkflow(deps)); + // Register the RECOMMENDER on this same long-lived engine (not a second + // ephemeral one) — so `runRecommenderForRepo`'s `engine.start("recommender")` + // actually runs, reusing the daemon's HookServer/sessionGate + dispatcherUrl. + // Per-repo settings/context resolve from the input repo's config at run time + // (`resolveRunSettings`/`gatherContext`), so one registration serves every + // managed repo — mirroring how the implementation workflow resolves per-repo. + engine.register( + createRecommenderWorkflow({ + db, + getAdapter, + sessionGate: deps.sessionGate, + tmux: { newSession, sendText, sendEnter, killSession }, + worktree: { createWorktree, destroyWorktree }, + resolveRepoPath: (repo) => { + const path = repoPaths.get(repo); + if (path === undefined) throw new Error(`no checkout path registered for repo ${repo}`); + return path; + }, + worktreeRoot: config.global.worktreeRoot, + dispatcherUrl: deps.dispatcherUrl, + stateIssue: ghStateIssueGateway, + surfaceProblem: ghSurfaceProblem, + triggerAutoDispatch: async ({ repo }) => scheduleAutoDispatch(repo), + gatherContext: (repo) => { + const cfg = loadRepoConfig(repo); + if (!cfg) throw new Error(`recommender: no config for repo ${repo}`); + return buildRecommenderContext({ + db, + repo, + adapters: Object.keys(cfg.adapters), + maxPerAdapter: cfg.limits?.maxConcurrentPerAdapter ?? {}, + repoMax: cfg.limits?.maxConcurrent ?? cfg.global.maxConcurrent, + globalMax: cfg.global.maxConcurrent, + }); + }, + resolveRunSettings: (repo) => { + const repoPath = repoPaths.get(repo); + const cfg = loadRepoConfig(repo); + if (repoPath === undefined || !cfg) { + throw new Error(`recommender: repo ${repo} is not registered/configured`); + } + return { + schemaPath: join(repoPath, "schemas", "state-issue.v1.md"), + config: { + defaultAdapter: cfg.global.defaultAdapter, + autoDispatch: cfg.recommender?.autoDispatch ?? false, + prMode: cfg.repo?.prMode ?? "worktree", + }, + repoConfig: { adapters: Object.keys(cfg.adapters) }, + agentTimeoutMs: cfg.recommender?.agentTimeoutMs, + }; + }, + }), + ); + // Watchdog cron: every 30s, correct transcript drift then reconcile every // launching/running workflow (launch-timeout, tmux liveness, idle detection, // sentinel re-arm). The reconcile logic is adapter-agnostic via getAdapter. diff --git a/packages/dispatcher/src/recommender-run.ts b/packages/dispatcher/src/recommender-run.ts index 79cb684c..f35de238 100644 --- a/packages/dispatcher/src/recommender-run.ts +++ b/packages/dispatcher/src/recommender-run.ts @@ -156,7 +156,7 @@ export async function resolveRecommenderOptions( } /** Default human surface: comment the problem on the state issue via `gh`. */ -async function ghSurfaceProblem(opts: { +export async function ghSurfaceProblem(opts: { repo: string; stateIssue: number; problem: string; diff --git a/packages/dispatcher/src/workflows/recommender.ts b/packages/dispatcher/src/workflows/recommender.ts index 3f4788c8..e9dbdb21 100644 --- a/packages/dispatcher/src/workflows/recommender.ts +++ b/packages/dispatcher/src/workflows/recommender.ts @@ -69,6 +69,14 @@ export type StateIssueReader = { readBody(repo: string, issueNumber: number): Promise; }; +/** The per-repo settings the recommender workflow resolves for each run. */ +export type RecommenderRunSettings = { + schemaPath: string; + config: RecommenderRunConfig; + repoConfig: RepoConfig; + agentTimeoutMs?: number; +}; + /** Everything the recommender workflow needs that is not part of its per-run input. */ export type RecommenderDeps = { db: Database; @@ -79,14 +87,27 @@ export type RecommenderDeps = { resolveRepoPath: (repo: string) => string; worktreeRoot: string; dispatcherUrl: string; - /** On-disk path to `state-issue.v1.md` the recommender is pointed at. */ - schemaPath: string; + /** + * On-disk path to `state-issue.v1.md`. Per-repo via {@link resolveRunSettings} + * on the daemon; the standalone runner supplies it statically. + */ + schemaPath?: string; /** Reads the state issue body — `prior_body` for the prompt, and the produced body to verify. */ stateIssue: StateIssueReader; - /** Configured adapter names, for `validate()` in the verify step. */ - repoConfig: RepoConfig; - /** The `config` block reported to the recommender. */ - config: RecommenderRunConfig; + /** Configured adapter names, for `validate()` in the verify step (static-runner path). */ + repoConfig?: RepoConfig; + /** The `config` block reported to the recommender (static-runner path). */ + config?: RecommenderRunConfig; + /** + * Per-repo run settings resolver — the **daemon path**. When set, the workflow + * resolves `schemaPath`/`config`/`repoConfig`/`agentTimeoutMs` from this for + * each run's `input.repo`, so ONE workflow registration on the daemon's + * long-lived engine serves every managed repo (mirrors how the implementation + * workflow resolves per-repo). When absent, the workflow falls back to the + * static fields above (the standalone `dispatchRecommender` path). Provide one + * or the other. + */ + resolveRunSettings?: (repo: string) => RecommenderRunSettings; /** * Gather the dispatcher-owned context (rate limits, in-flight, slots) verbatim. * Injected so tests stub it and the runner wires the real db/config-backed @@ -284,7 +305,30 @@ type VerifyResult = { ok: boolean; errors: string[] }; */ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow { const launchTimeout = deps.launchTimeoutMs ?? DEFAULT_LAUNCH_TIMEOUT_MS; - const agentTimeout = deps.agentTimeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS; + + /** + * Resolve the run's per-repo settings: the daemon's `resolveRunSettings(repo)` + * when wired (one registration serves every repo), else the static deps (the + * standalone runner). One path must be present — a missing both is a wiring bug. + */ + function runSettings(repo: string): RecommenderRunSettings { + if (deps.resolveRunSettings) return deps.resolveRunSettings(repo); + if ( + deps.schemaPath === undefined || + deps.config === undefined || + deps.repoConfig === undefined + ) { + throw new Error( + "recommender deps: provide resolveRunSettings (daemon) or schemaPath+config+repoConfig (standalone)", + ); + } + return { + schemaPath: deps.schemaPath, + config: deps.config, + repoConfig: deps.repoConfig, + agentTimeoutMs: deps.agentTimeoutMs, + }; + } /** Tear down the worktree + session. Both the final step and the prepare * compensation route here; idempotent so running it twice is safe. */ @@ -336,13 +380,14 @@ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow): Promise { const { handle } = ctx.steps["prepare-shallow-worktree"] as PrepareResult; const adapter = deps.getAdapter(ctx.input.adapter); + const agentTimeout = runSettings(ctx.input.repo).agentTimeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS; const sessionName = sessionNameFor(ctx.input); const sessionToken = crypto.randomUUID(); const tag = `[recommender:${sessionName}]`; @@ -413,7 +459,7 @@ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow { expect(name("owner/repo")).not.toBe(name("owner/rep o")); }); }); + +describe("recommender workflow — daemon path (resolveRunSettings, #135 fix)", () => { + // The bug this guards: the daemon used to fire the recommender on a *second* + // ephemeral engine that never processed the job, so no `recommender` row was + // ever created. The daemon now registers ONE workflow on its long-lived engine + // and resolves per-repo settings via `resolveRunSettings`. This proves that + // path actually RUNS: it creates the recommender row and drives to completion. + test("runs on the engine via per-repo resolveRunSettings and creates the recommender row", async () => { + const h = makeHarness({ autoDispatch: true, wireTrigger: true }); + const resolverCalls: string[] = []; + const daemonDeps: RecommenderDeps = { + ...h.deps, + // The daemon omits the static settings and resolves them per-repo instead. + schemaPath: undefined, + config: undefined, + repoConfig: undefined, + resolveRunSettings: (repo) => { + resolverCalls.push(repo); + return { + schemaPath: "/abs/schemas/state-issue.v1.md", + config: { defaultAdapter: "claude", autoDispatch: true, prMode: "worktree" }, + repoConfig: REPO_CONFIG, + agentTimeoutMs: 2000, + }; + }, + }; + + const id = await runToEnd(daemonDeps); + + const row = getWorkflow(db, id)!; + expect(row.state).toBe("completed"); // it actually ran on the engine + expect(row.kind).toBe("recommender"); // the row the old dead-engine path never created + expect(resolverCalls).toContain(REPO); // per-repo resolver drove the run, not static deps + // auto_dispatch came from the resolved per-repo config → the trigger fired. + expect(h.triggered).toEqual([{ repo: REPO, stateIssue: STATE_ISSUE }]); + }); + + test("a clear wiring error when neither resolveRunSettings nor static settings are provided", async () => { + const h = makeHarness(); + const broken: RecommenderDeps = { + ...h.deps, + schemaPath: undefined, + config: undefined, + repoConfig: undefined, + // resolveRunSettings deliberately absent → the build-prompt guard throws. + }; + const id = await runToEnd(broken); + // The guard fails the run (and compensation rolls the worktree back) rather + // than silently producing a half-run — exactly the failure mode we're fixing. + expect(["failed", "compensated"]).toContain(getWorkflow(db, id)!.state); + }); +}); From 8434ad51d8bff1d42314daf34b7dba889f669a21 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Mon, 25 May 2026 13:20:21 -0400 Subject: [PATCH 2/3] fix(recommender): clamp per-repo agent timeout to a ceiling the step backstop covers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit on #150: the step's registration-time timeout read the static agentTimeoutMs (unset → 15min default in daemon mode) while awaitStop read the per-repo resolveRunSettings value — a repo configured above the default would trip the generic step timeout before its own specific Stop-await. Add MAX_AGENT_TIMEOUT_MS (30min) ceiling: clamp the per-repo awaitStop to it and size the step backstop to it, so the internal timeout always fires first. --- .../dispatcher/src/workflows/recommender.ts | 19 +++++++++++++++++-- .../test/recommender-workflow.test.ts | 13 +++++++------ 2 files changed, 24 insertions(+), 8 deletions(-) diff --git a/packages/dispatcher/src/workflows/recommender.ts b/packages/dispatcher/src/workflows/recommender.ts index e9dbdb21..8e292c8b 100644 --- a/packages/dispatcher/src/workflows/recommender.ts +++ b/packages/dispatcher/src/workflows/recommender.ts @@ -135,6 +135,13 @@ const DEFAULT_LAUNCH_TIMEOUT_MS = 90_000; // rewriting the schema-strict state issue without finishing. Operators tune it // per repo via `[recommender] agent_timeout_minutes`. const DEFAULT_AGENT_TIMEOUT_MS = 15 * 60 * 1000; +// Hard ceiling on the per-repo agent timeout. The step's bunqueue `timeout` is a +// registration-time backstop and can't see the per-repo `resolveRunSettings` +// value (daemon mode), so it's sized for THIS ceiling; the per-repo `awaitStop` +// is clamped to it. Without the clamp, a repo configured above the default would +// trip the step timeout (a generic error) before its own `awaitStop` (specific) +// fired. Tune-up is allowed up to here, not past it. +const MAX_AGENT_TIMEOUT_MS = 30 * 60 * 1000; /** * Deterministic, repo-namespaced session name for the recommender's dedicated @@ -399,7 +406,12 @@ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow): Promise { const { handle } = ctx.steps["prepare-shallow-worktree"] as PrepareResult; const adapter = deps.getAdapter(ctx.input.adapter); - const agentTimeout = runSettings(ctx.input.repo).agentTimeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS; + // Clamp the per-repo timeout to the ceiling the step backstop is sized for, + // so the internal (specific-error) Stop-await always fires before the step's. + const agentTimeout = Math.min( + runSettings(ctx.input.repo).agentTimeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS, + MAX_AGENT_TIMEOUT_MS, + ); const sessionName = sessionNameFor(ctx.input); const sessionToken = crypto.randomUUID(); const tag = `[recommender:${sessionName}]`; @@ -518,7 +530,10 @@ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow { - // The step `timeout` is the hard cap; assert it via the built workflow's - // step config rather than wall-clock. Defaults: 90s launch + 15min agent - // (bumped from 5min, which was too tight against a real repo). + test("spawn-recommender-agent's step backstop is sized for the per-repo ceiling", () => { + // The step `timeout` is a registration-time backstop. It can't see the + // per-repo `resolveRunSettings` value (daemon mode), so it's sized for the + // 30-min hard ceiling (`MAX_AGENT_TIMEOUT_MS`) that the per-repo `awaitStop` + // is clamped to — guaranteeing the internal (specific) timeout fires first. const h = makeHarness(); delete (h.deps as { agentTimeoutMs?: number }).agentTimeoutMs; delete (h.deps as { launchTimeoutMs?: number }).launchTimeoutMs; const def = stepDef(h.deps, "spawn-recommender-agent"); expect(def).toBeDefined(); - // launch (90s) + agent (15min) + 30s backstop, per the factory. - expect(def!.timeout).toBe(90_000 + 15 * 60 * 1000 + 30_000); + // launch (90s) + ceiling (30min) + 30s backstop. + expect(def!.timeout).toBe(90_000 + 30 * 60 * 1000 + 30_000); }); test("prepare-shallow-worktree registers a compensation handler", () => { From 720172b87e4a723dd699463eedf6864ee03036e1 Mon Sep 17 00:00:00 2001 From: Justin Walsh Date: Mon, 25 May 2026 14:02:27 -0400 Subject: [PATCH 3/3] fix(recommender): freeze per-run settings once, thread through later steps CodeRabbit on #150: resolveRunSettings was re-called in build/spawn/verify/ trigger, so a live config edit mid-run could mix schemaPath/config/repoConfig/ autoDispatch/agentTimeoutMs within one execution. Resolve once in build-prompt, return it on BuildPromptResult, and read it from ctx in the later steps. --- .../dispatcher/src/workflows/recommender.ts | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/packages/dispatcher/src/workflows/recommender.ts b/packages/dispatcher/src/workflows/recommender.ts index 8e292c8b..b36478d0 100644 --- a/packages/dispatcher/src/workflows/recommender.ts +++ b/packages/dispatcher/src/workflows/recommender.ts @@ -294,7 +294,14 @@ export function buildRecommenderContext(opts: { } type PrepareResult = { handle: WorktreeHandle }; -type BuildPromptResult = { priorBody: string; promptText: string }; +// `settings` is resolved ONCE here and threaded to the later steps (spawn, +// verify, trigger) via ctx — so a live config edit mid-run can't mix different +// schemaPath/config/repoConfig/agentTimeoutMs values within one execution. +type BuildPromptResult = { + priorBody: string; + promptText: string; + settings: RecommenderRunSettings; +}; type VerifyResult = { ok: boolean; errors: string[] }; /** @@ -400,16 +407,17 @@ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow): Promise { const { handle } = ctx.steps["prepare-shallow-worktree"] as PrepareResult; + const { settings } = ctx.steps["build-prompt"] as BuildPromptResult; const adapter = deps.getAdapter(ctx.input.adapter); // Clamp the per-repo timeout to the ceiling the step backstop is sized for, // so the internal (specific-error) Stop-await always fires before the step's. const agentTimeout = Math.min( - runSettings(ctx.input.repo).agentTimeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS, + settings.agentTimeoutMs ?? DEFAULT_AGENT_TIMEOUT_MS, MAX_AGENT_TIMEOUT_MS, ); const sessionName = sessionNameFor(ctx.input); @@ -471,7 +479,8 @@ export function createRecommenderWorkflow(deps: RecommenderDeps): Workflow): Promise { const verify = ctx.steps["verify-state-issue-parses"] as VerifyResult; - // Gate on a clean parse. Phase 7 is read-only: the runner leaves - // `triggerAutoDispatch` unwired, so nothing dispatches regardless of config. - if (!verify.ok || !runSettings(ctx.input.repo).config.autoDispatch || !deps.triggerAutoDispatch) - return; + const { settings } = ctx.steps["build-prompt"] as BuildPromptResult; + // Gate on a clean parse + the frozen per-run auto_dispatch setting. + if (!verify.ok || !settings.config.autoDispatch || !deps.triggerAutoDispatch) return; await deps.triggerAutoDispatch({ repo: ctx.input.repo, stateIssue: ctx.input.stateIssue }); }