From 01c58cdc910d0af0426446a90e2d4e866d91e41a Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:19:41 +0100 Subject: [PATCH 1/2] fix(pi): retire uncertain model selections Co-authored-by: GPT 6.1 Sol Co-authored-by: Claude Opus 5.5 --- .../provider-pi/src/server/adapter.test.ts | 121 +++++++++++++++++- packages/provider-pi/src/server/adapter.ts | 102 ++++++++------- 2 files changed, 174 insertions(+), 49 deletions(-) diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index 67e7cc6aca8eb..b743c7d8d2abf 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -32,6 +32,7 @@ import * as Queue from "effect/Queue"; import * as Schema from "effect/Schema"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; @@ -167,7 +168,7 @@ interface FakePi { readonly failNextState: () => void; readonly failNextThinking: () => void; readonly deferNextLifecycle: (type: "switch_session" | "new_session" | "fork" | "abort") => void; - readonly deferNextModelSelection: () => void; + readonly deferNextModelSelection: (type?: "set_model" | "set_thinking_level") => void; readonly queueModels: (models: ReadonlyArray) => void; readonly vetoNextNewSession: () => void; readonly allRequests: () => ReadonlyArray; @@ -421,8 +422,8 @@ const makeFakePi: Effect.Effect = Effect.gen(function* () { deferNextLifecycle: (type) => { deferredLifecycle = type; }, - deferNextModelSelection: () => { - deferredLifecycle = "set_model"; + deferNextModelSelection: (type = "set_model") => { + deferredLifecycle = type; }, queueModels: (value) => { models = value; @@ -2128,6 +2129,120 @@ describe("PiAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(layerTest)), ); + it.effect.each([ + { operation: "set_model", restore: false, failure: "timeout" }, + { operation: "set_model", restore: false, failure: "interruption" }, + { operation: "set_thinking_level", restore: false, failure: "timeout" }, + { operation: "set_thinking_level", restore: false, failure: "interruption" }, + { operation: "set_model", restore: true, failure: "timeout" }, + { operation: "set_model", restore: true, failure: "interruption" }, + { operation: "set_thinking_level", restore: true, failure: "timeout" }, + { operation: "set_thinking_level", restore: true, failure: "interruption" }, + ] as const)( + "retires selection after $failure of $operation (baseline restore: $restore)", + ({ operation, restore, failure }) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + fake.queueState({ model: { provider: "anthropic", id: "baseline" } }); + const { runtime, takeEvent } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + if (restore) { + yield* startTurn(runtime, providerThread, "default", [], "First", { + ...modelSelection("anthropic/model-a"), + options: [{ id: "thinking", value: "low" }], + }); + yield* fake.takeRequest("set_model"); + yield* fake.takeRequest("set_thinking_level"); + yield* fake.takeRequest("prompt"); + yield* fake.emit({ type: "agent_start" }); + yield* fake.emit({ type: "agent_end", messages: [], willRetry: false }); + yield* fake.emit({ type: "agent_settled" }); + yield* takeEvent((event) => event.type === "turn.terminal"); + } + fake.deferNextModelSelection(operation); + const selection: ModelSelection = restore + ? modelSelection("default") + : { + ...modelSelection("anthropic/model-a"), + options: [{ id: "thinking", value: "low" }], + }; + const starting = yield* startTurn( + runtime, + providerThread, + "default", + [], + "Selection", + selection, + 2, + ).pipe(Effect.exit, Effect.forkChild); + const pending = yield* fake.takeRequest(operation); + if (failure === "timeout") { + yield* TestClock.adjust("15 seconds"); + const result = yield* Fiber.join(starting); + assert.equal(result._tag, "Failure"); + if (Exit.isFailure(result)) assert.match(Cause.pretty(result.cause), /PiRpcTimeoutError/); + } else { + yield* Fiber.interrupt(starting); + } + // A late acknowledgement cannot make the retired connection usable again. + yield* fake.emit({ type: "response", id: pending.id, command: operation, success: true }); + const retry = yield* startTurn(runtime, providerThread, "anthropic/model-a").pipe( + Effect.exit, + ); + assert.equal(retry._tag, "Failure"); + assert.deepEqual(fake.promptModels, restore ? ["anthropic/model-a"] : []); + yield* runtime.resumeThread({ providerThread }).pipe(Effect.flip); + + const freshFake = yield* makeFakePi; + const { runtime: freshRuntime } = yield* openRuntime(freshFake); + const freshThread = yield* freshRuntime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + yield* startTurn(freshRuntime, freshThread, "anthropic/model-b"); + yield* freshFake.takeRequest("prompt"); + assert.deepEqual(freshFake.promptModels, ["anthropic/model-b"]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + + it.effect.each(["Unknown model: anthropic/unknown", "No API key for anthropic"])( + "keeps the runtime usable after selection validation fails: %s", + (validationError) => + Effect.gen(function* () { + const fake = yield* makeFakePi; + const { runtime } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + fake.deferNextModelSelection(); + const starting = yield* startTurn(runtime, providerThread, "anthropic/unknown").pipe( + Effect.exit, + Effect.forkChild, + ); + const pending = yield* fake.takeRequest("set_model"); + yield* fake.emit({ + type: "response", + id: pending.id, + command: "set_model", + success: false, + error: validationError, + }); + const result = yield* Fiber.join(starting); + assert.equal(result._tag, "Failure"); + if (Exit.isFailure(result)) assert.include(Cause.pretty(result.cause), validationError); + yield* startTurn(runtime, providerThread, "anthropic/model-b"); + yield* fake.takeRequest("prompt"); + assert.deepEqual(fake.promptModels, ["anthropic/model-b"]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), + ); + it.effect("retires an interrupted switch before accepting further requests", () => 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 8ed02b6917fc0..126f6b74eb362 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -2542,16 +2542,49 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( return providerThread; }); - const applySelection = Effect.fnUntraced(function* (modelSelection: ModelSelection) { - const thinking = getModelSelectionStringOptionValue(modelSelection, "thinking"); - if (modelSelection.model === PI_INHERIT_MODEL_SLUG) { - if (appliedModel !== null && baselineModel !== null) { - const restoredModel = yield* request({ type: "set_model", ...baselineModel }); + const applySelection = Effect.fnUntraced( + function* (modelSelection: ModelSelection) { + const thinking = getModelSelectionStringOptionValue(modelSelection, "thinking"); + if (modelSelection.model === PI_INHERIT_MODEL_SLUG) { + if (appliedModel !== null && baselineModel !== null) { + const restoredModel = yield* request({ type: "set_model", ...baselineModel }); + appliedThinking = null; + contextWindow = rememberModelContextWindow(restoredModel); + appliedModel = null; + const updatedAt = yield* DateTime.now; + sessionEntity = { ...sessionEntity, model: PI_INHERIT_MODEL_SLUG, updatedAt }; + yield* emit({ + type: "provider_session.updated", + driver: PI_PROVIDER, + providerSession: sessionEntity, + }); + } + if ( + thinking === undefined && + baselineThinking !== null && + appliedThinking !== baselineThinking + ) { + yield* request({ type: "set_thinking_level", level: baselineThinking }); + appliedThinking = baselineThinking; + } + } else if (modelSelection.model !== appliedModel) { + const parsed = parsePiModelSlug(modelSelection.model); + if (parsed === null) { + return yield* protocolError( + `Pi model '${modelSelection.model}' must use provider/model format`, + ); + } + const selectedModel = yield* request({ + type: "set_model", + provider: parsed.provider, + modelId: parsed.modelId, + }); + // Pi resets thinking to the target model's configured default on set_model. appliedThinking = null; - contextWindow = rememberModelContextWindow(restoredModel); - appliedModel = null; + contextWindow = rememberModelContextWindow(selectedModel); + appliedModel = modelSelection.model; const updatedAt = yield* DateTime.now; - sessionEntity = { ...sessionEntity, model: PI_INHERIT_MODEL_SLUG, updatedAt }; + sessionEntity = { ...sessionEntity, model: modelSelection.model, updatedAt }; yield* emit({ type: "provider_session.updated", driver: PI_PROVIDER, @@ -2559,46 +2592,23 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( }); } if ( - thinking === undefined && - baselineThinking !== null && - appliedThinking !== baselineThinking + thinking !== undefined && + thinking !== appliedThinking && + PI_THINKING_LEVELS.has(thinking) ) { - yield* request({ type: "set_thinking_level", level: baselineThinking }); - appliedThinking = baselineThinking; - } - } else if (modelSelection.model !== appliedModel) { - const parsed = parsePiModelSlug(modelSelection.model); - if (parsed === null) { - return yield* protocolError( - `Pi model '${modelSelection.model}' must use provider/model format`, - ); + yield* request({ type: "set_thinking_level", level: thinking }); + appliedThinking = thinking; } - const selectedModel = yield* request({ - type: "set_model", - provider: parsed.provider, - modelId: parsed.modelId, - }); - // Pi resets thinking to the target model's configured default on set_model. - appliedThinking = null; - contextWindow = rememberModelContextWindow(selectedModel); - appliedModel = modelSelection.model; - const updatedAt = yield* DateTime.now; - sessionEntity = { ...sessionEntity, model: modelSelection.model, updatedAt }; - yield* emit({ - type: "provider_session.updated", - driver: PI_PROVIDER, - providerSession: sessionEntity, - }); - } - if ( - thinking !== undefined && - thinking !== appliedThinking && - PI_THINKING_LEVELS.has(thinking) - ) { - yield* request({ type: "set_thinking_level", level: thinking }); - appliedThinking = thinking; - } - }); + }, + (effect) => + effect.pipe( + // A timed-out or interrupted selection may still mutate Pi after the caller stops waiting. + Effect.tapError((error) => + error._tag === "PiRpcTimeoutError" ? connection.terminate : Effect.void, + ), + Effect.onInterrupt(() => connection.terminate), + ), + ); const resolvePromptPayload = Effect.fnUntraced(function* ( text: string, From a7ee6277f9b12c5db73558ab11a888f79fb1d5be Mon Sep 17 00:00:00 2001 From: Wout Stiens <71498452+StiensWout@users.noreply.github.com> Date: Sat, 7 Nov 2026 12:23:32 +0100 Subject: [PATCH 2/2] test(pi): trim selection retirement tests Drop the fresh-runtime tail that never touched the retired runtime and collapse the validation-error rows that exercised one code path. Co-authored-by: Claude Opus 5.5 Co-authored-by: GPT 6.1 Sol --- .../provider-pi/src/server/adapter.test.ts | 72 ++++++++----------- 1 file changed, 30 insertions(+), 42 deletions(-) diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index b743c7d8d2abf..94041f13b732d 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -2196,51 +2196,39 @@ describe("PiAdapterV2", () => { assert.equal(retry._tag, "Failure"); assert.deepEqual(fake.promptModels, restore ? ["anthropic/model-a"] : []); yield* runtime.resumeThread({ providerThread }).pipe(Effect.flip); - - const freshFake = yield* makeFakePi; - const { runtime: freshRuntime } = yield* openRuntime(freshFake); - const freshThread = yield* freshRuntime.ensureThread({ - threadId: THREAD_ID, - modelSelection: modelSelection("default"), - runtimePolicy, - }); - yield* startTurn(freshRuntime, freshThread, "anthropic/model-b"); - yield* freshFake.takeRequest("prompt"); - assert.deepEqual(freshFake.promptModels, ["anthropic/model-b"]); }).pipe(Effect.scoped, Effect.provide(layerTest)), ); - it.effect.each(["Unknown model: anthropic/unknown", "No API key for anthropic"])( - "keeps the runtime usable after selection validation fails: %s", - (validationError) => - Effect.gen(function* () { - const fake = yield* makeFakePi; - const { runtime } = yield* openRuntime(fake); - const providerThread = yield* runtime.ensureThread({ - threadId: THREAD_ID, - modelSelection: modelSelection("default"), - runtimePolicy, - }); - fake.deferNextModelSelection(); - const starting = yield* startTurn(runtime, providerThread, "anthropic/unknown").pipe( - Effect.exit, - Effect.forkChild, - ); - const pending = yield* fake.takeRequest("set_model"); - yield* fake.emit({ - type: "response", - id: pending.id, - command: "set_model", - success: false, - error: validationError, - }); - const result = yield* Fiber.join(starting); - assert.equal(result._tag, "Failure"); - if (Exit.isFailure(result)) assert.include(Cause.pretty(result.cause), validationError); - yield* startTurn(runtime, providerThread, "anthropic/model-b"); - yield* fake.takeRequest("prompt"); - assert.deepEqual(fake.promptModels, ["anthropic/model-b"]); - }).pipe(Effect.scoped, Effect.provide(layerTest)), + it.effect("keeps the runtime usable after Pi rejects a model selection", () => + Effect.gen(function* () { + const validationError = "Unknown model: anthropic/unknown"; + const fake = yield* makeFakePi; + const { runtime } = yield* openRuntime(fake); + const providerThread = yield* runtime.ensureThread({ + threadId: THREAD_ID, + modelSelection: modelSelection("default"), + runtimePolicy, + }); + fake.deferNextModelSelection(); + const starting = yield* startTurn(runtime, providerThread, "anthropic/unknown").pipe( + Effect.exit, + Effect.forkChild, + ); + const pending = yield* fake.takeRequest("set_model"); + yield* fake.emit({ + type: "response", + id: pending.id, + command: "set_model", + success: false, + error: validationError, + }); + const result = yield* Fiber.join(starting); + assert.equal(result._tag, "Failure"); + if (Exit.isFailure(result)) assert.include(Cause.pretty(result.cause), validationError); + yield* startTurn(runtime, providerThread, "anthropic/model-b"); + yield* fake.takeRequest("prompt"); + assert.deepEqual(fake.promptModels, ["anthropic/model-b"]); + }).pipe(Effect.scoped, Effect.provide(layerTest)), ); it.effect("retires an interrupted switch before accepting further requests", () =>