diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 8d2f2cb11..4d4060ce7 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -7,6 +7,11 @@ import { EventV2Bridge } from "@/event-v2-bridge" import { EventV2 } from "@opencode-ai/core/event" // altimate_change start — makeRuntime for the restored Promise wrappers (see bottom of file) import { makeRuntime } from "@/effect/run-service" +import { registerDisposer } from "@/effect/instance-registry" +import { Bus } from "@/bus" +import { BusEvent } from "@/bus/bus-event" +import { zod } from "@/util/effect-zod" +import z from "zod" // altimate_change end // Schemas — these are pure data; nothing checks class identity (see PR @@ -93,6 +98,41 @@ export const Event = { Rejected: EventV2.define({ type: "question.rejected", schema: Rejected.fields }), } +// altimate_change start — BusEvent mirrors of the question events. +// +// The EventV2 `Event` defs above publish to GlobalBus/EventV2 consumers only. +// The IDE webview subscribes to the `/event` SSE route, which is fed by the Bus +// *wildcard PubSub* (`Bus.publish`), a different channel. So `question.asked` +// never reached the webview → `pendingQuestions` stayed empty → the mcp-add +// question card had no request id to reply with → "submit does nothing". +// Publish these via `Bus.publish` too so they reach /event like every other +// webview-visible event. +// +// Reuse the structured Effect schemas (via the `zod` adapter) so the generated +// `/event` OpenAPI payloads match the SDK's typed shapes rather than `any`. +// +// Note: `EventV2Bridge.listen` already forwards these EventV2 events to GlobalBus +// and `Bus.publish` re-emits to GlobalBus too, so `/global/event` consumers see +// each question event twice (different top-level ids). This is intentional and +// verified harmless in-repo — TUI `sync.tsx` reconciles by request id, +// `notifications.ts` dedupes by id, and the trace consumer ignores question +// events. External subscribers that don't dedupe are the only residual concern. +const BusAsked = BusEvent.define("question.asked", zod(Request)) +const BusReplied = BusEvent.define("question.replied", zod(Replied)) +const BusRejected = BusEvent.define("question.rejected", zod(Rejected)) + +// Best-effort Bus mirror. A fire-and-forget /event notification must NEVER be +// able to abort core question settlement: `Effect.promise` turns a promise +// rejection into an unrecoverable fiber defect, and on the Deferred critical +// path that would skip `Deferred.succeed`/`Deferred.fail` and re-hang the tool +// on "Thinking…" — the exact failure this PR fixes. Recover any cause to a +// logged warning so publication can never block settlement. +const mirror = (def: D, properties: z.output) => + Effect.promise(() => Bus.publish(def, properties)).pipe( + Effect.catchCause((cause) => Effect.logWarning("question bus mirror failed", { type: def.type, cause })), + ) +// altimate_change end + export class RejectedError extends Schema.TaggedErrorClass()("QuestionRejectedError", {}) { override get message() { return "The user dismissed this question" @@ -108,9 +148,31 @@ interface PendingEntry { deferred: Deferred.Deferred, RejectedError> } -interface State { - pending: Map +// altimate_change start — process-global pending registry. +// +// The imperative `Question.ask()`/`reply()`/`list()` wrappers (bottom of file) +// get bundled into MORE THAN ONE module instance (proven: a module-scoped id +// differs between the tool's ask() and the HTTP route's reply()). Consistent +// `@/question` imports did NOT dedupe them. Each copy ran its own module-scoped +// `makeRuntime(...)` runtime with its own `InstanceState` cache, so the question +// TOOL registered the pending Deferred in one copy's map while the HTTP reply +// route looked it up in the OTHER copy's empty map — the Deferred never resolved +// and the `/discover-and-add-mcps` question hung on "Thinking…" after answering. +// +// Anchor the registry on `globalThis` so every module copy shares one Map. Keyed +// by instance directory so `list()` stays per-instance. +type PendingByDir = Map> +const pendingByDir: PendingByDir = ((globalThis as Record)["__altimateQuestionPending"] ??= + new Map>()) as PendingByDir +function pendingFor(directory: string): Map { + let map = pendingByDir.get(directory) + if (!map) { + map = new Map() + pendingByDir.set(directory, map) + } + return map } +// altimate_change end // Service @@ -134,31 +196,34 @@ export const layer = Layer.effect( Service, Effect.gen(function* () { const events = yield* EventV2Bridge.Service - const state = yield* InstanceState.make( - Effect.fn("Question.state")(function* () { - const state = { - pending: new Map(), - } - - yield* Effect.addFinalizer(() => - Effect.gen(function* () { - for (const item of state.pending.values()) { - yield* Deferred.fail(item.deferred, new RejectedError()) - } - state.pending.clear() - }), - ) - return state - }), - ) + // altimate_change start — clear a directory's pending questions when its + // instance is disposed/reloaded (mirrors the removed InstanceState finalizer) + // so entries in the process-global registry don't leak across instances. + const off = registerDisposer(async (directory) => { + const map = pendingByDir.get(directory) + if (!map) return + pendingByDir.delete(directory) + for (const { deferred } of map.values()) { + await Effect.runPromise( + Deferred.fail(deferred, new RejectedError()).pipe( + Effect.catchCause((cause) => Effect.logWarning("question cleanup failed on dispose", { cause })), + ), + ) + } + }) + yield* Effect.addFinalizer(() => Effect.sync(off)) + // altimate_change end const ask = Effect.fn("Question.ask")(function* (input: { sessionID: SessionID questions: ReadonlyArray tool?: Tool }) { - const pending = (yield* InstanceState.get(state)).pending + // altimate_change start — pending map lives in the globalThis registry (see top of file) + const directory = yield* InstanceState.directory + const pending = pendingFor(directory) + // altimate_change end const id = QuestionID.ascending() yield* Effect.logInfo("asking", { id, questions: input.questions.length }) @@ -171,6 +236,17 @@ export const layer = Layer.effect( } pending.set(id, { info, deferred }) yield* events.publish(Event.Asked, info) + // altimate_change start — also mirror on the Bus wildcard so the IDE webview + // (subscribed to /event) receives question.asked and can answer the card. + // Best-effort: a publish failure must not abort ask() before it registers + // the cleanup finalizer below (see `mirror`). + yield* mirror(BusAsked, { + id, + sessionID: input.sessionID, + questions: [...input.questions], + tool: input.tool, + }) + // altimate_change end return yield* Effect.ensuring( Deferred.await(deferred), @@ -184,7 +260,9 @@ export const layer = Layer.effect( requestID: QuestionID answers: ReadonlyArray }) { - const pending = (yield* InstanceState.get(state)).pending + // altimate_change start — pending map lives in the globalThis registry (see top of file) + const pending = pendingFor(yield* InstanceState.directory) + // altimate_change end const existing = pending.get(input.requestID) if (!existing) { yield* Effect.logWarning("reply for unknown request", { requestID: input.requestID }) @@ -198,10 +276,20 @@ export const layer = Layer.effect( answers: input.answers.map((a) => [...a]), }) yield* Deferred.succeed(existing.deferred, input.answers) + // altimate_change start — mirror on the Bus wildcard for /event (webview) clients, + // AFTER settling the Deferred and best-effort so a publish failure can't re-hang ask(). + yield* mirror(BusReplied, { + sessionID: existing.info.sessionID, + requestID: existing.info.id, + answers: input.answers.map((a) => [...a]), + }) + // altimate_change end }) const reject = Effect.fn("Question.reject")(function* (requestID: QuestionID) { - const pending = (yield* InstanceState.get(state)).pending + // altimate_change start — pending map lives in the globalThis registry (see top of file) + const pending = pendingFor(yield* InstanceState.directory) + // altimate_change end const existing = pending.get(requestID) if (!existing) { yield* Effect.logWarning("reject for unknown request", { requestID }) @@ -214,10 +302,16 @@ export const layer = Layer.effect( requestID: existing.info.id, }) yield* Deferred.fail(existing.deferred, new RejectedError()) + // altimate_change start — mirror on the Bus wildcard for /event (webview) clients, + // AFTER settling the Deferred and best-effort so a publish failure can't strand it. + yield* mirror(BusRejected, { sessionID: existing.info.sessionID, requestID: existing.info.id }) + // altimate_change end }) const list = Effect.fn("Question.list")(function* () { - const pending = (yield* InstanceState.get(state)).pending + // altimate_change start — pending map lives in the globalThis registry (see top of file) + const pending = pendingFor(yield* InstanceState.directory) + // altimate_change end return Array.from(pending.values(), (x) => x.info) })