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) } 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 } : {}), + } + }) +}