fix(core): coalesce duplicate form asks

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(
-    LayerNode.group([Tool.node, WebSearch.node, webSearchToolNode]),
-    [
-      [Permission.node, permission],
-      [WebSearch.node, websearch],
-      [Form.node, form],
-      [KV.node, kv],
-      [Image.node, imagePassthrough],
-    ],
-  ),
+  AppNodeBuilder.build(LayerNode.group([Tool.node, WebSearch.node, webSearchToolNode]), [
+    [Permission.node, permission],
+    [WebSearch.node, websearch],
+    [Form.node, form],
+    [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" },
+        },
+      ])
+    }),
+  )
 })