fix(core): serialize V2 prompt settlement around event publication #34853

Activity

1. rekram1-node commented on Jul 3, 2026

   

   rekram1-node

   on Jul 3, 2026

   ContributorAuthor

   More actions

   Related race: url-mode MCP elicitation completion can be silently dropped

   Found while reviewing the merged form work on v2 (85fde8c11d). Same family of problem — an ordering window widened by inline event publication — but on the create side rather than settlement.

   Scenario

   packages/core/src/mcp/index.ts, url-mode branch of elicitation.create:

   1. Handler synchronously generates formID and registers it in the urlElicitations map.
   2. forms.ask(...) then creates the form record asynchronously (persistence + Event.Created publish, whose listeners run inline and widen the window).
   3. If the server's elicitation/complete notification arrives between (1) and (2) completing, the complete handler finds the formID in the map but forms.reply fails with NotFoundError — which is swallowed by Effect.ignore (index.ts:~386).
   4. The completion is lost silently: the form stays pending forever and the originating MCP tool call hangs until the user manually cancels or the abort signal fires.

   Both handlers are independent Effect.runPromise calls off the SDK message loop (packages/core/src/mcp/client.ts:185-189), which does not serialize a request handler against subsequent notifications — so the interleaving only needs one await point inside forms.ask.

   Trigger likelihood

   Low: a legitimate flow requires the user to visit the URL, which requires the form UI, which requires the form to exist. Realistic triggers are servers that complete without user interaction (e.g. OAuth already satisfied server-side, completes within ms of issuing the request) or test harnesses.

   Correct fix

   Not a retry — this is an ordering problem, not a transient failure, and a retry only bounds the window with a timer while conflating "not created yet" with "never existed". The structural fix is to make the map value a Deferred<Form.ID>:

   • create: register the deferred synchronously (same position as today's set), create the form, succeed the deferred, then await the answer. ensuring cleanup fails the deferred and deletes the entry.
   • complete: map entry present means the elicitation is definitionally in-flight — await the deferred, then forms.reply. Correct by construction, no timing dependence. A failed deferred (create aborted/failed) surfaces and is ignored, matching today's settled-form behavior.

   Note the Effect.ignore on forms.reply must stay in some form: it also covers the legitimate case of a completion arriving after the form was already answered/cancelled.

   Secondary (even lower severity)

   If a server reuses an elicitationId while the first is pending, the second urlElicitations.set clobbers the entry and the first elicitation's ensuring delete then removes the second's mapping, losing its completion. Requires server misbehavior (IDs should be unique per spec).
2. opencode-agent

   added

   2.0

   coreAnything pertaining to core functionality of the application (opencode server stuff)Anything pertaining to core functionality of the application (opencode server stuff)

   bugSomething isn't workingSomething isn't working

   on Jul 5, 2026
3. sahiljagtap08 commented on Aug 23, 2026

   

   sahiljagtap08

   on Aug 23, 2026

   More actions

   I took a shot at this in #44340. It claims the pending entry before publishing so a concurrent reply gets NotFoundError instead of double settling, with rollback if the publish fails. Covers QuestionV2 and PermissionV2 including the cascade loops.
4. opencodejr commented on Oct 29, 2026

   

   opencodejrBot

   on Oct 29, 2026

   More actions

   Thanks, I reproduced this at the service level on the latest dev (f43563b). form.ts and bus.ts are the same in v2.2.0. permission.ts differs slightly there, but its reply path is still publish-then-delete.

   The script below registers a bus.listen listener that blocks during the first settlement event. While it's blocked, the script sends a second settlement for the same item:

   • Form, two replies (A, then B): both succeed and two form.replied events go out (A and B). The stored state ends up answered: A, even though B's event was published last, because whichever Cache.set commits last wins.
   • Form, reply + cancel: both succeed, one form.replied and one form.cancelled event go out, and the stored state is answered.
   • Permission, once + reject: both succeed and two permission.replied events go out (once, reject). The waiting assert ends up rejected, even though once was published first. The durable per-aggregate lock in Bus orders the two publishes but doesn't stop the second reply.

   I couldn't test Question separately because packages/core/src/question.ts doesn't exist on dev anymore.

   To run the script, copy it into packages/core/test/ and run bun test test/settlement-race.test.ts. Its three tests currently fail.

   Reproduction script (settlement-race.test.ts)

   // Repro for anomalyco/opencode#34853: concurrent settlement while a live listener delays Bus.publish.
   import { describe, expect } from "bun:test"
   import { Deferred, Effect, Exit, Fiber, Layer } from "effect"
   import { Agent } from "@opencode/core/agent"
   import { Database } from "@opencode/core/database/database"
   import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
   import { LayerNode } from "@opencode/util/effect/layer-node"
   import { Bus } from "@opencode/core/bus"
   import { Form } from "@opencode/core/form"
   import { Location } from "@opencode/core/location"
   import { Permission } from "@opencode/core/permission"
   import { PermissionSaved } from "@opencode/core/permission/saved"
   import { Project } from "@opencode/core/project"
   import { ProjectTable } from "@opencode/core/project/sql"
   import { AbsolutePath } from "@opencode/core/schema"
   import { Session } from "@opencode/core/session"
   import { SessionTable } from "@opencode/core/session/sql"
   import { SessionStore } from "@opencode/core/session/store"
   import { SessionSchema } from "@opencode/core/session/schema"
   import { location } from "./fixture/location"
   import { testEffect } from "./lib/effect"
   
   // A live listener that blocks while the first "replied" event is being published.
   const slowListener = (type: string) =>
     Effect.gen(function* () {
       const bus = yield* Bus.Service
       const entered = yield* Deferred.make<void>()
       const release = yield* Deferred.make<void>()
       const seen: unknown[] = []
       let first = true
       const unsubscribe = yield* bus.listen((event) => {
         if (event.type !== type) return Effect.void
         seen.push(event.data)
         if (!first) return Effect.void
         first = false
         return Effect.andThen(Deferred.succeed(entered, undefined), Deferred.await(release))
       })
       yield* Effect.addFinalizer(() => unsubscribe)
       return { entered, release, seen }
     })
   
   describe("Form settlement race", () => {
     const it = testEffect(AppNodeBuilder.build(LayerNode.group([Bus.node, Form.node])))
     it.effect("two concurrent replies both succeed", () =>
       Effect.gen(function* () {
         const service = yield* Form.Service
         const slow = yield* slowListener(Form.Event.Replied.type)
         const form = yield* service.create({
           sessionID: SessionSchema.ID.make("ses_test"),
           title: "race",
           fields: [{ key: "name", type: "string", required: true }],
         })
         const a = yield* service.reply({ id: form.id, answer: { name: "A" } }).pipe(Effect.exit, Effect.forkScoped)
         yield* Deferred.await(slow.entered)
         const b = yield* service.reply({ id: form.id, answer: { name: "B" } }).pipe(Effect.exit)
         yield* Deferred.succeed(slow.release, undefined)
         const aExit = yield* Fiber.join(a)
         const state = yield* service.state(form.id)
         console.log("Form: A =", Exit.isSuccess(aExit) ? "success" : "failure", "| B =", Exit.isSuccess(b) ? "success" : String(b.cause))
         console.log("Form: replied events =", JSON.stringify(slow.seen.map((d: any) => d.answer)), "| final state =", JSON.stringify(state))
         // Expected: exactly one reply wins, the other fails with AlreadySettledError, one Replied event.
         expect(slow.seen.length).toBe(1)
         expect(Exit.isSuccess(aExit) && Exit.isSuccess(b)).toBe(false)
       }),
     )
     it.effect("reply and cancel both succeed", () =>
       Effect.gen(function* () {
         const service = yield* Form.Service
         const bus = yield* Bus.Service
         const slow = yield* slowListener(Form.Event.Replied.type)
         const cancelled: unknown[] = []
         const un = yield* bus.listen((e) => (e.type === Form.Event.Cancelled.type ? Effect.sync(() => void cancelled.push(e.data)) : Effect.void))
         yield* Effect.addFinalizer(() => un)
         const form = yield* service.create({
           sessionID: SessionSchema.ID.make("ses_test"),
           title: "race",
           fields: [{ key: "name", type: "string", required: true }],
         })
         const a = yield* service.reply({ id: form.id, answer: { name: "A" } }).pipe(Effect.exit, Effect.forkScoped)
         yield* Deferred.await(slow.entered)
         const b = yield* service.cancel(form.id).pipe(Effect.exit)
         yield* Deferred.succeed(slow.release, undefined)
         const aExit = yield* Fiber.join(a)
         const state = yield* service.state(form.id)
         console.log("Form: reply =", Exit.isSuccess(aExit) ? "success" : "failure", "| cancel =", Exit.isSuccess(b) ? "success" : String(b.cause))
         console.log("Form: replied events =", slow.seen.length, "cancelled events =", cancelled.length, "| final state =", JSON.stringify(state))
         expect(slow.seen.length + cancelled.length).toBe(1)
       }),
     )
   })
   
   describe("Permission settlement race", () => {
     const current = Layer.succeed(Location.Service, Location.Service.of(location({ directory: AbsolutePath.make("/project") })))
     const it = testEffect(
       AppNodeBuilder.build(
         LayerNode.group([Database.node, Bus.node, SessionStore.node, PermissionSaved.node, Agent.node, Permission.node]),
         [Location.node.replace(current)],
       ),
     )
     it.effect("two concurrent replies both publish and succeed", () =>
       Effect.gen(function* () {
         const { db } = yield* Database.Service
         yield* db.insert(ProjectTable).values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }).onConflictDoNothing().run().pipe(Effect.orDie)
         yield* db.insert(SessionTable).values({ id: Session.ID.make("ses_test"), project_id: Project.ID.global, slug: "test", directory: "/project", title: "test", version: "test", agent: Agent.ID.make("test") }).onConflictDoNothing().run().pipe(Effect.orDie)
         yield* (yield* Agent.Service).transform((editor) => editor.update(Agent.ID.make("test"), (agent) => { agent.permissions = [] }))
         console.log("Permission: setup done")
         const service = yield* Permission.Service
         const bus = yield* Bus.Service
         const asked = yield* Deferred.make<Permission.Request>()
         const un = yield* bus.listen((e) => (e.type === Permission.Event.Asked.type ? Deferred.succeed(asked, e.data as Permission.Request).pipe(Effect.asVoid) : Effect.void))
         yield* Effect.addFinalizer(() => un)
         const slow = yield* slowListener(Permission.Event.Replied.type)
         const asker = yield* service.assert({ id: Permission.ID.create("per_test"), sessionID: Session.ID.make("ses_test"), action: "read", resources: ["src/index.ts"] }).pipe(Effect.exit, Effect.forkScoped)
         const request = yield* Deferred.await(asked)
         console.log("Permission: asked")
         const a = yield* service.reply({ requestID: request.id, reply: "once" }).pipe(Effect.exit, Effect.forkScoped)
         yield* Deferred.await(slow.entered)
         console.log("Permission: first reply publishing")
         const bFiber = yield* service.reply({ requestID: request.id, reply: "reject" }).pipe(Effect.exit, Effect.forkScoped)
         yield* Effect.promise(() => new Promise((r) => setTimeout(r, 50)))
         console.log("Permission: releasing listener")
         yield* Deferred.succeed(slow.release, undefined)
         const aExit = yield* Fiber.join(a)
         console.log("Permission: A joined")
         const b = yield* Fiber.join(bFiber)
         const askExit = yield* Fiber.join(asker)
         console.log("Permission: once =", Exit.isSuccess(aExit) ? "success" : "failure", "| reject =", Exit.isSuccess(b) ? "success" : String(b.cause))
         console.log("Permission: replied events =", JSON.stringify(slow.seen.map((d: any) => d.reply)), "| asker =", Exit.isSuccess(askExit) ? "allowed" : "rejected")
         expect(slow.seen.length).toBe(1)
       }),
     )
   })

   Run it: bun settlement-race.test.ts
5. opencodejr

   added

   reproduced

   on Oct 29, 2026
6. lilong555 commented on Nov 4, 2026

   

   lilong555

   on Nov 4, 2026

   More actions

   Hi @535628443 and @sahiljagtap08, thank you for your work on #52550 and #44340. Your per-request settlement approach was very helpful in investigating this issue.

   I have prepared and reviewed a follow-up on dev at b9d0982b that builds on that approach. It adds cleanup when close() skips an in-flight reply whose publication subsequently fails, and clears ask-only permission requests when shutdown publication itself fails.

   I compared the production patch from #52550 and the follow-up on that same base, with identical dependencies and tests using controlled listener failures. Four shutdown regression cases failed on the original patch and passed on the follow-up in each of 10 runs. The 15 settlement regression cases from #52550 passed on both implementations.

   Are either of you still working on this issue? I would be happy to share the reproductions and collaborate on your existing work. If a follow-up PR would be helpful, I will reference and credit both earlier contributions. Please let me know your preferred way to coordinate, and thanks again for laying the groundwork.