From 77f67d1766e6b2afa7e3b87a6da52cc652ceb8bf Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Mar 2026 10:44:34 -0800 Subject: [PATCH] Handle stale approval responses and cover Cursor resume IDs - Catch provider approval response failures and emit `provider.approval.respond.failed` - Auto-resolve approvals when provider reports unknown pending request IDs - Add tests for Cursor ACP resume via `resumeCursor.acpSessionId` and legacy `sessionId` --- .../Layers/ProviderCommandReactor.test.ts | 101 +++++++++++++++++- .../Layers/ProviderCommandReactor.ts | 97 ++++++++++++++++- .../src/provider/Layers/CursorAdapter.test.ts | 62 +++++++++++ 3 files changed, 253 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 2f0da9b9652ee6a0adba0495e4f0d52d41e8e84b..711394061752659809383846fd230aa7f47a0503 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -6,6 +6,7 @@ import type { ProviderRuntimeEvent, ProviderSession } from "@t3tools/contracts"; import { ApprovalRequestId, CommandId, + EventId, MessageId, ProjectId, ProviderSessionId, @@ -19,6 +20,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { ServerConfig } from "../../config.ts"; import { TextGenerationError } from "../../git/Errors.ts"; +import { ProviderAdapterRequestError } from "../../provider/Errors.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; @@ -43,10 +45,10 @@ const asApprovalRequestId = (value: string): ApprovalRequestId => const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asTurnId = (value: string): TurnId => TurnId.makeUnsafe(value); -async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise { +async function waitFor(predicate: () => boolean | Promise, timeoutMs = 2000): Promise { const deadline = Date.now() + timeoutMs; const poll = async (): Promise => { - if (predicate()) { + if (await predicate()) { return; } if (Date.now() >= deadline) { @@ -775,6 +777,101 @@ describe("ProviderCommandReactor", () => { }); }); + it("surfaces stale provider approval request failures and resolves stale pending approvals", async () => { + const harness = await createHarness(); + const now = new Date().toISOString(); + harness.respondToRequest.mockImplementation(() => + Effect.fail( + new ProviderAdapterRequestError({ + provider: "cursor", + method: "session/request_permission", + detail: "Unknown pending permission request: approval-request-1", + }), + ), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.session.set", + commandId: CommandId.makeUnsafe("cmd-session-set-for-approval-error"), + threadId: ThreadId.makeUnsafe("thread-1"), + session: { + threadId: ThreadId.makeUnsafe("thread-1"), + status: "running", + providerName: "cursor", + providerSessionId: asSessionId("sess-1"), + providerThreadId: ProviderThreadId.makeUnsafe("provider-thread-1"), + approvalPolicy: "on-request", + sandboxMode: "workspace-write", + activeTurnId: null, + lastError: null, + updatedAt: now, + }, + createdAt: now, + }), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.activity.append", + commandId: CommandId.makeUnsafe("cmd-approval-requested"), + threadId: ThreadId.makeUnsafe("thread-1"), + activity: { + id: EventId.makeUnsafe("activity-approval-requested"), + tone: "approval", + kind: "approval.requested", + summary: "Command approval requested", + payload: { + requestId: "approval-request-1", + requestKind: "command", + }, + turnId: null, + createdAt: now, + }, + createdAt: now, + }), + ); + + await Effect.runPromise( + harness.engine.dispatch({ + type: "thread.approval.respond", + commandId: CommandId.makeUnsafe("cmd-approval-respond-stale"), + threadId: ThreadId.makeUnsafe("thread-1"), + requestId: asApprovalRequestId("approval-request-1"), + decision: "acceptForSession", + createdAt: now, + }), + ); + + await waitFor(async () => { + const readModel = await Effect.runPromise(harness.engine.getReadModel()); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + if (!thread) return false; + return thread.activities.some((activity) => activity.kind === "provider.approval.respond.failed"); + }); + + const readModel = await Effect.runPromise(harness.engine.getReadModel()); + const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); + expect(thread).toBeDefined(); + + const failureActivity = thread?.activities.find( + (activity) => activity.kind === "provider.approval.respond.failed", + ); + expect(failureActivity).toBeDefined(); + expect(failureActivity?.payload).toMatchObject({ + requestId: "approval-request-1", + }); + + const resolvedActivity = thread?.activities.find( + (activity) => + activity.kind === "approval.resolved" && + typeof activity.payload === "object" && + activity.payload !== null && + (activity.payload as Record).requestId === "approval-request-1", + ); + expect(resolvedActivity).toBeDefined(); + }); + it("reacts to thread.session.stop by stopping provider session and clearing thread session state", async () => { const harness = await createHarness(); const now = new Date().toISOString(); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index e1ebea54e1ed604d9328a6e7607e402733a305f9..2946bab3ac3d9fa866ac7f076a6afbfe0168ca84 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -16,6 +16,7 @@ import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Stream } from "ef import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; import { GitCore } from "../../git/Services/GitCore.ts"; +import { ProviderAdapterRequestError } from "../../provider/Errors.ts"; import { TextGeneration } from "../../git/Services/TextGeneration.ts"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; @@ -71,6 +72,32 @@ const DEFAULT_SANDBOX_MODE: ProviderSandboxMode = "workspace-write"; const WORKTREE_BRANCH_PREFIX = "t3code"; const TEMP_WORKTREE_BRANCH_PATTERN = new RegExp(`^${WORKTREE_BRANCH_PREFIX}\\/[0-9a-f]{8}$`); +function toErrorMessage(error: unknown): string { + if (error instanceof Error && error.message.trim().length > 0) { + return error.message; + } + try { + return JSON.stringify(error); + } catch { + return String(error); + } +} + +function isUnknownPendingApprovalRequestError(error: unknown): boolean { + if (error instanceof ProviderAdapterRequestError) { + const detail = error.detail.toLowerCase(); + return ( + detail.includes("unknown pending approval request") || + detail.includes("unknown pending permission request") + ); + } + const message = toErrorMessage(error).toLowerCase(); + return ( + message.includes("unknown pending approval request") || + message.includes("unknown pending permission request") + ); +} + function isTemporaryWorktreeBranch(branch: string): boolean { return TEMP_WORKTREE_BRANCH_PATTERN.test(branch.trim().toLowerCase()); } @@ -127,6 +154,7 @@ const make = Effect.gen(function* () { readonly detail: string; readonly turnId: TurnId | null; readonly createdAt: string; + readonly requestId?: string; }) => orchestrationEngine.dispatch({ type: "thread.activity.append", @@ -139,6 +167,7 @@ const make = Effect.gen(function* () { summary: input.summary, payload: { detail: input.detail, + ...(input.requestId ? { requestId: input.requestId } : {}), }, turnId: input.turnId, createdAt: input.createdAt, @@ -146,6 +175,33 @@ const make = Effect.gen(function* () { createdAt: input.createdAt, }); + const appendApprovalResolvedActivity = (input: { + readonly threadId: ThreadId; + readonly requestId: string; + readonly decision: "accept" | "acceptForSession" | "decline" | "cancel"; + readonly detail: string; + readonly createdAt: string; + }) => + orchestrationEngine.dispatch({ + type: "thread.activity.append", + commandId: serverCommandId("approval-resolved"), + threadId: input.threadId, + activity: { + id: EventId.makeUnsafe(crypto.randomUUID()), + tone: "approval", + kind: "approval.resolved", + summary: "Approval resolved", + payload: { + requestId: input.requestId, + decision: input.decision, + detail: input.detail, + }, + turnId: null, + createdAt: input.createdAt, + }, + createdAt: input.createdAt, + }); + const setThreadSession = (input: { readonly threadId: ThreadId; readonly session: OrchestrationSession; @@ -488,14 +544,45 @@ const make = Effect.gen(function* () { detail: "No active provider session is bound to this thread.", turnId: null, createdAt: event.payload.createdAt, + requestId: event.payload.requestId, }); } - yield* providerService.respondToRequest({ - sessionId, - requestId: event.payload.requestId, - decision: event.payload.decision, - }); + yield* providerService + .respondToRequest({ + sessionId, + requestId: event.payload.requestId, + decision: event.payload.decision, + }) + .pipe( + Effect.catchCause((cause) => + Effect.gen(function* () { + const error = Cause.squash(cause); + const detail = toErrorMessage(error); + yield* appendProviderFailureActivity({ + threadId: event.payload.threadId, + kind: "provider.approval.respond.failed", + summary: "Provider approval response failed", + detail, + turnId: null, + createdAt: event.payload.createdAt, + requestId: event.payload.requestId, + }); + + if (!isUnknownPendingApprovalRequestError(error)) { + return; + } + + yield* appendApprovalResolvedActivity({ + threadId: event.payload.threadId, + requestId: event.payload.requestId, + decision: event.payload.decision, + detail: "Approval request is no longer pending in the provider session.", + createdAt: event.payload.createdAt, + }); + }), + ), + ); }); const processSessionStopRequested = Effect.fnUntraced(function* ( diff --git a/apps/server/src/provider/Layers/CursorAdapter.test.ts b/apps/server/src/provider/Layers/CursorAdapter.test.ts index 00aa44350b8105c9c247f17f16bb1ce95f3d087d..f2188ac007b1c8f16d48c409f62590432077c550 100644 --- a/apps/server/src/provider/Layers/CursorAdapter.test.ts +++ b/apps/server/src/provider/Layers/CursorAdapter.test.ts @@ -359,6 +359,68 @@ describe("CursorAdapterLive", () => { }).pipe(Effect.provide(layer)); }); + it.effect("resumes ACP session using resumeCursor.acpSessionId", () => { + const fake = new FakeCursorAcpProcess(); + const layer = makeCursorAdapterLive({ + createProcess: () => fake as never, + }); + + return Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const session = yield* adapter.startSession({ + provider: "cursor", + cwd: "/tmp/project", + resumeCursor: { + acpSessionId: "acp-session-resume", + }, + }); + + const methods = new Set(fake.requests.map((request) => request.method)); + assert.equal(methods.has("session/load"), true); + assert.equal(methods.has("session/new"), false); + + const loadRequest = fake.requests.find((request) => request.method === "session/load"); + assert.deepEqual(loadRequest?.params, { + sessionId: "acp-session-resume", + cwd: "/tmp/project", + mcpServers: [], + }); + assert.equal(session.threadId, "acp-session-resume"); + assert.deepEqual(session.resumeCursor, { + acpSessionId: "acp-session-resume", + }); + }).pipe(Effect.provide(layer)); + }); + + it.effect("accepts legacy resumeCursor.sessionId for ACP session resume", () => { + const fake = new FakeCursorAcpProcess(); + const layer = makeCursorAdapterLive({ + createProcess: () => fake as never, + }); + + return Effect.gen(function* () { + const adapter = yield* CursorAdapter; + const session = yield* adapter.startSession({ + provider: "cursor", + cwd: "/tmp/project", + resumeCursor: { + sessionId: "acp-session-legacy", + }, + }); + + const loadRequest = fake.requests.find((request) => request.method === "session/load"); + assert.deepEqual(loadRequest?.params, { + sessionId: "acp-session-legacy", + cwd: "/tmp/project", + mcpServers: [], + }); + assert.equal(session.threadId, "acp-session-legacy"); + assert.deepEqual(session.resumeCursor, { + acpSessionId: "acp-session-legacy", + }); + }).pipe(Effect.provide(layer)); + }); + it.effect("bridges permission requests to request.opened/request.resolved", () => { const fake = new FakeCursorAcpProcess(); const layer = makeCursorAdapterLive({ -- 2.54.0