From 8b6910732992521bcf907ce39101f8a633a5ba8d Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 26 Aug 2026 19:10:36 +0200 Subject: [PATCH] feat(agent): make durable drive total --- packages/agent/docs/runtime-simplification.md | 2 +- .../work-packages/05-direct-durable-drive.md | 4 +- packages/agent/src/harness/agent-harness.ts | 3 +- packages/agent/src/harness/runtime/drive.ts | 106 +++ .../src/harness/runtime/drive/deferred.ts | 58 +- .../src/harness/runtime/drive/reconcile.ts | 168 ++++ .../src/harness/runtime/drive/recovery.ts | 60 +- .../src/harness/runtime/drive/structural.ts | 48 +- .../src/harness/runtime/drive/terminal.ts | 5 - packages/agent/src/harness/runtime/lane.ts | 118 ++- packages/agent/src/harness/runtime/restore.ts | 11 +- packages/agent/src/harness/runtime/types.ts | 5 + packages/agent/src/harness/session/types.ts | 2 - packages/agent/src/harness/telemetry.ts | 2 +- .../harness/runtime/drive-generation.test.ts | 2 +- .../harness/runtime/drive-reconcile.test.ts | 747 ++++++++++++++++++ .../runtime/drive-retry-deferred.test.ts | 2 +- .../harness/runtime/drive-structural.test.ts | 167 +++- .../harness/runtime/drive-terminal.test.ts | 11 +- .../test/harness/runtime/drive-tools.test.ts | 8 +- .../agent/test/harness/runtime/lane.test.ts | 7 +- .../test/harness/runtime/restore.test.ts | 20 + .../agent/test/harness/runtime/watch.test.ts | 15 +- packages/agent/test/harness/types.test.ts | 2 +- .../src/experimental/harness-wire-adapter.ts | 20 +- .../experimental-harness-wire-adapter.test.ts | 13 + packages/protocol/src/harness.ts | 7 +- 27 files changed, 1470 insertions(+), 143 deletions(-) create mode 100644 packages/agent/src/harness/runtime/drive.ts create mode 100644 packages/agent/src/harness/runtime/drive/reconcile.ts create mode 100644 packages/agent/test/harness/runtime/drive-reconcile.test.ts diff --git a/packages/agent/docs/runtime-simplification.md b/packages/agent/docs/runtime-simplification.md index 5fc33a93d..bee079587 100644 --- a/packages/agent/docs/runtime-simplification.md +++ b/packages/agent/docs/runtime-simplification.md @@ -41,7 +41,7 @@ M6 established structural execution on the simplified substrate. Its first coher - per-request structural intent and usage settlement, attempt retry/recovery, hook decisions, compaction/navigation publication, and terminal cleanup; - focused coverage for threshold/overflow entry, split-turn request accounting, generated and hook results, retry/cap/recovery, model absence, mid-request durable cancellation, navigation, and preparation corruption. -R1a, R2, and R3 then moved queued input to the lane, replaced hydrated family outcomes with immutable operation records, and collapsed the family cross-product to 13 neutral leaves. R1b replaced the transitional checkpoint drains and marker fields with one shared atomic boundary planner, a transcript-derived threshold guard, and off-line finish-hook replanning. At the reviewed R1b checkpoint the runtime is 5,835 lines; `runtime/drive` is 3,870 lines; and the structural module is 1,170 lines, still down 290 lines (19.9%) from its 1,460-line foundation. The focused runtime set passes 194 tests across 15 files, with 44 memory/JSONL and 30 SQLite repository-conformance tests also passing. M7 reconciliation/total dispatch and M8 public surfaces have not started. +R1a, R2, and R3 then moved queued input to the lane, replaced hydrated family outcomes with immutable operation records, and collapsed the family cross-product to 13 neutral leaves. R1b replaced the transitional checkpoint drains and marker fields with one shared atomic boundary planner, a transcript-derived threshold guard, and off-line finish-hook replanning. M7 removed drained control, added atomic drain-and-return cancellation, reconciled all 13 leaves, and installed the total direct dispatcher. At the reviewed M7 checkpoint the runtime is 6,280 lines; `runtime/drive` is 4,115 lines; `structural.ts` is 1,194 lines, still down 266 lines (18.2%) from its 1,460-line foundation; and the new total dispatcher plus reconciliation are 274 lines. M8 public surfaces have not started. Public drive remains disabled. Keep the visible durable procedure order: `prepare → publish intent → perform effect → publish outcome`. diff --git a/packages/agent/docs/work-packages/05-direct-durable-drive.md b/packages/agent/docs/work-packages/05-direct-durable-drive.md index 9397dd8e8..410d3ef72 100644 --- a/packages/agent/docs/work-packages/05-direct-durable-drive.md +++ b/packages/agent/docs/work-packages/05-direct-durable-drive.md @@ -1,6 +1,6 @@ # WP05 — Direct durable drive -**Status: M5 complete; M6 structural foundation committed; R1a lane-owned inbox relocation, R2 record-only outcomes, reviewed R3 family-neutral leaves, and the reviewed R1b atomic boundary planner implemented; M7 reconciliation next, then M8 (public surfaces), M9 (documentation reconciliation), and M10 (provider KV-cache identity review).** +**Status: M5 complete; M6 structural coverage complete; R1a lane-owned inbox relocation, R2 record-only outcomes, reviewed R3 family-neutral leaves, the reviewed R1b atomic boundary planner, and reviewed M7 cancellation reconciliation/total dispatch implemented; M8 public surfaces next, then M9 documentation reconciliation and M10 provider KV-cache identity review.** WP06's Session/Branch/Lane separation is already part of the foundation. Public drive remains disabled until R1–R3, cancellation reconciliation, the total drive switch, every public execution surface, and the provider KV-cache identity review are complete. @@ -381,7 +381,7 @@ Everything the M6 foundation covered, re-expressed over neutral leaves, plus: re ### Goal -Make the internal graph total without public wiring. +**Implemented and reviewed.** Make the internal graph total without public wiring. Create `src/harness/runtime/drive/reconcile.ts`, `src/harness/runtime/drive.ts`, and focused reconciliation and switch tests. diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index d4a9fb530..a5eb4e414 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -212,7 +212,6 @@ export interface LaneSnapshot { status: OperationStatus; retry?: { attempt: number; maxAttempts: number; nextAttemptAt: number }; deferred?: { handle: DeferredHandle; poll: number }; - drained?: { steer: QueuedItem[]; followUp: QueuedItem[] }; streamingMessage?: AssistantMessage; runningTools: { toolCallId: string; @@ -241,7 +240,7 @@ export type HarnessEventPayload = | { type: "run_start"; runId: string } | { type: "run_resume"; runId: string } | { type: "run_suspend"; runId: string; reason: "deferred"; deferred: DeferredHandle } - | { type: "run_abort"; runId: string; steer: AgentMessage[]; followUp: AgentMessage[] } + | { type: "operation_abort"; operationId: string; steer: AgentMessage[]; followUp: AgentMessage[] } | ({ type: "run_end"; runId: string; fromTipId: string | null; tipId: string | null } & ( | { status: "completed" | "aborted"; error?: never } | { status: "failed"; error: OperationError } diff --git a/packages/agent/src/harness/runtime/drive.ts b/packages/agent/src/harness/runtime/drive.ts new file mode 100644 index 000000000..0a3a9729b --- /dev/null +++ b/packages/agent/src/harness/runtime/drive.ts @@ -0,0 +1,106 @@ +import type { DriveOutcome } from "../agent-harness.ts"; +import { AbortRequested } from "../execution/effect-gate.ts"; +import { SessionInvariantError } from "../session/session.ts"; +import { runCheckpoint, startRun } from "./drive/checkpoint.ts"; +import { runDeferred } from "./drive/deferred.ts"; +import { runGeneration } from "./drive/generation.ts"; +import { reconcileOperation } from "./drive/reconcile.ts"; +import { recoverAssistantGeneration } from "./drive/recovery.ts"; +import { + commitNavigation, + recoverStructuralGeneration, + runStructuralDecision, + runStructuralGeneration, + runStructuralRetryWait, +} from "./drive/structural.ts"; +import { runTools } from "./drive/tools.ts"; +import type { Lane } from "./lane.ts"; +import type { Drive, ProcedureResult } from "./types.ts"; + +function currentOperation(lane: Lane, drive: Drive) { + const operation = lane.state.operation; + if (operation === null || operation.meta.operationId !== drive.operationId) { + throw new SessionInvariantError(`Drive ${drive.operationId} has no matching current operation`); + } + return operation; +} + +/** Drive one installed pass through direct durable procedures until settlement or a durable wait. */ +export async function driveOperation( + lane: Lane, + drive: Drive, +): Promise { + let operation = currentOperation(lane, drive); + if (operation.state.control.status === "running") { + try { + await lane.hooks.runWithGate( + "before_drive", + { lane: lane.name, runId: drive.operationId, operation: operation.meta.intent.kind }, + drive.gate, + drive.context, + ); + } catch (error) { + if (!(error instanceof AbortRequested)) throw error; + await error.cancellation; + } + } + + for (;;) { + operation = currentOperation(lane, drive); + const state = operation.state; + let result: ProcedureResult; + try { + if (state.control.status === "cancel_requested") { + result = await reconcileOperation(lane, drive); + } else + switch (state.at) { + case "starting": + result = await startRun(lane, drive, state); + break; + case "checkpoint": + result = await runCheckpoint(lane, drive, state); + break; + case "assistant.ready": + case "assistant.retry_wait": + result = await runGeneration(lane, drive, state); + break; + case "assistant.effect_pending": + result = await recoverAssistantGeneration(lane, drive, state); + break; + case "tools": + result = await runTools(lane, drive, state); + break; + case "deferred.suspended": + case "deferred.effect_pending": + result = await runDeferred(lane, drive, state); + break; + case "summary.deciding": + result = await runStructuralDecision(lane, drive, state); + break; + case "summary.ready": + result = await runStructuralGeneration(lane, drive, state); + break; + case "summary.effect_pending": + result = await recoverStructuralGeneration(lane, drive, state); + break; + case "summary.retry_wait": + result = await runStructuralRetryWait(lane, drive, state); + break; + case "navigation.ready_to_commit": + result = await commitNavigation(lane, drive, state); + break; + } + } catch (error) { + if (!(error instanceof AbortRequested)) throw error; + await error.cancellation; + result = { kind: "continue" }; + } + + if (result.kind === "settled") return { kind: "settled", outcome: result.outcome }; + if (result.kind === "waiting") return result.outcome; + const next = currentOperation(lane, drive).state; + if (next === state && next.control.status !== "cancel_requested") { + throw new SessionInvariantError(`Drive procedure made no progress from ${state.at}`); + } + } +} diff --git a/packages/agent/src/harness/runtime/drive/deferred.ts b/packages/agent/src/harness/runtime/drive/deferred.ts index fa4552e0f..3b4240b8c 100644 --- a/packages/agent/src/harness/runtime/drive/deferred.ts +++ b/packages/agent/src/harness/runtime/drive/deferred.ts @@ -1,5 +1,5 @@ import type { Api, DeferredHandle, Model } from "@earendil-works/pi-ai"; -import { withAbortSignal } from "../../context.ts"; +import { type Context, withAbortSignal } from "../../context.ts"; import { consumeAssistantStream } from "../../execution/assistant.ts"; import { applyStreamOptionsPatch } from "../../hooks.ts"; import { SessionInvariantError } from "../../session/session.ts"; @@ -10,6 +10,7 @@ import { type OperationError, type OperationScope, operationScopeOf, + type SessionReader, type SettledAssistantMessage, } from "../../session/types.ts"; import { deleteList, pendingAssistantFrames } from "../../session/values.ts"; @@ -44,6 +45,33 @@ function configurationError(identity: LaneConfiguration["model"]): OperationErro }; } +export async function readDeferredSourceHandle( + reader: SessionReader, + deferred: DeferredLeaf, + context: Context, +): Promise { + const source = (await reader.getEntries([deferred.sourceEntryId], context)).get(deferred.sourceEntryId); + if ( + source?.type !== "message" || + source.message.role !== "assistant" || + source.message.stopReason !== "deferred" || + source.message.deferred === undefined + ) { + throw new SessionInvariantError(`Deferred source ${deferred.sourceEntryId} is missing its assistant handle`); + } + const handle = source.message.deferred; + const identity = deferred.configuration.model; + if ( + handle.id.length === 0 || + handle.provider !== identity.provider || + handle.modelId !== identity.modelId || + handle.api !== source.message.api + ) { + throw new SessionInvariantError(`Deferred source ${deferred.sourceEntryId} has an invalid handle`); + } + return handle; +} + async function readSourceHandle( lane: Lane, drive: Drive, @@ -51,30 +79,10 @@ async function readSourceHandle( ): Promise> { return lane.continueOperation( deferred, - async (_state, _current, _meta, reader) => { - const source = (await reader.getEntries([deferred.sourceEntryId], drive.context)).get(deferred.sourceEntryId); - if ( - source?.type !== "message" || - source.message.role !== "assistant" || - source.message.stopReason !== "deferred" || - source.message.deferred === undefined - ) { - throw new SessionInvariantError( - `Deferred source ${deferred.sourceEntryId} is missing its assistant handle`, - ); - } - const handle = source.message.deferred; - const identity = deferred.configuration.model; - if ( - handle.id.length === 0 || - handle.provider !== identity.provider || - handle.modelId !== identity.modelId || - handle.api !== source.message.api - ) { - throw new SessionInvariantError(`Deferred source ${deferred.sourceEntryId} has an invalid handle`); - } - return { kind: "return", result: handle }; - }, + async (_state, _current, _meta, reader) => ({ + kind: "return", + result: await readDeferredSourceHandle(reader, deferred, drive.context), + }), drive.context, ); } diff --git a/packages/agent/src/harness/runtime/drive/reconcile.ts b/packages/agent/src/harness/runtime/drive/reconcile.ts new file mode 100644 index 000000000..28ba8ad01 --- /dev/null +++ b/packages/agent/src/harness/runtime/drive/reconcile.ts @@ -0,0 +1,168 @@ +import type { DeferredHandle } from "@earendil-works/pi-ai"; +import type { HarnessEvent } from "../../agent-harness.ts"; +import { SessionInvariantError } from "../../session/session.ts"; +import type { + DeferredEffectPendingOperation, + DeferredSuspendedOperation, + OperationState, +} from "../../session/types.ts"; +import type { Lane } from "../lane.ts"; +import type { Drive, ProcedureResult } from "../types.ts"; +import { readDeferredSourceHandle } from "./deferred.ts"; +import { recoverCancelledAssistantEffect } from "./recovery.ts"; +import { operationCleanupWrites, operationResultRecord } from "./terminal.ts"; +import { runTools } from "./tools.ts"; + +async function cancelDeferredBestEffort( + lane: Lane, + drive: Drive, + deferred: DeferredSuspendedOperation | DeferredEffectPendingOperation, + handle: DeferredHandle, +): Promise { + const identity = deferred.configuration.model; + const model = lane.models.getModel(identity.provider, identity.modelId); + if (model === undefined) return; + try { + await lane.models.cancelDeferred(model, handle, { + signal: drive.closeSignal, + telemetryContext: drive.context.telemetryContext, + timeoutMs: deferred.streamOptions.timeoutMs, + maxRetries: deferred.streamOptions.maxRetries, + maxRetryDelayMs: deferred.streamOptions.maxRetryDelayMs, + headers: deferred.streamOptions.headers, + }); + } catch { + // Remote cancellation is best-effort; durable local reconciliation must continue. + } +} + +async function readDeferredHandle( + lane: Lane, + drive: Drive, + deferred: DeferredSuspendedOperation | DeferredEffectPendingOperation, +): Promise { + return lane.settleOperation( + deferred, + async (_state, _current, _meta, reader) => ({ + kind: "return", + result: await readDeferredSourceHandle(reader, deferred, drive.context), + }), + drive.context, + ); +} + +async function publishAbortedTerminal( + lane: Lane, + drive: Drive, + capability: OperationState, +): Promise { + return lane.settleOperation( + capability, + async (state, current, meta, reader) => { + if (current.control.status !== "cancel_requested") { + throw new SessionInvariantError("Cancellation reconciliation requires cancelled durable control"); + } + const record = operationResultRecord(meta, "aborted", state.tipId); + const cleanup = await operationCleanupWrites(reader, drive.operationId, current, drive.context); + const events: HarnessEvent[] = []; + if (meta.intent.kind === "run") { + switch (current.at) { + case "summary.deciding": + case "summary.ready": + case "summary.effect_pending": + case "summary.retry_wait": + if (current.task.boundary.kind !== "resume_checkpoint" || current.task.reason === undefined) { + throw new SessionInvariantError("Cancelled run summary has an invalid result boundary"); + } + events.push({ + type: "compaction_end", + lane: lane.name, + runId: drive.operationId, + reason: current.task.reason, + status: "aborted", + }); + break; + default: + break; + } + events.push({ + type: "run_end", + lane: lane.name, + runId: drive.operationId, + status: "aborted", + fromTipId: meta.sourceTipId, + tipId: state.tipId, + }); + } else if (meta.intent.kind === "compaction") { + events.push({ + type: "compaction_end", + lane: lane.name, + runId: drive.operationId, + reason: "manual", + status: "aborted", + }); + } else { + events.push({ + type: "navigation_end", + lane: lane.name, + runId: drive.operationId, + status: "aborted", + fromTipId: meta.sourceTipId, + tipId: state.tipId, + }); + } + return { + kind: "finish", + writes: cleanup, + record, + materialize: () => ({ kind: "settled", outcome: record }) as const, + events: () => events, + }; + }, + drive.context, + ); +} + +/** Advance one cancelled durable leaf without starting new ordinary work. */ +export async function reconcileOperation( + lane: Lane, + drive: Drive, +): Promise { + const operation = lane.state.operation; + if (operation === null || operation.meta.operationId !== drive.operationId) { + throw new SessionInvariantError(`Drive ${drive.operationId} has no matching operation to reconcile`); + } + if (operation.state.control.status !== "cancel_requested") { + throw new SessionInvariantError(`Operation ${drive.operationId} is not cancelled`); + } + drive.beginAbort(Promise.resolve()); + drive.signalAbort(); + + const state = operation.state; + switch (state.at) { + case "assistant.effect_pending": + return recoverCancelledAssistantEffect(lane, drive, state); + case "tools": + return runTools(lane, drive, state); + case "deferred.suspended": { + const handle = await readDeferredHandle(lane, drive, state); + await cancelDeferredBestEffort(lane, drive, state, handle); + return publishAbortedTerminal(lane, drive, state); + } + case "deferred.effect_pending": { + const handle = await readDeferredHandle(lane, drive, state); + await cancelDeferredBestEffort(lane, drive, state, handle); + return recoverCancelledAssistantEffect(lane, drive, state); + } + case "starting": + case "checkpoint": + case "assistant.ready": + case "assistant.retry_wait": + case "summary.deciding": + case "summary.ready": + case "summary.effect_pending": + case "summary.retry_wait": + case "navigation.ready_to_commit": + return publishAbortedTerminal(lane, drive, state); + } +} diff --git a/packages/agent/src/harness/runtime/drive/recovery.ts b/packages/agent/src/harness/runtime/drive/recovery.ts index dcab871c3..d193fdcc3 100644 --- a/packages/agent/src/harness/runtime/drive/recovery.ts +++ b/packages/agent/src/harness/runtime/drive/recovery.ts @@ -1,5 +1,10 @@ import { type AssistantMessage, reduceAssistantMessageFrames } from "@earendil-works/pi-ai"; -import type { AssistantEffectPendingOperation, SettledAssistantMessage } from "../../session/types.ts"; +import type { + AssistantEffectPendingOperation, + DeferredEffectPendingOperation, + LaneConfiguration, + SettledAssistantMessage, +} from "../../session/types.ts"; import type { Lane } from "../lane.ts"; import { readAssistantFrames } from "../progress.ts"; import type { Drive, ProcedureResult } from "../types.ts"; @@ -15,7 +20,7 @@ const ZERO_USAGE = { }; function interruptedAssistantMessage( - generation: AssistantEffectPendingOperation, + identity: LaneConfiguration["model"], partial: AssistantMessage | undefined, ): SettledAssistantMessage { const warning = @@ -25,8 +30,8 @@ function interruptedAssistantMessage( role: "assistant", content: [], api: "unknown", - provider: generation.generationContext.configuration.model.provider, - model: generation.generationContext.configuration.model.modelId, + provider: identity.provider, + model: identity.modelId, usage: ZERO_USAGE, stopReason: "error", errorMessage: warning, @@ -51,7 +56,10 @@ export async function recoverAssistantGeneration( + lane: Lane, + drive: Drive, + effect: AssistantEffectPendingOperation | DeferredEffectPendingOperation, +): Promise { + const frames = await lane.settleOperation( + effect, + async (_state, _current, _meta, reader) => ({ + kind: "return", + result: await readAssistantFrames(reader, drive.operationId, effect.responseEntryId, drive.context), + }), + drive.context, + ); + const identity = + effect.at === "assistant.effect_pending" + ? effect.generationContext.configuration.model + : effect.configuration.model; + const message = interruptedAssistantMessage(identity, reduceAssistantMessageFrames(frames)); + await lane.emitBatch( + [ + { + type: "message_start", + lane: lane.name, + runId: drive.operationId, + message, + recovery: true, + }, + { + type: "message_end", + lane: lane.name, + runId: drive.operationId, + message, + entryId: effect.responseEntryId, + recovery: true, + }, + ], + drive.context, + ); + return publishResponse(lane, drive, effect, message, { recovery: true }); +} diff --git a/packages/agent/src/harness/runtime/drive/structural.ts b/packages/agent/src/harness/runtime/drive/structural.ts index cd7662363..4edab345c 100644 --- a/packages/agent/src/harness/runtime/drive/structural.ts +++ b/packages/agent/src/harness/runtime/drive/structural.ts @@ -259,17 +259,9 @@ async function publishStructuralOutcome( const entryWriteIndex = writes.length; writes.push(insertEntry(entry), setValue(branchTip(lane.name), outcome.resultEntryId)); terminalTipId = outcome.resultEntryId; - baseEvents.push((commit) => [ - ...committedEntryEvents([entry], commit, lane.name, drive.operationId, entryWriteIndex), - { - type: "compaction_end", - lane: lane.name, - runId: drive.operationId, - reason: compactionReason(current.task), - status: "completed", - entryId: outcome.resultEntryId, - }, - ]); + baseEvents.push((commit) => + committedEntryEvents([entry], commit, lane.name, drive.operationId, entryWriteIndex), + ); } else if (outcome.kind === "branch_summary") { const boundary = navigationBoundary(current.task); const entry: NewEntry = { @@ -292,6 +284,37 @@ async function publishStructuralOutcome( ); } + const attempt = + current.at === "summary.ready" + ? current.nextAttempt + : current.at === "summary.effect_pending" + ? current.attempt + : undefined; + if (attempt !== undefined && attempt > 1) { + baseEvents.push(() => [ + { + type: "retry_end", + lane: lane.name, + runId: drive.operationId, + step: current.task.taskId, + attempt, + success: outcome.kind === "compaction" || outcome.kind === "branch_summary", + ...(outcome.kind === "failed" ? { finalError: outcome.error.message } : {}), + }, + ]); + } + if (outcome.kind === "compaction") { + baseEvents.push(() => [ + { + type: "compaction_end", + lane: lane.name, + runId: drive.operationId, + reason: compactionReason(current.task), + status: "completed", + entryId: outcome.resultEntryId, + }, + ]); + } const events = (commit: CommitResult): HarnessEvent[] => baseEvents.flatMap((materialize) => materialize(commit)); switch (current.task.boundary.kind) { @@ -406,7 +429,8 @@ async function publishStructuralOutcome( writes: [...writes, ...cleanup], record, materialize: () => ({ kind: "settled", outcome: record }) as const, - events: () => [ + events: (commit) => [ + ...events(commit), compactionEnd, { type: "run_end", diff --git a/packages/agent/src/harness/runtime/drive/terminal.ts b/packages/agent/src/harness/runtime/drive/terminal.ts index 8521af5f5..44f3dfff2 100644 --- a/packages/agent/src/harness/runtime/drive/terminal.ts +++ b/packages/agent/src/harness/runtime/drive/terminal.ts @@ -42,11 +42,6 @@ export async function operationCleanupWrites( if (call.status === "outcome_ready") pendingIds.add(call.resultEntryId); } } - if (state.control.status === "cancel_requested") { - for (const id of state.control.drainedSteer) pendingIds.add(id); - for (const id of state.control.drainedFollowUp) pendingIds.add(id); - } - let frameDelete: Write | undefined; if (state.at === "assistant.effect_pending" || state.at === "deferred.effect_pending") { frameDelete = deleteList(pendingAssistantFrames(operationId, state.responseEntryId)); diff --git a/packages/agent/src/harness/runtime/lane.ts b/packages/agent/src/harness/runtime/lane.ts index b4d7882ad..c8f42024f 100644 --- a/packages/agent/src/harness/runtime/lane.ts +++ b/packages/agent/src/harness/runtime/lane.ts @@ -7,6 +7,7 @@ import { } from "@earendil-works/pi-ai"; import type { AgentMessage, ThinkingLevel } from "../../types.ts"; import type { + AbortRequestResult, AgentLane, HarnessEvent, LaneConfigEventPayload, @@ -602,6 +603,104 @@ export class Lane implements AgentLane { throw new SliceNotImplemented("drive"); } + /** Package-private durable cancellation primitive. Public exposure remains guarded until M8. */ + async requestOperationAbort(operationId: string, context: Context): Promise { + if (this.closedError instanceof HarnessClosed) { + return Result.err(new Closed({ message: this.closedError.message })); + } + this.assertOpen(); + + const drive = this.activeDrive?.operationId === operationId ? this.activeDrive : undefined; + let resolveCancellation!: () => void; + let rejectCancellation!: (error: unknown) => void; + const cancellation = new Promise((resolve, reject) => { + resolveCancellation = resolve; + rejectCancellation = reject; + }); + void cancellation.catch(() => {}); + drive?.beginAbort(cancellation); + let gateSettled = false; + const settleGate = (signal: boolean): void => { + if (gateSettled) return; + gateSettled = true; + resolveCancellation(); + if (signal) drive?.signalAbort(); + }; + + try { + const result = await this.command(async (state, reader) => { + const operation = state.operation; + if (operation?.meta.operationId !== operationId) { + return { + kind: "return", + result: Result.err( + this.mismatch(operationId, operation?.meta.operationId ?? null, state.lastOperationId), + ), + }; + } + if (operation.state.control.status === "cancel_requested") { + return { + kind: "return", + result: Result.ok({ operationId, newlyRequested: false, steer: [], followUp: [] }), + }; + } + + const removed = state.inbox.filter((item) => item.kind === "steer" || item.kind === "followUp"); + const payloads = await Promise.all( + removed.map(async (item) => { + const stored = await reader.getValue(pendingEntry(item.entryId), context); + if (stored?.value.type !== "message") { + throw new SessionInvariantError( + `Pending ${item.kind} entry ${item.entryId} is missing its message`, + ); + } + return { item, message: stored.value.payload }; + }), + ); + const steer = payloads.filter(({ item }) => item.kind === "steer").map(({ message }) => message); + const followUp = payloads.filter(({ item }) => item.kind === "followUp").map(({ message }) => message); + const removedIds = new Set(removed.map((item) => item.entryId)); + const inbox = state.inbox.filter((item) => !removedIds.has(item.entryId)); + const operationState: OperationState = { + ...operation.state, + control: { status: "cancel_requested", requestedAt: Date.now() }, + }; + return { + kind: "commit", + writes: [ + ...removed.map((item) => deleteValue(pendingEntry(item.entryId))), + setValue(operationStateValue(operationId), operationState), + setValue(laneStateValue(this.name), { + currentOperationId: operationId, + lastOperationId: state.lastOperationId, + inbox, + }), + ], + next: { ...state, inbox, operation: { meta: operation.meta, state: operationState } }, + materialize: () => { + settleGate(true); + return Result.ok({ operationId, newlyRequested: true, steer, followUp }); + }, + events: () => [ + { + type: "operation_abort", + operationId, + steer, + followUp, + lane: this.name, + }, + ], + }; + }, context); + // A fresh Drive may observe an already-durable marker; pull its gate on the repeat path too. + settleGate(result.ok && result.value.newlyRequested === false); + return result; + } catch (error) { + if (!gateSettled) rejectCancellation(error); + throw error; + } + } + async requestAbort(..._args: Parameters): Promise { throw new SliceNotImplemented("requestAbort"); } @@ -895,23 +994,6 @@ export class Lane implements AgentLane { break; } - const drained = - operation.state.control.status === "cancel_requested" - ? { - steer: await readPendingMessages( - reader, - operation.state.control.drainedSteer, - "Drained steer entry", - context, - ), - followUp: await readPendingMessages( - reader, - operation.state.control.drainedFollowUp, - "Drained follow-up entry", - context, - ), - } - : undefined; operationSnapshot = { id: operation.meta.operationId, kind: operation.meta.intent.kind, @@ -919,7 +1001,6 @@ export class Lane implements AgentLane { status: operation.state.control.status === "cancel_requested" ? "aborting" : "open", ...(retry === undefined ? {} : { retry }), ...(deferred === undefined ? {} : { deferred }), - ...(drained === undefined ? {} : { drained }), ...(streamingMessage === undefined ? {} : { streamingMessage }), runningTools, }; @@ -1055,6 +1136,7 @@ export class Lane implements AgentLane { seal(error: Error): void { this.closedError ??= error; + this.activeDrive?.closeGate(error); } assertOpen(): void { diff --git a/packages/agent/src/harness/runtime/restore.ts b/packages/agent/src/harness/runtime/restore.ts index ec0bd1d3d..8b9e4df8f 100644 --- a/packages/agent/src/harness/runtime/restore.ts +++ b/packages/agent/src/harness/runtime/restore.ts @@ -26,9 +26,16 @@ function isSummaryState(state: OperationState): state is Extract void; private readonly rejectCompletion: (error: unknown) => void; @@ -111,6 +113,8 @@ export class Drive { const { gate, control } = createGate(); this.gate = gate; this.control = control; + this.closeController = new AbortController(); + this.closeSignal = this.closeController.signal; } settle(outcome: DriveOutcome): void { @@ -131,6 +135,7 @@ export class Drive { closeGate(error: Error): void { this.control.close(error); + if (!this.closeSignal.aborted) this.closeController.abort(error); } } diff --git a/packages/agent/src/harness/session/types.ts b/packages/agent/src/harness/session/types.ts index 0c8eb19e8..f4ace93a7 100644 --- a/packages/agent/src/harness/session/types.ts +++ b/packages/agent/src/harness/session/types.ts @@ -93,8 +93,6 @@ export type Control = | { status: "cancel_requested"; requestedAt: number; - drainedSteer: string[]; - drainedFollowUp: string[]; }; export interface OperationError { diff --git a/packages/agent/src/harness/telemetry.ts b/packages/agent/src/harness/telemetry.ts index 9eb6ee937..93785235f 100644 --- a/packages/agent/src/harness/telemetry.ts +++ b/packages/agent/src/harness/telemetry.ts @@ -164,7 +164,7 @@ const EVENT_TYPES = [ "run_start", "run_resume", "run_suspend", - "run_abort", + "operation_abort", "run_end", "fault", "handler_error", diff --git a/packages/agent/test/harness/runtime/drive-generation.test.ts b/packages/agent/test/harness/runtime/drive-generation.test.ts index d758d85f1..2e46e6e52 100644 --- a/packages/agent/test/harness/runtime/drive-generation.test.ts +++ b/packages/agent/test/harness/runtime/drive-generation.test.ts @@ -607,7 +607,7 @@ describe("runtime assistant generation", () => { if (operation === null) throw new Error("missing operation"); const nextState: OperationState = { ...operation.state, - control: { status: "cancel_requested", requestedAt: 10, drainedSteer: [], drainedFollowUp: [] }, + control: { status: "cancel_requested", requestedAt: 10 }, }; return { kind: "commit", diff --git a/packages/agent/test/harness/runtime/drive-reconcile.test.ts b/packages/agent/test/harness/runtime/drive-reconcile.test.ts new file mode 100644 index 000000000..db9de83a0 --- /dev/null +++ b/packages/agent/test/harness/runtime/drive-reconcile.test.ts @@ -0,0 +1,747 @@ +import { + createModels, + type DeferredHandle, + fauxAssistantMessage, + fauxProvider, + fauxToolCall, + type MutableModels, + type Provider, +} from "@earendil-works/pi-ai"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { HarnessEvent, WatchHandle } from "../../../src/harness/agent-harness.ts"; +import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; +import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; +import { HookRegistry } from "../../../src/harness/hooks.ts"; +import { reconcileOperation } from "../../../src/harness/runtime/drive/reconcile.ts"; +import { runStructuralDecision } from "../../../src/harness/runtime/drive/structural.ts"; +import { driveOperation } from "../../../src/harness/runtime/drive.ts"; +import { Lane } from "../../../src/harness/runtime/lane.ts"; +import { restoreLane } from "../../../src/harness/runtime/restore.ts"; +import { type Config, Drive } from "../../../src/harness/runtime/types.ts"; +import { insertEntry } from "../../../src/harness/session/commit.ts"; +import { MemoryStorage } from "../../../src/harness/session/memory.ts"; +import { StorageBackedSession } from "../../../src/harness/session/session.ts"; +import { InstrumentedStorage } from "../../../src/harness/session/testing/instrumented-storage.ts"; +import type { + Control, + DeferredEffectPendingOperation, + DeferredSuspendedOperation, + LaneConfiguration, + NewEntry, + OperationMeta, + OperationScope, + OperationState, + Session, + SummaryContext, + SummaryTask, + Write, +} from "../../../src/harness/session/types.ts"; +import * as storedValues from "../../../src/harness/session/values.ts"; +import type { AgentMessage } from "../../../src/types.ts"; + +const sessions: Session[] = []; +const operationId = "01950000-0000-7000-8000-000000000001"; + +interface Fixture { + lane: Lane; + drive: Drive; + session: Session; + storage: InstrumentedStorage; + models: MutableModels; + faux: ReturnType; + configuration: LaneConfiguration; + events: HarnessEvent[]; + hooks: HookRegistry; +} + +interface InstalledOperation { + state: OperationState; + intent: OperationMeta["intent"]; + entries?: NewEntry[]; + writes?: Write[]; + terminalEvents: HarnessEvent["type"][]; +} + +function unusedWatch(): WatchHandle { + throw new Error("watch is not used by reconciliation tests"); +} + +function cancelledControl(): Control { + return { status: "cancel_requested", requestedAt: 10 }; +} + +function scope(control: Control = cancelledControl()): OperationScope { + return { + control, + settings: { + compaction: DEFAULT_COMPACTION_SETTINGS, + steeringMode: "all", + followUpMode: "all", + toolExecution: "parallel", + }, + latestAssistantEntryId: null, + }; +} + +function user(content: string): AgentMessage { + return { role: "user", content, timestamp: 1 }; +} + +function summaryContext(configuration: LaneConfiguration): SummaryContext { + return { + resultEntryId: "summary-entry", + configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 10 }, + }; +} + +async function createFixture(): Promise { + const storage = new InstrumentedStorage(new MemoryStorage({ now: () => 100 })); + const session = new StorageBackedSession( + { id: `reconcile-${sessions.length}`, createdAt: 1, storageVersion: 1 }, + storage, + ); + sessions.push(session); + const faux = fauxProvider({ api: "faux", deferred: { pendingFetches: 0 } }); + const model = faux.getModel(); + const models = createModels(); + models.setProvider(faux.provider); + const configuration: LaneConfiguration = { + model: { provider: model.provider, modelId: model.id }, + thinkingLevel: "off", + activeToolNames: [], + }; + await session.mutate( + (mutator) => + mutator.commit( + [ + storedValues.setValue(storedValues.branchTip("main"), null), + storedValues.setValue(storedValues.laneConfig("main"), configuration), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: null, + lastOperationId: null, + inbox: [], + }), + ], + BACKGROUND_CONTEXT, + ), + BACKGROUND_CONTEXT, + ); + const hooks = new HookRegistry(() => {}); + const events: HarnessEvent[] = []; + const config: Config = { + tools: [], + resources: {}, + streamOptions: {}, + retryPolicy: { enabled: true, maxRetries: 1, baseDelayMs: 10 }, + compaction: DEFAULT_COMPACTION_SETTINGS, + steeringMode: "all", + followUpMode: "all", + toolExecution: "parallel", + toolContext: undefined, + systemPrompt: undefined, + toProviderMessages: (messages) => + messages.filter( + (message) => message.role === "user" || message.role === "assistant" || message.role === "toolResult", + ), + entryProjectors: {}, + }; + const lane = new Lane( + "main", + session, + models, + hooks, + await restoreLane(session, "main", BACKGROUND_CONTEXT), + (cause) => (cause instanceof Error ? cause : new Error(String(cause))), + (batch) => { + events.push(...structuredClone(batch)); + return Promise.resolve(); + }, + unusedWatch, + () => config, + ); + const drive = new Drive({ operationId }, BACKGROUND_CONTEXT); + lane.activeDrive = drive; + storage.clearCommitAttempts(); + return { lane, drive, session, storage, models, faux, configuration, events, hooks }; +} + +async function installOperation(fixture: Fixture, installed: InstalledOperation): Promise { + const entries = installed.entries ?? [ + { id: "tip", parentId: null, type: "message" as const, message: user("history") }, + ]; + const tipId = entries.at(-1)?.id ?? null; + const meta: OperationMeta = { + operationId, + lane: "main", + sourceTipId: tipId, + startedAt: 1, + intent: installed.intent, + }; + await fixture.lane.command( + (projection) => ({ + kind: "commit", + writes: [ + ...entries.map((entry) => insertEntry(entry)), + storedValues.setValue(storedValues.branchTip("main"), tipId), + ...(installed.writes ?? []), + storedValues.setValue(storedValues.operationMeta(operationId), meta), + storedValues.setValue(storedValues.operationState(operationId), installed.state), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + lastOperationId: projection.lastOperationId, + inbox: projection.inbox, + }), + ], + next: { ...projection, tipId, operation: { meta, state: installed.state } }, + materialize: () => undefined, + }), + BACKGROUND_CONTEXT, + ); + fixture.storage.clearCommitAttempts(); +} + +function deferredOperation( + fixture: Fixture, + effectPending: boolean, +): { state: DeferredSuspendedOperation | DeferredEffectPendingOperation; entry: NewEntry; handle: DeferredHandle } { + const model = fixture.faux.getModel(); + const handle = { provider: model.provider, modelId: model.id, api: model.api, id: "deferred-job" }; + const entry: NewEntry = { + id: "deferred-source", + parentId: null, + type: "message", + message: fauxAssistantMessage([], { stopReason: "deferred", deferred: handle }), + }; + const base = { + ...scope(), + stepId: "step", + sourceEntryId: entry.id, + poll: effectPending ? 1 : 0, + configuration: fixture.configuration, + streamOptions: {}, + }; + return { + state: effectPending + ? { + ...base, + at: "deferred.effect_pending", + responseEntryId: "deferred-response", + usageId: "deferred-usage", + } + : { ...base, at: "deferred.suspended" }, + entry, + handle, + }; +} + +function cases(fixture: Fixture): InstalledOperation[] { + const generation = { + stepId: "step", + triggerEntryId: "tip", + configuration: fixture.configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 10 }, + overflowRecoveryUsed: false, + }; + const runTask: SummaryTask = { + taskId: "run-summary", + reason: "threshold", + boundary: { + kind: "resume_checkpoint", + resumeAfter: { + continuation: { kind: "need_assistant", overflowRecoveryUsed: false }, + triggerEntryId: "tip", + }, + }, + }; + const compactionTask: SummaryTask = { taskId: "compaction", reason: "manual", boundary: { kind: "finish" } }; + const navigationTask: SummaryTask = { + taskId: "navigation", + boundary: { kind: "commit_navigation", targetId: "target" }, + }; + const suspended = deferredOperation(fixture, false); + const deferredEffect = deferredOperation(fixture, true); + return [ + { + state: { ...scope(), at: "starting" }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }, + { + state: { + ...scope(), + at: "checkpoint", + continuation: { kind: "need_assistant", overflowRecoveryUsed: false }, + triggerEntryId: "tip", + }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }, + { + state: { ...scope(), at: "assistant.ready", generationContext: generation, nextAttempt: 1 }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }, + { + state: { + ...scope(), + at: "assistant.effect_pending", + generationContext: generation, + attempt: 1, + responseEntryId: "assistant-response", + usageId: "assistant-usage", + intendedOutputLimit: 100, + contextWindow: 1_000, + }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + writes: [ + storedValues.appendList(storedValues.pendingAssistantFrames(operationId, "assistant-response"), { + type: "text_delta", + contentIndex: 0, + delta: "partial", + }), + ], + terminalEvents: ["run_end"], + }, + { + state: { + ...scope(), + at: "assistant.retry_wait", + generationContext: generation, + nextAttempt: 2, + notBefore: Date.now() + 100_000, + errorMessage: "retry", + }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }, + { + state: { + ...scope(), + at: "tools", + batch: { + assistantEntryId: "assistant", + configuration: fixture.configuration, + turnId: "turn", + calls: [{ status: "planned", sourceIndex: 0, resultEntryId: "tool-result" }], + }, + }, + intent: { kind: "run", promptEntryIds: [] }, + entries: [ + { + id: "assistant", + parentId: null, + type: "message", + message: fauxAssistantMessage(fauxToolCall("tool", {}), { stopReason: "toolUse" }), + }, + ], + terminalEvents: ["run_end"], + }, + { + state: suspended.state, + intent: { kind: "run", promptEntryIds: [] }, + entries: [suspended.entry], + terminalEvents: ["run_end"], + }, + { + state: deferredEffect.state, + intent: { kind: "run", promptEntryIds: [] }, + entries: [deferredEffect.entry], + terminalEvents: ["run_end"], + }, + { + state: { ...scope(), at: "summary.deciding", task: runTask }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["compaction_end", "run_end"], + }, + { + state: { + ...scope(), + at: "summary.ready", + task: compactionTask, + summaryContext: summaryContext(fixture.configuration), + nextAttempt: 1, + }, + intent: { kind: "compaction" }, + terminalEvents: ["compaction_end"], + }, + { + state: { + ...scope(), + at: "summary.effect_pending", + task: navigationTask, + summaryContext: summaryContext(fixture.configuration), + attempt: 1, + request: { index: 0, usageId: "usage" }, + usageIds: [], + }, + intent: { kind: "navigation", targetId: "target", summarize: true }, + terminalEvents: ["navigation_end"], + }, + { + state: { + ...scope(), + at: "summary.retry_wait", + task: runTask, + summaryContext: summaryContext(fixture.configuration), + nextAttempt: 2, + notBefore: Date.now() + 100_000, + errorMessage: "retry", + }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["compaction_end", "run_end"], + }, + { + state: { ...scope(), at: "navigation.ready_to_commit", targetId: "target" }, + intent: { kind: "navigation", targetId: "target", summarize: false }, + terminalEvents: ["navigation_end"], + }, + ]; +} + +afterEach(async () => { + vi.useRealTimers(); + vi.restoreAllMocks(); + for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); +}); + +describe("runtime total drive", () => { + it("drives an ordinary run through all direct procedures with one before_drive hook", async () => { + const fixture = await createFixture(); + await installOperation(fixture, { + state: { ...scope({ status: "running" }), at: "starting" }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }); + const beforeDrive = vi.fn(); + fixture.hooks.on("before_drive", beforeDrive); + fixture.faux.setResponses([fauxAssistantMessage("answer")]); + + expect(await driveOperation(fixture.lane, fixture.drive)).toMatchObject({ + kind: "settled", + outcome: { operationId, kind: "run", status: "completed" }, + }); + expect(beforeDrive).toHaveBeenCalledTimes(1); + expect(fixture.faux.state.callCount).toBe(1); + expect(fixture.lane.state.operation).toBeNull(); + }); + + it("leaves durable state unchanged when before_drive fails closed", async () => { + const fixture = await createFixture(); + const starting = { ...scope({ status: "running" }), at: "starting" } as const; + await installOperation(fixture, { + state: starting, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }); + fixture.hooks.on("before_drive", () => { + throw new Error("blocked drive"); + }); + + await expect(driveOperation(fixture.lane, fixture.drive)).rejects.toThrow("blocked drive"); + expect(fixture.lane.state.operation?.state).toEqual(starting); + expect(fixture.storage.getCommitAttempts()).toEqual([]); + }); +}); + +describe("runtime cancellation reconciliation", () => { + it("reconciles every durable leaf without ordinary hook admission", async () => { + for (let index = 0; index < 13; index++) { + const fixture = await createFixture(); + const installed = cases(fixture)[index]!; + await installOperation(fixture, installed); + const beforeDrive = vi.fn(); + fixture.hooks.on("before_drive", beforeDrive); + + const outcome = await driveOperation(fixture.lane, fixture.drive); + + expect(outcome).toMatchObject({ + kind: "settled", + outcome: { operationId, status: "aborted" }, + }); + expect(beforeDrive).not.toHaveBeenCalled(); + expect(fixture.lane.state.operation).toBeNull(); + expect(await fixture.lane.getResult(operationId, BACKGROUND_CONTEXT)).toMatchObject({ status: "aborted" }); + expect( + await fixture.session.getValue(storedValues.operationMeta(operationId), BACKGROUND_CONTEXT), + ).toBeUndefined(); + expect( + await fixture.session.getValue(storedValues.operationState(operationId), BACKGROUND_CONTEXT), + ).toBeUndefined(); + expect( + await fixture.session.scanValues(storedValues.operationToolArgsPrefix(operationId), BACKGROUND_CONTEXT), + ).toEqual([]); + expect( + await fixture.session.scanValues(storedValues.operationToolMemoPrefix(operationId), BACKGROUND_CONTEXT), + ).toEqual([]); + expect( + await fixture.session.scanValues(storedValues.operationPreparationPrefix(operationId), BACKGROUND_CONTEXT), + ).toEqual([]); + if (installed.state.at === "deferred.suspended" || installed.state.at === "deferred.effect_pending") { + expect(fixture.faux.state.cancelledDeferred).toHaveLength(1); + } + expect( + fixture.events + .filter((event) => installed.terminalEvents.includes(event.type)) + .slice(-installed.terminalEvents.length) + .map((event) => event.type), + ).toEqual(installed.terminalEvents); + } + }); + + it("durably drains abortable input once and preserves lane-owned input through terminal cleanup", async () => { + const fixture = await createFixture(); + await installOperation(fixture, { + state: { + ...scope({ status: "running" }), + at: "checkpoint", + continuation: { kind: "need_assistant", overflowRecoveryUsed: false }, + triggerEntryId: "tip", + }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }); + const queued = [ + { entryId: "steer", kind: "steer" as const, message: user("steer") }, + { entryId: "write", kind: "write" as const, message: user("write") }, + { entryId: "follow", kind: "followUp" as const, message: user("follow") }, + { entryId: "next", kind: "nextRun" as const, message: user("next") }, + ]; + await fixture.lane.command( + (state) => ({ + kind: "commit", + writes: [ + ...queued.map((item) => + storedValues.setValue(storedValues.pendingEntry(item.entryId), { + type: "message" as const, + payload: item.message, + }), + ), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + lastOperationId: null, + inbox: queued.map(({ entryId, kind }) => ({ entryId, kind })), + }), + ], + next: { ...state, inbox: queued.map(({ entryId, kind }) => ({ entryId, kind })) }, + materialize: () => undefined, + }), + BACKGROUND_CONTEXT, + ); + fixture.storage.clearCommitAttempts(); + + const requested = await fixture.lane.requestOperationAbort(operationId, BACKGROUND_CONTEXT); + expect(requested).toMatchObject({ + ok: true, + value: { + operationId, + newlyRequested: true, + steer: [{ content: "steer" }], + followUp: [{ content: "follow" }], + }, + }); + expect(fixture.drive.gate.signal.aborted).toBe(true); + expect(fixture.drive.closeSignal.aborted).toBe(false); + expect(fixture.lane.state.inbox).toEqual([ + { entryId: "write", kind: "write" }, + { entryId: "next", kind: "nextRun" }, + ]); + expect(await fixture.session.getValue(storedValues.pendingEntry("steer"), BACKGROUND_CONTEXT)).toBeUndefined(); + expect(await fixture.session.getValue(storedValues.pendingEntry("follow"), BACKGROUND_CONTEXT)).toBeUndefined(); + expect(fixture.events.filter((event) => event.type === "operation_abort")).toMatchObject([ + { operationId, steer: [{ content: "steer" }], followUp: [{ content: "follow" }] }, + ]); + const commitCount = fixture.storage.getCommitAttempts().length; + const eventCount = fixture.events.length; + expect(await fixture.lane.requestOperationAbort(operationId, BACKGROUND_CONTEXT)).toEqual({ + ok: true, + value: { operationId, newlyRequested: false, steer: [], followUp: [] }, + }); + expect(fixture.storage.getCommitAttempts()).toHaveLength(commitCount); + expect(fixture.events).toHaveLength(eventCount); + + expect(await driveOperation(fixture.lane, fixture.drive)).toMatchObject({ + kind: "settled", + outcome: { status: "aborted" }, + }); + expect(fixture.lane.state.inbox).toEqual([ + { entryId: "write", kind: "write" }, + { entryId: "next", kind: "nextRun" }, + ]); + expect(await fixture.session.getValue(storedValues.pendingEntry("write"), BACKGROUND_CONTEXT)).toBeDefined(); + expect(await fixture.session.getValue(storedValues.pendingEntry("next"), BACKGROUND_CONTEXT)).toBeDefined(); + fixture.storage.clearCommitAttempts(); + const stale = await fixture.lane.requestOperationAbort(operationId, BACKGROUND_CONTEXT); + expect(stale.ok).toBe(false); + if (stale.ok) throw new Error("settled abort unexpectedly succeeded"); + expect(stale.error._tag).toBe("OperationMismatch"); + expect(fixture.storage.getCommitAttempts()).toEqual([]); + }); + + it("marks cancellation without installing a Drive", async () => { + const fixture = await createFixture(); + await installOperation(fixture, { + state: { ...scope({ status: "running" }), at: "starting" }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }); + fixture.lane.activeDrive = undefined; + + expect(await fixture.lane.requestOperationAbort(operationId, BACKGROUND_CONTEXT)).toMatchObject({ + ok: true, + value: { newlyRequested: true }, + }); + expect(fixture.lane.activeDrive).toBeUndefined(); + expect(fixture.lane.state.operation?.state.control.status).toBe("cancel_requested"); + }); + + it("cancels an admitted retry timer only after the abort marker commits", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const fixture = await createFixture(); + const generation = { + stepId: "step", + triggerEntryId: "tip", + configuration: fixture.configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 2, baseDelayMs: 10 }, + overflowRecoveryUsed: false, + }; + await installOperation(fixture, { + state: { + ...scope({ status: "running" }), + at: "assistant.retry_wait", + generationContext: generation, + nextAttempt: 2, + notBefore: 2_000, + errorMessage: "retry", + }, + intent: { kind: "run", promptEntryIds: ["tip"] }, + terminalEvents: ["run_end"], + }); + const drive = new Drive({ operationId, waitForRetry: true }, BACKGROUND_CONTEXT); + fixture.lane.activeDrive = drive; + const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout"); + const running = driveOperation(fixture.lane, drive); + await vi.waitFor(() => expect(setTimeoutSpy).toHaveBeenCalled()); + + expect(await fixture.lane.requestOperationAbort(operationId, BACKGROUND_CONTEXT)).toMatchObject({ + ok: true, + value: { newlyRequested: true }, + }); + expect(await running).toMatchObject({ kind: "settled", outcome: { status: "aborted" } }); + expect(fixture.events.some((event) => event.type === "retry_start")).toBe(false); + }); + + it("drops a stale structural hook result when cancellation commits first", async () => { + const fixture = await createFixture(); + const task: SummaryTask = { taskId: "task", reason: "manual", boundary: { kind: "finish" } }; + const deciding = { ...scope({ status: "running" }), at: "summary.deciding", task } as const; + await installOperation(fixture, { + state: deciding, + intent: { kind: "compaction" }, + writes: [ + storedValues.setValue(storedValues.operationPreparation(operationId, task.taskId), { + kind: "compaction", + messagesToSummarize: [user("history")], + turnPrefixMessages: [], + retainedTail: [], + isSplitTurn: false, + tokensBefore: 100, + fileOps: { read: [], written: [], edited: [] }, + settings: DEFAULT_COMPACTION_SETTINGS, + }), + ], + terminalEvents: ["compaction_end"], + }); + let releaseHook!: () => void; + const hookStarted = new Promise((resolve) => { + fixture.hooks.on("before_compaction", async () => { + resolve(); + await new Promise((release) => { + releaseHook = release; + }); + return { + compaction: { summary: "stale", tokensBefore: 100, retainedTail: [] }, + }; + }); + }); + const running = runStructuralDecision(fixture.lane, fixture.drive, deciding); + await hookStarted; + await fixture.lane.requestOperationAbort(operationId, BACKGROUND_CONTEXT); + releaseHook(); + + expect(await running).toEqual({ kind: "continue" }); + expect(await fixture.session.findEntries({ type: "compaction" }, BACKGROUND_CONTEXT)).toEqual([]); + expect(await driveOperation(fixture.lane, fixture.drive)).toMatchObject({ + kind: "settled", + outcome: { status: "aborted" }, + }); + expect(fixture.events.at(-1)).toMatchObject({ type: "compaction_end", status: "aborted" }); + }); + + it("keeps the deferred cleanup signal separate from operation abort", () => { + const drive = new Drive({ operationId }, BACKGROUND_CONTEXT); + drive.beginAbort(Promise.resolve()); + drive.signalAbort(); + expect(drive.gate.signal.aborted).toBe(true); + expect(drive.closeSignal.aborted).toBe(false); + const closed = new Error("closed"); + drive.closeGate(closed); + expect(drive.closeSignal.aborted).toBe(true); + expect(drive.closeSignal.reason).toBe(closed); + }); + + it("can crash between cancelled response settlement and terminal cleanup", async () => { + const fixture = await createFixture(); + const installed = cases(fixture)[3]!; + await installOperation(fixture, installed); + + expect(await reconcileOperation(fixture.lane, fixture.drive)).toEqual({ kind: "continue" }); + expect(fixture.lane.state.operation?.state.at).toBe("checkpoint"); + expect(await fixture.session.getEntry("assistant-response", BACKGROUND_CONTEXT)).toMatchObject({ + type: "message", + message: { stopReason: "aborted" }, + }); + expect(await fixture.lane.getResult(operationId, BACKGROUND_CONTEXT)).toBeUndefined(); + expect( + await fixture.session.readList( + storedValues.pendingAssistantFrames(operationId, "assistant-response"), + undefined, + BACKGROUND_CONTEXT, + ), + ).toEqual([]); + + const resumed = new Drive({ operationId }, BACKGROUND_CONTEXT); + fixture.lane.activeDrive = resumed; + expect(await driveOperation(fixture.lane, resumed)).toMatchObject({ + kind: "settled", + outcome: { status: "aborted" }, + }); + }); + + it("ignores deferred-provider cancellation failure", async () => { + const fixture = await createFixture(); + const deferred = deferredOperation(fixture, false); + const provider: Provider = { + ...fixture.faux.provider, + cancelDeferred: async () => { + throw new Error("remote cancellation failed"); + }, + }; + fixture.models.setProvider(provider); + await installOperation(fixture, { + state: deferred.state, + intent: { kind: "run", promptEntryIds: [] }, + entries: [deferred.entry], + terminalEvents: ["run_end"], + }); + + expect(await driveOperation(fixture.lane, fixture.drive)).toMatchObject({ + kind: "settled", + outcome: { status: "aborted" }, + }); + }); +}); diff --git a/packages/agent/test/harness/runtime/drive-retry-deferred.test.ts b/packages/agent/test/harness/runtime/drive-retry-deferred.test.ts index 1bd02107d..3b1ce64d0 100644 --- a/packages/agent/test/harness/runtime/drive-retry-deferred.test.ts +++ b/packages/agent/test/harness/runtime/drive-retry-deferred.test.ts @@ -416,7 +416,7 @@ describe("runtime deferred polling", () => { if (operation === null) throw new Error("missing operation"); const nextState: OperationState = { ...operation.state, - control: { status: "cancel_requested", requestedAt: 10, drainedSteer: [], drainedFollowUp: [] }, + control: { status: "cancel_requested", requestedAt: 10 }, }; return { kind: "commit", diff --git a/packages/agent/test/harness/runtime/drive-structural.test.ts b/packages/agent/test/harness/runtime/drive-structural.test.ts index e040372bf..672456957 100644 --- a/packages/agent/test/harness/runtime/drive-structural.test.ts +++ b/packages/agent/test/harness/runtime/drive-structural.test.ts @@ -1,5 +1,5 @@ import { createModels, fauxAssistantMessage, fauxProvider, type MutableModels } from "@earendil-works/pi-ai"; -import { afterEach, describe, expect, it } from "vitest"; +import { afterEach, describe, expect, it, vi } from "vitest"; import type { HarnessEvent, WatchHandle } from "../../../src/harness/agent-harness.ts"; import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; @@ -269,12 +269,7 @@ async function cancelOperation(fixture: Fixture): Promise { if (operation === null) throw new Error("fixture has no operation"); const state: OperationState = { ...operation.state, - control: { - status: "cancel_requested", - requestedAt: 2, - drainedSteer: [], - drainedFollowUp: [], - }, + control: { status: "cancel_requested", requestedAt: 2 }, }; return { kind: "commit", @@ -286,6 +281,7 @@ async function cancelOperation(fixture: Fixture): Promise { } afterEach(async () => { + vi.useRealTimers(); for (const session of sessions.splice(0)) await session.close(BACKGROUND_CONTEXT); }); @@ -375,6 +371,29 @@ describe("runtime structural drive", () => { expect(fixture.events.some((event) => event.type === "compaction_start")).toBe(false); }); + it("rejects a missing threshold trigger when no newer compaction guards it", async () => { + const fixture = await createFixture(); + const model = fixture.faux.getModel(); + const checkpoint: CheckpointOperation = { + ...runScope({ enabled: true, reserveTokens: model.contextWindow, keepRecentTokens: 1 }), + at: "checkpoint", + continuation: { kind: "need_assistant", overflowRecoveryUsed: false }, + triggerEntryId: "missing-trigger", + }; + await installOperation( + fixture, + checkpoint, + { kind: "run", promptEntryIds: [] }, + { + entries: [{ id: "tip", parentId: null, type: "message", message: user("history") }], + }, + ); + + await expect(runCheckpoint(fixture.lane, fixture.drive, checkpoint)).rejects.toThrow( + "Checkpoint trigger missing-trigger is missing from its Branch", + ); + }); + it("finishes a may-finish run directly after threshold decline", async () => { const fixture = await createFixture(); const deciding = { @@ -1026,11 +1045,13 @@ describe("runtime structural drive", () => { expect(standalone.lane.state.operation).toBeNull(); }); - it("durably schedules retryable structural failures and exposes the retry wait", async () => { + it("durably brackets a delayed structural retry through success", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); const fixture = await createFixture(); const ready = summaryReady(runScope(), standaloneCompactionTask(), fixture.configuration, { maxAttempts: 2, - baseDelayMs: 10_000, + baseDelayMs: 10, }); await installOperation( fixture, @@ -1058,7 +1079,60 @@ describe("runtime structural drive", () => { notBefore: retry.notBefore, }, }); - expect(fixture.events.map((event) => event.type)).toContain("retry_scheduled"); + vi.setSystemTime(retry.notBefore); + expect(await runStructuralRetryWait(fixture.lane, fixture.drive, retry)).toEqual({ kind: "continue" }); + const second = currentState(fixture); + if (second.at !== "summary.ready") throw new Error("elapsed retry did not become ready"); + fixture.faux.setResponses([fauxAssistantMessage("summary")]); + expect(await runStructuralGeneration(fixture.lane, fixture.drive, second)).toMatchObject({ + kind: "settled", + outcome: { status: "completed" }, + }); + expect( + fixture.events + .filter((event) => event.type.startsWith("retry_")) + .map((event) => ({ type: event.type, ...("success" in event ? { success: event.success } : {}) })), + ).toEqual([{ type: "retry_scheduled" }, { type: "retry_start" }, { type: "retry_end", success: true }]); + }); + + it("closes an exhausted structural retry with its final error", async () => { + vi.useFakeTimers(); + vi.setSystemTime(2_000); + const fixture = await createFixture(); + const ready = summaryReady(runScope(), standaloneCompactionTask(), fixture.configuration, { + maxAttempts: 2, + baseDelayMs: 10, + }); + await installOperation( + fixture, + ready, + { kind: "compaction" }, + { + entries: [{ id: "tip", parentId: null, type: "message", message: user("history") }], + preparation: { taskId: "task", value: compactionPreparation() }, + }, + ); + fixture.faux.setResponses([ + fauxAssistantMessage("", { stopReason: "error", errorMessage: "rate limit exceeded" }), + ]); + await runStructuralGeneration(fixture.lane, fixture.drive, ready); + const retry = currentState(fixture); + if (retry.at !== "summary.retry_wait") throw new Error("retryable failure did not wait"); + vi.setSystemTime(retry.notBefore); + await runStructuralRetryWait(fixture.lane, fixture.drive, retry); + const second = currentState(fixture); + if (second.at !== "summary.ready") throw new Error("elapsed retry did not become ready"); + fixture.faux.setResponses([ + fauxAssistantMessage("", { stopReason: "error", errorMessage: "rate limit exceeded" }), + ]); + + expect(await runStructuralGeneration(fixture.lane, fixture.drive, second)).toMatchObject({ + kind: "settled", + outcome: { status: "failed" }, + }); + expect(fixture.events.filter((event) => event.type === "retry_end")).toMatchObject([ + { attempt: 2, success: false, finalError: expect.stringContaining("rate limit exceeded") }, + ]); }); it("finishes a standalone structural failure at the retry cap", async () => { @@ -1089,6 +1163,37 @@ describe("runtime structural drive", () => { expect(fixture.events.map((event) => event.type)).not.toContain("retry_scheduled"); }); + it("rejects invalid unsummarized navigation state without committing", async () => { + for (const invalid of ["missing", "source", "root_label"] as const) { + const fixture = await createFixture(); + const targetId = invalid === "missing" ? "missing" : invalid === "source" ? "source" : null; + const navigation = { + ...runScope(), + at: "navigation.ready_to_commit", + targetId, + ...(invalid === "root_label" ? { label: "invalid" } : {}), + } as const; + await installOperation( + fixture, + navigation, + { + kind: "navigation", + targetId, + summarize: false, + ...(invalid === "root_label" ? { label: "invalid" } : {}), + }, + { + entries: [{ id: "source", parentId: null, type: "message", message: user("source") }], + tipId: "source", + }, + ); + + await expect(commitNavigation(fixture.lane, fixture.drive, navigation)).rejects.toThrow(); + expect(fixture.storage.getCommitAttempts()).toEqual([]); + expect(fixture.lane.state.tipId).toBe("source"); + } + }); + it("moves an unsummarized navigation and cleans up in one terminal transaction", async () => { const fixture = await createFixture(); const navigation = { @@ -1275,6 +1380,48 @@ describe("runtime structural drive", () => { .flat() .some((write) => write.kind === "usage"), ).toBe(false); + expect(fixture.events.at(-1)).toMatchObject({ + type: "retry_scheduled", + attempt: 2, + recovery: true, + }); + }); + + it("terminal-fails an orphaned structural attempt at the retry cap", async () => { + const fixture = await createFixture(); + const effect = { + ...runScope(), + at: "summary.effect_pending", + task: standaloneCompactionTask(), + summaryContext: { + resultEntryId: "summary-entry", + configuration: fixture.configuration, + streamOptions: {}, + retryPolicy: { maxAttempts: 1, baseDelayMs: 10 }, + }, + attempt: 1, + request: { index: 0, usageId: "abandoned-usage" }, + usageIds: [], + } satisfies SummaryEffectPendingOperation; + await installOperation( + fixture, + effect, + { kind: "compaction" }, + { + entries: [{ id: "tip", parentId: null, type: "message", message: user("history") }], + preparation: { taskId: "task", value: compactionPreparation() }, + }, + ); + + expect(await recoverStructuralGeneration(fixture.lane, fixture.drive, effect)).toMatchObject({ + kind: "settled", + outcome: { + status: "failed", + error: { code: "structural_interrupted" }, + }, + }); + expect(fixture.lane.state.operation).toBeNull(); + expect(fixture.events.map((event) => event.type)).not.toContain("retry_scheduled"); }); it("rejects a preparation whose durable kind contradicts the structural state", async () => { diff --git a/packages/agent/test/harness/runtime/drive-terminal.test.ts b/packages/agent/test/harness/runtime/drive-terminal.test.ts index 50aa594d4..a2d6169ee 100644 --- a/packages/agent/test/harness/runtime/drive-terminal.test.ts +++ b/packages/agent/test/harness/runtime/drive-terminal.test.ts @@ -114,16 +114,11 @@ describe("runtime terminal cleanup mechanics", () => { usageId: "usage", intendedOutputLimit: 100, contextWindow: 1_000, - control: { - status: "cancel_requested", - requestedAt: 2, - drainedSteer: ["drained-steer"], - drainedFollowUp: ["drained-follow"], - }, + control: { status: "cancel_requested", requestedAt: 2 }, }; await seedLeftovers(session, operationId, state); await commit(session, [ - ...["steer", "follow", "write", "drained-steer", "drained-follow", "next"].map((id) => + ...["steer", "follow", "write", "next"].map((id) => storedValues.setValue(storedValues.pendingEntry(id), { type: "custom", customType: "test", @@ -157,8 +152,6 @@ describe("runtime terminal cleanup mechanics", () => { "value:delete:pi.op.preparation:run:task", "value:delete:pi.pending.tool_output:run:invocation", "list:delete:pi.pending.assistant_frame:run:response", - "value:delete:pi.pending.entry:drained-steer", - "value:delete:pi.pending.entry:drained-follow", ]); await commit(session, writes); for (const id of ["steer", "follow", "write", "next"]) { diff --git a/packages/agent/test/harness/runtime/drive-tools.test.ts b/packages/agent/test/harness/runtime/drive-tools.test.ts index 156888756..5b2f3dec7 100644 --- a/packages/agent/test/harness/runtime/drive-tools.test.ts +++ b/packages/agent/test/harness/runtime/drive-tools.test.ts @@ -142,9 +142,7 @@ async function createFixture(options: FixtureOptions): Promise { })); const run: ToolsOperation = { at: "tools", - control: options.cancelled - ? { status: "cancel_requested", requestedAt: 30, drainedSteer: [], drainedFollowUp: [] } - : { status: "running" }, + control: options.cancelled ? { status: "cancel_requested", requestedAt: 30 } : { status: "running" }, settings: { compaction: DEFAULT_COMPACTION_SETTINGS, steeringMode: "all", @@ -600,7 +598,7 @@ describe("durable tool batch", () => { if (operation === null) throw new Error("missing operation"); const nextRun: OperationState = { ...operation.state, - control: { status: "cancel_requested", requestedAt: 40, drainedSteer: [], drainedFollowUp: [] }, + control: { status: "cancel_requested", requestedAt: 40 }, }; return { kind: "commit", @@ -654,7 +652,7 @@ describe("durable tool batch", () => { if (operation === null) throw new Error("missing operation"); const nextRun: OperationState = { ...operation.state, - control: { status: "cancel_requested", requestedAt: 40, drainedSteer: [], drainedFollowUp: [] }, + control: { status: "cancel_requested", requestedAt: 40 }, }; return { kind: "commit", diff --git a/packages/agent/test/harness/runtime/lane.test.ts b/packages/agent/test/harness/runtime/lane.test.ts index b9c0c8b67..c43d5d417 100644 --- a/packages/agent/test/harness/runtime/lane.test.ts +++ b/packages/agent/test/harness/runtime/lane.test.ts @@ -315,12 +315,7 @@ describe("runtime Lane commands", () => { if (operation?.state.at !== "starting") throw new Error("missing accepted operation"); const cancelled: StartingOperation = { ...operation.state, - control: { - status: "cancel_requested", - requestedAt: 1, - drainedSteer: [], - drainedFollowUp: [], - }, + control: { status: "cancel_requested", requestedAt: 1 }, }; await lane.command( (state) => ({ diff --git a/packages/agent/test/harness/runtime/restore.test.ts b/packages/agent/test/harness/runtime/restore.test.ts index a6e21a0bf..5204c6a2a 100644 --- a/packages/agent/test/harness/runtime/restore.test.ts +++ b/packages/agent/test/harness/runtime/restore.test.ts @@ -215,11 +215,31 @@ describe("runtime lane restore", () => { state: { ...operationScope(), at: "navigation.ready_to_commit", targetId: null }, accepted: true, }, + { + intent: { kind: "navigation", targetId: "target", summarize: false }, + state: { ...operationScope(), at: "navigation.ready_to_commit", targetId: "different" }, + accepted: false, + }, + { + intent: { kind: "navigation", targetId: "target", summarize: false }, + state: summaryState(navigation), + accepted: false, + }, { intent: { kind: "navigation", targetId: "target", summarize: true }, state: summaryState(navigation), accepted: true, }, + { + intent: { kind: "navigation", targetId: "different", summarize: true }, + state: summaryState(navigation), + accepted: false, + }, + { + intent: { kind: "navigation", targetId: "target", summarize: true }, + state: { ...operationScope(), at: "navigation.ready_to_commit", targetId: "target" }, + accepted: false, + }, { intent: { kind: "navigation", targetId: "target", summarize: true }, state: summaryState(finish), diff --git a/packages/agent/test/harness/runtime/watch.test.ts b/packages/agent/test/harness/runtime/watch.test.ts index acfef0f19..0debfbf66 100644 --- a/packages/agent/test/harness/runtime/watch.test.ts +++ b/packages/agent/test/harness/runtime/watch.test.ts @@ -142,7 +142,7 @@ describe("runtime lane watch", () => { second.unsubscribe(); }); - it("dereferences queues, pending writes, deferred handles, and abort drains", async () => { + it("dereferences queues, pending writes, and deferred handles", async () => { const session = await createSession(); const handle = { provider: "provider", modelId: "model", api: "test", id: "deferred" }; const sourceMessage = fauxAssistantMessage([], { stopReason: "deferred", deferred: handle }); @@ -151,8 +151,6 @@ describe("runtime lane watch", () => { steer: "steer", follow: "follow", write: "write", - drainedSteer: "drained-steer", - drainedFollow: "drained-follow", }; const payload = (content: string) => ({ type: "message" as const, @@ -160,12 +158,7 @@ describe("runtime lane watch", () => { }); const operationId = session.idGenerator.next(); const state: OperationState = { - ...runScope({ - status: "cancel_requested", - requestedAt: 2, - drainedSteer: [ids.drainedSteer], - drainedFollowUp: [ids.drainedFollow], - }), + ...runScope({ status: "cancel_requested", requestedAt: 2 }), at: "deferred.suspended", stepId: "step", sourceEntryId: "source", @@ -216,10 +209,6 @@ describe("runtime lane watch", () => { id: operationId, status: "aborting", deferred: { handle, poll: 3 }, - drained: { - steer: [{ entryId: ids.drainedSteer }], - followUp: [{ entryId: ids.drainedFollow }], - }, runningTools: [], }); watch.unsubscribe(); diff --git a/packages/agent/test/harness/types.test.ts b/packages/agent/test/harness/types.test.ts index f6fadb52f..cc7617d06 100644 --- a/packages/agent/test/harness/types.test.ts +++ b/packages/agent/test/harness/types.test.ts @@ -397,7 +397,7 @@ it("covers Part 5 results, events, hooks, snapshots, tools, and stream options", | "run_start" | "run_resume" | "run_suspend" - | "run_abort" + | "operation_abort" | "run_end" | "fault" | "handler_error" diff --git a/packages/coding-agent/src/experimental/harness-wire-adapter.ts b/packages/coding-agent/src/experimental/harness-wire-adapter.ts index 94ee13c0d..ba203dea7 100644 --- a/packages/coding-agent/src/experimental/harness-wire-adapter.ts +++ b/packages/coding-agent/src/experimental/harness-wire-adapter.ts @@ -620,20 +620,6 @@ export function toWireLaneSnapshot(snapshot: HarnessLaneSnapshot): LaneSnapshot poll: snapshot.operation.deferred.poll, }, }), - ...(snapshot.operation.drained === undefined - ? {} - : { - drained: { - steer: snapshot.operation.drained.steer.map((item) => ({ - entryId: item.entryId, - message: toWireMessage(item.message), - })), - followUp: snapshot.operation.drained.followUp.map((item) => ({ - entryId: item.entryId, - message: toWireMessage(item.message), - })), - }, - }), ...(snapshot.operation.streamingMessage === undefined ? {} : { streamingMessage: toWireAssistantMessage(snapshot.operation.streamingMessage) }), @@ -685,10 +671,10 @@ export function toWireLaneEvent(event: HarnessEvent): LaneEvent | undefined { ...base, lane: event.lane, }; - case "run_abort": + case "operation_abort": return { - type: "run_abort", - runId: event.runId, + type: "operation_abort", + operationId: event.operationId, steer: event.steer.map(toWireMessage), followUp: event.followUp.map(toWireMessage), ...base, diff --git a/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts b/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts index 001fbfcc4..feaaecea4 100644 --- a/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts +++ b/packages/coding-agent/test/experimental-harness-wire-adapter.test.ts @@ -214,6 +214,19 @@ describe("Harness wire adapter", () => { expect(wire).toMatchObject({ lane: "main", transcript: [{ id: "entry-1" }] }); }); + test("projects family-neutral operation abort events", () => { + const event: HarnessEvent = { + type: "operation_abort", + lane: "main", + operationId: "operation-1", + steer: [{ role: "user", content: "steer", timestamp: 1 }], + followUp: [{ role: "user", content: "follow", timestamp: 2 }], + }; + const wire = toWireLaneEvent(event); + expect(Check(LaneEventSchema, wire)).toBe(true); + expect(wire).toMatchObject({ type: "operation_abort", operationId: "operation-1" }); + }); + test("projects assistant updates as compact frames", () => { const partial: AssistantMessage = { ...finalMessage, diff --git a/packages/protocol/src/harness.ts b/packages/protocol/src/harness.ts index c629aec51..312ab51fb 100644 --- a/packages/protocol/src/harness.ts +++ b/packages/protocol/src/harness.ts @@ -343,9 +343,6 @@ const LaneOperationSchema = StrictObject({ status: Type.Union([Type.Literal("running"), Type.Literal("open"), Type.Literal("aborting")]), startedAt: TimestampSchema, deferred: Type.Optional(StrictObject({ handle: DeferredHandleSchema, poll: Type.Integer({ minimum: 0 }) })), - drained: Type.Optional( - StrictObject({ steer: Type.Array(QueuedItemSchema), followUp: Type.Array(QueuedItemSchema) }), - ), streamingMessage: Type.Optional(AssistantMessageSchema), runningTools: Type.Array(RunningToolSchema), retry: Type.Optional( @@ -407,8 +404,8 @@ export const LaneEventSchema = Type.Union([ ...LaneEventBase, }), StrictObject({ - type: Type.Literal("run_abort"), - runId: IdSchema, + type: Type.Literal("operation_abort"), + operationId: IdSchema, steer: Type.Array(PromptMessageSchema), followUp: Type.Array(PromptMessageSchema), ...LaneEventBase,