diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index a5e2d53884eeaf4217499129e33d20a8a23d8823..944ffe2bee7c60b41d6b691bfdc4f2ffab206e4f 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -1045,97 +1045,112 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(layerTest)), ); - it.effect("defers a wake's late settlement until the joined user prompt is acknowledged", () => - Effect.gen(function* () { - const fake = yield* makeFakePi; - const offers = - yield* Queue.unbounded(); - const { runtime, takeEvent } = yield* openRuntime( - fake, - "default", - THREAD_ID, - SESSION_ID, - undefined, - { - offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid), - }, - ); - const providerThread = yield* runtime.ensureThread({ - threadId: THREAD_ID, - modelSelection: modelSelection("default"), - runtimePolicy, - }); - yield* fake.emit({ type: "agent_start" }); - yield* Queue.take(offers); - yield* startTurn(runtime, providerThread); - yield* fake.takeRequest("prompt"); - yield* fake.emit({ type: "agent_settled" }); - yield* fake.emit({ - type: "extension_ui_request", - method: "notify", - message: "Ordering barrier", - }); - yield* takeEvent( - (event) => - event.type === "turn_item.updated" && - event.turnItem.type === "dynamic_tool" && - event.turnItem.toolName === "notify", - ); - assert.equal(runtime.providerSession.status, "running"); - fake.queueState({ isStreaming: true }); - yield* fake.emit({ type: "response", command: "prompt", success: true }); - yield* fake.takeRequest("get_state"); - yield* fake.emit({ type: "agent_start" }); - yield* fake.emit({ type: "agent_settled" }); - yield* takeEvent((event) => event.type === "turn.terminal"); - }).pipe(Effect.scoped, Effect.provide(layerTest)), + it.effect.each([false, true])( + "defers a wake's late settlement until the joined user prompt is acknowledged, aborted=%s", + (aborted) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const offers = + yield* Queue.unbounded(); + const { runtime, takeEvent } = yield* openRuntime( + fake, + "default", + THREAD_ID, + SESSION_ID, + undefined, + { + offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid), + }, + ); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* fake.emit({ type: "agent_start" }); + yield* Queue.take(offers); + yield* startTurn(runtime, providerThread); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_settled", aborted }); + yield* fake.emit({ + type: "extension_ui_request", + method: "notify", + message: "Ordering barrier", + }); + yield* takeEvent( + (event) => + event.type === "turn_item.updated" && + event.turnItem.type === "dynamic_tool" && + event.turnItem.toolName === "notify", + ); + assert.equal(runtime.providerSession.status, "running"); + fake.queueState({ isStreaming: true }); + yield* fake.emit({ type: "response", command: "prompt", success: true }); + yield* fake.takeRequest("get_state"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ type: "agent_settled" }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), ); - it.effect("settles a command-only user prompt after taking a completed wake", () => - Effect.gen(function* () { - const fake = yield* makeFakePi; - const offers = - yield* Queue.unbounded(); - const { runtime, takeEvent } = yield* openRuntime( - fake, - "default", - THREAD_ID, - SESSION_ID, - undefined, - { - offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid), - }, - ); - const providerThread = yield* runtime.ensureThread({ - threadId: THREAD_ID, - modelSelection: modelSelection("default"), - runtimePolicy, - }); - yield* fake.emit({ type: "agent_start" }); - yield* Queue.take(offers); - yield* fake.emit({ type: "agent_settled" }); - yield* takeEvent( - (event) => - event.type === "provider_session.updated" && event.providerSession.status === "ready", - ); - yield* startTurn(runtime, providerThread, "default", [], "/hello"); - yield* fake.takeRequest("prompt"); - yield* fake.emit({ type: "response", command: "prompt", success: true }); - yield* takeEvent((event) => event.type === "turn.terminal"); - yield* startTurn( - runtime, - providerThread, - "default", - [], - "Stale continuation", - undefined, - 2, - THREAD_ID, - true, - ); - yield* takeEvent((event) => event.type === "turn.terminal"); - assert.equal(fake.allRequests().filter((request) => request["type"] === "prompt").length, 1); - }).pipe(Effect.scoped, Effect.provide(layerTest)), + it.effect.each([ + { aborted: false, resumes: false, expected: "completed" }, + { aborted: true, resumes: false, expected: "interrupted" }, + { aborted: true, resumes: true, expected: "completed" }, + ])( + "settles a completed wake, aborted=$aborted, resumes=$resumes", + ({ aborted, resumes, expected }) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const offers = + yield* Queue.unbounded(); + const { runtime, takeEvent } = yield* openRuntime( + fake, + "default", + THREAD_ID, + SESSION_ID, + undefined, + { + offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid), + }, + ); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* fake.emit({ type: "agent_start" }); + yield* Queue.take(offers); + yield* fake.emit({ type: "agent_settled", aborted }); + yield* takeEvent( + (event) => + event.type === "provider_session.updated" && event.providerSession.status === "ready", + ); + yield* startTurn(runtime, providerThread, "default", [], "/hello"); + yield* fake.takeRequest("prompt"); + if (resumes) yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ type: "response", command: "prompt", success: true }); + if (resumes) yield* fake.emit({ type: "agent_settled", aborted: false }); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue(terminal.type === "turn.terminal" && terminal.status === expected); + yield* startTurn( + runtime, + providerThread, + "default", + [], + "Stale continuation", + undefined, + 2, + THREAD_ID, + true, + ); + yield* takeEvent((event) => event.type === "turn.terminal"); + assert.equal( + fake.allRequests().filter((request) => request["type"] === "prompt").length, + 1, + ); + }).pipe(Effect.scoped, Effect.provide(layerTest)), ); it.effect("keeps a pending wake on its thread and lets Stop retire its process", () => @@ -3508,6 +3523,79 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(layerTest)), ); + it.effect.each([undefined, false, true])( + "preserves the native settlement outcome, aborted=%s", + (aborted) => + 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: "agent_settled", + ...(aborted === undefined ? {} : { aborted }), + }); + const completed = yield* takeEvent( + (event) => + event.type === "provider_turn.updated" && event.providerTurn.status !== "running", + ); + const status = aborted === true ? "interrupted" : "completed"; + assert.isTrue( + completed.type === "provider_turn.updated" && completed.providerTurn.status === status, + ); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue( + terminal.type === "turn.terminal" && + terminal.status === status && + terminal.failure === null, + ); + assert.equal(runtime.providerSession.status, "ready"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect("stops retry progress when Pi aborts the native run", () => + 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: "auto_retry_start", + attempt: 2, + maxAttempts: 5, + delayMs: 6_000, + errorMessage: "temporary network failure", + }); + const retrying = yield* takeEvent( + (event) => event.type === "turn_item.updated" && event.turnItem.type === "error", + ); + yield* fake.emit({ type: "agent_settled", aborted: true }); + const stopped = yield* takeEvent( + (event) => event.type === "turn_item.updated" && event.turnItem.type === "error", + ); + assert.isTrue( + retrying.type === "turn_item.updated" && + stopped.type === "turn_item.updated" && + stopped.turnItem.id === retrying.turnItem.id && + stopped.turnItem.status === "interrupted", + ); + const terminal = yield* takeEvent((event) => event.type === "turn.terminal"); + assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "interrupted"); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect("stops active retry progress when the turn is interrupted", () => Effect.gen(function* () { const fake = yield* makeFakePi; diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index 973ef4ebb6aa1321296ea76aaee94f6c19779683..7022be2135ab688d2a5dce19b7253e094727bd96 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -410,6 +410,8 @@ interface ActivePiTurn { */ readonly toolStartedAt: Map; interrupted: boolean; + /** Cancellation of the latest native run; new work can recover before settlement. */ + nativeAborted: boolean; /** * Whether any agent run activity was observed. Command-only prompts (pure * extension slash commands) never start an agent run and never emit @@ -1574,19 +1576,16 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( const turn = state.activeTurn; if (turn === null) return; state.activeTurn = null; + const interrupted = turn.interrupted || turn.nativeAborted; const completedAt = yield* DateTime.now; yield* completeOpenStreamItems(turn); if (turn.activeCompaction !== null) { - const status = turn.interrupted - ? "cancelled" - : turn.failure === null - ? "completed" - : "failed"; + const status = interrupted ? "cancelled" : turn.failure === null ? "completed" : "failed"; yield* emitCompaction(turn, turn.activeCompaction, status); turn.activeCompaction = null; } if (turn.activeProviderRetry !== null) { - if (turn.interrupted) { + if (interrupted) { yield* emitProviderRetry(turn, turn.activeProviderRetry, "interrupted", completedAt); turn.activeProviderRetry = null; } else if (turn.failure === null) { @@ -1600,7 +1599,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( const tokenUsage = readUsage ? yield* readTokenUsage(turn.latestCompactionAfterTokens, completedAt) : undefined; - const failure = turn.interrupted ? null : (turn.rejectedPromptFailure ?? turn.failure); + const failure = interrupted ? null : (turn.rejectedPromptFailure ?? turn.failure); yield* emit({ type: "provider_turn.updated", driver: PI_PROVIDER, @@ -1610,7 +1609,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( ...(treeRefs === null || treeRefs.nativeTurnRef === undefined ? {} : { nativeTurnRef: treeRefs.nativeTurnRef }), - status: turn.interrupted ? "interrupted" : failure !== null ? "failed" : "completed", + status: interrupted ? "interrupted" : failure !== null ? "failed" : "completed", completedAt, ...(tokenUsage === undefined ? {} : { tokenUsage }), }, @@ -1679,7 +1678,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( providerThreadId: state.providerThread.id, providerTurnId: turn.providerTurn.id, runOrdinal: turn.turnInput.runOrdinal, - status: turn.interrupted ? "interrupted" : "completed", + status: interrupted ? "interrupted" : "completed", failure: null, threadDisposition: "reusable", }); @@ -1799,6 +1798,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( return; } turn.sawAgentActivity = true; + turn.nativeAborted = false; turn.settleProbeGeneration += 1; return; } @@ -2018,6 +2018,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( return; } if (turn !== null) { + turn.nativeAborted = event["aborted"] === true; turn.settleWhenIdle = true; turn.settleProbeGeneration += 1; // A wake can settle while the newly joined user prompt is still @@ -2610,6 +2611,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( toolArgs: new Map(), toolStartedAt: new Map(), interrupted: false, + nativeAborted: false, sawAgentActivity: false, adoptedWake: false, promptMayBeCommandOnly: @@ -2682,6 +2684,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( if (wake !== null) { for (const event of wake.events) { if (!continuation && event["type"] === "agent_settled") { + activeTurn.nativeAborted = event["aborted"] === true; activeTurn.settleWhenIdle = true; continue; }