Let a person's message to chopin stop a background job instead of being refused

AGENTS.md

@@ -343,10 +343,18 @@ acknowledged at once and given to that turn (`PlannerSession.steer`, through the
 harness's own mid-run input: Atomic's `steer`, a `steer` frame to a remote client
 that offered `steer-turn`), which reads it at its next step as `@handle: text`, in the
 same turn. Steering takes the same write access as starting a turn: `chat:send` is
-checked before `Chat.send`, so a reader cannot steer. A harness that cannot take
-input mid-run (the Copilot SDK), or a running turn that is a background job's, refuses
-the message with `planner-busy`: "chopin can't take messages while it works on this
-machine. Stop it to ask now." (the server log names the harness). A machine that is away,
+checked before `Chat.send`, so a reader cannot steer. A background job never makes a
+person wait: a message to the Planner (or an `invoke_planner` message) while a job's turn
+runs, or while one is being handed the Chat, is acknowledged at once and the job gives way
+(`preempt` in `apps/server/src/chat/service.ts`). Its turn is stopped and starts the
+person's turn as it ends; a write its own tool has under way finishes first and stands as its
+result, and with none it goes back to pending, its attempts unspent, and runs again when that
+turn's end wakes the queue. One job (kind and target) gives way at most three times in a row;
+the next run is let finish, with the person's turn right behind it. A turn a person started, or
+one Chopin started for a Build or a comment, is never preempted. A harness that cannot take
+input mid-run (the Copilot SDK) refuses a message while a turn runs with `planner-busy`:
+"chopin can't take messages while it works on this machine. Stop it to ask now." (the server
+log names the harness). A machine that is away,
 in its hand-over window before the client resumes ("Still connecting to @mona's machine, so
 this message wasn't sent. Send it again in a moment."), or with none attached, refuses it with
 the connection's own words (`planner-not-attached`),
@@ -357,7 +365,7 @@ to be given to, and is answered as a passing trouble (`temporarily-unavailable`)
 turn ends before it can read the message, an idle Chat starts a turn for it and any other
 says plainly that Chopin did not read it. A comment note, an `invoke_planner as system` call
 and a task draft need a turn of their own, so a running turn refuses them before anything is
-saved. Only Chopin's own scheduling still waits for the running turn, in `chat.waiting`:
+saved. Only Chopin's own scheduling still waits for the running turn (or, for a job, a person's), in `chat.waiting`:
 background jobs and server-authored instructions nobody typed (initiative, an ended workflow
 run, readiness, decision follow-ups). `planner_deferred_turns` is no longer written or read;
 at start the writer-lease holder closes any row an earlier release left open

apps/server/src/agent/job-scope.ts

@@ -79,7 +79,25 @@ export async function runJobTool(
 	if (tool !== JOB_TOOLS[job.kind]) {
 		throw new Error("a background Planner job may run only its own tool");
 	}
+	// A job giving way to a person's turn writes nothing more; it runs again later.
+	if (chat.backgroundTurn?.preempted) {
+		throw new Error("a person's turn took the Planner; this job runs again later");
+	}
+	// Watched by a preemption, which lets a write under way finish before the turn stops.
+	let writing = record(chat, job, produce);
+	chat.jobWriting = writing;
+	try {
+		return await writing;
+	} finally {
+		if (chat.jobWriting === writing) chat.jobWriting = undefined;
+	}
+}
 
