From 58c68b8939c93ddcfa5c93934cc77d99df2255b4 Mon Sep 17 00:00:00 2001 From: Mike Olson Date: Fri, 25 Sep 2026 02:21:21 -0400 Subject: [PATCH] fix(pi): Fail turns that end on an unresolved abort --- .../Adapters/PiAdapterV2.test.ts | 497 +++++++++++++++++- .../orchestration-v2/Adapters/PiAdapterV2.ts | 22 +- .../pi_abort_outcome/scenarios.ndjson | 26 + 3 files changed, 542 insertions(+), 3 deletions(-) create mode 100644 apps/server/src/orchestration-v2/testkit/fixtures/pi_abort_outcome/scenarios.ndjson diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts index d044d51ee966..d0e50dcb0f09 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.test.ts @@ -1,6 +1,9 @@ import { assert, describe, it } from "@effect/vitest"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { + CommandId, + MessageId, + ProjectId, EnvironmentId, NodeId, ProviderInstanceId, @@ -24,6 +27,7 @@ import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as PlatformError from "effect/PlatformError"; +import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Sink from "effect/Sink"; @@ -51,6 +55,10 @@ import { type ProviderAdapterV2SessionRuntime, } from "../ProviderAdapter.ts"; import { handoffBudget } from "../ContextHandoffBudget.ts"; +import * as Orchestrator from "../Orchestrator.ts"; +import * as EffectWorker from "../EffectWorker.ts"; +import * as ProviderReplayHarness from "../testkit/ProviderReplayHarness.ts"; +import { checkpointWorkspace } from "../testkit/ReplayFixtureWorkspace.ts"; import { makePiAdapterV2, PI_PROVIDER, piLastErrorAt } from "./PiAdapterV2.ts"; import { makePiRpcConnection, type PiRpcRecord } from "./PiRpc.ts"; @@ -359,8 +367,12 @@ const openRuntime = Effect.fnUntraced(function* ( }); const emitted = yield* Queue.unbounded(); const eventsEnded = yield* Deferred.make(); + const observed: Array = []; yield* runtime.events.pipe( - Stream.runForEach((event) => Queue.offer(emitted, event)), + Stream.runForEach((event) => { + observed.push(event); + return Queue.offer(emitted, event); + }), Effect.ensuring(Deferred.succeed(eventsEnded, undefined)), Effect.forkScoped, ); @@ -371,7 +383,7 @@ const openRuntime = Effect.fnUntraced(function* ( if (predicate(event)) return event; } }); - return { runtime, takeEvent, eventsEnded: Deferred.await(eventsEnded) }; + return { runtime, takeEvent, observed, eventsEnded: Deferred.await(eventsEnded) }; }); const makeAppThread = Effect.fnUntraced(function* (model: string, threadId = THREAD_ID) { @@ -476,6 +488,426 @@ const expectModelFailure = (errorMessage: string) => ); }).pipe(Effect.scoped, Effect.provide(layerTest)); +const abortReplayCase = Schema.Struct({ + name: Schema.String, + events: Schema.Array(Schema.Record(Schema.String, Schema.Unknown)), + status: Schema.Literals(["failed", "completed"]), + error: Schema.NullOr(Schema.String), +}); +const decodeAbortReplayCase = Schema.decodeSync(Schema.fromJsonString(abortReplayCase)); +const abortReplayCaseNames = [ + "final-abort", + "recovered-stop", + "recovered-toolUse", + "recovered-length", + "recovered-then-second-abort", + "compaction-no-assistant", + "compaction-will-retry-no-assistant", + "cancelled-compaction", + "failed-compaction", + "custom-and-agent-start", + "compaction-and-assistant", + "noop-compaction-and-assistant", + "retry-abort-success-flag", + "retry-abort-specific-failure", + "specific-provider-failure", + "ordinary-retry", + "overflow-recovery", + "retry-exhaustion-compaction", + "normal-stop", + "normal-tool-use", + "pending-does-not-clear", + "deferred-does-not-clear", + "notifications", + "subordinate-abort", + "review-subordinate-after-last-main", + "review-notify-error-after-last-main", +]; + +// Synthetic protocol scenarios, not a wire capture. Only Pi's process is replaced; +// commands, ingestion, terminalization, checkpoints and projections are real. +describe("Pi abort outcome orchestrator replay", () => { + it.effect.each(abortReplayCaseNames)("%s", (name) => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const fixturePath = yield* path.fromFileUrl( + new URL("../testkit/fixtures/pi_abort_outcome/scenarios.ndjson", import.meta.url), + ); + const transcript = yield* fs.readFileString(fixturePath); + const cases = transcript + .trim() + .split("\n") + .map((line) => decodeAbortReplayCase(line)); + assert.sameMembers( + cases.map((entry) => entry.name), + abortReplayCaseNames, + ); + const scenario = cases.find((entry) => entry.name === name)!; + const fake = yield* makeFakePi; + const adapter = yield* makeAdapter(fake); + const observed: Array = []; + const registry = ProviderAdapterRegistry.layerSingle({ + ...adapter, + openSession: (input) => + adapter.openSession(input).pipe( + Effect.map((runtime) => ({ + ...runtime, + events: runtime.events.pipe( + Stream.tap((event) => Effect.sync(() => observed.push(event))), + ), + })), + ), + }); + const cwd = yield* checkpointWorkspace(`pi-abort-${scenario.name}`); + yield* Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const store = yield* ProjectionStore.ProjectionStoreV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("pi-abort:create"), + threadId: THREAD_ID, + projectId: ProjectId.make("pi-abort:project"), + createdBy: "user", + creationSource: "web", + title: "Pi abort replay", + modelSelection: modelSelection("default"), + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("pi-abort:message"), + threadId: THREAD_ID, + messageId: MessageId.make("pi-abort:message"), + createdBy: "user", + creationSource: "web", + text: "Respond", + attachments: [], + modelSelection: modelSelection("default"), + dispatchMode: { type: "start_immediately" }, + }); + yield* fake.takeRequest("prompt"); + for (const event of scenario.events) yield* fake.emit(event); + yield* fake.emit({ type: "agent_settled" }); + yield* orchestrator.streamStoredEvents.pipe( + Stream.filter( + ({ event }) => + event.type === "run.updated" && + (event.payload.status === "failed" || event.payload.status === "completed"), + ), + Stream.runHead, + ); + yield* worker.drain(); + const projection = yield* orchestrator.getThreadProjection(THREAD_ID); + assert.lengthOf(projection.runs, 1); + assert.equal(projection.runs[0]!.status, scenario.status); + assert.equal(projection.providerTurns.at(-1)?.status, scenario.status); + const terminals = observed.filter((event) => event.type === "turn.terminal"); + assert.lengthOf(terminals, 1); + assert.equal(terminals[0]?.threadDisposition, "reusable"); + const errors = observed.filter( + (event) => + event.type === "provider_session.updated" && event.providerSession.lastError !== null, + ); + assert.lengthOf(errors, scenario.error === null ? 0 : 1); + const session = projection.providerSessions.at(-1)!; + assert.equal(session.lastError, scenario.error); + if (scenario.error !== null) { + assert.isDefined(session.lastErrorAt); + assert.isNotNull(session.lastErrorAt); + const shell = (yield* store.getShellSnapshot()).threads.find( + (thread) => thread.id === THREAD_ID, + )!; + assert.equal(shell.lastError, scenario.error); + assert.equal(shell.lastErrorAt, DateTime.formatIso(session.lastErrorAt!)); + yield* TestClock.adjust("1 second"); + yield* manager.release({ providerSessionId: session.id, reason: "idle_timeout" }); + const after = (yield* store.getThreadProviderContext(THREAD_ID)).providerSessions.at(-1)!; + assert.equal(after.lastError, scenario.error); + assert.deepEqual(after.lastErrorAt, session.lastErrorAt); + } + assert.lengthOf( + fake.allRequests().filter((request) => request["type"] === "prompt"), + 1, + ); + }).pipe( + Effect.provide( + ProviderReplayHarness.layerWithRegistry( + { + name: `pi-abort-${scenario.name}`, + runtimePolicyOverride: { + cwd, + approvalPolicy: "never", + sandboxPolicy: { type: "readOnly" }, + }, + }, + registry, + ), + ), + ); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); +}); + +describe("Pi abort correlated RPC boundaries", () => { + it.effect("keeps Stop interrupted at confirmed idle after an assistant abort", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + const running = yield* takeEvent( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ); + assert.equal(running.type, "provider_turn.updated"); + if (running.type !== "provider_turn.updated") return; + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); + yield* runtime.readThreadSnapshot({ providerThread }); + yield* runtime.interruptTurn({ providerThread, providerTurnId: running.providerTurn.id }); + yield* fake.emit({ type: "agent_settled" }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(terminal.type, "turn.terminal"); + if (terminal.type !== "turn.terminal") return; + assert.equal(terminal.status, "interrupted"); + assert.isNull(terminal.failure); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("a failed idle probe cannot complete an unresolved abort", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); + fake.failNextState(); + yield* fake.emit({ type: "agent_settled" }); + yield* fake.takeRequest("get_state"); + yield* runtime.readThreadSnapshot({ providerThread }); + yield* fake.closeStdout; + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(terminal.type, "turn.terminal"); + if (terminal.type !== "turn.terminal") return; + assert.equal(terminal.status, "failed"); + assert.equal(terminal.failure?.class, "transport_error"); + assert.equal(terminal.failure?.message, "Pi process exited unexpectedly."); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect.each( + [false, true].flatMap((stop) => + (["streaming", "retry", "compaction"] as const).map((phase) => ({ stop, phase })), + ), + )("Stop $stop after abort during $phase", ({ stop, phase }) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + const running = yield* takeEvent( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status === "running", + ); + assert.equal(running.type, "provider_turn.updated"); + if (running.type !== "provider_turn.updated") return; + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); + if (phase === "compaction") yield* fake.emit({ type: "compaction_start", reason: "manual" }); + if (phase === "retry") + yield* fake.emit({ type: "auto_retry_start", attempt: 1, maxAttempts: 2, delayMs: 1 }); + yield* fake.emit({ + type: "extension_ui_request", + method: "notify", + message: "pump fence", + }); + yield* takeEvent( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "dynamic_tool" && + event.turnItem.toolName === "notify", + ); + if (stop) + yield* runtime.interruptTurn({ + providerThread, + providerTurnId: running.providerTurn.id, + }); + yield* fake.closeStdout; + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(terminal.type, "turn.terminal"); + if (terminal.type !== "turn.terminal") return; + assert.equal(terminal.status, stop ? "interrupted" : "failed"); + if (stop) assert.isNull(terminal.failure); + else { + assert.equal(terminal.failure?.class, "transport_error"); + assert.equal(terminal.failure?.message, "Pi process exited unexpectedly."); + } + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("invalidates a held abort idle probe when recovery starts", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent, observed } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); + fake.deferNextState(); + yield* fake.emit({ type: "agent_settled" }); + yield* fake.takeRequest("get_state"); + yield* fake.emit({ type: "compaction_start", reason: "manual" }); + yield* takeEvent( + (event) => event.type === "turn_item.updated" && event.turnItem.type === "compaction", + ); + fake.queueState({ isStreaming: true, isCompacting: false, pendingMessageCount: 0 }); + yield* fake.emit({ + type: "compaction_end", + result: { summary: "smaller" }, + willRetry: false, + }); + yield* fake.takeRequest("get_state"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.resolveDeferredState({ + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }); + // A subsequent correlated request puts the stale probe ahead of recovery's + // successful completion without relying on a wall-clock absence check. + yield* runtime.readThreadSnapshot({ providerThread }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "stop" }, + }); + yield* fake.emit({ type: "agent_settled" }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(terminal.type, "turn.terminal"); + if (terminal.type !== "turn.terminal") return; + assert.equal(terminal.status, "completed"); + assert.lengthOf( + observed.filter((event) => event.type === "turn.terminal"), + 1, + ); + assert.lengthOf( + observed.filter( + (event) => + event.type === "provider_session.updated" && event.providerSession.lastError !== null, + ), + 0, + ); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("publishes abort once and isolates duplicate settlement and late old-turn probes", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent, observed } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + const occurrences: Array = []; + for (const runOrdinal of [1, 2, 3]) { + yield* startTurn(runtime, providerThread, "default", [], "Respond", undefined, runOrdinal); + yield* fake.takeRequest("prompt"); + yield* TestClock.adjust("1 second"); + yield* fake.emit({ type: "agent_start" }); + if (runOrdinal === 2) { + yield* fake.resolveDeferredState({ + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }); + yield* runtime.readThreadSnapshot({ providerThread }); + } + const stopReason = runOrdinal === 2 ? "stop" : "aborted"; + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason }, + }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason }, + }); + if (runOrdinal === 1) { + fake.deferNextState(); + yield* fake.emit({ type: "agent_settled" }); + yield* fake.takeRequest("get_state"); + } + yield* fake.emit({ type: "agent_settled" }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(terminal.type, "turn.terminal"); + if (terminal.type !== "turn.terminal") return; + assert.equal(terminal.status, runOrdinal === 2 ? "completed" : "failed"); + assert.equal(terminal.runOrdinal, runOrdinal); + const updates = observed.filter((event) => event.type === "provider_session.updated"); + const session = updates.at(-1)!.providerSession; + if (runOrdinal === 2) { + assert.isNull(session.lastError); + assert.isNull(session.lastErrorAt); + } else { + assert.equal(session.lastError, "Pi aborted the response before completion."); + assert.isDefined(session.lastErrorAt); + assert.isNotNull(session.lastErrorAt); + occurrences.push(DateTime.toEpochMillis(session.lastErrorAt!)); + } + } + assert.lengthOf( + observed.filter((event) => event.type === "turn.terminal"), + 3, + ); + assert.lengthOf( + observed.filter( + (event) => + event.type === "provider_session.updated" && event.providerSession.lastError !== null, + ), + 2, + ); + assert.isAbove(occurrences[1]!, occurrences[0]!); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); +}); + describe("PiAdapterV2", () => { it("keeps the occurrence identity on same-text refreshes, including legacy null", () => { const now = DateTime.makeUnsafe("2026-09-20T12:00:00Z"); @@ -2433,6 +2865,10 @@ describe("PiAdapterV2", () => { yield* fake.takeRequest("prompt"); yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); // Extension ctx.compact() waits for this first settlement, then starts // compaction in a detached continuation. fake.queueState({ isStreaming: false, isCompacting: true, pendingMessageCount: 0 }); @@ -2459,6 +2895,10 @@ describe("PiAdapterV2", () => { event.turnItem.status === "completed", ); yield* fake.takeRequest("get_state"); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "stop" }, + }); fake.queueState({ isStreaming: false, isCompacting: false, pendingMessageCount: 0 }); yield* fake.emit({ type: "agent_settled" }); @@ -2792,3 +3232,56 @@ describe("PiRpc early process exit", () => { }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)), ); }); + +describe("B09 review probes", () => { + it.effect("review: late old-turn probe with a coinciding generation stays isolated", () => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime, takeEvent, observed } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(runtime, providerThread, "default", [], "Respond", undefined, 1); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); + fake.deferNextState(); + yield* fake.emit({ type: "agent_settled" }); + yield* fake.takeRequest("get_state"); + yield* fake.emit({ type: "agent_settled" }); + const first = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(first.type === "turn.terminal" ? first.status : null, "failed"); + yield* startTurn(runtime, providerThread, "default", [], "Respond", undefined, 2); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "aborted" }, + }); + yield* fake.emit({ type: "agent_start" }); + yield* fake.resolveDeferredState({ + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }); + yield* runtime.readThreadSnapshot({ providerThread }); + yield* fake.emit({ + type: "message_end", + message: { role: "assistant", content: [], stopReason: "stop" }, + }); + yield* fake.emit({ type: "agent_settled" }); + const second = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal(second.type === "turn.terminal" ? second.status : null, "completed"); + assert.equal(second.type === "turn.terminal" ? second.runOrdinal : null, 2); + assert.lengthOf( + observed.filter((event) => event.type === "turn.terminal"), + 2, + ); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); +}); diff --git a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts index 10fc3ea1b548..a04094bddc2d 100644 --- a/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/PiAdapterV2.ts @@ -318,6 +318,7 @@ interface ActivePiTurn { */ readonly toolStartedAt: Map; interrupted: boolean; + unresolvedAssistantAbort: boolean; /** * Whether any agent run activity was observed. Command-only prompts (pure * extension slash commands) never start an agent run and never emit @@ -1631,7 +1632,17 @@ export function makePiAdapterV2( const message = event["message"]; if (recordString(message, "role") !== "assistant") return; yield* completeOpenStreamItems(turn); - if (recordString(message, "stopReason") === "error" && turn.failure === null) { + const stopReason = recordString(message, "stopReason"); + if (stopReason === "aborted") { + turn.unresolvedAssistantAbort = true; + } else if ( + stopReason === "stop" || + stopReason === "toolUse" || + stopReason === "length" + ) { + turn.unresolvedAssistantAbort = false; + } + if (stopReason === "error" && turn.failure === null) { turn.failure = makeProviderFailure({ message: recordString(message, "errorMessage") ?? "Pi reported a model error.", class: "provider_error", @@ -1942,6 +1953,14 @@ export function makePiAdapterV2( (recordNumber(data, "pendingMessageCount") ?? 0) === 0 ) { turn.settleWhenIdle = false; + // An aborted attempt can recover through extension work. Classify it + // only after the current turn's idle probe confirms settlement. + if (!turn.interrupted && turn.failure === null && turn.unresolvedAssistantAbort) { + turn.failure = makeProviderFailure({ + message: "Pi aborted the response before completion.", + class: "provider_error", + }); + } if (state !== null) yield* finalizeTurn(state); } return; @@ -2357,6 +2376,7 @@ export function makePiAdapterV2( toolArgs: new Map(), toolStartedAt: new Map(), interrupted: false, + unresolvedAssistantAbort: false, sawAgentActivity: false, promptMayBeCommandOnly: compactCommand !== null || (payload?.message.trimStart().startsWith("/") ?? false), diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/pi_abort_outcome/scenarios.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/pi_abort_outcome/scenarios.ndjson new file mode 100644 index 000000000000..457ef1aeb371 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/pi_abort_outcome/scenarios.ndjson @@ -0,0 +1,26 @@ +{"name":"final-abort","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"recovered-stop","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name":"recovered-toolUse","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"toolUse"}}],"status":"completed","error":null} +{"name":"recovered-length","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"length"}}],"status":"completed","error":null} +{"name":"recovered-then-second-abort","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"toolUse"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"compaction-no-assistant","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","result":{"summary":"smaller","tokensBefore":10000,"estimatedTokensAfter":2000},"aborted":false,"willRetry":false}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"compaction-will-retry-no-assistant","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","result":{"summary":"smaller","tokensBefore":10000,"estimatedTokensAfter":2000},"aborted":false,"willRetry":true}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"cancelled-compaction","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","aborted":true}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"failed-compaction","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","errorMessage":"compact failed"}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"custom-and-agent-start","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"custom","content":[],"stopReason":"aborted","customType":"pi-xai-ws-loop-recovery"}},{"type":"agent_start"}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"compaction-and-assistant","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","result":{"summary":"smaller","tokensBefore":10000,"estimatedTokensAfter":2000},"aborted":false,"willRetry":false},{"type":"message_end","message":{"role":"custom","content":[],"stopReason":"aborted","customType":"pi-xai-ws-loop-recovery"}},{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name":"noop-compaction-and-assistant","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","errorMessage":"too small"},{"type":"message_end","message":{"role":"custom","content":[],"stopReason":"aborted","customType":"pi-xai-ws-loop-recovery"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name":"retry-abort-success-flag","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"capacity exhausted"}},{"type":"auto_retry_start","attempt":1,"maxAttempts":2,"delayMs":1,"errorMessage":"capacity exhausted"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"auto_retry_end","success":true,"attempt":1}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"retry-abort-specific-failure","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"capacity exhausted"}},{"type":"auto_retry_start","attempt":1,"maxAttempts":2,"delayMs":1,"errorMessage":"capacity exhausted"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"auto_retry_end","success":false,"finalError":"retry exhausted","attempt":2}],"status":"failed","error":"retry exhausted"} +{"name":"specific-provider-failure","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"capacity exhausted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}}],"status":"failed","error":"capacity exhausted"} +{"name":"ordinary-retry","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"capacity exhausted"}},{"type":"auto_retry_start","attempt":1,"maxAttempts":2,"delayMs":1,"errorMessage":"capacity exhausted"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}},{"type":"auto_retry_end","success":true,"attempt":1}],"status":"completed","error":null} +{"name":"overflow-recovery","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"capacity exhausted"}},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","result":{"summary":"smaller","tokensBefore":10000,"estimatedTokensAfter":2000},"aborted":false,"willRetry":true},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name":"retry-exhaustion-compaction","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"capacity exhausted"}},{"type":"auto_retry_start","attempt":1,"maxAttempts":2,"delayMs":1,"errorMessage":"capacity exhausted"},{"type":"auto_retry_end","success":false,"finalError":"retry exhausted","attempt":2},{"type":"compaction_start","reason":"manual"},{"type":"compaction_end","result":{"summary":"smaller","tokensBefore":10000,"estimatedTokensAfter":2000},"aborted":false,"willRetry":false}],"status":"failed","error":"retry exhausted"} +{"name":"normal-stop","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name":"normal-tool-use","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"toolUse"}}],"status":"completed","error":null} +{"name":"pending-does-not-clear","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"pending"}}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"deferred-does-not-clear","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"aborted"}},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"deferred"}}],"status":"failed","error":"Pi aborted the response before completion."} +{"name":"notifications","events":[{"type":"agent_start"},{"type":"extension_ui_request","id":"notify:info","method":"notify","notifyType":"info","message":"Recovery cooldown exhausted; response aborted"},{"type":"extension_ui_request","id":"notify:warning","method":"notify","notifyType":"warning","message":"Recovery cooldown exhausted; response aborted"},{"type":"extension_ui_request","id":"notify:error","method":"notify","notifyType":"error","message":"Recovery cooldown exhausted; response aborted"},{"type":"extension_error","extensionPath":"test","event":"turn_end","error":"aborted"},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name":"subordinate-abort","events":[{"type":"agent_start"},{"type":"message_end","message":{"role":"toolResult","content":[],"stopReason":"aborted"}},{"type":"tool_execution_start","toolCallId":"child","toolName":"subagent","args":{}},{"type":"tool_execution_end","toolCallId":"child","toolName":"subagent","result":{"content":[{"type":"text","text":"subagent aborted"}],"details":{"stopReason":"aborted"}},"isError":true},{"type":"message_end","message":{"role":"assistant","content":[],"stopReason":"stop"}}],"status":"completed","error":null} +{"name": "review-subordinate-after-last-main", "status": "completed", "error": null, "events": [{"type": "agent_start"}, {"type": "message_end", "message": {"role": "assistant", "content": [], "stopReason": "toolUse"}}, {"type": "tool_execution_start", "toolCallId": "child", "toolName": "subagent", "args": {}}, {"type": "tool_execution_end", "toolCallId": "child", "toolName": "subagent", "result": {"content": [{"type": "text", "text": "subagent aborted"}], "details": {"stopReason": "aborted"}}, "isError": true}, {"type": "message_end", "message": {"role": "toolResult", "content": [], "stopReason": "aborted"}}]} +{"name": "review-notify-error-after-last-main", "status": "completed", "error": null, "events": [{"type": "agent_start"}, {"type": "message_end", "message": {"role": "assistant", "content": [], "stopReason": "stop"}}, {"type": "extension_ui_request", "id": "n1", "method": "notify", "notifyType": "error", "message": "Loop recovery exhausted after repeated output; cooldown active."}, {"type": "extension_error", "extensionPath": "/x/pi-xai-ws", "event": "turn_end", "error": "aborted"}]}