import { createHash, randomUUID } from "node:crypto"; import { z } from "zod"; import { readJsonFile } from "../utils/fileStore.js"; import type { SessionWorkspaceContext } from "../runtime/conversationWorkspace.js"; import { SessionContextMismatchError } from "./metadataStore.js"; import { SessionWorkspaceStore } from "./workspace-store.js"; import { INTERACTION_VERSION, renderInteractionAnswer, interactionRequestSchema, interactionAnswersSchema, interactionBlocksTurn, interactionDraftInputSchema, interactionMutationInputSchema, interactionQuestionsSchema, interactionReplyInputSchema, type InteractionAnswer, type InteractionDelivery, type InteractionRequest, type InteractionResolvedAnswer } from "../chat/interaction-request-contract.js"; type Receipt = { operation: string; request_id: string; digest: string; result: InteractionRequest }; type State = { version: typeof INTERACTION_VERSION; requests: InteractionRequest[]; receipts: Receipt[] }; const filename = "interaction-requests.json"; export class InteractionRequestError extends Error { constructor(readonly code: string, readonly status: number = 409, readonly current?: InteractionRequest) { super(code); } } const initial = (): State => ({ version: INTERACTION_VERSION, requests: [], receipts: [] }); const parseState = (value: unknown): State => { if (value === undefined || value === null) return initial(); return z.object({ version: z.literal(INTERACTION_VERSION), requests: z.array(interactionRequestSchema).max(1000), receipts: z.array(z.object({ operation: z.string(), request_id: z.string().min(1).max(128), digest: z.string(), result: interactionRequestSchema }).strict()) }).strict().parse(value); }; const copy = (value: T): T => structuredClone(value); const validateAnswers = (record: InteractionRequest, answers: InteractionAnswer[], complete: boolean): InteractionResolvedAnswer[] => { const parsed = interactionAnswersSchema.parse(answers); const result = parsed.map((answer) => { const q = record.questions.find((q) => q.question_id === answer.question_id); if (!q || (!q.allow_custom && answer.custom_text.length > 0) || (q.mode === "single" && answer.selected_option_ids.length > 1)) throw new InteractionRequestError("invalid_answer", 400); const selected_options = answer.selected_option_ids.map((id) => { const option = q.options.find((o) => o.id === id); if (!option) throw new InteractionRequestError("invalid_option", 400); return option; }); return { ...answer, selected_options }; }); if (complete && record.questions.some((q) => { const a = result.find((a) => a.question_id === q.question_id); return !a || (q.required && !a.selected_option_ids.length && !a.custom_text.trim()); })) throw new InteractionRequestError("incomplete_answer", 400); if (Buffer.byteLength(JSON.stringify(result)) > 16384 || result.reduce((n, a) => n + a.custom_text.length + a.selected_options.reduce((m, o) => m + o.label.length, 0), 0) > 10000) throw new InteractionRequestError("answer_too_large", 400); return result; }; export class InteractionRequestStore { constructor(readonly workspaceStore: SessionWorkspaceStore) {} private async read(context: SessionWorkspaceContext) { if (!await this.workspaceStore.validate(context, filename)) throw new SessionContextMismatchError(); return parseState(await readJsonFile(this.workspaceStore.path(context, filename))); } private async mutate(context: SessionWorkspaceContext, operation: (state: State) => T): Promise { return this.workspaceStore.updateRecord(context, filename, async ({ current }) => { const state = parseState(current); const result = operation(state); state.receipts = state.receipts.slice(-32); while (state.receipts.length && Buffer.byteLength(JSON.stringify(state.receipts)) > 128 * 1024) state.receipts.shift(); if (state.requests.length > 1000 || Buffer.byteLength(JSON.stringify(state, null, 2)) > 2 * 1024 * 1024) throw new InteractionRequestError("interaction_capacity_exceeded", 413); return { result: copy(result), value: state }; }); } async get(context: SessionWorkspaceContext, id: string) { return copy((await this.read(context)).requests.find((r) => r.interaction_id === id) ?? null); } async forMessages(context: SessionWorkspaceContext, messageIds: ReadonlySet) { if (!messageIds.size) return []; return copy((await this.read(context)).requests.filter((record) => record.delivery && messageIds.has(record.delivery.native_message_id))); } async list(context: SessionWorkspaceContext, options: { limit?: number; cursor?: string } = {}) { const limit = z.number().int().min(1).max(50).parse(options.limit ?? 20); const requests = [...(await this.read(context)).requests].reverse(); const start = options.cursor ? requests.findIndex((r) => r.interaction_id === options.cursor) + 1 : 0; if (options.cursor && !start) throw new InteractionRequestError("invalid_cursor", 400); const items = requests.slice(start, start + limit); return copy({ items, next_cursor: start + limit < requests.length ? items.at(-1)?.interaction_id : undefined }); } async getBlocking(context: SessionWorkspaceContext) { return copy((await this.read(context)).requests.find(interactionBlocksTurn) ?? null); } async getForTurn(context: SessionWorkspaceContext, turnId: string) { return copy((await this.read(context)).requests.find((record) => record.turn_id === turnId) ?? null); } async create(context: SessionWorkspaceContext, input: { message_id: string; turn_id: string; call_id: string; questions: unknown }) { const parsed = z.object({ message_id: z.string().min(1).max(128), turn_id: z.string().min(1).max(128), call_id: z.string().min(1).max(128), questions: interactionQuestionsSchema }).strict().parse(input); return this.mutate(context, (state) => { const existing = state.requests.find((r) => r.turn_id === parsed.turn_id && r.call_id === parsed.call_id); if (existing) { if (JSON.stringify(existing.questions) !== JSON.stringify(parsed.questions) || existing.message_id !== parsed.message_id) throw new InteractionRequestError("idempotency_mismatch"); return existing; } if (state.requests.some(interactionBlocksTurn)) throw new InteractionRequestError("interaction_unresolved"); const record: InteractionRequest = { version: INTERACTION_VERSION, interaction_id: randomUUID(), ...parsed, status: "pending", revision: 0, draft_revision: 0, draft: [], created_at: new Date().toISOString() }; state.requests.push(record); return record; }); } private async apply(context: SessionWorkspaceContext, id: string, operation: string, input: unknown, draft: boolean, change: (record: InteractionRequest, input: Record) => void) { const parsed = (draft ? interactionDraftInputSchema : operation === "reply" ? interactionReplyInputSchema : interactionMutationInputSchema).parse(input); return this.mutate(context, (state) => { const digest = createHash("sha256").update(JSON.stringify(parsed)).digest("hex"); const previous = state.receipts.find((r) => r.request_id === parsed.request_id); if (previous) { if (previous.operation !== `${id}:${operation}` || previous.digest !== digest) throw new InteractionRequestError("idempotency_mismatch"); return state.requests.find((r) => r.interaction_id === id) ?? previous.result; } const record = state.requests.find((r) => r.interaction_id === id); if (!record) throw new InteractionRequestError("interaction_not_found", 404); if (record.status !== "pending") throw new InteractionRequestError("interaction_resolved", 409, copy(record)); const expected = "expected_draft_revision" in parsed ? parsed.expected_draft_revision : parsed.expected_revision; if (expected !== (draft ? record.draft_revision : record.revision)) throw new InteractionRequestError("revision_conflict", 409, copy(record)); change(record, parsed); if (draft) record.draft_revision++; else { record.revision++; record.resolved_at = new Date().toISOString(); } state.receipts.push({ operation: `${id}:${operation}`, request_id: parsed.request_id, digest, result: copy(record) }); return record; }); } updateDraft(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "draft", input, true, (r, p) => { validateAnswers(r, p.answers as InteractionAnswer[], false); r.draft = p.answers as InteractionAnswer[]; }); } reply(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "reply", input, false, (r, p) => { r.answers = validateAnswers(r, p.answers as InteractionAnswer[], true); r.status = "answered"; try { renderInteractionAnswer(r); } catch { throw new InteractionRequestError("answer_too_large", 400); } r.delivery = this.delivery(); }); } skip(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "skip", input, false, (r) => { if (r.questions.some((q) => q.required)) throw new InteractionRequestError("required_question", 400); r.status = "skipped"; r.answers = []; r.delivery = this.delivery(); }); } cancel(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "cancel", input, false, (r) => { r.status = "cancelled"; }); } private delivery(): InteractionDelivery { return { native_message_id: `msg_${randomUUID().replaceAll("-", "")}`, revision: 0, status: "queued" }; } async updateDelivery(context: SessionWorkspaceContext, id: string, input: { expected_revision: number; status: InteractionDelivery["status"]; response_id?: string; error_code?: string }) { const parsed = z.object({ expected_revision: z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER), status: z.enum(["queued", "dispatching", "delivered", "blocked", "cancelled"]), response_id: z.string().min(1).max(128).optional(), error_code: z.string().min(1).max(128).optional() }).strict().parse(input); return this.mutate(context, (state) => { const r = state.requests.find((r) => r.interaction_id === id); if (!r?.delivery) throw new InteractionRequestError("delivery_not_found", 404); if (r.delivery.revision !== parsed.expected_revision) throw new InteractionRequestError("revision_conflict", 409, copy(r)); if (["delivered", "cancelled"].includes(r.delivery.status)) throw new InteractionRequestError("delivery_resolved"); r.delivery = { native_message_id: r.delivery.native_message_id, revision: r.delivery.revision + 1, status: parsed.status, ...(parsed.response_id ? { response_id: parsed.response_id } : {}), ...(parsed.error_code ? { error_code: parsed.error_code } : {}) }; return r; }); } async resume(context: SessionWorkspaceContext, id: string, input: unknown) { const parsed = interactionMutationInputSchema.parse(input); return this.mutate(context, (state) => { const digest = createHash("sha256").update(JSON.stringify(parsed)).digest("hex"); const previous = state.receipts.find((r) => r.request_id === parsed.request_id); if (previous) { if (previous.operation !== `${id}:resume` || previous.digest !== digest) throw new InteractionRequestError("idempotency_mismatch"); return state.requests.find((r) => r.interaction_id === id) ?? previous.result; } const record = state.requests.find((r) => r.interaction_id === id); if (!record?.delivery) throw new InteractionRequestError("delivery_not_found", 404); if (record.delivery.revision !== parsed.expected_revision) throw new InteractionRequestError("revision_conflict", 409, copy(record)); state.receipts.push({ operation: `${id}:resume`, request_id: parsed.request_id, digest, result: copy(record) }); return record; }); } claimDelivery(context: SessionWorkspaceContext, id: string, input: { expected_revision: number; response_id: string }) { return this.updateDelivery(context, id, { ...input, status: "dispatching" }); } async invalidate(context: SessionWorkspaceContext, id: string, input: unknown) { return this.apply(context, id, "invalidate", input, false, (r) => { r.status = "invalidated"; }); } async cancelPending(context: SessionWorkspaceContext) { return this.mutate(context, (state) => { for (const r of state.requests) { if (r.status === "pending") { r.status = "cancelled"; r.revision++; r.resolved_at = new Date().toISOString(); } if (r.delivery && ["queued", "blocked"].includes(r.delivery.status)) { r.delivery.status = "cancelled"; r.delivery.revision++; } } return state.requests; }); } }