fix(recommender): run on the daemon's engine, not a standalone second one

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<boolean>;
+  /** 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<boolean> {
+  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<boolean>,
+): Promise<boolean> {
+  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 <repo>` — 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 <repo>` — 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<typeof loadConfig>;
   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<ReturnType<typeof dispatchRecommender>>;
+  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;
 }

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 {