From aa6778bf3a92eef254e0a2c20c76049c10126e1d Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 12:13:12 -0700 Subject: [PATCH 1/3] Add tests for ask_user postQuestion retry (CL-7248) A crash between posting the question card and persisting the gate must not mint a second card. These tests pin a stable questionId on retry and a no-op re-post of the same id. --- .../test/workflow-participant-routes.test.ts | 95 +++++++++++++++++++ packages/interaction-tools/src/client.test.ts | 29 ++++++ packages/interaction-tools/src/tool.test.ts | 51 +++++++++- 3 files changed, 173 insertions(+), 2 deletions(-) diff --git a/packages/chat/test/workflow-participant-routes.test.ts b/packages/chat/test/workflow-participant-routes.test.ts index 1127a2787..4b860a0af 100644 --- a/packages/chat/test/workflow-participant-routes.test.ts +++ b/packages/chat/test/workflow-participant-routes.test.ts @@ -266,6 +266,101 @@ describe("POST /participants/messages", () => { ).toBe(true); }); + test("re-posting a question block with the same questionId returns the existing card and does not insert a second message", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: TENANT.id, + workbenchId: "chan_1", + settings: { + "chat/kind": "workbench", + "chat/participants": [{ address: RUN_ADDRESS, handle: "myra" }], + }, + updatedBy: "prn_1", + }); + const roomMessages = createInMemoryRoomMessageStore(); + const app = buildApp({ store, roomMessages }); + const questionBlock = { + kind: "block" as const, + block: { + type: "question", + data: { + questionId: "q_retry1", + question: "Which environment?", + options: ["Staging", "Production"], + }, + }, + }; + const post = () => + app.request("/participants/messages", { + method: "POST", + headers: { "content-type": "application/json", ...AUTH_HEADERS }, + body: JSON.stringify({ parts: [questionBlock] }), + }); + + const first = await post(); + expect(first.status).toBe(201); + const firstBody = (await first.json()) as { id: string; createdAt: string }; + + const second = await post(); + expect(second.ok).toBe(true); + const secondBody = (await second.json()) as { + id: string; + createdAt: string; + }; + expect(secondBody.id).toBe(firstBody.id); + expect(secondBody.createdAt).toBe(firstBody.createdAt); + + const listed = await roomMessages.listMessages({ + tenantId: TENANT.id, + workbenchId: "chan_1", + }); + expect(listed.items).toHaveLength(1); + expect(listed.items[0]?.id).toBe(firstBody.id); + }); + + test("a different questionId still posts a second card", async () => { + const store = createInMemoryChatStore(); + await store.createWorkbenchSettings({ + tenantId: TENANT.id, + workbenchId: "chan_1", + settings: { + "chat/kind": "workbench", + "chat/participants": [{ address: RUN_ADDRESS, handle: "myra" }], + }, + updatedBy: "prn_1", + }); + const roomMessages = createInMemoryRoomMessageStore(); + const app = buildApp({ store, roomMessages }); + const post = (questionId: string) => + app.request("/participants/messages", { + method: "POST", + headers: { "content-type": "application/json", ...AUTH_HEADERS }, + body: JSON.stringify({ + parts: [ + { + kind: "block", + block: { + type: "question", + data: { + questionId, + question: "Which environment?", + options: ["Staging", "Production"], + }, + }, + }, + ], + }), + }); + + expect((await post("q_one")).status).toBe(201); + expect((await post("q_two")).status).toBe(201); + const listed = await roomMessages.listMessages({ + tenantId: TENANT.id, + workbenchId: "chan_1", + }); + expect(listed.items).toHaveLength(2); + }); + test("posting a connect-service block records the pending connection on the workbench and publishes chat.settings", async () => { const store = createInMemoryChatStore(); await store.createWorkbenchSettings({ diff --git a/packages/interaction-tools/src/client.test.ts b/packages/interaction-tools/src/client.test.ts index d6f63a10a..c1b4fc6e1 100644 --- a/packages/interaction-tools/src/client.test.ts +++ b/packages/interaction-tools/src/client.test.ts @@ -61,6 +61,35 @@ test("postQuestion posts a question block to the workflow-chat participants/mess ); }); +test("postQuestion stamps a caller-supplied questionId on the card instead of minting", async () => { + let seenBody: unknown; + const fetchImpl = (async (_url: string | URL, init?: RequestInit) => { + seenBody = JSON.parse(String(init?.body)); + return new Response( + JSON.stringify({ id: "msg_1", createdAt: "2026-08-17T00:00:00.000Z" }), + { status: 201 }, + ); + }) as unknown as typeof fetch; + + const result = await postQuestion(testConfig(fetchImpl), { + question: "Which environment?", + options: ["Staging", "Production"], + questionId: "q_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + }); + + const body = seenBody as { + parts: readonly { + kind: string; + block: { type: string; data: Record }; + }[]; + }; + expect(body.parts[0]?.block.data["questionId"]).toBe( + "q_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ); + expect(result.questionId).toBe("q_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"); + expect(result.messageId).toBe("msg_1"); +}); + test("a 404 from the route surfaces as NoOwnChannelError", async () => { const fetchImpl = (async () => new Response( diff --git a/packages/interaction-tools/src/tool.test.ts b/packages/interaction-tools/src/tool.test.ts index 4a9c6d9df..3ded2673e 100644 --- a/packages/interaction-tools/src/tool.test.ts +++ b/packages/interaction-tools/src/tool.test.ts @@ -79,11 +79,58 @@ test("ask_user posts a question block and suspends on a message_response gate", expect(decision.pendingOp.kind).toBe("message_response"); expect(decision.pendingOp.suspendedCall).toEqual(call); expect(decision.pendingOp.correlationId).toBe(decision.gate.correlationId); - // Minted by `postQuestion` (`q_`), not a separately-minted - // `crypto.randomUUID()` — the gate and the question card share one id. + // Derived from the tool-call id (`q_`), not minted per attempt — + // the gate and the question card share one id, and a retry of the same + // call reuses it. expect(decision.gate.correlationId).toMatch(/^q_[0-9a-f]{32}$/); }); +test("retrying ask_user for the same call reuses the questionId so a crash between post and suspend cannot orphan a second card", async () => { + const postedQuestionIds: string[] = []; + const fetchImpl = (async (_url: string | URL, init?: RequestInit) => { + const body = JSON.parse(String(init?.body)) as { + parts: readonly { + block: { data: Record }; + }[]; + }; + const questionId = body.parts[0]?.block.data["questionId"]; + if (typeof questionId === "string") postedQuestionIds.push(questionId); + return new Response( + JSON.stringify({ + id: `msg_${postedQuestionIds.length}`, + createdAt: "2026-08-17T00:00:00.000Z", + }), + { status: 201 }, + ); + }) as unknown as typeof fetch; + + const bundle = interactionTools(testEnv()); + const call = { + id: "call_1", + name: ASK_USER_TOOL, + arguments: { + question: "Which environment?", + options: ["Staging", "Production"], + }, + }; + + const first = await withFetch(fetchImpl, () => beforeTool(bundle, call)); + const second = await withFetch(fetchImpl, () => beforeTool(bundle, call)); + + expect(postedQuestionIds).toHaveLength(2); + const reusedId = postedQuestionIds[0]; + if (reusedId === undefined) { + throw new Error("expected a posted questionId"); + } + expect(postedQuestionIds[1]).toBe(reusedId); + expect(reusedId).toMatch(/^q_[0-9a-f]{32}$/); + if (first.type !== "suspend" || second.type !== "suspend") { + throw new Error("expected both retries to suspend"); + } + expect(first.gate.correlationId).toBe(reusedId); + expect(second.gate.correlationId).toBe(reusedId); +}); + test("ask_user rejects fewer than 2 options before ever posting", async () => { let posted = false; const fetchImpl = (async () => { From 01501bb218ce71ef77200f09c907419547c9687c Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 12:13:17 -0700 Subject: [PATCH 2/3] Make postQuestion idempotent on question id (CL-7248) Derive the card id from the tool-call id so a retry of the same ask_user call re-posts the same question. The participants/messages route returns the existing card instead of inserting a second one. --- .../chat/src/workflow-participant-routes.ts | 52 +++++++++++++++++++ packages/interaction-tools/src/client.ts | 52 ++++++++++++++----- packages/interaction-tools/src/tool.ts | 24 +++++---- 3 files changed, 106 insertions(+), 22 deletions(-) diff --git a/packages/chat/src/workflow-participant-routes.ts b/packages/chat/src/workflow-participant-routes.ts index f04312bb1..1a446d14d 100644 --- a/packages/chat/src/workflow-participant-routes.ts +++ b/packages/chat/src/workflow-participant-routes.ts @@ -60,6 +60,8 @@ import { } from "./workbench-service"; import type { ChatStore } from "./store"; import { Part, type Part as PartType } from "./parts"; +import { QuestionBlockData } from "./blocks"; +import type { RoomMessage, RoomMessageStore } from "./room-messages"; import { CONNECTIONS_PENDING_KEY, connectServiceConnectorIds, @@ -108,6 +110,37 @@ const MintDmInput = type({ definitionId: "string > 0" }); const PostMessageInput = type({ parts: Part.array() }); +function questionIdFromParts(parts: readonly PartType[]): string | undefined { + for (const part of parts) { + if (part.kind !== "block" || part.block.type !== "question") continue; + const parsed = QuestionBlockData(part.block.data); + if (parsed instanceof type.errors) continue; + return parsed.questionId; + } + return undefined; +} + +async function existingQuestionMessage( + roomMessages: Pick, + tenantId: string, + workbenchId: string, + questionId: string, +): Promise { + let cursor: string | undefined; + for (;;) { + const page = await roomMessages.listMessages({ + tenantId, + workbenchId, + ...(cursor !== undefined ? { cursor } : {}), + }); + for (const item of page.items) { + if (questionIdFromParts(item.parts) === questionId) return item; + } + if (page.nextCursor === undefined) return undefined; + cursor = page.nextCursor; + } +} + export type CreateWorkflowParticipantRoutesDeps = { readonly store: Pick< ChatStore, @@ -432,6 +465,25 @@ export function createWorkflowParticipantRoutes( ); } + // Re-posting a question card with the same questionId is a no-op that + // returns the existing message (CL-7248). `ask_user` derives that id + // from the tool-call id, so a crash between the first post and + // `suspendOnGate` retries the same id instead of orphaning a second + // card. Looked up before any other write so the retry also skips + // connect-service pending-set mutation. + const questionId = questionIdFromParts(body.parts); + if (questionId !== undefined) { + const existing = await existingQuestionMessage( + deps.roomMessages, + scope.tenantId, + workbench.workbenchId, + questionId, + ); + if (existing !== undefined) { + return c.json({ id: existing.id, createdAt: existing.createdAt }, 200); + } + } + // A connect-service card registers its connector on the room's own // settings before the message lands, so a later connection completing // in the browser can find this room and settle the card diff --git a/packages/interaction-tools/src/client.ts b/packages/interaction-tools/src/client.ts index b36a6bb00..10bf44591 100644 --- a/packages/interaction-tools/src/client.ts +++ b/packages/interaction-tools/src/client.ts @@ -4,6 +4,7 @@ // own channel" route the same bundle family (`@corbits/agent-directory-tools`) // already reaches for `participants/invite`. Same auth-header shape, same // error-handling pattern as `@corbits/agent-directory-tools`'s `client.ts`. +import { createHash } from "node:crypto"; import { type } from "arktype"; export interface AskUserClientConfig { @@ -19,6 +20,13 @@ export interface AskUserQuestion { readonly subtitle?: string; readonly options: readonly string[]; readonly allowFreeText?: boolean; + /** + * Stable id for this card. When omitted, `postQuestion` mints one. + * `ask_user` always supplies `questionIdForCall(call.id)` so a retry of + * the same tool call re-posts the same id — the participants/messages + * route treats that as a no-op and returns the existing card (CL-7248). + */ + readonly questionId?: string; } function authHeaders(config: AskUserClientConfig): Record { @@ -62,25 +70,43 @@ export class NoOwnChannelError extends Error {} const PostedMessageResponse = type({ id: "string", createdAt: "string" }); /** - * Posts a `question` block into the caller's own channel. Mints the - * block's `questionId` here (never trusts the model to supply a stable, - * collision-free id) and returns it: `beforeAskUser` (`./tool.ts`) reuses - * it verbatim as the `message_response` gate's own `correlationId`, and - * the block-response route persists (and later relays) an answer keyed on - * this same id as `blockId` — a question block's `blockId` IS its - * `questionId`, the same way a poll's is its `pollId` - * (`packages/chat/src/schema.ts`'s `block_responses` table comment). - * `@intx/hub-common`'s `generateId` is a closed enum of platform id kinds - * (vendored, read-only source) with no "question" entry, so this mints its - * own `q_`-prefixed id the same way `packages/chat/src/threads.ts`'s - * `thr_` ids do. + * Derives the question card's `questionId` from a tool-call id. The call + * id is assigned by the runtime, not the model, so it is stable across a + * crash-retry of the same `ask_user` invocation — hashing it into the + * `q_` shape `postQuestion` otherwise mints keeps the card and the + * `message_response` gate on one id without trusting model-supplied input. + */ +export function questionIdForCall(callId: string): string { + return `q_${createHash("sha256").update(callId).digest("hex").slice(0, 32)}`; +} + +/** + * Posts a `question` block into the caller's own channel. Uses a caller- + * supplied `questionId` when present (never trusts the model to supply a + * stable, collision-free id — `ask_user` derives it from the tool-call id) + * and otherwise mints one here. Returns the id: `beforeAskUser` + * (`./tool.ts`) reuses it verbatim as the `message_response` gate's own + * `correlationId`, and the block-response route persists (and later + * relays) an answer keyed on this same id as `blockId` — a question + * block's `blockId` IS its `questionId`, the same way a poll's is its + * `pollId` (`packages/chat/src/schema.ts`'s `block_responses` table + * comment). `@intx/hub-common`'s `generateId` is a closed enum of + * platform id kinds (vendored, read-only source) with no "question" + * entry, so a minted id is `q_`-prefixed the same way + * `packages/chat/src/threads.ts`'s `thr_` ids are. + * + * Re-posting the same `questionId` into the same workbench is a no-op: + * the participants/messages route returns the existing card rather than + * inserting a second one, so a crash between this post and + * `suspendOnGate` cannot orphan a duplicate on retry. */ export async function postQuestion( config: AskUserClientConfig, question: AskUserQuestion, ): Promise<{ readonly messageId: string; readonly questionId: string }> { const doFetch = config.fetchImpl ?? fetch; - const questionId = `q_${crypto.randomUUID().replace(/-/g, "")}`; + const questionId = + question.questionId ?? `q_${crypto.randomUUID().replace(/-/g, "")}`; const response = await doFetch( `${config.hubChatUrl}/api/workflow-chat/participants/messages`, { diff --git a/packages/interaction-tools/src/tool.ts b/packages/interaction-tools/src/tool.ts index 7a8804482..ca26971c9 100644 --- a/packages/interaction-tools/src/tool.ts +++ b/packages/interaction-tools/src/tool.ts @@ -17,7 +17,7 @@ import type { } from "@intx/types/runtime"; import { type } from "arktype"; -import { postQuestion, NoOwnChannelError } from "./client"; +import { postQuestion, NoOwnChannelError, questionIdForCall } from "./client"; import type { AskUserClientConfig } from "./client"; export const ASK_USER_TOOL = "ask_user"; @@ -81,7 +81,10 @@ async function beforeAskUser( let questionId: string; try { - ({ questionId } = await postQuestion(clientConfig(env), parsed)); + ({ questionId } = await postQuestion(clientConfig(env), { + ...parsed, + questionId: questionIdForCall(call.id), + })); } catch (err) { if (err instanceof NoOwnChannelError) { return { type: "block", reason: err.message }; @@ -92,13 +95,16 @@ async function beforeAskUser( }; } - // `postQuestion` already mints `questionId` and stamps it on the outbound - // question card's `data.questionId`; reusing it as the gate's own - // `correlationId` (rather than minting a second, unrelated id) is what - // lets the answer route resolve this exact gate later (CL-7191) — the - // block a person answers is keyed on `blockId`, which for a question - // block IS `questionId` (`packages/chat/src/schema.ts`'s "agent-authored - // pollId/formId" comment applies identically here). + // `postQuestion` stamps this `questionId` (derived from the tool-call + // id, not minted per attempt) on the outbound question card; reusing it + // as the gate's own `correlationId` (rather than minting a second, + // unrelated id) is what lets the answer route resolve this exact gate + // later (CL-7191) — the block a person answers is keyed on `blockId`, + // which for a question block IS `questionId` (`packages/chat/src/schema.ts`'s + // "agent-authored pollId/formId" comment applies identically here). The + // same derivation is what makes a crash between post and + // `suspendOnGate` retry-safe: the next attempt re-posts this id and the + // write path returns the existing card (CL-7248). const correlationId = questionId; const timeoutAt = Date.now() + ASK_USER_TIMEOUT_MS; const gateId = `pending-${correlationId}`; From 3ecea656dfd273118f50f6aa037e01e1d03025be Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 3 Sep 2026 15:06:16 -0700 Subject: [PATCH 3/3] Bump interaction-tools to 0.0.6 (CL-7248) --- bun.lock | 2 +- packages/interaction-tools/package.json | 2 +- workflows/assistant/src/index.ts | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/bun.lock b/bun.lock index 4f2d5eb98..8f7ceaced 100644 --- a/bun.lock +++ b/bun.lock @@ -888,7 +888,7 @@ }, "packages/interaction-tools": { "name": "@corbits/interaction-tools", - "version": "0.0.5", + "version": "0.0.6", "dependencies": { "@intx/agent": "workspace:*", "@intx/types": "workspace:*", diff --git a/packages/interaction-tools/package.json b/packages/interaction-tools/package.json index c5f1f4cc3..7eb48110d 100644 --- a/packages/interaction-tools/package.json +++ b/packages/interaction-tools/package.json @@ -2,7 +2,7 @@ "name": "@corbits/interaction-tools", "private": true, "description": "The ask_user tool: an @intx/agent bundle that poses an interview question in-thread as a question block and parks the turn on a message_response gate until the user answers", - "version": "0.0.5", + "version": "0.0.6", "license": "LGPL-2.1-or-later", "type": "module", "exports": { diff --git a/workflows/assistant/src/index.ts b/workflows/assistant/src/index.ts index 62c341e91..08572f1c5 100644 --- a/workflows/assistant/src/index.ts +++ b/workflows/assistant/src/index.ts @@ -52,7 +52,7 @@ export const ASSISTANT_TOOL_PACKAGE_PINS: readonly ToolPackagePin[] = [ { name: "@corbits/catalog-tools", version: "0.0.2" }, { name: "@corbits/skills-tools", version: "0.0.6" }, { name: "@corbits/mcp-tools", version: "0.0.11" }, - { name: "@corbits/interaction-tools", version: "0.0.5" }, + { name: "@corbits/interaction-tools", version: "0.0.6" }, { name: "@corbits/manus-tools", version: "0.0.11" }, { name: "@corbits/workflow-authoring-tools", version: "0.0.4" }, ];