From 677ea22da59c3ebc2eed58613c810b7bdc88cef9 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Mon, 24 Aug 2026 02:35:49 -0700 Subject: [PATCH 1/2] fix(graph): make a resume release dead inline reservations and let abort reach a parked run A resumed run held the budget reservation of every spawned child that never settled. An inline executor runs inside the process that spawned it, so a resume can prove it dead. The held reservation starved the pool and refused the next metered node. uncertainSpawnBudgets now keeps only runtimes that can re-attach. A parked waitForWakes run also ignored the abort signal. The wait now races the abort, so a host can shut a suspended run down. --- src/runtime/graph/scheduler.ts | 9 ++++++++- src/runtime/supervise/supervisor.ts | 9 ++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/src/runtime/graph/scheduler.ts b/src/runtime/graph/scheduler.ts index 42e6ccbd..7f4281b5 100644 --- a/src/runtime/graph/scheduler.ts +++ b/src/runtime/graph/scheduler.ts @@ -752,9 +752,16 @@ async function runGraphLoop( }) } rearmWakeSignal() + // An abort must reach a PARKED run too: nothing is live, no wake is coming, and the host is + // shutting the run down — the wait ends now and the loop exits on the aborted signal. + const abortWake = new Promise((fire) => { + if (abort.signal.aborted) fire() + else abort.signal.addEventListener('abort', () => fire(), { once: true }) + }) let pendingNext: Promise | null> | undefined while (!failure) { + if (abort.signal.aborted) break if (await expireDue()) continue if (wakes.length > 0) { await drainWakes() @@ -768,7 +775,7 @@ async function runGraphLoop( if (parked.length === 0) break // stuck or complete: `assembleGraphResult` classifies it if (options.waitForWakes) { rearmWakeSignal() - await wakeSignal + await Promise.race([wakeSignal, abortWake]) continue } // Offline (#976): no host will answer, so a `default` suspension resolves now; `wait` and a diff --git a/src/runtime/supervise/supervisor.ts b/src/runtime/supervise/supervisor.ts index 347eaa5d..6950b3a5 100644 --- a/src/runtime/supervise/supervisor.ts +++ b/src/runtime/supervise/supervisor.ts @@ -351,7 +351,14 @@ export function uncertainSpawnBudgets(events: SpawnEvent[]): Budget[] { return events .filter( (event): event is SpawnedEvent => - event.kind === 'spawned' && event.parent !== undefined && !terminal.has(event.id), + event.kind === 'spawned' && + event.parent !== undefined && + !terminal.has(event.id) && + // An `inline` executor runs inside the process that spawned it, so a resume can prove it + // dead rather than in-doubt: holding its reservation would charge the pool for work no + // process can ever finish. Only a runtime that can re-attach across a process boundary + // (bridge, sandbox) keeps its reservation charged as uncertain. + event.runtime !== 'inline', ) .map((event) => event.budget) } From 9c9c7524534a8757691cf9213f26a2d4c28a890f Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Mon, 24 Aug 2026 02:35:49 -0700 Subject: [PATCH 2/2] feat(graph): prove ADC's pr-review-with-approval workflow runs on the engine Lower the real template to an EngineGraphSpec with two host kinds. agent.run is metered into the conserved pool. decision parks the run; the human answer arrives as a resume. The if guard becomes the edge guard on the decision's data edge. The spike proves: approve and timeout paths, the park across a process restart, kill-anywhere replay at every journal boundary over host kinds, the pool cap acting as maxRunCostUsd, and the settles projected onto the actionResults shape the ADC run detail UI reads. An external effect is at-least-once across a crash; the fake host models the idempotency key that makes delivery exactly-once. --- tests/graph/adc-workflow-spike.test.ts | 438 ++++++++++++++++++++++++ tests/graph/fixtures/adc-workflow.ts | 446 +++++++++++++++++++++++++ 2 files changed, 884 insertions(+) create mode 100644 tests/graph/adc-workflow-spike.test.ts create mode 100644 tests/graph/fixtures/adc-workflow.ts diff --git a/tests/graph/adc-workflow-spike.test.ts b/tests/graph/adc-workflow-spike.test.ts new file mode 100644 index 00000000..0d09e0fb --- /dev/null +++ b/tests/graph/adc-workflow-spike.test.ts @@ -0,0 +1,438 @@ +/** + * The ADC integration spike, executed: agent-dev-container's real `pr-review-with-approval` + * template runs on the engine — approve and timeout paths, the human park across a process + * restart, kill-anywhere durability over HOST kinds (a settled agent run or posted review is + * never re-executed), the conserved pool capping the run the way `maxRunCostUsd` does, and the + * engine settles projected back onto the `actionResults` shape ADC's UI reads. + */ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterAll, describe, expect, it, vi } from 'vitest' +import { contentAddress } from '../../src/durable/content-address' +import { FileSpawnJournal, InMemoryResultBlobStore } from '../../src/durable/spawn-journal' +import { + agentKind, + createGraphEngine, + createGraphRun, + type GraphEngine, + runEngineGraph, + scriptKind, + subgraphKind, + supervisorKind, +} from '../../src/runtime/graph' +import type { SpawnEvent, SpawnJournal } from '../../src/runtime/supervise/types' +import { integrationInvokeKind } from './fixtures/adc-kinds' +import { + type AdcRunAgent, + actionResultsFromRun, + adcAgentRunKind, + adcDecisionKind, + lowerPrReviewWithApproval, + type PrTrigger, +} from './fixtures/adc-workflow' + +const REVIEW_TEXT = 'LGTM overall; one nit in src/scheduler.ts:41.' +const AGENT_COST = { costUsd: 0.42, inputTokens: 9_000, outputTokens: 1_200 } + +const trigger: PrTrigger = { + payload: { + pull_request: { number: 1417, title: 'fix: journal the woken seq' }, + repository: { + full_name: 'tangle-network/agent-runtime', + name: 'agent-runtime', + owner: { login: 'tangle-network' }, + }, + }, +} + +function fakeHost() { + const runAgent: AdcRunAgent = { + run: vi.fn(async () => ({ finalMessage: REVIEW_TEXT, ...AGENT_COST })), + } + // A crash between the provider call and its journaled settle replays the call: the engine is + // at-least-once at an external effect. The idempotency key is what makes delivery exactly-once + // — this fake models the key ADC's hub executor must accept, derived from the request. + const posted = new Map() + const integrations = { + deliveries: () => posted.size, + invoke: vi.fn(async (connector: string, operation: string, args: unknown) => { + const key = JSON.stringify([connector, operation, args]) + const existing = posted.get(key) + if (existing !== undefined) return existing + const response = { id: 991, state: 'COMMENTED' } + posted.set(key, response) + return response + }), + } + return { runAgent, integrations } +} + +function engine(host: ReturnType): GraphEngine { + return createGraphEngine({ + coreKinds: [ + agentKind({}), + supervisorKind({ + blobs: new InMemoryResultBlobStore(), + makeWorkerAgent: () => ({ name: 'x', act: async () => 1 }), + }), + scriptKind(), + subgraphKind(), + ], + kinds: [adcAgentRunKind(), adcDecisionKind(), integrationInvokeKind()], + effects: { runAgent: host.runAgent, integrations: host.integrations }, + }) +} + +const budget = { maxIterations: 40, maxTokens: 100_000, maxUsd: 5 } +const perNode = { maxIterations: 5, maxTokens: 50_000, maxUsd: 1 } + +const dirs: string[] = [] +afterAll(() => { + for (const dir of dirs) rmSync(dir, { recursive: true, force: true }) +}) +const journalPath = () => { + const dir = mkdtempSync(join(tmpdir(), 'adc-spike-')) + dirs.push(dir) + return join(dir, 'journal.jsonl') +} + +const EXPECTED_POST = { + owner: 'tangle-network', + repo: 'agent-runtime', + pull_number: 1417, + event: 'COMMENT', + body: REVIEW_TEXT, +} + +class KillError extends Error {} + +/** Allows the first `limit` appends, then kills the process stand-in at the boundary. */ +/** + * The park is observable only through the journal — the run handle has no "parked" signal, so a + * host does what ADC's decision store does: watch for the durable `waiting` event and read the + * token FROM it. Tokens are minted per VISIT: a crash before the park was journaled re-enters + * the node as a new visit with a new token, so a precomputed token is a stale ask. + */ +async function pendingParkToken(path: string, runId: string): Promise { + const events = (await new FileSpawnJournal(path).loadTree(runId)) ?? [] + const waiting = new Map() + const woken = new Set() + for (const event of events) { + if (event.kind === 'waiting' && event.spec.kind === 'token') { + waiting.set(event.id, event.spec.token) + } + if (event.kind === 'woken') woken.add(event.id) + } + for (const [id, token] of waiting) if (!woken.has(id)) return token + return undefined +} + +async function awaitParkToken(path: string, runId: string): Promise { + const deadline = Date.now() + 5_000 + while (Date.now() < deadline) { + const token = await pendingParkToken(path, runId) + if (token !== undefined) return token + await new Promise((resolve) => setTimeout(resolve, 20)) + } + throw new Error(`no pending park for ${runId} within 5s`) +} + +class KillingJournal implements SpawnJournal { + appends = 0 + constructor( + private readonly inner: SpawnJournal, + private readonly limit: number, + ) {} + loadTree(root: string) { + return this.inner.loadTree(root) + } + beginTree(root: string, at: string) { + return this.inner.beginTree(root, at) + } + async appendEvent(root: string, ev: SpawnEvent): Promise { + if (this.appends >= this.limit) + throw new KillError(`killed at journal append ${this.appends + 1}`) + this.appends += 1 + return this.inner.appendEvent(root, ev) + } +} + +describe('pr-review-with-approval on the engine', () => { + it('approve path: the human wake releases the guarded edge and the review is posted once', async () => { + const host = fakeHost() + const path = journalPath() + const run = createGraphRun(engine(host), lowerPrReviewWithApproval(trigger), 'review PR 1417', { + budget, + perNode, + journal: new FileSpawnJournal(path), + blobs: new InMemoryResultBlobStore(), + runId: 'approve-run', + waitForWakes: true, + finalizer: 'collectDelivered', + }) + + // The host reads the token from the durable `waiting` event; on an uninterrupted run it is + // also recomputable from run identity alone (visit 1 of the decision node). + const token = await awaitParkToken(path, 'approve-run') + expect(token).toBe( + contentAddress({ runId: 'approve-run', instance: 'step-2#1', kind: 'graph-suspension' }), + ) + await run.resume(token, { choice: 'approve', note: 'ship it' }) + + const result = await run.done + expect(result.kind).toBe('winner') + expect(host.runAgent.run).toHaveBeenCalledTimes(1) + expect(host.integrations.invoke).toHaveBeenCalledTimes(1) + expect(host.integrations.invoke).toHaveBeenCalledWith( + 'github', + 'pulls.reviews.create', + EXPECTED_POST, + ) + const settles = result.kind === 'winner' ? result.settles : [] + const decision = settles.find((settle) => settle.node === 'step-2') + expect(decision?.out).toEqual({ choice: 'approve', note: 'ship it' }) + }) + + it('timeout path: offline, the decision resolves to its default and the post is skipped — the run still completes', async () => { + const host = fakeHost() + const result = await runEngineGraph( + engine(host), + lowerPrReviewWithApproval(trigger), + 'review PR 1417', + { + budget, + perNode, + journal: new FileSpawnJournal(journalPath()), + blobs: new InMemoryResultBlobStore(), + runId: 'timeout-run', + finalizer: 'collectDelivered', + }, + ) + expect(result.kind).toBe('winner') + expect(host.runAgent.run).toHaveBeenCalledTimes(1) + expect(host.integrations.invoke).not.toHaveBeenCalled() + const settles = result.kind === 'winner' ? result.settles : [] + expect(settles.find((settle) => settle.node === 'step-2')?.out).toEqual({ + choice: 'reject', + timedOut: true, + }) + expect(settles.find((settle) => settle.node === 'step-3')).toBeUndefined() + + const rows = actionResultsFromRun(result) + expect(rows.map((row) => [row.nodeId, row.status])).toEqual([ + ['step-1', 'succeeded'], + ['step-2', 'succeeded'], + ['step-3', 'skipped'], + ]) + expect(rows[0]?.costUsd).toBe(AGENT_COST.costUsd) + }) + + it('kill-anywhere: at every journal boundary, a restart re-runs no settled host call — the agent never re-runs, the review never double-posts', async () => { + // Phase 0: the uninterrupted approve-path run — reference bytes and the append count. + const referenceHost = fakeHost() + const referencePath = journalPath() + const referenceJournal = new KillingJournal( + new FileSpawnJournal(referencePath), + Number.MAX_SAFE_INTEGER, + ) + const reference = createGraphRun( + engine(referenceHost), + lowerPrReviewWithApproval(trigger), + 'review PR 1417', + { + budget, + perNode, + journal: referenceJournal, + blobs: new InMemoryResultBlobStore(), + runId: 'kill-run', + waitForWakes: true, + finalizer: 'collectDelivered', + }, + ) + await reference.resume(await awaitParkToken(referencePath, 'kill-run'), { choice: 'approve' }) + const referenceResult = await reference.done + expect(referenceResult.kind).toBe('winner') + const referenceBytes = JSON.stringify( + referenceResult.kind === 'winner' ? referenceResult.out : undefined, + ) + const totalAppends = referenceJournal.appends + expect(totalAppends).toBeGreaterThan(10) + + for (let kill = 1; kill < totalAppends; kill += 1) { + const host = fakeHost() + const path = journalPath() + const blobs = new InMemoryResultBlobStore() + const spec = lowerPrReviewWithApproval(trigger) + const firstAbort = new AbortController() + const first = createGraphRun(engine(host), spec, 'review PR 1417', { + budget, + perNode, + journal: new KillingJournal(new FileSpawnJournal(path), kill), + blobs, + runId: 'kill-run', + waitForWakes: true, + finalizer: 'collectDelivered', + signal: firstAbort.signal, + }) + // The kill can land before the park, after it, or INSIDE the wake processing. The host + // flow is always: observe the durable waiting event, wake ITS token. When the kill eats + // the wake itself the run stays parked, so the process stand-in is shot down with the + // abort signal, the way a dead host process takes its run down. + const firstOutcome = first.done.then( + () => 'completed' as const, + (error) => (error instanceof KillError ? ('killed' as const) : ('down' as const)), + ) + const raced = await Promise.race([ + firstOutcome, + awaitParkToken(path, 'kill-run').then( + (token) => ({ token }), + () => 'no-park' as const, + ), + ]) + let killedInWake = false + if (typeof raced === 'object') { + const wakeError = await first.resume(raced.token, { choice: 'approve' }).then( + () => undefined, + (error: unknown) => error, + ) + killedInWake = wakeError instanceof KillError + if (killedInWake) firstAbort.abort('wake path killed') + } + const ended = await firstOutcome + expect( + ended === 'killed' || killedInWake, + `boundary ${kill} should kill (ended ${ended})`, + ).toBe(true) + + const agentCallsBeforeRestart = host.runAgent.run.mock.calls.length + const postCallsBeforeRestart = host.integrations.invoke.mock.calls.length + + // Which nodes SETTLED before the kill — their host calls may never happen again. + const journaled = (await new FileSpawnJournal(path).loadTree('kill-run')) ?? [] + const spawnedLabels = new Map( + journaled.flatMap((ev) => (ev.kind === 'spawned' ? [[ev.id, ev.label]] : [])), + ) + const settledNodes = new Set( + journaled.flatMap((ev) => + ev.kind === 'settled' && ev.id !== 'kill-run' + ? [String(spawnedLabels.get(ev.id) ?? '').split('#')[0] ?? ''] + : [], + ), + ) + + const restarted = createGraphRun(engine(host), spec, 'review PR 1417', { + budget, + perNode, + journal: new FileSpawnJournal(path), + blobs, + runId: 'kill-run', + resume: true, + waitForWakes: true, + finalizer: 'collectDelivered', + }) + // The restart may need a fresh wake — for the ORIGINAL token (park survived), or a NEW one + // (the crash landed before the park was durable, so the re-entered visit re-minted) — or + // none at all (the woken event survived). The journal, not a guess, says which. + const restartRace = await Promise.race([ + restarted.done.then(() => 'done' as const), + awaitParkToken(path, 'kill-run').then( + (token) => ({ token }), + () => 'no-park' as const, + ), + ]) + if (typeof restartRace === 'object') { + await restarted.resume(restartRace.token, { choice: 'approve' }).catch((error) => { + expect(String(error), `boundary ${kill} re-wake`).toMatch(/completed|already/) + }) + } + const resumed = await restarted.done + expect(resumed.kind, `boundary ${kill} result`).toBe('winner') + expect( + JSON.stringify(resumed.kind === 'winner' ? resumed.out : undefined), + `boundary ${kill} bytes`, + ).toBe(referenceBytes) + + if (settledNodes.has('step-1')) { + expect( + host.runAgent.run.mock.calls.length, + `boundary ${kill}: settled agent run re-executed`, + ).toBe(agentCallsBeforeRestart) + } + if (settledNodes.has('step-3')) { + expect( + host.integrations.invoke.mock.calls.length, + `boundary ${kill}: posted review re-posted`, + ).toBe(postCallsBeforeRestart) + } + // The absolute law regardless of where the kill landed: at most one DELIVERED post. A + // kill between the provider call and its settle may re-CALL (at-least-once), and the + // idempotency key collapses it to one delivery — the contract an ADC lift must keep. + expect( + host.integrations.deliveries(), + `boundary ${kill}: delivered posts`, + ).toBeLessThanOrEqual(1) + } + }, 180_000) + + it("budget cap: a pool smaller than the agent's spend fails the run the way maxRunCostUsd does", async () => { + const host = fakeHost() + const result = await runEngineGraph( + engine(host), + lowerPrReviewWithApproval(trigger), + 'review PR 1417', + { + budget: { maxIterations: 40, maxTokens: 100_000, maxUsd: 0.1 }, + perNode: { maxIterations: 5, maxTokens: 50_000, maxUsd: 0.1 }, + journal: new FileSpawnJournal(journalPath()), + blobs: new InMemoryResultBlobStore(), + runId: 'capped-run', + finalizer: 'collectDelivered', + }, + ) + expect(result.kind).toBe('no-winner') + expect(host.integrations.invoke).not.toHaveBeenCalled() + }) + + it('UI projection: the approve-path settles map onto the actionResults rows the run detail reads', async () => { + const host = fakeHost() + const path = journalPath() + const run = createGraphRun(engine(host), lowerPrReviewWithApproval(trigger), 'review PR 1417', { + budget, + perNode, + journal: new FileSpawnJournal(path), + blobs: new InMemoryResultBlobStore(), + runId: 'ui-run', + waitForWakes: true, + finalizer: 'collectDelivered', + }) + const uiToken = await awaitParkToken(path, 'ui-run') + await run.resume(uiToken, { choice: 'approve' }) + const result = await run.done + const rows = actionResultsFromRun(result) + expect(rows).toEqual([ + { + index: 0, + kind: 'agent.run', + nodeId: 'step-1', + status: 'succeeded', + output: { finalMessage: REVIEW_TEXT, costUsd: AGENT_COST.costUsd }, + costUsd: AGENT_COST.costUsd, + }, + { + index: 1, + kind: 'decision', + nodeId: 'step-2', + status: 'succeeded', + output: { choice: 'approve' }, + }, + { + index: 2, + kind: 'integration.invoke', + nodeId: 'step-3', + status: 'succeeded', + output: { id: 991, state: 'COMMENTED' }, + }, + ]) + }) +}) diff --git a/tests/graph/fixtures/adc-workflow.ts b/tests/graph/fixtures/adc-workflow.ts new file mode 100644 index 00000000..201c711a --- /dev/null +++ b/tests/graph/fixtures/adc-workflow.ts @@ -0,0 +1,446 @@ +/** + * The ADC substrate-swap spike: agent-dev-container's `pr-review-with-approval` workflow template + * (products/platform/api/src/lib/workflow-templates.ts), hand-lowered to the `EngineGraphSpec` + * its authoring compiler would emit. This is the proof that the engine can be the execution + * substrate under ADC's workflow product while ADC keeps its front-end and business kinds: + * + * - `agent.run` → a METERED host kind over ADC's `runAgent` dep; its spend settles into the + * one conserved pool, so ADC's `maxRunCostUsd` becomes `Budget.maxUsd`. + * - `decision` → a host kind that parks: the node settles a `SuspensionRequest`, the human + * answer arrives as `resume(token, payload)`, and ADC's `onTimeout: default` + * maps to the suspension's `onExpire: 'default'`. + * - `${steps.x}` → `data` edges; a single-field read is an edge projection, and a template + * assembling several sources is a pure `script` node (agent-runtime#971). + * - `if:` guards → the edge `guard` tree, which the engine adopted from ADC verbatim. + * + * The lowering targets the template's meaning, not its YAML: node ids keep ADC's positional + * `step-N` names so `${steps[N]}` and the UI's `nodeId` column stay recognizable. + */ + +import { contentAddress } from '../../../src/durable/content-address' +import { ValidationError } from '../../../src/errors' +import type { + EngineGraphSpec, + GraphNodeSettle, + GraphRunResult, + NodeKind, +} from '../../../src/runtime/graph' +import { suspended } from '../../../src/runtime/graph' +import type { + Agent, + AgentSpec, + Executor, + ExecutorResult, +} from '../../../src/runtime/supervise/types' + +/** What ADC's `deps.runAgent` looks like to the lowered `agent.run` node. */ +export interface AdcRunAgent { + run(request: AgentRunRequest): Promise +} + +export interface AgentRunRequest { + readonly profile: string + readonly prompt: string + readonly source: { readonly repo: string; readonly pr: number } + readonly maxRounds: number +} + +/** The slice of ADC's `RunAgentResult` the workflow surface reads. */ +export interface AgentRunOutcome { + readonly finalMessage: string + readonly costUsd: number + readonly inputTokens: number + readonly outputTokens: number +} + +export interface AgentRunKindConfig { + readonly profile: string + readonly maxRounds: number +} + +export interface DecisionKindConfig { + readonly title: string + readonly options: ReadonlyArray + readonly timeoutMs: number + /** ADC's `onTimeout: default` choice — the only offline-resolvable timeout policy. */ + readonly defaultChoice: string +} + +function asRecord(raw: unknown, context: string): Readonly> { + if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) { + throw new ValidationError(`${context}: config must be an object`) + } + return raw as Readonly> +} + +function requireString( + record: Readonly>, + key: string, + context: string, +): string { + const value = record[key] + if (typeof value !== 'string' || value.length === 0) { + throw new ValidationError(`${context}: ${key} must be a non-empty string`) + } + return value +} + +function requirePositiveInt( + record: Readonly>, + key: string, + context: string, +): number { + const value = record[key] + if (!Number.isSafeInteger(value) || (value as number) < 1) { + throw new ValidationError(`${context}: ${key} must be a positive integer`) + } + return value as number +} + +/** A leaf whose body is host code and whose spend is REAL: metered into the conserved pool. */ +function meteredLeaf( + name: string, + nodeKind: string, + body: ( + signal: AbortSignal, + ) => Promise<{ out: unknown; costUsd: number; inputTokens: number; outputTokens: number }>, +): Agent & { executorSpec: AgentSpec } { + let artifact: ExecutorResult | undefined + const executor: Executor = { + runtime: 'inline', + async execute(_task, signal) { + const { out, costUsd, inputTokens, outputTokens } = await body(signal) + artifact = { + outRef: contentAddress(out), + out, + spent: { + iterations: 1, + tokens: { input: inputTokens, output: outputTokens }, + usd: costUsd, + ms: 0, + }, + } + return artifact + }, + teardown: () => Promise.resolve({ destroyed: true }), + resultArtifact: () => { + if (!artifact) + throw new ValidationError(`${nodeKind}: resultArtifact() read before execute()`) + return artifact + }, + } + return { + name, + act: () => Promise.reject(new ValidationError(`${nodeKind}: act() is not the execution path`)), + executorSpec: { + profile: { name }, + harness: null, + executor, + execution: { correlation: { nodeKind } }, + }, + } +} + +/** A budget-exempt leaf, for host code that spends nothing (the decision park). */ +function exemptLeaf( + name: string, + nodeKind: string, + body: () => Promise, +): Agent & { executorSpec: AgentSpec } { + let artifact: ExecutorResult | undefined + const executor: Executor = { + runtime: 'inline', + budgetExempt: true, + async execute() { + const out = await body() + artifact = { + outRef: contentAddress(out), + out, + spent: { iterations: 0, tokens: { input: 0, output: 0 }, usd: 0, ms: 0 }, + } + return artifact + }, + teardown: () => Promise.resolve({ destroyed: true }), + resultArtifact: () => { + if (!artifact) + throw new ValidationError(`${nodeKind}: resultArtifact() read before execute()`) + return artifact + }, + } + return { + name, + act: () => Promise.reject(new ValidationError(`${nodeKind}: act() is not the execution path`)), + executorSpec: { + profile: { name }, + harness: null, + executor, + execution: { correlation: { nodeKind } }, + }, + } +} + +/** ADC's `agent.run`, lowered: one run through the host's `runAgent` dep, spend metered. */ +export function adcAgentRunKind(): NodeKind { + return { + id: 'adc.agent.run', + version: 1, + description: 'Run one ADC agent through the host runAgent dependency; spend is metered.', + validateConfig: (raw, context) => { + const record = asRecord(raw, context) + return { + profile: requireString(record, 'profile', context), + maxRounds: requirePositiveInt(record, 'maxRounds', context), + } + }, + configSchema: { + type: 'object', + properties: { profile: { type: 'string' }, maxRounds: { type: 'number' } }, + required: ['profile', 'maxRounds'], + additionalProperties: false, + }, + inputs: [{ name: 'request', schema: { type: 'object' } }], + outputs: [{ name: 'result', schema: { type: 'object' } }], + effects: ['runAgent'] as const, + onCrash: 'restart', + budget: 'metered', + run: ({ config, profile, inputs, effects }) => + meteredLeaf(profile.name ?? 'adc.agent.run', 'adc.agent.run/v1', async () => { + const request = inputs.request as { prompt: string; source: { repo: string; pr: number } } + const outcome = await (effects.runAgent as AdcRunAgent).run({ + profile: config.profile, + prompt: request.prompt, + source: request.source, + maxRounds: config.maxRounds, + }) + return { + out: { finalMessage: outcome.finalMessage, costUsd: outcome.costUsd }, + costUsd: outcome.costUsd, + inputTokens: outcome.inputTokens, + outputTokens: outcome.outputTokens, + } + }), + } +} + +/** ADC's `decision`, lowered: the node parks; a human answer is `resume(token, { choice })`. */ +export function adcDecisionKind(): NodeKind { + return { + id: 'adc.decision', + version: 1, + description: 'Park for a human choice; timeout resolves to the declared default.', + validateConfig: (raw, context) => { + const record = asRecord(raw, context) + const options = record.options + if ( + !Array.isArray(options) || + options.length < 2 || + options.some((option) => typeof option !== 'string' || option.length === 0) + ) { + throw new ValidationError(`${context}: options must be at least two non-empty strings`) + } + const defaultChoice = requireString(record, 'defaultChoice', context) + if (!options.includes(defaultChoice)) { + throw new ValidationError(`${context}: defaultChoice must be one of options`) + } + return { + title: requireString(record, 'title', context), + options: options as ReadonlyArray, + timeoutMs: requirePositiveInt(record, 'timeoutMs', context), + defaultChoice, + } + }, + configSchema: { + type: 'object', + properties: { + title: { type: 'string' }, + options: { type: 'array', items: { type: 'string' } }, + timeoutMs: { type: 'number' }, + defaultChoice: { type: 'string' }, + }, + required: ['title', 'options', 'timeoutMs', 'defaultChoice'], + additionalProperties: false, + }, + inputs: [{ name: 'prompt', schema: { type: 'string' } }], + outputs: [{ name: 'resolution', schema: { type: 'object' } }], + effects: [] as const, + onCrash: 'restart', + budget: 'exempt', + run: ({ config, profile }) => + exemptLeaf(profile.name ?? 'adc.decision', 'adc.decision/v1', async () => + suspended({ + onExpire: 'default', + expiresInMs: config.timeoutMs, + default: { choice: config.defaultChoice, timedOut: true }, + }), + ), + } +} + +/** The GitHub `pull_request: opened` trigger context the template consumes. */ +export interface PrTrigger { + readonly payload: { + readonly pull_request: { readonly number: number; readonly title: string } + readonly repository: { + readonly full_name: string + readonly name: string + readonly owner: { readonly login: string } + } + } +} + +/** + * `pr-review-with-approval`, lowered. The `${...}` templates become two pure script nodes (the + * compiler-emitted assemblies), the `if:` guard becomes the edge guard on the decision's data + * edge, and the decision is a forced terminal so a rejected review completes the run cleanly + * with the post skipped — ADC's semantics for a guarded-out step. + */ +export function lowerPrReviewWithApproval(trigger: PrTrigger): EngineGraphSpec { + return { + nodes: [ + { + id: 'trigger', + kind: 'script/v1', + config: { body: () => trigger, pure: true }, + }, + { + id: 'review-request', + kind: 'script/v1', + config: { + body: (inputs: Record) => { + const context = inputs.trigger as PrTrigger + return { + prompt: + `Review pull request #${context.payload.pull_request.number} ` + + 'and produce a complete structured review as your final output.', + source: { + repo: context.payload.repository.full_name, + pr: context.payload.pull_request.number, + }, + } + }, + pure: true, + }, + ports: { inputs: [{ name: 'trigger', schema: { type: 'object' } }] }, + }, + { + id: 'step-1', + kind: 'adc.agent.run/v1', + config: { profile: 'code-reviewer', maxRounds: 3 }, + }, + { + id: 'step-2', + kind: 'adc.decision/v1', + config: { + title: 'Post this PR review?', + options: ['approve', 'reject'], + timeoutMs: 24 * 60 * 60 * 1000, + defaultChoice: 'reject', + }, + terminal: true, + deliverable: { + check: (out: unknown) => typeof (out as { choice?: unknown }).choice === 'string', + describe: 'a resolved decision', + }, + }, + { + id: 'post-request', + kind: 'script/v1', + config: { + body: (inputs: Record) => { + const context = inputs.trigger as PrTrigger + const review = inputs.review as { finalMessage: string } + return { + owner: context.payload.repository.owner.login, + repo: context.payload.repository.name, + pull_number: context.payload.pull_request.number, + event: 'COMMENT', + body: review.finalMessage, + } + }, + pure: true, + }, + ports: { + inputs: [ + { name: 'trigger', schema: { type: 'object' } }, + { name: 'review', schema: { type: 'object' } }, + { name: 'approval', schema: { type: 'object' } }, + ], + }, + }, + { + id: 'step-3', + kind: 'integration.invoke/v1', + config: { connector: 'github', operation: 'pulls.reviews.create' }, + deliverable: { + check: (out: unknown) => out !== undefined, + describe: 'the posted review response', + }, + }, + ], + edges: [ + { kind: 'data', from: { node: 'trigger' }, to: { node: 'review-request', port: 'trigger' } }, + { kind: 'data', from: { node: 'review-request' }, to: { node: 'step-1', port: 'request' } }, + { + kind: 'data', + from: { node: 'step-1' }, + to: { node: 'step-2', port: 'prompt' }, + projection: { path: 'finalMessage' }, + }, + { kind: 'data', from: { node: 'trigger' }, to: { node: 'post-request', port: 'trigger' } }, + { kind: 'data', from: { node: 'step-1' }, to: { node: 'post-request', port: 'review' } }, + { + kind: 'data', + from: { node: 'step-2' }, + to: { node: 'post-request', port: 'approval' }, + guard: { path: 'out.choice', op: 'eq', value: 'approve' }, + }, + { kind: 'data', from: { node: 'post-request' }, to: { node: 'step-3', port: 'args' } }, + ], + } +} + +/** The row shape ADC's run detail UI reads (client/hooks/useWorkflowApi.ts), per template step. */ +export interface AdcActionResult { + readonly index: number + readonly kind: string + readonly nodeId: string + readonly status: 'succeeded' | 'failed' | 'skipped' | 'waiting' + readonly output?: unknown + readonly costUsd?: number +} + +const TEMPLATE_STEPS: ReadonlyArray<{ nodeId: string; kind: string }> = [ + { nodeId: 'step-1', kind: 'agent.run' }, + { nodeId: 'step-2', kind: 'decision' }, + { nodeId: 'step-3', kind: 'integration.invoke' }, +] + +/** Project an engine run onto ADC's `actionResults` — the substrate swap keeps the UI's shape. */ +export function actionResultsFromRun(result: GraphRunResult): AdcActionResult[] { + const byNode = new Map() + for (const settle of result.settles) byNode.set(settle.node, settle) + const waiting = new Set(result.kind === 'suspended' ? result.tokens.map(() => 'step-2') : []) + return TEMPLATE_STEPS.map((step, index) => { + const settle = byNode.get(step.nodeId) + if (!settle) { + return { + index, + kind: step.kind, + nodeId: step.nodeId, + status: waiting.has(step.nodeId) ? ('waiting' as const) : ('skipped' as const), + } + } + if (settle.status === 'down') { + return { index, kind: step.kind, nodeId: step.nodeId, status: 'failed' as const } + } + const costUsd = (settle.out as { costUsd?: number } | undefined)?.costUsd + return { + index, + kind: step.kind, + nodeId: step.nodeId, + status: 'succeeded' as const, + output: settle.out, + ...(typeof costUsd === 'number' ? { costUsd } : {}), + } + }) +}