Jev 6: orchestrate bounded message interpretation

apps/server/src/conversation-plan/interpret-batch.test.ts

@@ -0,0 +1,62 @@
+import { expect, test } from "bun:test";
+import { interpretMessage } from "./interpret";
+import { initialState } from "./domain";
+import type { JevQuestion } from "./jev";
+import { message, mockResult, seeded, settledBy } from "./interpret.test-fixtures";
+
+// Exact archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, pipeline.test.ts.
+// Complete callbacks and helpers retained; injected offline transport only.
+
+test("a failed target waits for its sibling before the next message starts", async () => {
+	let current = message(
+		"failed-batch",
+		"Sounds good to me. What about agents? Copilot?",
+		"Jules",
+	);
+	let next = message("next-message", "Should we use GitHub authentication?", "Mina");
+	let release!: (result: ReturnType<typeof mockResult>) => void;
+	let held = new Promise<ReturnType<typeof mockResult>>(resolve => release = resolve);
+	let heldQuestions!: Record<string, JevQuestion>;
+	let targetCalls = 0;
+	let nextStarted = false;
+	let first = interpretMessage({
+		channelId: "channel",
+		message: current,
+		recent: [],
+		state: settledBy(seeded(), "Mina"),
+		ask: request => {
+			if ("new_question" in request.questions) {
+				return Promise.resolve(mockResult(request.questions, { new_question: 0.95 }));
+			}
+			targetCalls++;
+			if ("c0_role" in request.questions) {
+				heldQuestions = request.questions;
+				return held;
+			}
+			if ("c1_role" in request.questions) return Promise.reject(new Error("target failed"));
+			return Promise.resolve(mockResult(request.questions, {}));
+		},
+	});
+	let second = first.then(() =>
+		interpretMessage({
+			channelId: "channel",
+			message: next,
+			recent: [current],
+			state: initialState(),
+			ask: async request => {
+				nextStarted = true;
+				return mockResult(request.questions, {});
+			},
+		})
+	);
+	await Bun.sleep(0);
+	expect(targetCalls).toBe(3);
+	expect(nextStarted).toBe(false);
+	release(mockResult(heldQuestions, {}));
+	let failed = await first;
+	await second;
+	expect(failed.events).toEqual([]);
+	expect(failed.analysis.status).toBe("failed");
+	expect(failed.analysis.passes).toHaveLength(1);
+	expect(nextStarted).toBe(true);
+});

apps/server/src/conversation-plan/interpret-order.test.ts

@@ -0,0 +1,41 @@
+import { expect, test } from "bun:test";
+import { postmarkId, sesId, stateWithOptions } from "./research-offers.test-fixtures";
+import { interpretMessage } from "./interpret";
+import { message, mockResult } from "./interpret.test-fixtures";
+
+// Preserve original targeting dispatch turn; helper extraction must introduce no extra await.
+test.each([false, true])(
+	"research eligible %s retains targeting dispatch before the next queued observation",
+	async eligible => {
+		let targetingCalls = 0;
+		let observe!: (count: number) => void;
+		let observed = new Promise<number>(resolve => observe = resolve);
+		let pending = interpretMessage({
+			channelId: "channel",
+			message: message("dispatch-order", "Should we use GitHub authentication?"),
+			recent: [],
+			state: stateWithOptions("auth", "Which authentication?", [
+				{ id: postmarkId, label: "GitHub" },
+				{ id: sesId, label: "Email" },
+			]),
+			ask: request => {
+				let triage = "new_question" in request.questions;
+				let research = "research_kind" in request.questions;
+				if ((triage && !eligible) || research) {
+					queueMicrotask(() => queueMicrotask(() => observe(targetingCalls)));
+				}
+				if (!triage && !research) targetingCalls++;
+				return Promise.resolve(
+					mockResult(
+						request.questions,
+						triage ? { new_question: 0.95, research_need: eligible ? 0.95 : 0.05 } : {},
+					),
+				);
+			},
+		});
+		expect(await observed).toBe(1);
+		let output = await pending;
+		expect(output.analysis.passes.map(pass => pass.stage)).toEqual(["triage", "targeting"]);
+		expect(output.researchOffer).toBeUndefined();
+	},
+);

