From 7eb99495b63f45ac03b2d245072958f8a8dff730 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 25 Oct 2026 21:59:37 +0000 Subject: [PATCH 1/3] 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 | 44 ++- .../provider-pi/src/server/adapter.test.ts | 24 +- packages/provider-pi/src/server/adapter.ts | 28 +- 11 files changed, 849 insertions(+), 128 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 847c2b4551c59..a9092ae960454 100644 --- a/packages/provider-muse/src/server/adapter.test.ts +++ b/packages/provider-muse/src/server/adapter.test.ts @@ -1028,7 +1028,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); @@ -1064,11 +1064,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); @@ -1076,35 +1081,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", () => @@ -1224,11 +1384,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, @@ -1236,6 +1391,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, @@ -1263,8 +1423,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)), ); @@ -1387,8 +1555,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 b29f8d51947bf..325e3d37686fa 100644 --- a/packages/provider-muse/src/server/adapter.ts +++ b/packages/provider-muse/src/server/adapter.ts @@ -223,13 +223,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; } @@ -253,6 +259,7 @@ const nativeRef = (nativeId: string) => ({ const backgroundTaskId = (item: MuseItem) => item.kind === "subagent" ? item.childSessionId || item.subagentId || item.itemId : item.itemId; 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", @@ -919,26 +926,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, @@ -1080,29 +1094,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"); }); /** @@ -1110,7 +1157,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 ( @@ -1123,11 +1172,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 = @@ -1247,11 +1319,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)), ), @@ -1265,20 +1348,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", @@ -1299,7 +1384,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)) @@ -2158,7 +2247,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, }; }); @@ -2306,7 +2399,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 = []; @@ -2338,11 +2432,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 94bdb56eb29ab..58970dc9d838e 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -413,6 +413,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 { @@ -1913,6 +1915,16 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function 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", @@ -1924,17 +1936,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, @@ -1966,12 +1985,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) { @@ -3530,6 +3544,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 }, @@ -3550,6 +3566,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 87e90e3258f4f..a4e79e65902b4 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -1453,13 +1453,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 d47b693f0e123..c3056e0cea3e2 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -3003,33 +3003,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 033dc6ecf199da91bfd74d91058f76c745c2f0b9 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 01:34:59 +0000 Subject: [PATCH 2/3] 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 a4e79e65902b4..902b22cfa9d0e 100644 --- a/packages/provider-pi/src/server/adapter.test.ts +++ b/packages/provider-pi/src/server/adapter.test.ts @@ -3649,9 +3649,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 c3056e0cea3e2..ea3cbc46e3ea9 100644 --- a/packages/provider-pi/src/server/adapter.ts +++ b/packages/provider-pi/src/server/adapter.ts @@ -1576,14 +1576,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( From a39a4e656cdef8dc03dc1de7e73415f72d2c4f1e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 07:49:00 +0000 Subject: [PATCH 3/3] test(server): drop replay tests duplicated by adapter coverage The Muse and Pi replay tests asserted the same preserved answers that the adapter unit tests and the extended muse_permission fixture already catch on main. Co-Authored-By: Claude Opus 5.5 --- .../MuseReplayFixtures.integration.test.ts | 271 +----------------- .../PiReplayFixtures.integration.test.ts | 102 +------ 2 files changed, 4 insertions(+), 369 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 f150135a814db..fee15966df4ed 100644 --- a/apps/server/src/orchestration-v2/testkit/MuseReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/MuseReplayFixtures.integration.test.ts @@ -1,16 +1,10 @@ -import { projectionFor } from "./fixtures/shared.ts"; -import type { ProviderReplayEntry } from "@t3tools/contracts"; -import { assert, describe, it } from "@effect/vitest"; +import { describe } from "@effect/vitest"; import { materializeMuseReplayWorkspace, MuseOrchestratorReplayHarness, } from "../Adapters/MuseAdapterV2.testkit.ts"; -import { - findReplayFixture, - itRunsReplayFixtures, - runReplayFixture, -} from "./ProviderReplayFixtures.ts"; +import { itRunsReplayFixtures, runReplayFixture } from "./ProviderReplayFixtures.ts"; describe("Muse replay fixtures", () => { itRunsReplayFixtures("muse", (input) => @@ -20,265 +14,4 @@ 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/PiReplayFixtures.integration.test.ts b/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts index eff829a450595..b9db455d1cffd 100644 --- a/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/PiReplayFixtures.integration.test.ts @@ -1,108 +1,10 @@ -import { projectionFor } from "./fixtures/shared.ts"; -import type { ProviderReplayEntry } from "@t3tools/contracts"; -import { assert, describe, it } from "@effect/vitest"; +import { describe } from "@effect/vitest"; import { PiOrchestratorReplayHarness } from "../Adapters/PiAdapterV2.testkit.ts"; -import { - findReplayFixture, - itRunsReplayFixtures, - runReplayFixture, -} from "./ProviderReplayFixtures.ts"; +import { 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), - ], - }; - }, - }); - }); });