fix(core): coalesce duplicate form asks

packages/core/src/form.ts

@@ -1,9 +1,10 @@
 export * as Form from "./form"
 
 import { Form } from "@opencode-ai/schema/form"
-import { Cache, Context, Deferred, Duration, Effect, Exit, Layer, Option, Schema } from "effect"
+import { Cache, Context, Deferred, Duration, Effect, Exit, Layer, Option, Ref, Schema } from "effect"
 import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
 import { Bus } from "./bus"
+import { KeyedMutex } from "./effect/keyed-mutex"
 
 const RETENTION = Duration.minutes(10)
 
@@ -12,6 +13,7 @@ export type ID = typeof ID.Type
 
 export const Info = Form.Info
 export type Info = typeof Info.Type
+const infoEquivalent = Schema.toEquivalence(Info)
 
 export const Field = Form.Field
 export type Field = Form.Field
@@ -94,12 +96,14 @@ interface Entry {
   readonly form: Info
   readonly state: State
   readonly deferred: Deferred.Deferred<TerminalState>
+  readonly waiters: Ref.Ref<number>
 }
 
 export const layer = Layer.effect(
   Service,
   Effect.gen(function* () {
     const bus = yield* Bus.Service
+    const locks = KeyedMutex.makeUnsafe<ID>()
     const forms = yield* Cache.makeWith<ID, Entry>(
       () => Effect.die(new Error("Form cache must be used via set/getSuccess, never get")),
       {
@@ -120,45 +124,77 @@ export const layer = Layer.effect(
       )
     })
 
-    const create = Effect.fn("Form.create")((input: CreateInput) =>
+    const settleCancelled = Effect.fn("Form.settleCancelled")((id: ID) =>
       Effect.uninterruptible(
         Effect.gen(function* () {
-          const id = input.id ?? ID.create()
+          const entry = yield* find(id)
-          const existing = yield* Cache.getSuccess(forms, id)
+          if (entry.state.status !== "pending") return yield* new AlreadySettledError({ id })
-          if (Option.isSome(existing)) return yield* new AlreadyExistsError({ id })
+          const next: TerminalState = { status: "cancelled" }
-          const invalid = validateFields(input.fields)
+          yield* bus.publish(Form.Event.Cancelled, { id, sessionID: entry.form.sessionID })
-          if (invalid) return yield* new InvalidFormError({ message: invalid })
+          yield* Cache.set(forms, id, { ...entry, state: next })
-          const form: Info = {
+          yield* Deferred.succeed(entry.deferred, next)
-            id,
-            sessionID: input.sessionID,
-            title: input.title,
-            ...(input.metadata === undefined ? {} : { metadata: input.metadata }),
-            fields: input.fields,
-          }
-          const entry: Entry = {
-            form,
-            state: { status: "pending" },
-            deferred: yield* Deferred.make<TerminalState>(),
-          }
-          yield* Cache.set(forms, id, entry)
-          yield* bus.publish(Form.Event.Created, { form }).pipe(Effect.onError(() => Cache.invalidate(forms, id)))
-          return form
         }),
       ),
     )
 
-    const ask = Effect.fn("Form.ask")((input: CreateInput) =>
+    const acquire = Effect.fn("Form.acquire")((input: CreateInput, id: ID, join: boolean, wait: boolean) =>
-      Effect.uninterruptibleMask((restore) =>
+      locks.withLock(id)(
-        Effect.gen(function* () {
+        Effect.uninterruptible(
-          const form = yield* create(input)
+          Effect.gen(function* () {
-          const entry = yield* find(form.id).pipe(Effect.orDie)
+            const existing = yield* Cache.getSuccess(forms, id)
-          return yield* restore(Deferred.await(entry.deferred)).pipe(
+            if (Option.isSome(existing)) {
-            Effect.onInterrupt(() => Effect.ignore(cancel(form.id))),
+              const expected = makeInfo(input, id)
-          )
+              if (join && infoEquivalent(existing.value.form, expected)) {
-        }),
+                if (wait) yield* Ref.update(existing.value.waiters, (count) => count + 1)
+                return existing.value
+              }
+              return yield* new AlreadyExistsError({ id })
+            }
+            const invalid = validateFields(input.fields)
+            if (invalid) return yield* new InvalidFormError({ message: invalid })
+            const form = makeInfo(input, id)
+            const entry: Entry = {
+              form,
+              state: { status: "pending" },
+              deferred: yield* Deferred.make<TerminalState>(),
+              waiters: yield* Ref.make(wait ? 1 : 0),
+            }
+            yield* Cache.set(forms, id, entry)
+            yield* bus.publish(Form.Event.Created, { form }).pipe(Effect.onError(() => Cache.invalidate(forms, id)))
+            return entry
+          }),
+        ),
       ),
     )
 
+    const create = Effect.fn("Form.create")((input: CreateInput) => {
+      const id = input.id ?? ID.create()
+      return acquire(input, id, false, false).pipe(Effect.map((entry) => entry.form))
+    })
+
+    const release = (entry: Entry, id: ID, interrupted: boolean) =>
+      locks.withLock(id)(
+        Ref.updateAndGet(entry.waiters, (count) => count - 1).pipe(
+          Effect.flatMap((count) =>
+            interrupted && count === 0 ? settleCancelled(id).pipe(Effect.ignore) : Effect.void,
+          ),
+        ),
+      )
+
+    const ask = Effect.fn("Form.ask")((input: CreateInput) => {
+      const id = input.id ?? ID.create()
+      return Effect.uninterruptibleMask((restore) =>
+        acquire(input, id, input.id !== undefined, true).pipe(
+          Effect.flatMap((entry) =>
+            restore(Deferred.await(entry.deferred)).pipe(
+              Effect.tap(() => release(entry, id, false)),
+              Effect.onInterrupt(() => release(entry, id, true)),
+            ),
+          ),
+        ),
+      )
+    })
+
     const get = Effect.fn("Form.get")(function* (id: ID) {
       return (yield* find(id)).form
     })
@@ -176,32 +212,27 @@ export const layer = Layer.effect(
     })
 
     const reply = Effect.fn("Form.reply")((input: ReplyInput) =>
-      Effect.uninterruptible(
+      locks.withLock(input.id)(
-        Effect.gen(function* () {
+        Effect.uninterruptible(
-          const entry = yield* find(input.id)
+          Effect.gen(function* () {
-          if (entry.state.status !== "pending") return yield* new AlreadySettledError({ id: input.id })
+            const entry = yield* find(input.id)
-          const invalid = validateAnswer(entry.form, input.answer)
+            if (entry.state.status !== "pending") return yield* new AlreadySettledError({ id: input.id })
-          if (invalid) return yield* new InvalidAnswerError({ id: input.id, message: invalid })
+            const invalid = validateAnswer(entry.form, input.answer)
-          const next: TerminalState = { status: "answered", answer: input.answer }
+            if (invalid) return yield* new InvalidAnswerError({ id: input.id, message: invalid })
-          yield* bus.publish(Form.Event.Replied, { id: input.id, sessionID: entry.form.sessionID, answer: input.answer })
+            const next: TerminalState = { status: "answered", answer: input.answer }
-          yield* Cache.set(forms, input.id, { ...entry, state: next })
+            yield* bus.publish(Form.Event.Replied, {
-          yield* Deferred.succeed(entry.deferred, next)
+              id: input.id,
-        }),
+              sessionID: entry.form.sessionID,
+              answer: input.answer,
+            })
+            yield* Cache.set(forms, input.id, { ...entry, state: next })
+            yield* Deferred.succeed(entry.deferred, next)
+          }),
+        ),
       ),
     )
 
-    const cancel = Effect.fn("Form.cancel")((id: ID) =>
+    const cancel = Effect.fn("Form.cancel")((id: ID) => locks.withLock(id)(settleCancelled(id)))
-      Effect.uninterruptible(
-        Effect.gen(function* () {
-          const entry = yield* find(id)
-          if (entry.state.status !== "pending") return yield* new AlreadySettledError({ id })
-          const next: TerminalState = { status: "cancelled" }
-          yield* bus.publish(Form.Event.Cancelled, { id, sessionID: entry.form.sessionID })
-          yield* Cache.set(forms, id, { ...entry, state: next })
-          yield* Deferred.succeed(entry.deferred, next)
-        }),
-      ),
-    )
 
     yield* Effect.addFinalizer(() =>
       Cache.values(forms).pipe(
@@ -223,6 +254,10 @@ export const locationLayer = layer
 
 export const node = makeLocationNode({ service: Service, layer, deps: [Bus.node] })
 
+function makeInfo(input: CreateInput, id: ID): Info {
+  return Info.make({ ...input, id })
+}
+
 function validateAnswer(form: Info, answer: Answer) {
   const fields = new Map(form.fields.map((field) => [field.key, field] as const))
   for (const key of Object.keys(answer)) {

packages/core/src/tool/plugin/websearch.ts

@@ -57,6 +57,7 @@ export const Plugin = {
                       const providers = (yield* ctx.websearch.providers()).data
                       if (providers.length === 0) return yield* new WebSearch.ProviderRequiredError()
                       const response = yield* forms.ask({
+                        id: Form.ID.create(`frm_websearch_provider_${context.sessionID}_${context.messageID}`),
                         sessionID: context.sessionID,
                         title: "Choose a provider so the agent can search the web",
                         metadata: { kind: "websearch.provider" },

packages/core/test/form.test.ts

@@ -40,6 +40,65 @@ describe("Form", () => {
     }),
   )
 
+  it.effect("joins concurrent asks for the same explicit form id", () =>
+    Effect.gen(function* () {
+      const service = yield* Form.Service
+      const bus = yield* Bus.Service
+      const created = yield* Deferred.make<void>()
+      const events: Form.Info[] = []
+      const unsubscribe = yield* bus.listen((event) => {
+        if (event.type !== Form.Event.Created.type) return Effect.void
+        const form = (event.data as { readonly form: Form.Info }).form
+        events.push(form)
+        return Deferred.succeed(created, undefined).pipe(Effect.asVoid)
+      })
+      yield* Effect.addFinalizer(() => unsubscribe)
+      const first = yield* service.ask(input).pipe(Effect.forkScoped)
+      const second = yield* service.ask(input).pipe(Effect.forkScoped)
+      yield* Deferred.await(created)
+
+      expect(events).toHaveLength(1)
+      expect((yield* service.list()).map((form) => form.id)).toEqual([formID])
+
+      yield* service.reply({ id: formID, answer: { name: "Ava" } })
+      const expected = { status: "answered", answer: { name: "Ava" } } as const
+      expect(yield* Fiber.join(first)).toEqual(expected)
+      expect(yield* Fiber.join(second)).toEqual(expected)
+      expect(yield* service.ask(input)).toEqual(expected)
+      expect(events).toHaveLength(1)
+    }),
+  )
+
+  it.effect("rejects conflicting asks for the same explicit form id", () =>
+    Effect.gen(function* () {
+      const service = yield* Form.Service
+      yield* service.create(input)
+
+      expect(yield* service.ask({ ...input, title: "Different form" }).pipe(Effect.flip)).toEqual(
+        new Form.AlreadyExistsError({ id: formID }),
+      )
+
+      yield* service.cancel(formID)
+    }),
+  )
+
+  it.effect("keeps a joined form pending when one asker is interrupted", () =>
+    Effect.gen(function* () {
+      const service = yield* Form.Service
+      yield* service.create(input)
+      const first = yield* service.ask(input).pipe(Effect.forkScoped)
+      const second = yield* service.ask(input).pipe(Effect.forkScoped)
+      yield* Effect.yieldNow
+      yield* Effect.yieldNow
+
+      yield* Fiber.interrupt(first)
+
+      expect(yield* service.state(formID)).toEqual({ status: "pending" })
+      yield* service.reply({ id: formID, answer: { name: "Ava" } })
+      expect(yield* Fiber.join(second)).toEqual({ status: "answered", answer: { name: "Ava" } })
+    }),
+  )
+
   it.effect("supports the temporary global mcp elicitation owner", () =>
     Effect.gen(function* () {
       const service = yield* Form.Service

packages/core/test/tool-websearch.test.ts

@@ -30,6 +30,8 @@ const webSearchToolNode = makeLocationNode({
 const sessionID = Session.ID.make("ses_websearch_test")
 const assertions: Permission.AssertInput[] = []
 const queries: WebSearch.Input[] = []
+const forms: Form.CreateInput[] = []
+let providerRequired = false
 let result = new WebSearch.Response({
   providerID: WebSearch.ID.make("exa"),
   results: [{ url: "https://example.com", title: "Search results", content: "search results", time: {} }],
@@ -38,6 +40,8 @@ let result = new WebSearch.Response({
 beforeEach(() => {
   assertions.length = 0
   queries.length = 0
+  forms.length = 0
+  providerRequired = false
   result = new WebSearch.Response({
     providerID: WebSearch.ID.make("exa"),
     results: [{ url: "https://example.com", title: "Search results", content: "search results", time: {} }],
@@ -60,11 +64,12 @@ const websearch = Layer.succeed(
   WebSearch.Service.of({
     transform: () => Effect.die("unused"),
     reload: () => Effect.die("unused"),
-    providers: () => Effect.succeed([]),
+    providers: () => Effect.succeed(providerRequired ? [{ id: WebSearch.ID.make("exa"), name: "Exa" }] : []),
     default: () => Effect.succeed(undefined),
     query: (input) =>
-      Effect.sync(() => {
+      Effect.gen(function* () {
         queries.push(input)
+        if (providerRequired && queries.length === 1) return yield* new WebSearch.ProviderRequiredError()
         return result
       }),
   }),
@@ -73,7 +78,11 @@ const form = Layer.succeed(
   Form.Service,
   Form.Service.of({
     create: () => Effect.die("unused"),
-    ask: () => Effect.die("unused"),
+    ask: (input) =>
+      Effect.sync(() => {
+        forms.push(input)
+        return { status: "answered" as const, answer: { provider: "exa" } }
+      }),
     get: () => Effect.die("unused"),
     list: () => Effect.die("unused"),
     state: () => Effect.die("unused"),
@@ -90,16 +99,13 @@ const kv = Layer.succeed(
   }),
 )
 const it = testEffect(
-  AppNodeBuilder.build(
+  AppNodeBuilder.build(LayerNode.group([Tool.node, WebSearch.node, webSearchToolNode]), [
-    LayerNode.group([Tool.node, WebSearch.node, webSearchToolNode]),
+    [Permission.node, permission],
-    [
+    [WebSearch.node, websearch],
-      [Permission.node, permission],
+    [Form.node, form],
-      [WebSearch.node, websearch],
+    [KV.node, kv],
-      [Form.node, form],
+    [Image.node, imagePassthrough],
-      [KV.node, kv],
+  ]),
-      [Image.node, imagePassthrough],
-    ],
-  ),
 )
 
 describe("WebSearchTool registration", () => {
@@ -202,4 +208,26 @@ describe("WebSearchTool registration", () => {
       })
     }),
   )
+
+  it.effect("gives provider selection a stable session-scoped form id", () =>
+    Effect.gen(function* () {
+      providerRequired = true
+      const registry = yield* Tool.Service
+
+      expect(
+        yield* executeTool(registry, {
+          sessionID,
+          ...toolIdentity,
+          call: { type: "tool-call", id: "call-provider", name: "websearch", input: { query: "effect" } },
+        }),
+      ).toMatchObject({ status: "completed" })
+      expect(forms).toMatchObject([
+        {
+          id: `frm_websearch_provider_${sessionID}_${toolIdentity.messageID}`,
+          sessionID,
+          metadata: { kind: "websearch.provider" },
+        },
+      ])
+    }),
+  )
 })