feat(pr): node-agnostic PR reconcile + hold-release firing (U4)

Adds PrReconciler — a per-repo, self-owned polling loop (started from the
runtime layer in project-engine.ts, NOT the scheduler) that ETag-probes
GitHub, deep-fetches on change, persists mirror state, clears unverified
on first reconcile, and fires releaseHeldTaskByEvent(github:pr-<event>)
for transitions (changes-requested/approved/conflict/conflict-cleared/
merged/closed). Drops terminal entities; persists an audit event on error.
GitHub ops injected via PrReconcileGithubOps at the 3 CLI sites; engine
never imports the dashboard client. scheduler.ts stays PR-free (R20),
pinned by a regression test. 8 tests.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

packages/cli/src/commands/daemon.ts

@@ -45,6 +45,7 @@ import {
   createGroupPrCallback,
   syncGroupPrCallback,
   createPrNodeGithubOps,
+  createPrReconcileGithubOps,
 } from "./task-lifecycle.js";
 import { promptForPort } from "./port-prompt.js";
 import { createReadOnlyProviderSettingsView } from "./provider-settings.js";
@@ -340,6 +341,7 @@ export async function runDaemon(opts: DaemonOptions = {}) {
     createGroupPr: createGroupPrCallback(githubClient),
     syncGroupPr: syncGroupPrCallback(githubClient),
     prNodeGithubOps: createPrNodeGithubOps(githubClient),
+    prReconcileGithubOps: createPrReconcileGithubOps(githubClient),
     getTaskMergeBlocker,
     onInsightRunProcessed: (s: unknown, r: unknown) => onMemoryInsightRunProcessed(s as ScheduledTask, r as AutomationRunResult),
   });

packages/cli/src/commands/dashboard.ts

@@ -55,6 +55,7 @@ import {
   createGroupPrCallback,
   syncGroupPrCallback,
   createPrNodeGithubOps,
+  createPrReconcileGithubOps,
 } from "./task-lifecycle.js";
 import { promptForPort } from "./port-prompt.js";
 import { ensureCwdProjectRegistered } from "./ensure-project-registered.js";
@@ -1616,6 +1617,7 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?:
       createGroupPr: createGroupPrCallback(githubClient),
       syncGroupPr: syncGroupPrCallback(githubClient),
       prNodeGithubOps: createPrNodeGithubOps(githubClient),
+      prReconcileGithubOps: createPrReconcileGithubOps(githubClient),
       getTaskMergeBlocker,
     });
 

packages/cli/src/commands/serve.ts

@@ -45,6 +45,7 @@ import {
   createGroupPrCallback,
   syncGroupPrCallback,
   createPrNodeGithubOps,
+  createPrReconcileGithubOps,
 } from "./task-lifecycle.js";
 import { promptForPort } from "./port-prompt.js";
 import { createReadOnlyProviderSettingsView } from "./provider-settings.js";
@@ -366,6 +367,7 @@ export async function runServe(
     createGroupPr: createGroupPrCallback(githubClient),
     syncGroupPr: syncGroupPrCallback(githubClient),
     prNodeGithubOps: createPrNodeGithubOps(githubClient),
+    prReconcileGithubOps: createPrReconcileGithubOps(githubClient),
     getTaskMergeBlocker,
     onInsightRunProcessed: (s: unknown, r: unknown) => onMemoryInsightRunProcessed(s as ScheduledTask, r as AutomationRunResult),
   });

packages/cli/src/commands/task-lifecycle.ts

@@ -27,7 +27,14 @@ import type { TaskStore } from "@fusion/core";
 import { resolveTaskMergeTarget, getCurrentRepo, isBranchGroupMemberLanded } from "@fusion/core";
 import type { Settings, TaskDetail, PrInfo, MergeResult, BranchGroup, BranchGroupPrState, Task } from "@fusion/core";
 import { activeSessionRegistry, resolveIntegrationBranch } from "@fusion/engine";
