diff --git a/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md b/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md index 9adcfaf..3e60ebd 100644 --- a/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md +++ b/skills/autopilot/docs/2026-08-31-cooperative-harness-execution-plan.md @@ -1,6 +1,6 @@ # Cooperative harness execution implementation plan -- **Status:** Approved design; PR 1 decision and promotion shutdown complete, runtime implementation not started +- **Status:** Approved design; PR 1 decision/promotion shutdown and PR 2 execution-assurance protocol complete; operator recovery and harness integration remain pending - **Date:** 2026-08-31 - **Audience:** Autopilot implementers and reviewers - **Related:** [Architecture](architecture.md), [continuity implementation plan](2026-08-30-continuity-evidence-implementation-plan.md), [durable event engine ADR](adr/0001-durable-event-engine.md) diff --git a/skills/autopilot/docs/architecture.md b/skills/autopilot/docs/architecture.md index 3f1b19a..5815f55 100644 --- a/skills/autopilot/docs/architecture.md +++ b/skills/autopilot/docs/architecture.md @@ -283,7 +283,19 @@ interface HarnessPort { } ``` -The capability manifest describes unattended execution, useful concurrency, event streaming, cancellation, restart reattachment, tool restrictions, and assurance level. +The capability manifest describes unattended execution, useful concurrency, event streaming, cancellation, tool restrictions, and assurance level. Its versioned execution-assurance profiles select implementation and review behavior independently: + +```ts +interface ExecutionAssurance { + schemaVersion: 1; + owner: "runtime" | "harness"; + continuity: "session" | "same-harness-instance" | "durable-subject"; + terminality: "cooperative" | "process-supervised"; + admission: "single-shot" | "idempotent"; +} +``` + +`restartReattachment` remains a protocol-v1 compatibility field for older manifests and journals. New attempts persist their selected assurance before launch in `ATTEMPT_STARTED`, then persist the exact adapter, backend, and subject identity in `ATTEMPT_EXECUTION_ADMITTED`. A lost single-shot admission or cooperative continuity boundary becomes `EXECUTION_STATE_UNKNOWN`; the engine never converts a missing reattachment into a new launch. Adapters return observations. They never write the journal or choose lifecycle transitions. On POSIX hosts, built-in CLI implementation executions run beneath a detached, attempt-scoped supervisor that owns the harness pipes and bounded output/activity capture. Before harness launch, a separately detached watchdog durably confirms readiness. The harness then joins the supervisor's known process group. All terminal publication is a watchdog-owned handshake: the supervisor publishes a bounded completion candidate, the watchdog terminates and confirms the group is quiescent, and only then publishes the durable result and terminal status. This also covers supervisor exit before child-identity publication. The reviewed Windows x64 Job Object helper will not be packaged because a custom process-management executable creates antivirus, application-reputation, architecture, and provenance risk. The source and runtime path remain temporarily for ordered removal, but the artifact-producing workflow is gone and Windows continues to report restart reattachment as unsupported. Windows currently uses `taskkill` only for the existing session-scoped direct-execution fallback. The approved replacement is version-pinned cooperative harness execution: only an exact terminal response from the uninterrupted harness instance may proceed to repository verification. Harness loss becomes `EXECUTION_STATE_UNKNOWN`, launches no replacement, and requires operator recovery. This boundary does not prove process-tree quiescence. The supervisor writes only fenced operational artifacts under `runs//executions//`; it cannot write `events.jsonl`, receipts, leases, snapshots, or Git state. On supported POSIX attempts, a fresh coordinator reconstructs the exact request from the journaled attempt and immutable context, reattaches to running or terminal supervisor artifacts, and waits for terminal process-tree evidence before allowing a replacement attempt. Review executions remain session-scoped. diff --git a/skills/autopilot/docs/implementation-plan.md b/skills/autopilot/docs/implementation-plan.md index 384aa96..f4c2374 100644 --- a/skills/autopilot/docs/implementation-plan.md +++ b/skills/autopilot/docs/implementation-plan.md @@ -1,6 +1,6 @@ # Harness-agnostic Autopilot implementation plan -- **Status:** Developer-preview implementation available; POSIX attempt-scoped implementation reattachment and controlled-fixture sealed restack successors are packaged. The reviewed Windows x64 Job Object artifact will not be packaged; Windows restart reattachment remains disabled while cooperative harness execution is implemented. +- **Status:** Developer-preview implementation available; POSIX attempt-scoped implementation reattachment, versioned per-mode execution assurance, exact admitted-subject journaling, and controlled-fixture sealed restack successors are packaged. The reviewed Windows x64 Job Object artifact will not be packaged; Windows restart reattachment remains disabled while operator recovery and cooperative harness integration are implemented. - **Date:** 2026-08-22 - **Audience:** Autopilot implementers and reviewers - **Governing design:** [Autopilot architecture](architecture.md) diff --git a/skills/autopilot/references/adapters.md b/skills/autopilot/references/adapters.md index a32b10a..cce6841 100644 --- a/skills/autopilot/references/adapters.md +++ b/skills/autopilot/references/adapters.md @@ -13,6 +13,12 @@ Adapters start fresh noninteractive sessions and normalize observations. They ca The adapter parser bounds output and rejects malformed JSON-mode output. The runtime ignores model completion claims and inspects the worktree directly. +### Execution assurance + +Protocol-v1 manifests may provide separate versioned execution-assurance profiles for implementation and review requests. Each profile records execution ownership (`runtime` or `harness`), continuity (`session`, `same-harness-instance`, or `durable-subject`), terminality (`cooperative` or `process-supervised`), and admission (`single-shot` or `idempotent`). Older manifests remain readable: `restartReattachment: true` maps to the existing runtime-owned process-supervised implementation path, while reviews and non-reattachable implementations remain session-scoped and cooperative. + +Before launch, `ATTEMPT_STARTED` persists the selected assurance and exact request identity. After launch returns, `ATTEMPT_EXECUTION_ADMITTED` binds the adapter and harness versions, adapter execution ID, backend ID, subject ID, and harness-instance ID when required. Losing a single-shot launch reply, changing any exact reattachment identity, or losing a cooperative harness instance becomes `EXECUTION_STATE_UNKNOWN`. A missing reattachment never falls back to `launch()`. + For a `review` gate, the runtime sends a separate role-scoped request with no writable roots or worker write/process grants. Claude Code receives only read/search tools, Codex uses its read-only sandbox, and direct Pi receives only its read tool. OpenCode and any ambient operating-system access remain cooperative. The adapter extracts exactly one structured review marker; missing, contradictory, malformed, truncated, timed-out, or inconclusive output is `UNVERIFIED`. The runtime compares the complete tree, HEAD, refs, and Git configuration before and after review and rejects any mutation. Version-pinned disposable exact-tree runs passed with Pi 0.84.4, Codex 0.151.0, and OpenCode 1.18.25. Claude Code 2.1.251 now reaches its API credential path, but the identity-linked key requires an `ANTHROPIC_WORKSPACE_ID` that is not present in the validation environment, so edit and review remain unverified. For Pi, Autopilot checks the standard Pi package directory for `pi-subagents` 0.53.0 or newer. When present, it loads only that extension and Autopilot's bridge, delegates the item to the resolved `worker` role through the public structured delegation API, and keeps the worker in Autopilot's existing worktree. The read-only exact-tree review role runs directly and is not subjected to the worker-only subagent terminal envelope. Autopilot does not install or update the extension. An older or absent installation uses the direct Pi process and reports that fallback through `doctor` and adapter limitations. diff --git a/skills/autopilot/runtime/dist/src/adapter-process.js b/skills/autopilot/runtime/dist/src/adapter-process.js index 32bd7b8..7fdf8c0 100644 --- a/skills/autopilot/runtime/dist/src/adapter-process.js +++ b/skills/autopilot/runtime/dist/src/adapter-process.js @@ -203,6 +203,31 @@ export class CliHarnessAdapter { eventStreaming: this.#configuration.expectsJsonLines, cancellation: this.#configuration.cancellation, restartReattachment: process.platform !== "win32" || windowsHelperSha256 !== undefined, + executionAssurance: { + schemaVersion: 1, + implementation: process.platform !== "win32" || windowsHelperSha256 !== undefined + ? { + schemaVersion: 1, + owner: "runtime", + continuity: "durable-subject", + terminality: "process-supervised", + admission: "idempotent", + } + : { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + review: { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + }, restrictions: this.#configuration.assurance, limitations: [ ...this.#configuration.limitations, @@ -266,6 +291,7 @@ export class CliHarnessAdapter { protocolVersion: 1, adapterExecutionId: handle.executionId, startedAt: handle.startedAt, + subject: { schemaVersion: 1, backendId: "process-supervisor", subjectId: handle.executionId }, supervisor: { schemaVersion: 1, directory: handle.directory, requestHash: handle.requestHash }, }; } @@ -301,7 +327,12 @@ export class CliHarnessAdapter { })); this.#executions.set(adapterExecutionId, { controller, promise }); this.#requests.set(adapterExecutionId, request); - return { protocolVersion: 1, adapterExecutionId, startedAt }; + return { + protocolVersion: 1, + adapterExecutionId, + startedAt, + subject: { schemaVersion: 1, backendId: "direct-process", subjectId: adapterExecutionId }, + }; } async reattach(request) { const environment = adapterEnvironment(request); @@ -318,6 +349,7 @@ export class CliHarnessAdapter { protocolVersion: 1, adapterExecutionId: handle.executionId, startedAt: handle.startedAt, + subject: { schemaVersion: 1, backendId: "process-supervisor", subjectId: handle.executionId }, supervisor: { schemaVersion: 1, directory: handle.directory, requestHash: handle.requestHash }, }; } diff --git a/skills/autopilot/runtime/dist/src/adapter-protocol.d.ts b/skills/autopilot/runtime/dist/src/adapter-protocol.d.ts index 038d52f..55b8969 100644 --- a/skills/autopilot/runtime/dist/src/adapter-protocol.d.ts +++ b/skills/autopilot/runtime/dist/src/adapter-protocol.d.ts @@ -1,4 +1,16 @@ import { type AssuranceLevel, type CapabilityGrant, type GrantFamily, type Predicate, type VerificationGate } from "./charter.js"; +export interface ExecutionAssurance { + readonly schemaVersion: 1; + readonly owner: "runtime" | "harness"; + readonly continuity: "session" | "same-harness-instance" | "durable-subject"; + readonly terminality: "cooperative" | "process-supervised"; + readonly admission: "single-shot" | "idempotent"; +} +export interface ExecutionAssuranceProfiles { + readonly schemaVersion: 1; + readonly implementation: ExecutionAssurance; + readonly review: ExecutionAssurance; +} export interface CapabilityManifest { readonly protocolVersion: 1; readonly adapterName: string; @@ -11,6 +23,7 @@ export interface CapabilityManifest { readonly eventStreaming: boolean; readonly cancellation: boolean; readonly restartReattachment: boolean; + readonly executionAssurance?: ExecutionAssuranceProfiles; readonly restrictions: "enforced" | "cooperative"; readonly limitations: readonly string[]; } @@ -81,6 +94,7 @@ export interface ReviewResult { export interface ExecutionRequest { readonly protocolVersion: 1; readonly role: "implementation" | "review"; + readonly executionAssurance?: ExecutionAssurance; readonly runId: string; readonly itemId: string; readonly attemptId: string; @@ -98,10 +112,17 @@ export interface ExecutionRequest { readonly maximumOutputBytes: number; readonly supervisionDirectory?: string; } +export interface ExecutionSubject { + readonly schemaVersion: 1; + readonly backendId: string; + readonly subjectId: string; + readonly harnessInstanceId?: string; +} export interface ExecutionHandle { readonly protocolVersion: 1; readonly adapterExecutionId: string; readonly startedAt: string; + readonly subject?: ExecutionSubject; readonly supervisor?: { readonly schemaVersion: 1; readonly directory: string; @@ -130,6 +151,8 @@ export interface HarnessPort { observe(handle: ExecutionHandle): Promise; cancel(handle: ExecutionHandle): Promise; } +export declare function executionAssuranceFor(manifest: CapabilityManifest, role: ExecutionRequest["role"]): ExecutionAssurance; +export declare function parseExecutionAssurance(value: unknown, label: string): ExecutionAssurance; export type AdapterMessage = { readonly protocolVersion: 1; readonly type: "capabilities"; diff --git a/skills/autopilot/runtime/dist/src/adapter-protocol.js b/skills/autopilot/runtime/dist/src/adapter-protocol.js index cc81c30..2314b10 100644 --- a/skills/autopilot/runtime/dist/src/adapter-protocol.js +++ b/skills/autopilot/runtime/dist/src/adapter-protocol.js @@ -1,6 +1,55 @@ import { GRANT_FAMILIES, } from "./charter.js"; import { AutopilotError } from "./errors.js"; import { expectBoolean, expectInteger, expectLiteral, expectRecord, expectString, expectStringArray } from "./json.js"; +const SESSION_COOPERATIVE_ASSURANCE = { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", +}; +function legacyExecutionAssurance(manifest) { + return { + schemaVersion: 1, + implementation: manifest.restartReattachment + ? { + schemaVersion: 1, + owner: "runtime", + continuity: "durable-subject", + terminality: "process-supervised", + admission: "idempotent", + } + : SESSION_COOPERATIVE_ASSURANCE, + review: SESSION_COOPERATIVE_ASSURANCE, + }; +} +export function executionAssuranceFor(manifest, role) { + return manifest.executionAssurance?.[role] ?? legacyExecutionAssurance(manifest)[role]; +} +export function parseExecutionAssurance(value, label) { + const object = expectRecord(value, label); + if (object.schemaVersion !== 1) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", `${label} schema version is not supported`); + } + return { + schemaVersion: 1, + owner: expectLiteral(object.owner, ["runtime", "harness"], `${label}.owner`), + continuity: expectLiteral(object.continuity, ["session", "same-harness-instance", "durable-subject"], `${label}.continuity`), + terminality: expectLiteral(object.terminality, ["cooperative", "process-supervised"], `${label}.terminality`), + admission: expectLiteral(object.admission, ["single-shot", "idempotent"], `${label}.admission`), + }; +} +function parseExecutionAssuranceProfiles(value) { + const object = expectRecord(value, "manifest.executionAssurance"); + if (object.schemaVersion !== 1) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", "execution assurance schema version is not supported"); + } + return { + schemaVersion: 1, + implementation: parseExecutionAssurance(object.implementation, "manifest.executionAssurance.implementation"), + review: parseExecutionAssurance(object.review, "manifest.executionAssurance.review"), + }; +} export function parseAdapterMessage(line, maximumBytes) { if (Buffer.byteLength(line) > maximumBytes) { throw new AutopilotError("ADAPTER_MALFORMED", "adapter message exceeds the configured line limit"); @@ -37,6 +86,9 @@ export function parseAdapterMessage(line, maximumBytes) { eventStreaming: expectBoolean(manifest.eventStreaming, "manifest.eventStreaming"), cancellation: expectBoolean(manifest.cancellation, "manifest.cancellation"), restartReattachment: expectBoolean(manifest.restartReattachment, "manifest.restartReattachment"), + ...(manifest.executionAssurance === undefined + ? {} + : { executionAssurance: parseExecutionAssuranceProfiles(manifest.executionAssurance) }), restrictions: expectLiteral(manifest.restrictions, ["cooperative", "enforced"], "manifest.restrictions"), limitations: expectStringArray(manifest.limitations, "manifest.limitations"), }, diff --git a/skills/autopilot/runtime/dist/src/engine.js b/skills/autopilot/runtime/dist/src/engine.js index e9e9c9f..91ce59e 100644 --- a/skills/autopilot/runtime/dist/src/engine.js +++ b/skills/autopilot/runtime/dist/src/engine.js @@ -1,6 +1,7 @@ import { randomUUID } from "node:crypto"; import { access, mkdir, readFile } from "node:fs/promises"; import { join, resolve } from "node:path"; +import { executionAssuranceFor } from "./adapter-protocol.js"; import { loadAmendmentContext } from "./amendment.js"; import { attemptContextHash, buildAttemptContext } from "./attempt-context.js"; import { createDeliveryAdapter } from "./delivery-adapters.js"; @@ -77,6 +78,7 @@ export class AutopilotEngine { #waitAbort; #activeHandles = new Map(); #implementationHandleIds = new Set(); + #cancellationRequestedHandleIds = new Set(); constructor(options) { this.#stateRoot = options.stateRoot; this.#runDirectory = options.runDirectory; @@ -95,14 +97,7 @@ export class AutopilotEngine { } this.#stopRequested = true; this.#waitAbort?.abort(); - await Promise.all([...this.#activeHandles.values()].map(async (handle) => { - try { - await this.#adapter.cancel(handle); - } - catch { - // The coordinator still stops after the bounded process deadline. - } - })); + await Promise.all([...this.#activeHandles.values()].map(async (handle) => await this.#requestHandleCancellation(handle))); } async requestPause() { if (this.#stopRequested || this.#projection.state === "SUCCEEDED" || this.#projection.state === "STOPPED") { @@ -123,9 +118,25 @@ export class AutopilotEngine { if (handle === undefined) { return []; } - return [this.#adapter.cancel(handle).catch(() => undefined)]; + return [this.#requestHandleCancellation(handle)]; })); } + async #requestHandleCancellation(handle) { + if (this.#cancellationRequestedHandleIds.has(handle.adapterExecutionId)) { + return; + } + this.#cancellationRequestedHandleIds.add(handle.adapterExecutionId); + try { + const result = await this.#adapter.cancel(handle); + if (!result.accepted) { + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + } + } + catch { + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + // Observation or unknown-state recovery still owns the execution boundary. + } + } async #blockItemForStop(item, attemptId) { if (!this.#stopRequested) { return false; @@ -145,6 +156,21 @@ export class AutopilotEngine { #hasUnobservedExecution() { return Object.values(this.#projection.items).some(({ attempts }) => attempts.length > 0 && attempts.at(-1)?.outcome === undefined); } + async #classifyUnobservedExecutionUnknown(reason) { + for (const item of Object.values(this.#projection.items)) { + const attempt = item.attempts.at(-1); + if (item.state === "ACTIVE" && attempt !== undefined && attempt.outcome === undefined) { + await this.#record({ + ...eventBase(reason, "reconciler"), + type: "ITEM_BLOCKED", + itemId: item.itemId, + attemptId: attempt.attemptId, + errorCode: "EXECUTION_STATE_UNKNOWN", + }); + } + } + return Object.values(this.#projection.items).some(({ blocker }) => blocker === "EXECUTION_STATE_UNKNOWN"); + } async #settlePauseIfRequested() { if (!this.#pauseRequested || this.#stopRequested) { return false; @@ -278,13 +304,43 @@ export class AutopilotEngine { async #observeRepository(worktreePath) { return await this.#withRepositoryLock(async () => await observeRepository(worktreePath, this.#managedBranchExpectations())); } + #executionAssurance(role) { + if (this.#manifest === undefined) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", "adapter capabilities have not been loaded"); + } + return executionAssuranceFor(this.#manifest, role); + } + #attemptExecutionAssurance(attempt) { + if (attempt.executionAssurance !== undefined) { + return attempt.executionAssurance; + } + return attempt.executionSupervised === true + ? { + schemaVersion: 1, + owner: "runtime", + continuity: "durable-subject", + terminality: "process-supervised", + admission: "idempotent", + } + : { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }; + } #executionSupervisionEnabled() { - return this.#manifest?.restartReattachment === true && this.#adapter.reattach !== undefined; + const assurance = this.#executionAssurance("implementation"); + return assurance.continuity === "durable-subject" + && assurance.terminality === "process-supervised" + && this.#adapter.reattach !== undefined; } #implementationRequest(item, attemptId, worktreePath, context, contextHash, deadline) { return { protocolVersion: 1, role: "implementation", + executionAssurance: this.#executionAssurance("implementation"), runId: this.#charter.runId, itemId: item.id, attemptId, @@ -302,6 +358,32 @@ export class AutopilotEngine { ...(this.#executionSupervisionEnabled() ? { supervisionDirectory: this.#runDirectory } : {}), }; } + async #recordExecutionAdmission(itemId, attemptId, handle) { + if (this.#manifest === undefined || handle.adapterExecutionId.length === 0) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "adapter execution admission identity is incomplete"); + } + const assurance = this.#executionAssurance("implementation"); + const subject = handle.subject ?? (assurance.owner === "runtime" + ? { schemaVersion: 1, backendId: "legacy-runtime-adapter", subjectId: handle.adapterExecutionId } + : undefined); + if (subject === undefined || subject.backendId.length === 0 || subject.subjectId.length === 0 + || (assurance.continuity === "same-harness-instance" && !subject.harnessInstanceId)) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "harness-owned execution subject identity was not captured"); + } + await this.#record({ + ...eventBase("Exact adapter execution subject admitted"), + type: "ATTEMPT_EXECUTION_ADMITTED", + itemId, + attemptId, + adapterName: this.#manifest.adapterName, + adapterVersion: this.#manifest.adapterVersion, + harnessVersion: this.#manifest.harnessVersion, + adapterExecutionId: handle.adapterExecutionId, + backendId: subject.backendId, + subjectId: subject.subjectId, + ...(subject.harnessInstanceId === undefined ? {} : { harnessInstanceId: subject.harnessInstanceId }), + }); + } #runtimeAuthorize(family, details = {}) { authorizeEffect({ family, actor: "runtime", ...details }, this.#requested, this.#charter.grants, RUNTIME_CAPABILITIES); } @@ -381,8 +463,11 @@ export class AutopilotEngine { || lease.attemptId !== attempt.attemptId || lease.epoch !== attempt.leaseEpoch) { throw new AutopilotError("JOURNAL_CORRUPT", "interrupted attempt writer lease changed identity"); } + const attemptAssurance = this.#attemptExecutionAssurance(attempt); if (itemProjection.state === "ACTIVE" && attempt.outcome === undefined - && attempt.executionSupervised === true && this.#executionSupervisionEnabled()) { + && attemptAssurance.continuity === "durable-subject" + && attemptAssurance.terminality === "process-supervised" + && this.#adapter.reattach !== undefined) { continue; } const observed = await this.#observeRepository(lease.worktreePath); @@ -611,7 +696,12 @@ export class AutopilotEngine { }); } async #reattachInterruptedExecution(item, attempt) { - if (attempt.outcome !== undefined || attempt.executionSupervised !== true || !this.#executionSupervisionEnabled()) { + const assurance = this.#attemptExecutionAssurance(attempt); + if (attempt.outcome !== undefined + || assurance.continuity !== "durable-subject" + || assurance.terminality !== "process-supervised" + || canonicalJson(assurance) !== canonicalJson(this.#executionAssurance("implementation")) + || this.#adapter.reattach === undefined) { return false; } const lease = await readLease(this.#runDirectory, item.id); @@ -629,17 +719,42 @@ export class AutopilotEngine { if (reattach === undefined) { return false; } - const existingHandle = await reattach.call(this.#adapter, request); - const handle = existingHandle ?? await this.#adapter.launch(request); + const handle = await reattach.call(this.#adapter, request); + if (handle === undefined) { + return false; + } + const admitted = attempt.execution; + const subject = handle.subject; + if (admitted !== undefined && (this.#manifest === undefined + || this.#manifest.adapterName !== admitted.adapterName + || this.#manifest.adapterVersion !== admitted.adapterVersion + || this.#manifest.harnessVersion !== admitted.harnessVersion + || handle.adapterExecutionId !== admitted.adapterExecutionId + || subject === undefined || subject.backendId !== admitted.backendId || subject.subjectId !== admitted.subjectId + || subject.harnessInstanceId !== admitted.harnessInstanceId)) { + return false; + } this.#activeHandles.set(handle.adapterExecutionId, handle); this.#implementationHandleIds.add(handle.adapterExecutionId); let observation; try { - observation = await this.#adapter.observe(handle); + if (this.#stopRequested || this.#pauseRequested) { + await this.#requestHandleCancellation(handle); + } + try { + observation = await this.#adapter.observe(handle); + } + catch (error) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "reattached execution did not produce an authoritative terminal observation", { cause: error instanceof Error ? error.message : String(error) }); + } } finally { this.#activeHandles.delete(handle.adapterExecutionId); this.#implementationHandleIds.delete(handle.adapterExecutionId); + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + } + if (observation.adapterExecutionId !== handle.adapterExecutionId) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "reattached observation changed exact execution identity"); } const attemptsDirectory = join(this.#runDirectory, "reports", "attempts"); const observationPath = join(attemptsDirectory, `${attempt.attemptId}.json`); @@ -662,7 +777,7 @@ export class AutopilotEngine { } await assertWritablePaths(lease.worktreePath, after.changedPaths, item.writableRoots); await this.#record({ - ...eventBase(stale ? "Late supervised adapter result quarantined" : "Supervised adapter execution reattached and observed", "reconciler"), + ...eventBase(stale ? "Late supervised adapter result quarantined" : "Exact supervised adapter execution reattached and observed", "reconciler"), type: "ATTEMPT_FINISHED", itemId: item.id, attemptId: attempt.attemptId, @@ -714,7 +829,16 @@ export class AutopilotEngine { continue; } if (itemProjection.state === "ACTIVE") { - const reattachedStatus = await this.#reattachInterruptedExecution(item, attempt); + let reattachedStatus; + try { + reattachedStatus = await this.#reattachInterruptedExecution(item, attempt); + } + catch (error) { + if (!(error instanceof AutopilotError) || error.code !== "EXECUTION_STATE_UNKNOWN") { + throw error; + } + reattachedStatus = false; + } if (reattachedStatus !== false) { if (reattachedStatus === "cancelled" && this.#projection.pauseRequestId !== undefined) { await this.#record({ @@ -1341,6 +1465,7 @@ export class AutopilotEngine { const handle = await this.#adapter.launch({ protocolVersion: 1, role: "review", + executionAssurance: this.#executionAssurance("review"), runId: this.#charter.runId, itemId: item.id, attemptId: `${attemptId}-review-${reviewKey}`, @@ -1364,6 +1489,7 @@ export class AutopilotEngine { } finally { this.#activeHandles.delete(handle.adapterExecutionId); + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); } const reviewObservationPath = join(attemptsDirectory, `${attemptId}.review-${reviewKey}-${contextHash.slice(0, 16)}.json`); await writeJsonAtomic(reviewObservationPath, adapterObservation); @@ -1919,6 +2045,7 @@ export class AutopilotEngine { contextHash, contextJournalSequence: context.sourceJournalSequence, executionSupervised: this.#executionSupervisionEnabled(), + executionAssurance: this.#executionAssurance("implementation"), deadline, evidence: [contextPath], idempotencyKey: `attempt:${this.#charter.runId}:${item.id}:${lease.epoch}`, @@ -1931,24 +2058,40 @@ export class AutopilotEngine { if (await this.#blockItemForStop(item, attemptId)) { return; } - const handle = await this.#adapter.launch(this.#implementationRequest(item, attemptId, worktreePath, context, contextHash, deadline)); + const assurance = this.#executionAssurance("implementation"); + let handle; + try { + handle = await this.#adapter.launch(this.#implementationRequest(item, attemptId, worktreePath, context, contextHash, deadline)); + } + catch (error) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "execution admission did not return an exact durable subject; launch retry is prohibited", { + cause: error instanceof Error ? error.message : String(error), + owner: assurance.owner, + admission: assurance.admission, + }); + } this.#activeHandles.set(handle.adapterExecutionId, handle); this.#implementationHandleIds.add(handle.adapterExecutionId); + await this.#recordExecutionAdmission(item.id, attemptId, handle); if (this.#stopRequested || this.#pauseRequested) { - try { - await this.#adapter.cancel(handle); - } - catch { - // Observation still owns bounded process cleanup. - } + await this.#requestHandleCancellation(handle); } let observation; try { - observation = await this.#adapter.observe(handle); + try { + observation = await this.#adapter.observe(handle); + } + catch (error) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "admitted execution did not produce an authoritative terminal observation", { cause: error instanceof Error ? error.message : String(error) }); + } } finally { this.#activeHandles.delete(handle.adapterExecutionId); this.#implementationHandleIds.delete(handle.adapterExecutionId); + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + } + if (observation.adapterExecutionId !== handle.adapterExecutionId) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "adapter observation changed exact execution identity"); } const observationPath = join(attemptsDirectory, `${attemptId}.json`); await writeJsonAtomic(observationPath, observation); @@ -2171,8 +2314,14 @@ export class AutopilotEngine { this.#manifest = await this.#adapter.describe(); } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + if (this.#hasUnobservedExecution() + && await this.#classifyUnobservedExecutionUnknown(`Exact adapter continuity could not be restored: ${reason}`)) { + await this.#waitForUnknownExecution(); + return await this.#writeReport(); + } await this.#record({ - ...eventBase(error instanceof Error ? error.message : String(error)), + ...eventBase(reason), type: "RUN_STOPPED", errorCode: error instanceof AutopilotError ? error.code : "ADAPTER_PREFLIGHT_FAILED", remediation: "Install or configure the selected harness adapter, then start a successor run.", @@ -2232,6 +2381,9 @@ export class AutopilotEngine { if (await this.#stopRunIfRequested()) { break; } + if (await this.#waitForUnknownExecution()) { + break; + } if (await this.#settlePauseIfRequested()) { break; } diff --git a/skills/autopilot/runtime/dist/src/events.d.ts b/skills/autopilot/runtime/dist/src/events.d.ts index dffce75..9a6d856 100644 --- a/skills/autopilot/runtime/dist/src/events.d.ts +++ b/skills/autopilot/runtime/dist/src/events.d.ts @@ -1,3 +1,4 @@ +import { type ExecutionAssurance } from "./adapter-protocol.js"; export type EventSource = "runtime" | "operator" | "reconciler"; export type WaitingDetails = { readonly kind: "operator-pause"; @@ -83,8 +84,20 @@ export type LifecycleEvent = (EventBase & { readonly contextHash?: string; readonly contextJournalSequence?: number; readonly executionSupervised?: boolean; + readonly executionAssurance?: ExecutionAssurance; readonly deadline: string; readonly idempotencyKey: string; +}) | (EventBase & { + readonly type: "ATTEMPT_EXECUTION_ADMITTED"; + readonly itemId: string; + readonly attemptId: string; + readonly adapterName: string; + readonly adapterVersion: string; + readonly harnessVersion: string; + readonly adapterExecutionId: string; + readonly backendId: string; + readonly subjectId: string; + readonly harnessInstanceId?: string; }) | (EventBase & { readonly type: "ATTEMPT_FINISHED"; readonly itemId: string; diff --git a/skills/autopilot/runtime/dist/src/events.js b/skills/autopilot/runtime/dist/src/events.js index 20b92ec..c8225c1 100644 --- a/skills/autopilot/runtime/dist/src/events.js +++ b/skills/autopilot/runtime/dist/src/events.js @@ -1,10 +1,11 @@ import { randomUUID } from "node:crypto"; +import { parseExecutionAssurance } from "./adapter-protocol.js"; import { AutopilotError } from "./errors.js"; import { expectBoolean, expectInteger, expectLiteral, expectRecord, expectString, expectStringArray } from "./json.js"; const EVENT_TYPES = [ "CHARTER_COMPILED", "RECONCILIATION_STARTED", "RECONCILIATION_COMPLETED", "RUN_PAUSE_REQUESTED", "RUN_WAITING", "RUN_WOKEN", "RUN_RESUMED", "RUN_VERIFYING", "RUN_SUCCEEDED", "RUN_STOPPED", "WRAP_UP_STARTED", "WORKTREE_ADOPTED", "ITEM_READY", - "ATTEMPT_STARTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", "ITEM_VERIFIED", "ITEM_SATISFIED", + "ATTEMPT_STARTED", "ATTEMPT_EXECUTION_ADMITTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", "ITEM_VERIFIED", "ITEM_SATISFIED", "ITEM_BLOCKED", "ITEM_ABANDONED", "RESTACK_DESCENDANT_STARTED", "RESTACK_DESCENDANT_TREE_PREPARED", "RESTACK_DESCENDANT_VERIFIED", "RESTACK_PROVIDER_HEAD_CONFIRMED", "RESTACK_DESCENDANT_SATISFIED", "RESTACK_DESCENDANT_BLOCKED", "EFFECT_INTENDED", @@ -177,9 +178,28 @@ export function parseLifecycleEvent(value) { ...(object.executionSupervised === undefined ? {} : { executionSupervised: expectBoolean(object.executionSupervised, "event.executionSupervised"), }), + ...(object.executionAssurance === undefined ? {} : { + executionAssurance: parseExecutionAssurance(object.executionAssurance, "event.executionAssurance"), + }), deadline: expectString(object.deadline, "event.deadline"), idempotencyKey: expectString(object.idempotencyKey, "event.idempotencyKey"), }; + case "ATTEMPT_EXECUTION_ADMITTED": + return { + ...base, + type, + itemId: expectString(object.itemId, "event.itemId"), + attemptId: expectString(object.attemptId, "event.attemptId"), + adapterName: expectString(object.adapterName, "event.adapterName"), + adapterVersion: expectString(object.adapterVersion, "event.adapterVersion"), + harnessVersion: expectString(object.harnessVersion, "event.harnessVersion"), + adapterExecutionId: expectString(object.adapterExecutionId, "event.adapterExecutionId"), + backendId: expectString(object.backendId, "event.backendId"), + subjectId: expectString(object.subjectId, "event.subjectId"), + ...(object.harnessInstanceId === undefined ? {} : { + harnessInstanceId: expectString(object.harnessInstanceId, "event.harnessInstanceId"), + }), + }; case "ATTEMPT_FINISHED": return { ...base, diff --git a/skills/autopilot/runtime/dist/src/reducer.d.ts b/skills/autopilot/runtime/dist/src/reducer.d.ts index 58cffe5..7e3d7d2 100644 --- a/skills/autopilot/runtime/dist/src/reducer.d.ts +++ b/skills/autopilot/runtime/dist/src/reducer.d.ts @@ -1,3 +1,4 @@ +import type { ExecutionAssurance } from "./adapter-protocol.js"; import type { RunCharter } from "./charter.js"; import type { LifecycleEvent, WaitingDetails } from "./events.js"; export type RunState = "COMPILED" | "RECONCILING" | "RUNNING" | "WAITING" | "VERIFYING" | "SUCCEEDED" | "STOPPED"; @@ -16,6 +17,16 @@ export interface AttemptProjection { readonly contextHash?: string; readonly contextJournalSequence?: number; readonly executionSupervised?: boolean; + readonly executionAssurance?: ExecutionAssurance; + readonly execution?: { + readonly adapterName: string; + readonly adapterVersion: string; + readonly harnessVersion: string; + readonly adapterExecutionId: string; + readonly backendId: string; + readonly subjectId: string; + readonly harnessInstanceId?: string; + }; readonly deadline: string; readonly idempotencyKey: string; readonly outcome?: "completed" | "failed" | "cancelled" | "timed-out" | "stale"; diff --git a/skills/autopilot/runtime/dist/src/reducer.js b/skills/autopilot/runtime/dist/src/reducer.js index 8f5f331..4242068 100644 --- a/skills/autopilot/runtime/dist/src/reducer.js +++ b/skills/autopilot/runtime/dist/src/reducer.js @@ -62,8 +62,9 @@ function assertRunTransition(current, event) { if (event.waiting?.kind === "operator-pause") { return "WAITING"; } - if (current !== "RUNNING" && current !== "WAITING") { - throw new AutopilotError("ILLEGAL_TRANSITION", `${event.type} requires RUNNING or WAITING, received ${current}`); + if (current !== "RUNNING" && current !== "WAITING" + && !(current === "RECONCILING" && event.waiting?.kind === "execution-unknown")) { + throw new AutopilotError("ILLEGAL_TRANSITION", `${event.type} requires RUNNING or WAITING, or RECONCILING for an unknown execution; received ${current}`); } return "WAITING"; case "RUN_WOKEN": @@ -127,11 +128,38 @@ function transitionItem(item, event) { ...(event.contextHash === undefined ? {} : { contextHash: event.contextHash }), ...(event.contextJournalSequence === undefined ? {} : { contextJournalSequence: event.contextJournalSequence }), ...(event.executionSupervised === undefined ? {} : { executionSupervised: event.executionSupervised }), + ...(event.executionAssurance === undefined ? {} : { executionAssurance: event.executionAssurance }), deadline: event.deadline, idempotencyKey: event.idempotencyKey, }, ], }; + case "ATTEMPT_EXECUTION_ADMITTED": { + if (item.state !== "ACTIVE") { + throw new AutopilotError("ILLEGAL_TRANSITION", `ATTEMPT_EXECUTION_ADMITTED cannot follow ${item.state}`); + } + const currentAttempt = item.attempts.at(-1); + if (currentAttempt?.attemptId !== event.attemptId || currentAttempt.execution !== undefined) { + throw new AutopilotError("ILLEGAL_TRANSITION", `execution admission is stale or duplicated for ${item.itemId}`); + } + return { + ...item, + attempts: item.attempts.map((attempt) => attempt.attemptId === event.attemptId + ? { + ...attempt, + execution: { + adapterName: event.adapterName, + adapterVersion: event.adapterVersion, + harnessVersion: event.harnessVersion, + adapterExecutionId: event.adapterExecutionId, + backendId: event.backendId, + subjectId: event.subjectId, + ...(event.harnessInstanceId === undefined ? {} : { harnessInstanceId: event.harnessInstanceId }), + }, + } + : attempt), + }; + } case "ATTEMPT_FINISHED": { if (item.state !== "ACTIVE") { throw new AutopilotError("ILLEGAL_TRANSITION", `ATTEMPT_FINISHED cannot follow ${item.state}`); @@ -347,7 +375,7 @@ export function reduce(projection, event) { const nextState = assertRunTransition(projection.state, event); let items = projection.items; const ordinaryItemLifecycle = [ - "DECISION_RECORDED", "ITEM_READY", "ATTEMPT_STARTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", + "DECISION_RECORDED", "ITEM_READY", "ATTEMPT_STARTED", "ATTEMPT_EXECUTION_ADMITTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", "ITEM_VERIFIED", "ITEM_SATISFIED", "ITEM_BLOCKED", "ITEM_ABANDONED", ]; if (event.itemId !== undefined && projection.restacks[event.itemId] !== undefined diff --git a/skills/autopilot/runtime/dist/src/report.d.ts b/skills/autopilot/runtime/dist/src/report.d.ts index 73f71ef..07497aa 100644 --- a/skills/autopilot/runtime/dist/src/report.d.ts +++ b/skills/autopilot/runtime/dist/src/report.d.ts @@ -1,3 +1,4 @@ +import type { ExecutionAssurance } from "./adapter-protocol.js"; import type { RunCharter } from "./charter.js"; import { type PredicateEvidenceEntry } from "./evidence-map.js"; import type { JournalRecord } from "./journal.js"; @@ -16,6 +17,16 @@ export interface RunReport { readonly blocker?: string; readonly attempts: number; readonly chargedAttempts: number; + readonly execution?: { + readonly assurance?: ExecutionAssurance; + readonly adapterName?: string; + readonly adapterVersion?: string; + readonly harnessVersion?: string; + readonly adapterExecutionId?: string; + readonly backendId?: string; + readonly subjectId?: string; + readonly harnessInstanceId?: string; + }; }[]; readonly restacks: readonly { readonly itemId: string; diff --git a/skills/autopilot/runtime/dist/src/report.js b/skills/autopilot/runtime/dist/src/report.js index 4990f8c..e3048d8 100644 --- a/skills/autopilot/runtime/dist/src/report.js +++ b/skills/autopilot/runtime/dist/src/report.js @@ -81,6 +81,8 @@ export async function writeReports(runDirectory, charter, projection, records, a state: projection.state, items: ordinaryWork.map((item) => { const itemProjection = projection.items[item.id]; + const lastAttempt = itemProjection?.attempts.at(-1); + const execution = lastAttempt?.execution; return { itemId: item.id, state: itemProjection?.state ?? "PENDING", @@ -89,6 +91,12 @@ export async function writeReports(runDirectory, charter, projection, records, a chargedAttempts: consumedAttempts(itemProjection), ...(itemProjection?.subject === undefined ? {} : { subject: itemProjection.subject }), ...(itemProjection?.blocker === undefined ? {} : { blocker: itemProjection.blocker }), + ...(lastAttempt?.executionAssurance === undefined && execution === undefined ? {} : { + execution: { + ...(lastAttempt?.executionAssurance === undefined ? {} : { assurance: lastAttempt.executionAssurance }), + ...(execution === undefined ? {} : execution), + }, + }), }; }), restacks: Object.values(projection.restacks).map((restack) => ({ diff --git a/skills/autopilot/runtime/schemas/adapter.schema.json b/skills/autopilot/runtime/schemas/adapter.schema.json index a6524da..2ffac16 100644 --- a/skills/autopilot/runtime/schemas/adapter.schema.json +++ b/skills/autopilot/runtime/schemas/adapter.schema.json @@ -48,6 +48,28 @@ } ], "$defs": { + "executionAssurance": { + "type": "object", + "additionalProperties": false, + "required": ["schemaVersion", "owner", "continuity", "terminality", "admission"], + "properties": { + "schemaVersion": { "const": 1 }, + "owner": { "enum": ["runtime", "harness"] }, + "continuity": { "enum": ["session", "same-harness-instance", "durable-subject"] }, + "terminality": { "enum": ["cooperative", "process-supervised"] }, + "admission": { "enum": ["single-shot", "idempotent"] } + } + }, + "executionAssuranceProfiles": { + "type": "object", + "additionalProperties": false, + "required": ["schemaVersion", "implementation", "review"], + "properties": { + "schemaVersion": { "const": 1 }, + "implementation": { "$ref": "#/$defs/executionAssurance" }, + "review": { "$ref": "#/$defs/executionAssurance" } + } + }, "manifest": { "type": "object", "additionalProperties": false, @@ -78,6 +100,7 @@ "eventStreaming": { "type": "boolean" }, "cancellation": { "type": "boolean" }, "restartReattachment": { "type": "boolean" }, + "executionAssurance": { "$ref": "#/$defs/executionAssuranceProfiles" }, "restrictions": { "enum": ["cooperative", "enforced"] }, "limitations": { "type": "array", "items": { "type": "string" } } } diff --git a/skills/autopilot/runtime/src/adapter-process.ts b/skills/autopilot/runtime/src/adapter-process.ts index 939fa6a..f470166 100644 --- a/skills/autopilot/runtime/src/adapter-process.ts +++ b/skills/autopilot/runtime/src/adapter-process.ts @@ -261,6 +261,31 @@ export class CliHarnessAdapter implements HarnessPort { eventStreaming: this.#configuration.expectsJsonLines, cancellation: this.#configuration.cancellation, restartReattachment: process.platform !== "win32" || windowsHelperSha256 !== undefined, + executionAssurance: { + schemaVersion: 1, + implementation: process.platform !== "win32" || windowsHelperSha256 !== undefined + ? { + schemaVersion: 1, + owner: "runtime", + continuity: "durable-subject", + terminality: "process-supervised", + admission: "idempotent", + } + : { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + review: { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + }, restrictions: this.#configuration.assurance, limitations: [ ...this.#configuration.limitations, @@ -345,6 +370,7 @@ export class CliHarnessAdapter implements HarnessPort { protocolVersion: 1, adapterExecutionId: handle.executionId, startedAt: handle.startedAt, + subject: { schemaVersion: 1, backendId: "process-supervisor", subjectId: handle.executionId }, supervisor: { schemaVersion: 1, directory: handle.directory, requestHash: handle.requestHash }, }; } @@ -380,7 +406,12 @@ export class CliHarnessAdapter implements HarnessPort { })); this.#executions.set(adapterExecutionId, { controller, promise }); this.#requests.set(adapterExecutionId, request); - return { protocolVersion: 1, adapterExecutionId, startedAt }; + return { + protocolVersion: 1, + adapterExecutionId, + startedAt, + subject: { schemaVersion: 1, backendId: "direct-process", subjectId: adapterExecutionId }, + }; } async reattach(request: ExecutionRequest): Promise { @@ -403,6 +434,7 @@ export class CliHarnessAdapter implements HarnessPort { protocolVersion: 1, adapterExecutionId: handle.executionId, startedAt: handle.startedAt, + subject: { schemaVersion: 1, backendId: "process-supervisor", subjectId: handle.executionId }, supervisor: { schemaVersion: 1, directory: handle.directory, requestHash: handle.requestHash }, }; } diff --git a/skills/autopilot/runtime/src/adapter-protocol.ts b/skills/autopilot/runtime/src/adapter-protocol.ts index 5f7d967..6c385d8 100644 --- a/skills/autopilot/runtime/src/adapter-protocol.ts +++ b/skills/autopilot/runtime/src/adapter-protocol.ts @@ -9,6 +9,20 @@ import { import { AutopilotError } from "./errors.js"; import { expectBoolean, expectInteger, expectLiteral, expectRecord, expectString, expectStringArray } from "./json.js"; +export interface ExecutionAssurance { + readonly schemaVersion: 1; + readonly owner: "runtime" | "harness"; + readonly continuity: "session" | "same-harness-instance" | "durable-subject"; + readonly terminality: "cooperative" | "process-supervised"; + readonly admission: "single-shot" | "idempotent"; +} + +export interface ExecutionAssuranceProfiles { + readonly schemaVersion: 1; + readonly implementation: ExecutionAssurance; + readonly review: ExecutionAssurance; +} + export interface CapabilityManifest { readonly protocolVersion: 1; readonly adapterName: string; @@ -21,6 +35,7 @@ export interface CapabilityManifest { readonly eventStreaming: boolean; readonly cancellation: boolean; readonly restartReattachment: boolean; + readonly executionAssurance?: ExecutionAssuranceProfiles; readonly restrictions: "enforced" | "cooperative"; readonly limitations: readonly string[]; } @@ -91,6 +106,7 @@ export interface ReviewResult { export interface ExecutionRequest { readonly protocolVersion: 1; readonly role: "implementation" | "review"; + readonly executionAssurance?: ExecutionAssurance; readonly runId: string; readonly itemId: string; readonly attemptId: string; @@ -109,10 +125,18 @@ export interface ExecutionRequest { readonly supervisionDirectory?: string; } +export interface ExecutionSubject { + readonly schemaVersion: 1; + readonly backendId: string; + readonly subjectId: string; + readonly harnessInstanceId?: string; +} + export interface ExecutionHandle { readonly protocolVersion: 1; readonly adapterExecutionId: string; readonly startedAt: string; + readonly subject?: ExecutionSubject; readonly supervisor?: { readonly schemaVersion: 1; readonly directory: string; @@ -145,6 +169,64 @@ export interface HarnessPort { cancel(handle: ExecutionHandle): Promise; } +const SESSION_COOPERATIVE_ASSURANCE: ExecutionAssurance = { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", +}; + +function legacyExecutionAssurance(manifest: Pick): ExecutionAssuranceProfiles { + return { + schemaVersion: 1, + implementation: manifest.restartReattachment + ? { + schemaVersion: 1, + owner: "runtime", + continuity: "durable-subject", + terminality: "process-supervised", + admission: "idempotent", + } + : SESSION_COOPERATIVE_ASSURANCE, + review: SESSION_COOPERATIVE_ASSURANCE, + }; +} + +export function executionAssuranceFor(manifest: CapabilityManifest, role: ExecutionRequest["role"]): ExecutionAssurance { + return manifest.executionAssurance?.[role] ?? legacyExecutionAssurance(manifest)[role]; +} + +export function parseExecutionAssurance(value: unknown, label: string): ExecutionAssurance { + const object = expectRecord(value, label); + if (object.schemaVersion !== 1) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", `${label} schema version is not supported`); + } + return { + schemaVersion: 1, + owner: expectLiteral(object.owner, ["runtime", "harness"], `${label}.owner`), + continuity: expectLiteral( + object.continuity, + ["session", "same-harness-instance", "durable-subject"], + `${label}.continuity`, + ), + terminality: expectLiteral(object.terminality, ["cooperative", "process-supervised"], `${label}.terminality`), + admission: expectLiteral(object.admission, ["single-shot", "idempotent"], `${label}.admission`), + }; +} + +function parseExecutionAssuranceProfiles(value: unknown): ExecutionAssuranceProfiles { + const object = expectRecord(value, "manifest.executionAssurance"); + if (object.schemaVersion !== 1) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", "execution assurance schema version is not supported"); + } + return { + schemaVersion: 1, + implementation: parseExecutionAssurance(object.implementation, "manifest.executionAssurance.implementation"), + review: parseExecutionAssurance(object.review, "manifest.executionAssurance.review"), + }; +} + export type AdapterMessage = | { readonly protocolVersion: 1; readonly type: "capabilities"; readonly manifest: CapabilityManifest } | { readonly protocolVersion: 1; readonly type: "started"; readonly executionId: string } @@ -186,6 +268,9 @@ export function parseAdapterMessage(line: string, maximumBytes: number): Adapter eventStreaming: expectBoolean(manifest.eventStreaming, "manifest.eventStreaming"), cancellation: expectBoolean(manifest.cancellation, "manifest.cancellation"), restartReattachment: expectBoolean(manifest.restartReattachment, "manifest.restartReattachment"), + ...(manifest.executionAssurance === undefined + ? {} + : { executionAssurance: parseExecutionAssuranceProfiles(manifest.executionAssurance) }), restrictions: expectLiteral(manifest.restrictions, ["cooperative", "enforced"], "manifest.restrictions"), limitations: expectStringArray(manifest.limitations, "manifest.limitations"), }, diff --git a/skills/autopilot/runtime/src/engine.ts b/skills/autopilot/runtime/src/engine.ts index 501ea13..7eb1178 100644 --- a/skills/autopilot/runtime/src/engine.ts +++ b/skills/autopilot/runtime/src/engine.ts @@ -4,11 +4,13 @@ import { join, resolve } from "node:path"; import type { AttemptContext, CapabilityManifest, + ExecutionAssurance, ExecutionHandle, ExecutionObservation, ExecutionRequest, HarnessPort, } from "./adapter-protocol.js"; +import { executionAssuranceFor } from "./adapter-protocol.js"; import { loadAmendmentContext, type AmendmentContext } from "./amendment.js"; import { attemptContextHash, buildAttemptContext } from "./attempt-context.js"; import type { GrantFamily, ReviewGate, RunCharter, WorkItem } from "./charter.js"; @@ -123,6 +125,7 @@ export class AutopilotEngine { #waitAbort: AbortController | undefined; readonly #activeHandles = new Map(); readonly #implementationHandleIds = new Set(); + readonly #cancellationRequestedHandleIds = new Set(); constructor(options: EngineOptions) { this.#stateRoot = options.stateRoot; @@ -143,13 +146,9 @@ export class AutopilotEngine { } this.#stopRequested = true; this.#waitAbort?.abort(); - await Promise.all([...this.#activeHandles.values()].map(async (handle) => { - try { - await this.#adapter.cancel(handle); - } catch { - // The coordinator still stops after the bounded process deadline. - } - })); + await Promise.all([...this.#activeHandles.values()].map(async (handle) => + await this.#requestHandleCancellation(handle) + )); } async requestPause(): Promise { @@ -171,10 +170,26 @@ export class AutopilotEngine { if (handle === undefined) { return []; } - return [this.#adapter.cancel(handle).catch(() => undefined)]; + return [this.#requestHandleCancellation(handle)]; })); } + async #requestHandleCancellation(handle: ExecutionHandle): Promise { + if (this.#cancellationRequestedHandleIds.has(handle.adapterExecutionId)) { + return; + } + this.#cancellationRequestedHandleIds.add(handle.adapterExecutionId); + try { + const result = await this.#adapter.cancel(handle); + if (!result.accepted) { + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + } + } catch { + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + // Observation or unknown-state recovery still owns the execution boundary. + } + } + async #blockItemForStop(item: WorkItem, attemptId?: string): Promise { if (!this.#stopRequested) { return false; @@ -198,6 +213,22 @@ export class AutopilotEngine { ); } + async #classifyUnobservedExecutionUnknown(reason: string): Promise { + for (const item of Object.values(this.#projection.items)) { + const attempt = item.attempts.at(-1); + if (item.state === "ACTIVE" && attempt !== undefined && attempt.outcome === undefined) { + await this.#record({ + ...eventBase(reason, "reconciler"), + type: "ITEM_BLOCKED", + itemId: item.itemId, + attemptId: attempt.attemptId, + errorCode: "EXECUTION_STATE_UNKNOWN", + }); + } + } + return Object.values(this.#projection.items).some(({ blocker }) => blocker === "EXECUTION_STATE_UNKNOWN"); + } + async #settlePauseIfRequested(): Promise { if (!this.#pauseRequested || this.#stopRequested) { return false; @@ -346,8 +377,39 @@ export class AutopilotEngine { ); } + #executionAssurance(role: ExecutionRequest["role"]): ExecutionAssurance { + if (this.#manifest === undefined) { + throw new AutopilotError("ADAPTER_UNSUPPORTED", "adapter capabilities have not been loaded"); + } + return executionAssuranceFor(this.#manifest, role); + } + + #attemptExecutionAssurance(attempt: RunProjection["items"][string]["attempts"][number]): ExecutionAssurance { + if (attempt.executionAssurance !== undefined) { + return attempt.executionAssurance; + } + return attempt.executionSupervised === true + ? { + schemaVersion: 1, + owner: "runtime", + continuity: "durable-subject", + terminality: "process-supervised", + admission: "idempotent", + } + : { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }; + } + #executionSupervisionEnabled(): boolean { - return this.#manifest?.restartReattachment === true && this.#adapter.reattach !== undefined; + const assurance = this.#executionAssurance("implementation"); + return assurance.continuity === "durable-subject" + && assurance.terminality === "process-supervised" + && this.#adapter.reattach !== undefined; } #implementationRequest( @@ -361,6 +423,7 @@ export class AutopilotEngine { return { protocolVersion: 1, role: "implementation", + executionAssurance: this.#executionAssurance("implementation"), runId: this.#charter.runId, itemId: item.id, attemptId, @@ -379,6 +442,33 @@ export class AutopilotEngine { }; } + async #recordExecutionAdmission(itemId: string, attemptId: string, handle: ExecutionHandle): Promise { + if (this.#manifest === undefined || handle.adapterExecutionId.length === 0) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "adapter execution admission identity is incomplete"); + } + const assurance = this.#executionAssurance("implementation"); + const subject = handle.subject ?? (assurance.owner === "runtime" + ? { schemaVersion: 1 as const, backendId: "legacy-runtime-adapter", subjectId: handle.adapterExecutionId } + : undefined); + if (subject === undefined || subject.backendId.length === 0 || subject.subjectId.length === 0 + || (assurance.continuity === "same-harness-instance" && !subject.harnessInstanceId)) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "harness-owned execution subject identity was not captured"); + } + await this.#record({ + ...eventBase("Exact adapter execution subject admitted"), + type: "ATTEMPT_EXECUTION_ADMITTED", + itemId, + attemptId, + adapterName: this.#manifest.adapterName, + adapterVersion: this.#manifest.adapterVersion, + harnessVersion: this.#manifest.harnessVersion, + adapterExecutionId: handle.adapterExecutionId, + backendId: subject.backendId, + subjectId: subject.subjectId, + ...(subject.harnessInstanceId === undefined ? {} : { harnessInstanceId: subject.harnessInstanceId }), + }); + } + #runtimeAuthorize(family: GrantFamily, details: Omit[0], "family" | "actor"> = {}): void { authorizeEffect({ family, actor: "runtime", ...details }, this.#requested, this.#charter.grants, RUNTIME_CAPABILITIES); } @@ -473,8 +563,11 @@ export class AutopilotEngine { || lease.attemptId !== attempt.attemptId || lease.epoch !== attempt.leaseEpoch) { throw new AutopilotError("JOURNAL_CORRUPT", "interrupted attempt writer lease changed identity"); } + const attemptAssurance = this.#attemptExecutionAssurance(attempt); if (itemProjection.state === "ACTIVE" && attempt.outcome === undefined - && attempt.executionSupervised === true && this.#executionSupervisionEnabled()) { + && attemptAssurance.continuity === "durable-subject" + && attemptAssurance.terminality === "process-supervised" + && this.#adapter.reattach !== undefined) { continue; } const observed = await this.#observeRepository(lease.worktreePath); @@ -728,7 +821,12 @@ export class AutopilotEngine { item: WorkItem, attempt: RunProjection["items"][string]["attempts"][number], ): Promise { - if (attempt.outcome !== undefined || attempt.executionSupervised !== true || !this.#executionSupervisionEnabled()) { + const assurance = this.#attemptExecutionAssurance(attempt); + if (attempt.outcome !== undefined + || assurance.continuity !== "durable-subject" + || assurance.terminality !== "process-supervised" + || canonicalJson(assurance) !== canonicalJson(this.#executionAssurance("implementation")) + || this.#adapter.reattach === undefined) { return false; } const lease = await readLease(this.#runDirectory, item.id); @@ -753,16 +851,44 @@ export class AutopilotEngine { if (reattach === undefined) { return false; } - const existingHandle = await reattach.call(this.#adapter, request); - const handle = existingHandle ?? await this.#adapter.launch(request); + const handle = await reattach.call(this.#adapter, request); + if (handle === undefined) { + return false; + } + const admitted = attempt.execution; + const subject = handle.subject; + if (admitted !== undefined && (this.#manifest === undefined + || this.#manifest.adapterName !== admitted.adapterName + || this.#manifest.adapterVersion !== admitted.adapterVersion + || this.#manifest.harnessVersion !== admitted.harnessVersion + || handle.adapterExecutionId !== admitted.adapterExecutionId + || subject === undefined || subject.backendId !== admitted.backendId || subject.subjectId !== admitted.subjectId + || subject.harnessInstanceId !== admitted.harnessInstanceId)) { + return false; + } this.#activeHandles.set(handle.adapterExecutionId, handle); this.#implementationHandleIds.add(handle.adapterExecutionId); let observation: ExecutionObservation; try { - observation = await this.#adapter.observe(handle); + if (this.#stopRequested || this.#pauseRequested) { + await this.#requestHandleCancellation(handle); + } + try { + observation = await this.#adapter.observe(handle); + } catch (error) { + throw new AutopilotError( + "EXECUTION_STATE_UNKNOWN", + "reattached execution did not produce an authoritative terminal observation", + { cause: error instanceof Error ? error.message : String(error) }, + ); + } } finally { this.#activeHandles.delete(handle.adapterExecutionId); this.#implementationHandleIds.delete(handle.adapterExecutionId); + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + } + if (observation.adapterExecutionId !== handle.adapterExecutionId) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "reattached observation changed exact execution identity"); } const attemptsDirectory = join(this.#runDirectory, "reports", "attempts"); const observationPath = join(attemptsDirectory, `${attempt.attemptId}.json`); @@ -785,7 +911,7 @@ export class AutopilotEngine { } await assertWritablePaths(lease.worktreePath, after.changedPaths, item.writableRoots); await this.#record({ - ...eventBase(stale ? "Late supervised adapter result quarantined" : "Supervised adapter execution reattached and observed", "reconciler"), + ...eventBase(stale ? "Late supervised adapter result quarantined" : "Exact supervised adapter execution reattached and observed", "reconciler"), type: "ATTEMPT_FINISHED", itemId: item.id, attemptId: attempt.attemptId, @@ -838,7 +964,15 @@ export class AutopilotEngine { continue; } if (itemProjection.state === "ACTIVE") { - const reattachedStatus = await this.#reattachInterruptedExecution(item, attempt); + let reattachedStatus: ExecutionObservation["status"] | "stale" | false; + try { + reattachedStatus = await this.#reattachInterruptedExecution(item, attempt); + } catch (error) { + if (!(error instanceof AutopilotError) || error.code !== "EXECUTION_STATE_UNKNOWN") { + throw error; + } + reattachedStatus = false; + } if (reattachedStatus !== false) { if (reattachedStatus === "cancelled" && this.#projection.pauseRequestId !== undefined) { await this.#record({ @@ -1557,6 +1691,7 @@ export class AutopilotEngine { const handle = await this.#adapter.launch({ protocolVersion: 1, role: "review", + executionAssurance: this.#executionAssurance("review"), runId: this.#charter.runId, itemId: item.id, attemptId: `${attemptId}-review-${reviewKey}`, @@ -1581,6 +1716,7 @@ export class AutopilotEngine { adapterObservation = await this.#adapter.observe(handle); } finally { this.#activeHandles.delete(handle.adapterExecutionId); + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); } const reviewObservationPath = join(attemptsDirectory, `${attemptId}.review-${reviewKey}-${contextHash.slice(0, 16)}.json`); await writeJsonAtomic(reviewObservationPath, adapterObservation); @@ -2211,6 +2347,7 @@ export class AutopilotEngine { contextHash, contextJournalSequence: context.sourceJournalSequence, executionSupervised: this.#executionSupervisionEnabled(), + executionAssurance: this.#executionAssurance("implementation"), deadline, evidence: [contextPath], idempotencyKey: `attempt:${this.#charter.runId}:${item.id}:${lease.epoch}`, @@ -2225,29 +2362,52 @@ export class AutopilotEngine { if (await this.#blockItemForStop(item, attemptId)) { return; } - const handle = await this.#adapter.launch(this.#implementationRequest( - item, - attemptId, - worktreePath, - context, - contextHash, - deadline, - )); + const assurance = this.#executionAssurance("implementation"); + let handle: ExecutionHandle; + try { + handle = await this.#adapter.launch(this.#implementationRequest( + item, + attemptId, + worktreePath, + context, + contextHash, + deadline, + )); + } catch (error) { + throw new AutopilotError( + "EXECUTION_STATE_UNKNOWN", + "execution admission did not return an exact durable subject; launch retry is prohibited", + { + cause: error instanceof Error ? error.message : String(error), + owner: assurance.owner, + admission: assurance.admission, + }, + ); + } this.#activeHandles.set(handle.adapterExecutionId, handle); this.#implementationHandleIds.add(handle.adapterExecutionId); + await this.#recordExecutionAdmission(item.id, attemptId, handle); if (this.#stopRequested || this.#pauseRequested) { - try { - await this.#adapter.cancel(handle); - } catch { - // Observation still owns bounded process cleanup. - } + await this.#requestHandleCancellation(handle); } let observation: ExecutionObservation; try { - observation = await this.#adapter.observe(handle); + try { + observation = await this.#adapter.observe(handle); + } catch (error) { + throw new AutopilotError( + "EXECUTION_STATE_UNKNOWN", + "admitted execution did not produce an authoritative terminal observation", + { cause: error instanceof Error ? error.message : String(error) }, + ); + } } finally { this.#activeHandles.delete(handle.adapterExecutionId); this.#implementationHandleIds.delete(handle.adapterExecutionId); + this.#cancellationRequestedHandleIds.delete(handle.adapterExecutionId); + } + if (observation.adapterExecutionId !== handle.adapterExecutionId) { + throw new AutopilotError("EXECUTION_STATE_UNKNOWN", "adapter observation changed exact execution identity"); } const observationPath = join(attemptsDirectory, `${attemptId}.json`); await writeJsonAtomic(observationPath, observation); @@ -2486,8 +2646,14 @@ export class AutopilotEngine { try { this.#manifest = await this.#adapter.describe(); } catch (error) { + const reason = error instanceof Error ? error.message : String(error); + if (this.#hasUnobservedExecution() + && await this.#classifyUnobservedExecutionUnknown(`Exact adapter continuity could not be restored: ${reason}`)) { + await this.#waitForUnknownExecution(); + return await this.#writeReport(); + } await this.#record({ - ...eventBase(error instanceof Error ? error.message : String(error)), + ...eventBase(reason), type: "RUN_STOPPED", errorCode: error instanceof AutopilotError ? error.code : "ADAPTER_PREFLIGHT_FAILED", remediation: "Install or configure the selected harness adapter, then start a successor run.", @@ -2546,6 +2712,9 @@ export class AutopilotEngine { if (await this.#stopRunIfRequested()) { break; } + if (await this.#waitForUnknownExecution()) { + break; + } if (await this.#settlePauseIfRequested()) { break; } diff --git a/skills/autopilot/runtime/src/events.ts b/skills/autopilot/runtime/src/events.ts index d17ded8..d315175 100644 --- a/skills/autopilot/runtime/src/events.ts +++ b/skills/autopilot/runtime/src/events.ts @@ -1,4 +1,5 @@ import { randomUUID } from "node:crypto"; +import { parseExecutionAssurance, type ExecutionAssurance } from "./adapter-protocol.js"; import { AutopilotError } from "./errors.js"; import { expectBoolean, expectInteger, expectLiteral, expectRecord, expectString, expectStringArray } from "./json.js"; @@ -67,9 +68,22 @@ export type LifecycleEvent = readonly contextHash?: string; readonly contextJournalSequence?: number; readonly executionSupervised?: boolean; + readonly executionAssurance?: ExecutionAssurance; readonly deadline: string; readonly idempotencyKey: string; }) + | (EventBase & { + readonly type: "ATTEMPT_EXECUTION_ADMITTED"; + readonly itemId: string; + readonly attemptId: string; + readonly adapterName: string; + readonly adapterVersion: string; + readonly harnessVersion: string; + readonly adapterExecutionId: string; + readonly backendId: string; + readonly subjectId: string; + readonly harnessInstanceId?: string; + }) | (EventBase & { readonly type: "ATTEMPT_FINISHED"; readonly itemId: string; @@ -168,7 +182,7 @@ export type LifecycleEvent = const EVENT_TYPES = [ "CHARTER_COMPILED", "RECONCILIATION_STARTED", "RECONCILIATION_COMPLETED", "RUN_PAUSE_REQUESTED", "RUN_WAITING", "RUN_WOKEN", "RUN_RESUMED", "RUN_VERIFYING", "RUN_SUCCEEDED", "RUN_STOPPED", "WRAP_UP_STARTED", "WORKTREE_ADOPTED", "ITEM_READY", - "ATTEMPT_STARTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", "ITEM_VERIFIED", "ITEM_SATISFIED", + "ATTEMPT_STARTED", "ATTEMPT_EXECUTION_ADMITTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", "ITEM_VERIFIED", "ITEM_SATISFIED", "ITEM_BLOCKED", "ITEM_ABANDONED", "RESTACK_DESCENDANT_STARTED", "RESTACK_DESCENDANT_TREE_PREPARED", "RESTACK_DESCENDANT_VERIFIED", "RESTACK_PROVIDER_HEAD_CONFIRMED", "RESTACK_DESCENDANT_SATISFIED", "RESTACK_DESCENDANT_BLOCKED", "EFFECT_INTENDED", @@ -344,9 +358,28 @@ export function parseLifecycleEvent(value: unknown): LifecycleEvent { ...(object.executionSupervised === undefined ? {} : { executionSupervised: expectBoolean(object.executionSupervised, "event.executionSupervised"), }), + ...(object.executionAssurance === undefined ? {} : { + executionAssurance: parseExecutionAssurance(object.executionAssurance, "event.executionAssurance"), + }), deadline: expectString(object.deadline, "event.deadline"), idempotencyKey: expectString(object.idempotencyKey, "event.idempotencyKey"), }; + case "ATTEMPT_EXECUTION_ADMITTED": + return { + ...base, + type, + itemId: expectString(object.itemId, "event.itemId"), + attemptId: expectString(object.attemptId, "event.attemptId"), + adapterName: expectString(object.adapterName, "event.adapterName"), + adapterVersion: expectString(object.adapterVersion, "event.adapterVersion"), + harnessVersion: expectString(object.harnessVersion, "event.harnessVersion"), + adapterExecutionId: expectString(object.adapterExecutionId, "event.adapterExecutionId"), + backendId: expectString(object.backendId, "event.backendId"), + subjectId: expectString(object.subjectId, "event.subjectId"), + ...(object.harnessInstanceId === undefined ? {} : { + harnessInstanceId: expectString(object.harnessInstanceId, "event.harnessInstanceId"), + }), + }; case "ATTEMPT_FINISHED": return { ...base, diff --git a/skills/autopilot/runtime/src/reducer.ts b/skills/autopilot/runtime/src/reducer.ts index 48703fc..031a1d5 100644 --- a/skills/autopilot/runtime/src/reducer.ts +++ b/skills/autopilot/runtime/src/reducer.ts @@ -1,3 +1,4 @@ +import type { ExecutionAssurance } from "./adapter-protocol.js"; import type { RunCharter } from "./charter.js"; import { AutopilotError } from "./errors.js"; import type { LifecycleEvent, WaitingDetails } from "./events.js"; @@ -19,6 +20,16 @@ export interface AttemptProjection { readonly contextHash?: string; readonly contextJournalSequence?: number; readonly executionSupervised?: boolean; + readonly executionAssurance?: ExecutionAssurance; + readonly execution?: { + readonly adapterName: string; + readonly adapterVersion: string; + readonly harnessVersion: string; + readonly adapterExecutionId: string; + readonly backendId: string; + readonly subjectId: string; + readonly harnessInstanceId?: string; + }; readonly deadline: string; readonly idempotencyKey: string; readonly outcome?: "completed" | "failed" | "cancelled" | "timed-out" | "stale"; @@ -155,8 +166,12 @@ function assertRunTransition(current: RunState, event: LifecycleEvent): RunState if (event.waiting?.kind === "operator-pause") { return "WAITING"; } - if (current !== "RUNNING" && current !== "WAITING") { - throw new AutopilotError("ILLEGAL_TRANSITION", `${event.type} requires RUNNING or WAITING, received ${current}`); + if (current !== "RUNNING" && current !== "WAITING" + && !(current === "RECONCILING" && event.waiting?.kind === "execution-unknown")) { + throw new AutopilotError( + "ILLEGAL_TRANSITION", + `${event.type} requires RUNNING or WAITING, or RECONCILING for an unknown execution; received ${current}`, + ); } return "WAITING"; case "RUN_WOKEN": @@ -221,11 +236,38 @@ function transitionItem(item: ItemProjection, event: LifecycleEvent): ItemProjec ...(event.contextHash === undefined ? {} : { contextHash: event.contextHash }), ...(event.contextJournalSequence === undefined ? {} : { contextJournalSequence: event.contextJournalSequence }), ...(event.executionSupervised === undefined ? {} : { executionSupervised: event.executionSupervised }), + ...(event.executionAssurance === undefined ? {} : { executionAssurance: event.executionAssurance }), deadline: event.deadline, idempotencyKey: event.idempotencyKey, }, ], }; + case "ATTEMPT_EXECUTION_ADMITTED": { + if (item.state !== "ACTIVE") { + throw new AutopilotError("ILLEGAL_TRANSITION", `ATTEMPT_EXECUTION_ADMITTED cannot follow ${item.state}`); + } + const currentAttempt = item.attempts.at(-1); + if (currentAttempt?.attemptId !== event.attemptId || currentAttempt.execution !== undefined) { + throw new AutopilotError("ILLEGAL_TRANSITION", `execution admission is stale or duplicated for ${item.itemId}`); + } + return { + ...item, + attempts: item.attempts.map((attempt) => attempt.attemptId === event.attemptId + ? { + ...attempt, + execution: { + adapterName: event.adapterName, + adapterVersion: event.adapterVersion, + harnessVersion: event.harnessVersion, + adapterExecutionId: event.adapterExecutionId, + backendId: event.backendId, + subjectId: event.subjectId, + ...(event.harnessInstanceId === undefined ? {} : { harnessInstanceId: event.harnessInstanceId }), + }, + } + : attempt), + }; + } case "ATTEMPT_FINISHED": { if (item.state !== "ACTIVE") { throw new AutopilotError("ILLEGAL_TRANSITION", `ATTEMPT_FINISHED cannot follow ${item.state}`); @@ -449,7 +491,7 @@ export function reduce(projection: RunProjection, event: LifecycleEvent): RunPro const nextState = assertRunTransition(projection.state, event); let items = projection.items; const ordinaryItemLifecycle = [ - "DECISION_RECORDED", "ITEM_READY", "ATTEMPT_STARTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", + "DECISION_RECORDED", "ITEM_READY", "ATTEMPT_STARTED", "ATTEMPT_EXECUTION_ADMITTED", "ATTEMPT_FINISHED", "ITEM_VERIFYING", "ATTEMPT_PAUSED", "ITEM_VERIFIED", "ITEM_SATISFIED", "ITEM_BLOCKED", "ITEM_ABANDONED", ]; if (event.itemId !== undefined && projection.restacks[event.itemId] !== undefined diff --git a/skills/autopilot/runtime/src/report.ts b/skills/autopilot/runtime/src/report.ts index 99aa277..37af2d4 100644 --- a/skills/autopilot/runtime/src/report.ts +++ b/skills/autopilot/runtime/src/report.ts @@ -1,5 +1,6 @@ import { mkdir, writeFile } from "node:fs/promises"; import { join } from "node:path"; +import type { ExecutionAssurance } from "./adapter-protocol.js"; import type { RunCharter } from "./charter.js"; import { projectPredicateEvidence, type PredicateEvidenceEntry } from "./evidence-map.js"; import type { JournalRecord } from "./journal.js"; @@ -21,6 +22,16 @@ export interface RunReport { readonly blocker?: string; readonly attempts: number; readonly chargedAttempts: number; + readonly execution?: { + readonly assurance?: ExecutionAssurance; + readonly adapterName?: string; + readonly adapterVersion?: string; + readonly harnessVersion?: string; + readonly adapterExecutionId?: string; + readonly backendId?: string; + readonly subjectId?: string; + readonly harnessInstanceId?: string; + }; }[]; readonly restacks: readonly { readonly itemId: string; @@ -167,6 +178,8 @@ export async function writeReports( state: projection.state, items: ordinaryWork.map((item) => { const itemProjection = projection.items[item.id]; + const lastAttempt = itemProjection?.attempts.at(-1); + const execution = lastAttempt?.execution; return { itemId: item.id, state: itemProjection?.state ?? "PENDING", @@ -175,6 +188,12 @@ export async function writeReports( chargedAttempts: consumedAttempts(itemProjection), ...(itemProjection?.subject === undefined ? {} : { subject: itemProjection.subject }), ...(itemProjection?.blocker === undefined ? {} : { blocker: itemProjection.blocker }), + ...(lastAttempt?.executionAssurance === undefined && execution === undefined ? {} : { + execution: { + ...(lastAttempt?.executionAssurance === undefined ? {} : { assurance: lastAttempt.executionAssurance }), + ...(execution === undefined ? {} : execution), + }, + }), }; }), restacks: Object.values(projection.restacks).map((restack) => ({ diff --git a/skills/autopilot/runtime/test/adapter-contract.test.ts b/skills/autopilot/runtime/test/adapter-contract.test.ts index 828c334..dc2dc4a 100644 --- a/skills/autopilot/runtime/test/adapter-contract.test.ts +++ b/skills/autopilot/runtime/test/adapter-contract.test.ts @@ -3,7 +3,7 @@ import { mkdir, mkdtemp, readFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { delimiter, join } from "node:path"; import { test } from "node:test"; -import { parseAdapterMessage, type ExecutionRequest } from "../src/adapter-protocol.js"; +import { executionAssuranceFor, parseAdapterMessage, type ExecutionRequest } from "../src/adapter-protocol.js"; import { CliHarnessAdapter, parseReviewResult } from "../src/adapter-process.js"; import { boundUtf8, runProcess, terminateProcessTree } from "../src/process.js"; import { windowsRestartReattachmentAvailable } from "../src/windows-job.js"; @@ -366,7 +366,7 @@ test("review result parser accepts one structured marker and rejects contradicto assert.equal(contradictory, undefined); }); -test("adapter protocol parses a complete capability manifest", () => { +test("adapter protocol derives execution assurance for a legacy capability manifest", () => { const message = parseAdapterMessage(JSON.stringify({ protocolVersion: 1, type: "capabilities", @@ -388,4 +388,57 @@ test("adapter protocol parses a complete capability manifest", () => { }), 4096); assert.equal(message.type, "capabilities"); + assert.ok(message.type === "capabilities"); + assert.deepEqual(executionAssuranceFor(message.manifest, "implementation"), { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }); +}); + +test("adapter protocol parses separate implementation and review assurance", () => { + const message = parseAdapterMessage(JSON.stringify({ + protocolVersion: 1, + type: "capabilities", + manifest: { + protocolVersion: 1, + adapterName: "fake", + adapterVersion: "2", + harnessVersion: "1", + families: ["files.read"], + assurance: "cooperative", + unattended: true, + maxConcurrency: 1, + eventStreaming: true, + cancellation: true, + restartReattachment: false, + executionAssurance: { + schemaVersion: 1, + implementation: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, + review: { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + }, + restrictions: "cooperative", + limitations: [], + }, + }), 4096); + + assert.equal(message.type, "capabilities"); + assert.ok(message.type === "capabilities"); + assert.equal(executionAssuranceFor(message.manifest, "implementation").owner, "harness"); + assert.equal(executionAssuranceFor(message.manifest, "implementation").continuity, "same-harness-instance"); + assert.equal(executionAssuranceFor(message.manifest, "review").continuity, "session"); }); diff --git a/skills/autopilot/runtime/test/engine.test.ts b/skills/autopilot/runtime/test/engine.test.ts index bd48a0b..bd8dd8e 100644 --- a/skills/autopilot/runtime/test/engine.test.ts +++ b/skills/autopilot/runtime/test/engine.test.ts @@ -43,14 +43,21 @@ async function waitForFile(path: string, timeoutMs = 5_000): Promise { class BlockingAdapter implements HarnessPort { cancelCalls = 0; readonly launched: Promise; + readonly observing: Promise; #resolveLaunched: (() => void) | undefined; + #resolveObserving: (() => void) | undefined; #resolveObservation: ((observation: ExecutionObservation) => void) | undefined; #cancelledHandle: ExecutionHandle | undefined; + #cancellationFailuresRemaining: number; - constructor() { + constructor(cancellationFailures = 0) { + this.#cancellationFailuresRemaining = cancellationFailures; this.launched = new Promise((resolve) => { this.#resolveLaunched = resolve; }); + this.observing = new Promise((resolve) => { + this.#resolveObserving = resolve; + }); } async describe(): Promise { @@ -77,6 +84,7 @@ class BlockingAdapter implements HarnessPort { } async observe(handle: ExecutionHandle): Promise { + this.#resolveObserving?.(); if (this.#cancelledHandle?.adapterExecutionId === handle.adapterExecutionId) { return this.#cancelledObservation(handle); } @@ -100,6 +108,10 @@ class BlockingAdapter implements HarnessPort { async cancel(handle: ExecutionHandle): Promise { this.cancelCalls += 1; + if (this.#cancellationFailuresRemaining > 0) { + this.#cancellationFailuresRemaining -= 1; + return { protocolVersion: 1, accepted: false }; + } this.#cancelledHandle = handle; this.#resolveObservation?.(this.#cancelledObservation(handle)); return { protocolVersion: 1, accepted: true }; @@ -178,7 +190,73 @@ class TrackingAdapter extends FakeAdapter { } } +class CooperativeHarnessAdapter extends TrackingAdapter { + reattachments = 0; + + override async describe(): Promise { + return { + ...await super.describe(), + adapterName: "cooperative-harness", + executionAssurance: { + schemaVersion: 1, + implementation: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, + review: { + schemaVersion: 1, + owner: "runtime", + continuity: "session", + terminality: "cooperative", + admission: "single-shot", + }, + }, + }; + } + + override async launch(request: ExecutionRequest): Promise { + const handle = await super.launch(request); + return { + ...handle, + subject: { + schemaVersion: 1, + backendId: "cooperative-test", + subjectId: `subject-${request.attemptId}`, + harnessInstanceId: "harness-instance-1", + }, + }; + } + + async reattach(_request: ExecutionRequest): Promise { + this.reattachments += 1; + return undefined; + } +} + +class LostAdmissionAdapter extends CooperativeHarnessAdapter { + override async launch(_request: ExecutionRequest): Promise { + this.launches += 1; + throw new Error("connection lost after request emission"); + } +} + +class LostObservationAdapter extends CooperativeHarnessAdapter { + override async observe(_handle: ExecutionHandle): Promise { + throw new Error("harness connection lost before terminal response"); + } +} + +class FailingDescribeAdapter extends CooperativeHarnessAdapter { + override async describe(): Promise { + throw new Error("owning harness instance is unavailable"); + } +} + class ReattachableAdapter extends TrackingAdapter { + cancelCalls = 0; reattachments = 0; readonly #reattachedId = "reattached-execution"; readonly #reattachedStatus: ExecutionObservation["status"]; @@ -197,6 +275,11 @@ class ReattachableAdapter extends TrackingAdapter { return { protocolVersion: 1, adapterExecutionId: this.#reattachedId, startedAt: new Date().toISOString() }; } + override async cancel(handle: ExecutionHandle): Promise { + this.cancelCalls += 1; + return await super.cancel(handle); + } + override async observe(handle: ExecutionHandle): Promise { if (handle.adapterExecutionId !== this.#reattachedId) { return await super.observe(handle); @@ -376,6 +459,37 @@ test("engine cancellation stops active work before verification or runtime effec assert.equal(finalJournal.records.some(({ event }) => event.type === "EFFECT_INTENDED"), false); }); +test("engine retries cancellation on stop after a pause cancellation was not accepted", async () => { + const repository = await createRepository(); + const charter = sealCharter(proposedCharter(repository.root, repository.baseCommit, "single", "run-cancel-retry")); + const stateRoot = await mkdtemp(join(tmpdir(), "autopilot-engine-cancel-retry-")); + const runDirectory = join(stateRoot, "runs", charter.runId); + await mkdir(join(runDirectory, "receipts"), { recursive: true }); + await writeImmutableJson(join(runDirectory, "charter.json"), charter); + const journalPath = join(runDirectory, "events.jsonl"); + await appendEvent(journalPath, { ...event("compiled"), type: "CHARTER_COMPILED" }); + const journal = await readJournal(journalPath); + const adapter = new BlockingAdapter(1); + const engine = new AutopilotEngine({ + stateRoot, + runDirectory, + charter, + adapter, + records: journal.records, + projection: rebuildProjection(charter, journal.records), + }); + + const running = engine.run(); + await adapter.observing; + await engine.requestPause(); + assert.equal(adapter.cancelCalls, 1); + await engine.requestStop(); + const report = await running; + + assert.equal(adapter.cancelCalls, 2); + assert.equal(report.state, "STOPPED"); +}); + test("engine pauses only after cancellation quiesces and resumes with an uncharged replacement attempt", async () => { const repository = await createRepository(); const charter = sealCharter(proposedCharter(repository.root, repository.baseCommit, "single", "run-pause-active")); @@ -507,11 +621,31 @@ test("engine does not launch a replacement when interrupted execution quiescence expectedBaseCommit: observation.headCommit, contextHash, contextJournalSequence: context.sourceJournalSequence, + executionAssurance: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, deadline: new Date(Date.now() + 30_000).toISOString(), idempotencyKey: "attempt:orphaned", }); + await appendEvent(journalPath, { + ...event("cooperative subject admitted before harness loss"), + type: "ATTEMPT_EXECUTION_ADMITTED", + itemId: item.id, + attemptId, + adapterName: "cooperative-harness", + adapterVersion: "1", + harnessVersion: "1", + adapterExecutionId: "request-orphaned", + backendId: "cooperative-test", + subjectId: "subject-orphaned", + harnessInstanceId: "harness-instance-old", + }); const journal = await readJournal(journalPath); - const adapter = new TrackingAdapter(); + const adapter = new CooperativeHarnessAdapter(); const engine = new AutopilotEngine({ stateRoot, runDirectory, @@ -531,11 +665,104 @@ test("engine does not launch a replacement when interrupted execution quiescence assert.equal(report.items[0]?.attempts, 1); assert.equal(report.items[0]?.chargedAttempts, 1); assert.equal(adapter.launches, 0); + assert.equal(adapter.reattachments, 0); assert.equal(finalJournal.records.some(({ event: lifecycleEvent }) => lifecycleEvent.type === "RUN_WAITING" && lifecycleEvent.waiting?.kind === "operator-pause" ), false); }); +test("engine treats lost single-shot admission as unknown without retrying launch", async () => { + const adapter = new LostAdmissionAdapter(); + const { report, runDirectory } = await runMode("single", false, false, adapter); + const journal = await readJournal(join(runDirectory, "events.jsonl")); + + assert.equal(report.state, "WAITING"); + assert.equal(report.waiting?.kind, "execution-unknown"); + assert.equal(report.items[0]?.blocker, "EXECUTION_STATE_UNKNOWN"); + assert.equal(report.items[0]?.execution?.assurance?.continuity, "same-harness-instance"); + assert.equal(adapter.launches, 1); + assert.equal(journal.records.filter(({ event: lifecycleEvent }) => + lifecycleEvent.type === "ATTEMPT_STARTED" + ).length, 1); + assert.equal(journal.records.some(({ event: lifecycleEvent }) => + lifecycleEvent.type === "ATTEMPT_EXECUTION_ADMITTED" + ), false); +}); + +test("engine treats a missing cooperative terminal response as unknown without replacement", async () => { + const adapter = new LostObservationAdapter(); + const { report, runDirectory } = await runMode("single", false, false, adapter); + const journal = await readJournal(join(runDirectory, "events.jsonl")); + + assert.equal(report.state, "WAITING"); + assert.equal(report.waiting?.kind, "execution-unknown"); + assert.equal(report.items[0]?.blocker, "EXECUTION_STATE_UNKNOWN"); + assert.equal(report.items[0]?.execution?.subjectId?.startsWith("subject-"), true); + assert.equal(adapter.launches, 1); + assert.equal(journal.records.filter(({ event: lifecycleEvent }) => + lifecycleEvent.type === "ATTEMPT_STARTED" + ).length, 1); + assert.equal(journal.records.some(({ event: lifecycleEvent }) => + lifecycleEvent.type === "ATTEMPT_EXECUTION_ADMITTED" + ), true); + assert.equal(journal.records.some(({ event: lifecycleEvent }) => + lifecycleEvent.type === "ATTEMPT_FINISHED" + ), false); +}); + +test("engine keeps an admitted execution unknown when adapter discovery fails on restart", async () => { + const repository = await createRepository(); + const charter = sealCharter(proposedCharter(repository.root, repository.baseCommit, "single", "run-describe-loss")); + const item = charter.work[0]; + assert.ok(item !== undefined); + const stateRoot = await mkdtemp(join(tmpdir(), "autopilot-engine-describe-loss-")); + const runDirectory = join(stateRoot, "runs", charter.runId); + await mkdir(join(runDirectory, "receipts"), { recursive: true }); + const journalPath = join(runDirectory, "events.jsonl"); + await appendEvent(journalPath, { ...event("compiled"), type: "CHARTER_COMPILED" }); + await appendEvent(journalPath, { ...event("reconciling"), type: "RECONCILIATION_STARTED" }); + await appendEvent(journalPath, { ...event("running"), type: "RECONCILIATION_COMPLETED" }); + await appendEvent(journalPath, { ...event("ready"), type: "ITEM_READY", itemId: item.id }); + const attemptId = "attempt-describe-loss"; + await appendEvent(journalPath, { + ...event("cooperative admission intent"), type: "ATTEMPT_STARTED", itemId: item.id, attemptId, leaseEpoch: 1, + expectedBaseCommit: repository.baseCommit, + executionAssurance: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, + deadline: new Date(Date.now() + 30_000).toISOString(), idempotencyKey: "attempt:describe-loss", + }); + await appendEvent(journalPath, { + ...event("cooperative subject admitted"), type: "ATTEMPT_EXECUTION_ADMITTED", itemId: item.id, attemptId, + adapterName: "cooperative-harness", adapterVersion: "1", harnessVersion: "1", + adapterExecutionId: "request-describe-loss", backendId: "cooperative-test", subjectId: "subject-describe-loss", + harnessInstanceId: "harness-instance-old", + }); + const journal = await readJournal(journalPath); + const adapter = new FailingDescribeAdapter(); + const engine = new AutopilotEngine({ + stateRoot, + runDirectory, + charter, + adapter, + records: journal.records, + projection: rebuildProjection(charter, journal.records), + }); + + const report = await engine.run(); + const finalJournal = await readJournal(journalPath); + + assert.equal(report.state, "WAITING"); + assert.equal(report.waiting?.kind, "execution-unknown"); + assert.equal(report.items[0]?.blocker, "EXECUTION_STATE_UNKNOWN"); + assert.equal(finalJournal.records.some(({ event: lifecycleEvent }) => lifecycleEvent.type === "RUN_STOPPED"), false); + assert.equal(adapter.launches, 0); +}); + test("engine reattaches an interrupted supervised execution before launching a replacement", async () => { const repository = await createRepository(); const charter = sealCharter(proposedCharter(repository.root, repository.baseCommit, "single", "run-supervised-execution")); @@ -680,6 +907,7 @@ test("engine preserves an uncharged pause when supervised cancellation is first assert.equal(report.items[0]?.attempts, 1); assert.equal(report.items[0]?.chargedAttempts, 0); assert.equal(adapter.reattachments, 1); + assert.equal(adapter.cancelCalls, 1); assert.equal(finalJournal.records.some(({ event: lifecycleEvent }) => lifecycleEvent.type === "ATTEMPT_PAUSED" && lifecycleEvent.attemptId === attemptId ), true); diff --git a/skills/autopilot/runtime/test/reducer.test.ts b/skills/autopilot/runtime/test/reducer.test.ts index 7843235..3ecd11c 100644 --- a/skills/autopilot/runtime/test/reducer.test.ts +++ b/skills/autopilot/runtime/test/reducer.test.ts @@ -284,6 +284,50 @@ test("reducer projects a nonterminal operator pause without charging its cancell assert.equal(projection.pauseRequestId, undefined); }); +test("reducer binds execution assurance and exact admitted subject to the current attempt", async () => { + const repository = await createRepository(); + const charter = sealCharter(proposedCharter(repository.root, repository.baseCommit)); + const attemptId = "attempt-admitted"; + let projection = initialProjection(charter); + for (const lifecycleEvent of [ + { ...base("reconcile"), type: "RECONCILIATION_STARTED" } as LifecycleEvent, + { ...base("running"), type: "RECONCILIATION_COMPLETED" } as LifecycleEvent, + { ...base("ready"), type: "ITEM_READY", itemId: "item-1" } as LifecycleEvent, + { + ...base("intent"), type: "ATTEMPT_STARTED", itemId: "item-1", attemptId, leaseEpoch: 1, + expectedBaseCommit: repository.baseCommit, + executionAssurance: { + schemaVersion: 1, + owner: "harness", + continuity: "same-harness-instance", + terminality: "cooperative", + admission: "single-shot", + }, + deadline: new Date(Date.now() + 1_000).toISOString(), idempotencyKey: "attempt:admitted", + } as LifecycleEvent, + { + ...base("admitted"), type: "ATTEMPT_EXECUTION_ADMITTED", itemId: "item-1", attemptId, + adapterName: "pi", adapterVersion: "2", harnessVersion: "0.84.4", + adapterExecutionId: "request-1", backendId: "pi-subagents", subjectId: "node-1", + harnessInstanceId: "owner-run-1", + } as LifecycleEvent, + ]) { + projection = reduce(projection, lifecycleEvent); + } + + const attempt = projection.items["item-1"]?.attempts.at(-1); + assert.equal(attempt?.executionAssurance?.continuity, "same-harness-instance"); + assert.equal(attempt?.execution?.backendId, "pi-subagents"); + assert.equal(attempt?.execution?.subjectId, "node-1"); + assert.equal(attempt?.execution?.harnessInstanceId, "owner-run-1"); + assert.throws(() => reduce(projection, { + ...base("duplicate"), type: "ATTEMPT_EXECUTION_ADMITTED", itemId: "item-1", attemptId, + adapterName: "pi", adapterVersion: "2", harnessVersion: "0.84.4", + adapterExecutionId: "request-1", backendId: "pi-subagents", subjectId: "node-1", + harnessInstanceId: "owner-run-1", + }), /duplicated/); +}); + test("reducer charges a naturally completed attempt when pause races with completion", async () => { const repository = await createRepository(); const charter = sealCharter(proposedCharter(repository.root, repository.baseCommit));