diff --git a/packages/cli/src/commands/daemon.ts b/packages/cli/src/commands/daemon.ts index fdabef21b7..6288d15d3f 100644 --- a/packages/cli/src/commands/daemon.ts +++ b/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), }); diff --git a/packages/cli/src/commands/dashboard.ts b/packages/cli/src/commands/dashboard.ts index 6ca6ab1b54..493ac48201 100644 --- a/packages/cli/src/commands/dashboard.ts +++ b/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, }); diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 3c7c5306cb..8a4bafc4cc 100644 --- a/packages/cli/src/commands/serve.ts +++ b/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), }); diff --git a/packages/cli/src/commands/task-lifecycle.ts b/packages/cli/src/commands/task-lifecycle.ts index d1f06da2a4..c869401f55 100644 --- a/packages/cli/src/commands/task-lifecycle.ts +++ b/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; updatePr(params: { owner?: string; repo?: string; number: number; title?: string; body?: string }): Promise; closePr(params: { number: number }): Promise; + /** 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, +): PrReconcileGithubOps { + return { + probe: (repo, prNumber, etag) => { + const { owner, name } = splitRepoSlug(repo); + return github.probePrChanged(owner, name, prNumber, etag); + }, + fetchPrState: async (repo, prNumber): Promise => { + 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 { try { const { stdout } = await execAsync(`git rev-list --count "${baseBranch}..${branch}"`, { cwd, timeout: 30_000 }); diff --git a/packages/engine/src/__tests__/pr-reconcile.test.ts b/packages/engine/src/__tests__/pr-reconcile.test.ts new file mode 100644 index 0000000000..d122685649 --- /dev/null +++ b/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)) => 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 = 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; + + 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 & { 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, + // Tiny intervals + a no-op timer keep the loop off the test clock. + setTimer: () => 0 as unknown as ReturnType, + 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/); + }); +}); diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index d931afcc20..e3e8a31d85 100644 --- a/packages/engine/src/index.ts +++ b/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, diff --git a/packages/engine/src/logger.ts b/packages/engine/src/logger.ts index 7bb561ab1b..d4f522aafb 100644 --- a/packages/engine/src/logger.ts +++ b/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"); diff --git a/packages/engine/src/pr-reconcile.ts b/packages/engine/src/pr-reconcile.ts new file mode 100644 index 0000000000..c96c731db0 --- /dev/null +++ b/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-")` — 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; +} + +// ── 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; + +/** + * 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; + /** Injected clock for the next-tick scheduler (defaults to setTimeout). */ + setTimer?: (fn: () => void, ms: number) => ReturnType; + clearTimer?: (handle: ReturnType) => 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-`. */ + 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; + 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; +} + +// ── 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; + private readonly clearTimer: (handle: ReturnType) => void; + + private readonly repos = new Map(); + 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 { + 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 { + const byRepo = new Map(); + 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 { + 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 { + // 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): 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. + } + } +} diff --git a/packages/engine/src/project-engine-manager.ts b/packages/engine/src/project-engine-manager.ts index 991a3ea03d..87ce037b9e 100644 --- a/packages/engine/src/project-engine-manager.ts +++ b/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, diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 9992b29c75..e54a6cf128 100644 --- a/packages/engine/src/project-engine.ts +++ b/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();