Simplify dispatch: agent-bound bus writer, plain JudgeDecision, no stray exports

packages/harness/README.md

@@ -78,8 +78,11 @@ the next round.
 
 `WriteAheadAgentBus` is an ordered append-only message log. It knows nothing
 about any actor: `append({ actor, kind, payload })` records a message with
-identity, ordering, and time, and `read(actor?)` returns the full history. An
-optional `allow` hook decides whether a given append or read may proceed.
+identity, ordering, and time, and `read(actor?)` returns the full history.
+`agent(actor)` binds one actor to the bus and returns a writer — `write(kind,
+payload?)` — so a caller that always writes as the same agent states the actor
+once. An optional `allow` hook decides whether a given append or read may
+proceed.
 Configure `redact` when payloads may contain sensitive application data.
 
 Storage is [unstorage](https://unstorage.unjs.io) — the bus owns no storage

packages/harness/src/bus.ts

@@ -9,6 +9,10 @@ export type AgentBusAccess =
 	| Readonly<{ operation: "append"; actor: string; kind: string }>
 	| Readonly<{ operation: "read"; actor?: string }>;
 
+/** Appends messages for one bound agent: the actor is fixed once, so each
+ * entry needs only its kind and optional payload. */
+export type AgentWriter = (kind: string, payload?: unknown) => Promise<AgentBusEntry>;
+
 /** Entries live under this key prefix in the configured storage. */
 const entryPrefix = "entry";
 
@@ -48,6 +52,12 @@ export class WriteAheadAgentBus {
 		this.#allow = settings.allow ?? (() => true);
 	}
 
+	/** Binds one agent to this bus: the returned writer appends that agent's
+	 * messages without restating the actor at every call site. */
+	agent(actor: string): AgentWriter {
+		return (kind, payload) => this.append({ actor, kind, ...(payload === undefined ? {} : { payload }) });
+	}
+
 	/** Appends one message and returns the recorded entry. */
 	async append(message: AgentMessage): Promise<AgentBusEntry> {
 		const parsed = agentMessage.parse(message);

packages/harness/src/dispatch.ts

@@ -1,7 +1,9 @@
 import type { WriteAheadAgentBus } from "./bus.js";
-import { judgeDecision, type AgentBusEntry, type JudgeDecision } from "./schema.js";
+import type { AgentBusEntry } from "./schema.js";
 
-export type { JudgeDecision } from "./schema.js";
+/** The only verdicts a judge may return. Gates and judges are typed against
+ * this union, so their outputs are used as returned — nothing re-parses them. */
+export type JudgeDecision = "pass" | "fail";
 
 /** Decides whether a proposed action may execute. In this harness the gate is
  * implemented by the judge — an ordinary actor whose verdict is recorded on
@@ -21,8 +23,21 @@ export class AgentActionDeniedError extends Error {
 	}
 }
 
-/** The message kind the gate's verdicts are recorded under. */
-export const decisionKind = "agent.decision";
+// The names the write-ahead convention itself writes, spelled once here. Bus
+// entries are serialized to storage, so these must stay plain strings —
+// symbols would not survive the round trip.
+const judgeActor = "judge";
+const decisionKind = "agent.decision";
+const failureOf = (kind: string) => `${kind}.failed`;
+const completionOf = (kind: string) => `${kind}.completed`;
+
+/** Records a verdict on the bus as the judge's own message. */
+export function recordDecision(
+	bus: WriteAheadAgentBus,
+	payload: Readonly<{ subject: string; decision: JudgeDecision; [detail: string]: unknown }>,
+): Promise<AgentBusEntry> {
+	return bus.agent(judgeActor)(decisionKind, payload);
+}
 
 /** The write-ahead convention, layered on top of the plain message bus:
  * record the intent, ask the gate, record the verdict as the judge's own
@@ -36,35 +51,33 @@ export async function dispatchAction<T>(
 	gate: ActionGate | undefined,
 	execute: () => Promise<T> | T,
 ): Promise<T> {
-	const action = await bus.append({ actor, kind, ...(payload === undefined ? {} : { payload }) });
+	const agent = bus.agent(actor);
+	const action = await agent(kind, payload);
 	if (gate) {
 		let decision: JudgeDecision;
 		try {
-			decision = judgeDecision.parse(await gate(action, await bus.read()));
+			decision = await gate(action, await bus.read());
 		} catch (error) {
 			// The gate error is the outcome; a failing failure record must not replace it.
-			await bus.append({
-				actor,
-				kind: `${kind}.failed`,
-				payload: { actionId: action.id, stage: "gate", message: errorMessage(error) },
-			}).catch(() => undefined);
+			await agent(failureOf(kind), { actionId: action.id, stage: "gate", message: errorMessage(error) })
+				.catch(() => undefined);
 			throw error;
 		}
-		await bus.append({ actor: "judge", kind: decisionKind, payload: { subject: "action", actionId: action.id, decision } });
+		await recordDecision(bus, { subject: "action", actionId: action.id, decision });
 		if (decision === "fail") throw new AgentActionDeniedError(action);
 	}
 	let result: T;
 	try {
 		result = await execute();
 	} catch (error) {
 		// Likewise: the execution error is the outcome, recorded best-effort.
-		await bus.append({ actor, kind: `${kind}.failed`, payload: { actionId: action.id, message: errorMessage(error) } })
+		await agent(failureOf(kind), { actionId: action.id, message: errorMessage(error) })
 			.catch(() => undefined);
 		throw error;
 	}
 	// Recorded after the fact, outside the catch: a failing completion append
 	// surfaces as a bus error, never as a failed action.
-	await bus.append({ actor, kind: `${kind}.completed`, payload: { actionId: action.id, result } });
+	await agent(completionOf(kind), { actionId: action.id, result });
 	return result;
 }
 

packages/harness/src/harness.ts

@@ -1,6 +1,17 @@
 import { WriteAheadAgentBus } from "./bus.js";
-import { dispatchAction, decisionKind, type ActionGate } from "./dispatch.js";
-import { candidateKey, judgeDecision, roundLimit, rubricText, type AgentBusEntry, type JudgeDecision } from "./schema.js";
+import { dispatchAction, recordDecision, type ActionGate, type JudgeDecision } from "./dispatch.js";
+import { candidateKey, roundLimit, rubricText, type AgentBusEntry } from "./schema.js";
+
+// The run's actors — named for the HarnessInput callbacks they run — and the
+// message kinds they write, spelled once here. Bus entries are serialized to
+// storage, so both must stay plain strings rather than symbols.
+const actors = { student: "student", teacher: "teacher", adversary: "adversary" } as const;
+const kinds = {
+	propose: "agent.propose",
+	assess: "agent.assess",
+	challenge: "agent.challenge",
+	reviseRubric: "agent.revise-rubric",
+} as const;
 
 /** Shapes the bus history handed to actors and the judge. The bus itself does
  * no context management: windowing, rolling summaries, or any other
@@ -151,8 +162,8 @@ export function defineTrainingHarness<TCandidate, TAssessment, TFeedback>(
 			): Promise<JudgeDecision> => {
 				const decision = judge === undefined
 					? fallback()
-					: judgeDecision.parse(await judge(Object.freeze({ ...request, context: await provide(await bus.read()) })));
-				await bus.append({ actor: "judge", kind: decisionKind, payload: { subject: request.subject, ...payload, decision } });
+					: await judge(Object.freeze({ ...request, context: await provide(await bus.read()) }));
+				await recordDecision(bus, { subject: request.subject, ...payload, decision });
 				return decision;
 			};
 			const result = (outcome: HarnessRun<TCandidate, TAssessment, TChallenge>["outcome"]) =>
@@ -169,14 +180,14 @@ export function defineTrainingHarness<TCandidate, TAssessment, TFeedback>(
 					context: await provide(await bus.read()),
 					...(input.signal === undefined ? {} : { signal: input.signal }),
 				});
-				const candidate = await dispatch("student", "agent.propose", { round, task: input.task, rubric, feedback },
+				const candidate = await dispatch(actors.student, kinds.propose, { round, task: input.task, rubric, feedback },
 					() => input.student(turn));
 				input.signal?.throwIfAborted();
 				const candidateId: string = candidateKey.parse(identify(candidate));
 				if (candidateId === previousCandidate) return result("stalled");
 				previousCandidate = candidateId;
 
-				const assessment = await dispatch("teacher", "agent.assess", { round, candidateId },
+				const assessment = await dispatch(actors.teacher, kinds.assess, { round, candidateId },
 					() => input.teacher(candidate, turn));
 				input.signal?.throwIfAborted();
 				const candidateDecision = await decide({ candidateId }, {
@@ -201,10 +212,10 @@ export function defineTrainingHarness<TCandidate, TAssessment, TFeedback>(
 					return result("accepted");
 				}
 
-				const challenge = await dispatch("adversary", "agent.challenge", { candidateId }, async () =>
+				const challenge = await dispatch(actors.adversary, kinds.challenge, { candidateId }, async () =>
 					adversary(candidate, {
 						task: input.task,
-						context: await provide(await bus.read("adversary")),
+						context: await provide(await bus.read(actors.adversary)),
 						...(input.signal === undefined ? {} : { signal: input.signal }),
 					}));
 				input.signal?.throwIfAborted();
@@ -224,7 +235,7 @@ export function defineTrainingHarness<TCandidate, TAssessment, TFeedback>(
 					return result("accepted");
 				}
 
-				const revision = await dispatch("teacher", "agent.revise-rubric", { round, candidateId }, async () =>
+				const revision = await dispatch(actors.teacher, kinds.reviseRubric, { round, candidateId }, async () =>
 					revise(challenge, {
 						task: input.task,
 						candidate,

packages/harness/src/index.ts

@@ -1,11 +1,11 @@
 export { WriteAheadAgentBus } from "./bus.js";
-export type { AgentBusAccess, AgentBusSettings } from "./bus.js";
+export type { AgentBusAccess, AgentBusSettings, AgentWriter } from "./bus.js";
 
-export { AgentActionDeniedError, decisionKind, dispatchAction } from "./dispatch.js";
-export type { ActionGate } from "./dispatch.js";
+export { AgentActionDeniedError, dispatchAction } from "./dispatch.js";
+export type { ActionGate, JudgeDecision } from "./dispatch.js";
 
-export { agentBusEntry, agentMessage, judgeDecision } from "./schema.js";
-export type { AbsolutePath, AgentBusEntry, AgentMessage, JudgeDecision } from "./schema.js";
+export { agentBusEntry, agentMessage } from "./schema.js";
+export type { AbsolutePath, AgentBusEntry, AgentMessage } from "./schema.js";
 
 export { defaultMaxRounds, defineTrainingHarness } from "./harness.js";
 export type {

packages/harness/src/schema.ts

@@ -33,9 +33,6 @@ export const agentBusEntry = agentMessage.extend({
 });
 export type AgentBusEntry = z.output<typeof agentBusEntry>;
 
-export const judgeDecision = z.enum(["pass", "fail"], { error: "judge must return exactly pass or fail" });
-export type JudgeDecision = z.output<typeof judgeDecision>;
-
 export const absolutePath = z.string()
 	.refine(isAbsolute, "path must be absolute")
 	.transform((path) => resolve(path))

packages/harness/test/harness.test.ts

@@ -147,6 +147,18 @@ describe("training harness", () => {
 		expect((await failed.read()).map(({ kind }) => kind)).toEqual(["test.failure", "test.failure.failed"]);
 	});
 
+	it("binds an agent writer so entries need only a kind and payload", async () => {
+		const bus = new WriteAheadAgentBus();
+		const student = bus.agent("student");
+		await student("test.note", { value: 1 });
+		await student("test.done");
+
+		expect((await bus.read("student")).map(({ actor, kind, payload }) => ({ actor, kind, payload }))).toEqual([
+			{ actor: "student", kind: "test.note", payload: { value: 1 } },
+			{ actor: "student", kind: "test.done", payload: undefined },
+		]);
+	});
+
 	it("refuses appends and reads the access hook denies", async () => {
 		const bus = new WriteAheadAgentBus({
 			allow: (access) => access.operation === "append" ? access.actor !== "intruder" : access.actor === undefined,