+async function record(
+	chat: Chat,
+	job: Job,
+	produce: (job: Job) => Promise<{ output: unknown } & Record<string, unknown>>,
+): Promise<unknown> {
 	let turn = chat.turn;
 	let id = job.id;
 	let { output, ...reply } = await produce(job);

apps/server/src/chat/job-fence.ts

@@ -6,8 +6,11 @@ export function watchJobAbort(chat: Chat, turn: JobTurn, signal: AbortSignal): (
 	let currentTurn = chat.turn;
 	function abort() {
 		if (chat.job !== turn.job || chat.turn !== currentTurn) return;
+		// A job that gave way to a person keeps what its own tool wrote, and was not interrupted.
+		if (turn.preempted) turn.output = chat.jobOutput;
 		chat.job = undefined;
 		chat.jobOutput = undefined;
+		if (turn.preempted) return;
 		chat.interruption ??= chat.closed
 			? "The document closed."
 			: signal.reason instanceof Error && signal.reason.name !== "AbortError"

apps/server/src/chat/job-queue.test.ts

@@ -31,7 +31,7 @@ test.each([false, true])(
 	},
 );
 
-test("a job waits for a human turn, and a person's message while a job runs is refused, without leaking job scope", async () => {
+test("a job waits for a human turn, and a person's message while a job runs preempts it, without leaking job scope", async () => {
 	let first = harness();
 	await Chat.send(first.context, socket(), member("Human first"));
 	await until(() => first.prompts.length === 1);
@@ -47,29 +47,25 @@ test("a job waits for a human turn, and a person's message while a job runs is r
 	let second = harness();
 	let ahead = Chat.job(second.context, JOB, "job first", "session");
 	await until(() => second.prompts.length === 1);
-	// A job's session is isolated: nobody talks to it, and nothing waits for it to end.
-	let refusals: Array<Record<string, unknown>> = [];
+	// A job's session is isolated: nobody talks to it, and a person's message takes its place.
+	let frames: Array<Record<string, unknown>> = [];
 	let sender = {
 		data: { handle: "ana", principalId: "U_ana" },
-		send: (frame: string) => refusals.push(JSON.parse(frame)),
+		send: (frame: string) => frames.push(JSON.parse(frame)),
 	} as unknown as Parameters<typeof Chat.send>[1];
 	await Chat.send(second.context, sender, member("Human second"));
-	expect(refusals).toEqual([
-		expect.objectContaining({
-			kind: "session:error",
-			message: "chopin can't take messages while it works on this machine. Stop it to ask now.",
-			code: "planner-busy",
-		}),
-	]);
+	expect(frames).toEqual([expect.objectContaining({ kind: "chat:send" })]);
+	expect(await ahead).toMatchObject({ status: "preempted" });
+	await until(() => second.prompts.length === 2);
+	// The person's turn has no job scope: the job's tool is refused to it.
+	expect(second.chat.job).toBeUndefined();
+	await expect(
+		runJobTool(second.chat, "refine_decision", async () => ({ output: { added: 0 } })),
+	).rejects.toThrow();
 	expect(second.chat.waiting).toHaveLength(0);
-	expect(second.chat.entries).toEqual([]);
-	await runJobTool(second.chat, "refine_decision", async () => ({ output: { added: 0 } }));
+	expect(second.steered).toEqual([]);
 	second.emit("session.idle");
-	expect(await ahead).toEqual({ status: "done", output: '{"added":0}' });
 	await until(() => !second.chat.busy);
-	expect(second.chat.job).toBeUndefined();
-	expect(second.prompts).toEqual(["job first"]);
-	expect(second.steered).toEqual([]);
 });
 
 test("archiving cancels a job queued behind a human turn without dropping human work", async () => {

apps/server/src/chat/job-queue.ts

@@ -11,6 +11,10 @@ export type JobTurn = {
 	prompt: string;
 	done: (outcome: JobOutcome) => void;
 	settled?: boolean;
+	/** A person's message cut its turn short; it runs again unless its own tool already wrote. */
+	preempted?: true;
+	/** What its own tool recorded before its turn was cut short. */
+	output?: string;
 };
 
 export function jobReason(error: unknown): string {

apps/server/src/chat/preempt.test.ts

@@ -0,0 +1,213 @@
+/**
+ * A person's message to the Planner is never refused because a background job is running: it
+ * preempts the job's turn, which goes back to wait and runs again once the person's turn is over.
+ */
+import { expect, test } from "bun:test";
+
+import * as Chat from "./service";
+import { runJobTool } from "../agent/job-scope";
+import { harness, installJobCleanup, JOB, member, until } from "./job.test-fixtures";
+
+import type { Socket } from "../wire";
+
+installJobCleanup();
+
+type Frame = { kind: string; [key: string]: unknown };
+
+function speaking(handle: string, frames: Frame[] = []): Socket {
+	return {
+		data: { handle, principalId: `U_${handle}` },
+		send: (raw: string) => frames.push(JSON.parse(raw) as Frame),
+	} as unknown as Socket;
+}
+
+test("a message to the Planner while a background job runs starts the person's turn at once, and the job waits to run again", async () => {
+	let h = harness();
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 1);
+	// A job's turn is quiet: Chat draws no turn for it and offers no Stop.
+	expect(h.chat.turn?.quiet).toBe(true);
+
+	let frames: Frame[] = [];
+	let ask = member("@chopin what is the plan?");
+	await Chat.send(h.context, speaking("bo", frames), ask);
+
+	// Accepted, never refused, and the job's turn gives way to the person's.
+	expect(frames).toEqual([expect.objectContaining({ kind: "chat:send", id: ask.requestId })]);
+	expect(await job).toEqual({ status: "preempted", reason: "A person asked the Planner." });
+	await until(() => h.prompts.length === 2);
+	expect(h.prompts[1]).toContain("what is the plan?");
+	expect(h.chat.job).toBeUndefined();
+	expect(h.chat.turn?.handle).toBe("bo");
+	// The person's turn is one people watch, and can talk to.
+	expect(h.chat.turn?.quiet).toBeUndefined();
+	expect(h.chat.steering).toBeDefined();
+	let shown = h.frames.filter(frame => frame.kind === "chat:state").at(-1);
+	expect(shown).toMatchObject({ busy: true, turn: { handle: "bo" } });
+	expect(shown?.turn).not.toHaveProperty("quiet");
+	h.emit("session.idle");
+	await until(() => !h.chat.busy);
+});
+
+test("a job whose own tool already wrote when a person asks is done, not run again", async () => {
+	let h = harness();
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 1);
+	await runJobTool(h.chat, "refine_decision", async () => ({ output: { added: 1 } }));
+
+	await Chat.send(h.context, speaking("bo"), member("@chopin now?"));
+	expect(await job).toEqual({ status: "done", output: '{"added":1}' });
+	await until(() => h.prompts.length === 2);
+	h.emit("session.idle");
+	await until(() => !h.chat.busy);
+});
+
+test("a write the job's tool has under way finishes whole and is its result; no write starts after", async () => {
+	let h = harness();
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 1);
+	let write = Promise.withResolvers<void>();
+	let writing = runJobTool(h.chat, "refine_decision", async () => {
+		await write.promise;
+		return { output: { added: 2 } };
+	});
+
+	await Chat.send(h.context, speaking("bo"), member("@chopin now?"));
+	// The job's turn is still there, waiting for its write: nothing else may write meanwhile.
+	await Bun.sleep(5);
+	expect(h.prompts).toHaveLength(1);
+	expect(h.aborts()).toBe(0);
+	await expect(
+		runJobTool(h.chat, "refine_decision", async () => ({ output: { added: 3 } })),
+	).rejects.toThrow("a person's turn took the Planner");
+
+	write.resolve();
+	expect(await writing).toMatchObject({ ok: true, added: 2 });
+	expect(await job).toEqual({ status: "done", output: '{"added":2}' });
+	await until(() => h.prompts.length === 2);
+	expect(h.prompts[1]).toContain("now?");
+	h.emit("session.idle");
+	await until(() => !h.chat.busy);
+});
+
+test("a job that keeps meeting messages gives way three times, then finishes with the person's turn right behind it", async () => {
+	let h = harness();
+	for (let round = 1; round <= 3; round++) {
+		let job = Chat.job(h.context, JOB, "job prompt", "session");
+		await until(() => h.prompts.length === 2 * round - 1);
+		await Chat.send(h.context, speaking("bo"), member(`@chopin question ${round}`));
+		expect(await job).toMatchObject({ status: "preempted" });
+		await until(() => h.prompts.length === 2 * round);
+		h.emit("session.idle");
+		await until(() => !h.chat.busy);
+	}
+
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 7);
+	// Run again after giving way, the job's turn is as quiet as ever.
+	expect(h.chat.turn?.quiet).toBe(true);
+	let frames: Frame[] = [];
+	let ask = member("@chopin question 4");
+	await Chat.send(h.context, speaking("bo", frames), ask);
+	// Still accepted at once; the job is let finish this time.
+	expect(frames).toEqual([expect.objectContaining({ kind: "chat:send", id: ask.requestId })]);
+	await Bun.sleep(5);
+	expect(h.prompts).toHaveLength(7);
+	expect(h.chat.job).toBe(JOB);
+	await runJobTool(h.chat, "refine_decision", async () => ({ output: { added: 0 } }));
+	h.emit("session.idle");
+	expect(await job).toEqual({ status: "done", output: '{"added":0}' });
+	await until(() => h.prompts.length === 8);
+	expect(h.prompts[7]).toContain("question 4");
+	h.emit("session.idle");
+	await until(() => !h.chat.busy);
+
+	// Finished, it starts over: the next run gives way again.
+	let next = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 9);
+	await Chat.send(h.context, speaking("bo"), member("@chopin question 5"));
+	expect(await next).toMatchObject({ status: "preempted" });
+	await until(() => h.prompts.length === 10);
+	h.emit("session.idle");
+	await until(() => !h.chat.busy);
+});
+
+test("a job being handed the Chat when a person asks waits behind their turn, then runs", async () => {
+	let h = harness();
+	await Chat.send(h.context, speaking("ana"), member("@chopin first"));
+	await until(() => h.prompts.length === 1);
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	expect(h.chat.waiting).toHaveLength(1);
+
+	// The save that hands the job the Chat is slow.
+	let persist = h.context.persist;
+	let handoff = Promise.withResolvers<void>();
+	let armed = true;
+	h.context.persist = async chat => {
+		if (armed && h.chat.starting) {
+			armed = false;
+			await handoff.promise;
+		}
+		await persist(chat);
+	};
+	h.emit("session.idle");
+	await until(() => !!h.chat.handoffJob);
+
+	let frames: Frame[] = [];
+	let ask = member("@chopin second");
+	await Chat.send(h.context, speaking("bo", frames), ask);
+	expect(frames).toEqual([expect.objectContaining({ kind: "chat:send", id: ask.requestId })]);
+	handoff.resolve();
+	await until(() => h.prompts.length === 2);
+	expect(h.prompts[1]).toContain("second");
+	expect(h.chat.waiting.map(item => item.job?.job)).toEqual([JOB]);
+
+	h.emit("session.idle");
+	await until(() => h.prompts.length === 3);
+	expect(h.prompts[2]).toBe("job prompt");
+	await runJobTool(h.chat, "refine_decision", async () => ({ output: { added: 0 } }));
+	h.emit("session.idle");
+	expect(await job).toEqual({ status: "done", output: '{"added":0}' });
+	await until(() => !h.chat.busy);
+});
+
+test("a harness that takes no message mid-run still starts the person's turn in place of a job", async () => {
+	let h = harness();
+	h.context.steers = undefined;
+	h.context.config.harness = "copilot-sdk";
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 1);
+	let frames: Frame[] = [];
+	let ask = member("@chopin now?");
+	await Chat.send(h.context, speaking("bo", frames), ask);
+	expect(frames).toEqual([expect.objectContaining({ kind: "chat:send", id: ask.requestId })]);
+	expect(await job).toMatchObject({ status: "preempted" });
+	await until(() => h.prompts.length === 2);
+	h.emit("session.idle");
+	await until(() => !h.chat.busy);
+});
+
+test("an MCP caller's instruction preempts a background job too, and is settled when its turn ends", async () => {
+	let h = harness();
+	let job = Chat.job(h.context, JOB, "job prompt", "session");
+	await until(() => h.prompts.length === 1);
+	let settled = false;
+	expect(
+		await Chat.invoke(
+			h.context,
+			{ id: "U_bo", login: "bo" },
+			"Plan this.",
+			undefined,
+			undefined,
+			() => {
+				settled = true;
+			},
+		),
+	).toBeUndefined();
+	expect(await job).toMatchObject({ status: "preempted" });
+	await until(() => h.prompts.length === 2);
+	expect(h.prompts[1]).toContain("Plan this.");
+	expect(settled).toBe(false);
+	h.emit("session.idle");
+	await until(() => settled);
+});