From cd96feaad01ece2539447bcbd9a80b2ec3a463fd Mon Sep 17 00:00:00 2001 From: Sarav Date: Thu, 16 Jul 2026 10:46:06 +0530 Subject: [PATCH 1/2] =?UTF-8?q?fix:=20[AI-7743]=20mcp-add=20question=20han?= =?UTF-8?q?gs=20=E2=80=94=20share=20question=20state=20+=20event=20bus=20a?= =?UTF-8?q?cross=20bundled=20module=20copies?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `/discover-and-add-mcps` rendered the "which MCP / what scope" question but, after answering, the chat sat on "Thinking…" and the server was never added. Root cause: after the v1.17.9 upstream merge, `src/question/index.ts` is bundled as TWO separate module instances (a module-scoped id differs between the tool's `ask()` and the HTTP route's `reply()`; Bun does not reliably dedupe it, and normalizing imports does NOT change that). Each copy ran its own `makeRuntime` runtime, which broke the flow two independent ways: 1. State split — the `question` tool registered the pending `Deferred` in one copy's `InstanceState` map, while `POST /question/:id/reply` looked it up in the OTHER copy's empty map. The reply 404'd and the `Deferred` never resolved, so the agent loop blocked forever ("Thinking…"). 2. Event never reached the IDE webview — `question.asked` was published only via `EventV2Bridge` -> `GlobalBus`, but the `/event` SSE stream the webview reads is fed by the Bus *wildcard PubSub* (`Bus.publish`). So `pendingQuestions` stayed empty and the answer card's Submit had no request id -> submitting did nothing. Fix (self-contained in `question/index.ts`): - Anchor the pending-question registry on `globalThis` (keyed by instance directory) so every bundled module copy shares one map. Restore per-instance cleanup via `registerDisposer` so entries don't leak across instances/tests. - Publish `question.asked`/`replied`/`rejected` via `Bus.publish` (added `BusEvent` mirrors) so they reach `/event` like every other webview-visible event. Every divergence from upstream is wrapped in `altimate_change start/end` markers (14/14 balanced) so future upstream merges are handled correctly; `--require-markers --strict` and the marker-integrity tests pass. Verified end-to-end in code-server: discover -> answer (Yes / Project) -> datamate written to `.altimate-code/altimate-code.json` (enabled) and the chat shows the success summary. Marker-integrity + question tests pass (134). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/opencode/src/question/index.ts | 132 +++++++++++++++++++----- 1 file changed, 108 insertions(+), 24 deletions(-) diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 8d2f2cb11..102d315a7 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -7,6 +7,10 @@ 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 z from "zod" // altimate_change end // Schemas — these are pure data; nothing checks class identity (see PR @@ -93,6 +97,39 @@ 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. (Schemas are loose — Bus.publish does not re-validate; +// they exist for the `type` string and typing.) +const BusAsked = BusEvent.define( + "question.asked", + z.object({ + id: QuestionID.zod, + sessionID: SessionID.zod, + questions: z.array(z.any()), + tool: z.object({ messageID: MessageID.zod, callID: z.string() }).optional(), + }), +) +const BusReplied = BusEvent.define( + "question.replied", + z.object({ + sessionID: SessionID.zod, + requestID: QuestionID.zod, + answers: z.array(z.array(z.string())), + }), +) +const BusRejected = BusEvent.define( + "question.rejected", + z.object({ sessionID: SessionID.zod, requestID: QuestionID.zod }), +) +// altimate_change end + export class RejectedError extends Schema.TaggedErrorClass()("QuestionRejectedError", {}) { override get message() { return "The user dismissed this question" @@ -108,9 +145,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 +193,30 @@ 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())).catch(() => {}) + } + }) + 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 +229,12 @@ export const layer = Layer.effect( } pending.set(id, { info, deferred }) yield* events.publish(Event.Asked, info) + // altimate_change start — also publish on the Bus wildcard so the IDE webview + // (subscribed to /event) receives question.asked and can answer the card. + yield* Effect.promise(() => + Bus.publish(BusAsked, { id, sessionID: input.sessionID, questions: [...input.questions], tool: input.tool }), + ) + // altimate_change end return yield* Effect.ensuring( Deferred.await(deferred), @@ -184,7 +248,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 }) @@ -197,11 +263,22 @@ export const layer = Layer.effect( requestID: existing.info.id, answers: input.answers.map((a) => [...a]), }) + // altimate_change start — mirror on the Bus wildcard for /event (webview) clients. + yield* Effect.promise(() => + Bus.publish(BusReplied, { + sessionID: existing.info.sessionID, + requestID: existing.info.id, + answers: input.answers.map((a) => [...a]), + }), + ) + // altimate_change end yield* Deferred.succeed(existing.deferred, input.answers) }) 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 }) @@ -213,11 +290,18 @@ export const layer = Layer.effect( sessionID: existing.info.sessionID, requestID: existing.info.id, }) + // altimate_change start — mirror on the Bus wildcard for /event (webview) clients. + yield* Effect.promise(() => + Bus.publish(BusRejected, { sessionID: existing.info.sessionID, requestID: existing.info.id }), + ) + // altimate_change end yield* Deferred.fail(existing.deferred, new RejectedError()) }) 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) }) From 0b5ecd12fe12aa2dcf2c97b16169281f3cefa4d6 Mon Sep 17 00:00:00 2001 From: Sarav Date: Thu, 23 Jul 2026 10:43:02 +0530 Subject: [PATCH 2/2] fix: [AI-7743] keep Bus mirror off the question Deferred critical path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review feedback (flagged by all reviewers): the `Bus.publish` mirrors added for /event (webview) clients sat on the Deferred critical path. Since `Effect.promise` converts a promise rejection into an unrecoverable fiber defect, a `Bus.publish` rejection could abort the fiber before `Deferred.succeed`/`Deferred.fail` ran — leaving the awaiting `Question.ask()` hung on "Thinking…", the exact failure this PR fixes. - Add a best-effort `mirror()` helper: `Effect.promise(() => Bus.publish(...))` recovered with `Effect.catchCause` to a logged warning, so a publish failure can never abort core question settlement. - Settle the Deferred FIRST, then mirror, in `reply()` and `reject()`. - `ask()` uses `mirror()` too, so a publish failure can't abort it before the cleanup finalizer is registered. - Define the Bus mirror schemas from the structured Effect schemas via the `zod` adapter (`zod(Request)` / `zod(Replied)` / `zod(Rejected)`) so the generated /event OpenAPI payloads match the SDK types (no `any`/loose shapes). - Log (not silently swallow) failures in the instance-dispose cleanup. - Document the intentional GlobalBus double-emit (verified harmless in-repo). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/opencode/src/question/index.ts | 94 ++++++++++++++----------- 1 file changed, 52 insertions(+), 42 deletions(-) diff --git a/packages/opencode/src/question/index.ts b/packages/opencode/src/question/index.ts index 102d315a7..4d4060ce7 100644 --- a/packages/opencode/src/question/index.ts +++ b/packages/opencode/src/question/index.ts @@ -10,6 +10,7 @@ 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 @@ -105,29 +106,31 @@ export const Event = { // 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. (Schemas are loose — Bus.publish does not re-validate; -// they exist for the `type` string and typing.) -const BusAsked = BusEvent.define( - "question.asked", - z.object({ - id: QuestionID.zod, - sessionID: SessionID.zod, - questions: z.array(z.any()), - tool: z.object({ messageID: MessageID.zod, callID: z.string() }).optional(), - }), -) -const BusReplied = BusEvent.define( - "question.replied", - z.object({ - sessionID: SessionID.zod, - requestID: QuestionID.zod, - answers: z.array(z.array(z.string())), - }), -) -const BusRejected = BusEvent.define( - "question.rejected", - z.object({ sessionID: SessionID.zod, requestID: QuestionID.zod }), -) +// 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", {}) { @@ -202,7 +205,11 @@ export const layer = Layer.effect( if (!map) return pendingByDir.delete(directory) for (const { deferred } of map.values()) { - await Effect.runPromise(Deferred.fail(deferred, new RejectedError())).catch(() => {}) + 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)) @@ -229,11 +236,16 @@ export const layer = Layer.effect( } pending.set(id, { info, deferred }) yield* events.publish(Event.Asked, info) - // altimate_change start — also publish on the Bus wildcard so the IDE webview + // altimate_change start — also mirror on the Bus wildcard so the IDE webview // (subscribed to /event) receives question.asked and can answer the card. - yield* Effect.promise(() => - Bus.publish(BusAsked, { id, sessionID: input.sessionID, questions: [...input.questions], tool: input.tool }), - ) + // 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( @@ -263,16 +275,15 @@ export const layer = Layer.effect( requestID: existing.info.id, answers: input.answers.map((a) => [...a]), }) - // altimate_change start — mirror on the Bus wildcard for /event (webview) clients. - yield* Effect.promise(() => - Bus.publish(BusReplied, { - sessionID: existing.info.sessionID, - requestID: existing.info.id, - answers: input.answers.map((a) => [...a]), - }), - ) - // altimate_change end 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) { @@ -290,12 +301,11 @@ export const layer = Layer.effect( sessionID: existing.info.sessionID, requestID: existing.info.id, }) - // altimate_change start — mirror on the Bus wildcard for /event (webview) clients. - yield* Effect.promise(() => - Bus.publish(BusRejected, { sessionID: existing.info.sessionID, requestID: existing.info.id }), - ) - // altimate_change end 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* () {