apps/server/src/conversation-plan/interpret-research.ts

@@ -0,0 +1,84 @@
+import type { Chat } from "@chopin/protocol";
+import type { JevResult } from "./jev";
+import type { InterpretInput, ResearchOfferCandidate } from "./interpret-types";
+import { confidentChoice, noul } from "./interpret-scoring";
+import { MAX_QUOTE_CANDIDATES } from "./quote-budget";
+import type { QuoteCandidate } from "./quotes";
+
+export type MemberResearchInput = InterpretInput & {
+	message: Chat.Entry & { author: Extract<Chat.Author, { kind: "member" }> };
+};
+
+/** The main interpreter retains the original guard, request await and research catch. */
+export function selectResearchOffer(
+	input: MemberResearchInput,
+	first: JevResult,
+	quotes: QuoteCandidate[],
+	result: JevResult,
+): ResearchOfferCandidate | undefined {
+	let researchOffer: ResearchOfferCandidate | undefined;
+	if (result.model === first.model) {
+		let answers = result.answers;
+		let quoteKey = confidentChoice(answers, "research_quote");
+		let threadId = confidentChoice(answers, "research_thread");
+		let firstId = confidentChoice(answers, "research_option_a");
+		let secondId = confidentChoice(answers, "research_option_b");
+		let kind = confidentChoice(answers, "research_kind");
+		let selectedQuote = quoteKey?.match(/^q(\d+)$/)?.[1];
+		let index = selectedQuote && Number(selectedQuote) < MAX_QUOTE_CANDIDATES
+			? selectedQuote
+			: undefined;
+		let quote = index === undefined ? undefined : quotes[Number(index)];
+		let answered = answers[`research_q${index}_answered`];
+		let scope = confidentChoice(answers, `research_q${index}_scope`);
+		let thread = input.state.threads.find(item => item.id === threadId);
+		let options = thread?.contributions.filter(item => item.kind === "option") ?? [];
+		let source = quote && {
+			messageId: input.message.id,
+			author: input.message.author,
+			quote: quote.quote,
+			start: quote.start,
+			end: quote.end,
+		};
+		let grounded = quote && thread && source
+			&& ["exploring", "leaning", "reopened"].includes(thread.status)
+			&& noul(first.answers, `c${index}_owned_unretracted`) >= 0.8
+			&& noul(answers, `research_q${index}_owned`) >= 0.8
+			&& answered?.type === "noul" && answered.noul <= 0.2
+			&& /\b(?:costs?|pric(?:e|es|ing)|egress|billing|fees?|rates?)\b/i.test(quote.quote);
+		if (
+			grounded && source && thread
+			&& firstId && secondId && firstId !== secondId
+			&& options.length === 2
+			&& options.some(item => item.id === firstId)
+			&& options.some(item => item.id === secondId)
+			&& kind === "current-cost-comparison"
+		) {
+			researchOffer = {
+				source,
+				threadId: thread.id,
+				optionIds: [firstId, secondId],
+			};
+		} else if (
+			grounded && source && thread && kind === "current-cost-concern"
+			&& [3, 4].includes(options.length)
+			&& input.state.threads.filter(item =>
+					["exploring", "leaning", "reopened"].includes(item.status)
+					&& [2, 3, 4].includes(
+						item.contributions.filter(value => value.kind === "option").length,
+					)
+				).length === 1
+			&& scope && scope !== "none"
+			&& (scope === "all-current" || options.some(item => item.id === scope))
+		) {
+			researchOffer = {
+				kind: "current-cost-concern",
+				source,
+				threadId: thread.id,
+				optionIds: options.map(item => item.id),
+				...(scope === "all-current" ? {} : { focusOptionId: scope }),
+			};
+		}
+	}
+	return researchOffer;
+}

