diff --git a/packages/harness/README.md b/packages/harness/README.md index 14d21a5..4007c99 100644 --- a/packages/harness/README.md +++ b/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 diff --git a/packages/harness/src/bus.ts b/packages/harness/src/bus.ts index 8a1166e..8935d3e 100644 --- a/packages/harness/src/bus.ts +++ b/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; + /** 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 { const parsed = agentMessage.parse(message); diff --git a/packages/harness/src/dispatch.ts b/packages/harness/src/dispatch.ts index 5f4efef..6a7a3f6 100644 --- a/packages/harness/src/dispatch.ts +++ b/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 { + 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,21 +51,19 @@ export async function dispatchAction( gate: ActionGate | undefined, execute: () => Promise | T, ): Promise { - 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; @@ -58,13 +71,13 @@ export async function dispatchAction( 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; } diff --git a/packages/harness/src/harness.ts b/packages/harness/src/harness.ts index e2a7362..480b16d 100644 --- a/packages/harness/src/harness.ts +++ b/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( ): Promise => { 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["outcome"]) => @@ -169,14 +180,14 @@ export function defineTrainingHarness( 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( 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( 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, diff --git a/packages/harness/src/index.ts b/packages/harness/src/index.ts index fbb71f1..a722ff2 100644 --- a/packages/harness/src/index.ts +++ b/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 { diff --git a/packages/harness/src/schema.ts b/packages/harness/src/schema.ts index 5a2db14..a36e321 100644 --- a/packages/harness/src/schema.ts +++ b/packages/harness/src/schema.ts @@ -33,9 +33,6 @@ export const agentBusEntry = agentMessage.extend({ }); export type AgentBusEntry = z.output; -export const judgeDecision = z.enum(["pass", "fail"], { error: "judge must return exactly pass or fail" }); -export type JudgeDecision = z.output; - export const absolutePath = z.string() .refine(isAbsolute, "path must be absolute") .transform((path) => resolve(path)) diff --git a/packages/harness/test/harness.test.ts b/packages/harness/test/harness.test.ts index 69ccc67..1707a67 100644 --- a/packages/harness/test/harness.test.ts +++ b/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,