From 81157c8aa36307bd7f4fd7f3bd1b2af522c13f40 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Mon, 27 Jul 2026 20:08:56 +0000 Subject: [PATCH] fix(core): coalesce duplicate form asks --- packages/core/src/form.ts | 143 +++++++++++++-------- packages/core/src/tool/plugin/websearch.ts | 1 + packages/core/test/form.test.ts | 59 +++++++++ packages/core/test/tool-websearch.test.ts | 54 ++++++-- 4 files changed, 190 insertions(+), 67 deletions(-) diff --git a/packages/core/src/form.ts b/packages/core/src/form.ts index 0a23ef3efe..7a265198fa 100644 --- a/packages/core/src/form.ts +++ b/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 + readonly waiters: Ref.Ref } export const layer = Layer.effect( Service, Effect.gen(function* () { const bus = yield* Bus.Service + const locks = KeyedMutex.makeUnsafe() const forms = yield* Cache.makeWith( () => 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 existing = yield* Cache.getSuccess(forms, id) - if (Option.isSome(existing)) return yield* new AlreadyExistsError({ id }) - const invalid = validateFields(input.fields) - if (invalid) return yield* new InvalidFormError({ message: invalid }) - const form: Info = { - 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(), - } - yield* Cache.set(forms, id, entry) - yield* bus.publish(Form.Event.Created, { form }).pipe(Effect.onError(() => Cache.invalidate(forms, id))) - return form + 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) }), ), ) - const ask = Effect.fn("Form.ask")((input: CreateInput) => - Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const form = yield* create(input) - const entry = yield* find(form.id).pipe(Effect.orDie) - return yield* restore(Deferred.await(entry.deferred)).pipe( - Effect.onInterrupt(() => Effect.ignore(cancel(form.id))), - ) - }), + const acquire = Effect.fn("Form.acquire")((input: CreateInput, id: ID, join: boolean, wait: boolean) => + locks.withLock(id)( + Effect.uninterruptible( + Effect.gen(function* () { + const existing = yield* Cache.getSuccess(forms, id) + if (Option.isSome(existing)) { + 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(), + 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( - Effect.gen(function* () { - const entry = yield* find(input.id) - if (entry.state.status !== "pending") return yield* new AlreadySettledError({ id: input.id }) - const invalid = validateAnswer(entry.form, input.answer) - if (invalid) return yield* new InvalidAnswerError({ id: input.id, message: invalid }) - const next: TerminalState = { status: "answered", answer: input.answer } - yield* bus.publish(Form.Event.Replied, { 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) - }), + locks.withLock(input.id)( + Effect.uninterruptible( + Effect.gen(function* () { + const entry = yield* find(input.id) + if (entry.state.status !== "pending") return yield* new AlreadySettledError({ id: input.id }) + const invalid = validateAnswer(entry.form, input.answer) + if (invalid) return yield* new InvalidAnswerError({ id: input.id, message: invalid }) + const next: TerminalState = { status: "answered", answer: input.answer } + yield* bus.publish(Form.Event.Replied, { + 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) => - 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) - }), - ), - ) + const cancel = Effect.fn("Form.cancel")((id: ID) => locks.withLock(id)(settleCancelled(id))) 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)) { diff --git a/packages/core/src/tool/plugin/websearch.ts b/packages/core/src/tool/plugin/websearch.ts index 3269e5dccb..1f8315aa32 100644 --- a/packages/core/src/tool/plugin/websearch.ts +++ b/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" }, diff --git a/packages/core/test/form.test.ts b/packages/core/test/form.test.ts index 5ee579668b..75fba2c22c 100644 --- a/packages/core/test/form.test.ts +++ b/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() + 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 diff --git a/packages/core/test/tool-websearch.test.ts b/packages/core/test/tool-websearch.test.ts index 9bd6815f22..a6038902b2 100644 --- a/packages/core/test/tool-websearch.test.ts +++ b/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" }, + }, + ]) + }), + ) })