apps/server/src/conversation-plan/interpret-scoring.ts

@@ -0,0 +1,23 @@
+import type { JevAnswer } from "./jev";
+
+export function noul(answers: Record<string, JevAnswer>, key: string): number {
+	let answer = answers[key];
+	return answer?.type === "noul" ? answer.noul : 0;
+}
+export function score(answers: Record<string, JevAnswer>, key: string): number {
+	let answer = answers[key];
+	return answer?.type === "score" ? answer.score : 0;
+}
+export function confidentChoice(
+	answers: Record<string, JevAnswer>,
+	key: string,
+): string | undefined {
+	let answer = answers[key];
+	if (answer?.type !== "choice") return;
+	let ranked = Object.values(answer.probabilities).sort((a, b) => b - a);
+	if (
+		answer.confidence < 0.8 || (ranked[0] ?? 0) < 0.8
+		|| (ranked[0] ?? 0) - (ranked[1] ?? 0) < 0.2
+	) return;
+	return answer.choice;
+}

apps/server/src/conversation-plan/interpret-source.test.ts

@@ -0,0 +1,69 @@
+import { expect, test } from "bun:test";
+import { interpretMessage } from "./interpret";
+import { initialState } from "./domain";
+import type { JevQuestion } from "./jev";
+import { message, mockResult, seeded, settledBy } from "./interpret.test-fixtures";
+import { extractQuotes } from "./quotes";
+
+// Exact archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, pipeline.test.ts.
+// Complete callbacks and helpers retained; injected offline transport only.
+
+test("rejects a fifth source quote before Jev or any partial event", async () => {
+	let current = message(
+		"five-quotes",
+		"Use GitHub OAuth. Use magic links. Send user tokens. Use a GitHub App. Keep credentials separate.",
+	);
+	let calls = 0;
+	expect(() => extractQuotes(current.text)).toThrow("source quote count exceeds 4");
+
+	let output = await interpretMessage({
+		channelId: "channel",
+		message: current,
+		recent: [],
+		state: initialState(),
+		ask: async request => {
+			calls++;
+			return mockResult(request.questions, {});
+		},
+	});
+
+	expect(calls).toBe(0);
+	expect(output.events).toEqual([]);
+	expect(output.analysis.status).toBe("failed");
+});
+
+test("a late retraction beyond the ownership context fails closed before Jev", async () => {
+	let calls = 0;
+	let ask = async (request: { questions: Record<string, JevQuestion> }) => {
+		calls++;
+		return mockResult(request.questions, {});
+	};
+	let tooLong = message(
+		"late-retraction",
+		`Sounds good to me. ${"x".repeat(4000)} Actually, no—I withdraw that.`,
+		"Jules",
+	);
+	let blocked = await interpretMessage({
+		channelId: "channel",
+		message: tooLong,
+		recent: [],
+		state: settledBy(seeded(), "Mina"),
+		ask,
+	});
+	expect(blocked.events).toEqual([]);
+	expect(blocked.analysis).toMatchObject({
+		status: "unlinked",
+		policyGate: "message exceeds ownership context",
+		passes: [],
+	});
+	expect(calls).toBe(0);
+	let boundary = message("exact-boundary", `${"x".repeat(3999)}.`, "Jules");
+	await interpretMessage({
+		channelId: "channel",
+		message: boundary,
+		recent: [],
+		state: initialState(),
+		ask,
+	});
+	expect(calls).toBe(1);
+});

apps/server/src/conversation-plan/interpret-targeting.test.ts

