diff --git a/apps/server/src/conversation-plan/effect-fields.ts b/apps/server/src/conversation-plan/effect-fields.ts new file mode 100644 index 000000000..bd939f221 --- /dev/null +++ b/apps/server/src/conversation-plan/effect-fields.ts @@ -0,0 +1,35 @@ +import { createHash } from "node:crypto"; +// Extracted from archive 446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.ts. + +export const MAX_EFFECTS = 1024; + +export function invalid(): never { + throw new Error("hosted channel has invalid conversation effects"); +} + +export function object( + value: unknown, + keys: string[], + optional: string[] = [], +): Record { + if (!value || typeof value !== "object" || Array.isArray(value)) return invalid(); + let record = value as Record; + if ( + Object.keys(record).some(key => !keys.includes(key) && !optional.includes(key)) + || keys.some(key => !Object.hasOwn(record, key)) + ) return invalid(); + return record; +} + +export function bounded(value: unknown, max: number): value is string { + return typeof value === "string" && !!value.trim() && value.length <= max; +} + +export function scopedKey(proposalId: string, triggerEventId: string): string { + let raw = triggerEventId === proposalId + ? `scoped-choice:${proposalId}:proposal` + : `scoped-choice:${proposalId}:agreement:${triggerEventId}`; + return raw.length <= 300 + ? raw + : `scoped-choice:${createHash("sha256").update(raw).digest("hex")}`; +} diff --git a/apps/server/src/conversation-plan/effect-outbox.ts b/apps/server/src/conversation-plan/effect-outbox.ts new file mode 100644 index 000000000..22cf97728 --- /dev/null +++ b/apps/server/src/conversation-plan/effect-outbox.ts @@ -0,0 +1,164 @@ +import { ULID } from "@chopin/dialect"; +import * as Question from "@chopin/question"; +import { assertSourceShape } from "./sources"; +import type { Effect } from "./effect-types"; +import { bounded, invalid, MAX_EFFECTS, object, scopedKey } from "./effect-fields"; +// Extracted from archive 446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.ts. + +export function restoreEffectOutbox(pendingValue: unknown, receiptValue: unknown): { + pending: Effect[]; + receipts: string[]; +} { + let pending = pendingValue === undefined ? [] : pendingValue; + let receipts = receiptValue === undefined ? [] : receiptValue; + if ( + !Array.isArray(pending) || pending.length > MAX_EFFECTS + || !Array.isArray(receipts) || receipts.length > MAX_EFFECTS + ) return invalid(); + let keys = new Set(); + for (let raw of pending) { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) return invalid(); + let base = raw as Record; + if (!bounded(base.key, 300) || keys.has(base.key)) return invalid(); + keys.add(base.key); + let item: Record; + if (base.kind === "insert-card") { + item = object(raw, ["key", "kind", "threadId", "header", "question", "options", "trigger"]); + if ( + item.key !== `insert:${item.threadId}` || !bounded(item.threadId, 200) + || !bounded(item.header, Question.limits.MAX_HEADER) + || !bounded(item.question, 500) || !bounded(item.trigger, 200) + || !Array.isArray(item.options) + || item.options.length > Question.limits.MAX_DECISION_OPTIONS + ) return invalid(); + let ids = new Set(); + for (let rawOption of item.options) { + let option = object(rawOption, ["id", "label"], ["source"]); + if ( + typeof option.id !== "string" || !ULID.test(option.id) + || !bounded(option.label, Question.limits.MAX_LABEL) + || ids.has(option.id as string) + ) return invalid(); + if (option.source !== undefined) { + try { + assertSourceShape(option.source); + if (option.source.role !== "option") return invalid(); + } catch { + return invalid(); + } + } + ids.add(option.id as string); + } + } else if (base.kind === "add-option") { + item = object(raw, ["key", "kind", "threadId", "optionId", "label", "trigger"], [ + "source", + ]); + if ( + item.key !== `option:${item.optionId}` || !bounded(item.threadId, 200) + || typeof item.optionId !== "string" || !ULID.test(item.optionId) + || !bounded(item.label, Question.limits.MAX_LABEL) + || !bounded(item.trigger, 200) + ) return invalid(); + if (item.source !== undefined) { + try { + assertSourceShape(item.source); + if (item.source.role !== "option") return invalid(); + } catch { + return invalid(); + } + } + } else if (base.kind === "suggest") { + item = object(raw, ["key", "kind", "threadId", "messageIds"], ["optionId"]); + if ( + !String(item.key).startsWith("suggest:") || !bounded(item.threadId, 200) + || item.optionId !== undefined + && (typeof item.optionId !== "string" || !ULID.test(item.optionId)) + || !Array.isArray(item.messageIds) || item.messageIds.length > 8 + || item.messageIds.some(id => !bounded(id, 200)) + || item.optionId === undefined && item.messageIds.length !== 0 + ) return invalid(); + } else if (base.kind === "prompt") { + item = object(raw, ["key", "kind", "threadId", "messageId", "generation"], [ + "optionId", + "sourceMessageIds", + ]); + if ( + !String(item.key).startsWith("prompt:") || String(item.key).length <= 7 + || !bounded(item.threadId, 200) + || item.optionId !== undefined + && (typeof item.optionId !== "string" || !ULID.test(item.optionId)) + || !bounded(item.messageId, 200) + || !Number.isSafeInteger(item.generation) || (item.generation as number) < 0 + || item.sourceMessageIds !== undefined && ( + !Array.isArray(item.sourceMessageIds) || item.sourceMessageIds.length > 8 + || item.sourceMessageIds.some(id => !bounded(id, 200)) + || new Set(item.sourceMessageIds).size !== item.sourceMessageIds.length + ) + ) return invalid(); + } else if (base.kind === "defer-prompt") { + item = object(raw, ["key", "kind", "threadId", "proposalId", "deferredEventId"]); + if ( + !bounded(item.threadId, 200) || !bounded(item.proposalId, 200) + || !bounded(item.deferredEventId, 200) + || item.key !== `defer-prompt:${item.deferredEventId}` + ) return invalid(); + } else if (base.kind === "scoped-choice") { + item = object(raw, [ + "key", + "kind", + "threadId", + "proposalId", + "cardId", + "optionId", + "label", + "scope", + "generation", + "triggerEventId", + "sources", + ]); + if ( + !bounded(item.threadId, 200) || !bounded(item.proposalId, 200) + || !bounded(item.cardId, 200) || !ULID.test(item.optionId as string) + || !bounded(item.label, Question.limits.MAX_LABEL) || item.scope !== "spike" + || !Number.isSafeInteger(item.generation) || (item.generation as number) < 0 + || !bounded(item.triggerEventId, 200) + || item.key !== scopedKey(item.proposalId, item.triggerEventId) + || !Array.isArray(item.sources) + || item.sources.length !== (item.triggerEventId === item.proposalId ? 1 : 2) + ) return invalid(); + try { + for (let [index, source] of (item.sources as unknown[]).entries()) { + assertSourceShape(source); + if ( + source.author.kind !== "member" + || index === 0 && source.role !== "support" + || index === 1 && !["support", "objection", "withdrawal"].includes(source.role) + ) return invalid(); + } + } catch { + return invalid(); + } + } else if (base.kind === "research") { + item = object(raw, ["key", "kind", "offerId"]); + if (!bounded(item.offerId, 200) || item.key !== `research:${item.offerId}`) { + return invalid(); + } + } else if (base.kind === "job") { + item = object(raw, ["key", "kind", "intent"], ["threadId"]); + let intent = object(item.intent, ["kind", "target", "trigger"]); + if ( + !String(item.key).startsWith("job:") + || item.threadId !== undefined && !bounded(item.threadId, 200) + || intent.kind === "prose" && !bounded(item.threadId, 200) + || !["heading", "refine", "suggest", "prose"].includes(intent.kind as string) + || !bounded(intent.target, 200) || !bounded(intent.trigger, 200) + ) return invalid(); + } else return invalid(); + } + let receiptKeys = new Set(); + for (let key of receipts) { + if (!bounded(key, 300) || receiptKeys.has(key) || keys.has(key)) return invalid(); + receiptKeys.add(key); + } + return { pending: pending as Effect[], receipts: receipts as string[] }; +} diff --git a/apps/server/src/conversation-plan/effect-projection.ts b/apps/server/src/conversation-plan/effect-projection.ts new file mode 100644 index 000000000..baf2908d4 --- /dev/null +++ b/apps/server/src/conversation-plan/effect-projection.ts @@ -0,0 +1,261 @@ +import type { ConversationPlan } from "@chopin/protocol"; +import { ULID } from "@chopin/dialect"; +import * as Question from "@chopin/question"; +import { currentScopedProposal, targetsScopedProposal } from "./events"; +import { QUESTION_SET_VERSION } from "./question-shared"; +import { effectivePreference } from "./preference"; +import { scopedKey } from "./effect-fields"; +import type { Effect } from "./effect-types"; +// Extracted from archive 446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.ts. + +function purposeful(analysis?: ConversationPlan.AnalysisRecord): boolean { + if ( + analysis?.questionSetVersion !== QUESTION_SET_VERSION + && analysis?.questionSetVersion !== "conversation-plan-4" + ) return false; + let triage = analysis.passes.find(pass => pass.stage === "triage"); + let answer = triage?.answers.enough_purpose; + return answer?.type === "noul" && answer.noul >= 0.75; +} + +export function effectsFor( + events: readonly ConversationPlan.Event[], + state: ConversationPlan.State, + analysis?: ConversationPlan.AnalysisRecord, + records?: ReadonlyMap, +): Effect[] { + let effects: Effect[] = []; + let opened = new Set( + events.filter(event => event.type === "thread.opened").map(event => event.threadId), + ); + for (let event of events) { + let thread = state.threads.find(item => item.id === event.threadId); + if (!thread) continue; + if (event.type === "settle.deferred") { + effects.push({ + key: `defer-prompt:${event.id}`, + kind: "defer-prompt", + threadId: thread.id, + proposalId: event.proposalId, + deferredEventId: event.id, + }); + } + if ( + event.type === "stance.changed" && event.position !== "support" + && event.source.author.kind === "member" + ) { + let participant = event.source.author.handle; + let eventIndex = state.events.findIndex(item => item.id === event.id); + let earlier = eventIndex < 0 ? [] : state.events.slice(0, eventIndex); + let proposal = earlier.findLast((item): item is Extract< + ConversationPlan.Event, + { type: "scoped-choice.proposed" } + > => + item.type === "scoped-choice.proposed" && item.threadId === thread.id + && (typeof event.scopedProposalId !== "string" + || item.id === event.scopedProposalId) + ); + if ( + !proposal || proposal.source.author.kind !== "member" + || !targetsScopedProposal(event, proposal.id, proposal.optionId) + || thread.questionnaireId !== proposal.cardId + || thread.pendingScopedChoice + && thread.pendingScopedChoice.proposalId !== proposal.id + || earlier.some(item => + item.type === "scoped-choice.saved" && item.proposalId === proposal.id + ) + || participant !== proposal.source.author.handle + && !earlier.some(item => + item.type === "scoped-choice.agreed" && item.proposalId === proposal.id + && item.source.author.kind === "member" + && item.source.author.handle === participant + ) + ) continue; + let record = records?.get(proposal.cardId); + if (records && !record) throw new Error("linked scoped choice card is missing"); + effects.push({ + key: scopedKey(proposal.id, event.id), + kind: "scoped-choice", + threadId: thread.id, + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, + generation: record?.history.length ?? 0, + triggerEventId: event.id, + sources: [proposal.source, event.source], + }); + continue; + } + if (event.type === "scoped-choice.proposed" || event.type === "scoped-choice.agreed") { + let proposal = currentScopedProposal(thread, state.events); + if ( + !proposal || !thread.questionnaireId || thread.questionnaireId !== proposal.cardId + || ["decided", "discarded"].includes(thread.status) + || state.events.some(item => + item.type === "scoped-choice.saved" && item.proposalId === proposal.id + ) + || event.type === "scoped-choice.proposed" && event.id !== proposal.id + || event.type === "scoped-choice.agreed" && event.proposalId !== proposal.id + || event.type === "scoped-choice.agreed" && ( + event.cardId !== proposal.cardId || event.optionId !== proposal.optionId + || event.label !== proposal.label || event.scope !== proposal.scope + ) + ) continue; + let record = records?.get(proposal.cardId); + if (records && !record) throw new Error("linked scoped choice card is missing"); + effects.push({ + key: scopedKey(proposal.id, event.id), + kind: "scoped-choice", + threadId: thread.id, + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, + generation: record?.history.length ?? 0, + triggerEventId: event.id, + sources: [ + proposal.source, + ...(event.type === "scoped-choice.agreed" + ? [event.source] + : []), + ], + }); + continue; + } + if (event.type === "thread.opened") { + if (thread.questionnaireId || ["decided", "discarded"].includes(thread.status)) continue; + effects.push({ + key: `insert:${thread.id}`, + kind: "insert-card", + threadId: thread.id, + header: thread.question.trim().slice(0, Question.limits.MAX_HEADER), + question: thread.question, + options: thread.contributions.filter(item => item.kind === "option" && ULID.test(item.id)) + .slice(0, Question.limits.MAX_DECISION_OPTIONS) + .map(item => { + let source = item.sources.find(source => source.role === "option"); + return { + id: item.id, + label: item.text.trim().slice(0, Question.limits.MAX_LABEL), + ...(source ? { source } : {}), + }; + }), + trigger: event.id, + }); + } else if ( + event.type === "option.added" + && (event.origin === "classifier" || !!event.source) + ) { + if (opened.has(thread.id) || !ULID.test(event.contribution.id)) continue; + let card = thread.questionnaireId ? records?.get(thread.questionnaireId) : undefined; + if (card?.definition?.questions[0]?.options.some(item => item.id === event.contribution.id)) { + continue; + } + effects.push({ + key: `option:${event.contribution.id}`, + kind: "add-option", + threadId: thread.id, + optionId: event.contribution.id, + label: event.contribution.text.trim().slice(0, Question.limits.MAX_LABEL), + trigger: event.id, + ...(event.source?.role === "option" ? { source: event.source } : {}), + }); + if (thread.questionnaireId && event.origin !== "human") { + if (!event.source) continue; + effects.push({ + key: `job:suggest:${thread.questionnaireId}:${event.id}`, + kind: "job", + threadId: thread.id, + intent: { + kind: "suggest", + target: thread.questionnaireId, + trigger: event.source.messageId, + }, + }); + } + } else if (event.type === "card.linked") { + let source = thread.questionSources[0]; + if (!source) continue; + effects.push({ + key: `job:refine:${event.questionnaireId}`, + kind: "job", + threadId: thread.id, + intent: { kind: "refine", target: event.questionnaireId, trigger: source.messageId }, + }); + } else if (event.type === "thread.leaning") { + if (!event.optionId || !ULID.test(event.optionId)) continue; + effects.push({ + key: `suggest:${event.id}`, + kind: "suggest", + threadId: thread.id, + optionId: event.optionId, + messageIds: "source" in event && event.source ? [event.source.messageId] : [], + }); + } + } + let reconciliations = new Map>(); + for (let event of events) { + if ( + event.type === "settle.suggested" || event.type === "settle.agreed" + || event.type === "settle.resumed" + ) { + reconciliations.set(event.threadId, event); + } else if (event.type === "stance.changed" && event.optionId) { + let thread = state.threads.find(item => item.id === event.threadId); + let actor = event.source.author; + let eventIndex = state.events.findIndex(item => item.id === event.id); + if (thread?.pendingSettle?.optionId !== event.optionId || actor.kind !== "member") continue; + if ( + state.events.slice(0, eventIndex < 0 ? undefined : eventIndex).some(item => + item.threadId === event.threadId + && (item.type === "settle.suggested" || item.type === "settle.agreed") + && item.optionId === event.optionId + && item.source.author.kind === "member" + && item.source.author.handle === actor.handle + ) + ) reconciliations.set(event.threadId, event); + } + } + for (let [threadId, trigger] of reconciliations) { + let thread = state.threads.find(item => item.id === threadId); + if ( + !thread?.pendingSettle || !ULID.test(thread.pendingSettle.optionId) + || ["decided", "discarded"].includes(thread.status) + ) continue; + let preference = effectivePreference(thread, state.events); + let optionId = preference?.optionId; + let messageIds = preference?.messageIds ?? []; + let record = thread.questionnaireId && records?.get(thread.questionnaireId); + if (thread.questionnaireId && !record) throw new Error("linked prompt card is missing"); + effects.push({ + key: `suggest:${trigger.id}`, + kind: "suggest", + threadId, + ...(optionId ? { optionId } : {}), + messageIds, + }); + effects.push({ + key: `prompt:${trigger.id}`, + kind: "prompt", + threadId, + ...(optionId ? { optionId } : {}), + messageId: trigger.source.messageId, + generation: record ? record.history.length : 0, + sourceMessageIds: messageIds, + }); + } + if (purposeful(analysis)) { + effects.push({ + key: "job:heading:document", + kind: "job", + intent: { kind: "heading", target: "document", trigger: analysis!.messageId }, + }); + } + return effects; +} diff --git a/apps/server/src/conversation-plan/effect-runner.ts b/apps/server/src/conversation-plan/effect-runner.ts new file mode 100644 index 000000000..fea56b503 --- /dev/null +++ b/apps/server/src/conversation-plan/effect-runner.ts @@ -0,0 +1,110 @@ +import type { Effect, EffectDeps } from "./effect-types"; +// Extracted from archive 446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.ts. + +export async function runEffects(deps: EffectDeps, effects: readonly Effect[]): Promise { + let completed = 0; + let blocked = new Set(); + for ( + let effect of [ + ...effects.filter(item => item.kind !== "job"), + ...effects.filter(item => item.kind === "job"), + ] + ) { + // A fresh, explicit writer action supplies the owner context for research. + if (effect.kind === "research") continue; + if (deps.applied(effect.key)) continue; + if ( + effect.kind !== "insert-card" && "threadId" in effect && effect.threadId + && blocked.has(effect.threadId) + ) continue; + try { + if (effect.kind === "insert-card") { + let target = deps.target(effect.threadId); + if (target.kind !== "unlinked") { + await deps.markApplied(effect.key); + completed++; + continue; + } + let card = await deps.insertCard({ + threadId: effect.threadId, + header: effect.header, + question: effect.question, + options: effect.options, + }); + await deps.link(effect.threadId, card); + } else if (effect.kind === "job") { + if (effect.intent.kind === "prose" && !effect.threadId) { + throw new Error("prose job effect has no linked thread"); + } + if (effect.threadId) { + let target = deps.target(effect.threadId); + if (target.kind === "unlinked") { + blocked.add(effect.threadId); + continue; + } + if ( + effect.intent.kind === "prose" + && !deps.proseReady?.( + effect.threadId, + effect.intent.target, + effect.intent.trigger, + ) + ) { + await deps.markApplied(effect.key); + completed++; + continue; + } + if ( + target.kind === "closed" + && effect.intent.kind !== "prose" + ) { + await deps.markApplied(effect.key); + completed++; + continue; + } + } + if (!deps.enqueueJob) continue; + await deps.enqueueJob(effect.intent, effect.key); + } else if (effect.kind === "scoped-choice") { + if (!deps.scopedChoice) continue; + await deps.scopedChoice(effect); + } else if (effect.kind === "defer-prompt") { + if (!deps.deferPrompt) continue; + await deps.deferPrompt(effect); + } else { + let target = deps.target(effect.threadId); + if (target.kind === "unlinked") { + blocked.add(effect.threadId); + continue; + } + if (target.kind === "open") { + if (effect.kind === "add-option") { + await deps.addOption(target.id, { + optionId: effect.optionId, + label: effect.label, + trigger: effect.trigger, + ...(effect.source ? { source: effect.source } : {}), + }); + } else if (effect.kind === "prompt") { + await deps.prompt(target.id, effect.generation, { + optionId: effect.optionId, + messageId: effect.messageId, + sourceMessageIds: effect.sourceMessageIds, + }); + } else { + await deps.suggest(target.id, { + optionId: effect.optionId, + messageIds: effect.messageIds, + }); + } + } + } + await deps.markApplied(effect.key); + completed++; + } catch (error) { + if ("threadId" in effect && effect.threadId) blocked.add(effect.threadId); + deps.report(error); + } + } + return completed; +} diff --git a/apps/server/src/conversation-plan/effect-types.ts b/apps/server/src/conversation-plan/effect-types.ts new file mode 100644 index 000000000..780c17536 --- /dev/null +++ b/apps/server/src/conversation-plan/effect-types.ts @@ -0,0 +1,104 @@ +import type { ConversationPlan } from "@chopin/protocol"; +// Extracted from archive 446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.ts. + +export type JobIntent = { + kind: "heading" | "refine" | "suggest" | "prose"; + target: string; + trigger: string; +}; + +export type Effect = + | { + key: string; + kind: "insert-card"; + threadId: string; + header: string; + question: string; + options: Array<{ id: string; label: string; source?: ConversationPlan.SourceRef }>; + trigger: string; + } + | { + key: string; + kind: "add-option"; + threadId: string; + optionId: string; + label: string; + trigger: string; + source?: ConversationPlan.SourceRef; + } + | { + key: string; + kind: "suggest"; + threadId: string; + optionId?: string; + messageIds: string[]; + } + | { + key: string; + kind: "prompt"; + threadId: string; + optionId?: string; + messageId: string; + generation: number; + /** Absent on legacy outbox entries; present for source-aware prompt deduplication. */ + sourceMessageIds?: string[]; + } + | { + key: string; + kind: "defer-prompt"; + threadId: string; + proposalId: string; + deferredEventId: string; + } + | { + key: string; + kind: "scoped-choice"; + threadId: string; + proposalId: string; + cardId: string; + optionId: string; + label: string; + scope: "spike"; + generation: number; + triggerEventId: string; + sources: ConversationPlan.SourceRef[]; + } + | { key: string; kind: "research"; offerId: string } + | { key: string; kind: "job"; intent: JobIntent; threadId?: string }; + +export type CardTarget = { kind: "unlinked" | "closed" } | { kind: "open"; id: string }; + +export type EffectDeps = { + applied(key: string): boolean; + markApplied(key: string): Promise; + target(threadId: string): CardTarget; + /** A decided prose job may pass a closed card only for its saved generation. */ + proseReady?(threadId: string, questionnaireId: string, trigger: string): boolean; + insertCard(input: { + threadId: string; + header: string; + question: string; + options: Array<{ id: string; label: string; source?: ConversationPlan.SourceRef }>; + }): Promise; + link(threadId: string, questionnaireId: string): Promise; + addOption(questionnaireId: string, input: { + optionId: string; + label: string; + trigger?: string; + source?: ConversationPlan.SourceRef; + }): Promise; + suggest( + questionnaireId: string, + input: { optionId?: string; messageIds: string[] }, + ): Promise; + prompt( + questionnaireId: string, + generation: number, + source?: { optionId?: string; messageId: string; sourceMessageIds?: string[] }, + ): Promise; + deferPrompt?(effect: Extract): Promise; + scopedChoice?(effect: Extract): Promise; + /** Omitted until the narrow Planner job queue arrives in Slice 5. */ + enqueueJob?(intent: JobIntent, effectKey: string): Promise; + report(error: unknown): void; +}; diff --git a/apps/server/src/conversation-plan/effects-interpretation.test.ts b/apps/server/src/conversation-plan/effects-interpretation.test.ts new file mode 100644 index 000000000..90c05753d --- /dev/null +++ b/apps/server/src/conversation-plan/effects-interpretation.test.ts @@ -0,0 +1,90 @@ +import { expect, test } from "bun:test"; +import { applyInference, initialState } from "./domain"; +import { effectsFor } from "./effects"; +import { interpretMessage } from "./interpret"; +import { message, mockResult, seeded } from "./interpret.test-fixtures"; +import { planEvents } from "./policy"; +import { d01RecordedOpening } from "./policy-terminal.test-fixtures"; + +// Whole pipeline callbacks3532/3554/4479 from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2. + +test("unlinked messages retain purpose evidence for later effects", async () => { + let calls = 0; + let output = await interpretMessage({ + channelId: "channel", + message: message("purpose", "We're making a shared planning document."), + recent: [], + state: seeded(), + ask: async (request) => { + calls++; + return mockResult(request.questions, { enough_purpose: 0.95 }); + }, + }); + expect(calls).toBe(1); + expect(output.events).toEqual([]); + expect(output.analysis.status).toBe("unlinked"); + expect(output.analysis.questionSetVersion).toBe("conversation-plan-8"); + expect(output.analysis.passes[0].answers.enough_purpose).toEqual({ + type: "noul", + noul: 0.95, + }); +}); + +test("an opening purpose queues a heading but no card even if Jev calls it a question", async () => { + let current = message("figma-purpose", "Okay we gotta figure out what we're doing for auth"); + let state = initialState(); + let calls = 0; + let output = await interpretMessage({ + channelId: "channel", + message: current, + recent: [], + state, + ask: async request => + mockResult( + request.questions, + calls++ === 0 + ? { enough_purpose: 0.98, new_question: 0.08, act: "question", significance: 2 } + : { c0_role: "question", c0_thread: "new" }, + ), + }); + expect(calls).toBe(2); + expect(output.analysis.passes[0]?.answers.act).toMatchObject({ choice: "question" }); + expect(output.events).toEqual([]); + let effects = effectsFor(output.events, state, { + ...output.analysis, + messageId: current.id, + eventIds: [], + }); + expect(effects).toMatchObject([{ + kind: "job", + intent: { kind: "heading", target: "document", trigger: current.id }, + }]); +}); + +test("D01's owned purpose and final question open one editor-context card", () => { + let input = d01RecordedOpening(); + expect(input.candidates.map(candidate => [candidate.quote, candidate.start, candidate.end])) + .toEqual([ + ["need to pick an editor before comments get any deeper.", 0, 54], + ["what are we comparing?", 55, 77], + ]); + let output = planEvents(input); + expect(output.events).toHaveLength(1); + expect(output.events[0]).toMatchObject({ + type: "thread.opened", + question: input.message.text, + source: { + messageId: input.message.id, + author: input.message.author, + quote: input.message.text, + start: 0, + end: input.message.text.length, + role: "question", + }, + }); + let accepted = applyInference(input.state, output.events[0]!, input.message); + expect(effectsFor(output.events, accepted)).toContainEqual(expect.objectContaining({ + kind: "insert-card", + header: input.message.text, + })); +}); diff --git a/apps/server/src/conversation-plan/effects-outbox.test.ts b/apps/server/src/conversation-plan/effects-outbox.test.ts new file mode 100644 index 000000000..4fe675a57 --- /dev/null +++ b/apps/server/src/conversation-plan/effects-outbox.test.ts @@ -0,0 +1,87 @@ +import { expect, test } from "bun:test"; +import { type Effect, restoreEffectOutbox } from "./effects"; +import { LATER, OPTION } from "./effects.test-fixtures"; + +// Whole callbacks from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts. +test("restored pending effects and receipts are strictly bounded", () => { + let pending: Effect = { + key: `option:${LATER}`, + kind: "add-option", + threadId: "t1", + optionId: LATER, + label: "Auth0", + trigger: "add-1", + }; + expect(restoreEffectOutbox([pending], ["insert:t1"])).toEqual({ + pending: [pending], + receipts: ["insert:t1"], + }); + expect(restoreEffectOutbox(undefined, undefined)).toEqual({ pending: [], receipts: [] }); + expect(() => restoreEffectOutbox([{ ...pending, surprise: true }], [])) + .toThrow("conversation effect"); + expect(() => restoreEffectOutbox([{ ...pending, optionId: "old-id" }], [])) + .toThrow("conversation effect"); + expect(() => restoreEffectOutbox([{ ...pending, optionId: [LATER] }], [])) + .toThrow("conversation effect"); + expect(() => + restoreEffectOutbox([{ + key: "insert:t1", + kind: "insert-card", + threadId: "t1", + header: "Auth", + question: "Which auth system?", + options: [{ id: [LATER], label: "Auth0" }], + trigger: "open-1", + }], []) + ).toThrow("conversation effect"); + expect(() => + restoreEffectOutbox([{ + key: "suggest:settle-1", + kind: "suggest", + threadId: "t1", + optionId: [LATER], + messageIds: ["m1"], + }], []) + ).toThrow("conversation effect"); + expect(() => restoreEffectOutbox([pending, pending], [])) + .toThrow("conversation effect"); + expect(() => restoreEffectOutbox([pending], [pending.key])) + .toThrow("conversation effect"); + expect(() => restoreEffectOutbox([pending], Array.from({ length: 1025 }, (_, i) => `k${i}`))) + .toThrow("conversation effect"); +}); + +test("restored prompt effects require a bounded captured generation", () => { + let prompt: Effect = { + key: "prompt:agree-1", + kind: "prompt", + threadId: "t1", + optionId: OPTION, + messageId: "m3", + generation: 0, + }; + let refreshed = { ...prompt, sourceMessageIds: ["m6"] }; + expect(restoreEffectOutbox([refreshed], [])).toEqual({ pending: [refreshed], receipts: [] }); + let cleared = { ...prompt, optionId: undefined, sourceMessageIds: [] }; + delete cleared.optionId; + expect(restoreEffectOutbox([cleared], [])).toEqual({ pending: [cleared], receipts: [] }); + expect(() => restoreEffectOutbox([{ ...prompt, sourceMessageIds: ["m6", "m6"] }], [])) + .toThrow("conversation effect"); + expect(restoreEffectOutbox([prompt], []).pending).toEqual([prompt]); + for (let generation of [undefined, -1, 0.5, 2 ** 53]) { + expect(() => restoreEffectOutbox([{ ...prompt, generation }], [])) + .toThrow("conversation effect"); + } + expect(() => restoreEffectOutbox([{ ...prompt, optionId: "old" }], [])) + .toThrow("conversation effect"); +}); + +test("restored prose effects require their linked thread", () => { + expect(() => + restoreEffectOutbox([{ + key: "job:prose:Q:1", + kind: "job", + intent: { kind: "prose", target: "Q", trigger: "decided:Q:1" }, + }], []) + ).toThrow("conversation effect"); +}); diff --git a/apps/server/src/conversation-plan/effects-projection.test.ts b/apps/server/src/conversation-plan/effects-projection.test.ts new file mode 100644 index 000000000..21ff8222c --- /dev/null +++ b/apps/server/src/conversation-plan/effects-projection.test.ts @@ -0,0 +1,180 @@ +import { expect, test } from "bun:test"; +import type { ConversationPlan } from "@chopin/protocol"; +import * as Question from "@chopin/question"; +import { initialState } from "./domain"; +import { effectsFor, recoverable } from "./effects"; +import { event, LATER, OPTION, thread } from "./effects.test-fixtures"; + +// Whole callbacks from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts. +test("an opened thread inserts only card-valid quoted options; legacy recovery stays empty", () => { + let open = event("thread.opened", "open-1", { question: "Which auth system?" }); + let state = { ...initialState(), threads: [thread()] }; + expect(effectsFor([open], state)).toMatchObject([{ + key: "insert:t1", + kind: "insert-card", + options: [{ id: OPTION, label: "GitHub Apps" }], + }]); + let old = thread({ + contributions: [{ + ...thread().contributions[0], + id: "classifier:historical", + }], + }); + let legacy = { ...initialState(), events: [open], threads: [old] }; + expect(recoverable(legacy)).toEqual([open]); + expect(effectsFor(recoverable(legacy), legacy)[0]).toMatchObject({ options: [] }); +}); + +test("card labels use the same bound on initial and later quoted options", () => { + let quote = "A".repeat(Question.limits.MAX_LABEL + 30); + let initial = thread({ contributions: [{ ...thread().contributions[0], text: quote }] }); + let state = { ...initialState(), threads: [initial] }; + let opened = event("thread.opened", "open-long"); + let added = event("option.added", "add-long", { + contribution: { id: LATER, text: quote, authoring: "quoted" }, + }); + expect(effectsFor([opened], state)[0]).toMatchObject({ + options: [{ id: OPTION, label: "A".repeat(Question.limits.MAX_LABEL) }], + }); + expect(effectsFor([added], state)[0]).toMatchObject({ + label: "A".repeat(Question.limits.MAX_LABEL), + }); + let spaced = ` ${" ".repeat(Question.limits.MAX_LABEL)}GitHub Apps `; + let spacedThread = thread({ + question: `${" ".repeat(Question.limits.MAX_HEADER)}Which auth system?`, + contributions: [{ ...thread().contributions[0], text: spaced }], + }); + let spacedState = { ...initialState(), threads: [spacedThread] }; + expect(effectsFor([opened], spacedState)[0]).toMatchObject({ + header: "Which auth system?", + options: [{ id: OPTION, label: "GitHub Apps" }], + }); + let spacedAdded = event("option.added", "add-spaced", { + contribution: { id: LATER, text: spaced, authoring: "quoted" }, + }); + expect(effectsFor([spacedAdded], spacedState)[0]).toMatchObject({ label: "GitHub Apps" }); +}); + +test("a pre-link option waits; link makes a separate refine job", () => { + let source: ConversationPlan.SourceRef = { + messageId: "option-message", + author: { kind: "member", handle: "jules" }, + quote: "Auth0", + start: 0, + end: 5, + role: "option", + }; + let option = event("option.added", "option-2", { + source, + contribution: { id: LATER, text: "Auth0", authoring: "quoted" }, + }); + let unlinked = { ...initialState(), threads: [thread()] }; + expect(effectsFor([option], unlinked).map(item => item.kind)).toEqual(["add-option"]); + let linked = { + ...initialState(), + threads: [thread({ + questionnaireId: "Q", + questionSources: [{ ...source, role: "question", messageId: "question-message" }], + })], + }; + expect(effectsFor([option], linked).map(item => item.kind)).toEqual([ + "add-option", + "job", + ]); + expect(effectsFor([event("card.linked", "link-1", { questionnaireId: "Q" })], linked)) + .toMatchObject([{ key: "job:refine:Q", kind: "job", intent: { kind: "refine" } }]); +}); + +test("sourced Planner chat options reach linked cards; source-free card options do not", () => { + let linked = { ...initialState(), threads: [thread({ questionnaireId: "Q" })] }; + let quoted = event("option.added", "planner-quoted", { + origin: "planner", + actor: { kind: "agent" }, + source: { + messageId: "agent-message", + author: { kind: "agent" }, + quote: "Auth0", + start: 0, + end: 5, + role: "option", + }, + contribution: { id: LATER, text: "Auth0", authoring: "quoted" }, + }); + let cardTool = event("option.added", "planner-card", { + origin: "planner", + actor: { kind: "agent" }, + contribution: { id: LATER, text: "Auth0", authoring: "scribe" }, + }); + expect(effectsFor([quoted], linked).map(item => item.kind)).toEqual(["add-option", "job"]); + expect(effectsFor([cardTool], linked)).toEqual([]); +}); + +test("new card jobs retain their event receipt but trigger from the source chat message", () => { + let source = { + messageId: "original-question", + author: { kind: "member" as const, handle: "jules" }, + quote: "Which auth system?", + start: 0, + end: 18, + role: "question" as const, + }; + let linked = { + ...initialState(), + threads: [thread({ questionnaireId: "Q", questionSources: [source] })], + }; + expect(effectsFor([event("card.linked", "link-event", { questionnaireId: "Q" })], linked)) + .toMatchObject([{ + key: "job:refine:Q", + intent: { kind: "refine", trigger: "original-question" }, + }]); + let quoted = event("option.added", "option-event", { + source: { ...source, messageId: "option-message", role: "option" }, + contribution: { id: LATER, text: "Auth0", authoring: "quoted" }, + }); + expect(effectsFor([quoted], linked)[1]).toMatchObject({ + key: "job:suggest:Q:option-event", + intent: { kind: "suggest", trigger: "option-message" }, + }); +}); + +test("unlinked purpose creates one heading intent without a thread event", () => { + let analysis = { + messageId: "m1", + questionSetVersion: "conversation-plan-8", + modelVersion: "jev", + status: "unlinked", + eventIds: [], + passes: [{ stage: "triage", answers: { enough_purpose: { type: "noul", noul: 0.75 } } }], + } as ConversationPlan.AnalysisRecord; + expect(effectsFor([], initialState(), analysis)).toMatchObject([{ + key: "job:heading:document", + kind: "job", + intent: { kind: "heading", target: "document", trigger: "m1" }, + }]); + expect(effectsFor([], initialState(), { + ...analysis, + questionSetVersion: "conversation-plan-4", + })).toMatchObject([{ key: "job:heading:document" }]); + expect( + effectsFor([], initialState(), { ...analysis, questionSetVersion: "conversation-plan-3" }), + ) + .toEqual([]); + expect(effectsFor([], initialState(), { + ...analysis, + passes: [{ stage: "triage", answers: { enough_purpose: { type: "noul", noul: 0.749 } } }], + })).toEqual([]); +}); + +test("new settle and leaning events do not target historical non-ULID options", () => { + let oldId = "classifier:historical-option"; + let state = { + ...initialState(), + threads: [thread({ + contributions: [{ ...thread().contributions[0], id: oldId }], + })], + }; + expect(effectsFor([event("settle.suggested", "settle-1", { optionId: oldId })], state)) + .toEqual([]); + expect(effectsFor([event("thread.leaning", "lean-1", { optionId: oldId })], state)) + .toEqual([]); +}); diff --git a/apps/server/src/conversation-plan/effects-retry.test.ts b/apps/server/src/conversation-plan/effects-retry.test.ts new file mode 100644 index 000000000..0918b03e3 --- /dev/null +++ b/apps/server/src/conversation-plan/effects-retry.test.ts @@ -0,0 +1,99 @@ +import { expect, test } from "bun:test"; +import { type Effect, runEffects } from "./effects"; +import { LATER, OPTION, sink } from "./effects.test-fixtures"; + +// Whole callbacks from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts. +test("a failed link retries the same card, then delivers options before refine", async () => { + let linked = false; + let failLink = true; + let failOption = true; + let state = sink({ + target: () => linked ? { kind: "open", id: "Q" } : { kind: "unlinked" }, + insertCard: async () => (state.calls.push("insert:Q"), "Q"), + link: async () => { + if (failLink) { + failLink = false; + throw new Error("link commit failed"); + } + linked = true; + state.calls.push("link:Q"); + }, + addOption: async () => { + if (failOption) { + failOption = false; + throw new Error("option commit failed"); + } + state.calls.push("option"); + }, + enqueueJob: async () => void state.calls.push("refine"), + }); + let effects: Effect[] = [ + { + key: "insert:t1", + kind: "insert-card", + threadId: "t1", + header: "Auth", + question: "Which auth system?", + options: [], + trigger: "open-1", + }, + { + key: "option:later", + kind: "add-option", + threadId: "t1", + optionId: LATER, + label: "Auth0", + trigger: "add-1", + }, + { + key: "job:refine:Q", + kind: "job", + threadId: "t1", + intent: { kind: "refine", target: "Q", trigger: "link-1" }, + }, + ]; + expect(await runEffects(state.deps, effects)).toBe(0); + expect(state.calls).toEqual(["insert:Q"]); + expect(await runEffects(state.deps, effects)).toBe(1); + expect(state.calls).toEqual(["insert:Q", "insert:Q", "link:Q"]); + expect(state.calls).not.toContain("refine"); + expect(await runEffects(state.deps, effects)).toBe(2); + expect(state.calls).toEqual(["insert:Q", "insert:Q", "link:Q", "option", "refine"]); +}); + +test("all quoted options project before a job, forwarding its raw intent and receipt key", async () => { + let state = sink({ + addOption: async (_id, input) => void state.calls.push(input.label), + enqueueJob: async (intent, key) => void state.calls.push(`${key}:${intent.trigger}`), + }); + let effects: Effect[] = [ + { + key: "option:first", + kind: "add-option", + threadId: "t1", + optionId: OPTION, + label: "First", + trigger: "event-first", + }, + { + key: "job:suggest:Q:event-first", + kind: "job", + threadId: "t1", + intent: { kind: "suggest", target: "Q", trigger: "event-first" }, + }, + { + key: "option:second", + kind: "add-option", + threadId: "t1", + optionId: LATER, + label: "Second", + trigger: "event-second", + }, + ]; + expect(await runEffects(state.deps, effects)).toBe(3); + expect(state.calls).toEqual([ + "First", + "Second", + "job:suggest:Q:event-first:event-first", + ]); +}); diff --git a/apps/server/src/conversation-plan/effects-runner.test.ts b/apps/server/src/conversation-plan/effects-runner.test.ts new file mode 100644 index 000000000..0d1b1b275 --- /dev/null +++ b/apps/server/src/conversation-plan/effects-runner.test.ts @@ -0,0 +1,215 @@ +import { expect, test } from "bun:test"; +import { initialState } from "./domain"; +import { type Effect, effectsFor, runEffects } from "./effects"; +import { event, LATER, OPTION, sink, thread } from "./effects.test-fixtures"; + +// Whole callbacks from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts. +test("mirroring an existing linked card option skips add and job but keeps the prompt", () => { + let source = { + messageId: "m5", + author: { kind: "member" as const, handle: "jules" }, + quote: "Let's just go with Auth0.", + start: 0, + end: 24, + role: "option" as const, + }; + let linked = { + ...initialState(), + threads: [thread({ + questionnaireId: "Q", + pendingSettle: { optionId: LATER, proposer: "jules", messageId: "m5" }, + contributions: [{ + id: LATER, + kind: "option", + text: "Auth0", + authoring: "scribe", + sources: [source], + actor: { kind: "classifier" }, + }], + })], + }; + let added = event("option.added", "card-mirror", { + source, + contribution: { id: LATER, text: "Auth0", authoring: "scribe" }, + }); + let settle = event("settle.suggested", "card-settle", { + source: { ...source, role: "resolution" }, + optionId: LATER, + }); + let records = new Map([["Q", { + history: [], + definition: { questions: [{ options: [{ id: LATER }] }] }, + }]]); + let effects = effectsFor([added, settle], linked, undefined, records); + expect(effects.some(item => item.kind === "add-option" || item.kind === "job")).toBe(false); + expect(effects.some(item => item.kind === "prompt")).toBe(true); +}); + +test("unlinked work defers, closed work is receipted, and absent job sink never spins", async () => { + let state = sink({ target: () => ({ kind: "unlinked" }) }); + let option: Effect = { + key: "option:2", + kind: "add-option", + threadId: "t1", + optionId: LATER, + label: "Auth0", + trigger: "option-2", + }; + let job: Effect = { + key: "job:heading:document", + kind: "job", + intent: { kind: "heading", target: "document", trigger: "m1" }, + }; + expect(await runEffects(state.deps, [option, job])).toBe(0); + expect(state.calls).toEqual([]); + expect([...state.receipts]).toEqual([]); + state.deps.target = () => ({ kind: "closed" }); + expect(await runEffects(state.deps, [option, job])).toBe(1); + expect([...state.receipts]).toEqual(["option:2"]); + state.deps.enqueueJob = async () => void state.calls.push("job"); + expect(await runEffects(state.deps, [option, job])).toBe(1); + expect(state.calls).toEqual(["job"]); +}); + +test("a closed card accepts only its current saved prose generation", async () => { + let calls: string[] = []; + let state = sink({ + target: () => ({ kind: "closed" }), + proseReady: (thread, id, trigger) => thread === "t1" && id === "Q" && trigger === "decided:Q:2", + enqueueJob: async intent => void calls.push(intent.trigger), + }); + let stale: Effect = { + key: "job:prose:Q:1", + kind: "job", + threadId: "t1", + intent: { kind: "prose", target: "Q", trigger: "decided:Q:1" }, + }; + let current: Effect = { + key: "job:prose:Q:2", + kind: "job", + threadId: "t1", + intent: { kind: "prose", target: "Q", trigger: "decided:Q:2" }, + }; + let mismatched: Effect = { + ...current, + key: "job:prose:Q:2:wrong-thread", + threadId: "t2", + }; + expect(await runEffects(state.deps, [stale, mismatched, current])).toBe(3); + expect(calls).toEqual(["decided:Q:2"]); + expect(state.receipts).toEqual(new Set([stale.key, mismatched.key, current.key])); + let reopened = sink({ + target: () => ({ kind: "open", id: "Q" }), + proseReady: () => false, + enqueueJob: async intent => void calls.push(intent.trigger), + }); + expect(await runEffects(reopened.deps, [current])).toBe(1); + expect(reopened.receipts).toEqual(new Set([current.key])); + expect(calls).toEqual(["decided:Q:2"]); + let malformedErrors: unknown[] = []; + let malformed = sink({ + enqueueJob: async intent => void calls.push(intent.trigger), + report: error => void malformedErrors.push(error), + }); + expect(await runEffects(malformed.deps, [{ ...current, threadId: undefined }])).toBe(0); + expect(malformed.calls).toEqual([]); + expect(malformedErrors).toHaveLength(1); +}); + +test("a pending suggestion waits for a link and targets the linked card exactly once", async () => { + let linked = false; + let delivered: Array<{ card: string; optionId?: string; messageIds: string[] }> = []; + let state = sink({ + target: () => linked ? { kind: "open", id: "Q" } : { kind: "unlinked" }, + suggest: async (card, input) => void delivered.push({ card, ...input }), + }); + let suggestion: Effect = { + key: "suggest:settle-1", + kind: "suggest", + threadId: "t1", + optionId: OPTION, + messageIds: ["m1"], + }; + expect(await runEffects(state.deps, [suggestion])).toBe(0); + expect(delivered).toEqual([]); + linked = true; + expect(await runEffects(state.deps, [suggestion])).toBe(1); + expect(delivered).toEqual([{ card: "Q", optionId: OPTION, messageIds: ["m1"] }]); + expect(await runEffects(state.deps, [suggestion])).toBe(0); + expect(delivered).toHaveLength(1); +}); + +test("failed insert retains its key; retry links and receipts it", async () => { + let attempts = 0; + let state = sink({ + target: () => ({ kind: "unlinked" }), + insertCard: async () => { + if (attempts++ === 0) throw new Error("storage unavailable"); + state.calls.push("insert"); + return "Q"; + }, + }); + let insert: Effect = { + key: "insert:t1", + kind: "insert-card", + threadId: "t1", + header: "Auth", + question: "Which auth system?", + options: [], + trigger: "open-1", + }; + expect(await runEffects(state.deps, [insert])).toBe(0); + expect([...state.receipts]).toEqual([]); + expect(await runEffects(state.deps, [insert])).toBe(1); + expect(state.calls).toEqual(["insert", "link"]); + expect([...state.receipts]).toEqual(["insert:t1"]); +}); + +test("closed inserts are terminal without creating a card", async () => { + let state = sink({ target: () => ({ kind: "closed" }) }); + let insert: Effect = { + key: "insert:t1", + kind: "insert-card", + threadId: "t1", + header: "Auth", + question: "Which auth system?", + options: [], + trigger: "open-1", + }; + expect(await runEffects(state.deps, [insert])).toBe(1); + expect(state.calls).toEqual([]); + expect([...state.receipts]).toEqual(["insert:t1"]); +}); + +test("a lost insert receipt recognizes the already-linked card on retry", async () => { + let linked = false; + let failReceipt = true; + let state = sink({ + target: () => linked ? { kind: "open", id: "Q" } : { kind: "unlinked" }, + link: async () => { + linked = true; + state.calls.push("link"); + }, + markApplied: async key => { + if (failReceipt) { + failReceipt = false; + throw new Error("receipt commit failed"); + } + state.receipts.add(key); + }, + }); + let insert: Effect = { + key: "insert:t1", + kind: "insert-card", + threadId: "t1", + header: "Auth", + question: "Which auth system?", + options: [], + trigger: "open-1", + }; + expect(await runEffects(state.deps, [insert])).toBe(0); + expect(state.calls).toEqual(["insert", "link"]); + expect(await runEffects(state.deps, [insert])).toBe(1); + expect(state.calls).toEqual(["insert", "link"]); + expect([...state.receipts]).toEqual(["insert:t1"]); +}); diff --git a/apps/server/src/conversation-plan/effects-scoped.test.ts b/apps/server/src/conversation-plan/effects-scoped.test.ts new file mode 100644 index 000000000..b1d77b678 --- /dev/null +++ b/apps/server/src/conversation-plan/effects-scoped.test.ts @@ -0,0 +1,75 @@ +import { expect, test } from "bun:test"; +import type { ConversationPlan } from "@chopin/protocol"; +import { effectsFor } from "./effects"; +import { + m7MessageText, + scopedAgreement, + scopedProposal, + scopedState, +} from "./effects.test-fixtures"; + +// Whole callbacks from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts. +test("a scoped choice proposal creates a keyed inline Save notice without a final prompt", () => { + let proposal = scopedProposal(); + let effects = effectsFor([proposal], scopedState(proposal)); + expect(effects).toHaveLength(1); + expect(effects[0]).toMatchObject({ + key: `scoped-choice:${proposal.id}:proposal`, + kind: "scoped-choice", + threadId: proposal.threadId, + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, + sources: [proposal.source], + }); + expect(effects.some(item => item.kind === "prompt")).toBe(false); +}); + +test("a scoped agreement refreshes the same proposal notice with both exact member sources", () => { + let proposal = scopedProposal(); + let agreement = scopedAgreement(proposal); + expect(agreement.source.quote).toBe( + m7MessageText.slice(agreement.source.start, agreement.source.end), + ); + let effects = effectsFor([agreement], scopedState(proposal, agreement)); + expect(effects).toHaveLength(1); + expect(effects[0]).toMatchObject({ + key: `scoped-choice:${proposal.id}:agreement:${agreement.id}`, + kind: "scoped-choice", + threadId: proposal.threadId, + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, + sources: [proposal.source, agreement.source], + }); + expect(effects.some(item => item.kind === "prompt")).toBe(false); +}); + +test("saving a scoped choice does not create another active Save notice", () => { + let proposal = scopedProposal(); + let saved: ConversationPlan.Event = { + id: "saved-m6", + type: "scoped-choice.saved", + threadId: proposal.threadId, + observedThreadVersion: 2, + origin: "human", + actor: { kind: "member", handle: "ana" }, + at: 8, + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, + sources: [proposal.source], + expectedGeneration: 0, + }; + let state = scopedState(proposal, saved); + state.threads[0]!.pendingScopedChoice = undefined; + let effects = effectsFor([saved], state); + expect(effects.map(item => String(item.kind))).not.toContain("scoped-choice"); + expect(effects.some(item => item.kind === "prompt")).toBe(false); +}); diff --git a/apps/server/src/conversation-plan/effects-settlement.test.ts b/apps/server/src/conversation-plan/effects-settlement.test.ts new file mode 100644 index 000000000..e475f16f8 --- /dev/null +++ b/apps/server/src/conversation-plan/effects-settlement.test.ts @@ -0,0 +1,137 @@ +import { expect, test } from "bun:test"; +import { initialState } from "./domain"; +import { type Effect, effectsFor, runEffects } from "./effects"; +import { event, OPTION, sink, thread } from "./effects.test-fixtures"; + +// Whole callbacks from archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts. +test("one sourced settle suggestion selects the card and queues a Save prompt", () => { + let suggestion = event("settle.suggested", "settle-1", { + optionId: OPTION, + source: { + messageId: "m2", + author: { kind: "member", handle: "mina" }, + quote: "Lets do GitHub Apps", + start: 0, + end: 18, + role: "resolution", + }, + }); + let linked = { + ...initialState(), + events: [suggestion], + threads: [thread({ + questionnaireId: "Q", + pendingSettle: { optionId: OPTION, proposer: "mina", messageId: "m2" }, + })], + }; + expect(effectsFor( + [suggestion], + linked, + undefined, + new Map([["Q", { history: [] }]]), + )).toEqual([{ + key: "suggest:settle-1", + kind: "suggest", + threadId: "t1", + optionId: OPTION, + messageIds: ["m2"], + }, { + key: "prompt:settle-1", + kind: "prompt", + threadId: "t1", + optionId: OPTION, + messageId: "m2", + generation: 0, + sourceMessageIds: ["m2"], + }]); +}); + +test("a sourced agreement queues a prompt bound to the card's current generation", () => { + let agreement = event("settle.agreed", "agree-1", { + optionId: OPTION, + source: { + messageId: "m3", + author: { kind: "member", handle: "jules" }, + quote: "Sounds good.", + start: 0, + end: 12, + role: "support", + }, + }); + let proposal = event("settle.suggested", "settle-1", { + optionId: OPTION, + source: { + messageId: "m2", + author: { kind: "member", handle: "mina" }, + quote: "Lets do GitHub Apps", + start: 0, + end: 18, + role: "resolution", + }, + }); + let linked = { + ...initialState(), + events: [proposal, agreement], + threads: [thread({ + questionnaireId: "Q", + pendingSettle: { optionId: OPTION, proposer: "mina", messageId: "m2" }, + })], + }; + expect(effectsFor( + [agreement], + linked, + undefined, + new Map([ + ["Q", { history: [{ at: 1 }, { at: 1 }] }], + ]), + )).toEqual([{ + key: "suggest:agree-1", + kind: "suggest", + threadId: "t1", + optionId: OPTION, + messageIds: ["m2", "m3"], + }, { + key: "prompt:agree-1", + kind: "prompt", + threadId: "t1", + optionId: OPTION, + messageId: "m3", + generation: 2, + sourceMessageIds: ["m2", "m3"], + }]); + let unlinked = { + ...linked, + threads: [thread({ pendingSettle: linked.threads[0].pendingSettle })], + }; + expect(effectsFor([agreement], unlinked, undefined, new Map())).toMatchObject([{}, { + kind: "prompt", + generation: 0, + }]); + expect(() => effectsFor([agreement], linked, undefined, new Map())) + .toThrow("linked prompt card is missing"); +}); + +test("a prompt defers until linked, receipts a closed card, and passes its captured generation", async () => { + let linked = false; + let delivered: Array<{ card: string; generation: number }> = []; + let state = sink({ + target: () => linked ? { kind: "open", id: "Q" } : { kind: "unlinked" }, + prompt: async (card, generation) => void delivered.push({ card, generation }), + }); + let prompt: Effect = { + key: "prompt:agree-1", + kind: "prompt", + threadId: "t1", + optionId: OPTION, + messageId: "m3", + generation: 0, + }; + expect(await runEffects(state.deps, [prompt])).toBe(0); + linked = true; + expect(await runEffects(state.deps, [prompt])).toBe(1); + expect(delivered).toEqual([{ card: "Q", generation: 0 }]); + expect([...state.receipts]).toEqual(["prompt:agree-1"]); + let closed = sink({ target: () => ({ kind: "closed" }) }); + expect(await runEffects(closed.deps, [prompt])).toBe(1); + expect(closed.calls).toEqual([]); +}); diff --git a/apps/server/src/conversation-plan/effects.test-fixtures.ts b/apps/server/src/conversation-plan/effects.test-fixtures.ts new file mode 100644 index 000000000..12005981a --- /dev/null +++ b/apps/server/src/conversation-plan/effects.test-fixtures.ts @@ -0,0 +1,136 @@ +import type { ConversationPlan } from "@chopin/protocol"; +import { initialState } from "./domain"; +import type { EffectDeps } from "./effects"; + +// Exact archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.test.ts helpers/data. + +export let OPTION = "01K0N4W3B7P27CBAEC7A8C8WEA"; + +export let LATER = "01K0N4W3B7P27CBAEC7A8C8WEB"; + +export let m7MessageText = "yep, Lexical for the spike. not a final library call yet."; + +export function thread(overrides: Partial = {}): ConversationPlan.Thread { + return { + id: "t1", + question: "Which auth system?", + questionSources: [], + questionAuthoring: "quoted", + status: "exploring", + contributions: [{ + id: OPTION, + kind: "option", + text: "GitHub Apps", + authoring: "quoted", + sources: [], + actor: { kind: "classifier" }, + }], + stances: [], + stanceHistory: [], + decisionHistory: [], + candidates: [], + version: 1, + ...overrides, + }; +} + +export function event(type: string, id: string, extra: Record = {}) { + return { + id, + type, + threadId: "t1", + observedThreadVersion: 1, + origin: "classifier", + actor: { kind: "classifier" }, + at: 1, + ...extra, + } as ConversationPlan.Event; +} + +export function sink(overrides: Partial = {}) { + let receipts = new Set(); + let calls: string[] = []; + let deps: EffectDeps = { + applied: key => receipts.has(key), + markApplied: async key => void receipts.add(key), + target: () => ({ kind: "open", id: "Q" }), + insertCard: async () => (calls.push("insert"), "Q"), + link: async () => void calls.push("link"), + addOption: async () => void calls.push("option"), + suggest: async () => void calls.push("suggest"), + prompt: async () => void calls.push("prompt"), + report: () => {}, + ...overrides, + }; + return { deps, receipts, calls }; +} + +export type ScopedProposal = Extract; + +export type ScopedAgreement = Extract; + +export let scopedProposal = (): ScopedProposal => ({ + id: "proposal-m6", + type: "scoped-choice.proposed", + threadId: "t1", + observedThreadVersion: 1, + origin: "classifier", + actor: { kind: "classifier" }, + at: 6, + source: { + messageId: "m6", + author: { kind: "member", handle: "mei" }, + quote: "I'd pick Lexical for the spike;", + start: 0, + end: "I'd pick Lexical for the spike;".length, + role: "support", + }, + cardId: "Q", + optionId: OPTION, + label: "Lexical", + scope: "spike", +}); + +export let scopedAgreement = (proposal: ScopedProposal): ScopedAgreement => ({ + id: "agreement-m7", + type: "scoped-choice.agreed", + threadId: proposal.threadId, + observedThreadVersion: 2, + origin: "classifier", + actor: { kind: "classifier" }, + at: 7, + source: { + messageId: "m7", + author: { kind: "member", handle: "rob" }, + quote: m7MessageText.slice(0, 27), + start: 0, + end: 27, + role: "support", + }, + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, +}); + +export let scopedState = ( + proposal: ScopedProposal, + ...later: ConversationPlan.Event[] +): ConversationPlan.State => ({ + ...initialState(), + events: [proposal, ...later], + threads: [thread({ + questionnaireId: "Q", + contributions: [{ ...thread().contributions[0]!, text: "Lexical" }], + pendingScopedChoice: { + proposalId: proposal.id, + cardId: proposal.cardId, + optionId: proposal.optionId, + label: proposal.label, + scope: proposal.scope, + proposer: "mei", + messageId: proposal.source.messageId, + }, + })], +}); diff --git a/apps/server/src/conversation-plan/effects.ts b/apps/server/src/conversation-plan/effects.ts new file mode 100644 index 000000000..6f6f1cfd7 --- /dev/null +++ b/apps/server/src/conversation-plan/effects.ts @@ -0,0 +1,16 @@ +import type { ConversationPlan } from "@chopin/protocol"; +export { MAX_EFFECTS } from "./effect-fields"; +export { restoreEffectOutbox } from "./effect-outbox"; +export { effectsFor } from "./effect-projection"; +export { runEffects } from "./effect-runner"; +export type { CardTarget, Effect, EffectDeps, JobIntent } from "./effect-types"; +// Extracted from archive 446a9779a937fa5be7cd3eb52fd7f3023d691ed2, effects.ts. + +export function recoverable(state: ConversationPlan.State): ConversationPlan.Event[] { + let open = new Set( + state.threads.filter(thread => + !thread.questionnaireId && ["exploring", "leaning", "reopened"].includes(thread.status) + ).map(thread => thread.id), + ); + return state.events.filter(event => event.type === "thread.opened" && open.has(event.threadId)); +} diff --git a/apps/server/src/conversation-plan/jobs-restore.test.ts b/apps/server/src/conversation-plan/jobs-restore.test.ts new file mode 100644 index 000000000..a089560c0 --- /dev/null +++ b/apps/server/src/conversation-plan/jobs-restore.test.ts @@ -0,0 +1,104 @@ +import { describe, expect, test } from "bun:test"; + +import * as Jobs from "./jobs"; +import { AT, card } from "./jobs.test-fixtures"; + +describe("Planner job queue", () => { + test("restore converts an interrupted running job to a failed, non-replayed job", () => { + let running = Jobs.start(Jobs.enqueue([], card("m1"), AT), "refine:W1:m1", AT); + let restored = Jobs.restore(JSON.parse(JSON.stringify(running))); + expect(restored[0]).toEqual({ + id: "refine:W1:m1", + kind: "refine", + target: "W1", + trigger: "m1", + status: "failed", + attempts: 0, + reason: "interrupted", + at: AT, + }); + expect(Jobs.next(restored)).toBeUndefined(); + expect(Jobs.restore(undefined)).toEqual([]); + }); + + test("restore rejects malformed structure, identity, bounds, and duplicate work", () => { + let pending = Jobs.enqueue([], card("m1"), AT)[0]!; + let cases: Array<[string, unknown]> = [ + ["null", null], + ["not an array", {}], + [ + "over capacity", + Array.from({ length: Jobs.MAX_JOBS + 1 }, (_, index) => ({ + ...pending, + id: `refine:W${index}:m1`, + target: `W${index}`, + })), + ], + ["missing field", [{ id: pending.id }]], + ["unknown field", [{ ...pending, extra: true }]], + ["wrong kind", [{ ...pending, kind: "other" }]], + ["wrong id", [{ ...pending, id: "refine:W1:other" }]], + ["blank target", [{ ...pending, target: " " }]], + ["long target", [{ ...pending, target: "x".repeat(201) }]], + ["blank trigger", [{ ...pending, trigger: " " }]], + ["long trigger", [{ ...pending, trigger: "x".repeat(201) }]], + ["bad timestamp", [{ ...pending, at: "tomorrow" }]], + ["noncanonical timestamp", [{ ...pending, at: "2026-09-25T10:00:00Z" }]], + ["negative attempts", [{ ...pending, attempts: -1 }]], + ["fractional attempts", [{ ...pending, attempts: 1.5 }]], + ["duplicate ids", [pending, { ...pending }]], + ]; + for (let [name, value] of cases) { + expect(() => Jobs.restore(value), name).toThrow("invalid Planner job"); + } + }); + + test("restore rejects impossible status details and concurrent runners", () => { + let pending = Jobs.enqueue([], card("m1"), AT)[0]!; + let running = Jobs.start([pending], pending.id, AT)[0]!; + let done = Jobs.settle([running], running.id, { status: "done", output: "{}" }, AT)[0]!; + let failed = Jobs.settle([running], running.id, { + status: "failed", + reason: "Planner unavailable", + }, AT)[0]!; + let skipped = Jobs.settle([running], running.id, { + status: "skipped", + reason: "AGENT=off", + }, AT)[0]!; + let cases: Array<[string, unknown]> = [ + ["pending reason", [{ ...pending, reason: "old" }]], + ["running output", [{ ...running, output: "{}" }]], + ["done missing output", [{ ...done, output: undefined }]], + ["done malformed JSON", [{ ...done, output: "not JSON" }]], + ["done huge output", [{ ...done, output: JSON.stringify("x".repeat(4096)) }]], + ["done reason", [{ ...done, reason: "old" }]], + ["done without attempt", [{ ...done, attempts: 0 }]], + ["failed missing reason", [{ ...failed, reason: undefined }]], + ["failed huge reason", [{ ...failed, reason: "x".repeat(501) }]], + ["failed output", [{ ...failed, output: "{}" }]], + ["failed without attempt", [{ ...failed, attempts: 0 }]], + ["skipped without attempt", [{ ...skipped, attempts: 0 }]], + ["two running", [running, { ...running, id: "refine:W2:m2", target: "W2", trigger: "m2" }]], + ["two pending card jobs", [pending, { + ...pending, + id: "suggest:W1:m2", + kind: "suggest", + trigger: "m2", + }]], + ]; + for (let [name, value] of cases) { + expect(() => Jobs.restore(value), name).toThrow("invalid Planner job"); + } + expect( + Jobs.restore([ + done, + { ...failed, id: "refine:W2:m2", target: "W2", trigger: "m2" }, + { ...skipped, id: "refine:W3:m3", target: "W3", trigger: "m3" }, + ]).map(job => job.status), + ).toEqual([ + "done", + "failed", + "skipped", + ]); + }); +}); diff --git a/apps/server/src/conversation-plan/jobs-validation.test.ts b/apps/server/src/conversation-plan/jobs-validation.test.ts new file mode 100644 index 000000000..6b1ec6c67 --- /dev/null +++ b/apps/server/src/conversation-plan/jobs-validation.test.ts @@ -0,0 +1,91 @@ +import { describe, expect, test } from "bun:test"; + +import * as Jobs from "./jobs"; +import { AT, card } from "./jobs.test-fixtures"; + +describe("Planner job queue", () => { + test("queue transitions reject data that could not be restored", () => { + let jobs = Jobs.enqueue([], card("m1"), AT); + expect(() => + Jobs.enqueue(jobs, { + kind: "heading", + target: "wrong", + trigger: "m2", + }, AT) + ).toThrow("invalid Planner job"); + expect(() => + Jobs.enqueue(jobs, { + kind: "refine", + target: "x".repeat(201), + trigger: "m2", + }, AT) + ).toThrow("invalid Planner job"); + expect(() => + Jobs.enqueue(jobs, { + kind: "refine", + target: "W2", + trigger: "x".repeat(201), + }, AT) + ).toThrow("invalid Planner job"); + expect(() => Jobs.enqueue(jobs, card("m2"), "yesterday")).toThrow("invalid Planner job"); + expect(() => Jobs.start(jobs, "refine:W1:m1", "yesterday")) + .toThrow("invalid Planner job"); + let running = Jobs.start(jobs, "refine:W1:m1", AT); + expect(() => + Jobs.settle(running, "refine:W1:m1", { + status: "done", + output: "not JSON", + }, AT) + ).toThrow("invalid Planner job"); + expect(() => + Jobs.settle(running, "refine:W1:m1", { + status: "done", + output: JSON.stringify("x".repeat(4096)), + }, AT) + ).toThrow("invalid Planner job"); + expect(() => + Jobs.settle(running, "refine:W1:m1", { + status: "failed", + reason: "x".repeat(501), + }, AT) + ).toThrow("invalid Planner job"); + let failed = Jobs.settle(running, "refine:W1:m1", { + status: "failed", + reason: "temporarily unavailable", + }, AT); + expect(() => Jobs.retry(failed, "refine:W1:m1", "yesterday")) + .toThrow("invalid Planner job"); + expect(jobs[0]?.status).toBe("pending"); + expect(running[0]?.status).toBe("running"); + }); + + test("retry waits while newer card work for the same target is pending", () => { + let failed = Jobs.settle( + Jobs.start(Jobs.enqueue([], card("m1"), AT), "refine:W1:m1", AT), + "refine:W1:m1", + { status: "failed", reason: "Planner unavailable" }, + AT, + ); + let jobs = Jobs.enqueue(failed, card("m2", "suggest"), AT); + expect(Jobs.retry(jobs, "refine:W1:m1", AT)).toBe(jobs); + let after = Jobs.settle(Jobs.start(jobs, "suggest:W1:m2", AT), "suggest:W1:m2", { + status: "done", + output: "{}", + }, AT); + expect(Jobs.retry(after, "refine:W1:m1", AT)[0]?.status).toBe("pending"); + }); + + test("card targets cannot make two intents share a colon-delimited job ID", () => { + let valid = { kind: "refine" as const, target: "W1", trigger: "m2:m3" }; + let ambiguous = { kind: "refine" as const, target: "W1:m2", trigger: "m3" }; + expect(Jobs.jobId(valid)).toBe("refine:W1:m2:m3"); + expect(() => Jobs.jobId(ambiguous)).toThrow("invalid Planner job"); + expect(() => Jobs.enqueue([], ambiguous, AT)).toThrow("invalid Planner job"); + let raw = { + ...Jobs.enqueue([], valid, AT)[0]!, + target: ambiguous.target, + trigger: ambiguous.trigger, + }; + expect(() => Jobs.restore([raw])).toThrow("invalid Planner job"); + }); +}); diff --git a/apps/server/src/conversation-plan/jobs.test-fixtures.ts b/apps/server/src/conversation-plan/jobs.test-fixtures.ts new file mode 100644 index 000000000..d033f7f86 --- /dev/null +++ b/apps/server/src/conversation-plan/jobs.test-fixtures.ts @@ -0,0 +1,5 @@ +export const AT = "2026-09-25T10:00:00.000Z"; + +export function card(trigger: string, kind: "refine" | "suggest" = "refine") { + return { kind, target: "W1", trigger }; +} diff --git a/apps/server/src/conversation-plan/jobs.test.ts b/apps/server/src/conversation-plan/jobs.test.ts new file mode 100644 index 000000000..ef2b6dfe5 --- /dev/null +++ b/apps/server/src/conversation-plan/jobs.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, test } from "bun:test"; + +import * as Jobs from "./jobs"; +import { AT, card } from "./jobs.test-fixtures"; + +describe("Planner job queue", () => { + test("a trigger mints one stable pending job", () => { + let once = Jobs.enqueue([], card("m1"), AT); + expect(once).toEqual([{ + id: "refine:W1:m1", + kind: "refine", + target: "W1", + trigger: "m1", + status: "pending", + attempts: 0, + at: AT, + }]); + expect(Jobs.enqueue(once, card("m1"), AT)).toBe(once); + }); + + test("pending card work coalesces, with refine taking the pending place", () => { + let jobs = Jobs.enqueue([], { kind: "heading", target: "document", trigger: "m0" }, AT); + jobs = Jobs.enqueue(jobs, card("m1", "suggest"), AT); + expect(Jobs.enqueue(jobs, card("m2", "suggest"), AT)).toBe(jobs); + jobs = Jobs.enqueue(jobs, card("m3", "refine"), AT); + expect(jobs.map(job => job.id)).toEqual([ + "heading:document:m0", + "refine:W1:m3", + ]); + expect(Jobs.enqueue(jobs, card("m4", "suggest"), AT)).toBe(jobs); + expect(Jobs.enqueue(jobs, card("m5", "refine"), AT)).toBe(jobs); + }); + + test("only the first pending job starts, and running work keeps later work queued", () => { + let jobs = Jobs.enqueue([], card("m1"), AT); + jobs = Jobs.enqueue(jobs, { kind: "suggest", target: "W2", trigger: "m2" }, AT); + expect(Jobs.next(jobs)?.id).toBe("refine:W1:m1"); + expect(Jobs.start(jobs, "suggest:W2:m2", AT)).toBe(jobs); + jobs = Jobs.start(jobs, "refine:W1:m1", AT); + expect(Jobs.next(jobs)).toBeUndefined(); + expect(Jobs.start(jobs, "suggest:W2:m2", AT)).toBe(jobs); + let queued = Jobs.enqueue(jobs, card("m3", "suggest"), AT); + expect(queued.map(job => [job.id, job.status])).toEqual([ + ["refine:W1:m1", "running"], + ["suggest:W2:m2", "pending"], + ["suggest:W1:m3", "pending"], + ]); + let done = Jobs.settle(queued, "refine:W1:m1", { status: "done", output: "{}" }, AT); + expect(done[0]).toMatchObject({ status: "done", attempts: 1, output: "{}" }); + expect(Jobs.next(done)?.id).toBe("suggest:W2:m2"); + }); + + test("full unfinished queue rejects new work but evicts the oldest done job", () => { + let jobs: Jobs.Job[] = []; + for (let index = 0; index < Jobs.MAX_JOBS; index++) { + jobs = Jobs.enqueue(jobs, { + kind: "suggest", + target: `W${index}`, + trigger: "m", + }, AT); + } + let same = Jobs.enqueue(jobs, { kind: "suggest", target: "W1", trigger: "new" }, AT); + expect(same).toBe(jobs); + let late = { kind: "refine" as const, target: "W129", trigger: "late" }; + expect(() => Jobs.enqueue(jobs, late, AT)).toThrow("Planner job queue is full"); + expect(jobs).toHaveLength(Jobs.MAX_JOBS); + jobs = Jobs.settle(Jobs.start(jobs, "suggest:W0:m", AT), "suggest:W0:m", { + status: "done", + output: "{}", + }, AT); + let grown = Jobs.enqueue(jobs, late, AT); + expect(grown).toHaveLength(Jobs.MAX_JOBS); + expect(grown.some(job => job.id === "suggest:W0:m")).toBe(false); + expect(grown.at(-1)?.id).toBe("refine:W129:late"); + }); + + test("128 skipped jobs leave room for new Planner work", () => { + let jobs: Jobs.Job[] = []; + for (let index = 0; index < Jobs.MAX_JOBS; index++) { + let intent = card(`m${index}`); + jobs = Jobs.enqueue(jobs, intent, AT); + jobs = Jobs.start(jobs, Jobs.jobId(intent), AT); + jobs = Jobs.settle(jobs, Jobs.jobId(intent), { + status: "skipped", + reason: "The Planner is off (AGENT=off).", + }, AT); + } + let next = Jobs.enqueue(jobs, card("m128"), AT); + expect(next).toHaveLength(Jobs.MAX_JOBS); + expect(next.some(job => job.id === "refine:W1:m0")).toBe(false); + expect(next.at(-1)).toMatchObject({ id: "refine:W1:m128", status: "pending" }); + expect(Jobs.enqueue(next, card("m128"), AT)).toBe(next); + }); + + test("evicting skipped history preserves retryable failed work", () => { + let failedIntent = card("failed"); + let jobs = Jobs.enqueue([], failedIntent, AT); + jobs = Jobs.start(jobs, Jobs.jobId(failedIntent), AT); + jobs = Jobs.settle(jobs, Jobs.jobId(failedIntent), { + status: "failed", + reason: "The Planner could not be reached.", + }, AT); + for (let index = 0; index < Jobs.MAX_JOBS - 1; index++) { + let intent = card(`skipped-${index}`); + jobs = Jobs.enqueue(jobs, intent, AT); + jobs = Jobs.start(jobs, Jobs.jobId(intent), AT); + jobs = Jobs.settle(jobs, Jobs.jobId(intent), { + status: "skipped", + reason: "The Planner is off (AGENT=off).", + }, AT); + } + let next = Jobs.enqueue(jobs, { kind: "suggest", target: "W2", trigger: "new" }, AT); + expect(next.some(job => job.id === Jobs.jobId(failedIntent))).toBe(true); + expect(next.some(job => job.id === "refine:W1:skipped-0")).toBe(false); + expect(Jobs.retry(next, Jobs.jobId(failedIntent), AT)[0]?.status).toBe("pending"); + }); + + test("only failed work can be retried, without forgetting its attempt", () => { + let jobs = Jobs.enqueue([], card("m1"), AT); + expect(Jobs.retry(jobs, "refine:W1:m1", AT)).toBe(jobs); + jobs = Jobs.start(jobs, "refine:W1:m1", AT); + expect(Jobs.retry(jobs, "refine:W1:m1", AT)).toBe(jobs); + jobs = Jobs.settle(jobs, "refine:W1:m1", { + status: "failed", + reason: "Planner unavailable", + }, AT); + expect(jobs[0]).toMatchObject({ status: "failed", reason: "Planner unavailable", attempts: 1 }); + expect(Jobs.next(jobs)).toBeUndefined(); + let retried = Jobs.retry(jobs, "refine:W1:m1", AT); + expect(retried[0]).toMatchObject({ status: "pending", attempts: 1 }); + expect(retried[0]?.reason).toBeUndefined(); + expect(Jobs.retry(retried, "refine:W1:m1", AT)).toBe(retried); + let skipped = Jobs.settle(Jobs.start(retried, "refine:W1:m1", AT), "refine:W1:m1", { + status: "skipped", + reason: "AGENT=off", + }, AT); + expect(skipped[0]).toMatchObject({ status: "skipped", attempts: 2 }); + expect(Jobs.retry(skipped, "refine:W1:m1", AT)).toBe(skipped); + }); + + test("refining and message diagnostics show only their matching work", () => { + let jobs = Jobs.enqueue([], card("m1", "suggest"), AT); + expect(Jobs.refining(jobs, "W1")).toBe(false); + jobs = Jobs.enqueue(jobs, { kind: "refine", target: "W2", trigger: "m2" }, AT); + expect(Jobs.refining(jobs, "W2")).toBe(true); + expect(Jobs.forTrigger(jobs, "m2").map(job => job.id)).toEqual(["refine:W2:m2"]); + expect(Jobs.forTrigger(jobs, "m3")).toEqual([]); + jobs = Jobs.settle(Jobs.start(jobs, "suggest:W1:m1", AT), "suggest:W1:m1", { + status: "done", + output: "{}", + }, AT); + jobs = Jobs.settle(Jobs.start(jobs, "refine:W2:m2", AT), "refine:W2:m2", { + status: "done", + output: "{}", + }, AT); + expect(Jobs.refining(jobs, "W2")).toBe(false); + expect(Jobs.forTrigger(jobs, "m2")).toHaveLength(1); + }); +}); diff --git a/apps/server/src/conversation-plan/jobs.ts b/apps/server/src/conversation-plan/jobs.ts new file mode 100644 index 000000000..dbbc68877 --- /dev/null +++ b/apps/server/src/conversation-plan/jobs.ts @@ -0,0 +1,189 @@ +/** Pure transitions for the durable Planner job queue. */ + +import type { ConversationPlan } from "@chopin/protocol"; +import type { JobIntent } from "./effects"; + +export type Job = ConversationPlan.Job; +export type JobOutcome = + | { status: "done"; output: string } + | { status: "failed" | "skipped"; reason: string }; + +export const MAX_JOBS = 128; +const MAX_TARGET = 200; +const MAX_TRIGGER = 200; +const MAX_REASON = 500; +const MAX_OUTPUT = 4096; +const KINDS = new Set(["heading", "refine", "suggest", "prose"]); +const STATUSES = new Set(["pending", "running", "done", "failed", "skipped"]); + +function invalid(): never { + throw new Error("invalid Planner job in sidecar"); +} + +function bounded(value: unknown, max: number): value is string { + return typeof value === "string" && !!value.trim() && value.length <= max; +} + +function isoTime(value: unknown): value is string { + return typeof value === "string" + && /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/.test(value) + && Number.isFinite(Date.parse(value)) + && new Date(value).toISOString() === value; +} + +function jsonOutput(value: unknown): value is string { + if (!bounded(value, MAX_OUTPUT)) return false; + try { + JSON.parse(value); + return true; + } catch { + return false; + } +} + +export function jobId(intent: JobIntent): string { + if (intent.target.includes(":")) invalid(); + return `${intent.kind}:${intent.target}:${intent.trigger}`; +} + +export function enqueue(jobs: Job[], intent: JobIntent, at: string): Job[] { + if ( + !KINDS.has(intent.kind) || !bounded(intent.target, MAX_TARGET) + || !bounded(intent.trigger, MAX_TRIGGER) + || (intent.kind === "heading") !== (intent.target === "document") + || !isoTime(at) + ) invalid(); + let id = jobId(intent); + if (jobs.some(job => job.id === id)) return jobs; + if (intent.kind === "refine" || intent.kind === "suggest") { + let pending = jobs.findIndex(job => + (job.kind === "refine" || job.kind === "suggest") + && job.target === intent.target && job.status === "pending" + ); + if (pending >= 0) { + if (intent.kind === "suggest" || jobs[pending]!.kind === "refine") return jobs; + return jobs.map((job, index) => + index === pending ? { ...job, id, kind: "refine", trigger: intent.trigger, at } : job + ); + } + } + let available = jobs; + if (jobs.length >= MAX_JOBS) { + let oldestTerminal = jobs.findIndex(job => job.status === "done" || job.status === "skipped"); + if (oldestTerminal < 0) throw new Error("Planner job queue is full"); + available = jobs.filter((_, index) => index !== oldestTerminal); + } + return [...available, { id, ...intent, status: "pending", attempts: 0, at }]; +} + +export function next(jobs: Job[]): Job | undefined { + if (jobs.some(job => job.status === "running")) return; + return jobs.find(job => job.status === "pending"); +} + +export function start(jobs: Job[], id: string, at: string): Job[] { + if (next(jobs)?.id !== id) return jobs; + if (!isoTime(at)) invalid(); + return jobs.map(job => job.id === id ? { ...job, status: "running", at } : job); +} + +export function settle(jobs: Job[], id: string, outcome: JobOutcome, at: string): Job[] { + let running = jobs.find(job => job.id === id && job.status === "running"); + if (!running) return jobs; + if ( + !isoTime(at) || running.attempts >= Number.MAX_SAFE_INTEGER + || (outcome.status === "done" + ? !jsonOutput(outcome.output) + : !["failed", "skipped"].includes(outcome.status) + || !bounded(outcome.reason, MAX_REASON)) + ) invalid(); + return jobs.map(job => { + if (job.id !== id) return job; + let { reason: _reason, output: _output, ...rest } = job; + return outcome.status === "done" + ? { ...rest, status: "done", attempts: job.attempts + 1, output: outcome.output, at } + : { ...rest, status: outcome.status, attempts: job.attempts + 1, reason: outcome.reason, at }; + }); +} + +export function retry(jobs: Job[], id: string, at: string): Job[] { + let failed = jobs.find(job => job.id === id && job.status === "failed"); + if (!failed) return jobs; + if ( + (failed.kind === "refine" || failed.kind === "suggest") + && jobs.some(job => + job.status === "pending" && (job.kind === "refine" || job.kind === "suggest") + && job.target === failed.target + ) + ) return jobs; + if (!isoTime(at)) invalid(); + return jobs.map(job => { + if (job.id !== id) return job; + let { reason: _reason, ...rest } = job; + return { ...rest, status: "pending", at }; + }); +} + +export function refining(jobs: Job[], questionnaireId: string): boolean { + return jobs.some(job => + job.kind === "refine" && job.target === questionnaireId + && (job.status === "pending" || job.status === "running") + ); +} + +export function forTrigger(jobs: Job[], messageId: string): Job[] { + return jobs.filter(job => job.trigger === messageId); +} + +export function restore(value: unknown): Job[] { + if (value === undefined) return []; + if (!Array.isArray(value) || value.length > MAX_JOBS) invalid(); + let ids = new Set(); + let pendingCards = new Set(); + let running = 0; + for (let raw of value) { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) invalid(); + let job = raw as Record; + let required = ["id", "kind", "target", "trigger", "status", "attempts", "at"]; + if ( + required.some(key => !Object.hasOwn(job, key)) + || Object.keys(job).some(key => ![...required, "reason", "output"].includes(key)) + || !KINDS.has(job.kind as string) + || !STATUSES.has(job.status as string) + || !bounded(job.target, MAX_TARGET) + || (job.target as string).includes(":") + || !bounded(job.trigger, MAX_TRIGGER) + || !bounded(job.id, 420) + || job.id !== `${job.kind}:${job.target}:${job.trigger}` + || ids.has(job.id) + || !Number.isSafeInteger(job.attempts) || (job.attempts as number) < 0 + || !isoTime(job.at) + || (job.kind === "heading") !== (job.target === "document") + ) invalid(); + ids.add(job.id as string); + let hasReason = Object.hasOwn(job, "reason"); + let hasOutput = Object.hasOwn(job, "output"); + if (job.status === "pending" || job.status === "running") { + if (hasReason || hasOutput) invalid(); + if (job.status === "running" && ++running > 1) invalid(); + if ( + job.status === "pending" && (job.kind === "refine" || job.kind === "suggest") + ) { + if (pendingCards.has(job.target as string)) invalid(); + pendingCards.add(job.target as string); + } + } else if (job.status === "done") { + if (hasReason || !hasOutput || !jsonOutput(job.output) || job.attempts === 0) { + invalid(); + } + } else { + if ( + hasOutput || !hasReason || !bounded(job.reason, MAX_REASON) + || job.attempts === 0 && (job.status !== "failed" || job.reason !== "interrupted") + ) invalid(); + } + } + return value.map((job: Job) => + job.status === "running" ? { ...job, status: "failed", reason: "interrupted" } : job + ); +} diff --git a/apps/server/src/conversation-plan/pipeline-recovery-effects.test.ts b/apps/server/src/conversation-plan/pipeline-recovery-effects.test.ts new file mode 100644 index 000000000..9461e34eb --- /dev/null +++ b/apps/server/src/conversation-plan/pipeline-recovery-effects.test.ts @@ -0,0 +1,135 @@ +import { expect, test } from "bun:test"; +import { applyInference, initialState } from "./domain"; +import { effectsFor } from "./effects"; +import { interpretMessage } from "./interpret"; +import { planEvents } from "./policy"; +import { extractQuotes } from "./quotes"; +import { mockResult } from "./interpret.test-fixtures"; +import { confidentChoice, first, follow, message } from "./policy-initial.test-fixtures"; +import { d01LinkedEditorCard } from "./pipeline-linked-option.test-fixtures"; + +test("a distinct new option is still added without claiming an existing card option", () => { + let { state, threadId, options, linkedCards } = d01LinkedEditorCard(); + let quote = "I'd pick Slate for the spike;"; + let current = message("d01-m6-new", quote, "Mei"); + let result = planEvents({ + channelId: "channel", + message: current, + state, + linkedCards, + first: { + ...first({ new_option: 0.95, c0_owned_unretracted: 0.95 }), + act: confidentChoice("proposal"), + thread_target: confidentChoice(threadId), + }, + candidates: [{ + quote, + start: 0, + end: quote.length, + answers: { + ...follow({ role: "option", thread: threadId }), + option: confidentChoice("new"), + new_option: { type: "noul", noul: 0.95 }, + planning_substance: { type: "noul", noul: 0.95 }, + duplicate: { type: "noul", noul: 0.05 }, + }, + }], + }); + let added = result.events.filter(event => event.type === "option.added"); + expect(added).toHaveLength(1); + expect(added[0]).toMatchObject({ + type: "option.added", + threadId, + source: { messageId: current.id, quote, role: "option" }, + contribution: { text: quote, targetId: threadId, authoring: "quoted" }, + }); + let addedOption = added[0]; + if (addedOption?.type !== "option.added") throw new Error("a distinct option was not added"); + expect(options.some(option => option.id === addedOption.contribution.id)).toBe(false); + expect(state.threads[0]!.contributions.filter(item => item.kind === "option")).toEqual([]); +}); + +test("an explicit multi-option question opens one thread with its exact quoted options", async () => { + let recent = [message("figma-purpose", "Okay we gotta figure out what we're doing for auth")]; + let current = message( + "figma-options", + "Should we use an off the shelf solution like Auth0? Or roll our own? Or use something like GitHub auth?", + "Jules", + ); + let state = initialState(); + let calls = 0; + let output = await interpretMessage({ + channelId: "channel", + message: current, + recent, + state, + ask: async request => + mockResult( + request.questions, + calls++ === 0 + ? { + new_question: 0.97, + act: "question", + thread_target: "new", + new_option: 0.96, + significance: 2, + } + : { + c0_role: "option", + c0_thread: "new", + c0_new_option: 0.96, + c0_duplicate: 0.84, + c1_role: "option", + c1_thread: "none", + c1_new_option: 0.93, + c2_role: "option", + c2_thread: "new", + c2_new_option: 0.96, + }, + ), + }); + expect(calls).toBe(4); + expect(output.events.map(event => event.type)).toEqual([ + "thread.opened", + "option.added", + "option.added", + "option.added", + ]); + let opened = output.events[0]!; + expect(opened).toMatchObject({ + question: current.text, + source: { quote: current.text, start: 0, end: current.text.length }, + }); + let quotes = extractQuotes(current.text); + expect(output.events.slice(1).map(event => "source" in event ? event.source?.quote : "")) + .toEqual(quotes.map(quote => quote.quote)); + let accepted = output.events.reduce( + (next, event) => applyInference(next, event, current), + state, + ); + expect(effectsFor(output.events, accepted)).toMatchObject([{ + kind: "insert-card", + options: quotes.map(quote => ({ label: quote.quote })), + }]); + let targeting = output.analysis.passes[1]!.answers; + let replay = planEvents({ + channelId: "channel", + message: current, + state, + first: output.analysis.passes[0]!.answers, + candidates: quotes.map((quote, index) => ({ + ...quote, + answers: Object.fromEntries( + Object.entries(targeting) + .filter(([key]) => key.startsWith(`c${index}_`)) + .map(([key, answer]) => [key.slice(3), answer]), + ), + })), + }); + expect(replay.events.map(event => event.id)).toEqual(output.events.map(event => event.id)); + let duplicate = replay.events.reduce( + (next, event) => applyInference(next, event, current), + accepted, + ); + expect(duplicate.events).toEqual(accepted.events); +}); diff --git a/apps/server/src/conversation-plan/pipeline-source-opening.test.ts b/apps/server/src/conversation-plan/pipeline-source-opening.test.ts new file mode 100644 index 000000000..b574a03bc --- /dev/null +++ b/apps/server/src/conversation-plan/pipeline-source-opening.test.ts @@ -0,0 +1,36 @@ +import { expect, test } from "bun:test"; +import { applyInference, initialState } from "./domain"; +import { effectsFor } from "./effects"; +import { interpretMessage } from "./interpret"; +import { mockResult } from "./interpret.test-fixtures"; +import { message } from "./policy-initial.test-fixtures"; + +test.each( + [ + ["interrogative choice", "Should we use Auth0 or build our own authentication?", 0.97], + ["declarative choice", "We need to decide between Auth0 and a custom login system.", 0.93], + ["specific unknown", "How should we handle account recovery?", 0.65], + ] as const, +)("an actual %s can still open a card", async (_, text, newQuestion) => { + let current = message(`actual-${newQuestion}`, text); + let state = initialState(); + let calls = 0; + let output = await interpretMessage({ + channelId: "channel", + message: current, + recent: [], + state, + ask: async request => + mockResult( + request.questions, + calls++ === 0 + ? { enough_purpose: 0.98, new_question: newQuestion, significance: 2 } + : { c0_role: "question", c0_thread: "new" }, + ), + }); + expect(output.events.map(event => event.type)).toEqual(["thread.opened"]); + let accepted = applyInference(state, output.events[0]!, current); + expect(effectsFor(output.events, accepted).map(effect => effect.kind)).toContain( + "insert-card", + ); +}); diff --git a/apps/server/src/questions/card-actions.test.ts b/apps/server/src/questions/card-actions.test.ts new file mode 100644 index 000000000..ee63c3050 --- /dev/null +++ b/apps/server/src/questions/card-actions.test.ts @@ -0,0 +1,109 @@ +import { expect, test } from "bun:test"; + +import { + captureCardAction, + decisionGeneration, + proseJobTrigger, + restorePendingCardActions, +} from "./card-actions"; + +import type { Record } from "./records"; + +const CARD = "01K0N4W3B7P27CBAEC7A8C8WEA"; +const OPTION = "01K0N4W3B7P27CBAEC7A8C8WEB"; + +function record(history = 0): Record { + return { + id: CARD, + origin: "conversation", + threadId: "thread-a", + status: "open", + definition: { questions: [] }, + history: Array.from({ length: history }, () => ({ choices: [], owner: "ana", at: 1 })), + optionOrigins: {}, + editors: [], + } as Record; +} + +test("card actions capture immutable generations, actors, times and choices", () => { + expect(decisionGeneration(record())).toBe(1); + expect(decisionGeneration(record(1))).toBe(2); + expect(proseJobTrigger(CARD, decisionGeneration(record(1)))).toBe(`decided:${CARD}:2`); + let first = captureCardAction( + record(), + { + kind: "decided", + id: CARD, + actor: "ana", + optionIds: [OPTION], + }, + 9, + "GitHub Apps", + ); + let reopen = captureCardAction(record(), { + kind: "reopened", + id: CARD, + actor: "ben", + }, 10); + let second = captureCardAction( + record(1), + { + kind: "decided", + id: CARD, + actor: "cy", + optionIds: [OPTION], + }, + 11, + "GitHub Apps", + ); + expect([first?.id, reopen?.id, second?.id]).toEqual([ + `card:${CARD}:decided:1`, + `card:${CARD}:reopened:1`, + `card:${CARD}:decided:2`, + ]); + expect([first?.actor, reopen?.actor, second?.actor]).toEqual(["ana", "ben", "cy"]); + expect([first?.at, reopen?.at, second?.at]).toEqual([9, 10, 11]); + expect(first).toMatchObject({ optionIds: [OPTION], text: "GitHub Apps" }); + expect(restorePendingCardActions([first!, reopen!, second!])).toEqual([ + first!, + reopen!, + second!, + ]); +}); + +test("chat-origin and unlinked cards have no mirror action", () => { + expect(captureCardAction({ ...record(), threadId: undefined }, { + kind: "discarded", + id: CARD, + actor: "ana", + }, 9)).toBeUndefined(); + expect(captureCardAction(record(), { + kind: "option-added", + id: CARD, + actor: "chopin", + origin: "chat", + optionId: OPTION, + label: "GitHub Apps", + }, 9)).toBeUndefined(); +}); + +test("pending action restore rejects duplicates, missing payload and unbounded actors", () => { + let action = captureCardAction( + record(), + { + kind: "decided", + id: CARD, + actor: "ana", + optionIds: [OPTION], + }, + 9, + "GitHub Apps", + )!; + expect(() => restorePendingCardActions([action, action])).toThrow(/card actions/); + expect(() => restorePendingCardActions([{ ...action, text: undefined }])).toThrow( + /card actions/, + ); + expect(() => restorePendingCardActions([{ ...action, actor: "x".repeat(201) }])).toThrow( + /card actions/, + ); +}); diff --git a/apps/server/src/questions/card-actions.ts b/apps/server/src/questions/card-actions.ts new file mode 100644 index 000000000..e06cc2bb6 --- /dev/null +++ b/apps/server/src/questions/card-actions.ts @@ -0,0 +1,159 @@ +import { ULID } from "@chopin/dialect"; +import * as Question from "@chopin/question"; + +import type { Record as CardRecord } from "./records"; +import type { CardEvent } from "./card-event"; + +const MAX_PENDING_CARD_ACTIONS = 1024; +const MAX_DECISION_TEXT = 500; + +type Base = { + id: string; + cardId: string; + threadId: string; + actor: string; + at: number; +}; + +/** Stable identity of the currently saved decision; reopen records the prior answer in history. */ +export function decisionGeneration(record: Pick): number { + return record.history.length + 1; +} + +export function proseJobTrigger(cardId: string, generation: number): string { + return `decided:${cardId}:${generation}`; +} + +export type PendingCardAction = + & Base + & ( + | { kind: "decided"; generation: number; optionIds: string[]; text: string } + | { kind: "reopened"; generation: number } + | { kind: "discarded" } + | { kind: "option-added"; optionId: string; label: string; origin: "human" | "planner" } + ); + +/** Capture the event before the card's fenced commit, using card history rather than observer state. */ +export function captureCardAction( + record: CardRecord, + event: CardEvent, + at: number, + text?: string, +): PendingCardAction | undefined { + if (!record.threadId) return; + if (event.id !== record.id) throw new Error("card action does not match its record"); + let base = { + cardId: record.id, + threadId: record.threadId, + actor: event.actor, + at, + }; + switch (event.kind) { + case "decided": { + if (!text) throw new Error("card decision text is missing"); + let generation = decisionGeneration(record); + let summary = text.length > MAX_DECISION_TEXT + ? `${text.slice(0, MAX_DECISION_TEXT - 1).replace(/[\uD800-\uDBFF]$/, "")}…` + : text; + return { + ...base, + id: `card:${record.id}:decided:${generation}`, + kind: "decided", + generation, + optionIds: [...event.optionIds], + text: summary, + }; + } + case "reopened": { + let generation = decisionGeneration(record); + return { + ...base, + id: `card:${record.id}:reopened:${generation}`, + kind: "reopened", + generation, + }; + } + case "discarded": + return { ...base, id: `card:${record.id}:discarded`, kind: "discarded" }; + case "option-added": + if (event.origin === "chat") return; + return { + ...base, + id: `card:${record.id}:option:${event.optionId}`, + kind: "option-added", + optionId: event.optionId, + label: event.label, + origin: event.origin, + }; + } +} + +export function appendCardAction( + pending: readonly PendingCardAction[], + action: PendingCardAction | undefined, +): PendingCardAction[] { + if (!action) return [...pending]; + if (pending.some(item => item.id === action.id)) throw new Error("duplicate pending card action"); + if (pending.length >= MAX_PENDING_CARD_ACTIONS) throw new Error("pending card actions are full"); + return [...pending, action]; +} + +function invalid(): never { + throw new Error("hosted channel has invalid card actions"); +} + +function bounded(value: unknown, max: number): value is string { + return typeof value === "string" && !!value.trim() && value.length <= max; +} + +function exact(value: Record, keys: string[]): boolean { + return Object.keys(value).sort().join(",") === keys.sort().join(","); +} + +/** Legacy sidecars may omit the queue; present entries must be complete and strictly bounded. */ +export function restorePendingCardActions(value: unknown): PendingCardAction[] { + if (value === undefined) return []; + if (!Array.isArray(value) || value.length > MAX_PENDING_CARD_ACTIONS) invalid(); + let ids = new Set(); + for (let raw of value) { + if (!raw || typeof raw !== "object" || Array.isArray(raw)) invalid(); + let item = raw as Record; + let base = ["id", "cardId", "threadId", "actor", "at", "kind"]; + if ( + !bounded(item.cardId, 200) || !ULID.test(item.cardId) + || !bounded(item.threadId, 200) || !bounded(item.actor, 200) + || !Number.isSafeInteger(item.at) || (item.at as number) < 0 + || !bounded(item.id, 200) || ids.has(item.id) + ) invalid(); + ids.add(item.id); + if (item.kind === "decided") { + if ( + !exact(item, [...base, "generation", "optionIds", "text"]) + || !Number.isSafeInteger(item.generation) || (item.generation as number) < 1 + || item.id !== `card:${item.cardId}:decided:${item.generation}` + || !Array.isArray(item.optionIds) + || item.optionIds.length > Question.limits.MAX_DECISION_OPTIONS + || item.optionIds.some(id => typeof id !== "string" || !ULID.test(id)) + || new Set(item.optionIds).size !== item.optionIds.length + || !bounded(item.text, MAX_DECISION_TEXT) + ) invalid(); + } else if (item.kind === "reopened") { + if ( + !exact(item, [...base, "generation"]) + || !Number.isSafeInteger(item.generation) || (item.generation as number) < 1 + || item.id !== `card:${item.cardId}:reopened:${item.generation}` + ) invalid(); + } else if (item.kind === "discarded") { + if (!exact(item, base) || item.id !== `card:${item.cardId}:discarded`) invalid(); + } else if (item.kind === "option-added") { + if ( + !exact(item, [...base, "optionId", "label", "origin"]) + || typeof item.optionId !== "string" || !ULID.test(item.optionId) + || item.id !== `card:${item.cardId}:option:${item.optionId}` + || !bounded(item.label, Question.limits.MAX_LABEL) + || !["human", "planner"].includes(item.origin as string) + ) invalid(); + } else invalid(); + } + return value as PendingCardAction[]; +} diff --git a/apps/server/src/questions/card-event.ts b/apps/server/src/questions/card-event.ts new file mode 100644 index 000000000..89c42aaa0 --- /dev/null +++ b/apps/server/src/questions/card-event.ts @@ -0,0 +1,13 @@ +export type CardEvent = + | { kind: "decided"; id: string; threadId?: string; actor: string; optionIds: string[] } + | { kind: "reopened"; id: string; threadId?: string; actor: string } + | { kind: "discarded"; id: string; threadId?: string; actor: string } + | { + kind: "option-added"; + id: string; + threadId?: string; + actor: string; + optionId: string; + label: string; + origin: "chat" | "planner" | "human"; + };