diff --git a/packages/cli/src/commands/daemon.ts b/packages/cli/src/commands/daemon.ts index 27d1af4f67..fdabef21b7 100644 --- a/packages/cli/src/commands/daemon.ts +++ b/packages/cli/src/commands/daemon.ts @@ -44,6 +44,7 @@ import { processPullRequestMergeTask, createGroupPrCallback, syncGroupPrCallback, + createPrNodeGithubOps, } from "./task-lifecycle.js"; import { promptForPort } from "./port-prompt.js"; import { createReadOnlyProviderSettingsView } from "./provider-settings.js"; @@ -338,6 +339,7 @@ export async function runDaemon(opts: DaemonOptions = {}) { processPullRequestMergeTask(s, wd, taskId, githubClient, getTaskMergeBlocker, pool), createGroupPr: createGroupPrCallback(githubClient), syncGroupPr: syncGroupPrCallback(githubClient), + prNodeGithubOps: createPrNodeGithubOps(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 0618905751..6ca6ab1b54 100644 --- a/packages/cli/src/commands/dashboard.ts +++ b/packages/cli/src/commands/dashboard.ts @@ -54,6 +54,7 @@ import { processPullRequestMergeTask, createGroupPrCallback, syncGroupPrCallback, + createPrNodeGithubOps, } from "./task-lifecycle.js"; import { promptForPort } from "./port-prompt.js"; import { ensureCwdProjectRegistered } from "./ensure-project-registered.js"; @@ -1614,6 +1615,7 @@ export async function runDashboard(port: number, opts: { paused?: boolean; dev?: processPullRequestMergeTask(s, wd, taskId, githubClient, getTaskMergeBlocker, pool), createGroupPr: createGroupPrCallback(githubClient), syncGroupPr: syncGroupPrCallback(githubClient), + prNodeGithubOps: createPrNodeGithubOps(githubClient), getTaskMergeBlocker, }); diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 8f01b2c798..3c7c5306cb 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -44,6 +44,7 @@ import { processPullRequestMergeTask, createGroupPrCallback, syncGroupPrCallback, + createPrNodeGithubOps, } from "./task-lifecycle.js"; import { promptForPort } from "./port-prompt.js"; import { createReadOnlyProviderSettingsView } from "./provider-settings.js"; @@ -364,6 +365,7 @@ export async function runServe( processPullRequestMergeTask(s, wd, taskId, githubClient, getTaskMergeBlocker, pool), createGroupPr: createGroupPrCallback(githubClient), syncGroupPr: syncGroupPrCallback(githubClient), + prNodeGithubOps: createPrNodeGithubOps(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 be0004c70e..d1f06da2a4 100644 --- a/packages/cli/src/commands/task-lifecycle.ts +++ b/packages/cli/src/commands/task-lifecycle.ts @@ -27,7 +27,7 @@ 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 } from "@fusion/engine"; +import type { CreateGroupPrFn, SyncGroupPrFn, WorktreePool, PrNodeGithubOps } from "@fusion/engine"; /** * Minimal interface for GitHub operations needed by the PR merge workflow. @@ -43,7 +43,7 @@ interface GitHubOperations { mergeReady: boolean; blockingReasons: string[]; }>; - mergePr(params: { number: number; method?: "merge" | "squash" | "rebase" }): Promise; + mergePr(params: { number: number; method?: "merge" | "squash" | "rebase"; expectedHeadOid?: string }): Promise; 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; @@ -303,6 +303,90 @@ export function syncGroupPrCallback( }; } +/** Best-effort resolve the head commit OID for a branch (so `pr-merge` can pass + * `expectedHeadOid`). Returns undefined on any failure — the merge then runs + * without the stale-head guard, which the reconcile still corroborates. */ +async function resolveBranchHeadOid(cwd: string, branch: string): Promise { + try { + const { stdout } = await execAsync(`git rev-parse "${branch}"`, { cwd, timeout: 30_000 }); + const oid = stdout.trim(); + return oid.length > 0 ? oid : undefined; + } catch { + return undefined; + } +} + +/** Structural detection of the dashboard `PrStaleHeadError` without importing the + * class (task-lifecycle.ts deliberately has no @fusion/dashboard dependency). */ +function isStaleHeadError(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code?: unknown }).code === "stale-head" + ); +} + +/** + * Build the `prNodeGithubOps` engine callbacks (U3) backing the `pr-create` / + * `pr-respond` / `pr-merge` workflow nodes. Closes over a GitHub client so the + * engine never imports the dashboard `GitHubClient` (FN-3049). Mirrors + * `createGroupPrCallback` / `syncGroupPrCallback`. + * + * - resolvePrSource: derives the single-task PR source identity (repo from the + * per-process repo, head branch from the task branch-naming convention). + * - createPr: pushes the task branch to origin, opens the PR, resolves the head OID. + * - mergePr: merges with `expectedHeadOid`; a `PrStaleHeadError` (detected + * structurally) maps to `{ status: "stale-head" }` so the node routes the race. + * - respond: omitted in U3 (U5 wires the real review-response run); the node then + * falls back to its inert `disagreed-only` default. + */ +export function createPrNodeGithubOps( + github: Pick, +): PrNodeGithubOps { + return { + resolvePrSource: (task) => { + const repo = getCurrentRepo(); + const repoSlug = repo ? `${repo.owner}/${repo.repo}` : ""; + return { + sourceType: "task", + sourceId: task.id, + repo: repoSlug, + headBranch: getTaskBranchName(task.id), + }; + }, + createPr: async ({ task, entity }) => { + const cwd = process.cwd(); + const headBranch = entity.headBranch || getTaskBranchName(task.id); + await pushTaskBranchToOrigin(cwd, headBranch); + const created = await github.createPr({ + title: task.title ?? `Task ${task.id}`, + body: task.description ?? "", + head: headBranch, + base: entity.baseBranch, + }); + const headOid = await resolveBranchHeadOid(cwd, headBranch); + return { prNumber: created.number, prUrl: created.url, headOid }; + }, + mergePr: async ({ entity }) => { + if (entity.prNumber == null) { + throw new Error(`pr-merge: entity ${entity.id} has no persisted prNumber`); + } + try { + await github.mergePr({ + number: entity.prNumber, + method: "squash", + expectedHeadOid: entity.headOid, + }); + return { status: "merged-requested" }; + } catch (err) { + if (isStaleHeadError(err)) return { status: "stale-head" }; + throw err; + } + }, + }; +} + 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/core/src/index.ts b/packages/core/src/index.ts index 999eb211cb..65dbc70d59 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -522,6 +522,23 @@ export { isBranchGroupMemberLanded, isBranchGroupComplete, } from "./branch-group-completion.js"; +export type { + PrEntity, + PrEntityCreateInput, + PrEntityUpdate, + PrEntityState, + PrEntitySourceType, + PrReviewDecision, + PrChecksRollup, + PrThreadState, + PrThreadOutcome, +} from "./types.js"; +export { + isPrEntityActive, + isPrBacked, + isPrEntityActionable, + isPrEntityAutoMergeReady, +} from "./pr-entity.js"; export { findVitestProcessIds, type FindVitestProcessIdsOptions, diff --git a/packages/core/src/workflow-ir-types.ts b/packages/core/src/workflow-ir-types.ts index b96e538fb5..e96a0eeba6 100644 --- a/packages/core/src/workflow-ir-types.ts +++ b/packages/core/src/workflow-ir-types.ts @@ -1,9 +1,11 @@ /** Node kinds. v1 kinds (start/prompt/script/gate/end) plus the v2 additions: - * `hold` (passive dwell column states), `split`/`join` (parallel fan-out), and - * the step-inversion additions (FN step-inversion, KTD-3/4/12/15): - * `foreach` (runtime-expanding per-step template region), `step-review` - * (per-step review verdicts as outcome edges), `parse-steps` (graph-native - * step-list parsing), and `code` (sandboxed TypeScript). */ + * `hold` (passive dwell column states), `split`/`join` (parallel fan-out), the + * step-inversion additions (FN step-inversion, KTD-3/4/12/15): `foreach` + * (runtime-expanding per-step template region), `step-review` (per-step review + * verdicts as outcome edges), `parse-steps` (graph-native step-list parsing), + * and `code` (sandboxed TypeScript); and the unified PR-entity additions (U3): + * `pr-create` (open/reuse the PR + write the entity), `pr-respond` (the + * review-response run), and `pr-merge` (tool-side merge with expectedHeadOid). */ export type WorkflowIrNodeKind = | "start" | "prompt" @@ -16,7 +18,10 @@ export type WorkflowIrNodeKind = | "foreach" | "step-review" | "parse-steps" - | "code"; + | "code" + | "pr-create" + | "pr-respond" + | "pr-merge"; export interface WorkflowIrNode { id: string; diff --git a/packages/engine/src/__tests__/pr-nodes.test.ts b/packages/engine/src/__tests__/pr-nodes.test.ts new file mode 100644 index 0000000000..07fb0bc426 --- /dev/null +++ b/packages/engine/src/__tests__/pr-nodes.test.ts @@ -0,0 +1,212 @@ +/** + * U3 — PR node handlers (pr-create / pr-respond / pr-merge). + * + * Covers: pr-create success→open, pr-create failure→failed (routable, never + * throws), create idempotent re-entry, pr-merge stale-head→value:"stale-head" + * with no `merged` write, pr-merge does-not-write-merged on success, unverified + * entity not actioned, and unwired deps fail closed (value:"pr-nodes-unwired"). + * + * The handlers run against a real in-memory TaskStore (U1 store CRUD) and fakes + * for the injected GitHub callbacks — the engine never touches a real client. + */ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync } from "node:fs"; +import { rm } from "node:fs/promises"; +import { join } from "node:path"; +import { tmpdir } from "node:os"; +import { TaskStore } from "@fusion/core"; +import type { TaskDetail, WorkflowIrNode } from "@fusion/core"; + +import { + createPrNodeHandlers, + type PrMergeCallResult, + type PrNodeDeps, + type PrSourceDescriptor, +} from "../pr-nodes.js"; +import { createDefaultNodeHandlers, createNoopLegacySeams } from "../workflow-node-handlers.js"; +import type { WorkflowNodeExecutionContext } from "../workflow-graph-executor.js"; + +function makeTmpDir(): string { + return mkdtempSync(join(tmpdir(), "fusion-pr-nodes-test-")); +} + +const SOURCE: PrSourceDescriptor = { + sourceType: "task", + sourceId: "T-1", + repo: "owner/repo", + headBranch: "fusion/t-1", +}; + +function ctx(taskId = "T-1"): WorkflowNodeExecutionContext { + return { + task: { id: taskId } as unknown as TaskDetail, + settings: undefined, + context: {}, + }; +} + +const NODE = { id: "n", kind: "pr-create" } as WorkflowIrNode; + +describe("PR node handlers (U3)", () => { + let rootDir: string; + let store: TaskStore; + + beforeEach(async () => { + rootDir = makeTmpDir(); + store = new TaskStore(rootDir, join(rootDir, ".fusion-global")); + await store.init(); + }); + + afterEach(async () => { + store.close(); + await rm(rootDir, { recursive: true, force: true, maxRetries: 5, retryDelay: 50 }); + }); + + function deps(overrides: Partial = {}): PrNodeDeps { + return { + getStore: () => store, + resolvePrSource: () => SOURCE, + createPr: async () => ({ prNumber: 42, prUrl: "https://github.com/owner/repo/pull/42", headOid: "abc123" }), + mergePr: async () => ({ status: "merged-requested" }) as PrMergeCallResult, + ...overrides, + }; + } + + it("pr-create success → entity open with persisted PR fields, value:open", async () => { + const handlers = createPrNodeHandlers(deps()); + const result = await handlers["pr-create"](NODE, ctx()); + expect(result).toEqual({ outcome: "success", value: "open" }); + + const entity = store.getActivePrEntityBySource("task", "T-1"); + expect(entity?.state).toBe("open"); + expect(entity?.prNumber).toBe(42); + expect(entity?.prUrl).toBe("https://github.com/owner/repo/pull/42"); + expect(entity?.headOid).toBe("abc123"); + }); + + it("pr-create failure → entity failed + failureReason, value:failed (routable, never throws)", async () => { + // Pre-create the entity so we hold its id (the failed row leaves the active set). + const seeded = store.ensurePrEntityForSource(SOURCE); + const handlers = createPrNodeHandlers( + deps({ + createPr: async () => { + throw new Error("boom-create"); + }, + }), + ); + const result = await handlers["pr-create"](NODE, ctx()); + // Failure is a ROUTABLE success-outcome with value:"failed", not a throw. + expect(result).toEqual({ outcome: "success", value: "failed" }); + + // `failed` is terminal, so the entity is no longer "active" — but it exists. + expect(store.getActivePrEntityBySource("task", "T-1")).toBeNull(); + const failed = store.getPrEntity(seeded.id); + expect(failed?.state).toBe("failed"); + expect(failed?.failureReason).toContain("boom-create"); + expect(failed?.prNumber).toBeUndefined(); + }); + + it("pr-create idempotent re-entry on an already-open entity is a no-op", async () => { + const createPr = vi.fn(async () => ({ prNumber: 7, prUrl: "u", headOid: "h" })); + const handlers = createPrNodeHandlers(deps({ createPr })); + + const first = await handlers["pr-create"](NODE, ctx()); + expect(first.value).toBe("open"); + expect(createPr).toHaveBeenCalledTimes(1); + + const second = await handlers["pr-create"](NODE, ctx()); + expect(second).toEqual({ outcome: "success", value: "open" }); + // Re-entry must NOT call GitHub again, and must NOT mint a second entity. + expect(createPr).toHaveBeenCalledTimes(1); + }); + + it("pr-merge stale head → value:stale-head, entity stays open, no merged write", async () => { + // Seed an open, verified entity. + const created = store.ensurePrEntityForSource({ ...SOURCE, state: "open", prNumber: 9 }); + store.updatePrEntity(created.id, { headOid: "stale" }); + + const handlers = createPrNodeHandlers( + deps({ mergePr: async () => ({ status: "stale-head" }) as PrMergeCallResult }), + ); + const result = await handlers["pr-merge"]({ id: "m", kind: "pr-merge" } as WorkflowIrNode, ctx()); + expect(result).toEqual({ outcome: "success", value: "stale-head" }); + + const entity = store.getPrEntity(created.id); + expect(entity?.state).toBe("open"); // never advanced to merged + }); + + it("pr-merge success emits merged-requested and does NOT write merged (reconcile corroborates)", async () => { + const created = store.ensurePrEntityForSource({ ...SOURCE, state: "open", prNumber: 9 }); + store.updatePrEntity(created.id, { headOid: "tip" }); + + const mergePr = vi.fn(async () => ({ status: "merged-requested" }) as PrMergeCallResult); + const handlers = createPrNodeHandlers(deps({ mergePr })); + const result = await handlers["pr-merge"]({ id: "m", kind: "pr-merge" } as WorkflowIrNode, ctx()); + expect(result).toEqual({ outcome: "success", value: "merged-requested" }); + // expectedHeadOid is passed from the entity's headOid. + expect(mergePr).toHaveBeenCalledWith(expect.objectContaining({ expectedHeadOid: "tip" })); + + const entity = store.getPrEntity(created.id); + expect(entity?.state).toBe("open"); // node never writes merged + }); + + it("unverified entity is not merged or responded to — emits a benign outcome", async () => { + const created = store.ensurePrEntityForSource({ + ...SOURCE, + state: "open", + prNumber: 9, + unverified: true, + }); + + const mergePr = vi.fn(async () => ({ status: "merged-requested" }) as PrMergeCallResult); + const respond = vi.fn(async () => ({ value: "fixed" as const })); + const handlers = createPrNodeHandlers(deps({ mergePr, respond })); + + const merge = await handlers["pr-merge"]({ id: "m", kind: "pr-merge" } as WorkflowIrNode, ctx()); + expect(merge).toEqual({ outcome: "success", value: "not-actionable" }); + expect(mergePr).not.toHaveBeenCalled(); + + const resp = await handlers["pr-respond"]({ id: "r", kind: "pr-respond" } as WorkflowIrNode, ctx()); + expect(resp).toEqual({ outcome: "success", value: "not-actionable" }); + expect(respond).not.toHaveBeenCalled(); + + const entity = store.getPrEntity(created.id); + expect(entity?.state).toBe("open"); + }); + + it("pr-respond default (no respond dep) is inert: value:disagreed-only + bumps responseRounds", async () => { + const created = store.ensurePrEntityForSource({ ...SOURCE, state: "open", prNumber: 9 }); + expect(store.getPrEntity(created.id)?.responseRounds).toBe(0); + + const handlers = createPrNodeHandlers(deps()); // no respond + const result = await handlers["pr-respond"]({ id: "r", kind: "pr-respond" } as WorkflowIrNode, ctx()); + expect(result).toEqual({ outcome: "success", value: "disagreed-only" }); + + expect(store.getPrEntity(created.id)?.responseRounds).toBe(1); + }); + + it("pr-respond delegates to the injected respond callback", async () => { + store.ensurePrEntityForSource({ ...SOURCE, state: "open", prNumber: 9 }); + const respond = vi.fn(async () => ({ value: "fixed" as const, contextPatch: { k: "v" } })); + const handlers = createPrNodeHandlers(deps({ respond })); + + const result = await handlers["pr-respond"]({ id: "r", kind: "pr-respond" } as WorkflowIrNode, ctx()); + expect(result).toEqual({ outcome: "success", value: "fixed", contextPatch: { k: "v" } }); + expect(respond).toHaveBeenCalledTimes(1); + }); + + it("unwired pr-* deps fail closed (value:pr-nodes-unwired)", async () => { + // createDefaultNodeHandlers with no prNodes dep → the three kinds fail closed. + const handlers = createDefaultNodeHandlers(createNoopLegacySeams(), undefined, {}); + for (const kind of ["pr-create", "pr-respond", "pr-merge"] as const) { + const result = await handlers[kind]({ id: kind, kind } as WorkflowIrNode, ctx()); + expect(result).toEqual({ outcome: "failure", value: "pr-nodes-unwired" }); + } + }); + + it("createDefaultNodeHandlers wires real pr-* handlers when prNodes is supplied", async () => { + const handlers = createDefaultNodeHandlers(createNoopLegacySeams(), undefined, { prNodes: deps() }); + const result = await handlers["pr-create"](NODE, ctx()); + expect(result).toEqual({ outcome: "success", value: "open" }); + }); +}); diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 3a29414187..7e0f1fa35b 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -1116,6 +1116,11 @@ export interface TaskExecutorOptions { onAgentText?: (taskId: string, delta: string) => void; onAgentTool?: (taskId: string, toolName: string) => void; autoRecoveryDispatcher?: AutoRecoveryDispatcher; + /** PR-entity node deps (U3): assembled `PrNodeDeps` (store + injected GitHub + * callbacks) for the `pr-create`/`pr-respond`/`pr-merge` workflow nodes. The + * runtime binds the store and threads the CLI-injected ops. Absent → the pr-* + * node kinds fail closed. */ + prNodes?: import("./pr-nodes.js").PrNodeDeps; } export class TaskExecutor { @@ -3346,6 +3351,9 @@ export class TaskExecutor { // Step-inversion (KTD-15, U14): code node runner — esbuild compile + // child-process execution with the harness contract. runCode: this.buildCodeNodeRunner(), + // PR-entity nodes (U3): pr-create/pr-respond/pr-merge handler deps — + // engine-owned store + CLI-injected GitHub callbacks. Absent → fail closed. + prNodes: this.options.prNodes, // Step-inversion (KTD-11, U10): worktree isolation + ordered integration + // parallel scheduling. Per-instance worktrees branched off the task's main // branch tip; integration rebases each branch in step order; the projection diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index dec5919a47..d931afcc20 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -46,6 +46,20 @@ export { type CodeNodeRunner, type DefaultNodeHandlerDeps, } from "./workflow-node-handlers.js"; +export { + createPrNodeHandlers, + buildPrNodeDeps, + type PrNodeDeps, + type PrNodeGithubOps, + type PrNodeStore, + type PrSourceDescriptor, + type PrCreateCallInput, + type PrCreateCallResult, + type PrMergeCallInput, + type PrMergeCallResult, + type PrRespondCallInput, + type PrRespondCallResult, +} from "./pr-nodes.js"; export { WorkflowGraphTaskRunner, type WorkflowGraphRunDisposition, diff --git a/packages/engine/src/pr-nodes.ts b/packages/engine/src/pr-nodes.ts new file mode 100644 index 0000000000..dcd1e70003 --- /dev/null +++ b/packages/engine/src/pr-nodes.ts @@ -0,0 +1,326 @@ +// PR node handlers for the unified PR-entity review loop (U3). +// +// Three first-class node kinds whose handlers own the PR side effects and emit +// outcomes the graph routes on: +// - pr-create : open (or reuse) the PR and write the entity to `open` /`failed` +// - pr-respond : run the review-response loop body (U5 fills the real body; +// U3 delegates to an injected callback defaulting to a no-op) +// - pr-merge : tool-side merge with `expectedHeadOid` (reconcile, U4, +// corroborates the terminal `merged` write — the node never does) +// +// All handlers are idempotent, fast (no indefinite waits — those are holds, U4), +// and fail-closed. The engine NEVER imports the dashboard GitHubClient: every +// GitHub side effect is an injected callback wired from the CLI composition layer +// (mirroring how `createGroupPr`/`syncGroupPr` are wired). That keeps the engine +// free of the dashboard dependency (FN-3049: static imports only, no dashboard +// client) and unit-testable with fakes. + +import { + isPrEntityActionable, + type PrEntity, + type PrEntityCreateInput, + type PrEntityUpdate, + type TaskDetail, + type WorkflowIrNode, +} from "@fusion/core"; + +import type { WorkflowNodeHandler } from "./workflow-graph-executor.js"; + +/** + * The narrow slice of the store the PR node handlers need. Declared structurally + * (not as the full `TaskStore`) so the engine stays decoupled from the concrete + * store and the handlers stay trivially fakeable in tests. + */ +export interface PrNodeStore { + /** Create-or-reuse the single non-terminal entity for a source (AE6 idempotency). */ + ensurePrEntityForSource(input: PrEntityCreateInput): PrEntity; + getPrEntity(id: string): PrEntity | null; + getActivePrEntityBySource(sourceType: PrEntity["sourceType"], sourceId: string): PrEntity | null; + updatePrEntity(id: string, patch: PrEntityUpdate): PrEntity; +} + +/** Identity of the PR an entity is created for, resolved from the task + node. */ +export interface PrSourceDescriptor extends PrEntityCreateInput {} + +/** Input for the injected `createPr` callback (the dashboard GitHubClient wrapper). */ +export interface PrCreateCallInput { + task: TaskDetail; + node: WorkflowIrNode; + entity: PrEntity; +} + +/** Result of a successful PR creation — the GitHub-mirror fields the node persists. */ +export interface PrCreateCallResult { + prNumber: number; + prUrl: string; + /** Resolved head commit OID, persisted so `pr-merge` can pass `expectedHeadOid`. */ + headOid?: string; +} + +/** Input for the injected `mergePr` callback. */ +export interface PrMergeCallInput { + task: TaskDetail; + node: WorkflowIrNode; + entity: PrEntity; + /** The head OID the merge is gated on (defeats the push/merge race, U2/U6). */ + expectedHeadOid?: string; +} + +/** + * Discriminated result of the injected `mergePr` callback. The callback wraps the + * dashboard `mergePr`, which throws `PrStaleHeadError` on a head-moved race; the + * wrapper catches it and returns `{ status: "stale-head" }` so the engine never + * imports the dashboard error class. Any other failure should be thrown so the + * handler classifies it as a benign retryable outcome. + */ +export type PrMergeCallResult = + | { status: "merged-requested" } + | { status: "stale-head" }; + +/** Input for the injected `respond` callback (U5 implements the real body). */ +export interface PrRespondCallInput { + task: TaskDetail; + node: WorkflowIrNode; + entity: PrEntity; + context: Record; +} + +/** + * Result of the injected `respond` callback. `outcome` is the routing value the + * `pr-respond` node emits (drives the bounded-rework edge back to await-review): + * - "fixed" : a fix was pushed; loop back to await-review + * - "disagreed-only" : nothing actionable / all threads disagreed; leave open + */ +export interface PrRespondCallResult { + value: "fixed" | "disagreed-only"; + contextPatch?: Record; +} + +/** + * Dependencies the PR node handlers close over. All injected from the CLI + * composition layer where importing the dashboard GitHubClient IS allowed; the + * engine receives only plain callbacks + a structural store accessor. + */ +export interface PrNodeDeps { + /** Structural store accessor (the engine already owns the store instance). */ + getStore(): PrNodeStore; + /** + * Resolve the PR source identity for a `pr-create` node from the task + node. + * The CLI wiring derives sourceType/sourceId (task id or branch-group id), + * repo, and head/base branch from the task's branch-naming + tracking config. + */ + resolvePrSource(task: TaskDetail, node: WorkflowIrNode): Promise | PrSourceDescriptor; + /** Open the PR on GitHub. Throws on failure (the node records `failed`). */ + createPr(input: PrCreateCallInput): Promise; + /** Merge the PR tool-side with `expectedHeadOid`. Returns a discriminated result. */ + mergePr(input: PrMergeCallInput): Promise; + /** + * Run the review-response body (U5). Defaults to a no-op returning + * `disagreed-only` when omitted, so U3 ships a routable-but-inert pr-respond. + */ + respond?: (input: PrRespondCallInput) => Promise; + /** Optional audit sink, called with a stable reason on every routable failure. */ + audit?: (reason: string, detail: string) => void; +} + +/** + * The CLI-injected slice of {@link PrNodeDeps}: only the GitHub side-effect + * callbacks (which close over the dashboard `GitHubClient`) plus the source + * resolver and audit sink. The engine binds `getStore` itself (it owns the store + * instance) via {@link buildPrNodeDeps}, so the CLI layer never needs a store + * reference. Mirrors how `createGroupPr`/`syncGroupPr` are injected as plain + * callbacks from the CLI composition layer. + */ +export interface PrNodeGithubOps { + resolvePrSource: PrNodeDeps["resolvePrSource"]; + createPr: PrNodeDeps["createPr"]; + mergePr: PrNodeDeps["mergePr"]; + respond?: PrNodeDeps["respond"]; + audit?: PrNodeDeps["audit"]; +} + +/** + * Assemble full {@link PrNodeDeps} from the engine-owned store + the CLI-injected + * GitHub ops. Used by the runtime/executor wiring so the CLI layer stays free of + * any store reference and the engine never imports the dashboard client. + */ +export function buildPrNodeDeps(getStore: () => PrNodeStore, ops: PrNodeGithubOps): PrNodeDeps { + return { + getStore, + resolvePrSource: ops.resolvePrSource, + createPr: ops.createPr, + mergePr: ops.mergePr, + respond: ops.respond, + audit: ops.audit, + }; +} + +function classifyError(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} + +/** + * Build the three PR node handlers from injected deps. Mirrors the seam-injection + * pattern (`createStepReviewHandler` / `createParseStepsHandler`): the engine + * graph layer stays engine-agnostic and unit-testable with fakes. + */ +export function createPrNodeHandlers(deps: PrNodeDeps): Record< + "pr-create" | "pr-respond" | "pr-merge", + WorkflowNodeHandler +> { + const audit = (reason: string, detail: string): void => { + try { + deps.audit?.(reason, detail); + } catch { + // Audit must never affect the run. + } + }; + + // ── pr-create ────────────────────────────────────────────────────────────── + // Ensure the entity in `creating`, call GitHub, flip to `open` on success or + // `failed` (routable, NOT a thrown error) on failure. Re-entry on an already + // open entity is a no-op emitting value:"open" (AE6 create-or-reuse idempotency). + const prCreate: WorkflowNodeHandler = async (node, ctx) => { + const store = deps.getStore(); + + let source: PrSourceDescriptor; + try { + source = await deps.resolvePrSource(ctx.task, node); + } catch (err) { + const detail = `pr-create node '${node.id}' could not resolve PR source: ${classifyError(err)}`; + audit("pr-create-source-error", detail); + // No entity yet → fail closed with a routable outcome. + return { outcome: "failure", value: "source-error" }; + } + + // Create-or-reuse the single live entity (the store enforces the partial + // unique index, so re-entry never mints a second entity). + const entity = store.ensurePrEntityForSource({ + ...source, + state: source.state ?? "creating", + }); + + // Idempotent re-entry: an already-open entity with a persisted PR is a no-op. + if (entity.state === "open" && entity.prNumber != null) { + return { outcome: "success", value: "open" }; + } + + // Ensure the row is in `creating` before the side effect (so a crash mid-flight + // leaves a recoverable state, not a stale `failed`). + const creating = entity.state === "creating" ? entity : store.updatePrEntity(entity.id, { state: "creating" }); + + let created: PrCreateCallResult; + try { + created = await deps.createPr({ task: ctx.task, node, entity: creating }); + } catch (err) { + const reason = classifyError(err); + audit("pr-create-failed", `pr-create node '${node.id}' creation failed: ${reason}`); + // Failure is a ROUTABLE outcome — the graph routes on value:"failed". Record + // the classified reason and the failed state; never throw. + store.updatePrEntity(creating.id, { state: "failed", failureReason: reason }); + return { outcome: "success", value: "failed" }; + } + + store.updatePrEntity(creating.id, { + state: "open", + prNumber: created.prNumber, + prUrl: created.prUrl, + headOid: created.headOid ?? null, + }); + return { outcome: "success", value: "open" }; + }; + + // ── pr-merge ─────────────────────────────────────────────────────────────── + // Merge tool-side with `expectedHeadOid` from the entity. Does NOT write the + // terminal `merged` state — the reconcile (U4) corroborates that from GitHub. + // A stale-head race emits value:"stale-head" leaving the entity open; a clean + // merge request emits value:"merged-requested". + const prMerge: WorkflowNodeHandler = async (node, ctx) => { + const store = deps.getStore(); + const entity = store.getActivePrEntityBySource("task", ctx.task.id) + ?? store.getActivePrEntityBySource("branch-group", ctx.task.id); + + if (!entity) { + audit("pr-merge-no-entity", `pr-merge node '${node.id}' found no live PR entity for task ${ctx.task.id}`); + return { outcome: "failure", value: "no-entity" }; + } + + // Unverified entities (imported legacy state GitHub has not corroborated) are + // a hard gate (R19): never merge on stale state — emit a benign outcome. + if (!isPrEntityActionable(entity)) { + audit("pr-merge-not-actionable", `pr-merge node '${node.id}' entity ${entity.id} not actionable (unverified/terminal)`); + return { outcome: "success", value: "not-actionable" }; + } + + let result: PrMergeCallResult; + try { + result = await deps.mergePr({ + task: ctx.task, + node, + entity, + expectedHeadOid: entity.headOid, + }); + } catch (err) { + // A non-stale merge error is benign/retryable — never throw out of the + // handler, and never write `merged`. Route a routable failure value. + const reason = classifyError(err); + audit("pr-merge-error", `pr-merge node '${node.id}' merge failed: ${reason}`); + return { outcome: "failure", value: "merge-error" }; + } + + if (result.status === "stale-head") { + // The head moved since we read `expectedHeadOid`; leave the entity open so a + // re-evaluation merges against the new head. Never write `merged`. + return { outcome: "success", value: "stale-head" }; + } + + // Merge requested cleanly. Do NOT write `merged` here — reconcile corroborates. + return { outcome: "success", value: "merged-requested" }; + }; + + // ── pr-respond ───────────────────────────────────────────────────────────── + // Delegate to the injected `respond` callback (U5 implements the real body). + // Defaults to a no-op returning value:"disagreed-only". Increments the entity's + // responseRounds (the R8 iteration-cap counter, survives restart). + const prRespond: WorkflowNodeHandler = async (node, ctx) => { + const store = deps.getStore(); + const entity = store.getActivePrEntityBySource("task", ctx.task.id) + ?? store.getActivePrEntityBySource("branch-group", ctx.task.id); + + if (!entity) { + audit("pr-respond-no-entity", `pr-respond node '${node.id}' found no live PR entity for task ${ctx.task.id}`); + return { outcome: "failure", value: "no-entity" }; + } + + // Unverified/terminal entities are not responded to (R19 hard gate). + if (!isPrEntityActionable(entity)) { + audit("pr-respond-not-actionable", `pr-respond node '${node.id}' entity ${entity.id} not actionable (unverified/terminal)`); + return { outcome: "success", value: "not-actionable" }; + } + + // Bump the rework-cycle counter (R8 cap backing; persisted). + store.updatePrEntity(entity.id, { responseRounds: entity.responseRounds + 1 }); + + if (!deps.respond) { + // U3 default: inert but routable. U5 wires the real review-response run. + return { outcome: "success", value: "disagreed-only" }; + } + + let result: PrRespondCallResult; + try { + result = await deps.respond({ task: ctx.task, node, entity, context: ctx.context }); + } catch (err) { + const reason = classifyError(err); + audit("pr-respond-error", `pr-respond node '${node.id}' response run failed: ${reason}`); + return { outcome: "failure", value: "respond-error" }; + } + + return { outcome: "success", value: result.value, contextPatch: result.contextPatch }; + }; + + return { + "pr-create": prCreate, + "pr-respond": prRespond, + "pr-merge": prMerge, + }; +} diff --git a/packages/engine/src/project-engine-manager.ts b/packages/engine/src/project-engine-manager.ts index 9feef342f8..991a3ea03d 100644 --- a/packages/engine/src/project-engine-manager.ts +++ b/packages/engine/src/project-engine-manager.ts @@ -38,6 +38,7 @@ export interface EngineManagerOptions { processPullRequestMerge?: ProjectEngineOptions["processPullRequestMerge"]; createGroupPr?: ProjectEngineOptions["createGroupPr"]; syncGroupPr?: ProjectEngineOptions["syncGroupPr"]; + prNodeGithubOps?: ProjectEngineOptions["prNodeGithubOps"]; getTaskMergeBlocker?: ProjectEngineOptions["getTaskMergeBlocker"]; onInsightRunProcessed?: ProjectEngineOptions["onInsightRunProcessed"]; } @@ -485,6 +486,7 @@ export class ProjectEngineManager { processPullRequestMerge: this.options.processPullRequestMerge, createGroupPr: this.options.createGroupPr, syncGroupPr: this.options.syncGroupPr, + prNodeGithubOps: this.options.prNodeGithubOps, 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 a379a234fc..9992b29c75 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -17,6 +17,7 @@ import { InProcessRuntime } from "./runtimes/in-process-runtime.js"; 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 { PrCommentHandler } from "./pr-comment-handler.js"; import { NtfyNotifier } from "./notifier.js"; import { NotificationService, OAuthAlertStateStore, OAuthExpiryMonitor, OAuthValidityLogger } from "./notification/index.js"; @@ -220,6 +221,15 @@ export interface ProjectEngineOptions { * the PR body. */ syncGroupPr?: SyncGroupPrFn; + /** + * PR-entity node GitHub ops (U3): the injected `createPr`/`mergePr`/`respond` + * callbacks (+ source resolver + audit) that back the `pr-create`/`pr-respond`/ + * `pr-merge` workflow nodes. Injected from the CLI layer because they close + * over the dashboard `GitHubClient`; the engine must not statically import it + * (FN-3049). Mirrors `createGroupPr`/`syncGroupPr`. When absent, the pr-* node + * kinds fail closed (value:"pr-nodes-unwired"). + */ + prNodeGithubOps?: PrNodeGithubOps; /** * Returns the merge blocker reason for a task, or null/undefined if * the task is eligible for merge. Imported from @fusion/core. @@ -364,10 +374,14 @@ export class ProjectEngine { centralCore: CentralCore, private options: ProjectEngineOptions = {}, ) { - // Pass through externalTaskStore to the runtime config if provided - const runtimeConfig: ProjectRuntimeConfig = options.externalTaskStore - ? { ...config, externalTaskStore: options.externalTaskStore } - : config; + // Pass through externalTaskStore + PR node GitHub ops (U3) to the runtime + // config. The runtime binds the engine-owned store and hands the assembled + // PrNodeDeps to the executor's workflow-graph runner. + const runtimeConfig: ProjectRuntimeConfig = { + ...config, + ...(options.externalTaskStore ? { externalTaskStore: options.externalTaskStore } : {}), + ...(options.prNodeGithubOps ? { prNodeGithubOps: options.prNodeGithubOps } : {}), + }; this.runtime = new InProcessRuntime(runtimeConfig, centralCore); // Let the runtime's SelfHealingManager re-enqueue tasks directly into our // auto-merge queue when it clears a stale `merging` status, instead of diff --git a/packages/engine/src/project-runtime.ts b/packages/engine/src/project-runtime.ts index 1c09ee07c9..00c7b9a960 100644 --- a/packages/engine/src/project-runtime.ts +++ b/packages/engine/src/project-runtime.ts @@ -55,6 +55,14 @@ export interface ProjectRuntimeConfig { * Useful when the caller (e.g. dashboard.ts) owns and watches the store. */ externalTaskStore?: TaskStore; + /** + * PR-entity node GitHub ops (U3): the injected `createPr`/`mergePr`/`respond` + * callbacks (+ source resolver + audit) for the `pr-create`/`pr-respond`/ + * `pr-merge` workflow nodes. Threaded from the CLI layer; the runtime binds the + * engine-owned store and hands the assembled deps to the executor. Absent → the + * pr-* node kinds fail closed. + */ + prNodeGithubOps?: import("./pr-nodes.js").PrNodeGithubOps; } /** diff --git a/packages/engine/src/runtimes/in-process-runtime.ts b/packages/engine/src/runtimes/in-process-runtime.ts index da1639bcff..8eae3d7663 100644 --- a/packages/engine/src/runtimes/in-process-runtime.ts +++ b/packages/engine/src/runtimes/in-process-runtime.ts @@ -17,6 +17,7 @@ import { Scheduler } from "../scheduler.js"; import type { PrMonitor, PrComment } from "../pr-monitor.js"; import type { PrInfo } from "@fusion/core"; import { TaskExecutor, type TaskExecutorOptions } from "../executor.js"; +import { buildPrNodeDeps } from "../pr-nodes.js"; import { WorktreePool, isGitRepository, type PoolInvariantViolation } from "../worktree-pool.js"; import { AgentSemaphore } from "../concurrency.js"; import { HeartbeatMonitor, HeartbeatTriggerScheduler, type WakeContext } from "../agent-heartbeat.js"; @@ -442,6 +443,7 @@ export class InProcessRuntime } } + const prNodeGithubOps = this.config.prNodeGithubOps; const executorOptions: TaskExecutorOptions = { semaphore: this.globalSemaphore, pool: this.worktreePool, @@ -451,6 +453,13 @@ export class InProcessRuntime messageStore: this.messageStore, missionStore, reflectionService, + // PR-entity nodes (U3): assemble the handler deps from the CLI-injected + // GitHub ops (createPr/mergePr/respond) + the engine-owned store. The CLI + // layer never holds a store reference; the engine binds it here. Absent + // ops → undefined → the pr-* node kinds fail closed. + prNodes: prNodeGithubOps + ? buildPrNodeDeps(() => this.taskStore, prNodeGithubOps) + : undefined, onSliceComplete: (slice) => { void this.scheduler.onSliceComplete(slice); }, diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 1649dfa917..23f29bfc5d 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -11,6 +11,7 @@ import { type WorkflowCustomNodeRunner, type WorkflowLegacySeams, } from "./workflow-node-handlers.js"; +import type { PrNodeDeps } from "./pr-nodes.js"; import { runSplitJoin, type BranchEnvironment, @@ -56,6 +57,9 @@ export interface WorkflowGraphExecutorDeps { /** Step-inversion (U14, KTD-15): runner for the `code` node (esbuild compile + * child-process execution). Absent → a code node fails cleanly. */ runCode?: CodeNodeRunner; + /** PR-entity nodes (U3): deps for `pr-create`/`pr-respond`/`pr-merge` (injected + * GitHub callbacks + store accessor). Absent → the pr-* kinds fail cleanly. */ + prNodes?: PrNodeDeps; maxRetriesPerNode?: number; /** Per-branch run-state persistence (U13). Optional — fully in-memory without it. */ branchPersistence?: WorkflowBranchPersistence; @@ -145,6 +149,7 @@ export class WorkflowGraphExecutor { ...createDefaultNodeHandlers(deps.seams ?? createNoopLegacySeams(), deps.runCustomNode, { parseSteps: deps.parseStepsDeps, runCode: deps.runCode, + prNodes: deps.prNodes, }), ...(deps.handlers ?? {}), }; diff --git a/packages/engine/src/workflow-graph-task-runner.ts b/packages/engine/src/workflow-graph-task-runner.ts index 5cbaf23fb3..c879f6e2dd 100644 --- a/packages/engine/src/workflow-graph-task-runner.ts +++ b/packages/engine/src/workflow-graph-task-runner.ts @@ -15,6 +15,7 @@ import type { WorkflowBranchSemaphore, } from "./workflow-graph-branches.js"; import type { ForeachEnvironment, WorkflowStepInstancePersistence } from "./workflow-graph-foreach.js"; +import type { PrNodeDeps } from "./pr-nodes.js"; // (Both types are also used as values in the side-effect tracking wrappers below.) /** @@ -69,6 +70,9 @@ export interface WorkflowGraphTaskRunnerDeps { /** Step-inversion (U14, KTD-15): `code` node runner. Additive; a workflow with * no code node never invokes it. */ runCode?: CodeNodeRunner; + /** PR-entity nodes (U3): deps for `pr-create`/`pr-respond`/`pr-merge`. Additive; + * a workflow with no pr-* node never invokes them; absent → they fail closed. */ + prNodes?: PrNodeDeps; /** Step-inversion (KTD-11, U10): worktree-isolation + parallel-scheduling deps. * Additive; a shared-isolation foreach never invokes them. */ allocateInstanceWorktree?: ForeachEnvironment["allocateInstanceWorktree"]; @@ -192,6 +196,7 @@ export class WorkflowGraphTaskRunner { onReworkReset: this.deps.onReworkReset, parseStepsDeps: this.deps.parseStepsDeps, runCode: this.deps.runCode, + prNodes: this.deps.prNodes, // Step-inversion (KTD-11, U10): worktree isolation + parallel scheduling. allocateInstanceWorktree: this.deps.allocateInstanceWorktree, resolveIntegrationBase: this.deps.resolveIntegrationBase, diff --git a/packages/engine/src/workflow-node-handlers.ts b/packages/engine/src/workflow-node-handlers.ts index 40ab61869f..f867a18136 100644 --- a/packages/engine/src/workflow-node-handlers.ts +++ b/packages/engine/src/workflow-node-handlers.ts @@ -2,6 +2,7 @@ import { WorkflowIrError, getStepParser } from "@fusion/core"; import type { TaskDetail, TaskStep, WorkflowIrNode } from "@fusion/core"; import type { WorkflowNodeHandler, WorkflowNodeResult } from "./workflow-graph-executor.js"; +import { createPrNodeHandlers, type PrNodeDeps } from "./pr-nodes.js"; export type WorkflowSeamName = "planning" | "execute" | "review" | "merge" | "schedule" | "step-execute"; @@ -517,6 +518,8 @@ export interface DefaultNodeHandlerDeps { parseSteps?: ParseStepsHandlerDeps; /** code node runner (U14). When absent, a code node fails cleanly. */ runCode?: CodeNodeRunner; + /** PR node deps (U3). When absent, the three pr-* kinds fail cleanly. */ + prNodes?: PrNodeDeps; } export function createDefaultNodeHandlers( @@ -524,7 +527,15 @@ export function createDefaultNodeHandlers( runCustomNode?: WorkflowCustomNodeRunner, deps?: DefaultNodeHandlerDeps, ): Record< - "prompt" | "script" | "gate" | "step-review" | "parse-steps" | "code", + | "prompt" + | "script" + | "gate" + | "step-review" + | "parse-steps" + | "code" + | "pr-create" + | "pr-respond" + | "pr-merge", WorkflowNodeHandler > { const promptLike = createPromptLikeHandler(seams, runCustomNode); @@ -533,6 +544,16 @@ export function createDefaultNodeHandlers( const parseSteps: WorkflowNodeHandler = deps?.parseSteps ? createParseStepsHandler(deps.parseSteps) : async () => ({ outcome: "failure", value: "parse-steps-unwired" }); + // PR nodes without deps fail closed (mirrors parse-steps): a pr-* node reached + // without GitHub wiring must NOT silently succeed — it would route an + // unverified PR side effect forward. + const prNodes: Record<"pr-create" | "pr-respond" | "pr-merge", WorkflowNodeHandler> = deps?.prNodes + ? createPrNodeHandlers(deps.prNodes) + : { + "pr-create": async () => ({ outcome: "failure", value: "pr-nodes-unwired" }), + "pr-respond": async () => ({ outcome: "failure", value: "pr-nodes-unwired" }), + "pr-merge": async () => ({ outcome: "failure", value: "pr-nodes-unwired" }), + }; return { prompt: promptLike, script: promptLike, @@ -540,6 +561,7 @@ export function createDefaultNodeHandlers( "step-review": createStepReviewHandler(seams), "parse-steps": parseSteps, code: createCodeNodeHandler(deps?.runCode), + ...prNodes, }; }