Jev: orchestrate durable conversation processing and Planner jobs

apps/server/src/chat/job-authenticated-memory.test-fixtures.ts

@@ -0,0 +1,97 @@
+import * as Chat from "./service";
+import * as Plan from "../plan/service";
+import { Sessions } from "../auth/session";
+import { ActiveOwnerBindings } from "../agent/active-owner";
+import { createPlannerAgent } from "../harness/agents";
+import { openPlannerSession } from "../harness/session";
+import { headingHarness } from "./job-heading-harness.test-fixtures";
+import type { ConversationPlan } from "@chopin/protocol";
+import type { HostedAuth } from "../auth/routes";
+import type { openPlan } from "../testing/plan";
+
+function clock(): Date {
+	return new Date();
+}
+
+async function noop(_output?: unknown): Promise<void> {}
+
+export async function authenticatedMemory(
+	opened: Awaited<ReturnType<typeof openPlan>>,
+	kind: ConversationPlan.JobKind = "heading",
+	call?: { name: string; input: Record<string, unknown> },
+) {
+	let plan = opened.plan;
+	let sessions = new Sessions(opened.storage, true, clock);
+	let session = await sessions.issue("U_test", {
+		accessToken: "test-token",
+		accessExpiresIn: 28_800,
+		refreshToken: "test-refresh",
+		refreshExpiresIn: 15_897_600,
+	});
+	let repository = {
+		id: opened.channel.repositoryId,
+		owner: "owner",
+		name: "repository",
+		defaultBranch: "main",
+	};
+	let auth = {
+		storage: opened.storage,
+		sessions,
+		clock,
+		github: {
+			repositoryAccess: async () => ({ ...repository, permissions: { push: true, admin: false } }),
+		},
+		admission: { allowed: async () => true },
+	} as unknown as HostedAuth;
+	let owners = new ActiveOwnerBindings(auth);
+	let startCheck: () => Promise<void> = noop;
+	let resultCheck: (output: unknown) => Promise<void> = noop;
+	let driver = headingHarness(() => startCheck(), output => resultCheck(output), call);
+	let headingAgent = createPlannerAgent(driver.fake, kind);
+	let sandboxDestroys = 0;
+	let context: Chat.Room = {
+		chat: plan.chat,
+		plan,
+		server: opened.server,
+		room: opened.channel.id,
+		config: { agent: true } as Chat.Room["config"],
+		auth,
+		repository,
+		claimantSessionId: session.id,
+		activeOwner: () => owners.resolve(opened.channel.id),
+		persist: () => Plan.persist(plan),
+		openPlannerSession: (owner, channel) =>
+			openPlannerSession(owner, channel, {
+				headingAgent,
+				refineAgent: headingAgent,
+				proseAgent: headingAgent,
+				githubTools: async () => ({ ok: true, value: {} }),
+				createSandbox: async () =>
+					({
+						defaultWorkingDirectory: "/tmp",
+						async run() {
+							return { exitCode: 0, stdout: "", stderr: "" };
+						},
+						async destroy() {
+							sandboxDestroys++;
+						},
+					}) as never,
+				registerCredential: () => () => {},
+			}),
+	};
+
+	return {
+		context,
+		driver,
+		sessionId: session.id,
+		revokeAll: () => owners.revokeAll(),
+		invalidateOwner: () => owners.revokeCredential(session.id, 1),
+		sandboxDestroys: () => sandboxDestroys,
+		onStart(check: typeof startCheck) {
+			startCheck = check;
+		},
+		onResult(check: typeof resultCheck) {
+			resultCheck = check;
+		},
+	};
+}

apps/server/src/chat/job-card-memory.test-fixtures.ts

