authorgravatar for 71498452+StiensWout@users.noreply.github.comWout Stiens <71498452+StiensWout@users.noreply.github.com> 2026-10-11 00:46:57+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-10 17:46:57-05:00
log7e8071f1f1566220fb52cbfaaf97a2bc4e93a193
tree8ee4b159160fb59adb11f6413f5d1b18f2d23dcc
parente408cac0d7d903ccfcecfb906d4e48292294ab55
signature Signed by PGP key B5690EEEBB952194

fix(pi): preserve native abort outcomes (#17853)

2 files changed, 189 insertions(+), 98 deletions(-)

packages/provider-pi/src/server/adapter.test.ts+177-89
...@@ -1045,97 +1045,112 @@ describe("PiAdapterV2", () => {...@@ -1045,97 +1045,112 @@ describe("PiAdapterV2", () => {
1045 }).pipe(Effect.scoped, Effect.provide(layerTest)),1045 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1046 );1046 );
10471047
1048 it.effect("defers a wake's late settlement until the joined user prompt is acknowledged", () =>1048 it.effect.each([false, true])(
1049 Effect.gen(function* () {1049 "defers a wake's late settlement until the joined user prompt is acknowledged, aborted=%s",
1050 const fake = yield* makeFakePi;1050 (aborted) =>
1051 const offers =1051 Effect.gen(function* () {
1052 yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();1052 const fake = yield* makeFakePi;
1053 const { runtime, takeEvent } = yield* openRuntime(1053 const offers =
1054 fake,1054 yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1055 "default",1055 const { runtime, takeEvent } = yield* openRuntime(
1056 THREAD_ID,1056 fake,
1057 SESSION_ID,1057 "default",
1058 undefined,1058 THREAD_ID,
1059 {1059 SESSION_ID,
1060 offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),1060 undefined,
1061 },1061 {
1062 );1062 offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1063 const providerThread = yield* runtime.ensureThread({1063 },
1064 threadId: THREAD_ID,1064 );
1065 modelSelection: modelSelection("default"),1065 const providerThread = yield* runtime.ensureThread({
1066 runtimePolicy,1066 threadId: THREAD_ID,
1067 });1067 modelSelection: modelSelection("default"),
1068 yield* fake.emit({ type: "agent_start" });1068 runtimePolicy,
1069 yield* Queue.take(offers);1069 });
1070 yield* startTurn(runtime, providerThread);1070 yield* fake.emit({ type: "agent_start" });
1071 yield* fake.takeRequest("prompt");1071 yield* Queue.take(offers);
1072 yield* fake.emit({ type: "agent_settled" });1072 yield* startTurn(runtime, providerThread);
1073 yield* fake.emit({1073 yield* fake.takeRequest("prompt");
1074 type: "extension_ui_request",1074 yield* fake.emit({ type: "agent_settled", aborted });
1075 method: "notify",1075 yield* fake.emit({
1076 message: "Ordering barrier",1076 type: "extension_ui_request",
1077 });1077 method: "notify",
1078 yield* takeEvent(1078 message: "Ordering barrier",
1079 (event) =>1079 });
1080 event.type === "turn_item.updated" &&1080 yield* takeEvent(
1081 event.turnItem.type === "dynamic_tool" &&1081 (event) =>
1082 event.turnItem.toolName === "notify",1082 event.type === "turn_item.updated" &&
1083 );1083 event.turnItem.type === "dynamic_tool" &&
1084 assert.equal(runtime.providerSession.status, "running");1084 event.turnItem.toolName === "notify",
1085 fake.queueState({ isStreaming: true });1085 );
1086 yield* fake.emit({ type: "response", command: "prompt", success: true });1086 assert.equal(runtime.providerSession.status, "running");
1087 yield* fake.takeRequest("get_state");1087 fake.queueState({ isStreaming: true });
1088 yield* fake.emit({ type: "agent_start" });1088 yield* fake.emit({ type: "response", command: "prompt", success: true });
1089 yield* fake.emit({ type: "agent_settled" });1089 yield* fake.takeRequest("get_state");
1090 yield* takeEvent((event) => event.type === "turn.terminal");1090 yield* fake.emit({ type: "agent_start" });
1091 }).pipe(Effect.scoped, Effect.provide(layerTest)),1091 yield* fake.emit({ type: "agent_settled" });
1092 const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
1093 assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "completed");
1094 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1092 );1095 );
10931096
1094 it.effect("settles a command-only user prompt after taking a completed wake", () =>1097 it.effect.each([
1095 Effect.gen(function* () {1098 { aborted: false, resumes: false, expected: "completed" },
1096 const fake = yield* makeFakePi;1099 { aborted: true, resumes: false, expected: "interrupted" },
1097 const offers =1100 { aborted: true, resumes: true, expected: "completed" },
1098 yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();1101 ])(
1099 const { runtime, takeEvent } = yield* openRuntime(1102 "settles a completed wake, aborted=$aborted, resumes=$resumes",
1100 fake,1103 ({ aborted, resumes, expected }) =>
1101 "default",1104 Effect.gen(function* () {
1102 THREAD_ID,1105 const fake = yield* makeFakePi;
1103 SESSION_ID,1106 const offers =
1104 undefined,1107 yield* Queue.unbounded<ProviderContinuationRequests.ProviderContinuationRequest>();
1105 {1108 const { runtime, takeEvent } = yield* openRuntime(
1106 offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),1109 fake,
1107 },1110 "default",
1108 );1111 THREAD_ID,
1109 const providerThread = yield* runtime.ensureThread({1112 SESSION_ID,
1110 threadId: THREAD_ID,1113 undefined,
1111 modelSelection: modelSelection("default"),1114 {
1112 runtimePolicy,1115 offer: (request) => Queue.offer(offers, request).pipe(Effect.asVoid),
1113 });1116 },
1114 yield* fake.emit({ type: "agent_start" });1117 );
1115 yield* Queue.take(offers);1118 const providerThread = yield* runtime.ensureThread({
1116 yield* fake.emit({ type: "agent_settled" });1119 threadId: THREAD_ID,
1117 yield* takeEvent(1120 modelSelection: modelSelection("default"),
1118 (event) =>1121 runtimePolicy,
1119 event.type === "provider_session.updated" && event.providerSession.status === "ready",1122 });
1120 );1123 yield* fake.emit({ type: "agent_start" });
1121 yield* startTurn(runtime, providerThread, "default", [], "/hello");1124 yield* Queue.take(offers);
1122 yield* fake.takeRequest("prompt");1125 yield* fake.emit({ type: "agent_settled", aborted });
1123 yield* fake.emit({ type: "response", command: "prompt", success: true });1126 yield* takeEvent(
1124 yield* takeEvent((event) => event.type === "turn.terminal");1127 (event) =>
1125 yield* startTurn(1128 event.type === "provider_session.updated" && event.providerSession.status === "ready",
1126 runtime,1129 );
1127 providerThread,1130 yield* startTurn(runtime, providerThread, "default", [], "/hello");
1128 "default",1131 yield* fake.takeRequest("prompt");
1129 [],1132 if (resumes) yield* fake.emit({ type: "agent_start" });
1130 "Stale continuation",1133 yield* fake.emit({ type: "response", command: "prompt", success: true });
1131 undefined,1134 if (resumes) yield* fake.emit({ type: "agent_settled", aborted: false });
1132 2,1135 const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
1133 THREAD_ID,1136 assert.isTrue(terminal.type === "turn.terminal" && terminal.status === expected);
1134 true,1137 yield* startTurn(
1135 );1138 runtime,
1136 yield* takeEvent((event) => event.type === "turn.terminal");1139 providerThread,
1137 assert.equal(fake.allRequests().filter((request) => request["type"] === "prompt").length, 1);1140 "default",
1138 }).pipe(Effect.scoped, Effect.provide(layerTest)),1141 [],
1142 "Stale continuation",
1143 undefined,
1144 2,
1145 THREAD_ID,
1146 true,
1147 );
1148 yield* takeEvent((event) => event.type === "turn.terminal");
1149 assert.equal(
1150 fake.allRequests().filter((request) => request["type"] === "prompt").length,
1151 1,
1152 );
1153 }).pipe(Effect.scoped, Effect.provide(layerTest)),
1139 );1154 );
11401155
1141 it.effect("keeps a pending wake on its thread and lets Stop retire its process", () =>1156 it.effect("keeps a pending wake on its thread and lets Stop retire its process", () =>
...@@ -3508,6 +3523,79 @@ describe("PiAdapterV2", () => {...@@ -3508,6 +3523,79 @@ describe("PiAdapterV2", () => {
3508 }).pipe(Effect.scoped, Effect.provide(layerTest)),3523 }).pipe(Effect.scoped, Effect.provide(layerTest)),
3509 );3524 );
35103525
3526 it.effect.each([undefined, false, true])(
3527 "preserves the native settlement outcome, aborted=%s",
3528 (aborted) =>
3529 Effect.gen(function* () {
3530 const fake = yield* makeFakePi;
3531 const { runtime, takeEvent } = yield* openRuntime(fake);
3532 const providerThread = yield* runtime.ensureThread({
3533 threadId: THREAD_ID,
3534 modelSelection: modelSelection("default"),
3535 runtimePolicy,
3536 });
3537 yield* startTurn(runtime, providerThread);
3538 yield* fake.takeRequest("prompt");
3539 yield* fake.emit({ type: "agent_start" });
3540 yield* fake.emit({
3541 type: "agent_settled",
3542 ...(aborted === undefined ? {} : { aborted }),
3543 });
3544 const completed = yield* takeEvent(
3545 (event) =>
3546 event.type === "provider_turn.updated" && event.providerTurn.status !== "running",
3547 );
3548 const status = aborted === true ? "interrupted" : "completed";
3549 assert.isTrue(
3550 completed.type === "provider_turn.updated" && completed.providerTurn.status === status,
3551 );
3552 const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3553 assert.isTrue(
3554 terminal.type === "turn.terminal" &&
3555 terminal.status === status &&
3556 terminal.failure === null,
3557 );
3558 assert.equal(runtime.providerSession.status, "ready");
3559 }).pipe(Effect.scoped, Effect.provide(layerTest)),
3560 );
3561
3562 it.effect("stops retry progress when Pi aborts the native run", () =>
3563 Effect.gen(function* () {
3564 const fake = yield* makeFakePi;
3565 const { runtime, takeEvent } = yield* openRuntime(fake);
3566 const providerThread = yield* runtime.ensureThread({
3567 threadId: THREAD_ID,
3568 modelSelection: modelSelection("default"),
3569 runtimePolicy,
3570 });
3571 yield* startTurn(runtime, providerThread);
3572 yield* fake.takeRequest("prompt");
3573 yield* fake.emit({ type: "agent_start" });
3574 yield* fake.emit({
3575 type: "auto_retry_start",
3576 attempt: 2,
3577 maxAttempts: 5,
3578 delayMs: 6_000,
3579 errorMessage: "temporary network failure",
3580 });
3581 const retrying = yield* takeEvent(
3582 (event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3583 );
3584 yield* fake.emit({ type: "agent_settled", aborted: true });
3585 const stopped = yield* takeEvent(
3586 (event) => event.type === "turn_item.updated" && event.turnItem.type === "error",
3587 );
3588 assert.isTrue(
3589 retrying.type === "turn_item.updated" &&
3590 stopped.type === "turn_item.updated" &&
3591 stopped.turnItem.id === retrying.turnItem.id &&
3592 stopped.turnItem.status === "interrupted",
3593 );
3594 const terminal = yield* takeEvent((event) => event.type === "turn.terminal");
3595 assert.isTrue(terminal.type === "turn.terminal" && terminal.status === "interrupted");
3596 }).pipe(Effect.scoped, Effect.provide(layerTest)),
3597 );
3598
3511 it.effect("stops active retry progress when the turn is interrupted", () =>3599 it.effect("stops active retry progress when the turn is interrupted", () =>
3512 Effect.gen(function* () {3600 Effect.gen(function* () {
3513 const fake = yield* makeFakePi;3601 const fake = yield* makeFakePi;
packages/provider-pi/src/server/adapter.ts+12-9
...@@ -410,6 +410,8 @@ interface ActivePiTurn {...@@ -410,6 +410,8 @@ interface ActivePiTurn {
410 */410 */
411 readonly toolStartedAt: Map<string, DateTime.Utc>;411 readonly toolStartedAt: Map<string, DateTime.Utc>;
412 interrupted: boolean;412 interrupted: boolean;
413 /** Cancellation of the latest native run; new work can recover before settlement. */
414 nativeAborted: boolean;
413 /**415 /**
414 * Whether any agent run activity was observed. Command-only prompts (pure416 * Whether any agent run activity was observed. Command-only prompts (pure
415 * extension slash commands) never start an agent run and never emit417 * extension slash commands) never start an agent run and never emit
...@@ -1574,19 +1576,16 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1574,19 +1576,16 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
1574 const turn = state.activeTurn;1576 const turn = state.activeTurn;
1575 if (turn === null) return;1577 if (turn === null) return;
1576 state.activeTurn = null;1578 state.activeTurn = null;
1579 const interrupted = turn.interrupted || turn.nativeAborted;
1577 const completedAt = yield* DateTime.now;1580 const completedAt = yield* DateTime.now;
1578 yield* completeOpenStreamItems(turn);1581 yield* completeOpenStreamItems(turn);
1579 if (turn.activeCompaction !== null) {1582 if (turn.activeCompaction !== null) {
1580 const status = turn.interrupted1583 const status = interrupted ? "cancelled" : turn.failure === null ? "completed" : "failed";
1581 ? "cancelled"
1582 : turn.failure === null
1583 ? "completed"
1584 : "failed";
1585 yield* emitCompaction(turn, turn.activeCompaction, status);1584 yield* emitCompaction(turn, turn.activeCompaction, status);
1586 turn.activeCompaction = null;1585 turn.activeCompaction = null;
1587 }1586 }
1588 if (turn.activeProviderRetry !== null) {1587 if (turn.activeProviderRetry !== null) {
1589 if (turn.interrupted) {1588 if (interrupted) {
1590 yield* emitProviderRetry(turn, turn.activeProviderRetry, "interrupted", completedAt);1589 yield* emitProviderRetry(turn, turn.activeProviderRetry, "interrupted", completedAt);
1591 turn.activeProviderRetry = null;1590 turn.activeProviderRetry = null;
1592 } else if (turn.failure === null) {1591 } else if (turn.failure === null) {
...@@ -1600,7 +1599,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1600,7 +1599,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
1600 const tokenUsage = readUsage1599 const tokenUsage = readUsage
1601 ? yield* readTokenUsage(turn.latestCompactionAfterTokens, completedAt)1600 ? yield* readTokenUsage(turn.latestCompactionAfterTokens, completedAt)
1602 : undefined;1601 : undefined;
1603 const failure = turn.interrupted ? null : (turn.rejectedPromptFailure ?? turn.failure);1602 const failure = interrupted ? null : (turn.rejectedPromptFailure ?? turn.failure);
1604 yield* emit({1603 yield* emit({
1605 type: "provider_turn.updated",1604 type: "provider_turn.updated",
1606 driver: PI_PROVIDER,1605 driver: PI_PROVIDER,
...@@ -1610,7 +1609,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1610,7 +1609,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
1610 ...(treeRefs === null || treeRefs.nativeTurnRef === undefined1609 ...(treeRefs === null || treeRefs.nativeTurnRef === undefined
1611 ? {}1610 ? {}
1612 : { nativeTurnRef: treeRefs.nativeTurnRef }),1611 : { nativeTurnRef: treeRefs.nativeTurnRef }),
1613 status: turn.interrupted ? "interrupted" : failure !== null ? "failed" : "completed",1612 status: interrupted ? "interrupted" : failure !== null ? "failed" : "completed",
1614 completedAt,1613 completedAt,
1615 ...(tokenUsage === undefined ? {} : { tokenUsage }),1614 ...(tokenUsage === undefined ? {} : { tokenUsage }),
1616 },1615 },
...@@ -1679,7 +1678,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1679,7 +1678,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
1679 providerThreadId: state.providerThread.id,1678 providerThreadId: state.providerThread.id,
1680 providerTurnId: turn.providerTurn.id,1679 providerTurnId: turn.providerTurn.id,
1681 runOrdinal: turn.turnInput.runOrdinal,1680 runOrdinal: turn.turnInput.runOrdinal,
1682 status: turn.interrupted ? "interrupted" : "completed",1681 status: interrupted ? "interrupted" : "completed",
1683 failure: null,1682 failure: null,
1684 threadDisposition: "reusable",1683 threadDisposition: "reusable",
1685 });1684 });
...@@ -1799,6 +1798,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -1799,6 +1798,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
1799 return;1798 return;
1800 }1799 }
1801 turn.sawAgentActivity = true;1800 turn.sawAgentActivity = true;
1801 turn.nativeAborted = false;
1802 turn.settleProbeGeneration += 1;1802 turn.settleProbeGeneration += 1;
1803 return;1803 return;
1804 }1804 }
...@@ -2018,6 +2018,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -2018,6 +2018,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
2018 return;2018 return;
2019 }2019 }
2020 if (turn !== null) {2020 if (turn !== null) {
2021 turn.nativeAborted = event["aborted"] === true;
2021 turn.settleWhenIdle = true;2022 turn.settleWhenIdle = true;
2022 turn.settleProbeGeneration += 1;2023 turn.settleProbeGeneration += 1;
2023 // A wake can settle while the newly joined user prompt is still2024 // A wake can settle while the newly joined user prompt is still
...@@ -2610,6 +2611,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -2610,6 +2611,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
2610 toolArgs: new Map(),2611 toolArgs: new Map(),
2611 toolStartedAt: new Map(),2612 toolStartedAt: new Map(),
2612 interrupted: false,2613 interrupted: false,
2614 nativeAborted: false,
2613 sawAgentActivity: false,2615 sawAgentActivity: false,
2614 adoptedWake: false,2616 adoptedWake: false,
2615 promptMayBeCommandOnly:2617 promptMayBeCommandOnly:
...@@ -2682,6 +2684,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (...@@ -2682,6 +2684,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* (
2682 if (wake !== null) {2684 if (wake !== null) {
2683 for (const event of wake.events) {2685 for (const event of wake.events) {
2684 if (!continuation && event["type"] === "agent_settled") {2686 if (!continuation && event["type"] === "agent_settled") {
2687 activeTurn.nativeAborted = event["aborted"] === true;
2685 activeTurn.settleWhenIdle = true;2688 activeTurn.settleWhenIdle = true;
2686 continue;2689 continue;
2687 }2690 }