diff --git a/apps/mobile/src/lib/threadActivity.test.ts b/apps/mobile/src/lib/threadActivity.test.ts index 557403072..c71e8c381 100644 --- a/apps/mobile/src/lib/threadActivity.test.ts +++ b/apps/mobile/src/lib/threadActivity.test.ts @@ -144,6 +144,68 @@ describe("pending user input answers", () => { }); describe("pending approvals", () => { + it.each([{}, { requestType: "unknown" }])( + "exposes legacy OpenCode approvals without a known request kind: %j", + (legacyPayload) => { + const requested = makeActivity({ + id: EventId.make("approval-legacy"), + kind: "approval.requested", + summary: "Approval requested", + createdAt: "2026-08-24T00:00:00.000Z", + payload: { requestId: "per-legacy", detail: "*", ...legacyPayload }, + }); + + expect(derivePendingApprovals([requested])).toEqual([ + { + requestId: "per-legacy", + requestKind: "command", + createdAt: requested.createdAt, + detail: "*", + }, + ]); + }, + ); + + it.each(["tool_user_input", "auth_tokens_refresh"])( + "does not turn %s into an approval", + (requestType) => { + const activity = makeActivity({ + id: EventId.make("approval-non-approval"), + kind: "approval.requested", + summary: "Approval requested", + createdAt: "2026-08-24T00:00:00.000Z", + payload: { requestId: "not-an-approval", requestType }, + }); + + expect(derivePendingApprovals([activity])).toEqual([]); + }, + ); + + it.each(["approval.resolved", "provider.approval.respond.failed"])( + "removes legacy approvals after %s", + (kind) => { + const requested = makeActivity({ + id: EventId.make("approval-legacy-open"), + kind: "approval.requested", + summary: "Approval requested", + createdAt: "2026-08-24T00:00:00.000Z", + payload: { requestId: "per-legacy", requestType: "unknown" }, + }); + const resolved = makeActivity({ + id: EventId.make("approval-legacy-resolved"), + kind, + summary: "Approval resolved", + createdAt: "2026-08-24T00:00:01.000Z", + payload: { + requestId: "per-legacy", + detail: "Unknown pending permission request: per-legacy", + }, + }); + + expect(derivePendingApprovals([requested, resolved])).toEqual([]); + }, + ); + it("keeps app access approvals and persistence choices from remote environments", () => { const options = [ { decision: "decline", label: "Decline" }, diff --git a/apps/mobile/src/lib/threadActivity.ts b/apps/mobile/src/lib/threadActivity.ts index 20316cfd3..491b71f94 100644 --- a/apps/mobile/src/lib/threadActivity.ts +++ b/apps/mobile/src/lib/threadActivity.ts @@ -1647,10 +1647,16 @@ export function derivePendingApprovals( ? payload.options.filter(isProviderApprovalOption) : undefined; - if (activity.kind === "approval.requested" && requestId && requestKind) { + if ( + activity.kind === "approval.requested" && + requestId && + payload?.requestType !== "tool_user_input" && + payload?.requestType !== "auth_tokens_refresh" + ) { openByRequestId.set(requestId, { requestId, - requestKind, + // Older OpenCode requests can have no recognized approval kind. + requestKind: requestKind ?? "command", createdAt: activity.createdAt, ...(detail ? { detail } : {}), ...(appName ? { appName } : {}), diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index b374849f5..667432b69 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -1,4 +1,5 @@ import { + ApprovalRequestId, CheckpointRef, CommandId, CorrelationId, @@ -2863,7 +2864,7 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }), ); - it.effect("ignores non-stale provider approval response failures", () => + it.effect("restores pending approvals when a provider reply fails", () => Effect.gen(function* () { const projectionPipeline = yield* OrchestrationProjectionPipeline; const eventStore = yield* OrchestrationEventStore; @@ -2948,6 +2949,24 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { }, }); + yield* appendAndProject({ + type: "thread.approval-response-requested", + eventId: EventId.make("evt-nonstale-approval-response"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-nonstale-approval"), + occurredAt: "2026-02-26T12:45:02.500Z", + commandId: CommandId.make("cmd-nonstale-approval-response"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-nonstale-approval-response"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-nonstale-approval"), + requestId: ApprovalRequestId.make("approval-request-nonstale-existing"), + decision: "accept", + createdAt: "2026-02-26T12:45:02.500Z", + }, + }); + yield* appendAndProject({ type: "thread.activity-appended", eventId: EventId.make("evt-nonstale-approval-4"), @@ -3040,6 +3059,67 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => { WHERE thread_id = 'thread-nonstale-approval' `; assert.deepEqual(threadRows, [{ pendingApprovalCount: 1 }]); + + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-nonstale-approval-resolved"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-nonstale-approval"), + occurredAt: "2026-02-26T12:45:05.000Z", + commandId: CommandId.make("cmd-nonstale-approval-resolved"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-nonstale-approval-resolved"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-nonstale-approval"), + activity: { + id: EventId.make("activity-nonstale-approval-resolved"), + tone: "approval", + kind: "approval.resolved", + summary: "Approval resolved", + payload: { + requestId: "approval-request-nonstale-existing", + decision: "accept", + }, + turnId: null, + createdAt: "2026-02-26T12:45:05.000Z", + }, + }, + }); + + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-nonstale-approval-late-failure"), + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-nonstale-approval"), + occurredAt: "2026-02-26T12:45:06.000Z", + commandId: CommandId.make("cmd-nonstale-approval-late-failure"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-nonstale-approval-late-failure"), + metadata: {}, + payload: { + threadId: ThreadId.make("thread-nonstale-approval"), + activity: { + id: EventId.make("activity-nonstale-approval-late-failure"), + tone: "error", + kind: "provider.approval.respond.failed", + summary: "Provider approval response failed", + payload: { + requestId: "approval-request-nonstale-existing", + detail: "Provider timed out while responding to approval request", + }, + turnId: null, + createdAt: "2026-02-26T12:45:06.000Z", + }, + }, + }); + + const resolvedThreadRows = yield* sql<{ readonly pendingApprovalCount: number }>` + SELECT pending_approval_count AS "pendingApprovalCount" + FROM projection_threads + WHERE thread_id = 'thread-nonstale-approval' + `; + assert.deepEqual(resolvedThreadRows, [{ pendingApprovalCount: 0 }]); }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 57109a914..d56d94b2c 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1813,6 +1813,44 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti }); return; } + if (Option.isNone(existingRow) || existingRow.value.status !== "resolved") { + return; + } + + // Sending a reply clears the badge before the provider accepts it. + // A failed reply must restore the request unless a terminal event + // already closed it, including a reply from another client. + const requestActivities = (yield* projectionThreadActivityRepository.listByThreadId({ + threadId: existingRow.value.threadId, + })).filter((activity) => extractActivityRequestId(activity.payload) === requestId); + const wasRequested = requestActivities.some( + (activity) => activity.kind === "approval.requested", + ); + const wasResolved = requestActivities.some((activity) => { + if (activity.kind === "approval.resolved") { + return true; + } + if (activity.kind !== "provider.approval.respond.failed") { + return false; + } + const activityPayload = + typeof activity.payload === "object" && activity.payload !== null + ? (activity.payload as Record) + : null; + return isStalePendingApprovalFailureDetail( + typeof activityPayload?.detail === "string" + ? activityPayload.detail.toLowerCase() + : null, + ); + }); + if (wasRequested && !wasResolved) { + yield* projectionPendingApprovalRepository.upsert({ + ...existingRow.value, + status: "pending", + decision: null, + resolvedAt: null, + }); + } return; } // Only approval-requested activities should create pending-approval diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 454e8cc14..297d432d8 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1582,6 +1582,217 @@ describe("ProviderRuntimeIngestion", () => { ).toEqual(["retry output"]); }); + it.each([ + { delivery: "buffered", enableLegacyTokenStreaming: false }, + { delivery: "streamed", enableLegacyTokenStreaming: true }, + ])("settles OpenCode aborted turns and saves $delivery assistant text", async (settings) => { + const harness = await createHarness({ + serverSettings: { enableLegacyTokenStreaming: settings.enableLegacyTokenStreaming }, + }); + const threadId = asThreadId("thread-1"); + const turnId = asTurnId("opencode-aborted-turn"); + const base = { + provider: ProviderDriverKind.make("opencode"), + threadId, + turnId, + createdAt: "2026-01-01T00:00:01.000Z", + }; + harness.emit({ ...base, type: "turn.started", eventId: asEventId("opencode-started") }); + harness.emit({ + ...base, + type: "content.delta", + eventId: asEventId("opencode-partial-text"), + itemId: asItemId("opencode-text-part"), + payload: { streamKind: "assistant_text", delta: "Work before the stop." }, + }); + harness.emit({ + ...base, + type: "turn.aborted", + eventId: asEventId("opencode-aborted"), + createdAt: "2026-01-01T00:00:02.000Z", + payload: { reason: "Interrupted by user." }, + }); + + await harness.drain(); + const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); + expect(thread?.session).toMatchObject({ + status: "interrupted", + activeTurnId: null, + lastError: null, + }); + expect(thread?.latestTurn).toMatchObject({ + turnId, + state: "interrupted", + completedAt: "2026-01-01T00:00:02.000Z", + }); + expect(thread?.messages).toEqual([ + expect.objectContaining({ + role: "assistant", + turnId, + text: "Work before the stop.", + streaming: false, + }), + ]); + }); + + it.each([ + { source: "the previous turn", turnId: asTurnId("opencode-stopped-turn") }, + { source: "an unspecified turn", turnId: undefined }, + ])("ignores late OpenCode aborts for $source across newer turns", async (lateAbort) => { + const harness = await createHarness({ + serverSettings: { enableLegacyTokenStreaming: true }, + }); + const threadId = asThreadId("thread-1"); + const stoppedTurnId = asTurnId("opencode-stopped-turn"); + const nextTurnId = asTurnId("opencode-next-turn"); + const base = { + provider: ProviderDriverKind.make("opencode"), + threadId, + createdAt: "2026-01-01T00:00:01.000Z", + }; + harness.emit({ + ...base, + type: "turn.started", + eventId: asEventId("opencode-first-started"), + turnId: stoppedTurnId, + }); + harness.emit({ + ...base, + type: "turn.aborted", + eventId: asEventId("opencode-first-aborted"), + turnId: stoppedTurnId, + payload: { reason: "Interrupted by user." }, + }); + await harness.drain(); + await harness.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("opencode-next-request"), + threadId, + message: { + messageId: asMessageId("opencode-next-message"), + role: "user", + text: "Start the next turn.", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: base.createdAt, + }); + harness.emit({ + ...base, + type: "session.state.changed", + eventId: asEventId("opencode-next-starting"), + payload: { state: "starting" }, + }); + harness.emit({ + ...base, + type: "turn.started", + eventId: asEventId("opencode-next-started"), + turnId: nextTurnId, + }); + harness.emit({ + ...base, + type: "content.delta", + eventId: asEventId("opencode-next-partial-text"), + turnId: nextTurnId, + itemId: asItemId("opencode-next-text-part"), + payload: { streamKind: "assistant_text", delta: "The next turn is running." }, + }); + await harness.drain(); + + harness.emit({ + ...base, + type: "turn.aborted", + eventId: asEventId("opencode-late-abort"), + ...(lateAbort.turnId ? { turnId: lateAbort.turnId } : {}), + payload: { reason: "Interrupted by user." }, + }); + await harness.drain(); + + const thread = (await harness.readModel()).threads.find((entry) => entry.id === threadId); + expect(thread?.session).toMatchObject({ status: "running", activeTurnId: nextTurnId }); + expect(thread?.latestTurn).toMatchObject({ turnId: nextTurnId, state: "running" }); + expect(thread?.messages.filter((message) => message.role === "assistant")).toEqual([ + expect.objectContaining({ + turnId: nextTurnId, + text: "The next turn is running.", + streaming: true, + }), + ]); + + harness.emit({ + ...base, + type: "turn.completed", + eventId: asEventId("opencode-next-completed"), + turnId: nextTurnId, + createdAt: "2026-01-01T00:00:02.000Z", + payload: { state: "completed" }, + }); + await harness.drain(); + + const pendingAt = "2026-01-01T00:00:03.000Z"; + for (const hasPendingStart of [false, true]) { + if (hasPendingStart) { + await harness.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make("opencode-pending-start"), + threadId, + message: { + messageId: asMessageId("opencode-pending-message"), + role: "user", + text: "Start another turn.", + attachments: [], + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + createdAt: pendingAt, + }); + harness.emit({ + ...base, + type: "session.state.changed", + eventId: asEventId("opencode-pending-starting"), + createdAt: pendingAt, + payload: { state: "starting" }, + }); + } + harness.emit({ + ...base, + type: "turn.aborted", + eventId: asEventId(`opencode-late-abort-after-completion-${hasPendingStart}`), + ...(lateAbort.turnId ? { turnId: lateAbort.turnId } : {}), + createdAt: "2026-01-01T00:00:04.000Z", + payload: { reason: "Interrupted by user." }, + }); + await harness.drain(); + + const completedThread = (await harness.readModel()).threads.find( + (entry) => entry.id === threadId, + ); + expect(completedThread?.session).toMatchObject({ + status: hasPendingStart ? "starting" : "ready", + activeTurnId: null, + }); + expect(completedThread?.latestTurn).toMatchObject({ turnId: nextTurnId, state: "completed" }); + } + + harness.emit({ + ...base, + type: "turn.started", + eventId: asEventId("opencode-pending-started"), + turnId: asTurnId("opencode-pending-turn"), + createdAt: "2026-01-01T00:00:05.000Z", + }); + await harness.drain(); + const startedThread = (await harness.readModel()).threads.find( + (entry) => entry.id === threadId, + ); + expect(startedThread?.latestTurn).toMatchObject({ + turnId: asTurnId("opencode-pending-turn"), + state: "running", + requestedAt: pendingAt, + }); + }); + it("applies provider session.state.changed transitions directly", async () => { const harness = await createHarness(); const waitingAt = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 61683577f..e837137e3 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2485,13 +2485,17 @@ const make = Effect.gen(function* () { createdAt: now, }); }); + const isTerminalTurn = event.type === "turn.completed" || event.type === "turn.aborted"; + const isCompactedThreadState = + event.type === "thread.state.changed" && event.payload.state === "compacted"; const pendingTurnStart = event.type === "session.started" || event.type === "session.state.changed" || event.type === "session.exited" || event.type === "thread.started" || event.type === "turn.started" || - event.type === "turn.completed" + isTerminalTurn || + isCompactedThreadState ? yield* projectionTurnRepository.getPendingTurnStartByThreadId({ threadId: thread.id, }) @@ -2552,6 +2556,7 @@ const make = Effect.gen(function* () { if (!conflictsWithActiveTurn) return true; return conflictingTurnStartIsPendingTurnStart; case "turn.completed": + case "turn.aborted": if (conflictsWithActiveTurn || missingTurnForActiveTurn) { return false; } @@ -2559,14 +2564,10 @@ const make = Effect.gen(function* () { if (activeTurnId !== null && eventTurnId !== undefined) { return sameId(activeTurnId, eventTurnId); } - // No active turn tracked: accept only completions that name their - // turn (covers a real completion whose turn.started was lost). An - // untargeted completion cannot prove it belongs to any turn this - // thread ran — the known emitter was the Claude resume handshake - // (system/init + result(num_turns: 0)), which is not a turn at - // all — and applying it here stomps the "starting" lifecycle - // state while a turn start is pending. - return eventTurnId !== undefined; + // A named completion can recover a lost turn.started event. + // An abort needs an active turn so a delayed stop cannot replace + // a ready session or clear a newer pending start. + return event.type === "turn.completed" && eventTurnId !== undefined; default: return true; } @@ -2599,7 +2600,7 @@ const make = Effect.gen(function* () { event.type === "session.exited" || event.type === "thread.started" || event.type === "turn.started" || - event.type === "turn.completed" + isTerminalTurn ) { const status = (() => { switch (event.type) { @@ -2611,6 +2612,8 @@ const make = Effect.gen(function* () { return "running"; case "session.exited": return "stopped"; + case "turn.aborted": + return "interrupted"; case "turn.completed": return normalizeRuntimeTurnState(event.payload.state) === "failed" ? "error" @@ -2625,7 +2628,7 @@ const make = Effect.gen(function* () { const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) - : event.type === "turn.completed" || event.type === "session.exited" + : isTerminalTurn || event.type === "session.exited" ? null : event.type === "session.state.changed" && !sessionStatusAllowsActiveTurn( @@ -2639,7 +2642,7 @@ const make = Effect.gen(function* () { : event.type === "turn.completed" && normalizeRuntimeTurnState(event.payload.state) === "failed" ? (event.payload.errorMessage ?? thread.session?.lastError ?? "Turn failed") - : status === "ready" + : status === "ready" || status === "interrupted" ? null : (thread.session?.lastError ?? null); @@ -2905,7 +2908,7 @@ const make = Effect.gen(function* () { }); } - if (event.type === "turn.completed") { + if (isTerminalTurn) { const detailedThread = yield* getLoadedThreadDetail(); const messages = detailedThread?.messages ?? []; const proposedPlans = detailedThread?.proposedPlans ?? []; @@ -3048,7 +3051,7 @@ const make = Effect.gen(function* () { } else if (!conflictsWithActiveTurn) { if (event.type === "turn.plan.updated") { threadPlanProgress.recordPlanProgress(thread.id, event.payload.plan); - } else if (event.type === "turn.completed" || event.type === "turn.aborted") { + } else if (isTerminalTurn && shouldApplyThreadLifecycle) { threadPlanProgress.clearThreadPlanProgress(thread.id); } } diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 90ea21623..1720ee736 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -15,7 +15,11 @@ import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { beforeEach } from "vite-plus/test"; -import type { PermissionRequest, QuestionRequest } from "@opencode-ai/sdk/v2"; +import type { + Event as OpenCodeEvent, + PermissionRequest, + QuestionRequest, +} from "@opencode-ai/sdk/v2"; import { ApprovalRequestId, @@ -70,6 +74,11 @@ const runtimeMock = { abortImplementation: null as | ((sessionID: string, signal?: AbortSignal) => Promise) | null, + sessionChildrenCalls: [] as string[], + sessionChildrenById: new Map>(), + sessionChildrenImplementation: null as + | ((sessionID: string) => Promise>) + | null, closeCalls: [] as string[], revertCalls: [] as Array<{ sessionID: string; messageID?: string }>, messageCalls: [] as Array<{ sessionID: string; messageID: string }>, @@ -79,22 +88,30 @@ const runtimeMock = { promptAsyncImplementation: null as (() => Promise) | null, autoPromptEcho: true, autoConnect: true, + endEventStream: false, promptEchoEvents: [] as Array, closeError: null as Error | null, messages: [] as MessageEntry[], subscribedEvents: [] as Array>, eventSubscribeObserved: null as (() => void) | null, + eventStreamError: null as ((cause: unknown) => void) | null, permissionReplyCalls: [] as Array<{ requestID: string; reply: string }>, + permissionReplyImplementation: null as ((signal?: AbortSignal) => Promise) | null, + permissionReplySignals: [] as AbortSignal[], questionReplyCalls: [] as Array<{ requestID: string; answers: ReadonlyArray>; }>, + questionReplyImplementation: null as ((signal?: AbortSignal) => Promise) | null, sessionStatus: "idle" as "idle" | "busy", sessionStatusFailures: 0, sessionStatusCalls: 0, sessionStatusImplementation: null as (() => Promise) | null, sessionGetIds: [] as string[], sessionGetObserved: null as ((sessionID: string) => void) | null, + sessionGetImplementation: null as + | ((sessionID: string, signal?: AbortSignal) => Promise) + | null, missingSessionIds: new Set(), transientErrorSessionIds: new Set(), sessionDirectoryById: new Map(), @@ -117,6 +134,9 @@ const runtimeMock = { this.state.abortCalls.length = 0; this.state.abortSignals.length = 0; this.state.abortImplementation = null; + this.state.sessionChildrenCalls.length = 0; + this.state.sessionChildrenById.clear(); + this.state.sessionChildrenImplementation = null; this.state.closeCalls.length = 0; this.state.revertCalls.length = 0; this.state.messageCalls.length = 0; @@ -126,19 +146,25 @@ const runtimeMock = { this.state.promptAsyncImplementation = null; this.state.autoPromptEcho = true; this.state.autoConnect = true; + this.state.endEventStream = false; this.state.promptEchoEvents.length = 0; this.state.closeError = null; this.state.messages = []; this.state.subscribedEvents = []; this.state.eventSubscribeObserved = null; + this.state.eventStreamError = null; this.state.permissionReplyCalls.length = 0; + this.state.permissionReplyImplementation = null; + this.state.permissionReplySignals.length = 0; this.state.questionReplyCalls.length = 0; + this.state.questionReplyImplementation = null; this.state.sessionStatus = "idle"; this.state.sessionStatusFailures = 0; this.state.sessionStatusCalls = 0; this.state.sessionStatusImplementation = null; this.state.sessionGetIds.length = 0; this.state.sessionGetObserved = null; + this.state.sessionGetImplementation = null; this.state.missingSessionIds.clear(); this.state.transientErrorSessionIds.clear(); this.state.sessionDirectoryById.clear(); @@ -210,9 +236,12 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { data: { id: runtimeMock.state.createdSessionIds.shift() ?? `${baseUrl}/session` }, }; }, - get: async ({ sessionID }: { sessionID: string }) => { + get: async ({ sessionID }: { sessionID: string }, options?: { signal?: AbortSignal }) => { runtimeMock.state.sessionGetIds.push(sessionID); runtimeMock.state.sessionGetObserved?.(sessionID); + if (runtimeMock.state.sessionGetImplementation) { + await runtimeMock.state.sessionGetImplementation(sessionID, options?.signal); + } // The real client is `throwOnError: true`: non-2xx rejects rather // than resolving, so missing → 404 throw, transient → 500 throw. if (runtimeMock.state.transientErrorSessionIds.has(sessionID)) { @@ -252,6 +281,20 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { runtimeMock.state.abortSignals.push(options.signal); } await runtimeMock.state.abortImplementation?.(sessionID, options?.signal); + runtimeMock.state.pendingPermissions = runtimeMock.state.pendingPermissions.filter( + (request) => request.sessionID !== sessionID, + ); + runtimeMock.state.pendingQuestions = runtimeMock.state.pendingQuestions.filter( + (request) => request.sessionID !== sessionID, + ); + }, + children: async ({ sessionID }: { sessionID: string }) => { + runtimeMock.state.sessionChildrenCalls.push(sessionID); + return { + data: runtimeMock.state.sessionChildrenImplementation + ? await runtimeMock.state.sessionChildrenImplementation(sessionID) + : (runtimeMock.state.sessionChildrenById.get(sessionID) ?? []), + }; }, status: async () => { runtimeMock.state.sessionStatusCalls += 1; @@ -333,19 +376,60 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { }, }, event: { - subscribe: async () => { + subscribe: async ( + _input: unknown, + options?: { signal?: AbortSignal; onSseError?: (cause: unknown) => void }, + ) => { runtimeMock.state.eventSubscribeObserved?.(); + runtimeMock.state.eventStreamError = options?.onSseError ?? null; return { stream: (async function* () { - if (runtimeMock.state.autoConnect) { - yield { id: "evt-auto-connected", type: "server.connected", properties: {} }; - } - for (const event of runtimeMock.state.subscribedEvents) { - const resolved = await event; - while (runtimeMock.state.promptEchoEvents.length > 0) { - yield runtimeMock.state.promptEchoEvents.shift(); + const aborted = promiseWithResolvers(); + const onAbort = () => aborted.resolve(undefined); + options?.signal?.addEventListener("abort", onAbort, { once: true }); + try { + if (runtimeMock.state.autoConnect) { + yield { id: "evt-auto-connected", type: "server.connected", properties: {} }; + } + for (const event of runtimeMock.state.subscribedEvents) { + if (options?.signal?.aborted) return; + const resolved = await Promise.race([event, aborted.promise]); + if (options?.signal?.aborted) return; + while (runtimeMock.state.promptEchoEvents.length > 0) { + yield runtimeMock.state.promptEchoEvents.shift(); + } + const nativeEvent = resolved as OpenCodeEvent; + if (nativeEvent.type === "permission.asked") { + runtimeMock.state.pendingPermissions = + runtimeMock.state.pendingPermissions.filter( + (request) => request.id !== nativeEvent.properties.id, + ); + runtimeMock.state.pendingPermissions.push(nativeEvent.properties); + } else if (nativeEvent.type === "permission.replied") { + runtimeMock.state.pendingPermissions = + runtimeMock.state.pendingPermissions.filter( + (request) => request.id !== nativeEvent.properties.requestID, + ); + } else if (nativeEvent.type === "question.asked") { + runtimeMock.state.pendingQuestions = runtimeMock.state.pendingQuestions.filter( + (request) => request.id !== nativeEvent.properties.id, + ); + runtimeMock.state.pendingQuestions.push(nativeEvent.properties); + } else if ( + nativeEvent.type === "question.replied" || + nativeEvent.type === "question.rejected" + ) { + runtimeMock.state.pendingQuestions = runtimeMock.state.pendingQuestions.filter( + (request) => request.id !== nativeEvent.properties.requestID, + ); + } + yield resolved; + } + if (!runtimeMock.state.endEventStream && !options?.signal?.aborted) { + await aborted.promise; } - yield resolved; + } finally { + options?.signal?.removeEventListener("abort", onAbort); } })(), }; @@ -360,8 +444,18 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { : runtimeMock.state.pendingPermissions, }; }, - reply: async ({ requestID, reply }: { requestID: string; reply: string }) => { + reply: async ( + { requestID, reply }: { requestID: string; reply: string }, + options?: { signal?: AbortSignal }, + ) => { runtimeMock.state.permissionReplyCalls.push({ requestID, reply }); + if (options?.signal) runtimeMock.state.permissionReplySignals.push(options.signal); + if (runtimeMock.state.permissionReplyImplementation) { + await runtimeMock.state.permissionReplyImplementation(options?.signal); + } + runtimeMock.state.pendingPermissions = runtimeMock.state.pendingPermissions.filter( + (request) => request.id !== requestID, + ); }, }, question: { @@ -373,14 +467,21 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { : runtimeMock.state.pendingQuestions, }; }, - reply: async ({ - requestID, - answers, - }: { - requestID: string; - answers: ReadonlyArray>; - }) => { + reply: async ( + { + requestID, + answers, + }: { + requestID: string; + answers: ReadonlyArray>; + }, + options?: { signal?: AbortSignal }, + ) => { runtimeMock.state.questionReplyCalls.push({ requestID, answers }); + await runtimeMock.state.questionReplyImplementation?.(options?.signal); + runtimeMock.state.pendingQuestions = runtimeMock.state.pendingQuestions.filter( + (request) => request.id !== requestID, + ); }, }, }) as unknown as ReturnType, @@ -1131,6 +1232,9 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { it.effect("stops a configured-server session without trying to own server lifecycle", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; + const rootSessionId = "http://127.0.0.1:9999/session"; + runtimeMock.state.sessionChildrenById.set(rootSessionId, [{ id: "ses_stop_child" }]); + runtimeMock.state.sessionChildrenById.set("ses_stop_child", [{ id: "ses_stop_grandchild" }]); yield* adapter.startSession({ provider: ProviderDriverKind.make("opencode"), threadId: asThreadId("thread-opencode"), @@ -1140,10 +1244,11 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { yield* adapter.stopSession(asThreadId("thread-opencode")); NodeAssert.deepEqual(runtimeMock.state.startCalls, []); - NodeAssert.deepEqual( - runtimeMock.state.abortCalls.includes("http://127.0.0.1:9999/session"), - true, - ); + NodeAssert.deepEqual(runtimeMock.state.abortCalls, [ + rootSessionId, + "ses_stop_child", + "ses_stop_grandchild", + ]); }), ); @@ -2517,6 +2622,414 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect.each([ + { permission: "external_directory", decision: "accept", reply: "once" }, + { permission: "doom_loop", decision: "acceptForSession", reply: "always" }, + { permission: "todowrite", decision: "decline", reply: "reject" }, + { permission: "webfetch", decision: "cancel", reply: "reject" }, + { permission: "custom_tool", decision: "accept", reply: "once" }, + ] as const)( + "shows $permission approval and resolves its $decision reply without SSE", + ({ permission, decision, reply }) => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId(`thread-permission-${permission}`); + const request = { + ...permissionRequest(`per_${permission}`, "http://127.0.0.1:9999/session"), + permission, + patterns: ["*"], + }; + runtimeMock.state.subscribedEvents = [ + { + id: "evt-permission", + type: "permission.asked", + properties: request, + } satisfies OpenCodeEvent, + ]; + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + }); + const opened = Option.getOrThrow(yield* Fiber.join(openedFiber)); + NodeAssert.ok(opened.type === "request.opened"); + NodeAssert.equal(opened.payload.requestType, "command_execution_approval"); + NodeAssert.equal( + opened.payload.detail, + `${permission.replaceAll("_", " ")}\n\nAllow for workspace also permits matching requests in other OpenCode sessions in this workspace.`, + ); + NodeAssert.deepEqual( + opened.payload.options?.map((option) => option.label), + ["Allow once", "Allow for workspace", "Deny"], + ); + const resolvedFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "request.resolved", + ), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make(request.id), decision); + const resolved = Option.getOrThrow(yield* Fiber.join(resolvedFiber)); + NodeAssert.equal(resolved.requestId, request.id); + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make(request.id), decision); + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, [ + { requestID: request.id, reply }, + ]); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("keeps a permission reply retryable after its HTTP request times out", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-permission-timeout"); + const request = permissionRequest("per_timeout", "http://127.0.0.1:9999/session"); + const replyStarted = promiseWithResolvers(); + runtimeMock.state.permissionReplyImplementation = async () => { + replyStarted.resolve(undefined); + await new Promise(() => {}); + }; + runtimeMock.state.subscribedEvents = [ + { id: "evt-ask", type: "permission.asked", properties: request }, + ]; + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + }); + yield* Fiber.join(openedFiber); + const replyFiber = yield* adapter + .respondToRequest(threadId, ApprovalRequestId.make(request.id), "accept") + .pipe(Effect.exit, Effect.forkChild); + yield* Effect.promise(() => replyStarted.promise); + yield* Effect.yieldNow; + yield* advanceTestClock(10_000); + NodeAssert.equal(Exit.isFailure(yield* Fiber.join(replyFiber)), true); + NodeAssert.equal(runtimeMock.state.permissionReplySignals[0]?.aborted, true); + runtimeMock.state.permissionReplyImplementation = null; + const resolvedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.resolved"), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make(request.id), "accept"); + NodeAssert.equal(Option.getOrThrow(yield* Fiber.join(resolvedFiber)).requestId, request.id); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("keeps a recovering permission retryable until its native request is loaded", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-permission-recovering"); + const request = permissionRequest("per_recovering", "ses_resumed"); + const listStarted = promiseWithResolvers(); + const releaseList = promiseWithResolvers(); + runtimeMock.state.permissionListImplementation = async () => { + listStarted.resolve(undefined); + return await releaseList.promise; + }; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + resumeCursor: { schemaVersion: 1, sessionId: request.sessionID }, + }); + yield* Effect.promise(() => listStarted.promise); + const reply = yield* adapter + .respondToRequest(threadId, ApprovalRequestId.make(request.id), "accept") + .pipe(Effect.result); + NodeAssert.equal(reply._tag, "Failure"); + if (reply._tag === "Failure" && reply.failure._tag === "ProviderAdapterRequestError") { + NodeAssert.match(reply.failure.detail, /still loading/); + } + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.runHead, + Effect.forkChild, + ); + releaseList.resolve([request]); + yield* Fiber.join(openedFiber); + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make(request.id), "accept"); + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, [ + { requestID: request.id, reply: "once" }, + ]); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("closes missing permissions and questions after reconnect", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-missing-requests"); + const sessionID = "http://127.0.0.1:9999/session"; + const reconnect = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + { + id: "evt-permission", + type: "permission.asked", + properties: permissionRequest("per_missing", sessionID), + }, + { + id: "evt-question", + type: "question.asked", + properties: questionRequest("que_missing", sessionID), + }, + reconnect.promise, + ]; + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "request.opened" || event.type === "user-input.requested"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + }); + yield* Fiber.join(openedFiber); + const resolvedFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "request.resolved" || event.type === "user-input.resolved"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + runtimeMock.state.pendingPermissions = []; + runtimeMock.state.pendingQuestions = []; + reconnect.resolve({ id: "evt-reconnected", type: "server.connected", properties: {} }); + const resolved = yield* Fiber.join(resolvedFiber); + NodeAssert.deepEqual( + resolved.map((event) => event.requestId), + ["per_missing", "que_missing"], + ); + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, []); + NodeAssert.deepEqual(runtimeMock.state.questionReplyCalls, []); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("closes pending requests after Stop and ignores late requests from that turn", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-stop-requests"); + const sessionID = "http://127.0.0.1:9999/session"; + const startRequests = promiseWithResolvers(); + const lateRequests = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + startRequests.promise, + { + id: "evt-question", + type: "question.asked", + properties: questionRequest("que_stop", sessionID), + }, + lateRequests.promise, + { + id: "evt-late-question", + type: "question.asked", + properties: questionRequest("que_late", sessionID), + }, + { id: "evt-drained", type: "session.compacted", properties: { sessionID } }, + ]; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Work", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "request.opened" || event.type === "user-input.requested"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + startRequests.resolve({ + id: "evt-permission", + type: "permission.asked", + properties: permissionRequest("per_stop", sessionID), + }); + yield* Fiber.join(openedFiber); + const stoppedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "turn.aborted"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.interruptTurn(threadId, turn.turnId); + const stopped = yield* Fiber.join(stoppedFiber); + NodeAssert.deepEqual( + stopped.map((event) => event.type), + ["request.resolved", "user-input.resolved", "turn.aborted"], + ); + const lateFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "thread.state.changed"), + Stream.runCollect, + Effect.forkChild, + ); + lateRequests.resolve({ + id: "evt-late-permission", + type: "permission.asked", + properties: permissionRequest("per_late", sessionID), + }); + const late = yield* Fiber.join(lateFiber); + NodeAssert.deepEqual( + late.map((event) => event.type), + ["thread.state.changed"], + ); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("keeps progress live during automatic approval and never reopens a finished turn", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-auto-approval-progress"); + const sessionID = "http://127.0.0.1:9999/session"; + const ask = promiseWithResolvers(); + const idle = promiseWithResolvers(); + const replyStarted = promiseWithResolvers(); + const releaseReply = promiseWithResolvers(); + runtimeMock.state.permissionReplyImplementation = async () => { + replyStarted.resolve(undefined); + await releaseReply.promise; + throw new Error("reply response lost"); + }; + runtimeMock.state.subscribedEvents = [ask.promise, idle.promise]; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId, + input: "Work", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const completedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + ask.resolve({ + id: "evt-ask", + type: "permission.asked", + properties: permissionRequest("per_slow_auto", sessionID), + }); + yield* Effect.promise(() => replyStarted.promise); + idle.resolve({ + id: "evt-idle", + type: "session.status", + properties: { sessionID, status: { type: "idle" } }, + }); + const completed = yield* Fiber.join(completedFiber); + NodeAssert.equal( + completed.some((event) => event.type === "request.opened"), + false, + ); + const remainingFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "session.exited"), + Stream.runCollect, + Effect.forkChild, + ); + releaseReply.resolve(undefined); + yield* advanceTestClock(10_000); + yield* adapter.stopSession(threadId); + const remaining = yield* Fiber.join(remainingFiber); + NodeAssert.equal( + remaining.some((event) => event.type === "request.opened"), + false, + ); + }), + ); + + it.effect("keeps automatic approval fallback available after a steer", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-auto-approval-steer"); + const ask = promiseWithResolvers(); + const replyStarted = promiseWithResolvers(); + const releaseReply = promiseWithResolvers(); + runtimeMock.state.sessionStatus = "busy"; + runtimeMock.state.permissionReplyImplementation = async () => { + replyStarted.resolve(undefined); + await releaseReply.promise; + throw new Error("reply failed"); + }; + runtimeMock.state.subscribedEvents = [ask.promise]; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const modelSelection = createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ); + const turn = yield* adapter.sendTurn({ threadId, input: "Work", modelSelection }); + ask.resolve({ + id: "evt-ask", + type: "permission.asked", + properties: permissionRequest("per_steer_auto", "http://127.0.0.1:9999/session"), + }); + yield* Effect.promise(() => replyStarted.promise); + const steered = yield* adapter.sendTurn({ + threadId, + input: "Keep the change small", + modelSelection, + }); + NodeAssert.equal(steered.turnId, turn.turnId); + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.runHead, + Effect.forkChild, + ); + releaseReply.resolve(undefined); + NodeAssert.equal( + Option.getOrThrow(yield* Fiber.join(openedFiber)).requestId, + "per_steer_auto", + ); + runtimeMock.state.permissionReplyImplementation = null; + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make("per_steer_auto"), "accept"); + yield* adapter.stopSession(threadId); + }), + ); + it.effect("routes child-session approval requests and replies through the parent thread", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; @@ -2609,21 +3122,226 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("routes child-session questions and replies through the parent thread", () => - Effect.gen(function* () { - const adapter = yield* OpenCodeAdapter; - const threadId = asThreadId("thread-child-question"); - const questionReply = promiseWithResolvers(); - runtimeMock.state.subscribedEvents = [ - { - id: "evt-child-created", - type: "session.created", - properties: { - sessionID: "ses_child_question", - info: { - id: "ses_child_question", - parentID: "http://127.0.0.1:9999/session", - title: "Child session", + it.effect.each([ + { + name: "a doom-loop ask on the parent session", + requestId: "per_doom_loop", + sessionID: "http://127.0.0.1:9999/session", + permission: "doom_loop", + patterns: ["bash"], + always: [] as string[], + }, + { + name: "a child-session ask", + requestId: "per_child_full", + sessionID: "ses_child_full", + permission: "read", + patterns: ["/repo/settings.env"], + always: ["/repo/settings.env"], + }, + ])( + "auto-approves $name in full access", + ({ requestId, sessionID, permission, patterns, always }) => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId(`thread-full-access-${requestId}`); + const replyStarted = promiseWithResolvers(); + runtimeMock.state.permissionReplyImplementation = async () => + replyStarted.resolve(undefined); + runtimeMock.state.subscribedEvents = [ + { + id: "evt-child-created", + type: "session.created", + properties: { + sessionID: "ses_child_full", + info: { + id: "ses_child_full", + parentID: "http://127.0.0.1:9999/session", + title: "Child session", + }, + }, + }, + { + id: "evt-permission", + type: "permission.asked", + properties: { id: requestId, sessionID, permission, patterns, metadata: {}, always }, + }, + replyStarted.promise.then(() => ({ + id: "evt-permission-replied", + type: "permission.replied", + properties: { sessionID, requestID: requestId, reply: "once" }, + })), + // The suppressed ask emits nothing, so an empty question serves as a + // sentinel that closes the collected stream once the pump is past it. + { + id: "evt-sentinel-question", + type: "question.asked", + properties: { + id: "que_sentinel", + sessionID: "http://127.0.0.1:9999/session", + questions: [], + }, + }, + ]; + + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "user-input.requested"), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, [ + { requestID: requestId, reply: "once" }, + ]); + NodeAssert.equal( + events.some((event) => event.type === "request.opened"), + false, + ); + NodeAssert.equal( + events.some((event) => event.type === "request.resolved"), + false, + ); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("surfaces the approval when the full-access auto-reply fails", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-full-access-reply-failed"); + runtimeMock.state.permissionReplyImplementation = async () => { + throw new Error("reply failed"); + }; + runtimeMock.state.subscribedEvents = [ + { + id: "evt-doom-loop", + type: "permission.asked", + properties: { + id: "per_doom_loop_failed", + sessionID: "http://127.0.0.1:9999/session", + permission: "doom_loop", + patterns: ["bash"], + metadata: {}, + always: [], + }, + }, + ]; + + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.take(1), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const opened = Option.getOrUndefined( + yield* Fiber.join(openedFiber).pipe(Effect.timeout("1 second")), + ); + NodeAssert.equal(opened?.requestId, "per_doom_loop_failed"); + // Exactly one auto-reply attempt: the fallback surfaces the dialog + // instead of retrying the reply. + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, [ + { requestID: "per_doom_loop_failed", reply: "once" }, + ]); + + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("does not reopen a failed full-access auto-reply after its terminal reply", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-full-access-reply-failed-after-terminal"); + const childId = "ses_full_access_terminal_child"; + const request = permissionRequest("per_failed_after_terminal", childId); + const ancestryAttempted = promiseWithResolvers(); + const releaseReply = promiseWithResolvers(); + // The ask arrives from a child whose ancestry lookup is failing, so it + // is handled on a retry fiber. The terminal reply lands while that + // fiber's auto-reply is still in flight; the reply then fails. The + // request must neither reopen nor emit a stray resolution. + runtimeMock.state.sessionParentById.set(childId, "http://127.0.0.1:9999/session"); + runtimeMock.state.transientErrorSessionIds.add(childId); + runtimeMock.state.sessionGetObserved = (sessionID) => { + if (sessionID === childId) { + ancestryAttempted.resolve(undefined); + } + }; + runtimeMock.state.permissionReplyImplementation = async () => { + await releaseReply.promise; + throw new Error("reply failed"); + }; + const terminalEvent = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + { id: "evt-ask", type: "permission.asked", properties: request }, + terminalEvent.promise, + ]; + + const requestEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "request.opened" || event.type === "request.resolved"), + ), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + yield* Effect.promise(() => ancestryAttempted.promise); + runtimeMock.state.transientErrorSessionIds.delete(childId); + yield* advanceTestClock(250); + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, [ + { requestID: request.id, reply: "once" }, + ]); + + // Drain the microtask queue so the pump has consumed the terminal reply + // before the in-flight auto-reply is allowed to fail. + terminalEvent.resolve({ + id: "evt-reply", + type: "permission.replied", + properties: { sessionID: childId, requestID: request.id, reply: "once" }, + }); + yield* Effect.promise(() => new Promise((resolve) => setImmediate(resolve))); + releaseReply.resolve(undefined); + yield* advanceTestClock(250); + + NodeAssert.equal(requestEventsFiber.pollUnsafe(), undefined); + yield* Fiber.interrupt(requestEventsFiber); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("routes child-session questions and replies through the parent thread", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-child-question"); + const questionReply = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [ + { + id: "evt-child-created", + type: "session.created", + properties: { + sessionID: "ses_child_question", + info: { + id: "ses_child_question", + parentID: "http://127.0.0.1:9999/session", + title: "Child session", }, }, }, @@ -2739,53 +3457,68 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); - it.effect("retries ancestry for one live child request after a transient failure", () => - Effect.gen(function* () { - const adapter = yield* OpenCodeAdapter; - const threadId = asThreadId("thread-child-request-ancestry-retry"); - const parentId = "http://127.0.0.1:9999/session"; - const ancestryAttempted = promiseWithResolvers(); - runtimeMock.state.sessionParentById.set("ses_existing_child", parentId); - runtimeMock.state.transientErrorSessionIds.add("ses_existing_child"); - runtimeMock.state.sessionGetObserved = (sessionID) => { - if (sessionID === "ses_existing_child") { - ancestryAttempted.resolve(undefined); + it.effect.each(["failure", "timeout"] as const)( + "retries ancestry for a child request after a transient %s", + (lookupFailure) => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId(`thread-child-request-ancestry-retry-${lookupFailure}`); + const parentId = "http://127.0.0.1:9999/session"; + const ancestryAttempted = promiseWithResolvers(); + runtimeMock.state.sessionParentById.set("ses_existing_child", parentId); + runtimeMock.state.transientErrorSessionIds.add("ses_existing_child"); + let lookupSignal: AbortSignal | undefined; + if (lookupFailure === "timeout") { + runtimeMock.state.sessionGetImplementation = async (_sessionID, signal) => { + lookupSignal = signal; + await new Promise(() => {}); + }; } - }; - runtimeMock.state.subscribedEvents = [ - { - id: "evt-existing-child-permission", - type: "permission.asked", - properties: permissionRequest("per_retry", "ses_existing_child"), - }, - ]; + runtimeMock.state.sessionGetObserved = (sessionID) => { + if (sessionID === "ses_existing_child") { + ancestryAttempted.resolve(undefined); + } + }; + runtimeMock.state.subscribedEvents = [ + { + id: "evt-existing-child-permission", + type: "permission.asked", + properties: permissionRequest("per_retry", "ses_existing_child"), + }, + ]; - const eventsFiber = yield* adapter.streamEvents.pipe( - Stream.filter( - (event) => - event.threadId === threadId && - (event.type === "runtime.warning" || event.type === "request.opened"), - ), - Stream.take(2), - Stream.runCollect, - Effect.forkChild, - ); - yield* adapter.startSession({ - provider: ProviderDriverKind.make("opencode"), - threadId, - runtimeMode: "approval-required", - }); - yield* Effect.promise(() => ancestryAttempted.promise); - runtimeMock.state.transientErrorSessionIds.delete("ses_existing_child"); - yield* advanceTestClock(250); + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "runtime.warning" || event.type === "request.opened"), + ), + Stream.take(2), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + }); + yield* Effect.promise(() => ancestryAttempted.promise); + if (lookupFailure === "timeout") { + yield* Effect.yieldNow; + yield* advanceTestClock(10_000); + NodeAssert.equal(lookupSignal?.aborted, true); + runtimeMock.state.sessionGetImplementation = null; + } + runtimeMock.state.transientErrorSessionIds.delete("ses_existing_child"); + yield* advanceTestClock(250); - const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); - NodeAssert.deepEqual( - events.map((event) => event.type), - ["runtime.warning", "request.opened"], - ); - yield* adapter.respondToRequest(threadId, ApprovalRequestId.make("per_retry"), "accept"); - }), + const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); + NodeAssert.deepEqual( + events.map((event) => event.type), + ["runtime.warning", "request.opened"], + ); + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make("per_retry"), "accept"); + }), ); it.effect("does not resurrect a recovered child request after its live reply", () => @@ -2840,7 +3573,8 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const response = yield* Effect.exit( adapter.respondToRequest(threadId, ApprovalRequestId.make(stale.id), "accept"), ); - NodeAssert.equal(Exit.isFailure(response), true); + NodeAssert.equal(Exit.isSuccess(response), true); + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, []); }), ); @@ -2892,7 +3626,8 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const response = yield* Effect.exit( adapter.respondToRequest(threadId, ApprovalRequestId.make(request.id), "accept"), ); - NodeAssert.equal(Exit.isFailure(response), true); + NodeAssert.equal(Exit.isSuccess(response), true); + NodeAssert.deepEqual(runtimeMock.state.permissionReplyCalls, []); }), ); @@ -3007,6 +3742,262 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("stops the full OpenCode child tree before it completes the interrupt", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-interrupt-child-tree"); + const parentAbortEvent = promiseWithResolvers(); + const markerEvent = promiseWithResolvers(); + const parentAbortStarted = promiseWithResolvers(); + const parentAbortRelease = promiseWithResolvers(); + const childAbortStarted = promiseWithResolvers(); + const childAbortRelease = promiseWithResolvers(); + const rootSessionId = "http://127.0.0.1:9999/session"; + runtimeMock.state.subscribedEvents = [parentAbortEvent.promise, markerEvent.promise]; + runtimeMock.state.sessionChildrenById.set(rootSessionId, [ + { id: "ses_child_a" }, + { id: "ses_child_b" }, + ]); + runtimeMock.state.sessionChildrenById.set("ses_child_a", [{ id: "ses_grandchild" }]); + runtimeMock.state.sessionChildrenById.set("ses_unrelated", [{ id: "ses_unrelated_child" }]); + runtimeMock.state.abortImplementation = async (sessionID) => { + if (sessionID === rootSessionId) { + parentAbortStarted.resolve(undefined); + await parentAbortRelease.promise; + } + if (sessionID === "ses_child_a") { + childAbortStarted.resolve(undefined); + await childAbortRelease.promise; + } + }; + + const markerFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => event.threadId === threadId && event.type === "thread.metadata.updated", + ), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Run child agents", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + + const interruptFiber = yield* adapter + .interruptTurn(threadId, turn.turnId) + .pipe(Effect.result, Effect.forkChild); + yield* Effect.promise(() => parentAbortStarted.promise); + runtimeMock.state.sessionChildrenById.get(rootSessionId)?.push({ id: "ses_late_child" }); + parentAbortEvent.resolve({ + id: "evt-parent-aborted", + type: "session.error", + properties: { + sessionID: rootSessionId, + error: { name: "MessageAbortedError", data: { message: "Aborted" } }, + }, + }); + markerEvent.resolve({ + id: "evt-after-parent-abort", + type: "session.updated", + properties: { info: { id: rootSessionId, title: "Parent abort received" } }, + }); + yield* Fiber.join(markerFiber); + + NodeAssert.equal(interruptFiber.pollUnsafe(), undefined); + yield* Effect.promise(() => childAbortStarted.promise); + NodeAssert.equal(interruptFiber.pollUnsafe(), undefined); + NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_unrelated"), false); + NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_unrelated_child"), false); + const sessionsDuringCleanup = yield* adapter.listSessions(); + const sessionDuringCleanup = sessionsDuringCleanup.find( + (candidate) => candidate.threadId === threadId, + ); + NodeAssert.equal(sessionDuringCleanup?.status, "running"); + NodeAssert.equal(sessionDuringCleanup?.activeTurnId, turn.turnId); + const nextTurnFiber = yield* adapter + .sendTurn({ + threadId, + input: "Start after every child stops", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }) + .pipe(Effect.forkChild); + yield* Effect.yieldNow; + NodeAssert.equal(runtimeMock.state.promptCalls.length, 1); + + childAbortRelease.resolve(undefined); + parentAbortRelease.resolve(undefined); + const result = yield* Fiber.join(interruptFiber); + const nextTurn = yield* Fiber.join(nextTurnFiber); + NodeAssert.equal(result._tag, "Success"); + NodeAssert.notEqual(nextTurn.turnId, turn.turnId); + NodeAssert.equal(runtimeMock.state.promptCalls.length, 2); + NodeAssert.equal(runtimeMock.state.abortCalls[0], rootSessionId); + NodeAssert.deepEqual( + new Set(runtimeMock.state.abortCalls.slice(1)), + new Set(["ses_child_a", "ses_child_b", "ses_grandchild", "ses_late_child"]), + ); + NodeAssert.deepEqual( + new Set(runtimeMock.state.sessionChildrenCalls), + new Set([rootSessionId, "ses_child_a", "ses_child_b", "ses_grandchild", "ses_late_child"]), + ); + const sessions = yield* adapter.listSessions(); + const session = sessions.find((candidate) => candidate.threadId === threadId); + NodeAssert.equal(session?.status, "running"); + NodeAssert.equal(session?.activeTurnId, nextTurn.turnId); + + runtimeMock.state.abortImplementation = null; + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("limits SDK requests across the full OpenCode child tree", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-interrupt-child-request-limit"); + const rootSessionId = "http://127.0.0.1:9999/session"; + const requestRelease = promiseWithResolvers(); + const limitReached = promiseWithResolvers(); + let inFlight = 0; + let maxInFlight = 0; + const holdRequest = async (result: T): Promise => { + inFlight += 1; + maxInFlight = Math.max(maxInFlight, inFlight); + if (inFlight === 8) { + limitReached.resolve(undefined); + } + await requestRelease.promise; + inFlight -= 1; + return result; + }; + + const children = Array.from({ length: 8 }, (_, index) => ({ id: `ses_child_${index}` })); + runtimeMock.state.sessionChildrenById.set(rootSessionId, children); + for (const child of children.slice(1)) { + runtimeMock.state.sessionChildrenById.set( + child.id, + Array.from({ length: 8 }, (_, index) => ({ id: `${child.id}_nested_${index}` })), + ); + } + runtimeMock.state.abortImplementation = async (sessionID) => { + if (sessionID.includes("_nested_")) { + await holdRequest(undefined); + } + }; + runtimeMock.state.sessionChildrenImplementation = async (sessionID) => { + if (sessionID === "ses_child_0") { + return await holdRequest([]); + } + return runtimeMock.state.sessionChildrenById.get(sessionID) ?? []; + }; + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Run a nested child tree", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + + const interruptFiber = yield* adapter + .interruptTurn(threadId, turn.turnId) + .pipe(Effect.forkChild); + yield* Effect.promise(() => limitReached.promise); + yield* Effect.yieldNow; + + NodeAssert.equal(inFlight, 8); + NodeAssert.equal(maxInFlight, 8); + + requestRelease.resolve(undefined); + yield* Fiber.join(interruptFiber); + + runtimeMock.state.abortImplementation = null; + runtimeMock.state.sessionChildrenImplementation = null; + runtimeMock.state.sessionChildrenById.clear(); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("attempts every child abort and fails the interrupt when one child abort fails", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-interrupt-child-failure"); + const rootSessionId = "http://127.0.0.1:9999/session"; + const failingChildStarted = promiseWithResolvers(); + const failingChildRelease = promiseWithResolvers(); + const siblingAbortStarted = promiseWithResolvers(); + runtimeMock.state.sessionChildrenById.set(rootSessionId, [ + { id: "ses_failing_child" }, + { id: "ses_surviving_sibling" }, + ]); + runtimeMock.state.abortImplementation = async (sessionID) => { + if (sessionID === "ses_failing_child") { + failingChildStarted.resolve(undefined); + await failingChildRelease.promise; + throw new Error("child abort failed"); + } + if (sessionID === "ses_surviving_sibling") { + siblingAbortStarted.resolve(undefined); + } + }; + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Run child agents", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + + const interruptFiber = yield* adapter + .interruptTurn(threadId, turn.turnId) + .pipe(Effect.result, Effect.forkChild); + yield* Effect.promise(() => failingChildStarted.promise); + yield* Effect.promise(() => siblingAbortStarted.promise); + NodeAssert.equal(interruptFiber.pollUnsafe(), undefined); + failingChildRelease.resolve(undefined); + const result = yield* Fiber.join(interruptFiber); + + NodeAssert.equal(result._tag, "Failure"); + if (result._tag === "Failure") { + NodeAssert.equal(result.failure._tag, "ProviderAdapterRequestError"); + NodeAssert.equal(result.failure.detail, "child abort failed"); + } + NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_failing_child"), true); + NodeAssert.equal(runtimeMock.state.abortCalls.includes("ses_surviving_sibling"), true); + const sessions = yield* adapter.listSessions(); + const session = sessions.find((candidate) => candidate.threadId === threadId); + NodeAssert.equal(session?.status, "running"); + NodeAssert.equal(session?.activeTurnId, turn.turnId); + + runtimeMock.state.abortImplementation = null; + yield* adapter.stopSession(threadId); + }), + ); + it.effect("keeps an idle event from completing a turn while its abort request is pending", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; @@ -4286,11 +5277,20 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { const threadId = asThreadId("thread-interrupt-provider-error"); const errorEvent = promiseWithResolvers(); const abortStarted = promiseWithResolvers(); - const abortRelease = promiseWithResolvers(); + const childAbortStarted = promiseWithResolvers(); + const childAbortRelease = promiseWithResolvers(); + const rootSessionId = "http://127.0.0.1:9999/session"; runtimeMock.state.subscribedEvents = [errorEvent.promise]; - runtimeMock.state.abortImplementation = async () => { - abortStarted.resolve(undefined); - await abortRelease.promise; + runtimeMock.state.sessionChildrenById.set(rootSessionId, [{ id: "ses_error_child" }]); + runtimeMock.state.abortImplementation = async (sessionID) => { + if (sessionID === rootSessionId) { + abortStarted.resolve(undefined); + await new Promise(() => {}); + } + if (sessionID === "ses_error_child") { + childAbortStarted.resolve(undefined); + await childAbortRelease.promise; + } }; const eventsFiber = yield* adapter.streamEvents.pipe( @@ -4321,16 +5321,14 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { id: "evt-provider-error-after-stop", type: "session.error", properties: { - sessionID: "http://127.0.0.1:9999/session", + sessionID: rootSessionId, error: { name: "APIError", data: { message: "Upstream failed", isRetryable: false }, }, }, }); - yield* Effect.yieldNow; - abortRelease.resolve(undefined); - yield* Fiber.join(interruptFiber); + yield* Effect.promise(() => childAbortStarted.promise); const events = Array.from(yield* Fiber.join(eventsFiber).pipe(Effect.timeout("1 second"))); NodeAssert.deepEqual( @@ -4349,7 +5347,41 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { failed?.type === "turn.completed" ? failed.payload.state : undefined, "failed", ); + const sessionsDuringCleanup = yield* adapter.listSessions(); + const sessionDuringCleanup = sessionsDuringCleanup.find( + (candidate) => candidate.threadId === threadId, + ); + NodeAssert.equal(sessionDuringCleanup?.status, "error"); + NodeAssert.equal(sessionDuringCleanup?.activeTurnId, undefined); + + const secondInterruptFiber = yield* adapter.interruptTurn(threadId).pipe(Effect.forkChild); + const nextTurnFiber = yield* adapter + .sendTurn({ + threadId, + input: "Start after child cleanup", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }) + .pipe(Effect.forkChild); + yield* Effect.yieldNow; + NodeAssert.equal( + runtimeMock.state.abortCalls.filter((sessionID) => sessionID === rootSessionId).length, + 1, + ); + NodeAssert.equal(secondInterruptFiber.pollUnsafe(), undefined); + NodeAssert.equal(nextTurnFiber.pollUnsafe(), undefined); + NodeAssert.equal(runtimeMock.state.promptCalls.length, 1); + + childAbortRelease.resolve(undefined); + yield* Fiber.join(interruptFiber); + yield* Fiber.join(secondInterruptFiber); + const nextTurn = yield* Fiber.join(nextTurnFiber); + NodeAssert.notEqual(nextTurn.turnId, turn.turnId); + NodeAssert.equal(runtimeMock.state.promptCalls.length, 2); + runtimeMock.state.abortImplementation = null; yield* adapter.stopSession(threadId); }), ); @@ -4720,6 +5752,260 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect("maps native task progress only while a turn is active", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-native-progress"); + const sessionID = "http://127.0.0.1:9999/session"; + const startProgress = promiseWithResolvers(); + const finishTurn = promiseWithResolvers(); + const lateProgress = promiseWithResolvers(); + const todos = [ + { content: "Read files", status: "completed", priority: "high" }, + { content: "Fix OpenCode", status: "in_progress", priority: "high" }, + { content: "Run tests", status: "pending", priority: "medium" }, + { content: "Old task", status: "cancelled", priority: "low" }, + { content: "Unknown task", status: "future-status", priority: "low" }, + ]; + const todoEvent = { + id: "evt-todos", + type: "todo.updated", + properties: { sessionID, todos }, + } satisfies OpenCodeEvent; + runtimeMock.state.subscribedEvents = [ + startProgress.promise, + { ...todoEvent, id: "evt-duplicate-todos" }, + { + ...todoEvent, + id: "evt-child-todos", + properties: { + sessionID: "child-session", + todos: [{ content: "Child task", status: "pending", priority: "low" }], + }, + }, + ...["todowrite", "bash"].map( + (tool) => + ({ + id: `evt-${tool}`, + type: "message.part.updated", + properties: { + sessionID, + time: 2, + part: { + id: `part-${tool}`, + sessionID, + messageID: "msg-tools", + type: "tool", + callID: `call-${tool}`, + tool, + state: { + status: "completed", + input: tool === "bash" ? { command: "pwd" } : { todos }, + output: tool === "bash" ? "/repo\n" : "Tasks updated", + title: tool === "bash" ? "Working directory" : "Tasks updated", + metadata: {}, + time: { start: 1, end: 2 }, + }, + }, + }, + }) satisfies OpenCodeEvent, + ), + finishTurn.promise, + lateProgress.promise, + { id: "evt-progress-drained", type: "session.compacted", properties: { sessionID } }, + ]; + const eventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter( + (event) => + event.threadId === threadId && + (event.type === "turn.plan.updated" || event.type === "item.completed"), + ), + Stream.take(3), + Stream.runCollect, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Work through the task list", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + startProgress.resolve(todoEvent); + const events = yield* Fiber.join(eventsFiber); + const plan = events.find((event) => event.type === "turn.plan.updated"); + NodeAssert.equal(plan?.turnId, turn.turnId); + NodeAssert.deepEqual(plan?.payload.plan, [ + { step: "Read files", status: "completed" }, + { step: "Fix OpenCode", status: "inProgress" }, + { step: "Run tests", status: "pending" }, + { step: "Unknown task", status: "pending" }, + ]); + const tools = events.filter((event) => event.type === "item.completed"); + NodeAssert.equal(tools[0]?.payload.itemType, "dynamic_tool_call"); + NodeAssert.equal(tools[1]?.payload.title, "Working directory"); + NodeAssert.partialDeepStrictEqual(tools[1]?.payload.data, { + command: "pwd", + result: "/repo\n", + }); + const completedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.runHead, + Effect.forkChild, + ); + finishTurn.resolve({ + id: "evt-progress-completed", + type: "session.status", + properties: { sessionID, status: { type: "idle" } }, + }); + yield* Fiber.join(completedFiber); + const lateEventsFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "thread.state.changed"), + Stream.runCollect, + Effect.forkChild, + ); + lateProgress.resolve({ ...todoEvent, id: "evt-late-todos" }); + NodeAssert.deepEqual( + (yield* Fiber.join(lateEventsFiber)).map((event) => event.type), + ["thread.state.changed"], + ); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("warns on disconnection and recovers a completion missed during reconnect", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-reconnect-completion"); + const reconnect = promiseWithResolvers(); + runtimeMock.state.subscribedEvents = [reconnect.promise]; + runtimeMock.state.sessionStatus = "busy"; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Work", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const warningFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "runtime.warning"), + Stream.runHead, + Effect.forkChild, + ); + runtimeMock.state.eventStreamError?.(new Error("socket closed")); + const warning = Option.getOrThrow(yield* Fiber.join(warningFiber)); + NodeAssert.ok(warning.type === "runtime.warning"); + NodeAssert.equal(warning.payload.message, "OpenCode connection lost. Reconnecting."); + const completedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "turn.completed"), + Stream.runHead, + Effect.forkChild, + ); + runtimeMock.state.sessionStatus = "idle"; + reconnect.resolve({ + id: "evt-reconnected", + type: "server.connected", + properties: {}, + } satisfies OpenCodeEvent); + NodeAssert.equal(Option.getOrThrow(yield* Fiber.join(completedFiber)).turnId, turn.turnId); + NodeAssert.equal( + (yield* adapter.listSessions()).find((session) => session.threadId === threadId)?.status, + "ready", + ); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect( + "ends a running session on clean stream closure without discarding unresolved permissions", + () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-stream-closed"); + const endStream = promiseWithResolvers(); + const request = permissionRequest("per_disconnect", "http://127.0.0.1:9999/session"); + runtimeMock.state.pendingPermissions = [request]; + runtimeMock.state.subscribedEvents = [endStream.promise]; + const openedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.runHead, + Effect.forkChild, + ); + const session = yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + }); + yield* Fiber.join(openedFiber); + yield* adapter.sendTurn({ + threadId, + input: "Work", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }); + const exitedFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "session.exited"), + Stream.runCollect, + Effect.forkChild, + ); + runtimeMock.state.endEventStream = true; + runtimeMock.state.abortImplementation = async () => { + throw new Error("server unreachable"); + }; + endStream.resolve({ + id: "evt-busy", + type: "session.status", + properties: { sessionID: request.sessionID, status: { type: "busy" } }, + }); + const exited = yield* Fiber.join(exitedFiber); + NodeAssert.equal( + exited.some((event) => event.type === "request.resolved"), + false, + ); + NodeAssert.match( + exited.find((event) => event.type === "runtime.error")?.payload.message ?? "", + /event stream ended/, + ); + NodeAssert.equal(yield* adapter.hasSession(threadId), false); + runtimeMock.state.endEventStream = false; + runtimeMock.state.subscribedEvents = []; + runtimeMock.state.abortImplementation = null; + const recoveredFiber = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId && event.type === "request.opened"), + Stream.runHead, + Effect.forkChild, + ); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "approval-required", + resumeCursor: session.resumeCursor, + }); + NodeAssert.equal( + Option.getOrThrow(yield* Fiber.join(recoveredFiber)).requestId, + request.id, + ); + yield* adapter.respondToRequest(threadId, ApprovalRequestId.make(request.id), "accept"); + yield* adapter.stopSession(threadId); + }), + ); + it.effect("lets OpenCode own session title generation and emits title metadata updates", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 580d99b2f..6d7a3cd8a 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -58,6 +58,8 @@ import { } from "../opencodeRuntime.ts"; import * as Option from "effect/Option"; +const encodePlanFingerprint = Schema.encodeSync(Schema.fromJsonString(Schema.Unknown)); + const PROVIDER = ProviderDriverKind.make("opencode"); /** @@ -193,8 +195,10 @@ const decodeOpenCodeSessionStatusMap = Schema.decodeUnknownOption(OpenCodeSessio interface OpenCodeCancellation { readonly turnId: TurnId | undefined; + readonly acknowledgment: Deferred.Deferred; readonly completion: Deferred.Deferred; acknowledged?: boolean; + turnSettled?: boolean; deferredIdleEvent?: OpenCodeSessionStatusEvent; } @@ -329,6 +333,7 @@ interface OpenCodeSessionContext { readonly openCodeSessionId: string; readonly relatedSessionIds: Set; readonly resolvedRequestIds: Set; + readonly autoRepliedRequestIds: Set; readonly emittedTerminalRequestIds: Set; readonly requestRelationRetries: Map; readonly pendingPermissions: Map; @@ -341,6 +346,7 @@ interface OpenCodeSessionContext { activeTurnId: TurnId | undefined; activeAgent: string | undefined; activeVariant: string | undefined; + lastPlanFingerprint: string | undefined; cancellation: OpenCodeCancellation | undefined; interruptedTurnId: TurnId | undefined; reconcileIdleStatus: boolean; @@ -416,6 +422,9 @@ type EventBaseInput = { function toToolLifecycleItemType(toolName: string): ToolLifecycleItemType { const normalized = toolName.toLowerCase(); + if (normalized === "todowrite" || normalized === "todoread") { + return "dynamic_tool_call"; + } if (normalized.includes("bash") || normalized.includes("command")) { return "command_execution"; } @@ -448,16 +457,15 @@ function toToolLifecycleItemType(toolName: string): ToolLifecycleItemType { function mapPermissionToRequestType( permission: string, -): "command_execution_approval" | "file_read_approval" | "file_change_approval" | "unknown" { +): "command_execution_approval" | "file_read_approval" | "file_change_approval" { switch (permission) { - case "bash": - return "command_execution_approval"; case "read": return "file_read_approval"; case "edit": return "file_change_approval"; default: - return "unknown"; + // Every OpenCode permission needs an actionable approval in each client. + return "command_execution_approval"; } } @@ -705,10 +713,88 @@ const failPendingOpenCodeCancellation = Effect.fn("failPendingOpenCodeCancellati ).pipe(Effect.ignore); }); -const abortOpenCodeSessionForTeardown = (context: OpenCodeSessionContext) => - runOpenCodeSdk("session.abort", (signal) => +const abortOpenCodeDescendants = Effect.fn("abortOpenCodeDescendants")(function* ( + context: OpenCodeSessionContext, +) { + const visited = new Set([context.openCodeSessionId]); + const requestSemaphore = Semaphore.makeUnsafe(8); + + const visit = ( + sessionId: string, + abortSession: boolean, + ): Effect.Effect => + Effect.gen(function* () { + let firstFailure: OpenCodeRuntimeError | undefined; + if (abortSession) { + const abortResult = yield* requestSemaphore + .withPermit( + runOpenCodeSdk("session.abort", (signal) => + context.client.session.abort({ sessionID: sessionId }, { signal }), + ), + ) + .pipe( + Effect.catchIf( + (cause) => isOpenCodeNotFound(cause), + () => Effect.void, + ), + Effect.result, + ); + if (abortResult._tag === "Failure") { + firstFailure = abortResult.failure; + } + } + + const childrenResult = yield* requestSemaphore + .withPermit( + runOpenCodeSdk("session.children", (signal) => + context.client.session.children({ sessionID: sessionId }, { signal }), + ), + ) + .pipe( + Effect.catchIf( + (cause) => isOpenCodeNotFound(cause), + () => Effect.void, + ), + Effect.result, + ); + if (childrenResult._tag === "Failure") { + return firstFailure ?? childrenResult.failure; + } + + const children = childrenResult.success?.data ?? []; + const newChildren = children.filter((child) => { + if (visited.has(child.id)) { + return false; + } + visited.add(child.id); + return true; + }); + const childFailures = yield* Effect.forEach(newChildren, (child) => visit(child.id, true), { + concurrency: 8, + }); + firstFailure ??= childFailures.find((failure) => failure !== undefined); + return firstFailure; + }); + + const firstFailure = yield* visit(context.openCodeSessionId, false); + if (firstFailure) { + return yield* firstFailure; + } +}); + +const abortOpenCodeSessionForTeardown = Effect.fn("abortOpenCodeSessionForTeardown")(function* ( + context: OpenCodeSessionContext, +) { + // Stop the parent before the snapshot so it cannot add another child after + // the adapter reads the tree. + yield* runOpenCodeSdk("session.abort", (signal) => context.client.session.abort({ sessionID: context.openCodeSessionId }, { signal }), ).pipe(Effect.timeout("1 second"), Effect.ignore({ log: true })); + yield* abortOpenCodeDescendants(context).pipe( + Effect.timeout("1 second"), + Effect.ignore({ log: true }), + ); +}); const cancelPendingOpenCodePrompt = Effect.fn("cancelPendingOpenCodePrompt")(function* ( context: OpenCodeSessionContext, @@ -929,6 +1015,18 @@ export function makeOpenCodeAdapter( runtimeEvents, sessionIncarnationId === undefined ? event : { ...event, sessionIncarnationId }, ).pipe(Effect.asVoid); + // Synchronous publish for callers that must not yield between a state + // check and the enqueue, e.g. reopening an approval only if its terminal + // event has not landed yet. Stamps the incarnation exactly like `emit`. + const emitUnsafe = ( + sessionIncarnationId: ProviderSession["sessionIncarnationId"], + event: ProviderRuntimeEvent, + ) => { + Queue.offerUnsafe( + runtimeEvents, + sessionIncarnationId === undefined ? event : { ...event, sessionIncarnationId }, + ); + }; const writeNativeEvent = ( threadId: ThreadId, event: { @@ -983,6 +1081,10 @@ export function makeOpenCodeAdapter( context.interruptedTurnId = undefined; context.awaitingBusyAfterInterruption = false; context.reconcileIdleStatus = false; + for (const requestId of context.autoRepliedRequestIds) { + context.emittedTerminalRequestIds.add(requestId); + } + context.autoRepliedRequestIds.clear(); applyProviderSessionUpdate( context, { status: "ready" }, @@ -992,6 +1094,7 @@ export function makeOpenCodeAdapter( if (pendingIdleReconciliation?.fiber) { yield* Fiber.interrupt(pendingIdleReconciliation.fiber); } + yield* schedulePendingRequestRecovery(context); yield* emit(context.sessionIncarnationId, { ...(yield* buildEventBase({ threadId: context.session.threadId, @@ -1346,6 +1449,7 @@ export function makeOpenCodeAdapter( { clearActiveTurnId: true, clearLastError: true }, ); } + yield* clearPendingOpenCodeRequests(context, { type: "session.abort" }); yield* emit(context.sessionIncarnationId, { ...(yield* buildEventBase({ threadId: context.session.threadId, @@ -1505,7 +1609,19 @@ export function makeOpenCodeAdapter( const seen = new Set(); const getSession = (sessionID: string) => - runOpenCodeSdk("session.get", () => context.client.session.get({ sessionID })).pipe( + runOpenCodeSdk("session.get", (signal) => + context.client.session.get({ sessionID }, { signal }), + ).pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + new OpenCodeRuntimeError({ + operation: "session.get", + detail: "OpenCode session ancestry lookup did not complete within 10 seconds.", + }), + ), + }), Effect.catchIf( (cause) => isOpenCodeNotFound(cause), () => Effect.succeed(undefined), @@ -1537,6 +1653,80 @@ export function makeOpenCodeAdapter( return false; }); + const openPermissionRequest = Effect.fn("openPermissionRequest")(function* ( + context: OpenCodeSessionContext, + request: PermissionRequest, + raw: unknown, + ) { + const base = yield* buildEventBase({ + threadId: context.session.threadId, + turnId: context.activeTurnId, + requestId: request.id, + raw, + }); + const stopped = yield* Ref.get(context.stopped); + if ( + stopped || + context.emittedTerminalRequestIds.has(request.id) || + context.pendingPermissions.has(request.id) + ) { + return; + } + const patterns = request.patterns.filter((pattern) => pattern !== "*"); + const detail = + request.permission === "bash" && patterns.length > 0 + ? patterns.join("\n") + : [request.permission.replaceAll("_", " "), ...patterns].join("\n"); + context.autoRepliedRequestIds.delete(request.id); + context.pendingPermissions.set(request.id, request); + emitUnsafe(context.sessionIncarnationId, { + ...base, + type: "request.opened", + payload: { + requestType: mapPermissionToRequestType(request.permission), + detail: `${detail}\n\nAllow for workspace also permits matching requests in other OpenCode sessions in this workspace.`, + args: request.metadata, + options: [ + { decision: "accept", label: "Allow once" }, + { + decision: "acceptForSession", + label: "Allow for workspace", + }, + { decision: "decline", label: "Deny" }, + ], + }, + }); + }); + + // Full access means the user already granted everything, but two upstream + // paths never consult the session ruleset we send: doom-loop detection + // (evaluated against the agent ruleset only) and subagent sessions (which + // keep only deny and external-directory rules). Answer those asks here. + // + // Reply "once", not "always": OpenCode stores "always" grants per + // directory, so on a shared external server an "always" from a full-access + // thread would silently widen what a supervised thread on the same + // directory is allowed to do. + const autoReplyFullAccess = Effect.fn("autoReplyFullAccess")(function* ( + context: OpenCodeSessionContext, + request: PermissionRequest, + raw: unknown, + ) { + const replied = yield* runOpenCodeSdk("permission.reply", (signal) => + context.client.permission.reply({ requestID: request.id, reply: "once" }, { signal }), + ).pipe( + Effect.timeout("10 seconds"), + Effect.as(true), + Effect.orElseSucceed(() => false), + ); + if (!replied) { + // Fall back to the dialog. The id stays resolved so a recovered copy + // of this ask cannot reopen after the user answers; + // `pendingPermissions` gates re-asks while the dialog is open. + yield* openPermissionRequest(context, request, raw); + } + }); + const emitPendingOpenCodeRequest = Effect.fn("emitPendingOpenCodeRequest")(function* ( context: OpenCodeSessionContext, event: OpenCodeAskedRequestEvent, @@ -1545,26 +1735,26 @@ export function makeOpenCodeAdapter( if (context.resolvedRequestIds.has(event.properties.id)) { return; } + if (context.activeTurnId === undefined && context.reconcileIdleStatus) { + context.resolvedRequestIds.add(event.properties.id); + return; + } if (event.type === "permission.asked") { const request = event.properties; if (context.pendingPermissions.has(request.id)) { return; } - context.pendingPermissions.set(request.id, request); - yield* emit(context.sessionIncarnationId, { - ...(yield* buildEventBase({ - threadId: context.session.threadId, - turnId: context.activeTurnId, - requestId: request.id, - raw, - })), - type: "request.opened", - payload: { - requestType: mapPermissionToRequestType(request.permission), - detail: request.patterns.length > 0 ? request.patterns.join("\n") : request.permission, - args: request.metadata, - }, - }); + if (context.session.runtimeMode === "full-access") { + // Reply outside the event pump so a slow HTTP response cannot hide + // progress, terminal replies, or the acknowledgment for Stop. + context.resolvedRequestIds.add(request.id); + context.autoRepliedRequestIds.add(request.id); + yield* autoReplyFullAccess(context, request, raw).pipe( + Effect.forkIn(context.sessionScope), + ); + return; + } + yield* openPermissionRequest(context, request, raw); return; } @@ -1572,14 +1762,19 @@ export function makeOpenCodeAdapter( if (context.pendingQuestions.has(request.id)) { return; } + const base = yield* buildEventBase({ + threadId: context.session.threadId, + turnId: context.activeTurnId, + requestId: request.id, + raw, + }); + const stopped = yield* Ref.get(context.stopped); + if (stopped || context.resolvedRequestIds.has(request.id)) { + return; + } context.pendingQuestions.set(request.id, request); - yield* emit(context.sessionIncarnationId, { - ...(yield* buildEventBase({ - threadId: context.session.threadId, - turnId: context.activeTurnId, - requestId: request.id, - raw, - })), + emitUnsafe(context.sessionIncarnationId, { + ...base, type: "user-input.requested", payload: { questions: normalizeQuestionRequest(request) }, }); @@ -1600,23 +1795,32 @@ export function makeOpenCodeAdapter( const emitTerminalOpenCodeRequest = Effect.fn("emitTerminalOpenCodeRequest")(function* ( context: OpenCodeSessionContext, event: OpenCodeTerminalRequestEvent, + raw: unknown = event, ) { const requestId = event.properties.requestID; if (context.emittedTerminalRequestIds.has(requestId)) { return; } + if (context.autoRepliedRequestIds.delete(requestId)) { + context.emittedTerminalRequestIds.add(requestId); + return; + } + const base = yield* buildEventBase({ + threadId: context.session.threadId, + turnId: context.activeTurnId, + requestId, + raw, + }); + if (context.emittedTerminalRequestIds.has(requestId)) return; context.emittedTerminalRequestIds.add(requestId); if (event.type === "permission.replied") { - yield* emit(context.sessionIncarnationId, { - ...(yield* buildEventBase({ - threadId: context.session.threadId, - turnId: context.activeTurnId, - requestId, - raw: event, - })), + const request = context.pendingPermissions.get(requestId); + context.pendingPermissions.delete(requestId); + emitUnsafe(context.sessionIncarnationId, { + ...base, type: "request.resolved", payload: { - requestType: "unknown", + requestType: request ? mapPermissionToRequestType(request.permission) : "unknown", decision: mapPermissionDecision(event.properties.reply), }, }); @@ -1624,6 +1828,7 @@ export function makeOpenCodeAdapter( } const request = context.pendingQuestions.get(requestId); + context.pendingQuestions.delete(requestId); const answers = event.type === "question.replied" && request ? Object.fromEntries( @@ -1633,18 +1838,77 @@ export function makeOpenCodeAdapter( ]), ) : {}; - yield* emit(context.sessionIncarnationId, { - ...(yield* buildEventBase({ - threadId: context.session.threadId, - turnId: context.activeTurnId, - requestId, - raw: event, - })), + emitUnsafe(context.sessionIncarnationId, { + ...base, type: "user-input.resolved", payload: { answers }, }); }); + const closePendingOpenCodeRequests = Effect.fn("closePendingOpenCodeRequests")(function* ( + context: OpenCodeSessionContext, + permissions: ReadonlyArray, + questions: ReadonlyArray, + raw: unknown, + ) { + for (const request of permissions) { + if (!context.pendingPermissions.has(request.id)) continue; + yield* resolvePendingOpenCodeRequest(context, request.id); + const base = yield* buildEventBase({ + threadId: context.session.threadId, + turnId: context.activeTurnId, + requestId: request.id, + raw, + }); + if (context.emittedTerminalRequestIds.has(request.id)) continue; + context.pendingPermissions.delete(request.id); + context.emittedTerminalRequestIds.add(request.id); + emitUnsafe(context.sessionIncarnationId, { + ...base, + type: "request.resolved", + payload: { requestType: mapPermissionToRequestType(request.permission) }, + }); + } + for (const request of questions) { + if (!context.pendingQuestions.has(request.id)) continue; + yield* resolvePendingOpenCodeRequest(context, request.id); + const base = yield* buildEventBase({ + threadId: context.session.threadId, + turnId: context.activeTurnId, + requestId: request.id, + raw, + }); + if (context.emittedTerminalRequestIds.has(request.id)) continue; + context.pendingQuestions.delete(request.id); + context.emittedTerminalRequestIds.add(request.id); + emitUnsafe(context.sessionIncarnationId, { + ...base, + type: "user-input.resolved", + payload: { answers: {} }, + }); + } + }); + + const clearPendingOpenCodeRequests = Effect.fn("clearPendingOpenCodeRequests")(function* ( + context: OpenCodeSessionContext, + raw: unknown, + ) { + context.pendingRequestRecovery = undefined; + for (const requestId of context.requestRelationRetries.keys()) { + yield* resolvePendingOpenCodeRequest(context, requestId); + } + for (const requestId of context.autoRepliedRequestIds) { + context.emittedTerminalRequestIds.add(requestId); + } + context.autoRepliedRequestIds.clear(); + yield* closePendingOpenCodeRequests( + context, + [...context.pendingPermissions.values()], + [...context.pendingQuestions.values()], + raw, + ); + }); + const scheduleRequestRelationRetry = Effect.fn("scheduleRequestRelationRetry")(function* ( context: OpenCodeSessionContext, event: OpenCodeRoutedRequestEvent, @@ -1732,10 +1996,21 @@ export function makeOpenCodeAdapter( const run = Effect.gen(function* () { let retryCount = 0; while (context.pendingRequestRecovery === recovery) { - const responses = yield* Effect.all({ - permissions: runOpenCodeSdk("permission.list", () => context.client.permission.list()), - questions: runOpenCodeSdk("question.list", () => context.client.question.list()), - }).pipe( + // Only requests pending before the snapshot can be closed by it. + const priorPermissions = [...context.pendingPermissions.values()]; + const priorQuestions = [...context.pendingQuestions.values()]; + const responses = yield* Effect.all( + { + permissions: runOpenCodeSdk("permission.list", (signal) => + context.client.permission.list(undefined, { signal }), + ), + questions: runOpenCodeSdk("question.list", (signal) => + context.client.question.list(undefined, { signal }), + ), + }, + { concurrency: 2 }, + ).pipe( + Effect.timeout("10 seconds"), Effect.match({ onFailure: (cause) => ({ type: "failure" as const, cause }), onSuccess: (value) => ({ type: "success" as const, value }), @@ -1779,6 +2054,14 @@ export function makeOpenCodeAdapter( yield* Effect.sleep(`${delayMs} millis`); continue; } + const permissionIds = new Set(permissions.map((request) => request.id)); + const questionIds = new Set(questions.map((request) => request.id)); + yield* closePendingOpenCodeRequests( + context, + priorPermissions.filter((request) => !permissionIds.has(request.id)), + priorQuestions.filter((request) => !questionIds.has(request.id)), + { type: "pending-requests.recovered" }, + ); yield* Effect.forEach( permissions, (request) => @@ -1848,6 +2131,9 @@ export function makeOpenCodeAdapter( yield* schedulePendingRequestRecovery(context); if (!isFirstConnection) { yield* schedulePromptAdmissionRecovery(context, event); + if (context.activeTurnId !== undefined && context.promptAdmission === undefined) { + yield* scheduleIdleReconciliation(context, context.activeTurnId, event); + } } return; } @@ -1924,6 +2210,7 @@ export function makeOpenCodeAdapter( context.awaitingBusyAfterInterruption) && (event.type === "message.part.delta" || event.type === "message.part.updated" || + event.type === "todo.updated" || (event.type === "message.updated" && event.properties.info.role === "assistant")); if (suppressInterruptedParentOutput) { return; @@ -1950,6 +2237,25 @@ export function makeOpenCodeAdapter( break; } + case "session.compacted": { + // Surfaces OpenCode context compaction the same way Claude's + // compact_boundary does: ingestion turns it into a "Context + // compacted" row for the thread. + yield* emit(context.sessionIncarnationId, { + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + raw: event, + })), + type: "thread.state.changed", + payload: { + state: "compacted", + detail: event, + }, + }); + break; + } + case "message.updated": { const promptAdmission = context.promptAdmission; if ( @@ -1988,7 +2294,11 @@ export function makeOpenCodeAdapter( case "message.part.delta": { const existingPart = context.partById.get(event.properties.partID); - if (!existingPart) { + if ( + !existingPart || + (existingPart.type !== "text" && existingPart.type !== "reasoning") || + event.properties.field !== "text" + ) { break; } const role = messageRoleForPart(context, existingPart); @@ -2043,7 +2353,9 @@ export function makeOpenCodeAdapter( if (part.type === "tool") { const itemType = toToolLifecycleItemType(part.tool); const title = - part.state.status === "running" ? (part.state.title ?? part.tool) : part.tool; + part.state.status === "running" || part.state.status === "completed" + ? (part.state.title ?? part.tool) + : part.tool; const detail = detailFromToolPart(part); const payload = { itemType, @@ -2057,6 +2369,14 @@ export function makeOpenCodeAdapter( data: { tool: part.tool, state: part.state, + ...(typeof part.state.input.command === "string" + ? { command: part.state.input.command } + : {}), + ...(itemType === "file_change" ? { input: part.state.input } : {}), + ...(part.state.status === "completed" && + (itemType === "command_execution" || itemType === "mcp_tool_call") + ? { result: part.state.output } + : {}), }, }; const runtimeEvent: ProviderRuntimeEvent = { @@ -2087,7 +2407,6 @@ export function makeOpenCodeAdapter( } case "permission.replied": { - context.pendingPermissions.delete(event.properties.requestID); yield* emitTerminalOpenCodeRequest(context, event); break; } @@ -2099,18 +2418,46 @@ export function makeOpenCodeAdapter( case "question.replied": { yield* emitTerminalOpenCodeRequest(context, event); - context.pendingQuestions.delete(event.properties.requestID); break; } case "question.rejected": { - context.pendingQuestions.delete(event.properties.requestID); yield* emitTerminalOpenCodeRequest(context, event); break; } + case "todo.updated": { + if (turnId === undefined) break; + const plan = event.properties.todos + .filter((todo) => todo.status !== "cancelled") + .map((todo) => ({ + step: trimText(todo.content) ?? "Task", + status: + todo.status === "completed" + ? ("completed" as const) + : todo.status === "in_progress" + ? ("inProgress" as const) + : ("pending" as const), + })); + const fingerprint = encodePlanFingerprint([turnId, plan]); + const base = yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + raw: event, + }); + // Session-wide task updates must not reopen progress after a turn ends. + if (context.activeTurnId !== turnId || context.lastPlanFingerprint === fingerprint) break; + context.lastPlanFingerprint = fingerprint; + emitUnsafe(context.sessionIncarnationId, { + ...base, + type: "turn.plan.updated", + payload: { plan }, + }); + break; + } + case "session.status": { - if (event.properties.status.type === "busy") { + if (event.properties.status.type === "busy" || event.properties.status.type === "retry") { if (turnId === undefined) { break; } @@ -2135,7 +2482,7 @@ export function makeOpenCodeAdapter( })), type: "runtime.warning", payload: { - message: event.properties.status.message, + message: `OpenCode retry ${event.properties.status.attempt}: ${event.properties.status.message}`, detail: event.properties.status, }, }); @@ -2173,13 +2520,12 @@ export function makeOpenCodeAdapter( if (isOpenCodeAbortError(event.properties.error)) { if (cancellation !== undefined && cancellation.turnId === undefined) { cancellation.acknowledged = true; - context.cancellation = undefined; - context.reconcileIdleStatus = true; - yield* Deferred.succeed(cancellation.completion, undefined).pipe(Effect.ignore); + yield* Deferred.succeed(cancellation.acknowledgment, undefined).pipe(Effect.ignore); break; } if (activeTurnId !== undefined && cancellation?.turnId === activeTurnId) { - yield* interruptOpenCodeTurn(context, activeTurnId, event); + cancellation.acknowledged = true; + yield* Deferred.succeed(cancellation.acknowledgment, undefined).pipe(Effect.ignore); break; } if (context.interruptedTurnId !== undefined || context.reconcileIdleStatus) { @@ -2187,14 +2533,19 @@ export function makeOpenCodeAdapter( } } yield* cancelIdleReconciliation(context); - if (activeTurnId !== undefined && cancellation?.turnId === activeTurnId) { - context.cancellation = undefined; - yield* Deferred.succeed(cancellation.completion, undefined).pipe(Effect.ignore); + const terminalCancellation = + activeTurnId !== undefined && cancellation?.turnId === activeTurnId + ? cancellation + : undefined; + if (terminalCancellation) { + terminalCancellation.turnSettled = true; + terminalCancellation.acknowledged = true; } context.activeTurnId = undefined; context.activeAgent = undefined; context.activeVariant = undefined; context.reconcileIdleStatus = false; + yield* schedulePendingRequestRecovery(context); yield* updateProviderSession( context, { @@ -2229,6 +2580,11 @@ export function makeOpenCodeAdapter( detail: event.properties.error, }, }); + if (terminalCancellation) { + yield* Deferred.succeed(terminalCancellation.acknowledgment, undefined).pipe( + Effect.ignore, + ); + } break; } @@ -2243,9 +2599,29 @@ export function makeOpenCodeAdapter( // shutdown) and cancels the in-flight `event.subscribe` fetch so // the async iterable unwinds cleanly. const eventsAbortController = new AbortController(); - yield* Scope.addFinalizer( - context.sessionScope, - Effect.sync(() => eventsAbortController.abort()), + let lastStreamError: unknown; + let warnedAboutDisconnect = false; + const streamErrors = yield* Queue.unbounded(); + yield* Scope.addFinalizer(context.sessionScope, Queue.shutdown(streamErrors)); + yield* Stream.fromQueue(streamErrors).pipe( + Stream.runForEach((cause) => + Effect.gen(function* () { + if (warnedAboutDisconnect) return; + warnedAboutDisconnect = true; + yield* emit(context.sessionIncarnationId, { + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId: context.activeTurnId, + })), + type: "runtime.warning", + payload: { + message: "OpenCode connection lost. Reconnecting.", + detail: openCodeRuntimeErrorDetail(cause), + }, + }); + }), + ), + Effect.forkIn(context.sessionScope), ); // Fibers forked into `context.sessionScope` are interrupted @@ -2254,6 +2630,10 @@ export function makeOpenCodeAdapter( runOpenCodeSdk("event.subscribe", () => context.client.event.subscribe(undefined, { signal: eventsAbortController.signal, + onSseError: (cause) => { + lastStreamError = cause; + Queue.offerUnsafe(streamErrors, cause); + }, }), ), (subscription) => @@ -2265,7 +2645,13 @@ export function makeOpenCodeAdapter( detail: openCodeRuntimeErrorDetail(cause), cause, }), - ).pipe(Stream.runForEach((event) => handleSubscribedEvent(context, event))), + ).pipe( + Stream.runForEach((event) => { + if (event.type === "server.connected") lastStreamError = undefined; + if (event.type === "server.connected") warnedAboutDisconnect = false; + return handleSubscribedEvent(context, event); + }), + ), ).pipe( Effect.exit, Effect.flatMap((exit) => @@ -2275,12 +2661,14 @@ export function makeOpenCodeAdapter( if (eventsAbortController.signal.aborted || (yield* Ref.get(context.stopped))) { return; } - if (Exit.isFailure(exit)) { - yield* emitUnexpectedExit( - context, - openCodeRuntimeErrorDetail(Cause.squash(exit.cause)), - ); - } + yield* emitUnexpectedExit( + context, + Exit.isFailure(exit) + ? openCodeRuntimeErrorDetail(Cause.squash(exit.cause)) + : lastStreamError !== undefined + ? `OpenCode event stream disconnected: ${openCodeRuntimeErrorDetail(lastStreamError)}` + : "OpenCode event stream ended unexpectedly. Send another message to reconnect.", + ); }), ), Effect.forkIn(context.sessionScope), @@ -2299,6 +2687,12 @@ export function makeOpenCodeAdapter( Effect.forkIn(context.sessionScope), ); } + // Scope finalizers run in reverse order. Abort the pending read before + // interrupting the pump, whose iterator.return() waits for that read. + yield* Scope.addFinalizer( + context.sessionScope, + Effect.sync(() => eventsAbortController.abort()), + ); }); const startSession: OpenCodeAdapterShape["startSession"] = Effect.fn("startSession")( @@ -2485,6 +2879,7 @@ export function makeOpenCodeAdapter( openCodeSessionId: started.openCodeSession.id, relatedSessionIds: new Set([started.openCodeSession.id]), resolvedRequestIds: new Set(), + autoRepliedRequestIds: new Set(), emittedTerminalRequestIds: new Set(), requestRelationRetries: new Map(), pendingPermissions: new Map(), @@ -2497,6 +2892,7 @@ export function makeOpenCodeAdapter( activeTurnId: undefined, activeAgent: undefined, activeVariant: undefined, + lastPlanFingerprint: undefined, cancellation: undefined, interruptedTurnId: undefined, reconcileIdleStatus: false, @@ -2937,14 +3333,12 @@ export function makeOpenCodeAdapter( return; } const existingCancellation = context.cancellation; - if ( - existingCancellation !== undefined && - existingCancellation.turnId === interruptedTurnId - ) { + if (existingCancellation !== undefined) { return yield* Deferred.await(existingCancellation.completion); } const cancellation: OpenCodeCancellation = { turnId: interruptedTurnId, + acknowledgment: Deferred.makeUnsafe(), completion: Deferred.makeUnsafe(), }; context.cancellation = cancellation; @@ -2957,10 +3351,11 @@ export function makeOpenCodeAdapter( yield* Deferred.await(promptAdmission.submissionSettled); } - const abortOutcome = yield* Effect.raceFirst( + const parentAbortOutcome = yield* Effect.raceFirst( runOpenCodeSdk("session.abort", (signal) => context.client.session.abort({ sessionID: context.openCodeSessionId }, { signal }), ).pipe( + Effect.asVoid, Effect.timeout("10 seconds"), Effect.catchTags({ OpenCodeRuntimeError: (cause) => Effect.fail(toRequestError(cause)), @@ -2977,33 +3372,67 @@ export function makeOpenCodeAdapter( Effect.exit, Effect.map((exit) => ({ source: "request" as const, exit })), ), + Effect.raceFirst( + Deferred.await(cancellation.acknowledgment).pipe( + Effect.map(() => ({ source: "acknowledgment" as const })), + ), + Deferred.await(cancellation.completion).pipe( + Effect.exit, + Effect.map((exit) => ({ source: "completion" as const, exit })), + ), + ), + ); + if (parentAbortOutcome.source === "completion") { + return Exit.isFailure(parentAbortOutcome.exit) + ? yield* Effect.failCause(parentAbortOutcome.exit.cause) + : undefined; + } + const parentAbortExit = + parentAbortOutcome.source === "request" ? parentAbortOutcome.exit : Exit.void; + + const descendantAbortOutcome = yield* Effect.raceFirst( + abortOpenCodeDescendants(context).pipe( + Effect.timeout("10 seconds"), + Effect.catchTags({ + OpenCodeRuntimeError: (cause) => Effect.fail(toRequestError(cause)), + TimeoutError: (cause) => + Effect.fail( + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session.abort", + detail: "OpenCode child session cleanup did not complete within 10 seconds.", + cause, + }), + ), + }), + Effect.exit, + Effect.map((exit) => ({ source: "request" as const, exit })), + ), Deferred.await(cancellation.completion).pipe( Effect.exit, - Effect.map((exit) => ({ source: "event" as const, exit })), + Effect.map((exit) => ({ source: "completion" as const, exit })), ), ); - if (abortOutcome.source === "event") { - return Exit.isFailure(abortOutcome.exit) - ? yield* Effect.failCause(abortOutcome.exit.cause) + if (descendantAbortOutcome.source === "completion") { + return Exit.isFailure(descendantAbortOutcome.exit) + ? yield* Effect.failCause(descendantAbortOutcome.exit.cause) : undefined; } - const abortExit = abortOutcome.exit; - if (Exit.isFailure(abortExit)) { - if (interruptedTurnId && context.interruptedTurnId === interruptedTurnId) { - yield* Deferred.succeed(cancellation.completion, undefined).pipe(Effect.ignore); - return; - } - if (cancellation.turnId === undefined && cancellation.acknowledged) { - if (context.cancellation === cancellation) { - context.cancellation = undefined; - context.reconcileIdleStatus = true; - } - yield* Deferred.succeed(cancellation.completion, undefined).pipe(Effect.ignore); - return; - } + + const parentAbortFailed = Exit.isFailure(parentAbortExit) && !cancellation.acknowledged; + const failedExit = parentAbortFailed + ? parentAbortExit + : Exit.isFailure(descendantAbortOutcome.exit) + ? descendantAbortOutcome.exit + : undefined; + if (failedExit !== undefined && Exit.isFailure(failedExit)) { if (context.cancellation === cancellation) { context.cancellation = undefined; - if (cancellation.turnId !== undefined && cancellation.deferredIdleEvent) { + if ( + parentAbortFailed && + cancellation.turnId !== undefined && + cancellation.deferredIdleEvent + ) { yield* scheduleIdleReconciliation( context, cancellation.turnId, @@ -3011,16 +3440,19 @@ export function makeOpenCodeAdapter( ); } } - yield* Deferred.done(cancellation.completion, abortExit).pipe(Effect.ignore); - return yield* Effect.failCause(abortExit.cause); + yield* Deferred.done(cancellation.completion, failedExit).pipe(Effect.ignore); + return yield* Effect.failCause(failedExit.cause); } if (context.cancellation === cancellation) { - if (cancellation.turnId !== undefined) { + if (cancellation.turnSettled) { + context.cancellation = undefined; + } else if (cancellation.turnId !== undefined) { yield* interruptOpenCodeTurn(context, cancellation.turnId); } else { context.cancellation = undefined; context.reconcileIdleStatus = true; + yield* clearPendingOpenCodeRequests(context, { type: "session.abort" }); } } yield* Deferred.succeed(cancellation.completion, undefined).pipe(Effect.ignore); @@ -3031,20 +3463,52 @@ export function makeOpenCodeAdapter( "respondToRequest", )(function* (threadId, requestId, decision) { const context = yield* ensureSessionContext(sessions, threadId); - if (!context.pendingPermissions.has(requestId)) { + const request = context.pendingPermissions.get(requestId); + if (!request) { + if (context.emittedTerminalRequestIds.has(requestId)) return; return yield* new ProviderAdapterRequestError({ provider: PROVIDER, method: "permission.reply", - detail: `Unknown pending permission request: ${requestId}`, + detail: + context.pendingRequestRecovery || context.requestRelationRetries.has(requestId) + ? "OpenCode is still loading this permission request. Try again." + : `Unknown pending permission request: ${requestId}`, }); } - yield* runOpenCodeSdk("permission.reply", () => - context.client.permission.reply({ - requestID: requestId, - reply: toOpenCodePermissionReply(decision), + const reply = toOpenCodePermissionReply(decision); + yield* runOpenCodeSdk("permission.reply", (signal) => + context.client.permission.reply( + { + requestID: requestId, + reply, + }, + { signal }, + ), + ).pipe( + Effect.mapError(toRequestError), + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "permission.reply", + detail: "OpenCode permission reply did not complete within 10 seconds.", + }), + ), }), - ).pipe(Effect.mapError(toRequestError)); + ); + yield* resolvePendingOpenCodeRequest(context, requestId); + yield* emitTerminalOpenCodeRequest( + context, + { + id: `reply:${requestId}`, + type: "permission.replied", + properties: { sessionID: request.sessionID, requestID: requestId, reply }, + }, + { type: "permission.reply", requestID: requestId, reply }, + ); }); const respondToUserInput: OpenCodeAdapterShape["respondToUserInput"] = Effect.fn( @@ -3053,19 +3517,54 @@ export function makeOpenCodeAdapter( const context = yield* ensureSessionContext(sessions, threadId); const request = context.pendingQuestions.get(requestId); if (!request) { + if (context.emittedTerminalRequestIds.has(requestId)) return; return yield* new ProviderAdapterRequestError({ provider: PROVIDER, method: "question.reply", - detail: `Unknown pending user-input request: ${requestId}`, + detail: + context.pendingRequestRecovery || context.requestRelationRetries.has(requestId) + ? "OpenCode is still loading this question. Try again." + : `Unknown pending user-input request: ${requestId}`, }); } - yield* runOpenCodeSdk("question.reply", () => - context.client.question.reply({ - requestID: requestId, - answers: toOpenCodeQuestionAnswers(request, answers), + const questionAnswers = toOpenCodeQuestionAnswers(request, answers); + yield* runOpenCodeSdk("question.reply", (signal) => + context.client.question.reply( + { + requestID: requestId, + answers: questionAnswers, + }, + { signal }, + ), + ).pipe( + Effect.mapError(toRequestError), + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => + Effect.fail( + new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "question.reply", + detail: "OpenCode question reply did not complete within 10 seconds.", + }), + ), }), - ).pipe(Effect.mapError(toRequestError)); + ); + yield* resolvePendingOpenCodeRequest(context, requestId); + yield* emitTerminalOpenCodeRequest( + context, + { + id: `reply:${requestId}`, + type: "question.replied", + properties: { + sessionID: request.sessionID, + requestID: requestId, + answers: questionAnswers, + }, + }, + { type: "question.reply", requestID: requestId }, + ); }); const stopSession: OpenCodeAdapterShape["stopSession"] = Effect.fn("stopSession")( diff --git a/apps/server/src/provider/opencodeRuntime.environment.test.ts b/apps/server/src/provider/opencodeRuntime.environment.test.ts index 584a9d80f..680032de0 100644 --- a/apps/server/src/provider/opencodeRuntime.environment.test.ts +++ b/apps/server/src/provider/opencodeRuntime.environment.test.ts @@ -1,12 +1,24 @@ import type { OpencodeClient } from "@opencode-ai/sdk/v2"; +import * as NodeServices from "@effect/platform-node/NodeServices"; import { it as effectIt } from "@effect/vitest"; +import { + HostProcessEnvironment, + HostProcessExecutablePath, + HostProcessPlatform, +} from "@t3tools/shared/hostProcess"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Path from "effect/Path"; import * as TestClock from "effect/testing/TestClock"; +import { FetchHttpClient, HttpClient } from "effect/unstable/http"; import { describe, expect, it } from "vite-plus/test"; import { + OpenCodeRuntime, OpenCodeRuntimeError, + OpenCodeRuntimeLive, resolveOpenCodeConfigContent, resolveOpenCodeServerPassword, verifyOpenCodeServerVersion, @@ -150,3 +162,78 @@ describe("verifyOpenCodeServerVersion", () => { }).pipe(Effect.provide(TestClock.layer())), ); }); + +describe("OpenCode server output", () => { + effectIt.live( + "drains stdout and stderr after startup so server requests can finish", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const environment = yield* HostProcessEnvironment; + const executablePath = yield* HostProcessExecutablePath; + const platform = yield* HostProcessPlatform; + const tempDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-opencode-output-" }); + const isWindows = platform === "win32"; + const binaryPath = path.join(tempDir, isWindows ? "opencode.cmd" : "opencode"); + const scriptPath = path.join(tempDir, "opencode.mjs"); + + yield* fs.writeFileString( + scriptPath, + `import { createServer } from "node:http"; +const writeOutput = (stream) => new Promise((resolve, reject) => { + stream.write("x".repeat(2 * 1024 * 1024), (error) => error ? reject(error) : resolve()); +}); +const server = createServer(async (request, response) => { + if (request.url.startsWith("/global/health")) { + response.setHeader("Content-Type", "application/json"); + response.end(JSON.stringify({ healthy: true, version: "1.14.19" })); + return; + } + await Promise.all([writeOutput(process.stdout), writeOutput(process.stderr)]); + response.end("drained"); +}); +server.listen(0, "127.0.0.1", () => { + process.stdout.write("opencode server listening on http://127.0.0.1:" + server.address().port + "\\n"); +}); +`, + ); + yield* fs.writeFileString( + binaryPath, + [ + ...(isWindows ? ["@echo off"] : ["#!/bin/sh"]), + isWindows + ? '"%T3_TEST_NODE_BINARY%" "%T3_TEST_OPENCODE_SCRIPT%" %*' + : 'exec "$T3_TEST_NODE_BINARY" "$T3_TEST_OPENCODE_SCRIPT" "$@"', + "", + ].join("\n"), + ); + if (!isWindows) { + yield* fs.chmod(binaryPath, 0o755); + } + + const runtime = yield* OpenCodeRuntime; + const server = yield* runtime.startOpenCodeServerProcess({ + binaryPath, + directory: tempDir, + port: 0, + environment: { + ...environment, + T3_TEST_NODE_BINARY: executablePath, + T3_TEST_OPENCODE_SCRIPT: scriptPath, + }, + }); + const response = yield* HttpClient.get(`${server.url}/output`); + + expect(yield* response.text).toBe("drained"); + expect(yield* server.isRunning).toBe(true); + }).pipe( + Effect.scoped, + Effect.provide([ + OpenCodeRuntimeLive.pipe(Layer.provideMerge(NodeServices.layer)), + FetchHttpClient.layer, + ]), + ), + 10_000, + ); +}); diff --git a/apps/server/src/provider/opencodeRuntime.inventory.test.ts b/apps/server/src/provider/opencodeRuntime.inventory.test.ts index 2a878a24a..39ffec743 100644 --- a/apps/server/src/provider/opencodeRuntime.inventory.test.ts +++ b/apps/server/src/provider/opencodeRuntime.inventory.test.ts @@ -1,12 +1,14 @@ import * as NodeAssert from "node:assert/strict"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import type { OpencodeClient } from "@opencode-ai/sdk/v2"; +import { createOpencodeClient, type OpencodeClient } from "@opencode-ai/sdk/v2"; import { it } from "@effect/vitest"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as FileSystem from "effect/FileSystem"; import * as Layer from "effect/Layer"; import * as Path from "effect/Path"; +import * as Queue from "effect/Queue"; import { HostProcessEnvironment, HostProcessExecutablePath, @@ -18,6 +20,44 @@ import { OpenCodeRuntime, OpenCodeRuntimeLive } from "./opencodeRuntime.ts"; const testLayer = OpenCodeRuntimeLive.pipe(Layer.provideMerge(NodeServices.layer)); it.layer(testLayer)("OpenCodeRuntime inventory", (it) => { + it.effect("aborts pending SDK requests when inventory loading is interrupted", () => + Effect.gen(function* () { + const runtime = yield* OpenCodeRuntime; + const started = yield* Queue.make(); + const aborted = yield* Queue.make(); + const client = createOpencodeClient({ + baseUrl: "http://opencode.test", + fetch: Object.assign( + (input: string | Request | URL) => { + const request = input instanceof Request ? input : new Request(input.toString()); + return new Promise((_resolve, reject) => { + request.signal.addEventListener( + "abort", + () => { + Queue.offerUnsafe(aborted, new URL(request.url).pathname); + reject(request.signal.reason); + }, + { once: true }, + ); + Queue.offerUnsafe(started, undefined); + }); + }, + { preconnect: () => undefined }, + ), + }); + + const inventoryFiber = yield* runtime.loadOpenCodeInventory(client).pipe(Effect.forkChild); + yield* Queue.takeN(started, 3); + yield* Fiber.interrupt(inventoryFiber); + + NodeAssert.deepEqual((yield* Queue.takeAll(aborted)).toSorted(), [ + "/agent", + "/provider", + "/skill", + ]); + }), + ); + it.effect("keeps provider inventory when agent discovery fails", () => Effect.gen(function* () { const runtime = yield* OpenCodeRuntime; diff --git a/apps/server/src/provider/opencodeRuntime.permissions.test.ts b/apps/server/src/provider/opencodeRuntime.permissions.test.ts index a7837f670..a6ae1fbe0 100644 --- a/apps/server/src/provider/opencodeRuntime.permissions.test.ts +++ b/apps/server/src/provider/opencodeRuntime.permissions.test.ts @@ -1,5 +1,6 @@ import * as NodeAssert from "node:assert/strict"; +import * as RegExpUtils from "effect/RegExp"; import { describe, it } from "vite-plus/test"; import { buildOpenCodePermissionRules, toOpenCodePermissionReply } from "./opencodeRuntime.ts"; @@ -7,9 +8,14 @@ import { buildOpenCodePermissionRules, toOpenCodePermissionReply } from "./openc function actionFor( runtimeMode: Parameters[0], permission: string, + target = "*", ) { - return buildOpenCodePermissionRules(runtimeMode).find((rule) => rule.permission === permission) - ?.action; + // OpenCode uses the last matching rule. Its wildcards match directory separators. + return buildOpenCodePermissionRules(runtimeMode).findLast( + (rule) => + (rule.permission === "*" || rule.permission === permission) && + new RegExp(`^${RegExpUtils.escape(rule.pattern).replaceAll("\\*", ".*")}$`, "s").test(target), + )?.action; } describe("buildOpenCodePermissionRules", () => { @@ -27,12 +33,38 @@ describe("buildOpenCodePermissionRules", () => { NodeAssert.equal(actionFor("auto", "edit"), "ask"); }); - it("keeps asking for everything else in the auto modes", () => { - for (const runtimeMode of ["auto-accept-edits", "auto"] as const) { + it("allows workspace reads and task updates without asking in supervised modes", () => { + for (const runtimeMode of ["approval-required", "auto-accept-edits", "auto"] as const) { + for (const permission of ["read", "glob", "grep", "lsp", "skill", "todowrite"]) { + NodeAssert.equal(actionFor(runtimeMode, permission, "src/index.ts"), "allow"); + } + } + }); + + it("preserves OpenCode's environment-file approval rules", () => { + for (const runtimeMode of ["approval-required", "auto-accept-edits", "auto"] as const) { + for (const target of [ + ".env", + ".env.local", + "config/service.env", + "config/service.env.local", + ]) { + NodeAssert.equal(actionFor(runtimeMode, "read", target), "ask"); + } + for (const target of [".env.example", "config/service.env.example"]) { + NodeAssert.equal(actionFor(runtimeMode, "read", target), "allow"); + } + } + }); + + it("still asks before commands, network access, external directories and unknown tools", () => { + for (const runtimeMode of ["approval-required", "auto-accept-edits", "auto"] as const) { NodeAssert.equal(actionFor(runtimeMode, "bash"), "ask"); NodeAssert.equal(actionFor(runtimeMode, "webfetch"), "ask"); + NodeAssert.equal(actionFor(runtimeMode, "websearch"), "ask"); NodeAssert.equal(actionFor(runtimeMode, "external_directory"), "ask"); - NodeAssert.equal(actionFor(runtimeMode, "*"), "ask"); + NodeAssert.equal(actionFor(runtimeMode, "doom_loop"), "ask"); + NodeAssert.equal(actionFor(runtimeMode, "custom_tool"), "ask"); } }); @@ -45,14 +77,13 @@ describe("buildOpenCodePermissionRules", () => { }); describe("toOpenCodePermissionReply", () => { - it("maps every accept-family decision to an approval", () => { - NodeAssert.equal(toOpenCodePermissionReply("accept"), "once"); - NodeAssert.equal(toOpenCodePermissionReply("acceptForSession"), "always"); - NodeAssert.equal(toOpenCodePermissionReply("acceptAlways"), "always"); - }); - - it("still rejects the decline and cancel decisions", () => { - NodeAssert.equal(toOpenCodePermissionReply("decline"), "reject"); - NodeAssert.equal(toOpenCodePermissionReply("cancel"), "reject"); + it.each([ + ["accept", "once"], + ["acceptForSession", "always"], + ["acceptAlways", "always"], + ["decline", "reject"], + ["cancel", "reject"], + ] as const)("maps %s to %s", (decision, reply) => { + NodeAssert.equal(toOpenCodePermissionReply(decision), reply); }); }); diff --git a/apps/server/src/provider/opencodeRuntime.ts b/apps/server/src/provider/opencodeRuntime.ts index 8076fce25..05553fd89 100644 --- a/apps/server/src/provider/opencodeRuntime.ts +++ b/apps/server/src/provider/opencodeRuntime.ts @@ -80,6 +80,7 @@ export function resolveOpenCodeServerPassword( const OPENCODE_SERVER_READY_PREFIX = "opencode server listening"; const DEFAULT_OPENCODE_SERVER_TIMEOUT_MS = 30_000; const DEFAULT_HOSTNAME = "127.0.0.1"; +const OPENCODE_SERVER_STARTUP_MAX_OUTPUT_CHARS = 64 * 1024; const OPENCODE_SKILL_DISCOVERY_MAX_OUTPUT_BYTES = 8 * 1024 * 1024; export interface OpenCodeServerProcess { readonly url: string; @@ -486,8 +487,19 @@ export function buildOpenCodePermissionRules(runtimeMode: RuntimeMode): Permissi // reviewer, OpenCode among them, fall back to Supervised for that mode. const editAction = runtimeMode === "auto-accept-edits" ? "allow" : "ask"; + // Session rules override OpenCode's agent defaults. Allow reads and task + // updates, but keep its default approval rules for environment files. return [ { permission: "*", pattern: "*", action: "ask" }, + { permission: "read", pattern: "*", action: "allow" }, + { permission: "read", pattern: "*.env", action: "ask" }, + { permission: "read", pattern: "*.env.*", action: "ask" }, + { permission: "read", pattern: "*.env.example", action: "allow" }, + { permission: "glob", pattern: "*", action: "allow" }, + { permission: "grep", pattern: "*", action: "allow" }, + { permission: "lsp", pattern: "*", action: "allow" }, + { permission: "skill", pattern: "*", action: "allow" }, + { permission: "todowrite", pattern: "*", action: "allow" }, { permission: "bash", pattern: "*", action: "ask" }, { permission: "edit", pattern: "*", action: editAction }, { permission: "webfetch", pattern: "*", action: "ask" }, @@ -700,18 +712,24 @@ const makeOpenCodeRuntime = Effect.gen(function* () { ); yield* Scope.addFinalizer(runtimeScope, terminateChild); - const stdoutRef = yield* Ref.make(""); - const stderrRef = yield* Ref.make(""); + const stdoutRef = yield* Ref.make(""); + const stderrRef = yield* Ref.make(""); const readyDeferred = yield* Deferred.make(); const setReadyFromStdoutChunk = (chunk: string) => - Ref.updateAndGet(stdoutRef, (stdout) => `${stdout}${chunk}`).pipe( - Effect.flatMap((nextStdout) => { - const parsed = parseServerUrlFromOutput(nextStdout); - return parsed - ? Deferred.succeed(readyDeferred, parsed).pipe(Effect.ignore) - : Effect.void; - }), + Ref.modify(stdoutRef, (stdout) => { + if (stdout === null) { + return [null, null] as const; + } + const nextStdout = `${stdout}${chunk}`; + return [ + parseServerUrlFromOutput(nextStdout), + nextStdout.slice(-OPENCODE_SERVER_STARTUP_MAX_OUTPUT_CHARS), + ] as const; + }).pipe( + Effect.flatMap((parsed) => + parsed ? Deferred.succeed(readyDeferred, parsed).pipe(Effect.ignore) : Effect.void, + ), ); const stdoutFiber = yield* child.stdout.pipe( @@ -722,7 +740,13 @@ const makeOpenCodeRuntime = Effect.gen(function* () { ); const stderrFiber = yield* child.stderr.pipe( Stream.decodeText(), - Stream.runForEach((chunk) => Ref.update(stderrRef, (stderr) => `${stderr}${chunk}`)), + Stream.runForEach((chunk) => + Ref.update(stderrRef, (stderr) => + stderr === null + ? null + : `${stderr}${chunk}`.slice(-OPENCODE_SERVER_STARTUP_MAX_OUTPUT_CHARS), + ), + ), Effect.ignore, Effect.forkIn(runtimeScope), ); @@ -730,8 +754,8 @@ const makeOpenCodeRuntime = Effect.gen(function* () { const exitFiber = yield* child.exitCode.pipe( Effect.flatMap((code) => Effect.gen(function* () { - const stdout = yield* Ref.get(stdoutRef); - const stderr = yield* Ref.get(stderrRef); + const stdout = (yield* Ref.get(stdoutRef)) ?? ""; + const stderr = (yield* Ref.get(stderrRef)) ?? ""; const exitCode = Number(code); yield* Deferred.fail( readyDeferred, @@ -757,14 +781,11 @@ const makeOpenCodeRuntime = Effect.gen(function* () { Deferred.await(readyDeferred).pipe(Effect.timeoutOption(timeoutMs)), ); - // Startup-time fibers are no longer needed once ready has resolved (either - // way). The exit fiber is only interrupted on failure; on success it keeps - // the caller's `exitCode` effect observable until the scope closes. - yield* Fiber.interrupt(stdoutFiber).pipe(Effect.ignore); - yield* Fiber.interrupt(stderrFiber).pipe(Effect.ignore); + if (Exit.isFailure(readyExit) || Option.isNone(readyExit.value)) { + yield* Fiber.interruptAll([stdoutFiber, stderrFiber, exitFiber]).pipe(Effect.ignore); + } if (Exit.isFailure(readyExit)) { - yield* Fiber.interrupt(exitFiber).pipe(Effect.ignore); const squashed = Cause.squash(readyExit.cause); return yield* ensureRuntimeError( "startOpenCodeServerProcess", @@ -775,13 +796,18 @@ const makeOpenCodeRuntime = Effect.gen(function* () { const readyOption = readyExit.value; if (Option.isNone(readyOption)) { - yield* Fiber.interrupt(exitFiber).pipe(Effect.ignore); return yield* new OpenCodeRuntimeError({ operation: "startOpenCodeServerProcess", detail: `Timed out waiting for OpenCode server start after ${timeoutMs}ms.`, }); } + // Keep draining both pipes until the process scope closes. Stopping the + // readers can block OpenCode when its output buffers fill. Startup output + // is no longer needed, so discard later output instead of retaining it. + yield* Ref.set(stdoutRef, null); + yield* Ref.set(stderrRef, null); + const url = readyOption.value; const version = yield* verifyOpenCodeServerVersion( createOpenCodeSdkClient({ @@ -847,7 +873,7 @@ const makeOpenCodeRuntime = Effect.gen(function* () { }; const loadProviders = (client: OpencodeClient) => - runOpenCodeSdk("provider.list", () => client.provider.list()).pipe( + runOpenCodeSdk("provider.list", (signal) => client.provider.list(undefined, { signal })).pipe( Effect.filterMapOrFail( (list) => list.data @@ -863,13 +889,13 @@ const makeOpenCodeRuntime = Effect.gen(function* () { ); const loadAgents = (client: OpencodeClient) => - runOpenCodeSdk("app.agents", () => client.app.agents()).pipe( + runOpenCodeSdk("app.agents", (signal) => client.app.agents(undefined, { signal })).pipe( Effect.map((result) => result.data ?? []), Effect.orElseSucceed((): ReadonlyArray => []), ); const loadSkills = (client: OpencodeClient) => - runOpenCodeSdk("app.skills", () => client.app.skills()).pipe( + runOpenCodeSdk("app.skills", (signal) => client.app.skills(undefined, { signal })).pipe( Effect.map((result) => (result.data ?? []).map((skill) => ({ name: skill.name, diff --git a/apps/web/src/session-logic.test.ts b/apps/web/src/session-logic.test.ts index 6c83c18ca..1a7c51176 100644 --- a/apps/web/src/session-logic.test.ts +++ b/apps/web/src/session-logic.test.ts @@ -64,6 +64,37 @@ function makeActivity(overrides: { } describe("derivePendingApprovals", () => { + it.each([{}, { requestType: "unknown" }])( + "exposes legacy OpenCode approvals without a known request kind: %j", + (legacyPayload) => { + const requested = makeActivity({ + kind: "approval.requested", + payload: { requestId: "per-legacy", detail: "*", ...legacyPayload }, + }); + + expect(derivePendingApprovals([requested])).toEqual([ + { + requestId: "per-legacy", + requestKind: "command", + createdAt: requested.createdAt, + detail: "*", + }, + ]); + }, + ); + + it.each(["tool_user_input", "auth_tokens_refresh"])( + "does not turn %s into an approval", + (requestType) => { + const activity = makeActivity({ + kind: "approval.requested", + payload: { requestId: "not-an-approval", requestType }, + }); + + expect(derivePendingApprovals([activity])).toEqual([]); + }, + ); + it("tracks open approvals and removes resolved ones", () => { const activities: OrchestrationThreadActivity[] = [ makeActivity({ @@ -92,7 +123,7 @@ describe("derivePendingApprovals", () => { kind: "approval.requested", summary: "File-change approval requested", tone: "approval", - payload: { requestId: "req-2", requestKind: "file-change" }, + payload: { requestId: "req-2", requestType: "unknown" }, }), ]; @@ -201,7 +232,7 @@ describe("derivePendingApprovals", () => { tone: "approval", payload: { requestId: "req-stale-1", - requestKind: "command", + requestType: "unknown", }, }), makeActivity({ diff --git a/apps/web/src/session-logic.ts b/apps/web/src/session-logic.ts index 55e4d61ba..fabb6a4e4 100644 --- a/apps/web/src/session-logic.ts +++ b/apps/web/src/session-logic.ts @@ -484,10 +484,16 @@ export function derivePendingApprovals( ? payload.options.filter(isProviderApprovalOption) : undefined; - if (activity.kind === "approval.requested" && requestId && requestKind) { + if ( + activity.kind === "approval.requested" && + requestId && + payload?.requestType !== "tool_user_input" && + payload?.requestType !== "auth_tokens_refresh" + ) { openByRequestId.set(requestId, { requestId, - requestKind, + // Older OpenCode requests can have no recognized approval kind. + requestKind: requestKind ?? "command", createdAt: activity.createdAt, ...(detail ? { detail } : {}), ...(appName ? { appName } : {}), diff --git a/docs/internals/providers.md b/docs/internals/providers.md index f5bd33b2b..2c0776749 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -227,6 +227,12 @@ connection, while OpenCode stores MCP connections by directory. Sharing these ch without changing MCP routing would let two threads in one directory replace each other's connection. +Chat adapters send the runtime mode as a session ruleset, but upstream OpenCode evaluates +doom-loop and subagent asks against the agent ruleset only. In full access the adapter answers +those asks itself so the user never sees an approval they already granted. It replies `once` +rather than `always` because OpenCode stores `always` grants per directory, and on a shared +external server that would widen what a supervised thread in the same directory may do. + OpenCode loads its catalog through the HTTP API when an enabled provider instance starts. The provider registry keeps the snapshot in memory and persists it in the existing per-instance cache. Each `subscribeServerConfig` connection refreshes all providers, so a client reconnect reloads the diff --git a/docs/user/providers-opencode.md b/docs/user/providers-opencode.md index f3d9b5dc9..65a783c04 100644 --- a/docs/user/providers-opencode.md +++ b/docs/user/providers-opencode.md @@ -17,6 +17,47 @@ With a server URL, Pylon connects to that external server and uses only the pass provider settings. It does not send a local `OPENCODE_SERVER_PASSWORD` to an external server. OpenCode uses this password for HTTP Basic authentication. +## Approvals + +In **Supervised** and **Auto** modes, OpenCode can read normal project files, search files, load +skills, and update its task list without approval. Files such as `.env` and `.env.local` still +require approval. `.env.example` does not. OpenCode does not have an AI approval reviewer, so +**Auto** uses the same permission rules as **Supervised**. + +OpenCode asks before it runs commands, edits files, accesses the web, or accesses directories +outside the workspace. **Auto-accept edits** also permits file edits without approval. +**Full access** permits all these actions. Questions that need your answer can still appear. + +An **Approval** badge means OpenCode needs a decision. Open the thread to see the action and +choose one of these options: + +- **Allow once** permits this request. +- **Allow for workspace** permits matching requests in other OpenCode sessions in the same + workspace. It is not limited to the current thread. +- **Deny** rejects this request. Use **Stop** to stop the whole turn. + +If a connection error prevents the reply, the approval stays available so you can try again. + +## Progress + +Pylon shows OpenCode's response text and tool results while work runs. The web and desktop apps +also show its task-list progress in the Tasks tab, turn summary, and sidebar working line. +Cancelled steps disappear from the list. A task-list update does not require approval. + +If the OpenCode connection closes unexpectedly, Pylon shows an error. Send another prompt to +reconnect to the same OpenCode session. + +## Stop a turn + +When you select **Stop**, Pylon stops the main OpenCode session and all nested child sessions. +Pylon waits for this cleanup before it marks the turn as stopped or sends the next prompt. It +does not stop unrelated OpenCode sessions. After Stop succeeds, pending approvals and questions +are cleared. + +Stop reports an error if OpenCode cannot stop the main session or list or stop a child session. +When Pylon closes an OpenCode session, it also tries to stop the child sessions, but this +teardown is best effort. + ## Refresh the model list Pylon loads the model list when an enabled OpenCode provider starts and keeps the list in its diff --git a/packages/client-runtime/src/work-log/presentation.test.ts b/packages/client-runtime/src/work-log/presentation.test.ts index 8d179c417..18e22dd18 100644 --- a/packages/client-runtime/src/work-log/presentation.test.ts +++ b/packages/client-runtime/src/work-log/presentation.test.ts @@ -6,10 +6,49 @@ import { commandDetailRepeatsCommand, extractCommandOutputText, resolveViewedImageAsset, + summarizeToolGroup, toolGroupAction, + toolGroupSummaryKind, workEntryViewedImagePath, + type WorkLogPresentationEntry, } from "./presentation.js"; +describe("summarizeToolGroup", () => { + it.each(["command", "file-read", "file-change"])( + "keeps %s approvals out of tool execution counts", + (requestKind) => { + const approvals = [ + { + label: "Approval requested", + sourceActivityKind: "approval.requested", + tone: "info", + requestKind, + }, + { + label: "Approval resolved", + sourceActivityKind: "approval.resolved", + tone: "info", + requestKind, + }, + { + label: "Provider approval response failed", + sourceActivityKind: "provider.approval.respond.failed", + tone: "error", + }, + ] satisfies WorkLogPresentationEntry[]; + + expect( + summarizeToolGroup([ + ...approvals, + { label: "Read", tone: "tool", itemType: "dynamic_tool_call" }, + ]), + ).toBe("Received 3 updates and used 1 tool"); + expect(summarizeToolGroup(approvals)).toBe("Received 3 updates"); + expect(toolGroupSummaryKind(approvals)).toBe("update"); + }, + ); +}); + describe("command work-log details", () => { it("extracts Claude result blocks and projected output", () => { expect( diff --git a/packages/client-runtime/src/work-log/presentation.ts b/packages/client-runtime/src/work-log/presentation.ts index 1528b34d1..89cfe2540 100644 --- a/packages/client-runtime/src/work-log/presentation.ts +++ b/packages/client-runtime/src/work-log/presentation.ts @@ -211,6 +211,13 @@ export function workLogEntryIsLocalCodeSearch(entry: WorkLogPresentationEntry): } export function toolGroupAction(entry: WorkLogPresentationEntry): ToolGroupAction { + if ( + entry.sourceActivityKind === "approval.requested" || + entry.sourceActivityKind === "approval.resolved" || + entry.sourceActivityKind === "provider.approval.respond.failed" + ) { + return "update"; + } if ( entry.requestKind === "file-read" || entry.itemType === "image_view" ||