@@ -0,0 +1,131 @@
+import { expect } from "bun:test";
+import * as Chat from "./service";
+import * as Plan from "../plan/service";
+import * as Questions from "../questions/service";
+import { openPlan } from "../testing/plan";
+import { authenticatedMemory } from "./job-authenticated-memory.test-fixtures";
+import { createPlannerJobs } from "../conversation-plan/planner-jobs";
+import { PROMPT_FOR } from "../conversation-plan/job-prompts";
+import { JOB_TOOLS } from "../agent/job-scope";
+import { documentTools } from "../agent/tools";
+import { scopedJobTools } from "../agent/job-tools";
+
+export const CARD = "01K0N4TR8K7JGM4R1J7PW4R8YJ";
+export const QUESTION = "01K0N4V4E7Y6P4MJ5WD8XZF3B2";
+export const OPTION = "01K0N4W3B7P27CBAEC7A8C8WEA";
+const WIDGET = `<Questionnaire id="${CARD}" status="decided" thread="t1">
+<Question id="${QUESTION}" header="Auth" prompt="Which authentication approach?" multiple="false">
+<Option id="${OPTION}" label="GitHub Apps" />
+<Answer value="GitHub Apps" choices="${OPTION}" />
+</Question>
+</Questionnaire>
+`;
+
+export async function cardMemory(kind: "refine" | "suggest" | "prose") {
+	let answered = Questions.normalizeRecord({
+		id: CARD,
+		definition: {
+			questions: [{
+				id: QUESTION,
+				header: "Auth",
+				question: "Which authentication approach?",
+				multiple: false,
+				options: [{ id: OPTION, label: "GitHub Apps", description: "" }],
+			}],
+		},
+		status: "answered",
+		origin: "conversation",
+		threadId: "t1",
+		owner: "mina",
+		decidedAt: 1_758_645_000,
+		choices: [OPTION],
+	});
+	let opened = kind === "prose"
+		? await openPlan(`Context.\n\n${WIDGET}`, { questions: [answered] })
+		: await openPlan("Opening prose.\n\nLater prose.\n");
+	let plan = opened.plan;
+	let id = kind === "prose"
+		? CARD
+		: await Questions.insertConversationCard(plan, opened.server, opened.channel.id, {
+			threadId: "thread-a",
+			header: "Authentication",
+			question: "What auth system should we use?",
+			options: [],
+		});
+	plan.chat.entries.push({
+		id: "m1",
+		author: { kind: "member", handle: "ana" },
+		text: "We need to choose authentication.",
+		ts: 1,
+	});
+	await Plan.persist(plan);
+	let raw = kind === "prose"
+		? { revision: plan.revision, id, text: "We chose GitHub Apps." }
+		: kind === "refine"
+		? { revision: plan.revision, id, title: "How should people sign in?", add_options: [] }
+		: {
+			revision: plan.revision,
+			id,
+			add_options: [{ label: "Passkeys", rationale: "Avoids passwords for account access." }],
+		};
+	let identity = await authenticatedMemory(opened, kind, { name: JOB_TOOLS[kind], input: raw });
+	let errors: unknown[] = [];
+	identity.onStart(async () => {
+		let before = Plan.source(plan);
+		let guarded = scopedJobTools(documentTools, Chat.documentRoom(identity.context));
+		let result = await guarded.edit_plan!.execute!(
+			{
+				revision: plan.revision,
+				operations: [{ op: "insert_root", source: "Unauthorized paragraph." }],
+			},
+			{
+				toolCallId: "foreign",
+				messages: [],
+				context: { room: Chat.documentRoom(identity.context) },
+			} as never,
+		);
+		expect(String(result)).toContain(`use only ${JOB_TOOLS[kind]}`);
+		expect(Plan.source(plan)).toBe(before);
+	});
+	identity.onResult(async output => {
+		expect(String(output)).toContain('"ok": true');
+		identity.driver.finish();
+	});
+	let jobs = createPlannerJobs({
+		plan,
+		exclusive: action => Plan.exclusive(plan, action),
+		persist: () => Plan.persistExclusive(plan),
+		runner: (job, prompt, signal) =>
+			Chat.job(identity.context, job, prompt, identity.sessionId, signal),
+		prompt: job => PROMPT_FOR[job.kind]?.(plan, job),
+		publishJobs: current =>
+			opened.server.publish(
+				opened.channel.id,
+				JSON.stringify({ kind: "conversation-plan:jobs", jobs: current }),
+			),
+		publishMeta: target => Questions.announce(plan, opened.server, opened.channel.id, target),
+		activity: async text => {
+			await Chat.noticeExclusive(identity.context, text);
+		},
+		onError: error => errors.push(error),
+	});
+	let trigger = kind === "prose" ? `decided:${CARD}:1` : "m1";
+	return {
+		...identity,
+		opened,
+		plan,
+		jobs,
+		id,
+		errors,
+		async run() {
+			await jobs.enqueue({ kind, target: id, trigger });
+			await jobs.idle();
+		},
+		async close() {
+			jobs.stop();
+			identity.revokeAll();
+			await jobs.idle();
+			await Plan.close(plan);
+		},
+	};
+}

apps/server/src/chat/job-card-memory.test.ts

@@ -0,0 +1,54 @@
+import { installJobCleanup } from "./job.test-fixtures";
+import { expect, test } from "bun:test";
+import * as Plan from "../plan/service";
+import { cardMemory } from "./job-card-memory.test-fixtures";
+
+installJobCleanup();
+
+for (let kind of ["refine", "suggest", "prose"] as const) {
+	test(`current ${kind} Harness profile executes its own durable card tool and reopens without replay`, async () => {
+		let h = await cardMemory(kind);
+		try {
+			await h.run();
+			expect(h.plan.conversationPlanJobs).toMatchObject([{ status: "done", attempts: 1 }]);
+			let record = h.plan.records.get(h.id)!;
+			if (kind === "refine") {
+				expect(record.definition.questions[0]?.question).toBe("How should people sign in?");
+			}
+			if (kind === "suggest") {
+				expect(record.definition.questions[0]?.options.map(option => option.label)).toEqual([
+					"Passkeys",
+				]);
+			}
+			if (kind === "prose") {
+				expect(Plan.source(h.plan)).toContain("We chose GitHub Apps.");
+				expect(record.prose).toHaveLength(1);
+				expect(record.prose?.[0]?.orphaned).not.toBe(true);
+			}
+			expect(h.plan.chat.entries).toHaveLength(2);
+			expect(h.plan.chat.entries[0]?.text).toBe("We need to choose authentication.");
+			expect(h.plan.chat.entries[1]?.author.kind).toBe("system");
+			expect(h.plan.chat.entries.some(entry => entry.tools?.length || entry.streaming)).toBe(false);
+			expect(
+				h.opened.broadcasts.some(frame =>
+					frame.kind === "chat:tool" || frame.kind === "chat:delta"
+				),
+			).toBe(false);
+			expect(h.plan.chat.job).toBeUndefined();
+			expect(h.errors).toEqual([]);
+			await h.close();
+			let reopened = await Plan.open(h.opened.channel.id, h.opened.backend, h.opened.server);
+			try {
+				expect(reopened.records.get(h.id)?.definition).toEqual(record.definition);
+				expect(reopened.conversationPlanJobs).toMatchObject([{ status: "done", attempts: 1 }]);
+				expect(reopened.chat.waiting).toEqual([]);
+				expect(h.driver.starts()).toBe(1);
+				if (kind === "prose") expect(Plan.source(reopened)).toContain("We chose GitHub Apps.");
+			} finally {
+				await Plan.close(reopened);
+			}
+		} finally {
+			if (!h.plan.chat.closed) await h.close();
+		}
+	});
+}