From 3e6b4fec3d1d6b7acb311f36c0ac88295809cc19 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 25 Oct 2026 21:59:37 +0000 Subject: [PATCH 1/2] fix(server): preserve answered approvals through provider acknowledgements Co-authored-by: Alex Southwell <4596216+saphid@users.noreply.github.com> --- .../MuseReplayFixtures.integration.test.ts | 271 +++++++++++++++++- ...OpenCodeReplayFixtures.integration.test.ts | 65 ++++- .../PiReplayFixtures.integration.test.ts | 102 ++++++- .../fixtures/muse_permission/output.ts | 8 + .../opencode_child_approval/output.ts | 2 + .../plan_questions/opencode_output.ts | 10 + .../provider-muse/src/server/adapter.test.ts | 260 ++++++++++++++--- packages/provider-muse/src/server/adapter.ts | 163 ++++++++--- .../provider-opencode/src/server/adapter.ts | 54 ++-- .../provider-pi/src/server/adapter.test.ts | 24 +- packages/provider-pi/src/server/adapter.ts | 28 +- 11 files changed, 854 insertions(+), 133 deletions(-) diff --git a/apps/server/src/orchestration-v2/testkit/MuseReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/MuseReplayFixtures.integration.test.ts index fee15966df4ed..f150135a814db 100644 --- a/apps/server/src/orchestration-v2/testkit/MuseReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/MuseReplayFixtures.integration.test.ts @@ -1,10 +1,16 @@ -import { describe } from "@effect/vitest"; +import { projectionFor } from "./fixtures/shared.ts"; +import type { ProviderReplayEntry } from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; import { materializeMuseReplayWorkspace, MuseOrchestratorReplayHarness, } from "../Adapters/MuseAdapterV2.testkit.ts"; -import { itRunsReplayFixtures, runReplayFixture } from "./ProviderReplayFixtures.ts"; +import { + findReplayFixture, + itRunsReplayFixtures, + runReplayFixture, +} from "./ProviderReplayFixtures.ts"; describe("Muse replay fixtures", () => { itRunsReplayFixtures("muse", (input) => @@ -14,4 +20,265 @@ describe("Muse replay fixtures", () => { materializeWorkspace: materializeMuseReplayWorkspace, }), ); + it.effect( + "keeps Muse answers and decisions across native acknowledgement and turn cleanup", + () => { + const fixture = findReplayFixture("simple"); + const provider = fixture?.providers.find((candidate) => candidate.driver === "muse"); + assert.isDefined(fixture); + assert.isDefined(provider); + const answers = { approach: ["Small", "Keep tests focused"] }; + return runReplayFixture({ + harness: MuseOrchestratorReplayHarness, + materializeWorkspace: materializeMuseReplayWorkspace, + fixtureName: fixture.name, + buildInput: () => ({ + ...fixture.buildInput(), + steps: [ + ...fixture.buildInput().steps, + { type: "answer_next_user_input_request", answers }, + { type: "answer_next_user_input_request", answers }, + { type: "approve_next_runtime_request", decision: "acceptForSession" }, + ], + }), + driver: { + ...provider, + assertOutput: (result) => { + const projection = projectionFor(result, fixture.name); + assert.equal(projection.runs.at(-1)?.status, "completed"); + assert.lengthOf(projection.runtimeRequests, 3); + const observed = []; + const expected = []; + for (const nativeId of ["question-ack", "question-no-ack", "approval-no-ack"]) { + const request = projection.runtimeRequests.find( + (candidate) => candidate.nativeRequestRef?.nativeId === nativeId, + ); + assert.isDefined(request); + const card = projection.turnItems.find((item) => item.nodeId === request.nodeId); + const node = projection.nodes.find((candidate) => candidate.id === request.nodeId); + const isQuestion = request.kind === "user_input"; + const response = isQuestion ? { answers } : { decision: "acceptForSession" as const }; + const submitted = result.domainEvents.find( + (event) => + event.type === "runtime-request.updated" && + event.payload.id === request.id && + event.payload.status === "resolved", + ); + assert.isDefined(submitted, `${nativeId} was persisted before native delivery`); + assert.deepInclude(submitted.payload, response); + if (!isQuestion) { + const assistantItemIndex = result.domainEvents.findIndex( + (event) => + event.type === "turn-item.updated" && + event.payload.type === "assistant_message", + ); + const submittedIndex = result.domainEvents.indexOf(submitted); + assert.isAbove(assistantItemIndex, submittedIndex); + // The next assistant item is persisted after stageResolved, before turn cleanup. + const beforeAssistant = result.domainEvents.slice(0, assistantItemIndex); + const requestAtUpdate = beforeAssistant.findLast( + (event) => + event.type === "runtime-request.updated" && event.payload.id === request.id, + ); + const nodeAtUpdate = beforeAssistant.findLast( + (event) => event.type === "node.updated" && event.payload.id === request.nodeId, + ); + const cardAtUpdate = beforeAssistant.findLast( + (event) => + event.type === "turn-item.updated" && event.payload.nodeId === request.nodeId, + ); + assert.deepInclude( + requestAtUpdate?.payload, + { status: "resolved", decision: "acceptForSession" }, + "stageResolved must preserve the persisted T3 decision before turn cleanup", + ); + assert.deepInclude(nodeAtUpdate?.payload, { status: "completed" }); + assert.deepInclude(cardAtUpdate?.payload, { status: "completed" }); + } + observed.push({ + nativeId, + status: request.status, + answers: request.answers, + decision: request.decision, + nodeStatus: node?.status, + cardStatus: card?.status, + questionAnswer: + card?.type === "user_input_request" ? card.questionAnswer : undefined, + }); + expected.push({ + nativeId, + status: "resolved", + answers: isQuestion ? answers : undefined, + decision: isQuestion ? undefined : "acceptForSession", + nodeStatus: "completed", + cardStatus: "completed", + questionAnswer: isQuestion + ? { + requestId: request.id, + answers, + attachmentsByQuestionId: {}, + questionTextById: { approach: "Which approach?" }, + } + : undefined, + } as const); + } + assert.deepEqual(observed, expected); + }, + }, + transformTranscript: (transcript) => { + const at = transcript.entries.findIndex( + (entry) => entry.type === "emit_inbound" && entry.label === "turn/started", + ); + const started = transcript.entries[at]; + assert.isDefined(started); + if (started.type !== "emit_inbound") assert.fail("Missing Muse turn/started"); + const { sessionId, turnId } = ( + started.frame as { + readonly params: { readonly sessionId: string; readonly turnId: string }; + } + ).params; + const metadata = transcript.metadata as { + readonly commandIds: ReadonlyArray>; + }; + return { + ...transcript, + metadata: { + ...transcript.metadata, + commandIds: [ + [...metadata.commandIds[0]!, "answer-ack", "answer-no-ack", "approve-no-ack"], + ], + }, + entries: [ + ...transcript.entries.slice(0, at + 1), + ...["question-ack", "question-no-ack"].flatMap( + (userInputId, index): ProviderReplayEntry[] => [ + { + type: "emit_inbound", + frame: { + jsonrpc: "2.0", + method: "userInput/requested", + params: { + sessionId, + turnId, + userInputId, + questions: [ + { + id: "approach", + header: "Approach", + question: "Which approach?", + options: [{ label: "Small" }, { label: "Large" }], + selection: { mode: "single" }, + }, + ], + }, + }, + }, + { + type: "expect_outbound", + frame: { + jsonrpc: "2.0", + id: 4 + index, + method: "userInput/answer", + params: { + sessionId, + userInputId, + commandId: index === 0 ? "answer-ack" : "answer-no-ack", + answers: [ + { + questionId: "approach", + selectedLabel: "Small", + note: "Keep tests focused", + }, + ], + }, + }, + }, + { type: "emit_inbound", frame: { jsonrpc: "2.0", id: 4 + index, result: {} } }, + ...(index === 0 + ? [ + { + type: "emit_inbound" as const, + frame: { + jsonrpc: "2.0", + method: "userInput/settled", + params: { sessionId, turnId, userInputId }, + }, + }, + ] + : []), + ], + ), + { + type: "emit_inbound", + frame: { + jsonrpc: "2.0", + method: "approval/requested", + params: { + sessionId, + turnId, + approvalId: "approval-no-ack", + currentRequirementId: { approvalId: "approval-no-ack", sourceIndex: 0 }, + availableChoices: [ + { + choiceId: "session", + label: "Allow for session", + decision: "approvedForSession", + scope: "session", + }, + ], + subject: { kind: "shell", command: "touch result.txt" }, + }, + }, + }, + { + type: "expect_outbound", + frame: { + jsonrpc: "2.0", + id: 6, + method: "approval/decide", + params: { + sessionId, + approvalId: "approval-no-ack", + commandId: "approve-no-ack", + requirementId: { approvalId: "approval-no-ack", sourceIndex: 0 }, + choiceId: "session", + }, + }, + }, + { type: "emit_inbound", frame: { jsonrpc: "2.0", id: 6, result: {} } }, + { + type: "emit_inbound", + frame: { + jsonrpc: "2.0", + method: "approval/updated", + params: { + sessionId, + turnId, + approvalId: "approval-no-ack", + currentRequirementId: { approvalId: "approval-no-ack", sourceIndex: 0 }, + availableChoices: [ + { + choiceId: "session", + label: "Allow for session", + decision: "approvedForSession", + scope: "session", + }, + ], + subject: { kind: "shell", command: "touch result.txt" }, + change: { + kind: "stageResolved", + requirementId: { approvalId: "approval-no-ack", sourceIndex: 0 }, + choiceId: "session", + decision: "approvedForSession", + }, + }, + }, + }, + ...transcript.entries.slice(at + 1), + ], + }; + }, + }); + }, + ); }); diff --git a/apps/server/src/orchestration-v2/testkit/OpenCodeReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/OpenCodeReplayFixtures.integration.test.ts index f9a594042b712..48c4f723d2a64 100644 --- a/apps/server/src/orchestration-v2/testkit/OpenCodeReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/OpenCodeReplayFixtures.integration.test.ts @@ -1,4 +1,5 @@ -import { describe, it } from "@effect/vitest"; +import { projectionFor } from "./fixtures/shared.ts"; +import { assert, describe, it } from "@effect/vitest"; import type { ProviderReplayEntry } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -59,6 +60,68 @@ describe("OpenCode replay fixtures", () => { }), }), ); + it.effect("keeps an OpenCode decline when the turn ends without acknowledgement", () => { + const fixture = findReplayFixture("opencode_child_approval"); + const provider = fixture?.providers.find((candidate) => candidate.driver === "opencode"); + assert.isDefined(fixture); + assert.isDefined(provider); + return runReplayFixture({ + harness: OpenCodeOrchestratorReplayHarness, + fixtureName: fixture.name, + buildInput: () => ({ + ...fixture.buildInput(), + steps: fixture + .buildInput() + .steps.map((step) => + step.type === "approve_next_runtime_request" ? { ...step, decision: "decline" } : step, + ), + }), + driver: { + ...provider, + assertOutput: (result) => { + const projection = projectionFor(result, fixture.name); + assert.lengthOf(projection.runtimeRequests, 1); + const request = projection.runtimeRequests[0]!; + assert.deepInclude(request, { status: "resolved", decision: "decline" }); + assert.equal( + projection.nodes.find((node) => node.id === request.nodeId)?.status, + "cancelled", + ); + assert.equal( + projection.turnItems.find((item) => item.nodeId === request.nodeId)?.status, + "cancelled", + ); + assert.equal(projection.runs.at(-1)?.status, "completed"); + }, + }, + transformTranscript: (transcript) => { + assert.isTrue( + transcript.entries.some( + (entry) => entry.type === "expect_outbound" && entry.label === "permission.reply", + ), + ); + assert.isTrue( + transcript.entries.some( + (entry) => entry.type === "emit_inbound" && entry.label === "child.permission.replied", + ), + ); + return { + ...transcript, + entries: transcript.entries + .filter( + (entry) => + entry.type !== "emit_inbound" || entry.label !== "child.permission.replied", + ) + .map((entry) => { + if (entry.type !== "expect_outbound" || entry.label !== "permission.reply") + return entry; + const frame = entry.frame as { readonly input: object }; + return { ...entry, frame: { ...frame, input: { ...frame.input, reply: "reject" } } }; + }), + }; + }, + }); + }); }); function undecodableEvent(entry: ProviderReplayEntry): ProviderReplayEntry { diff --git a/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts index b9db455d1cffd..eff829a450595 100644 --- a/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts @@ -1,10 +1,108 @@ -import { describe } from "@effect/vitest"; +import { projectionFor } from "./fixtures/shared.ts"; +import type { ProviderReplayEntry } from "@t3tools/contracts"; +import { assert, describe, it } from "@effect/vitest"; import { PiOrchestratorReplayHarness } from "../Adapters/PiAdapterV2.testkit.ts"; -import { itRunsReplayFixtures, runReplayFixture } from "./ProviderReplayFixtures.ts"; +import { + findReplayFixture, + itRunsReplayFixtures, + runReplayFixture, +} from "./ProviderReplayFixtures.ts"; describe("Pi replay fixtures", () => { itRunsReplayFixtures("pi", (input) => runReplayFixture({ ...input, harness: PiOrchestratorReplayHarness }), ); + it.effect("keeps Pi decisions and text answers after the provider turn ends", () => { + const fixture = findReplayFixture("simple"); + const provider = fixture?.providers.find((candidate) => candidate.driver === "pi"); + assert.isDefined(fixture); + assert.isDefined(provider); + const answers = { "ui-input": "Keep the existing tests" }; + return runReplayFixture({ + harness: PiOrchestratorReplayHarness, + fixtureName: fixture.name, + buildInput: () => ({ + ...fixture.buildInput(), + steps: [ + ...fixture.buildInput().steps, + { type: "approve_next_runtime_request", decision: "accept" }, + { type: "approve_next_runtime_request", decision: "decline" }, + { type: "answer_next_user_input_request", answers }, + ], + }), + driver: { + ...provider, + assertOutput: (result) => { + const projection = projectionFor(result, fixture.name); + assert.lengthOf(projection.runtimeRequests, 3); + for (const [nativeId, response, status] of [ + ["ui-accept", { decision: "accept" }, "completed"], + ["ui-decline", { decision: "decline" }, "cancelled"], + ["ui-input", { answers }, "completed"], + ] as const) { + const request = projection.runtimeRequests.find( + (candidate) => candidate.nativeRequestRef?.nativeId === nativeId, + ); + assert.isDefined(request); + assert.deepInclude(request, { status: "resolved", ...response }); + assert.equal( + projection.nodes.find((node) => node.id === request.nodeId)?.status, + status, + ); + assert.equal( + projection.turnItems.find((item) => item.nodeId === request.nodeId)?.status, + status, + ); + } + assert.equal(projection.runs.at(-1)?.status, "completed"); + }, + }, + transformTranscript: (transcript) => { + const at = transcript.entries.findIndex( + (entry) => entry.type === "emit_inbound" && entry.label === "turn_start", + ); + assert.isAtLeast(at, 0, "Pi fixture must contain turn_start"); + return { + ...transcript, + entries: [ + ...transcript.entries.slice(0, at + 1), + ...(["accept", "decline"] as const).flatMap((decision): ProviderReplayEntry[] => [ + { + type: "emit_inbound", + frame: { + type: "extension_ui_request", + id: `ui-${decision}`, + method: "confirm", + title: "Allow bash?", + }, + }, + { + type: "expect_outbound", + frame: { + type: "extension_ui_response", + id: `ui-${decision}`, + confirmed: decision === "accept", + }, + }, + ]), + { + type: "emit_inbound", + frame: { + type: "extension_ui_request", + id: "ui-input", + method: "input", + title: "How?", + }, + }, + { + type: "expect_outbound", + frame: { type: "extension_ui_response", id: "ui-input", value: answers["ui-input"] }, + }, + ...transcript.entries.slice(at + 1), + ], + }; + }, + }); + }); }); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/muse_permission/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/muse_permission/output.ts index ac3d52d293c5a..349b4967b4149 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/muse_permission/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/muse_permission/output.ts @@ -38,6 +38,14 @@ export function assertMusePermissionOutput( ["command", "resolved", "decline"], ], ); + for (const request of requests) { + const status = request.decision === "decline" ? "cancelled" : "completed"; + assert.equal(projection.nodes.find((node) => node.id === request.nodeId)?.status, status); + assert.equal( + projection.turnItems.find((item) => item.nodeId === request.nodeId)?.status, + status, + ); + } const approval = projection.turnItems.find((item) => item.type === "approval_request"); const options = approval?.type === "approval_request" ? (approval.options ?? []) : []; assert.includeDeepMembers( diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/opencode_child_approval/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/opencode_child_approval/output.ts index 59f70a52453f6..5e9bca90ebf13 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/opencode_child_approval/output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/opencode_child_approval/output.ts @@ -33,6 +33,8 @@ export function assertOpenCodeChildApprovalOutput( assert.lengthOf(requests, 1); assert.equal(requests[0]?.kind, "command"); assert.equal(requests[0]?.status, "resolved"); + assert.equal(requests[0]?.decision, "accept"); + assert.equal(projection.nodes.find((node) => node.id === approval.nodeId)?.status, "completed"); assert.lengthOf(projection.subagents, 1); assert.equal(projection.subagents[0]?.status, "completed"); diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/plan_questions/opencode_output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/plan_questions/opencode_output.ts index e8b72ff5b99e2..a9a994683bba1 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/plan_questions/opencode_output.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/plan_questions/opencode_output.ts @@ -4,6 +4,7 @@ import type { ProviderReplayTranscript } from "@t3tools/contracts"; import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts"; import { assertPlanQuestionsOutputBase } from "./codex_output.ts"; import { projectionFor } from "../shared.ts"; +import { planQuestionsInput } from "./input.ts"; export function assertOpenCodePlanQuestionsOutput( result: OrchestratorV2ScenarioResult, @@ -14,6 +15,15 @@ export function assertOpenCodePlanQuestionsOutput( const requestItem = projection.turnItems.find((item) => item.type === "user_input_request"); assert.equal(requestItem?.questions[0]?.id, "question-0-schema-vs-flexibility"); assert.equal(requestItem?.status, "completed"); + const answer = planQuestionsInput().steps.find( + (step) => step.type === "answer_next_user_input_request", + ); + assert.isDefined(answer); + assert.deepEqual(projection.runtimeRequests[0]?.answers, answer.answers); + assert.equal( + projection.nodes.find((node) => node.id === requestItem?.nodeId)?.status, + "completed", + ); assert.lengthOf( projection.turnItems.filter( (item) => item.type === "dynamic_tool" && item.toolName === "question", diff --git a/packages/provider-muse/src/server/adapter.test.ts b/packages/provider-muse/src/server/adapter.test.ts index 7950ef28cb93a..b5ca681aa3a62 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -985,7 +985,7 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); - it.effect("uses the latest native approval requirement and waits for its resolution event", () => + it.effect("uses the latest native approval requirement and cleans up on native resolution", () => Effect.gen(function* () { const fake = yield* makeFakeMuse(); const harness = yield* makeHarness(fake); @@ -1021,11 +1021,16 @@ describe("MuseAdapterV2", () => { ), ); yield* fake.emit("approval/resolved", { approvalId: "approval-1", turnId: nativeId }); - const resolved = yield* harness.takeEvent( - "runtime_request.updated", - (event) => event.runtimeRequest.status === "resolved", + yield* harness.takeEvent( + "provider_session.updated", + (event) => event.providerSession.status === "running", + ); + assert.isFalse( + harness.allEvents.some( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "resolved", + ), ); - assert.strictEqual(resolved.runtimeRequest.id, first.runtimeRequest.id); const repeated = yield* harness.runtime .respondToRuntimeRequest({ requestId: first.runtimeRequest.id, decision: "accept" }) .pipe(Effect.exit); @@ -1033,35 +1038,190 @@ describe("MuseAdapterV2", () => { }).pipe(Effect.scoped, Effect.provide(testLayer)), ); - it.effect("keeps the user's decision when Muse settles before acknowledging it", () => - Effect.gen(function* () { - const fake = yield* makeFakeMuse(); - const harness = yield* makeHarness(fake); - const { nativeId } = yield* startConversation(harness, fake); - yield* fake.emit("approval/requested", approval(nativeId)); - const pending = yield* harness.takeEvent( - "runtime_request.updated", - (event) => event.runtimeRequest.status === "pending", - ); - const acknowledgement = pendingPromise>(); - fake.queueResponse("approval/decide", acknowledgement.promise); - const responding = yield* harness.runtime - .respondToRuntimeRequest({ requestId: pending.runtimeRequest.id, decision: "decline" }) - .pipe(Effect.forkScoped); - yield* fake.takeCall("approval/decide"); - yield* fake.emit("approval/resolved", { - approvalId: "approval-1", - turnId: nativeId, - decision: "abort", - }); - const resolved = yield* harness.takeEvent( - "runtime_request.updated", - (event) => event.runtimeRequest.status === "resolved", - ); - assert.strictEqual(resolved.runtimeRequest.decision, "decline"); - acknowledgement.resolve({}); - yield* Fiber.join(responding); - }).pipe(Effect.scoped, Effect.provide(testLayer)), + it.effect( + "does not replace the submitted decision when Muse settles before acknowledging it", + () => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const harness = yield* makeHarness(fake); + const { nativeId } = yield* startConversation(harness, fake); + yield* fake.emit("approval/requested", approval(nativeId)); + const pending = yield* harness.takeEvent( + "runtime_request.updated", + (event) => event.runtimeRequest.status === "pending", + ); + const acknowledgement = pendingPromise>(); + fake.queueResponse("approval/decide", acknowledgement.promise); + const responding = yield* harness.runtime + .respondToRuntimeRequest({ requestId: pending.runtimeRequest.id, decision: "decline" }) + .pipe(Effect.forkScoped); + yield* fake.takeCall("approval/decide"); + yield* fake.emit("approval/resolved", { + approvalId: "approval-1", + turnId: nativeId, + decision: "abort", + }); + yield* harness.takeEvent( + "provider_session.updated", + (event) => event.providerSession.status === "running", + ); + acknowledgement.resolve({}); + yield* Fiber.join(responding); + assert.isFalse( + harness.allEvents.some( + (event) => + event.type === "runtime_request.updated" && + event.runtimeRequest.status === "resolved", + ), + ); + }).pipe(Effect.scoped, Effect.provide(testLayer)), + ); + + it.effect.each([true, false])( + "only opens the next approval stage when the in-flight decision permits it, terminal=%s", + (terminal) => + Effect.gen(function* () { + const fake = yield* makeFakeMuse(); + const harness = yield* makeHarness(fake); + const { nativeId } = yield* startConversation(harness, fake); + const firstStage = { + ...approval(nativeId), + subject: { kind: "shell", command: "touch first.txt && touch second.txt" }, + availableChoices: [ + ...approval(nativeId).availableChoices, + { + choiceId: "session", + label: "Allow for session", + decision: "approvedForSession", + scope: "session", + }, + ], + }; + const milestone = Effect.fnUntraced(function* (text: string) { + yield* fake.emit("item/completed", { + turnId: nativeId, + item: { + itemId: text, + turnId: nativeId, + kind: "agentMessage", + revision: 1, + status: "completed", + text, + }, + }); + yield* harness.takeEvent("message.updated", (event) => event.message.text === text); + }); + const readSnapshot = () => + harness.runtime.readThreadSnapshot!({ providerThread: harness.providerThread }); + yield* fake.emit("approval/requested", firstStage); + const first = (yield* harness.takeEvent("runtime_request.updated")).runtimeRequest; + yield* harness.takeEvent( + "provider_session.updated", + (event) => event.providerSession.status === "waiting", + ); + const afterFirstCard = harness.allEvents.length; + const acknowledgement = pendingPromise>(); + fake.queueResponse("approval/decide", acknowledgement.promise); + const responding = yield* harness.runtime + .respondToRuntimeRequest({ requestId: first.id, decision: "acceptForSession" }) + .pipe(Effect.forkScoped); + const firstDecision = yield* fake.takeCall("approval/decide"); + assert.deepStrictEqual(firstDecision.params.requirementId, firstStage.currentRequirementId); + assert.strictEqual(firstDecision.params.choiceId, "session"); + assert.containSubset((yield* readSnapshot()).runtimeRequests, [ + { id: first.id, status: "resolved", decision: "acceptForSession" }, + ]); + + const nextStage = { + ...firstStage, + currentRequirementId: { approvalId: "approval-1", sourceIndex: 1 }, + change: { kind: "stageResolved" }, + }; + yield* fake.emit("approval/updated", nextStage); + // Ingestion must progress while the decide reply is held, without locking notifications. + yield* milestone("update ingested before decide acknowledgement"); + acknowledgement.resolve({ + approvalId: "approval-1", + commandId: "decision-1", + status: "accepted", + terminal, + }); + yield* Fiber.join(responding); + yield* milestone("decide acknowledgement handled"); + const afterAcknowledgement = yield* readSnapshot(); + const laterEvents = harness.allEvents.slice(afterFirstCard); + const nextRequests = laterEvents.filter( + (event) => event.type === "runtime_request.updated", + ); + if (terminal) { + assert.lengthOf(nextRequests, 0, "A terminal decision must not create a phantom stage"); + assert.isFalse( + laterEvents.some( + (event) => + (event.type === "node.updated" && event.node.kind === "approval_request") || + (event.type === "turn_item.updated" && + event.turnItem.type === "approval_request") || + (event.type === "provider_session.updated" && + event.providerSession.status === "waiting"), + ), + ); + assert.isFalse(afterAcknowledgement.runtimeRequests.some((r) => r.status === "pending")); + } else { + assert.lengthOf(nextRequests, 1); + const next = nextRequests[0]!.runtimeRequest; + assert.notStrictEqual(next.id, first.id); + assert.notStrictEqual(next.nodeId, first.nodeId); + assert.strictEqual(next.status, "pending"); + assert.deepStrictEqual(afterAcknowledgement.runtimeRequests, [next]); + const repeated = yield* harness.runtime + .respondToRuntimeRequest({ requestId: first.id, decision: "acceptForSession" }) + .pipe(Effect.exit); + assert.strictEqual(repeated._tag, "Failure"); + assert.lengthOf( + fake.calls.filter((call) => call.method === "approval/decide"), + 1, + ); + fake.queueResponse("approval/decide", { + approvalId: "approval-1", + commandId: "decision-2", + status: "accepted", + terminal: true, + }); + yield* harness.runtime.respondToRuntimeRequest({ + requestId: next.id, + decision: "decline", + }); + const nextDecision = yield* fake.takeCall("approval/decide"); + assert.deepStrictEqual(nextDecision.params.requirementId, nextStage.currentRequirementId); + assert.strictEqual(nextDecision.params.choiceId, "deny"); + assert.containSubset((yield* readSnapshot()).runtimeRequests, [ + { id: next.id, status: "resolved", decision: "decline" }, + ]); + } + yield* fake.emit("approval/resolved", { + approvalId: "approval-1", + turnId: nativeId, + decision: "approved", + }); + yield* harness.takeEvent( + "provider_session.updated", + (event) => event.providerSession.status === "running", + ); + assert.isEmpty((yield* readSnapshot()).runtimeRequests); + assert.isFalse( + harness.allEvents + .slice(afterFirstCard) + .some( + (event) => + (event.type === "runtime_request.updated" && + (event.runtimeRequest.id === first.id || + event.runtimeRequest.status !== "pending")) || + (event.type === "node.updated" && event.node.id === first.nodeId) || + (event.type === "turn_item.updated" && event.turnItem.nodeId === first.nodeId), + ), + "Native updates must not replace either T3-owned answer or the original node/card", + ); + }).pipe(Effect.scoped, Effect.provide(testLayer)), ); it.effect("lets a user turn take a held Muse report turn, approvals included", () => @@ -1181,11 +1341,6 @@ describe("MuseAdapterV2", () => { subagentOrigin: child.subagentOrigin, currentRequirementId: { approvalId: "approval-1", sourceIndex: 1 }, }); - const second = yield* fake.takeCall("approval/decide"); - assert.deepStrictEqual(second.params.requirementId, { - approvalId: "approval-1", - sourceIndex: 1, - }); outdated.reject( new MspError({ code: -32053, @@ -1193,6 +1348,11 @@ describe("MuseAdapterV2", () => { data: { kind: "approvalRequirementStale" }, }), ); + const second = yield* fake.takeCall("approval/decide"); + assert.deepStrictEqual(second.params.requirementId, { + approvalId: "approval-1", + sourceIndex: 1, + }); current.reject( new MspError({ code: -32052, @@ -1220,8 +1380,16 @@ describe("MuseAdapterV2", () => { assert.strictEqual(decline.params.choiceId, "deny"); assert.deepStrictEqual(decline.params.requirementId, second.params.requirementId); yield* fake.emit("approval/resolved", child); - const resolved = yield* harness.takeEvent("runtime_request.updated"); - assert.strictEqual(resolved.runtimeRequest.decision, "decline"); + yield* harness.takeEvent( + "provider_session.updated", + (event) => event.providerSession.status === "running", + ); + assert.isFalse( + harness.allEvents.some( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "resolved", + ), + ); }).pipe(Effect.scoped, Effect.provide(testLayer)), ); @@ -1344,8 +1512,14 @@ describe("MuseAdapterV2", () => { answers: response.params.answers, }); yield* harness.takeEvent( - "runtime_request.updated", - (event) => event.runtimeRequest.status === "resolved", + "provider_session.updated", + (event) => event.providerSession.status === "running", + ); + assert.isFalse( + harness.allEvents.some( + (event) => + event.type === "runtime_request.updated" && event.runtimeRequest.status === "resolved", + ), ); }).pipe(Effect.scoped, Effect.provide(testLayer)), ); diff --git a/packages/provider-muse/src/server/adapter.ts b/packages/provider-muse/src/server/adapter.ts index cee5218b877f0..130f54dbe615b 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -222,13 +222,19 @@ interface ActiveTurn { } interface PendingRequest { - readonly native: - | { type: "approval"; value: MuseApproval } - | { type: "question"; value: MuseUserInput }; + native: { type: "approval"; value: MuseApproval } | { type: "question"; value: MuseUserInput }; request: OrchestrationV2RuntimeRequest; node: OrchestrationV2ExecutionNode; item: OrchestrationV2TurnItem; - response?: Pick; + response?: { + readonly owner: "orchestrator" | "adapter"; + readonly value: Pick; + readonly approval?: { + readonly requirementId: MuseApproval["currentRequirementId"]; + inFlight: boolean; + terminal: boolean; + }; + }; shown: boolean; } @@ -249,6 +255,7 @@ const nativeRef = (nativeId: string) => ({ strength: "strong" as const, }); const recordSchema = Schema.Record(Schema.String, Schema.Unknown); +const isAlreadyTerminal = Schema.is(Schema.Struct({ kind: Schema.Literal("alreadyTerminal") })); const INFORMATIONAL_NOTIFICATIONS = new Set([ "session/contextUsage", "session/tokenUsage", @@ -902,26 +909,33 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( entry: PendingRequest, status: "resolved" | "cancelled", ) { + pending.delete(entry.request.id); + // The orchestrator already persisted its response, node and card before delivery. + if (entry.response?.owner === "orchestrator") return; const time = yield* DateTime.now; + const decision = entry.response?.value.decision; + const settledStatus = + status === "resolved" && decision !== "decline" && decision !== "cancel" + ? "completed" + : "cancelled"; entry.request = { ...entry.request, - ...(status === "resolved" ? entry.response : undefined), + ...(status === "resolved" ? entry.response?.value : undefined), status, resolvedAt: time, responseCapability: { type: "not_resumable", reason: "This Muse request has ended." }, }; entry.node = { ...entry.node, - status: status === "resolved" ? "completed" : "cancelled", + status: settledStatus, completedAt: time, }; entry.item = { ...entry.item, - status: status === "resolved" ? "completed" : "cancelled", + status: settledStatus, completedAt: time, updatedAt: time, }; - pending.delete(entry.request.id); yield* emit({ type: "runtime_request.updated", driver: MUSE_PROVIDER, @@ -1060,29 +1074,62 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const decideApproval = Effect.fnUntraced(function* ( entry: PendingRequest, decision: ProviderApprovalDecision | undefined, + owner: "orchestrator" | "adapter", ) { if (entry.native.type !== "approval") return; const approval = entry.native.value; const choice = decision && museApprovalChoices(approval).get(decision); if (!choice) return yield* protocolError("Muse did not offer this approval decision"); // Recorded first: Muse can settle the approval before it acknowledges the command. - entry.response = { decision }; + const roundTrip = { + requirementId: entry.response?.approval?.requirementId ?? approval.currentRequirementId, + inFlight: true, + terminal: entry.response?.approval?.terminal ?? false, + }; + const response = { + owner, + value: { decision, resolvedAt: yield* DateTime.now }, + approval: roundTrip, + }; + entry.response = response; yield* request("approval/decide", { approvalId: approval.approvalId, - requirementId: approval.currentRequirementId, + requirementId: roundTrip.requirementId, choiceId: choice.choiceId, - }).pipe(Effect.tapError(() => Effect.sync(() => delete entry.response))); + }).pipe( + Effect.onExit((exit) => + Effect.gen(function* () { + if (pending.get(entry.request.id) !== entry || entry.response !== response) return; + roundTrip.inFlight = false; + roundTrip.terminal ||= exit._tag === "Success" && exit.value.terminal === true; + if (exit._tag === "Failure" && owner === "adapter" && !roundTrip.terminal) { + delete entry.response; + if ( + entry.native.type === "approval" && + (entry.native.value.currentRequirementId.approvalId !== + roundTrip.requirementId.approvalId || + entry.native.value.currentRequirementId.sourceIndex !== + roundTrip.requirementId.sourceIndex) + ) + yield* publishRequest(entry.native); + return; + } + // Notifications keep flowing during the RPC; release an absorbed stage only now. + if (!roundTrip.terminal) yield* publishRequest(entry.native); + }).pipe(eventPermit.withPermits(1)), + ), + ); }); const showRequest = Effect.fnUntraced(function* (entry: PendingRequest) { entry.shown = true; + yield* emit({ type: "node.updated", driver: MUSE_PROVIDER, node: entry.node }); + yield* emit({ type: "turn_item.updated", driver: MUSE_PROVIDER, turnItem: entry.item }); yield* emit({ type: "runtime_request.updated", driver: MUSE_PROVIDER, threadId: entry.node.threadId, runtimeRequest: entry.request, }); - yield* emit({ type: "node.updated", driver: MUSE_PROVIDER, node: entry.node }); - yield* emit({ type: "turn_item.updated", driver: MUSE_PROVIDER, turnItem: entry.item }); yield* updateSession("waiting"); }); /** @@ -1090,7 +1137,9 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( * may end long before the child asks, and T3 keeps a runless request on the * thread until Muse settles it. */ - const publishRequest = Effect.fnUntraced(function* (native: PendingRequest["native"]) { + const publishRequest = Effect.fnUntraced(function* ( + native: PendingRequest["native"], + ): Effect.fn.Return { const child = native.type === "approval" && native.value.subagentOrigin !== undefined; const turn = child ? undefined : active; if ( @@ -1103,11 +1152,34 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( const nativeId = native.type === "approval" ? native.value.approvalId : native.value.userInputId; if (turn?.settledRequests.has(`${native.type}:${nativeId}`)) return; - const previous = [...pending.values()].find( + let previous = [...pending.values()].find( (entry) => entry.native.type === native.type && entry.request.nativeRequestRef?.nativeId === nativeId, ); + if (previous?.response) { + const answeredRequirement = + previous.response.approval?.requirementId ?? + (previous.native.type === "approval" + ? previous.native.value.currentRequirementId + : undefined); + if ( + answeredRequirement && + native.type === "approval" && + !previous.response.approval?.inFlight && + !previous.response.approval?.terminal && + (answeredRequirement.approvalId !== native.value.currentRequirementId.approvalId || + answeredRequirement.sourceIndex !== native.value.currentRequirementId.sourceIndex) + ) { + // A new requirement needs its own answer without replacing the recorded one. + yield* resolvePending(previous, "resolved"); + previous = undefined; + } else { + // A terminal ack can close even the new frontier named by a trailing update. + previous.native = native; + return; + } + } const providerTurnId = turn?.providerTurn.id ?? null; const threadId = turn?.input.threadId ?? thread.appThreadId ?? input.threadId; const requestId = @@ -1227,11 +1299,22 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( pending.set(requestId, entry); if (entry.shown) yield* showRequest(entry); if (autoApprove) - yield* decideApproval(entry, "accept").pipe( - Effect.catch(() => + yield* decideApproval(entry, "accept", "adapter").pipe( + Effect.catch((error) => Effect.gen(function* () { - // Native updates may settle or replace this entry while the RPC is in flight. - if (!closed && !broken && pending.get(requestId) === entry) + // A stale requirement is replaced by a native update, not a user answer. + if ( + error.payload instanceof MspError && + error.payload.kind === "approvalRequirementStale" + ) + return; + if ( + !closed && + !broken && + pending.get(requestId) === entry && + !entry.shown && + !entry.response?.approval?.terminal + ) yield* showRequest(entry); }).pipe(eventPermit.withPermits(1)), ), @@ -1245,20 +1328,22 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ) { switch (method) { case "approval/requested": - case "approval/updated": - if ( - method === "approval/updated" && - ![...pending.values()].some( - (entry) => - entry.native.type === "approval" && - entry.native.value.approvalId === params.approvalId, - ) - ) + case "approval/updated": { + const previous = [...pending.values()].find( + (entry) => + entry.native.type === "approval" && + entry.native.value.approvalId === params.approvalId, + ); + if (method === "approval/updated" && isAlreadyTerminal(params.change)) { + if (previous?.response?.approval) previous.response.approval.terminal = true; return; + } + if (method === "approval/updated" && !previous) return; return yield* publishRequest({ type: "approval", value: yield* decode(MuseApproval, params), }); + } case "userInput/requested": return yield* publishRequest({ type: "question", @@ -1279,7 +1364,11 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( method === "approval/resolved" && typeof params.decision === "string" ? museApprovalDecision({ decision: params.decision, scope: "once" }) : undefined; - if (!entry.response && native) entry.response = { decision: native }; + if (!entry.response && native) + entry.response = { + owner: "adapter", + value: { decision: native, resolvedAt: yield* DateTime.now }, + }; yield* resolvePending(entry, "resolved"); } if (![...pending.values()].some((candidate) => candidate.shown)) @@ -2134,7 +2223,11 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ), runtimeRequests: [...pending.values()] .filter((entry) => entry.shown) - .map((entry) => entry.request), + .map((entry) => + entry.response?.owner === "orchestrator" + ? { ...entry.request, ...entry.response.value, status: "resolved" as const } + : entry.request, + ), providerPayload: items, }; }); @@ -2282,7 +2375,8 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( function* (args) { const entry = pending.get(args.requestId); if (!entry) return yield* protocolError("This Muse request is no longer pending"); - if (entry.native.type === "approval") yield* decideApproval(entry, args.decision); + if (entry.native.type === "approval") + yield* decideApproval(entry, args.decision, "orchestrator"); else { const questionRequest = entry.native.value; const answers = []; @@ -2314,11 +2408,14 @@ export const makeMuseAdapterV2 = Effect.fn("makeMuseAdapterV2")(function* ( ...(freeText ? (selected.length ? { note: freeText } : { freeText }) : {}), }); } - if (args.answers) entry.response = { answers: args.answers }; + entry.response = { + owner: "orchestrator", + value: { answers: args.answers, resolvedAt: yield* DateTime.now }, + }; yield* request("userInput/answer", { userInputId: questionRequest.userInputId, answers, - }).pipe(Effect.tapError(() => Effect.sync(() => delete entry.response))); + }); } }, (effect, args) => diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index d872b723df793..b09e675bb43a3 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -411,6 +411,8 @@ interface PendingOpenCodeRequest { readonly createdAt: DateTime.Utc; readonly permission?: PermissionRequest; readonly question?: QuestionRequest; + // Retained across delivery failures so late acknowledgements preserve T3's answer. + response?: Pick; } export interface OpenCodeAdapterV2Options { @@ -1862,19 +1864,29 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }, }); yield* emitProviderEvent({ - type: "runtime_request.updated", + type: "turn_item.updated", driver: OPENCODE_PROVIDER, - threadId: turn.threadId, - runtimeRequest, + turnItem: runtimeRequestTurnItem(pending, "waiting", null, now), }); yield* emitProviderEvent({ - type: "turn_item.updated", + type: "runtime_request.updated", driver: OPENCODE_PROVIDER, - turnItem: runtimeRequestTurnItem(pending, "waiting", null, now), + threadId: turn.threadId, + runtimeRequest, }); yield* updateProviderSession("waiting", null); }); + const updateRequestWaitingStatus = () => + updateProviderSession( + Array.from(pendingRequests.values()).some( + (pending) => pending.turn.isRoot && pending.response === undefined, + ) + ? "waiting" + : "running", + null, + ); + const resolveRuntimeRequest = Effect.fnUntraced(function* ( nativeRequestId: string, status: "resolved" | "cancelled", @@ -1886,17 +1898,24 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function if (current !== undefined) { const resolved: OrchestrationV2RuntimeRequest = { ...current, - status, + ...pending.response, + status: pending.response === undefined ? status : "resolved", resolvedAt: now, }; pending.state.runtimeRequests.set(String(pending.requestId), resolved); - yield* emitProviderEvent({ - type: "runtime_request.updated", - driver: OPENCODE_PROVIDER, - threadId: pending.turn.threadId, - runtimeRequest: resolved, - }); + if (pending.response === undefined) { + yield* emitProviderEvent({ + type: "runtime_request.updated", + driver: OPENCODE_PROVIDER, + threadId: pending.turn.threadId, + runtimeRequest: resolved, + }); + } } + pendingRequests.delete(String(pending.requestId)); + pendingRequestsByNativeId.delete(nativeRequestId); + // T3 persisted its response, card and node before delivery; native settlement is only cleanup. + if (pending.response !== undefined) return yield* updateRequestWaitingStatus(); yield* emitProviderEvent({ type: "node.updated", driver: OPENCODE_PROVIDER, @@ -1928,12 +1947,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function now, ), }); - pendingRequests.delete(String(pending.requestId)); - pendingRequestsByNativeId.delete(nativeRequestId); - const hasOtherPending = Array.from(pendingRequests.values()).some( - (candidate) => candidate.turn.isRoot, - ); - if (!hasOtherPending) yield* updateProviderSession("running", null); + yield* updateRequestWaitingStatus(); }); const resolveSessionOwner = Effect.fnUntraced(function* (sessionId: string) { @@ -3446,6 +3460,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function pending.question, requestInput.answers, ); + pending.response = { answers: requestInput.answers }; + yield* updateRequestWaitingStatus(); yield* sdkCall( "question.reply", { requestID: pending.nativeRequestId, answers }, @@ -3466,6 +3482,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function ); } const reply = OpenCodeRuntime.toOpenCodePermissionReply(requestInput.decision); + pending.response = { decision: requestInput.decision }; + yield* updateRequestWaitingStatus(); yield* sdkCall( "permission.reply", { requestID: pending.nativeRequestId, reply }, diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index a8a664d8b1cb6..416deeb95994d 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -1016,13 +1016,23 @@ describe("PiAdapterV2", () => { const prompt = yield* fake.takeRequest("prompt"); assert.equal(prompt["message"], "Continue the review"); assert.equal(prompt["streamingBehavior"], "steer"); - const resolved = yield* takeEvent( - (event) => - event.type === "runtime_request.updated" && event.runtimeRequest.id === dialog!.id, - ); - assert.isTrue( - resolved.type === "runtime_request.updated" && - resolved.runtimeRequest.status === "resolved", + yield* fake.emit({ + type: "extension_ui_request", + id: "next-dialog", + method: "confirm", + title: "Continue?", + }); + yield* takeEvent((event) => { + if (event.type !== "runtime_request.updated") return false; + requests.push(event); + return event.runtimeRequest.nativeRequestRef?.nativeId === "next-dialog"; + }); + assert.deepEqual( + requests + .filter((event) => event.runtimeRequest.id === dialog!.id) + .map((event) => event.runtimeRequest.status), + ["pending"], + "answering the model dialog must not replace the orchestrator's record", ); }).pipe(Effect.scoped, Effect.provide(layerTest)), ); diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index dc71c6dd4413d..3a45b0132b9c2 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -2699,33 +2699,7 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( if (pending.method === "confirm" && requestInput.decision === "acceptForSession") { sessionApprovals.add(pending.approvalKey); } - const resolvedAt = yield* DateTime.now; - pending.runtimeRequest = { - ...pending.runtimeRequest, - status: "resolved", - resolvedAt, - }; - yield* emit({ - type: "runtime_request.updated", - driver: PI_PROVIDER, - threadId: pending.node.threadId, - runtimeRequest: pending.runtimeRequest, - }); - yield* emit({ - type: "node.updated", - driver: PI_PROVIDER, - node: { ...pending.node, status: "completed", completedAt: resolvedAt }, - }); - yield* emit({ - type: "turn_item.updated", - driver: PI_PROVIDER, - turnItem: { - ...pending.turnItem, - status: "completed", - completedAt: resolvedAt, - updatedAt: resolvedAt, - }, - }); + // T3 already persisted the answer and card; cached updates would overwrite them. }, (effect, requestInput) => effect.pipe( From e44aa96ece6d53fb950350dc70398706ca6be497 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 01:34:59 +0000 Subject: [PATCH 2/2] fix(pi): publish approval cards before actionable requests Co-authored-by: Alex Southwell <4596216+saphid@users.noreply.github.com> --- .../provider-pi/src/server/adapter.test.ts | 28 +++++++++++++++++-- packages/provider-pi/src/server/adapter.ts | 4 +-- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/packages/provider-pi/src/server/adapter.test.ts b/packages/provider-pi/src/server/adapter.test.ts index 416deeb95994d..b46e25657834d 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -3136,9 +3136,31 @@ describe("PiAdapterV2", () => { title: "Run project extensions?", message: "This project has .pi/extensions.", }); - const pending = yield* takeEvent( - (event) => - event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending", + const artifacts: ProviderAdapter.ProviderAdapterV2Event[] = []; + const pending = yield* takeEvent((event) => { + artifacts.push(event); + return ( + event.type === "runtime_request.updated" && event.runtimeRequest.status === "pending" + ); + }); + if (pending.type !== "runtime_request.updated") return; + assert.isTrue( + artifacts.some( + (event) => + event.type === "node.updated" && + event.node.id === pending.runtimeRequest.nodeId && + event.node.status === "waiting", + ), + "the node must exist before an immediately answerable request", + ); + assert.isTrue( + artifacts.some( + (event) => + event.type === "turn_item.updated" && + event.turnItem.nodeId === pending.runtimeRequest.nodeId && + event.turnItem.status === "waiting", + ), + "the card must exist before an immediately answerable request", ); const requestId = pending.type === "runtime_request.updated" ? pending.runtimeRequest.id : undefined; diff --git a/packages/provider-pi/src/server/adapter.ts b/packages/provider-pi/src/server/adapter.ts index 3a45b0132b9c2..08c9fe2cca92e 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -1348,14 +1348,14 @@ export const makePiAdapterV2 = Effect.fn("makePiAdapterV2")(function* ( turnItem, }; pendingPrompts.set(String(requestId), pending); + yield* emit({ type: "node.updated", driver: PI_PROVIDER, node }); + yield* emit({ type: "turn_item.updated", driver: PI_PROVIDER, turnItem }); yield* emit({ type: "runtime_request.updated", driver: PI_PROVIDER, threadId, runtimeRequest, }); - yield* emit({ type: "node.updated", driver: PI_PROVIDER, node }); - yield* emit({ type: "turn_item.updated", driver: PI_PROVIDER, turnItem }); if (expiresAt !== null) { const now = yield* DateTime.now; pending.expirationFiber = yield* Effect.sleep(