feat(dispatcher): human-in-the-loop and review-driven resume flow

packages/adapters/claude/src/classify.ts

@@ -1,6 +1,11 @@
 import { existsSync, readFileSync } from "node:fs";
 import { join } from "node:path";
-import type { HookPayload, RateLimitDetection, StopClassification } from "@middle/core";
+import type {
+  BlockedSentinel,
+  HookPayload,
+  RateLimitDetection,
+  StopClassification,
+} from "@middle/core";
 
 const USAGE_LIMIT_RE = /You've hit your usage limit\. Resets at (.+?)\./;
 
@@ -27,7 +32,8 @@ export function classifyStop(opts: {
   const middleDir = join(opts.worktree, ".middle");
 
   if (opts.sentinelPresent) {
-    return { kind: "asked-question", sentinelPath: join(middleDir, "blocked.json") };
+    const sentinelPath = join(middleDir, "blocked.json");
+    return { kind: "asked-question", sentinelPath, sentinel: readBlockedSentinel(sentinelPath) };
   }
 
   const match = USAGE_LIMIT_RE.exec(readTail(opts.transcriptPath));
@@ -68,6 +74,27 @@ function readTail(path: string): string {
   }
 }
 
+/**
+ * Read and tolerantly parse the `.middle/blocked.json` question sentinel so the
+ * workflow can surface the agent's question (and any context) to the human —
+ * e.g. posted on the Epic when it parks on `asked-question`. Returns `null` when
+ * the file is missing, unreadable, not JSON, or carries no string `question`:
+ * the Stop is still classified `asked-question` (the sentinel's *presence* is
+ * the signal), the contents are just best-effort.
+ */
+function readBlockedSentinel(path: string): BlockedSentinel | null {
+  try {
+    const parsed = JSON.parse(readFileSync(path, "utf8")) as Record<string, unknown>;
+    if (typeof parsed.question !== "string" || parsed.question.length === 0) return null;
+    const context = typeof parsed.context === "string" ? parsed.context : undefined;
+    return context === undefined
+      ? { question: parsed.question }
+      : { question: parsed.question, context };
+  } catch {
+    return null;
+  }
+}
+
 function readFailedReason(path: string): string {
   try {
     const parsed = JSON.parse(readFileSync(path, "utf8")) as { reason?: unknown };

packages/adapters/claude/test/adapter.test.ts

@@ -171,8 +171,12 @@ function writeMiddleDir(): { cwd: string; middle: string; transcript: string } {
 }
 
 describe("classifyStop", () => {
-  test("sentinelPresent → asked-question, with the worktree-anchored blocked.json path", () => {
-    const { cwd, transcript } = writeMiddleDir();
+  test("sentinelPresent → asked-question, surfacing the blocked.json path + question/context", () => {
+    const { cwd, middle, transcript } = writeMiddleDir();
+    writeFileSync(
+      join(middle, "blocked.json"),
+      JSON.stringify({ question: "Use option A or B?", context: "Both pass typecheck." }),
+    );
     const result = claudeAdapter.classifyStop({
       payload: { cwd },
       transcriptPath: transcript,
@@ -182,6 +186,25 @@ describe("classifyStop", () => {
     expect(result.kind).toBe("asked-question");
     if (result.kind === "asked-question") {
       expect(result.sentinelPath).toBe(join(cwd, ".middle", "blocked.json"));
+      expect(result.sentinel).toEqual({
+        question: "Use option A or B?",
+        context: "Both pass typecheck.",
+      });
+    }
+  });
+
+  test("asked-question tolerates a malformed/contentless blocked.json (sentinel → null)", () => {
+    const { cwd, middle, transcript } = writeMiddleDir();
+    writeFileSync(join(middle, "blocked.json"), "{ not valid json");
+    const result = claudeAdapter.classifyStop({
+      payload: { cwd },
+      transcriptPath: transcript,
+      sentinelPresent: true,
+      worktree: cwd,
+    });
+    expect(result.kind).toBe("asked-question");
+    if (result.kind === "asked-question") {
+      expect(result.sentinel).toBeNull();
     }
   });
 

packages/core/src/adapter.ts

@@ -86,9 +86,21 @@ export type TranscriptState = {
   lastToolUse: string | null;
 };
 
+/**
+ * The contents of a `.middle/blocked.json` question sentinel: the question the
+ * agent needs answered to proceed, plus optional supporting context the human
+ * needs to answer it. The skill writes this when it parks on `asked-question`.
+ * Parsed tolerantly — a sentinel that is missing or malformed yields `null` on
+ * the classification rather than failing the Stop.
+ */
+export type BlockedSentinel = {
+  question: string;
+  context?: string;
+};
+
 export type StopClassification =
   | { kind: "done" } // agent marked the PR ready
-  | { kind: "asked-question"; sentinelPath: string }
+  | { kind: "asked-question"; sentinelPath: string; sentinel: BlockedSentinel | null }
   | { kind: "rate-limited"; resetAt: string /* ISO */ }
   | { kind: "bare-stop" } // stopped, no sentinel, not done
   | { kind: "failed"; reason: string };

packages/core/src/index.ts

@@ -25,6 +25,7 @@ export type {
   LaunchOpts,
   TranscriptState,
   StopClassification,
+  BlockedSentinel,
   RateLimitDetection,
 } from "./adapter.ts";
 

packages/dispatcher/src/db/migrations/002_waitfor_fired.sql

@@ -0,0 +1,9 @@
+-- 002_waitfor_fired.sql
+-- The GitHub poller (Phase 5) fires a workflow's resume signal once per park.
+-- `fired_at` records when a signal was fired so a subsequent poll pass does not
+-- re-fire the same wait before the workflow has resumed and consumed the row.
+-- A fresh park (next review round) deletes-and-reinserts the row, clearing it.
+
+ALTER TABLE waitfor_signals ADD COLUMN fired_at INTEGER;
+
+INSERT OR IGNORE INTO schema_version VALUES (2);

packages/dispatcher/src/dispatch.ts

@@ -168,6 +168,11 @@ export async function dispatchEpic(opts: DispatchEpicOptions): Promise<DispatchE
         resolveRepoPath: () => opts.repoPath,
         worktreeRoot: opts.worktreeRoot,
         dispatcherUrl: `http://127.0.0.1:${hookServer.port}`,
+        // Resume hand-off: a continuation round re-enters the same workflow on
+        // this engine (Phase 8 hosts parked executions on a long-lived engine).
+        enqueueContinuation: async (input) => {
+          await engine.start("implementation", input);
+        },
         planCommentReader: ghGitHub,
         agentLogin,
         // Positive done-signal (#80): a bare-stop only completes if the Epic

packages/dispatcher/src/main.ts

@@ -13,8 +13,11 @@ import { Engine } from "bunqueue/workflow";
 import { openAndMigrate } from "./db.ts";
 import { HookServer } from "./hook-server.ts";
 import { DbHookStore } from "./hook-store.ts";
+import { ghPollGateway } from "./poller-gateway.ts";
+import { startPoller } from "./poller-cron.ts";
 import { killSession, status } from "./tmux.ts";
 import { startWatchdog } from "./watchdog-cron.ts";
+import { RESUME_EVENT } from "./workflows/implementation.ts";
 
 /** Phase 2 adapter registry — only `claude` is implemented. */
 function getAdapter(name: string): AgentAdapter {
@@ -44,6 +47,18 @@ async function main(): Promise<void> {
     getAdapter,
   });
 
+  // GitHub poller: every 60s, for each parked workflow with an armed wait, fire
+  // its resume signal when the unblocking event appears (a human reply, or a PR
+  // review verdict). `fireSignal` delivers it to the engine that hosts the
+  // parked execution. NOTE: routing dispatches through this long-lived engine
+  // (so parked executions live here to be resumed) is the Phase 8 auto-dispatch
+  // integration; the poller + signal seam are in place ahead of it.
+  const stopPoller = await startPoller({
+    db,
+    github: ghPollGateway,
+    fireSignal: (workflowId, payload) => engine.signal(workflowId, RESUME_EVENT, payload),
+  });
+
   console.log(
     `middle dispatcher up — hooks on :${hookServer.port}, db ${config.global.dbPath}`,
   );
@@ -59,6 +74,11 @@ async function main(): Promise<void> {
     } catch (error) {
       console.error(`shutdown: stopWatchdog failed — ${(error as Error).message}`);
     }
+    try {
+      await stopPoller();
+    } catch (error) {
+      console.error(`shutdown: stopPoller failed — ${(error as Error).message}`);
+    }
     try {
       hookServer.stop();
     } catch (error) {

packages/dispatcher/src/poller-cron.ts

@@ -0,0 +1,34 @@
+import { Bunqueue } from "bunqueue/client";
+import { runPoller, type PollerDeps } from "./poller.ts";
+
+/**
+ * How often the poller checks GitHub for resume triggers. Slower than the
+ * watchdog (30s) — a human reply or a review verdict is not latency-sensitive,
+ * and a gentler cadence is kinder to GitHub rate limits.
+ */
+export const POLLER_INTERVAL_MS = 60_000;
+
+/**
+ * Stand up the GitHub poller as a bunqueue cron: every {@link POLLER_INTERVAL_MS}
+ * it runs one {@link runPoller} pass over parked workflows with an armed wait,
+ * firing the resume signal when the unblocking event appears. Returns a stop
+ * function that tears the cron down. The pass is resilient on its own (per-
+ * workflow failures are isolated); this wrapper guards the whole pass too so a
+ * thrown pass never crashes the cron worker.
+ */
+export async function startPoller(deps: PollerDeps): Promise<() => Promise<void>> {
+  const queue = new Bunqueue("middle-poller", {
+    embedded: true,
+    processor: async () => {
+      try {
+        await runPoller(deps);
+      } catch (error) {
+        console.error(`[poller] pass failed: ${(error as Error).message}`);
+      }
+    },
+  });
+  await queue.every("poller-tick", POLLER_INTERVAL_MS);
+  return async () => {
+    await queue.close(true);
+  };
+}

packages/dispatcher/src/poller-gateway.ts

@@ -0,0 +1,119 @@
+import type { GitHubPollGateway, IssueComment, PrReview, PrSnapshot } from "./poller.ts";
+
+/**
+ * The production {@link GitHubPollGateway} — reads issue comments and PR review
+ * state through the `gh` CLI. The poller's logic is unit-tested against an
+ * injected stub gateway; this is the thin subprocess glue that backs it in the
+ * dispatcher. Read-only: the poller never writes to GitHub.
+ */
+
+async function gh(argv: string[]): Promise<string> {
+  const proc = Bun.spawn(["gh", ...argv], { stdout: "pipe", stderr: "pipe", stdin: "ignore" });
+  const [stdout, stderr] = await Promise.all([
+    new Response(proc.stdout).text(),
+    new Response(proc.stderr).text(),
+  ]);
+  if ((await proc.exited) !== 0) {
+    throw new Error(`gh ${argv.join(" ")} failed: ${stderr.trim()}`);
+  }
+  return stdout;
+}
+
+function isBotLogin(login: string, type: string | undefined): boolean {
+  return type === "Bot" || login.endsWith("[bot]");
+}
+
+export const ghPollGateway: GitHubPollGateway = {
+  async listIssueComments(repo: string, issueNumber: number): Promise<IssueComment[]> {
+    // `--slurp` wraps the per-page arrays into one outer array; `gh` without it
+    // emits one JSON array *per page*, which `JSON.parse` chokes on past page 1.
+    const out = await gh([
+      "api",
+      "--paginate",
+      "--slurp",
+      `repos/${repo}/issues/${issueNumber}/comments`,
+    ]);
+    const rows = (
+      JSON.parse(out) as Array<
+        Array<{
+          id: number;
+          body: string;
+          created_at: string;
+          user: { login: string; type?: string } | null;
+        }>
+      >
+    ).flat();
+    return rows.map((r) => ({
+      id: r.id,
+      body: r.body ?? "",
+      createdAt: Date.parse(r.created_at),
+      authorLogin: r.user?.login ?? "",
+      authorIsBot: isBotLogin(r.user?.login ?? "", r.user?.type),
+    }));
+  },
+
+  async findPrForEpic(repo: string, epicNumber: number): Promise<PrSnapshot | null> {
+    // The Epic's one PR closes the Epic — find the open PR referencing it.
+    const listOut = await gh([
+      "pr",
+      "list",
+      "--repo",
+      repo,
+      "--state",
+      "open",
+      "--search",
+      `in:body Closes #${epicNumber}`,
+      "--json",
+      "number",
+    ]);
+    const prs = JSON.parse(listOut) as Array<{ number: number }>;
+    const prNumber = prs[0]?.number;
+    if (prNumber === undefined) return null;
+
+    const viewOut = await gh([
+      "pr",
+      "view",
+      String(prNumber),
+      "--repo",
+      repo,
+      "--json",
+      "reviewDecision,labels",
+    ]);
+    const view = JSON.parse(viewOut) as {
+      reviewDecision: string | null;
+      labels: Array<{ name: string }>;
+    };
+
+    const reviewsOut = await gh([
+      "api",
+      "--paginate",
+      "--slurp",
+      `repos/${repo}/pulls/${prNumber}/reviews`,
+    ]);
+    const reviewRows = (
+      JSON.parse(reviewsOut) as Array<
+        Array<{
+          id: number;
+          state: string;
+          body: string;
+          submitted_at: string | null;
+          user: { login: string } | null;
+        }>
+      >
+    ).flat();
+    const reviews: PrReview[] = reviewRows.map((r) => ({
+      id: r.id,
+      state: r.state,
+      body: r.body ?? "",
+      submittedAt: r.submitted_at ? Date.parse(r.submitted_at) : 0,
+      authorLogin: r.user?.login ?? "",
+    }));
+
+    return {
+      number: prNumber,
+      reviewDecision: view.reviewDecision ?? null,
+      reviews,
+      labels: view.labels.map((l) => l.name),
+    };
+  },
+};