-import type { CreateGroupPrFn, SyncGroupPrFn, WorktreePool, PrNodeGithubOps } from "@fusion/engine";
+import type {
+  CreateGroupPrFn,
+  SyncGroupPrFn,
+  WorktreePool,
+  PrNodeGithubOps,
+  PrReconcileGithubOps,
+  PrReconcileFetchResult,
+} from "@fusion/engine";
 
 /**
  * Minimal interface for GitHub operations needed by the PR merge workflow.
@@ -47,6 +54,13 @@ interface GitHubOperations {
   getPrStatus(owner: string, repo: string, number: number): Promise<PrInfo>;
   updatePr(params: { owner?: string; repo?: string; number: number; title?: string; body?: string }): Promise<PrInfo>;
   closePr(params: { number: number }): Promise<PrInfo>;
+  /** ETag-conditional change probe (U2/U4); 304 ⇒ unchanged, rate-limit-free. */
+  probePrChanged(
+    owner: string | undefined,
+    repo: string | undefined,
+    number: number,
+    etag?: string,
+  ): Promise<{ changed: boolean; etag?: string }>;
 }
 
 /**
@@ -387,6 +401,67 @@ export function createPrNodeGithubOps(
   };
 }
 
+/**
+ * Parse the entity's `owner/repo` repo slug into its components, tolerating an
+ * empty/single-segment value (returns undefined owner/repo so the client falls
+ * back to its configured repo).
+ */
+function splitRepoSlug(repo: string): { owner: string | undefined; name: string | undefined } {
+  const [owner, name] = repo.split("/");
+  return { owner: owner || undefined, name: name || undefined };
+}
+
+/** Map a GitHub `PrStatus` to the reconcile fetch result's coarse PR state. */
+function mapPrStatusToFetchState(status: PrInfo["status"]): "open" | "merged" | "closed" {
+  if (status === "merged") return "merged";
+  if (status === "closed") return "closed";
+  // "open" and "draft" both reconcile as open.
+  return "open";
+}
+
+/**
+ * Build the `prReconcileGithubOps` engine callbacks (U4) backing the
+ * node-agnostic {@link PrReconciler}. Closes over the dashboard `GitHubClient`
+ * so the engine never imports it (FN-3049), exactly like
+ * {@link createPrNodeGithubOps}. Wired at the same three CLI composition sites.
+ *
+ * - probe: ETag-conditional change probe (304 ⇒ unchanged ⇒ skip deep-fetch).
+ * - fetchPrState: deep-fetch the GitHub-corroborated mirror. A 404 (PR not
+ *   found) maps to `{ exists: false }` so the reconcile clears fictional
+ *   unverified entities (R19).
+ */
+export function createPrReconcileGithubOps(
+  github: Pick<GitHubOperations, "probePrChanged" | "getPrStatus">,
+): PrReconcileGithubOps {
+  return {
+    probe: (repo, prNumber, etag) => {
+      const { owner, name } = splitRepoSlug(repo);
+      return github.probePrChanged(owner, name, prNumber, etag);
+    },
+    fetchPrState: async (repo, prNumber): Promise<PrReconcileFetchResult> => {
+      const { owner, name } = splitRepoSlug(repo);
+      let info: PrInfo;
+      try {
+        info = await github.getPrStatus(owner ?? "", name ?? "", prNumber);
+      } catch (err) {
+        // A 404 / "not found" means there is no PR behind this entity.
+        const message = err instanceof Error ? err.message : String(err);
+        if (/not found|404/i.test(message)) return { exists: false };
+        throw err;
+      }
+      return {
+        exists: true,
+        prState: mapPrStatusToFetchState(info.status),
+        prNumber: info.number,
+        prUrl: info.url,
+        mergeable: info.mergeable,
+        checksRollup: info.checkRollup,
+        reviewDecision: info.lastReviewDecision ?? null,
+      };
+    },
+  };
+}
+
 async function hasCommitsRelativeToBranch(cwd: string, branch: string, baseBranch: string): Promise<boolean> {
   try {
     const { stdout } = await execAsync(`git rev-list --count "${baseBranch}..${branch}"`, { cwd, timeout: 30_000 });

packages/engine/src/__tests__/pr-reconcile.test.ts

@@ -0,0 +1,262 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { mkdtempSync, readFileSync } from "node:fs";
+import { rm } from "node:fs/promises";
+import { join } from "node:path";
+import { tmpdir } from "node:os";
+import { fileURLToPath } from "node:url";
+import { TaskStore } from "@fusion/core";
+import type { PrEntity } from "@fusion/core";
+import {
+  PrReconciler,
+  deriveTransitions,
+  type PrReconcileFetchResult,
+  type PrReconcileGithubOps,
+} from "../pr-reconcile.js";
+
+function makeTmpDir(prefix: string): string {
+  return mkdtempSync(join(tmpdir(), prefix));
+}
+
+/** A fake GitHub ops with scriptable probe + deep-fetch responses, recording calls. */
+function makeFakeOps(): {
+  ops: PrReconcileGithubOps;
+  probeCalls: Array<{ repo: string; prNumber: number; etag?: string }>;
+  fetchCalls: Array<{ repo: string; prNumber: number }>;
+  setProbe: (changed: boolean, etag?: string) => void;
+  setFetch: (result: PrReconcileFetchResult | (() => Promise<PrReconcileFetchResult>)) => void;
+  failFetch: (message: string) => void;
+} {
+  const probeCalls: Array<{ repo: string; prNumber: number; etag?: string }> = [];
+  const fetchCalls: Array<{ repo: string; prNumber: number }> = [];
+  let probeResult: { changed: boolean; etag?: string } = { changed: true, etag: "etag-1" };
+  let fetchImpl: () => Promise<PrReconcileFetchResult> = async () => ({ exists: true, prState: "open" });
+
+  return {
+    probeCalls,
+    fetchCalls,
+    setProbe: (changed, etag) => {
+      probeResult = { changed, etag };
+    },
+    setFetch: (result) => {
+      fetchImpl = typeof result === "function" ? result : async () => result;
+    },
+    failFetch: (message) => {
+      fetchImpl = async () => {
+        throw new Error(message);
+      };
+    },
+    ops: {
+      probe: async (repo, prNumber, etag) => {
+        probeCalls.push({ repo, prNumber, etag });
+        return probeResult;
+      },
+      fetchPrState: async (repo, prNumber) => {
+        fetchCalls.push({ repo, prNumber });
+        return fetchImpl();
+      },
+    },
+  };
+}
+
+describe("PrReconciler (U4 — node-agnostic GitHub reconcile)", () => {
+  let rootDir: string;
+  let globalDir: string;
+  let store: TaskStore;
+  let release: ReturnType<typeof vi.fn>;
+
+  beforeEach(async () => {
+    rootDir = makeTmpDir("kb-engine-pr-reconcile-");
+    globalDir = makeTmpDir("kb-engine-pr-reconcile-global-");
+    store = new TaskStore(rootDir, globalDir, { inMemoryDb: true });
+    await store.init();
+    release = vi.fn(async () => ({ released: true }));
+  });
+
+  afterEach(async () => {
+    store.close();
+    await rm(rootDir, { recursive: true, force: true });
+    await rm(globalDir, { recursive: true, force: true });
+  });
+
+  function seedEntity(overrides: Partial<PrEntity> & { sourceId: string; prNumber?: number }): PrEntity {
+    const entity = store.ensurePrEntityForSource({
+      sourceType: overrides.sourceType ?? "task",
+      sourceId: overrides.sourceId,
+      repo: overrides.repo ?? "owner/repo",
+      headBranch: overrides.headBranch ?? `fusion/${overrides.sourceId}`,
+      state: overrides.state ?? "open",
+      prNumber: overrides.prNumber,
+      unverified: overrides.unverified ?? false,
+    });
+    // Apply mirror fields that ensure-create does not take.
+    if (
+      overrides.reviewDecision !== undefined ||
+      overrides.mergeable !== undefined ||
+      overrides.prUrl !== undefined ||
+      overrides.state !== undefined
+    ) {
+      return store.updatePrEntity(entity.id, {
+        state: overrides.state,
+        reviewDecision: overrides.reviewDecision,
+        mergeable: overrides.mergeable ?? undefined,
+        prUrl: overrides.prUrl ?? undefined,
+      });
+    }
+    return entity;
+  }
+
+  function makeReconciler(ops: PrReconcileGithubOps): PrReconciler {
+    return new PrReconciler({
+      store,
+      ops,
+      releaseByEvent: release as unknown as (taskId: string, tag: string) => Promise<unknown>,
+      // Tiny intervals + a no-op timer keep the loop off the test clock.
+      setTimer: () => 0 as unknown as ReturnType<typeof setTimeout>,
+      clearTimer: () => {},
+    });
+  }
+
+  it("AE4: PR merged on GitHub → fires github:pr-merged + entity becomes terminal (drops from poll)", async () => {
+    seedEntity({ sourceId: "TASK-1", prNumber: 10, state: "open" });
+    const fake = makeFakeOps();
+    fake.setFetch({ exists: true, prState: "merged", prNumber: 10 });
+    const reconciler = makeReconciler(fake.ops);
+
+    const fired = await reconciler.reconcileRepoOnce("owner/repo");
+
+    expect(fired.map((t) => t.event)).toEqual(["merged"]);
+    expect(release).toHaveBeenCalledWith("TASK-1", "github:pr-merged");
+
+    const entity = store.getActivePrEntityBySource("task", "TASK-1");
+    expect(entity).toBeNull(); // now merged ⇒ not active ⇒ out of the poll set.
+    expect(store.listActivePrEntities()).toHaveLength(0);
+  });
+
+  it("changes-requested on GitHub → fires github:pr-changes-requested", async () => {
+    seedEntity({ sourceId: "TASK-2", prNumber: 11, state: "open", reviewDecision: null });
+    const fake = makeFakeOps();
+    fake.setFetch({ exists: true, prState: "open", prNumber: 11, reviewDecision: "CHANGES_REQUESTED" });
+    const reconciler = makeReconciler(fake.ops);
+
+    const fired = await reconciler.reconcileRepoOnce("owner/repo");
+
+    expect(fired.map((t) => t.event)).toEqual(["changes-requested"]);
+    expect(release).toHaveBeenCalledWith("TASK-2", "github:pr-changes-requested");
+    expect(store.getActivePrEntityBySource("task", "TASK-2")?.reviewDecision).toBe("CHANGES_REQUESTED");
+  });
+
+  it("unverified entity with no real PR → cleared on first poll, NOT advanced on stale state (R19)", async () => {
+    const seeded = seedEntity({ sourceId: "TASK-3", prNumber: 999, state: "open", unverified: true });
+    const fake = makeFakeOps();
+    fake.setFetch({ exists: false }); // no PR behind it.
+    const reconciler = makeReconciler(fake.ops);
+
+    const fired = await reconciler.reconcileRepoOnce("owner/repo");
+
+    expect(fired).toHaveLength(0);
+    expect(release).not.toHaveBeenCalled(); // never advanced on stale state.
+    expect(store.getActivePrEntityBySource("task", "TASK-3")).toBeNull(); // cleared (closed).
+    expect(store.getPrEntity(seeded.id)?.state).toBe("closed");
+    expect(store.getPrEntity(seeded.id)?.unverified).toBe(false);
+
+    const audit = store.getRunAuditEvents({ agentId: "pr-reconcile" });
+    expect(audit.some((e) => e.mutationType === "pr-reconcile:cleared-fiction")).toBe(true);
+  });
+
+  it("N entities in one repo → one batched probe PER ENTITY but a single tick (rate-limit batching)", async () => {
+    seedEntity({ sourceId: "TASK-A", prNumber: 21, state: "open" });
+    seedEntity({ sourceId: "TASK-B", prNumber: 22, state: "open" });
+    seedEntity({ sourceId: "TASK-C", prNumber: 23, state: "open" });
+    const fake = makeFakeOps();
+    fake.setProbe(false); // 304 unchanged for all.
+    const reconciler = makeReconciler(fake.ops);
+
+    await reconciler.reconcileRepoOnce("owner/repo");
+
+    // All three probed in the single tick for the one repo; no deep-fetch (304).
+    expect(fake.probeCalls).toHaveLength(3);
+    expect(fake.fetchCalls).toHaveLength(0);
+    // The repo grouping ran once for the whole repo (single tick, not per-entity ticks).
+    expect(reconciler.getTrackedRepos()).toEqual(["owner/repo"]);
+  });
+
+  it("probe 304 → no deep-fetch, no writes", async () => {
+    const seeded = seedEntity({ sourceId: "TASK-4", prNumber: 30, state: "open", reviewDecision: null });
+    const beforeUpdatedAt = seeded.updatedAt;
+    const fake = makeFakeOps();
+    fake.setProbe(false);
+    const reconciler = makeReconciler(fake.ops);
+
+    const fired = await reconciler.reconcileRepoOnce("owner/repo");
+
+    expect(fired).toHaveLength(0);
+    expect(fake.fetchCalls).toHaveLength(0);
+    expect(release).not.toHaveBeenCalled();
+    expect(store.getActivePrEntityBySource("task", "TASK-4")?.updatedAt).toBe(beforeUpdatedAt);
+  });
+
+  it("deep-fetch error → persisted audit event + poller survives (backoff)", async () => {
+    seedEntity({ sourceId: "TASK-5", prNumber: 40, state: "open" });
+    const fake = makeFakeOps();
+    fake.failFetch("boom: github 500");
+    const reconciler = makeReconciler(fake.ops);
+
+    // Must not throw — the loop records the error and continues.
+    await expect(reconciler.reconcileRepoOnce("owner/repo")).resolves.toEqual([]);
+
+    const audit = store.getRunAuditEvents({ agentId: "pr-reconcile" });
+    const errEvent = audit.find((e) => e.mutationType === "pr-reconcile:error");
+    expect(errEvent).toBeTruthy();
+    expect(JSON.stringify(errEvent?.metadata)).toContain("boom: github 500");
+
+    // Entity remains active (poller survives, did not corrupt state).
+    expect(store.getActivePrEntityBySource("task", "TASK-5")).toBeTruthy();
+  });
+
+  it("deriveTransitions: terminal short-circuits, review + conflict are independent", () => {
+    const base = {
+      id: "x",
+      sourceType: "task",
+      sourceId: "t",
+      repo: "owner/repo",
+      headBranch: "h",
+      state: "open",
+      autoMerge: false,
+      unverified: false,
+      responseRounds: 0,
+      createdAt: 0,
+      updatedAt: 0,
+    } as PrEntity;
+
+    expect(deriveTransitions(base, { exists: true, prState: "merged" }).map((t) => t.event)).toEqual(["merged"]);
+    expect(deriveTransitions(base, { exists: true, prState: "closed" }).map((t) => t.event)).toEqual(["closed"]);
+
+    // Both a review change and a conflict can fire on one pass.
+    const both = deriveTransitions(base, {
+      exists: true,
+      prState: "open",
+      reviewDecision: "APPROVED",
+      mergeable: "conflicting",
+    });
+    expect(both.map((t) => t.event).sort()).toEqual(["approved", "conflict"]);
+
+    // conflict-cleared only when transitioning FROM conflicting → clean.
+    const cleared = deriveTransitions({ ...base, mergeable: "conflicting" }, {
+      exists: true,
+      prState: "open",
+      mergeable: "clean",
+    });
+    expect(cleared.map((t) => t.event)).toEqual(["conflict-cleared"]);
+
+    // UNKNOWN mergeable never maps to conflict.
+    expect(
+      deriveTransitions(base, { exists: true, prState: "open", mergeable: "unknown" }).map((t) => t.event),
+    ).toEqual([]);
+  });
+
+  it("REGRESSION (R20): scheduler.ts contains zero PR symbols", () => {
+    const schedulerPath = fileURLToPath(new URL("../scheduler.ts", import.meta.url));
+    const source = readFileSync(schedulerPath, "utf8");
+    expect(source).not.toMatch(/pr-create|pr-respond|pull_request|PrEntity|pr-reconcile|PrReconciler/);
+  });
+});

packages/engine/src/index.ts

@@ -60,6 +60,18 @@ export {
   type PrRespondCallInput,
   type PrRespondCallResult,
 } from "./pr-nodes.js";
+export {
+  PrReconciler,
+  deriveTransitions,
+  type PrReconcileGithubOps,
+  type PrReconcileFetchResult,
+  type PrReconcileStore,
+  type PrReconcilerOptions,
+  type PrReconcileIntervals,
+  type PrReconcileTransition,
+  type PrReleaseByEventFn,
+  type ResolveGroupReleaseTaskFn,
+} from "./pr-reconcile.js";
 export {
   WorkflowGraphTaskRunner,
   type WorkflowGraphRunDisposition,

packages/engine/src/logger.ts

@@ -82,6 +82,7 @@ export const reviewerLog = createLogger("reviewer");
 
 /** Logger for the PR monitor subsystem. */
 export const prMonitorLog = createLogger("pr-monitor");
+export const prReconcileLog = createLogger("pr-reconcile");
 
 /** Logger for the project runtime subsystem. */
 export const runtimeLog = createLogger("runtime");

packages/engine/src/pr-reconcile.ts

@@ -0,0 +1,521 @@
+/**
+ * Node-agnostic GitHub reconcile (PR-lifecycle-as-workflow-nodes, U4).
+ *
+ * This is the per-repo, node-kind-agnostic poller that corroborates each active
+ * {@link PrEntity} against GitHub and fires the *generic* external-event hold
+ * releases that advance whatever card is parked in a PR-await hold. It is the
+ * load-bearing R20 invariant made concrete: the scheduler contains ZERO PR
+ * knowledge — this reconciler lives in the PR feature's own module and is
+ * started/stopped from the RUNTIME layer (project-engine), never from
+ * `scheduler.ts`.
+ *
+ * Shape mirrors {@link PrMonitor} (adaptive interval map, exponential backoff,
+ * injected GitHub ops, start/stop/stopAll) but operates per-repo (not per-task)
+ * so N entities in one repo cost one ETag probe per tick, not N (rate-limit
+ * safety, R17).
+ *
+ * Flow per repo per tick:
+ *   1. ETag probe (304 is rate-limit-free) — if unchanged, no deep-fetch / no
+ *      writes for that entity.
+ *   2. On change, deep-fetch the mirror state.
+ *   3. Persist the mirror (state, prNumber/prUrl/headOid, mergeable, checks,
+ *      reviewDecision) via {@link TaskStore.updatePrEntity}; clear `unverified`
+ *      on the first successful reconcile.
+ *   4. If an unverified entity has NO real PR on GitHub, transition it to
+ *      `closed` (fiction cleared) and DO NOT advance it on stale state (R19).
+ *   5. For each detected transition fire
+ *      `releaseHeldTaskByEvent(store, taskId, "github:pr-<event>")` — the
+ *      generic sweep moves the card; it never learns PR semantics.
+ *   6. Drop terminal (merged/closed) entities from the poll set (R18).
+ *   7. Every caught error persists an audit event (silent catch-and-continue is
+ *      the documented stall mode) and the repo backs off; the poller survives.
+ */
+
+import type { PrEntity, PrConflictState, PrChecksRollup, PrReviewDecision } from "@fusion/core";
+import { isPrEntityActive } from "@fusion/core";
+import { prReconcileLog } from "./logger.js";
+import { releaseHeldTaskByEvent } from "./hold-release.js";
+
+// ── Injected GitHub ops (node-agnostic; engine never imports the dashboard) ────
+
+/** Result of a deep-fetch of a single PR's GitHub-corroborated mirror state. */
+export interface PrReconcileFetchResult {
+  /**
+   * Whether the PR actually exists on GitHub. `false` means there is no PR
+   * behind this entity (the fiction case for unverified imported entities, R19).
+   */
+  exists: boolean;
+  /** Open / merged / closed (draft maps to open for reconcile purposes). */
+  prState?: "open" | "merged" | "closed";
+  prNumber?: number;
+  prUrl?: string;
+  headOid?: string;
+  mergeable?: PrConflictState;
+  checksRollup?: PrChecksRollup;
+  reviewDecision?: PrReviewDecision;
+}
+
+/**
+ * The CLI-injected GitHub callbacks backing the reconcile. Mirrors
+ * {@link PrNodeGithubOps}: only plain callbacks that close over the dashboard
+ * `GitHubClient`; the engine receives no client reference. Wired alongside
+ * `prNodeGithubOps` at the three CLI composition sites.
+ */
+export interface PrReconcileGithubOps {
+  /**
+   * ETag-conditional change probe. `changed:false` (HTTP 304) is rate-limit-free
+   * and means the caller may skip the deep-fetch. Returns a fresh `etag` to
+   * store for the next probe.
+   */
+  probe(repo: string, prNumber: number, etag?: string): Promise<{ changed: boolean; etag?: string }>;
+  /** Deep-fetch the GitHub-corroborated mirror state for a PR. */
+  fetchPrState(repo: string, prNumber: number): Promise<PrReconcileFetchResult>;
+}
+
+// ── Store + release seams (kept structural so the reconciler is unit-testable) ─
+
+/** The slice of the task store the reconciler reads/writes. */
+export interface PrReconcileStore {
+  listActivePrEntities(): PrEntity[];
+  getPrEntity(id: string): PrEntity | null;
+  updatePrEntity(id: string, patch: import("@fusion/core").PrEntityUpdate): PrEntity;
+  recordRunAuditEvent?: (input: import("@fusion/core").RunAuditEventInput) => unknown;
+}
+
+/**
+ * Release function injected for testability. Defaults to the real
+ * {@link releaseHeldTaskByEvent}, which only acts on `external-event` holds (a
+ * no-op otherwise, so firing it for a task that is not parked in an await hold
+ * is harmless). Tests inject a spy to assert the transition→event-tag mapping
+ * without driving a full workflow graph.
+ */
+export type PrReleaseByEventFn = (taskId: string, eventTag: string) => Promise<unknown>;
+
+/**
+ * Resolve a branch-group entity to a representative task id to release, or
+ * `null` to skip release for that group. v1 has no group→task resolver wired
+ * (documented choice): groups persist their reconciled mirror state but do not
+ * fire hold releases until a resolver is injected.
+ */
+export type ResolveGroupReleaseTaskFn = (entity: PrEntity) => string | null;
+
+export interface PrReconcilerOptions {
+  store: PrReconcileStore;
+  ops: PrReconcileGithubOps;
+  /** Defaults to {@link releaseHeldTaskByEvent} bound to the store. */
+  releaseByEvent?: PrReleaseByEventFn;
+  /** Branch-group → representative task resolver (v1: omitted ⇒ skip groups). */
+  resolveGroupReleaseTask?: ResolveGroupReleaseTaskFn;
+  /** Override cadence/backoff knobs (tests use tiny intervals). */
+  intervals?: Partial<PrReconcileIntervals>;
+  /** Injected clock for the next-tick scheduler (defaults to setTimeout). */
+  setTimer?: (fn: () => void, ms: number) => ReturnType<typeof setTimeout>;
+  clearTimer?: (handle: ReturnType<typeof setTimeout>) => void;
+}
+
+export interface PrReconcileIntervals {
+  /** ~15-30s when there is recent activity. */
+  active: number;
+  /** ~60-120s when idle. */
+  idle: number;
+  /** 5min when dormant (no changes for a while). */
+  dormant: number;
+  /** Max backoff cap. */
+  maxBackoff: number;
+  /** Errors before a repo is considered failing (still survives, just backs off). */
+  maxConsecutiveErrors: number;
+}
+
+const DEFAULT_INTERVALS: PrReconcileIntervals = {
+  active: 20 * 1000,
+  idle: 90 * 1000,
+  dormant: 5 * 60 * 1000,
+  maxBackoff: 15 * 60 * 1000,
+  maxConsecutiveErrors: 5,
+};
+
+// ── Transition → event-tag mapping (the load-bearing semantics) ────────────────
+
+/** A detected GitHub-state transition for one entity, with its release event tag. */
+export interface PrReconcileTransition {
+  event:
+    | "merged"
+    | "closed"
+    | "changes-requested"
+    | "approved"
+    | "conflict"
+    | "conflict-cleared";
+  /** The hold-release event tag: `github:pr-<event>`. */
+  tag: string;
+  /** Whether this transition makes the entity terminal (drop from poll). */
+  terminal: boolean;
+}
+
+/**
+ * Derive the list of transitions between a previously-persisted entity mirror
+ * and a freshly-fetched GitHub state. Pure + exported for unit testing.
+ *
+ * Ordering: terminal states (merged/closed) short-circuit — once merged/closed,
+ * review/conflict transitions are irrelevant. Otherwise review-decision and
+ * mergeability transitions are independent and may both fire.
+ */
+export function deriveTransitions(prev: PrEntity, next: PrReconcileFetchResult): PrReconcileTransition[] {
+  const tag = (event: PrReconcileTransition["event"]): string => `github:pr-${event}`;
+
+  if (next.prState === "merged") {
+    return [{ event: "merged", tag: tag("merged"), terminal: true }];
+  }
+  if (next.prState === "closed") {
+    return [{ event: "closed", tag: tag("closed"), terminal: true }];
+  }
+
+  const out: PrReconcileTransition[] = [];
+
+  // Review decision transitions (fire only on the edge into the new state).
+  if (next.reviewDecision !== undefined && next.reviewDecision !== prev.reviewDecision) {
+    if (next.reviewDecision === "CHANGES_REQUESTED") {
+      out.push({ event: "changes-requested", tag: tag("changes-requested"), terminal: false });
+    } else if (next.reviewDecision === "APPROVED") {
+      out.push({ event: "approved", tag: tag("approved"), terminal: false });
+    }
+  }
+
+  // Mergeability transitions. "conflicting" is the only conflict signal that
+  // fires a conflict release; UNKNOWN never maps to conflict (never gates as
+  // conflicting). Clearing FROM conflicting back to clean fires conflict-cleared.
+  if (next.mergeable !== undefined && next.mergeable !== prev.mergeable) {
+    if (next.mergeable === "conflicting") {
+      out.push({ event: "conflict", tag: tag("conflict"), terminal: false });
+    } else if (prev.mergeable === "conflicting" && next.mergeable === "clean") {
+      out.push({ event: "conflict-cleared", tag: tag("conflict-cleared"), terminal: false });
+    }
+  }
+
+  return out;
+}
+
+// ── Per-repo tracking ──────────────────────────────────────────────────────────
+
+interface RepoTracker {
+  repo: string;
+  /** Per-entity ETag for conditional probes (entityId → etag). */
+  etags: Map<string, string>;
+  consecutiveErrors: number;
+  /** True if the last tick saw a change (drives active cadence). */
+  active: boolean;
+  /** Ticks with no change (drives idle → dormant). */
+  quietTicks: number;
+  timer?: ReturnType<typeof setTimeout>;
+}
+
+// ── The reconciler ─────────────────────────────────────────────────────────────
+
+export class PrReconciler {
+  private readonly store: PrReconcileStore;
+  private readonly ops: PrReconcileGithubOps;
+  private readonly releaseByEvent: PrReleaseByEventFn;
+  private readonly resolveGroupReleaseTask?: ResolveGroupReleaseTaskFn;
+  private readonly intervals: PrReconcileIntervals;
+  private readonly setTimer: (fn: () => void, ms: number) => ReturnType<typeof setTimeout>;
+  private readonly clearTimer: (handle: ReturnType<typeof setTimeout>) => void;
+
+  private readonly repos = new Map<string, RepoTracker>();
+  private running = false;
+
+  constructor(options: PrReconcilerOptions) {
+    this.store = options.store;
+    this.ops = options.ops;
+    this.releaseByEvent =
+      options.releaseByEvent ??
+      ((taskId, eventTag) =>
+        releaseHeldTaskByEvent(this.store as unknown as import("@fusion/core").TaskStore, taskId, eventTag));
+    this.resolveGroupReleaseTask = options.resolveGroupReleaseTask;
+    this.intervals = { ...DEFAULT_INTERVALS, ...(options.intervals ?? {}) };
+    this.setTimer = options.setTimer ?? ((fn, ms) => setTimeout(fn, ms));
+    this.clearTimer = options.clearTimer ?? ((h) => clearTimeout(h));
+  }
+
+  /**
+   * Start the reconciler. Schedules the first tick for every repo that currently
+   * has active entities, then re-derives the repo set on each tick (so newly
+   * created entities are picked up and terminal ones drop out).
+   */
+  start(): void {
+    if (this.running) return;
+    this.running = true;
+    this.syncReposAndSchedule();
+    prReconcileLog.log("PR reconcile started");
+  }
+
+  /** Stop a single repo's polling. */
+  stopRepo(repo: string): void {
+    const tracker = this.repos.get(repo);
+    if (tracker?.timer) this.clearTimer(tracker.timer);
+    this.repos.delete(repo);
+  }
+
+  /** Stop all polling. */
+  stopAll(): void {
+    for (const tracker of this.repos.values()) {
+      if (tracker.timer) this.clearTimer(tracker.timer);
+    }
+    this.repos.clear();
+    this.running = false;
+    prReconcileLog.log("PR reconcile stopped");
+  }
+
+  /** Currently tracked repos (for tests/observability). */
+  getTrackedRepos(): string[] {
+    return [...this.repos.keys()];
+  }
+
+  /**
+   * Run exactly one tick for one repo. Exposed for deterministic testing without
+   * the timer loop. Returns the transitions fired this tick (across entities).
+   */
+  async reconcileRepoOnce(repo: string): Promise<PrReconcileTransition[]> {
+    const tracker = this.repos.get(repo) ?? this.ensureTracker(repo);
+    return this.tickRepo(tracker);
+  }
+
+  // ── Internals ────────────────────────────────────────────────────────────────
+
+  private ensureTracker(repo: string): RepoTracker {
+    let tracker = this.repos.get(repo);
+    if (!tracker) {
+      tracker = { repo, etags: new Map(), consecutiveErrors: 0, active: true, quietTicks: 0 };
+      this.repos.set(repo, tracker);
+    }
+    return tracker;
+  }
+
+  /** Group active entities by repo; ensure a tracker + scheduled tick per repo. */
+  private syncReposAndSchedule(): void {
+    if (!this.running) return;
+    const byRepo = this.groupActiveByRepo();
+    // Drop repos with no active entities.
+    for (const repo of [...this.repos.keys()]) {
+      if (!byRepo.has(repo)) this.stopRepo(repo);
+    }
+    for (const repo of byRepo.keys()) {
+      const tracker = this.ensureTracker(repo);
+      if (!tracker.timer) this.scheduleNextTick(tracker);
+    }
+  }
+
+  private groupActiveByRepo(): Map<string, PrEntity[]> {
+    const byRepo = new Map<string, PrEntity[]>();
+    let entities: PrEntity[];
+    try {
+      entities = this.store.listActivePrEntities();
+    } catch (err) {
+      prReconcileLog.error("Failed to list active PR entities:", err);
+      return byRepo;
+    }
+    for (const entity of entities) {
+      if (!isPrEntityActive(entity)) continue; // R18: terminal entities are out.
+      if (!entity.repo) continue;
+      const list = byRepo.get(entity.repo) ?? [];
+      list.push(entity);
+      byRepo.set(entity.repo, list);
+    }
+    return byRepo;
+  }
+
+  private resolveInterval(tracker: RepoTracker): number {
+    let base: number;
+    if (tracker.active) base = this.intervals.active;
+    else if (tracker.quietTicks >= 3) base = this.intervals.dormant;
+    else base = this.intervals.idle;
+
+    if (tracker.consecutiveErrors > 0) {
+      const mult = Math.pow(2, Math.min(tracker.consecutiveErrors, 5));
+      base = Math.min(base * mult, this.intervals.maxBackoff);
+    }
+    return base;
+  }
+
+  private scheduleNextTick(tracker: RepoTracker): void {
+    if (!this.running) return;
+    const interval = this.resolveInterval(tracker);
+    tracker.timer = this.setTimer(() => {
+      void this.tickRepo(tracker).finally(() => {
+        // Re-derive the repo set (pick up new entities, drop emptied repos),
+        // then reschedule this repo if it still has work.
+        tracker.timer = undefined;
+        this.syncReposAndSchedule();
+      });
+    }, interval);
+  }
+
+  /**
+   * One reconcile pass over all active entities in a single repo. Per-repo error
+   * handling: a deep-fetch error on one entity is recorded as an audit event and
+   * bumps the repo's backoff, but the loop continues to the next entity and the
+   * poller survives.
+   */
+  private async tickRepo(tracker: RepoTracker): Promise<PrReconcileTransition[]> {
+    const byRepo = this.groupActiveByRepo();
+    const entities = byRepo.get(tracker.repo) ?? [];
+    if (entities.length === 0) {
+      this.stopRepo(tracker.repo);
+      return [];
+    }
+
+    const fired: PrReconcileTransition[] = [];
+    let sawChange = false;
+    let sawError = false;
+
+    for (const entity of entities) {
+      try {
+        const transitions = await this.reconcileEntity(entity, tracker);
+        if (transitions === "changed" || transitions.length > 0) sawChange = true;
+        if (Array.isArray(transitions)) fired.push(...transitions);
+      } catch (err) {
+        sawError = true;
+        this.recordError(entity, err);
+      }
+    }
+
+    // Cadence + backoff bookkeeping.
+    if (sawError) {
+      tracker.consecutiveErrors += 1;
+    } else {
+      tracker.consecutiveErrors = 0;
+    }
+    if (sawChange) {
+      tracker.active = true;
+      tracker.quietTicks = 0;
+    } else {
+      tracker.active = false;
+      tracker.quietTicks += 1;
+    }
+
+    return fired;
+  }
+
+  /**
+   * Reconcile one entity. Returns the transitions fired, or the literal
+   * `"changed"` when GitHub changed but produced no card-advancing transition
+   * (still counts as activity for cadence). Throws on deep-fetch error so the
+   * repo loop can record it and back off.
+   */
+  private async reconcileEntity(
+    entity: PrEntity,
+    tracker: RepoTracker,
+  ): Promise<PrReconcileTransition[] | "changed"> {
+    // An entity without a PR number can only be reconciled by source-of-truth
+    // existence: for unverified imports with no number, treat as fiction.
+    if (entity.prNumber == null) {
+      if (entity.unverified) {
+        this.clearFiction(entity);
+        return [];
+      }
+      // Verified entity still mid-create (no number yet): nothing to reconcile.
+      return [];
+    }
+
+    // 1. ETag-cheap probe. 304 ⇒ unchanged ⇒ no deep-fetch / no writes.
+    const probe = await this.ops.probe(entity.repo, entity.prNumber, tracker.etags.get(entity.id));
+    if (probe.etag) tracker.etags.set(entity.id, probe.etag);
+    if (!probe.changed) return [];
+
+    // 2. Deep-fetch the mirror state (may throw → caller records + backs off).
+    const fetched = await this.ops.fetchPrState(entity.repo, entity.prNumber);
+
+    // 3. Fiction: unverified entity whose PR does not actually exist (R19). Clear
+    //    it to a terminal/cleared state and DO NOT advance it on stale state.
+    if (!fetched.exists) {
+      if (entity.unverified) {
+        this.clearFiction(entity);
+      } else {
+        // A verified entity that vanished from GitHub: treat as closed.
+        this.store.updatePrEntity(entity.id, { state: "closed", unverified: false });
+      }
+      return [];
+    }
+
+    // 4. Derive transitions BEFORE persisting (compare against the prior mirror).
+    const transitions = deriveTransitions(entity, fetched);
+
+    // 5. Persist the corroborated mirror; clear `unverified` on first success.
+    const nextState =
+      fetched.prState === "merged" ? "merged" : fetched.prState === "closed" ? "closed" : entity.state;
+    this.store.updatePrEntity(entity.id, {
+      state: nextState,
+      prNumber: fetched.prNumber ?? entity.prNumber,
+      prUrl: fetched.prUrl ?? null,
+      headOid: fetched.headOid ?? null,
+      mergeable: fetched.mergeable ?? null,
+      checksRollup: fetched.checksRollup ?? null,
+      reviewDecision: fetched.reviewDecision,
+      unverified: false,
+    });
+
+    // 6. Fire the generic external-event releases. The unverified gate (R19) is
+    //    already cleared above only AFTER a real PR was corroborated, so a
+    //    just-cleared entity may legitimately advance on this same pass.
+    for (const transition of transitions) {
+      const taskId = this.resolveReleaseTaskId(entity);
+      if (taskId) {
+        try {
+          await this.releaseByEvent(taskId, transition.tag);
+        } catch (err) {
+          // A release failure must not abort reconcile; record + continue.
+          this.recordError(entity, err, `release:${transition.tag}`);
+        }
+      }
+      // 7. Terminal transition ⇒ entity is now terminal; it drops from the poll
+      //    set on the next groupActiveByRepo() pass (R18). Clear its ETag.
+      if (transition.terminal) tracker.etags.delete(entity.id);
+    }
+
+    return transitions.length > 0 ? transitions : "changed";
+  }
+
+  /** Resolve the task id whose hold should be released for this entity. */
+  private resolveReleaseTaskId(entity: PrEntity): string | null {
+    if (entity.sourceType === "task") return entity.sourceId;
+    // branch-group: requires an injected resolver; otherwise skip (v1 choice).
+    if (this.resolveGroupReleaseTask) return this.resolveGroupReleaseTask(entity);
+    return null;
+  }
+
+  /**
+   * Clear a fictional unverified entity (no real PR behind it, R19): transition
+   * to `closed` and never advance it on stale state.
+   */
+  private clearFiction(entity: PrEntity): void {
+    this.store.updatePrEntity(entity.id, {
+      state: "closed",
+      unverified: false,
+      failureReason: "reconcile: no PR exists on GitHub (cleared fictional unverified entity)",
+    });
+    this.recordAudit(entity, "pr-reconcile:cleared-fiction", {
+      prNumber: entity.prNumber ?? null,
+    });
+    prReconcileLog.log(`Cleared fictional unverified PR entity ${entity.id} (no real PR)`);
+  }
+
+  private recordError(entity: PrEntity, err: unknown, phase = "deep-fetch"): void {
+    const message = err instanceof Error ? err.message : String(err);
+    prReconcileLog.error(`PR reconcile error (${phase}) for entity ${entity.id}: ${message}`);
+    this.recordAudit(entity, "pr-reconcile:error", { phase, error: message });
+  }
+
+  private recordAudit(entity: PrEntity, mutationType: string, metadata: Record<string, unknown>): void {
+    try {
+      void this.store.recordRunAuditEvent?.({
+        taskId: entity.sourceType === "task" ? entity.sourceId : undefined,
+        agentId: "pr-reconcile",
+        runId: `pr-reconcile:${entity.id}`,
+        domain: "database",
+        mutationType,
+        target: entity.id,
+        metadata: { repo: entity.repo, entityId: entity.id, ...metadata },
+      });
+    } catch {
+      // Audit is best-effort, but a thrown audit must never break the poller.
+    }
+  }
+}

packages/engine/src/project-engine-manager.ts

@@ -39,6 +39,7 @@ export interface EngineManagerOptions {
   createGroupPr?: ProjectEngineOptions["createGroupPr"];
   syncGroupPr?: ProjectEngineOptions["syncGroupPr"];
   prNodeGithubOps?: ProjectEngineOptions["prNodeGithubOps"];
+  prReconcileGithubOps?: ProjectEngineOptions["prReconcileGithubOps"];
   getTaskMergeBlocker?: ProjectEngineOptions["getTaskMergeBlocker"];
   onInsightRunProcessed?: ProjectEngineOptions["onInsightRunProcessed"];
 }
@@ -487,6 +488,7 @@ export class ProjectEngineManager {
       createGroupPr: this.options.createGroupPr,
       syncGroupPr: this.options.syncGroupPr,
       prNodeGithubOps: this.options.prNodeGithubOps,
+      prReconcileGithubOps: this.options.prReconcileGithubOps,
       getTaskMergeBlocker: this.options.getTaskMergeBlocker,
       onInsightRunProcessed: this.options.onInsightRunProcessed,
       ...overrides,

packages/engine/src/project-engine.ts

@@ -18,6 +18,7 @@ import type { WorktreePool } from "./worktree-pool.js";
 import type { ProjectRuntimeConfig } from "./project-runtime.js";
 import { PrMonitor } from "./pr-monitor.js";
 import type { PrNodeGithubOps } from "./pr-nodes.js";
+import { PrReconciler, type PrReconcileGithubOps } from "./pr-reconcile.js";
 import { PrCommentHandler } from "./pr-comment-handler.js";
 import { NtfyNotifier } from "./notifier.js";
 import { NotificationService, OAuthAlertStateStore, OAuthExpiryMonitor, OAuthValidityLogger } from "./notification/index.js";
@@ -230,6 +231,15 @@ export interface ProjectEngineOptions {
    * kinds fail closed (value:"pr-nodes-unwired").
    */
   prNodeGithubOps?: PrNodeGithubOps;
+  /**
+   * Node-agnostic GitHub reconcile ops (U4): the injected ETag-probe +
+   * deep-fetch callbacks backing {@link PrReconciler}. Injected from the CLI
+   * layer for the same FN-3049 reason as {@link prNodeGithubOps}. When present,
+   * the runtime layer (this engine, NOT the scheduler) starts a per-repo
+   * reconcile that fires the generic external-event hold releases advancing
+   * PR-await cards. When absent, no reconcile runs.
+   */
+  prReconcileGithubOps?: PrReconcileGithubOps;
   /**
    * Returns the merge blocker reason for a task, or null/undefined if
    * the task is eligible for merge. Imported from @fusion/core.
@@ -267,6 +277,7 @@ export class ProjectEngine {
   private runtime: InProcessRuntime;
   private started = false;
   private prMonitor?: PrMonitor;
+  private prReconciler?: PrReconciler;
   private prCommentHandler?: PrCommentHandler;
   private notifier?: NtfyNotifier;
   private notificationService?: NotificationService;
@@ -471,6 +482,19 @@ export class ProjectEngine {
         this.prCommentHandler!.createFollowUpTask(taskId, prInfo, comments),
     });
 
+    // 2b. Node-agnostic GitHub reconcile (U4). Started HERE in the runtime layer,
+    // NOT in scheduler.ts (R20 invariant: the scheduler stays PR-ignorant). The
+    // reconciler keys on active PR entities, fires generic external-event hold
+    // releases, and persists audit on error. Only runs when the CLI injected the
+    // probe/deep-fetch ops.
+    if (this.options.prReconcileGithubOps) {
+      this.prReconciler = new PrReconciler({
+        store,
+        ops: this.options.prReconcileGithubOps,
+      });
+      this.prReconciler.start();
+    }
+
     // 3. Initialize notification services (unless caller manages them externally)
     if (!this.options.skipNotifier) {
       const agentStore = this.runtime.getAgentStore();
@@ -728,6 +752,8 @@ export class ProjectEngine {
     }
 
     // Stop auxiliary subsystems
+    this.prReconciler?.stopAll();
+    this.prReconciler = undefined;
     this.oauthExpiryMonitor?.stop();
     this.oauthValidityLogger?.stop();
     this.notificationService?.stop();