@@ -0,0 +1,113 @@
+import { expect, test } from "bun:test";
+import { interpretMessage } from "./interpret";
+import type { JevQuestion } from "./jev";
+import { message, mockResult, seeded, settledBy, stateWithOption } from "./interpret.test-fixtures";
+
+// Exact archive446a9779a937fa5be7cd3eb52fd7f3023d691ed2, pipeline.test.ts.
+// Complete callbacks and helpers retained; injected offline transport only.
+
+test("interprets a declarative pair for a different open topic through both Jev passes", async () => {
+	let state = stateWithOption(
+		"Which search service should we use?",
+		"search-thread",
+		"existing-search",
+		"Use PostgreSQL search.",
+	);
+	let current = message("search-alternatives", "By search I mean Algolia or Meilisearch.");
+	let calls = 0;
+	let output = await interpretMessage({
+		channelId: "channel",
+		message: current,
+		recent: [],
+		state,
+		ask: async request =>
+			mockResult(
+				request.questions,
+				calls++ === 0
+					? { new_option: 0.05, thread_target: "search-thread", significance: 2 }
+					: {
+						c0_role: "none",
+						c1_role: "none",
+						c0_thread: "search-thread",
+						c1_thread: "search-thread",
+						c0_new_option: 0.05,
+						c1_new_option: 0.85,
+					},
+			),
+	});
+	expect(calls).toBe(3);
+	expect(output.events.map(event => event.type)).toEqual(["option.added", "option.added"]);
+	expect(output.events.map(event => "source" in event && event.source)).toMatchObject([
+		{ quote: "Algolia", start: 17, end: 24, role: "option" },
+		{ quote: "Meilisearch", start: 28, end: 39, role: "option" },
+	]);
+});
+
+test("a later target failure or inconsistent model cannot publish partial events", async () => {
+	let current = message("partial", "Sounds good to me. What about agents? Copilot?", "Jules");
+	for (let failure of ["transport", "model"] as const) {
+		let calls = 0;
+		let output = await interpretMessage({
+			channelId: "channel",
+			message: current,
+			recent: [message("proposal", "Let's use GitHub.", "Mina")],
+			state: settledBy(seeded(), "Mina"),
+			ask: async request => {
+				calls++;
+				if (calls === 4 && failure === "transport") throw new Error("synthetic failure");
+				let result = mockResult(request.questions, {
+					new_question: 0.95,
+					c0_owned_unretracted: 0.95,
+					c1_owned_unretracted: 0.95,
+					c2_owned_unretracted: 0.95,
+					c0_role: "support",
+					c0_thread: "thread-a",
+					c0_support: 0.95,
+					c0_agrees_with_settle: 0.95,
+					c1_role: "question",
+					c1_thread: "new",
+				});
+				if (calls === 3 && failure === "model") result.model = "other-model";
+				return result;
+			},
+		});
+		expect(calls).toBe(4);
+		expect(output.events).toEqual([]);
+		expect(output.analysis.status).toBe("failed");
+		expect(output.analysis.passes).toHaveLength(1);
+	}
+});
+
+test("out-of-order target replies retain their quote-specific answers", async () => {
+	let current = message(
+		"out-of-order",
+		"Sounds good to me. What about agents? Copilot?",
+		"Jules",
+	);
+	let release: Array<(result: ReturnType<typeof mockResult>) => void> = [];
+	let requests: Array<Record<string, JevQuestion>> = [];
+	let interpretation = interpretMessage({
+		channelId: "channel",
+		message: current,
+		recent: [],
+		state: settledBy(seeded(), "Mina"),
+		ask: request => {
+			if ("new_question" in request.questions) {
+				return Promise.resolve(mockResult(request.questions, { new_question: 0.95 }));
+			}
+			requests.push(request.questions);
+			return new Promise(resolve => release.push(resolve));
+		},
+	});
+	await Bun.sleep(0);
+	expect(requests).toHaveLength(3);
+	for (let index of [2, 1, 0]) {
+		release[index]!(mockResult(requests[index]!, { [`c${index}_new_option`]: (index + 1) / 10 }));
+	}
+	let output = await interpretation;
+	expect(output.analysis.passes.map(pass => pass.stage)).toEqual(["triage", "targeting"]);
+	let answers = output.analysis.passes[1]!.answers;
+	for (let index = 0; index < 3; index++) {
+		expect(answers[`c${index}_new_option`]).toEqual({ type: "noul", noul: (index + 1) / 10 });
+	}
+});