|
| 1 | +import type { PlanIdentity } from '../core/identity.ts'; |
| 2 | +import type { PlanContext } from '../core/plan.ts'; |
| 3 | +import type { InvocationHandle, InvocationInput, TaskClone } from '../agents/contract.ts'; |
| 4 | +import { prepareExecution } from '../core/execution-prompt.ts'; |
| 5 | +import { auditRun, type ChangeManifest } from '../core/run-audit.ts'; |
| 6 | +import { FinishFailure, type PreparedAttempt, type RunnerCoordinator, type RunnerDeps } from './coordinator.ts'; |
| 7 | +import type { AttemptRecord, Store } from './store.ts'; |
| 8 | + |
| 9 | +/** |
| 10 | + * F2b: per-item execution and the runner's commit step (design, "How codeboost runs a plan"; plan-format.md, "After |
| 11 | + * each run"). The workspace operations are D's (#66); until they exist this module is exercised with a fake. |
| 12 | + */ |
| 13 | +export interface WorkspaceRef { readonly clone: TaskClone; readonly storage: unknown } |
| 14 | +export interface TaskWorkspace { |
| 15 | + /** A fresh task filesystem from the recorded trusted head; never a reset of a used one. Abortable; settles only when its work stopped. */ |
| 16 | + materialize(attempt: AttemptRecord, head: string, signal: AbortSignal): Promise<WorkspaceRef>; |
| 17 | + /** No-follow snapshot of every declared symlink target, taken before launch. */ |
| 18 | + snapshotDeclaredLinks(workspace: WorkspaceRef, paths: readonly string[], signal: AbortSignal): Promise<unknown>; |
| 19 | + /** The change manifest after the agent settled, plus a digest the commit step must match. */ |
| 20 | + inspectChanges(workspace: WorkspaceRef, input: { baseHead: string; linkSnapshot: unknown }, signal: AbortSignal): Promise<ChangeManifest & { digest: string }>; |
| 21 | + /** Undo agent commits (keeping changes), stage exactly `paths`, commit with hooks off; refuse if the tree no longer matches `digest`. */ |
| 22 | + commit(workspace: WorkspaceRef, input: { baseHead: string; paths: readonly string[]; message: string; trailers: Readonly<Record<string, string>>; digest: string }, signal: AbortSignal): Promise<string>; |
| 23 | + release(workspace: WorkspaceRef): Promise<void>; |
| 24 | +} |
| 25 | +/** D's start call for an execute/fix phase with this prompt; returns at once (see #51). */ |
| 26 | +export type AgentLauncher = (input: InvocationInput, prompt: string, workspace: WorkspaceRef) => InvocationHandle; |
| 27 | +/** Trusted runner-side sources for a task. Issue text and lessons are untrusted data inside the prompt. */ |
| 28 | +export interface ExecutionSources { |
| 29 | + planContext(identity: PlanIdentity): PlanContext; |
| 30 | + issue(identity: PlanIdentity): { number: number; title: string; body: string; comments: readonly string[] }; |
| 31 | + lessons(identity: PlanIdentity): readonly string[]; |
| 32 | + vendor(identity: PlanIdentity): 'claude' | 'codex'; |
| 33 | +} |
| 34 | +/** Prefix of the diagnostic for an audit safety violation; the executor moves the task to needs human on it. */ |
| 35 | +export const SAFETY_VIOLATION = 'Safety violation:'; |
| 36 | +export interface ExecutionResult { head: string; unchanged: boolean; inScope: string[]; outOfScope: string[] } |
| 37 | +interface Private { workspace: WorkspaceRef; prompt: string; baseHead: string; linkSnapshot: unknown } |
| 38 | + |
| 39 | +/** RunnerDeps for execute attempts: fresh workspace, prompt, agent, then audit and the runner's own commit. */ |
| 40 | +export function executionDeps(store: Store, workspace: TaskWorkspace, launch: AgentLauncher, sources: ExecutionSources): RunnerDeps { |
| 41 | + const identityOf = (attempt: AttemptRecord): PlanIdentity => findIdentity(store, attempt); |
| 42 | + return { |
| 43 | + async prepare(attempt, signal) { |
| 44 | + if (attempt.kind !== 'execute' || !attempt.item) throw new Error('Execution deps run execute attempts for one plan item.'); |
| 45 | + const identity = identityOf(attempt), plan = store.getPlan(identity, attempt.context.planRevision); |
| 46 | + const item = plan.items.find(entry => entry.id === attempt.item)!; |
| 47 | + const context = sources.planContext(identity); |
| 48 | + const baseHead = store.getSnapshot(identity, attempt.context.snapshotId).head; |
| 49 | + const request = prepareExecution({ identity, attemptId: attempt.id, mode: 'execute', plan, itemId: item.id, |
| 50 | + issue: sources.issue(identity), approvedLessons: sources.lessons(identity), allowedCommands: context.allowedCommands }); |
| 51 | + const ws = await workspace.materialize(attempt, baseHead, signal); |
| 52 | + const declaredLinks = item.files.map(file => file.path).filter(path => context.baseEntries.some(entry => entry.kind === 'symlink' && context.pathKey(entry.path) === context.pathKey(path))); |
| 53 | + const linkSnapshot = await workspace.snapshotDeclaredLinks(ws, declaredLinks, signal); |
| 54 | + const data: Private = { workspace: ws, prompt: request.prompt, baseHead, linkSnapshot }; |
| 55 | + return { clone: ws.clone, vendor: sources.vendor(identity), approvedArgv: request.approvedArgv, private: data }; |
| 56 | + }, |
| 57 | + async cleanupPreparation() { /* host-side files belong to D's materialize; task storage waits for release */ }, |
| 58 | + start(input, prepared) { const data = prepared.private as Private; return launch(input, data.prompt, data.workspace); }, |
| 59 | + validate() { throw new Error('Execute attempts publish through finish().'); }, |
| 60 | + async finish(attempt, _result, prepared, signal) { |
| 61 | + const data = prepared.private as Private, identity = identityOf(attempt); |
| 62 | + const plan = store.getPlan(identity, attempt.context.planRevision), item = plan.items.find(entry => entry.id === attempt.item)!; |
| 63 | + const manifest = await workspace.inspectChanges(data.workspace, { baseHead: data.baseHead, linkSnapshot: data.linkSnapshot }, signal); |
| 64 | + const outcome = auditRun(item, manifest, sources.planContext(identity).pathKey); |
| 65 | + if (outcome.kind === 'violation') throw new FinishFailure(`${SAFETY_VIOLATION} ${outcome.violations.join(' ')}`); |
| 66 | + if (outcome.unchanged) return { value: { head: data.baseHead, unchanged: true, inScope: [], outOfScope: [] } satisfies ExecutionResult }; |
| 67 | + const head = await workspace.commit(data.workspace, { |
| 68 | + baseHead: data.baseHead, paths: [...outcome.inScope, ...outcome.outOfScope], digest: manifest.digest, |
| 69 | + message: `${item.id}: ${item.title}`, trailers: { 'Plan-Item': item.id, 'Plan-Revision': `r${plan.revision}` }, |
| 70 | + }, signal); |
| 71 | + const snapshot = store.getSnapshot(identity, attempt.context.snapshotId); |
| 72 | + return { |
| 73 | + value: { head, unchanged: false, inScope: outcome.inScope, outOfScope: outcome.outOfScope } satisfies ExecutionResult, |
| 74 | + history: { base: snapshot.base, head, entries: [{ sha: head, owner: item.id, origin: 'owned', sourceSha: null }] }, |
| 75 | + }; |
| 76 | + }, |
| 77 | + async release(_attempt, prepared) { |
| 78 | + const data = prepared.private as Private; |
| 79 | + await workspace.release(data.workspace); |
| 80 | + }, |
| 81 | + }; |
| 82 | +} |
| 83 | +/** The plan identity that owns an attempt; attempts are stored per plan key. */ |
| 84 | +function findIdentity(store: Store, attempt: AttemptRecord): PlanIdentity { |
| 85 | + const key = store.attemptOwner(attempt.id); |
| 86 | + if (!key) throw new Error('Unknown attempt.'); |
| 87 | + const [repositoryId, taskId, planId] = JSON.parse(key) as string[]; |
| 88 | + return { repositoryId: repositoryId!, taskId: taskId!, planId: planId! }; |
| 89 | +} |
| 90 | + |
| 91 | +export type ExecutionOutcome = |
| 92 | + | { kind: 'executed'; items: string[]; unchanged: string[] } |
| 93 | + | { kind: 'needs amendment'; item: string; outOfScope: string[]; checkpointId: string } |
| 94 | + | { kind: 'needs human'; item: string; reason: string } |
| 95 | + | { kind: 'stopped'; item: string; state: string; reason: string | null }; |
| 96 | + |
| 97 | +/** |
| 98 | + * Runs a task's plan items in order, one execute attempt each. Stops at the first item that does not complete cleanly: |
| 99 | + * out-of-scope files pause the task in needs amendment with a checkpoint; a safety violation moves it to needs human. |
| 100 | + */ |
| 101 | +export class ItemExecutor { |
| 102 | + #store: Store; #runner: RunnerCoordinator; #sources: ExecutionSources; |
| 103 | + #deadlineMs: number; |
| 104 | + constructor(store: Store, runner: RunnerCoordinator, sources: ExecutionSources, deadlineMs = 10 * 60_000) { |
| 105 | + this.#store = store; this.#runner = runner; this.#sources = sources; this.#deadlineMs = deadlineMs; |
| 106 | + } |
| 107 | + async runTask(identity: PlanIdentity, options: { fromItem?: string } = {}): Promise<ExecutionOutcome> { |
| 108 | + const plan = this.#store.getPlan(identity); |
| 109 | + const start = options.fromItem ? plan.items.findIndex(item => item.id === options.fromItem) : 0; |
| 110 | + if (start < 0) throw new Error('Unknown plan item.'); |
| 111 | + const done: string[] = [], unchanged: string[] = []; |
| 112 | + for (const item of plan.items.slice(start)) { |
| 113 | + const attempt = this.#runner.start(identity, { |
| 114 | + expectedStateVersion: this.#store.getTask(identity).stateVersion, kind: 'execute', item: item.id, |
| 115 | + expectedContext: this.#store.currentContext(identity), deadline: Date.now() + this.#deadlineMs, |
| 116 | + }); |
| 117 | + await this.#runner.settled(identity); |
| 118 | + const row = this.#store.getAttempt(identity, attempt.id); |
| 119 | + if (row.state !== 'completed') { |
| 120 | + if (row.state === 'failed' && row.diagnostic?.startsWith(SAFETY_VIOLATION)) { |
| 121 | + this.#store.transitionTask(identity, this.#store.getTask(identity).stateVersion, 'needs human'); |
| 122 | + return { kind: 'needs human', item: item.id, reason: row.diagnostic }; |
| 123 | + } |
| 124 | + return { kind: 'stopped', item: item.id, state: row.state, reason: row.diagnostic }; |
| 125 | + } |
| 126 | + const result = row.result as ExecutionResult; |
| 127 | + done.push(item.id); |
| 128 | + if (result.unchanged) unchanged.push(item.id); |
| 129 | + if (result.outOfScope.length) { |
| 130 | + const view = { revision: this.#store.getPlan(identity).revision, snapshotId: this.#store.getSnapshot(identity).id }; |
| 131 | + const checkpoint = this.#store.recordCheckpoint(identity, view, { |
| 132 | + item: item.id, baseEntries: this.#sources.planContext(identity).baseEntries, |
| 133 | + completedItems: plan.items.slice(0, plan.items.indexOf(item) + 1).map(entry => entry.id), outOfScopePaths: result.outOfScope, |
| 134 | + }); |
| 135 | + this.#store.transitionTask(identity, this.#store.getTask(identity).stateVersion, 'needs amendment'); |
| 136 | + return { kind: 'needs amendment', item: item.id, outOfScope: result.outOfScope, checkpointId: checkpoint.id }; |
| 137 | + } |
| 138 | + } |
| 139 | + return { kind: 'executed', items: done, unchanged }; |
| 140 | + } |
| 141 | +} |
0 commit comments