diff --git a/apps/brunch-agent/evals/petrinaut/README.md b/apps/brunch-agent/evals/petrinaut/README.md index e30004163dc..11933ccf4af 100644 --- a/apps/brunch-agent/evals/petrinaut/README.md +++ b/apps/brunch-agent/evals/petrinaut/README.md @@ -45,6 +45,22 @@ yarn workspace @apps/brunch-agent probe:turn pharma-cold-chain openai/gpt-6.1-so The arguments are the case, the `provider/id` model and the thinking level; they default to `pharma-cold-chain` and Brunch's default model and thinking level. The provider's API key must be in the environment. Each probe spends one live model turn and requires the same model and spend authorization as a persona run. Output stays under `apps/brunch-agent/.data-wipe-me/turn-probes/` and is not committed. A probe is an observation of one turn, not a quality grade. +## Run packets + +Every persona run and probe ends by writing `evidence/packet.json` and `evidence/packet.md` into its directory. The packet accounts for each exchange's time, which is what a change to turn time is judged by: + +- active time (wall time less approval waits) and time until the first reply text; +- model steps, each split into time before first output (hidden reasoning), tool-argument streaming and reply text, and tool phases; +- each step's shape (its tool categories, in order); +- the tool mix, skill activations, Ledger commit sizes, net growth, tokens and cost; +- what it could not account for: exchanges that never settled (left out of the time sums), steps cut off mid-message or mid-tools, and stored records it could not read. + +[`run-packet.ts`](./run-packet.ts) reads `conversation.db` and, when present, the server log's `flue.submission chronology` lines, `run.json` and the evidence files. Without the server log, which the probe does not keep, argument-streaming times are unknown. To write or refresh packets for retained runs, which spends no model turns: + +```sh +yarn workspace @apps/brunch-agent packet:run .data-wipe-me/persona-runs/run-XXXXXX [...] +``` + ## Scoring Each maintained case also keeps a `key.md` for scoring a run by hand: what the person believes, where the pack puts that under pressure, and what Brunch must tell apart, respect or keep conditional. Neither runner loads it. There are no graders or campaign protocols. A persona run or a probe is an observation, not an automatic quality grade or acceptance decision. diff --git a/apps/brunch-agent/evals/petrinaut/persona/README.md b/apps/brunch-agent/evals/petrinaut/persona/README.md index eaa1455ef0c..f952a5983f2 100644 --- a/apps/brunch-agent/evals/petrinaut/persona/README.md +++ b/apps/brunch-agent/evals/petrinaut/persona/README.md @@ -50,7 +50,7 @@ The helper speaks a Pi-style JSONL protocol to the bridge. Commands are `{ "id"? The persona stops by its own rule: when the person has what they came for, when the conversation is going nowhere, or when they run out of patience. It then runs `persona end` and tells you why it stopped. You, or an agent supervising the run, can force a stop at any point, for example after a chosen number of turns, with `/bin/persona end`, or with Ctrl-C in the launcher when the agent runs in its own pane. -**Persona runs have no automatic accounting cutoff.** The launcher disables the campaign accounting wrapper for Brunch even if `BRUNCH_STEP_A_ACCOUNTING` was inherited. There is no built-in turn-count limit. A live-provider run requires explicit model and spend authorization; the existence of this command grants none. +**Persona runs have no automatic accounting cutoff.** The launcher disables the campaign accounting wrapper for Brunch even if `BRUNCH_STEP_A_ACCOUNTING` was inherited. `--max-messages ` stops the run once Brunch has replied to the persona's nth message; the case's opening is not counted, and the run packet is written as on any other stop. A live-provider run requires explicit model and spend authorization; the existence of this command grants none. ## Guidance provenance @@ -123,6 +123,7 @@ Each launch prints its directory under `apps/brunch-agent/.data-wipe-me/persona- - `persona-brief.md`, `resume-brief.md` and `bin/persona`: the agent's private brief, latest resume notice and bridge helper. - `bridge-log.jsonl`: every utterance Brunch admitted, with the persona agent settings that wrote each persona utterance (empty when you started the agent yourself), used to reconcile resume; and every destructive edit the launcher allowed, with the items its approval panel listed. - `evidence/`: canonical snapshot and derived transcript, tool trace, compiled `ledger.md` with its `ledger.json` commits, and bound `net.json`; refreshed after completed turns and on shutdown. +- `evidence/packet.json` and `packet.md`: the run packet, written on shutdown (see [Run packets](../README.md#run-packets)). - Service logs, only for services this launch started. The persistent Chrome profile lives outside the checkout; its path is in `run.json`. Older runs may also contain `pi/`, `persona-input.md`, `usage-ledger.json` or `attempt-ledger.md`; leave them untouched as historical evidence. diff --git a/apps/brunch-agent/evals/petrinaut/persona/launch.ts b/apps/brunch-agent/evals/petrinaut/persona/launch.ts index 4c5343dc7fb..d42172396d5 100644 --- a/apps/brunch-agent/evals/petrinaut/persona/launch.ts +++ b/apps/brunch-agent/evals/petrinaut/persona/launch.ts @@ -32,6 +32,7 @@ import { DEFAULT_CHAT_THINKING, } from "../../../src/shared/chat-model.ts"; import { agentOwnershipHeaders } from "../../../src/shared/conversation/identity.ts"; +import { writeRunPacket } from "../run-packet.ts"; import { openPersonaBrowserBridge } from "./browser-bridge.ts"; import { PersonaBrowserTurnError, @@ -286,6 +287,8 @@ type PersonaLaunchOptions = { readonly agent?: PersonaAgentSettings; /** Wait for Enter before the first message; only possible on a TTY. */ readonly recordingPause?: boolean; + /** Stop the run once Brunch has replied to this many persona messages. */ + readonly maxMessages?: number; }; /** @@ -302,6 +305,7 @@ const launchPersona = async ({ roles = resolvePersonaRoleSettings(), agent, recordingPause: pauseForRecording = true, + maxMessages, }: PersonaLaunchOptions) => { const manifest = resume ? await verifyGuidanceResume(resume.config.guidance) @@ -385,25 +389,46 @@ const launchPersona = async ({ report(`Run: ${run}`); const stop = new AbortController(); const started: ChildProcess[] = []; - let servicesStopRequested = false; + /** Whether the signal reached the process group; false once it has gone. */ + const signalGroup = (pid: number, signal: NodeJS.Signals | 0) => { + try { + process.kill(-pid, signal); + return true; + } catch (error) { + if ( + !(error instanceof Error) || + !("code" in error) || + error.code !== "ESRCH" + ) { + throw error; + } + return false; + } + }; + let servicesStopped: Promise | undefined; + /** Ends the owned process groups and settles once every one has exited. */ const stopStartedServices = () => { - if (servicesStopRequested) return; - servicesStopRequested = true; - for (const child of started) { - if (!child.pid || child.exitCode !== null || child.signalCode !== null) - continue; - try { - process.kill(-child.pid, "SIGTERM"); - } catch (error) { - if ( - !(error instanceof Error) || - !("code" in error) || - error.code !== "ESRCH" - ) { - throw error; + servicesStopped ??= (async () => { + const groups = started.flatMap((child) => + child.pid !== undefined && + child.exitCode === null && + child.signalCode === null && + signalGroup(child.pid, "SIGTERM") + ? [child.pid] + : [], + ); + // Services flush their logs and the conversation store on SIGTERM; + // one that has not exited by then is killed so the launcher can finish. + const deadline = Date.now() + 10_000; + while (groups.some((pid) => signalGroup(pid, 0))) { + if (Date.now() > deadline) { + for (const pid of groups) signalGroup(pid, "SIGKILL"); + return; } + await delay(100); } - } + })(); + return servicesStopped; }; let browser: | Awaited> @@ -427,6 +452,7 @@ const launchPersona = async ({ }; process.on("SIGINT", onInterruptKey); process.on("SIGTERM", interrupt); + process.on("SIGHUP", interrupt); try { const preflight = await execute( process.execPath, @@ -505,6 +531,11 @@ const launchPersona = async ({ executablePath: chromeExecutable, headless: false, viewport: null, + // The launcher's own handlers stop the run and close the browser; Playwright's + // would kill it and exit before cleanup retains the run. + handleSIGINT: false, + handleSIGTERM: false, + handleSIGHUP: false, args: [ "--remote-debugging-address=127.0.0.1", "--remote-debugging-port=0", @@ -619,6 +650,7 @@ const launchPersona = async ({ url: opened.session.url, headers: agentOwnershipHeaders(opened.session), }); + let repliedMessages = 0; bridge = await openPersonaBrowserBridge({ prompt: async (message, turn) => { await verifyGuidanceResume(manifest); @@ -656,6 +688,13 @@ const launchPersona = async ({ documentId, join(run, "evidence"), ); + repliedMessages += 1; + if (maxMessages !== undefined && repliedMessages >= maxMessages) + // After the bridge has delivered this reply. + setImmediate(() => { + report(`Message limit reached: ${maxMessages} persona messages.`); + interrupt(); + }); return { text: result.reply.text, submissionIds: result.submissionIds, @@ -760,8 +799,9 @@ const launchPersona = async ({ }); report(`Launcher Stop: ${stopDisposition.settlement}.`); // Settlement is durable now; stop services before slower browser cleanup - // so terminal Ctrl-C cannot orphan the launcher's process groups. - stopStartedServices(); + // so terminal Ctrl-C cannot orphan the launcher's process groups. The + // packet step below awaits the same stop and reports any failure. + stopStartedServices().catch(() => undefined); } await bridge?.close(); if (agentProcess?.kind === "herdr") { @@ -787,7 +827,17 @@ const launchPersona = async ({ await browser?.close(); } } finally { - stopStartedServices(); + await stopStartedServices(); + // After the services stop, the server log and conversation store are final. + await writeRunPacket(run) + .then(() => refreshProofManifest(join(run, "evidence"))) + .then( + () => report(`Run packet: ${join(run, "evidence", "packet.md")}`), + (error: unknown) => + process.stderr.write( + `Could not write the run packet (retry with \`yarn workspace @apps/brunch-agent packet:run ${run}\`): ${String(error)}\n`, + ), + ); report(`Retained run: ${run}`); } if (agentProcess?.kind === "foreground" && !lifecycle.agentExited) { @@ -797,6 +847,7 @@ const launchPersona = async ({ } finally { process.removeListener("SIGINT", onInterruptKey); process.removeListener("SIGTERM", interrupt); + process.removeListener("SIGHUP", interrupt); } } }; @@ -820,13 +871,14 @@ if ( "persona-model": { type: "string" }, "persona-thinking": { type: "string" }, "skip-recording-pause": { type: "boolean" }, + "max-messages": { type: "string" }, resume: { type: "string" }, help: { type: "boolean", short: "h" }, }, }); if (values.help) { report( - `Usage: yarn brunch:persona --case [--objective ] [--route ] [--initial-net ] [--interview-length quick|standard|thorough|deep] [--brunch-model ] [--brunch-thinking ] [agent options]\nDiscover cases: yarn brunch:persona --list-cases\nDefault: empty net on /; optional --initial-net stages a model and is not a from-scratch run. --interview-length turns on the website's Interview length at that level for the whole run; without it the run has none. --objective is a private aim for the person, phrased in their own terms; without it the person's goal comes from the case. --objective is fresh-run-only and is neither retained nor reapplied on resume. Starts owned services, a fresh headed Chrome window and a browser bridge; on an interactive terminal, pauses for Enter before sending anything (--skip-recording-pause skips it). Defaults: Brunch ${DEFAULT_CHAT_MODEL} ${DEFAULT_CHAT_THINKING}. Native usage is retained; ordinary external-agent runs have no automatic budget cutoff. Requires macOS Chrome, unused ${brunchEnv.chatPort}/${brunchEnv.panelPort}, and Brunch's provider API key. The persona stops by its own rule; Ctrl-C here or /bin/persona end stops the run at any point, stopping owned resources and retaining run data.`, + `Usage: yarn brunch:persona --case [--objective ] [--route ] [--initial-net ] [--interview-length quick|standard|thorough|deep] [--brunch-model ] [--brunch-thinking ] [agent options]\nDiscover cases: yarn brunch:persona --list-cases\nDefault: empty net on /; optional --initial-net stages a model and is not a from-scratch run. --interview-length turns on the website's Interview length at that level for the whole run; without it the run has none. --objective is a private aim for the person, phrased in their own terms; without it the person's goal comes from the case. --objective is fresh-run-only and is neither retained nor reapplied on resume. Starts owned services, a fresh headed Chrome window and a browser bridge; on an interactive terminal, pauses for Enter before sending anything (--skip-recording-pause skips it). --max-messages stops the run once Brunch has replied to the persona's nth message (the case's opening is not counted). Defaults: Brunch ${DEFAULT_CHAT_MODEL} ${DEFAULT_CHAT_THINKING}. Native usage is retained; ordinary external-agent runs have no automatic budget cutoff. Requires macOS Chrome, unused ${brunchEnv.chatPort}/${brunchEnv.panelPort}, and Brunch's provider API key. The persona stops by its own rule; Ctrl-C here or /bin/persona end stops the run at any point, stopping owned resources and retaining run data.`, ); report( `Agent options:\n --agent ${personaAgentPresets.join("|")} Start that agent with the persona brief.\n --agent-command '' Start any other agent; {prompt} becomes the quoted launch prompt.\n --persona-model Passed to the --agent preset's own model flag.\n --persona-thinking Passed to pi's --thinking (only with --agent pi).\nThe agent runs in a Herdr pane when HERDR_ENV=1, otherwise in this terminal. Without an agent option the launcher prints the launch prompt for you to give any agent. The agent talks to Brunch through /bin/persona (say, transcript, state, end, rpc).`, @@ -862,6 +914,13 @@ if ( }) : undefined; const recordingPause = !values["skip-recording-pause"]; + const maxMessages = + values["max-messages"] === undefined + ? undefined + : Number(values["max-messages"]); + const invalidMaxMessages = + maxMessages !== undefined && + !(Number.isInteger(maxMessages) && maxMessages > 0); const resumeRun = async (resumeDirectory: string) => { if ( values.case || @@ -883,6 +942,7 @@ if ( resume: retained, agent: selectedAgent, recordingPause, + maxMessages, }); }; const freshRun = async (caseDirectory: string) => @@ -903,12 +963,17 @@ if ( }), agent: agent() ?? {}, recordingPause, + maxMessages, }); - const task = values.resume - ? resumeRun(values.resume) - : directory - ? freshRun(directory) - : Promise.reject(new Error("Supply --case ")); + const task = invalidMaxMessages + ? Promise.reject( + new Error("--max-messages takes a positive whole number"), + ) + : values.resume + ? resumeRun(values.resume) + : directory + ? freshRun(directory) + : Promise.reject(new Error("Supply --case ")); await task.catch((error: unknown) => { process.stderr.write( `${error instanceof Error ? error.message : "Persona launch failed"}\n`, diff --git a/apps/brunch-agent/evals/petrinaut/run-packet-cli.ts b/apps/brunch-agent/evals/petrinaut/run-packet-cli.ts new file mode 100644 index 00000000000..fbf96bbd4c7 --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet-cli.ts @@ -0,0 +1,30 @@ +/** + * Writes `evidence/packet.json` and `evidence/packet.md` for each given + * persona run or turn-probe directory. Reads retained files only; spends no + * model turns. + * + * yarn workspace @apps/brunch-agent packet:run [ …] + */ +import { resolve } from "node:path"; + +import { writeRunPacket } from "./run-packet.ts"; + +const directories = process.argv + .slice(2) + .map((directory) => resolve(directory)); +if (directories.length === 0) { + process.stderr.write("Usage: packet:run [ …]\n"); + process.exit(2); +} +const written = await Promise.all( + directories.map(async (directory) => ({ + directory, + packet: await writeRunPacket(directory), + })), +); +for (const { directory, packet } of written) { + const { summary } = packet; + process.stdout.write( + `${packet.label}: ${summary.exchanges} exchanges, median ${Math.round((summary.activePerExchangeMs.median ?? 0) / 1000)} s, ${summary.stepsPerExchange?.toFixed(1) ?? "–"} steps per exchange → ${resolve(directory, "evidence", "packet.md")}\n`, + ); +} diff --git a/apps/brunch-agent/evals/petrinaut/run-packet.ts b/apps/brunch-agent/evals/petrinaut/run-packet.ts new file mode 100644 index 00000000000..76646fc8daa --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet.ts @@ -0,0 +1,32 @@ +/** + * A run's analysable packet: per-exchange turn time split into hidden + * reasoning, argument streaming, reply text and tool phases, with the tool + * mix, skill activations, Ledger commit sizes, net growth and cost. Reads a + * persona run or turn-probe directory and writes `evidence/packet.json` and + * `evidence/packet.md` into it. + */ +import { mkdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { + buildRunPacket, + type RunPacket, +} from "./run-packet/build-run-packet.ts"; +import { readRunDirectory } from "./run-packet/read-run-directory.ts"; +import { renderRunPacket } from "./run-packet/render-run-packet.ts"; + +export type { RunPacket } from "./run-packet/build-run-packet.ts"; + +export const writeRunPacket = async (directory: string): Promise => { + const packet = buildRunPacket(await readRunDirectory(directory)); + const evidence = join(directory, "evidence"); + await mkdir(evidence, { recursive: true }); + await Promise.all([ + writeFile( + join(evidence, "packet.json"), + `${JSON.stringify(packet, null, 2)}\n`, + ), + writeFile(join(evidence, "packet.md"), renderRunPacket(packet)), + ]); + return packet; +}; diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/build-run-packet.test.ts b/apps/brunch-agent/evals/petrinaut/run-packet/build-run-packet.test.ts new file mode 100644 index 00000000000..ab917d02191 --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/build-run-packet.test.ts @@ -0,0 +1,620 @@ +import { expect, test } from "vitest"; + +import { petrinautContextualUserMessageBody } from "@hashintel/brunch-agent"; + +import { buildRunPacket, type StreamEvent } from "./build-run-packet.ts"; +import { renderRunPacket } from "./render-run-packet.ts"; + +import type { LoggedSubmissionChronology } from "./parse-chronology-log.ts"; + +const t0 = Date.parse("2026-10-15T10:00:00.000Z"); + +const event = ( + type: string, + offsetMs: number, + payload: Record = {}, + submissionId = "sub-1", +): StreamEvent => ({ + type, + at: t0 + offsetMs, + submissionId, + payload: { + type, + timestamp: new Date(t0 + offsetMs).toISOString(), + ...payload, + }, +}); + +const placeResult = { + brunchBrowserResult: true, + output: { applied: true }, + metadata: { + documentRevision: { before: "a", after: "b" }, + readBack: { + places: [{ id: "p1" }, { id: "p2" }], + transitions: [{ id: "t1", inputArcs: [{}], outputArcs: [{}] }], + }, + }, +}; + +// Step 1 builds and commits; step 2 replies. +const events: StreamEvent[] = [ + event("user_message", 0, { + content: [{ type: "text", text: "We run two\nbays." }], + }), + event("agent_start_run", 100), + event("assistant_message_started", 1_100, { turnId: "turn-1" }), + event("assistant_tool_call", 9_000, { + turnId: "turn-1", + toolCallId: "call-place", + name: "addPlace", + arguments: { name: "Bay" }, + }), + event("assistant_tool_call", 9_000, { + turnId: "turn-1", + toolCallId: "call-commit", + name: "ledger_commit", + arguments: { entities: [] }, + }), + event("assistant_message_completed", 9_100, { + turnId: "turn-1", + usage: { + input: 3, + output: 200, + cacheRead: 1_000, + cacheWrite: 50, + reasoning: 120, + cost: { total: 0.01 }, + }, + }), + event("tool_outcome", 9_400, { + toolCallId: "call-place", + toolName: "addPlace", + isError: false, + durationMs: 250, + output: placeResult, + }), + event("tool_outcome", 9_150, { + toolCallId: "call-commit", + toolName: "ledger_commit", + isError: false, + durationMs: 5, + output: { status: "recorded", ids: ["e1", "e2", "c1"] }, + }), + event("tool_results_committed", 9_500, { turnId: "turn-1" }), + event("assistant_message_started", 10_500, { turnId: "turn-2" }), + event("assistant_reasoning_started", 11_000, { blockId: "r1" }), + event("assistant_reasoning_completed", 13_000, { blockId: "r1" }), + event("assistant_text_started", 15_500, { blockId: "x1" }), + event("assistant_text_completed", 17_500, { blockId: "x1" }), + event("assistant_message_completed", 17_600, { + turnId: "turn-2", + usage: { input: 2, output: 80, cacheRead: 1_200, cacheWrite: 0 }, + }), + event("submission_settled", 17_800, { outcome: "completed" }), +]; + +const chronology: LoggedSubmissionChronology = { + submissionId: "sub-1", + outcome: "completed", + turns: [ + { + turnId: "turn-1", + purpose: "agent", + // With the run's start, so deltas from either origin coincide. + startedAt: new Date(t0 + 100).toISOString(), + timeToFirstEventMs: 1_000, + durationMs: 9_000, + terminal: "turn", + isError: false, + inputTokens: 3, + cacheReadTokens: 1_000, + outputTokens: 200, + requestShape: null, + responseShape: null, + toolCalls: [ + { + toolCallId: "call-place", + toolName: "addPlace", + firstDeltaMs: 6_000, + lastDeltaMs: 6_500, + deltaCount: 10, + argumentChars: 40, + maxGapMs: 100, + lastDeltaToTerminalMs: 0, + }, + { + toolCallId: "call-commit", + toolName: "ledger_commit", + firstDeltaMs: 6_600, + lastDeltaMs: 8_900, + deltaCount: 300, + argumentChars: 3_600, + maxGapMs: 100, + lastDeltaToTerminalMs: 0, + }, + ], + }, + ], +}; + +const input = { + label: "run-test", + metadata: null, + submissions: [ + { submissionId: "sub-1", acceptedAt: t0, settledAt: t0 + 18_000 }, + { submissionId: "sub-2", acceptedAt: t0 + 30_000, settledAt: t0 + 31_000 }, + ], + events: [ + ...events, + event("agent_start_run", 30_000, {}, "sub-2"), + event("assistant_message_started", 30_200, { turnId: "turn-3" }, "sub-2"), + event( + "assistant_tool_call", + 30_400, + { + toolCallId: "call-approve", + name: "applyAutoLayout", + arguments: { askUserFirst: true }, + }, + "sub-2", + ), + event("assistant_message_completed", 30_500, { turnId: "turn-3" }, "sub-2"), + event( + "tool_outcome", + 30_900, + { + toolCallId: "call-approve", + isError: false, + durationMs: 45_000, + }, + "sub-2", + ), + event("tool_results_committed", 30_900, {}, "sub-2"), + ], + chronology: new Map([["sub-1", chronology]]), + skipped: { batches: 0, events: 0, chronologyLines: 0 }, + finalLedger: null, + finalNet: null, +}; + +test("an exchange's wall time splits into hidden reasoning, argument streaming, reply text and tool phases", () => { + const packet = buildRunPacket(input); + const [exchange] = packet.exchanges; + + expect(exchange).toMatchObject({ + index: 1, + userExcerpt: "We run two bays.", + outcome: "completed", + wallMs: 18_000, + untilFirstTextMs: 15_500, + toolPhaseMs: 400, + stepShape: ["ledger+net-change", "reply"], + }); + expect(exchange?.steps).toMatchObject([ + { + durationMs: 9_000, + // The first argument delta, 6 s after the request. + beforeOutputMs: 6_000, + argumentStreamingMs: 2_900, + toolPhaseMs: 400, + }, + { + durationMs: 8_100, + beforeOutputMs: 6_000, + reasoningSummaryMs: 2_000, + textMs: 2_000, + argumentStreamingMs: null, + toolPhaseMs: 0, + }, + ]); + expect(exchange?.modelMs).toBe(17_100); + expect(exchange?.otherMs).toBe(500); +}); + +test("tool calls are categorised, and Ledger commits counted with their argument size", () => { + const [exchange] = buildRunPacket(input).exchanges; + + expect(exchange?.categories.ledger).toEqual({ + calls: 1, + executionMs: 5, + argumentChars: 3_600, + }); + expect(exchange?.categories["net-change"].calls).toBe(1); + expect(exchange?.ledger).toEqual({ + commits: 1, + recorded: 1, + refused: 0, + recordsAdded: 3, + argumentChars: 3_600, + }); + expect(exchange?.netChanges).toBe(1); + expect(exchange?.netAfter).toEqual({ places: 2, transitions: 1, arcs: 2 }); + expect(exchange?.usage).toMatchObject({ + outputTokens: 280, + cacheWriteTokens: 50, + costUsd: 0.01, + }); +}); + +test("an approval wait is kept out of Brunch's active time", () => { + const packet = buildRunPacket(input); + const second = packet.exchanges[1]; + + expect(second).toMatchObject({ + thinkTimeBeforeMs: 12_000, + longToolCalls: [{ name: "applyAutoLayout", executionMs: 45_000 }], + approvalWaitMs: 400, + activeMs: 600, + // Net size carries over from the exchange that last showed it. + netAfter: { places: 2, transitions: 1, arcs: 2 }, + }); + expect(packet.summary).toMatchObject({ + exchanges: 2, + wallMs: 19_000, + activeMs: 18_600, + steps: 3, + stepsByToolCount: { none: 1, one: 1, several: 1 }, + ledgerCommits: 1, + longToolCalls: 1, + }); +}); + +test("an approval stopped before its tool returned is kept out of active time", () => { + const packet = buildRunPacket({ + ...input, + submissions: [ + { submissionId: "stop", acceptedAt: t0, settledAt: t0 + 60_000 }, + ], + events: [ + event("agent_start_run", 0, {}, "stop"), + event("assistant_message_started", 200, { turnId: "turn-a" }, "stop"), + event( + "assistant_tool_call", + 400, + { toolCallId: "call-remove", name: "removeElements", arguments: {} }, + "stop", + ), + event("assistant_message_completed", 1_000, { turnId: "turn-a" }, "stop"), + event("submission_settled", 60_000, { outcome: "aborted" }, "stop"), + ], + chronology: new Map(), + }); + const [exchange] = packet.exchanges; + + expect(exchange?.steps).toMatchObject([ + { durationMs: 1_000, toolPhaseMs: 59_000, cutOffIn: "tools" }, + ]); + expect(exchange).toMatchObject({ + outcome: "aborted", + wallMs: 60_000, + approvalWaitMs: 59_000, + activeMs: 1_000, + }); + expect(packet.summary.incomplete).toEqual({ + unsettledExchanges: 0, + cutOffMessages: 0, + cutOffToolPhases: 1, + }); +}); + +test("a message cut off mid-stream keeps its model time up to settlement", () => { + const packet = buildRunPacket({ + ...input, + submissions: [ + { submissionId: "stop", acceptedAt: t0, settledAt: t0 + 5_000 }, + ], + events: [ + event("agent_start_run", 0, {}, "stop"), + event("assistant_message_started", 100, { turnId: "turn-a" }, "stop"), + event("assistant_text_started", 2_000, { blockId: "x1" }, "stop"), + event("submission_settled", 5_000, { outcome: "aborted" }, "stop"), + ], + chronology: new Map(), + }); + const [exchange] = packet.exchanges; + + expect(exchange?.steps).toMatchObject([ + { durationMs: 5_000, toolPhaseMs: 0, cutOffIn: "message" }, + ]); + expect(exchange).toMatchObject({ modelMs: 5_000, otherMs: 0 }); + expect(packet.summary.incomplete.cutOffMessages).toBe(1); +}); + +test("a step after a cut-off one starts where the cut-off one ends, so no span counts twice", () => { + const packet = buildRunPacket({ + ...input, + submissions: [ + { submissionId: "gap", acceptedAt: t0, settledAt: t0 + 10_000 }, + ], + events: [ + event("agent_start_run", 0, {}, "gap"), + // Cut off in its tools: no `tool_results_committed` follows. + event("assistant_message_started", 200, { turnId: "turn-a" }, "gap"), + event( + "assistant_tool_call", + 400, + { toolCallId: "call-a", name: "addPlace", arguments: {} }, + "gap", + ), + event("assistant_message_completed", 1_000, { turnId: "turn-a" }, "gap"), + // Cut off mid-message: no `assistant_message_completed` follows. + event("assistant_message_started", 3_000, { turnId: "turn-b" }, "gap"), + event("assistant_message_started", 6_000, { turnId: "turn-c" }, "gap"), + event("assistant_message_completed", 9_000, { turnId: "turn-c" }, "gap"), + event("submission_settled", 10_000, { outcome: "completed" }, "gap"), + ], + chronology: new Map(), + }); + const [exchange] = packet.exchanges; + + expect(exchange?.steps).toMatchObject([ + { durationMs: 1_000, toolPhaseMs: 2_000, cutOffIn: "tools" }, + { durationMs: 3_000, toolPhaseMs: 0, cutOffIn: "message" }, + { durationMs: 3_000, toolPhaseMs: 0, cutOffIn: null }, + ]); + expect(exchange).toMatchObject({ + wallMs: 10_000, + modelMs: 7_000, + toolPhaseMs: 2_000, + otherMs: 1_000, + }); +}); + +test("an unsettled exchange is counted but left out of the time sums, and the packet says so", () => { + const complete = buildRunPacket(input); + const packet = buildRunPacket({ + ...input, + submissions: [ + ...input.submissions, + { submissionId: "sub-3", acceptedAt: t0 + 40_000, settledAt: null }, + ], + events: [ + ...input.events, + event("agent_start_run", 40_000, {}, "sub-3"), + event("assistant_message_started", 40_100, { turnId: "turn-4" }, "sub-3"), + event( + "assistant_message_completed", + 45_000, + { turnId: "turn-4", usage: { output: 10 } }, + "sub-3", + ), + ], + skipped: { batches: 1, events: 0, chronologyLines: 0 }, + }); + + expect(packet.version).toBe(2); + expect(packet.sources.skipped).toEqual({ + batches: 1, + events: 0, + chronologyLines: 0, + }); + expect(packet.summary).toMatchObject({ + exchanges: 3, + incomplete: { unsettledExchanges: 1 }, + steps: 4, + wallMs: complete.summary.wallMs, + activeMs: complete.summary.activeMs, + modelMs: complete.summary.modelMs, + usage: { outputTokens: complete.summary.usage.outputTokens + 10 }, + }); + + const markdown = renderRunPacket(packet); + expect(markdown).toContain( + "- Incomplete run: 1 exchange never settled, left out of the time sums; 1 stream batch unreadable, skipped.", + ); + expect(markdown).toContain("| 3 (unsettled) |"); +}); + +test("without a chronology, argument streaming is unknown rather than zero", () => { + const packet = buildRunPacket({ ...input, chronology: new Map() }); + + expect(packet.sources.chronology).toBe(false); + expect(packet.summary.argumentStreamingMs).toBeNull(); + expect(packet.exchanges[0]?.steps[0]?.argumentStreamingMs).toBeNull(); + // The db's tool calls arrive with the completed message, so the step's + // output is only known to start at its end. + expect(packet.exchanges[0]?.steps[0]?.beforeOutputMs).toBe(9_000); +}); + +test("a skill activated again is a reload, marked when a compaction came between", () => { + const activation = (offsetMs: number, submissionId: string, id: string) => [ + event("assistant_message_started", offsetMs, {}, submissionId), + event( + "assistant_tool_call", + offsetMs, + { + toolCallId: id, + name: "activate_skill", + arguments: { name: "eliciting" }, + }, + submissionId, + ), + ]; + const packet = buildRunPacket({ + ...input, + submissions: [ + { submissionId: "a", acceptedAt: t0, settledAt: t0 + 10 }, + { submissionId: "b", acceptedAt: t0 + 20, settledAt: t0 + 30 }, + { submissionId: "c", acceptedAt: t0 + 40, settledAt: t0 + 50 }, + ], + events: [ + ...activation(1, "a", "call-a"), + ...activation(21, "b", "call-b"), + event("compaction", 41, { tokensBefore: 180_000 }, "c"), + ...activation(42, "c", "call-c"), + ], + chronology: new Map(), + }); + + expect(packet.exchanges.map((exchange) => exchange.skillReloads)).toEqual([ + [], + [{ name: "eliciting", afterCompaction: false }], + [{ name: "eliciting", afterCompaction: true }], + ]); + expect(packet.exchanges[2]?.compactions).toEqual([{ tokensBefore: 180_000 }]); + expect(packet.summary).toMatchObject({ + skillReloads: 2, + skillReloadsAfterCompaction: 1, + compactions: 1, + }); +}); + +const chronologyTurn = ( + turnId: string, + startedAt: number | undefined, + toolCalls: readonly { + toolCallId: string; + firstDeltaMs: number; + lastDeltaMs: number; + }[], +): LoggedSubmissionChronology["turns"][number] => ({ + turnId, + purpose: "agent", + ...(startedAt === undefined + ? {} + : { startedAt: new Date(t0 + startedAt).toISOString() }), + timeToFirstEventMs: null, + durationMs: 0, + terminal: "turn", + isError: false, + inputTokens: null, + cacheReadTokens: null, + outputTokens: null, + requestShape: null, + responseShape: null, + toolCalls: toolCalls.map((call) => ({ + ...call, + toolName: "addPlace", + deltaCount: 2, + argumentChars: 10, + maxGapMs: 0, + lastDeltaToTerminalMs: 0, + })), +}); + +// Each turn starts after its step's request: 500 ms after the run starts, +// and 500 ms after the first step's tool results are committed. +const offsetEvents: StreamEvent[] = [ + event("agent_start_run", 100, {}, "offset"), + event("assistant_message_started", 1_000, { turnId: "turn-a" }, "offset"), + event("assistant_text_started", 3_000, { blockId: "xa" }, "offset"), + event("assistant_text_completed", 3_100, { blockId: "xa" }, "offset"), + event( + "assistant_tool_call", + 3_200, + { toolCallId: "call-a", name: "addPlace", arguments: {} }, + "offset", + ), + event("assistant_message_completed", 3_200, { turnId: "turn-a" }, "offset"), + event("tool_results_committed", 3_500, {}, "offset"), + event("assistant_message_started", 4_500, { turnId: "turn-b" }, "offset"), + event("assistant_text_started", 7_000, { blockId: "xb" }, "offset"), + event("assistant_text_completed", 7_100, { blockId: "xb" }, "offset"), + event( + "assistant_tool_call", + 7_200, + { toolCallId: "call-b", name: "addPlace", arguments: {} }, + "offset", + ), + event("assistant_message_completed", 7_200, { turnId: "turn-b" }, "offset"), + event("tool_results_committed", 7_300, {}, "offset"), + event("submission_settled", 7_400, { outcome: "completed" }, "offset"), +]; + +const offsetPacket = (withStartedAt: boolean) => + buildRunPacket({ + ...input, + submissions: [ + { submissionId: "offset", acceptedAt: t0, settledAt: t0 + 7_400 }, + ], + events: offsetEvents, + chronology: new Map([ + [ + "offset", + { + submissionId: "offset", + outcome: "completed", + turns: [ + chronologyTurn("turn-a", withStartedAt ? 600 : undefined, [ + { toolCallId: "call-a", firstDeltaMs: 400, lastDeltaMs: 1_400 }, + ]), + chronologyTurn("turn-b", withStartedAt ? 4_000 : undefined, [ + { toolCallId: "call-b", firstDeltaMs: 1_000, lastDeltaMs: 2_000 }, + ]), + ], + }, + ], + ]), + }); + +test("the first step's tool-argument deltas are measured from the run's start, not the turn's", () => { + const [step] = offsetPacket(true).exchanges[0]?.steps ?? []; + + expect(step).toMatchObject({ + // The turn starts 500 ms after the run; its first delta 400 ms later. + beforeOutputMs: 900, + argumentStreamingMs: 1_000, + toolCalls: [{ firstDeltaMs: 900, lastDeltaMs: 1_900 }], + }); +}); + +test("a later step's tool-argument deltas are measured from the committed tool results, not the turn's start", () => { + const step = offsetPacket(true).exchanges[0]?.steps[1]; + + expect(step).toMatchObject({ + // The turn starts 500 ms after the commit; its first delta 1 s later. + beforeOutputMs: 1_500, + argumentStreamingMs: 1_000, + toolCalls: [{ firstDeltaMs: 1_500, lastDeltaMs: 2_500 }], + }); +}); + +test("a chronology logged without turn start times leaves tool deltas unplaced but still measures their streaming", () => { + const steps = offsetPacket(false).exchanges[0]?.steps; + + expect(steps).toMatchObject([ + { + // Reply text only: 2.9 s after the run started. + beforeOutputMs: 2_900, + argumentStreamingMs: 1_000, + toolCalls: [{ firstDeltaMs: null, lastDeltaMs: null }], + }, + { + beforeOutputMs: 3_500, + argumentStreamingMs: 1_000, + toolCalls: [{ firstDeltaMs: null, lastDeltaMs: null }], + }, + ]); +}); + +test("a user message sent with submission context is excerpted as the user's own text", () => { + const packet = buildRunPacket({ + ...input, + submissions: [ + { submissionId: "context", acceptedAt: t0, settledAt: t0 + 100 }, + ], + events: [ + event( + "user_message", + 0, + { + content: [ + { + type: "text", + text: petrinautContextualUserMessageBody({ + userText: "We run two\nbays.", + diagnosticsContext: "", + submissionContext: { interviewLength: "short" }, + }), + }, + ], + }, + "context", + ), + ], + chronology: new Map(), + }); + + expect(packet.exchanges[0]?.userExcerpt).toBe("We run two bays."); +}); diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/build-run-packet.ts b/apps/brunch-agent/evals/petrinaut/run-packet/build-run-packet.ts new file mode 100644 index 00000000000..70a8c871827 --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/build-run-packet.ts @@ -0,0 +1,874 @@ +import { brunchTools, petrinautUserMessageText } from "@hashintel/brunch-agent"; + +import { + type ToolCategory, + toolCategories, + toolCategoryOf, +} from "./tool-categories.ts"; + +import type { LoggedSubmissionChronology } from "./parse-chronology-log.ts"; + +/** One conversation-stream event, with its parsed timestamp. */ +export interface StreamEvent { + readonly type: string; + readonly at: number; + readonly submissionId: string; + readonly payload: Readonly>; +} + +export interface RunSubmission { + readonly submissionId: string; + readonly acceptedAt: number; + readonly settledAt: number | null; +} + +export interface RunMetadata { + readonly caseName: string | null; + readonly brunchModel: string | null; + readonly brunchThinking: string | null; + readonly personaAgent: string | null; + readonly guidanceSha256: string | null; + readonly createdAt: string | null; +} + +export interface NetSize { + readonly places: number; + readonly transitions: number; + readonly arcs: number; +} + +export interface FinalNetSummary extends NetSize { + readonly tokenTypes: number; + readonly colouredPlaces: number; + readonly parameters: number; + readonly differentialEquations: number; + readonly metrics: number; + readonly scenarios: number; +} + +export interface FinalLedgerSummary { + readonly acceptedCommits: number; + readonly entities: number; + readonly claims: number; + readonly obligations: number; +} + +/** Stored records the reader could not use, and so left out of the packet. */ +export interface SkippedRecords { + /** Conversation-stream batches whose data did not parse. */ + readonly batches: number; + /** Events without a type or timestamp. */ + readonly events: number; + /** Server-log chronology lines that did not parse. */ + readonly chronologyLines: number; +} + +export interface RunPacketInput { + readonly label: string; + readonly metadata: RunMetadata | null; + readonly submissions: readonly RunSubmission[]; + readonly events: readonly StreamEvent[]; + /** Server-log chronology by submission; absent when the run kept no server log. */ + readonly chronology: ReadonlyMap; + readonly skipped: SkippedRecords; + readonly finalLedger: FinalLedgerSummary | null; + readonly finalNet: FinalNetSummary | null; +} + +export interface Usage { + readonly inputTokens: number; + readonly outputTokens: number; + readonly cacheReadTokens: number; + readonly cacheWriteTokens: number; + readonly reasoningTokens: number; + readonly costUsd: number; +} + +export interface PacketToolCall { + readonly toolCallId: string; + readonly name: string; + readonly category: ToolCategory; + readonly argumentChars: number; + /** + * From the server-log chronology: milliseconds from the step's request to + * the first and last argument delta. Null without a chronology, or when it + * predates turn start times and so cannot place its deltas. + */ + readonly firstDeltaMs: number | null; + readonly lastDeltaMs: number | null; + readonly executionMs: number | null; + readonly isError: boolean | null; +} + +export interface PacketStep { + readonly turnId: string; + /** + * Request sent (the previous step's tool results committed, or the run + * started) to message completed. A message that never completed ends where + * the next step starts, or else at settlement or the exchange's last event. + */ + readonly durationMs: number; + /** + * Request sent to the first reply text or tool-argument delta: hidden + * reasoning, time to first event and any reasoning summary. Tool-argument + * deltas count only when the chronology places them (see `firstDeltaMs`). + */ + readonly beforeOutputMs: number; + readonly reasoningSummaryMs: number; + readonly textMs: number; + /** First to last tool-argument delta; null without a server-log chronology. */ + readonly argumentStreamingMs: number | null; + /** + * Message completed to tool results committed: execution, browser round + * trips and approval waits. Tools never committed end where the message + * would. + */ + readonly toolPhaseMs: number; + /** Where the step was cut off: before its message completed, before its tool results were committed, or not at all. */ + readonly cutOffIn: "message" | "tools" | null; + readonly toolCalls: readonly PacketToolCall[]; + readonly usage: Usage; +} + +export interface CategoryTotals { + readonly calls: number; + readonly executionMs: number; + readonly argumentChars: number; +} + +export interface PacketExchange { + readonly index: number; + readonly submissionId: string; + readonly outcome: string | null; + readonly userExcerpt: string; + /** Previous settlement to this submission's acceptance: the person's (or persona's) time to answer. */ + readonly thinkTimeBeforeMs: number | null; + readonly wallMs: number | null; + /** Acceptance to the first reply text: what the person waits before reading anything. */ + readonly untilFirstTextMs: number | null; + readonly modelMs: number; + readonly beforeOutputMs: number; + readonly argumentStreamingMs: number | null; + readonly textMs: number; + readonly toolPhaseMs: number; + readonly otherMs: number | null; + readonly steps: readonly PacketStep[]; + readonly categories: Readonly>; + readonly skillActivations: readonly string[]; + /** Activations of a skill already activated earlier in the run, and whether a compaction came between. */ + readonly skillReloads: readonly { name: string; afterCompaction: boolean }[]; + readonly skillResourceReads: readonly string[]; + /** Context compactions during the exchange, with the estimated context size that triggered each. */ + readonly compactions: readonly { tokensBefore: number | null }[]; + readonly ledger: { + readonly commits: number; + readonly recorded: number; + readonly refused: number; + readonly recordsAdded: number; + readonly argumentChars: number; + }; + readonly netChanges: number; + readonly netAfter: NetSize | null; + /** Tool calls that took at least `longToolCallMs`, usually waiting on the person's approval. */ + readonly longToolCalls: readonly { name: string; executionMs: number }[]; + /** + * Tool phases of the steps holding a long tool call, or cut off at least + * `longToolCallMs` into tools that never returned. + */ + readonly approvalWaitMs: number; + /** Wall time less approval waits: what Brunch itself spent. */ + readonly activeMs: number | null; + /** Each step's tool categories (`+`-joined, led by `text` when it also wrote reply text), or `reply` for a step without tools. */ + readonly stepShape: readonly string[]; + readonly usage: Usage; +} + +export interface Distribution { + readonly median: number | null; + readonly p90: number | null; + readonly max: number | null; +} + +export interface RunPacket { + readonly version: 2; + readonly label: string; + readonly metadata: RunMetadata | null; + readonly sources: { + readonly chronology: boolean; + readonly skipped: SkippedRecords; + }; + /** + * Time sums cover settled exchanges only, since an unsettled exchange has no + * wall or active time to share them against. Counts, usage and + * distributions of first-text time cover every exchange. + */ + readonly summary: { + readonly exchanges: number; + readonly incomplete: { + /** Exchanges with no settlement, left out of the time sums. */ + readonly unsettledExchanges: number; + /** Steps whose message never completed. */ + readonly cutOffMessages: number; + /** Steps whose tool results were never committed. */ + readonly cutOffToolPhases: number; + }; + readonly wallMs: number; + readonly approvalWaitMs: number; + /** Wall time less approval waits; shares and distributions use it. */ + readonly activeMs: number; + readonly activePerExchangeMs: Distribution; + readonly untilFirstTextMs: Distribution; + readonly steps: number; + readonly stepsPerExchange: number | null; + readonly stepsByToolCount: { + readonly none: number; + readonly one: number; + readonly several: number; + }; + readonly modelMs: number; + readonly beforeOutputMs: number; + readonly argumentStreamingMs: number | null; + readonly textMs: number; + readonly toolPhaseMs: number; + readonly categories: Readonly>; + readonly skillActivations: Readonly>; + readonly skillReloads: number; + readonly skillReloadsAfterCompaction: number; + readonly skillResourceReads: number; + readonly compactions: number; + readonly ledgerCommits: number; + readonly ledgerRefused: number; + readonly ledgerArgumentChars: number; + readonly longToolCalls: number; + readonly usage: Usage; + }; + readonly finalLedger: FinalLedgerSummary | null; + readonly finalNet: FinalNetSummary | null; + readonly exchanges: readonly PacketExchange[]; +} + +const longToolCallMs = 30_000; + +const zeroUsage: Usage = { + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheWriteTokens: 0, + reasoningTokens: 0, + costUsd: 0, +}; + +const addUsage = (left: Usage, right: Usage): Usage => ({ + inputTokens: left.inputTokens + right.inputTokens, + outputTokens: left.outputTokens + right.outputTokens, + cacheReadTokens: left.cacheReadTokens + right.cacheReadTokens, + cacheWriteTokens: left.cacheWriteTokens + right.cacheWriteTokens, + reasoningTokens: left.reasoningTokens + right.reasoningTokens, + costUsd: left.costUsd + right.costUsd, +}); + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const numberField = (record: unknown, key: string): number => { + if (!isRecord(record)) return 0; + const value = record[key]; + return typeof value === "number" ? value : 0; +}; + +const stringField = (record: unknown, key: string): string | undefined => { + if (!isRecord(record)) return undefined; + const value = record[key]; + return typeof value === "string" ? value : undefined; +}; + +const usageOf = (payload: Readonly>): Usage => { + const usage = payload.usage; + return { + inputTokens: numberField(usage, "input"), + outputTokens: numberField(usage, "output"), + cacheReadTokens: numberField(usage, "cacheRead"), + cacheWriteTokens: numberField(usage, "cacheWrite"), + reasoningTokens: numberField(usage, "reasoning"), + costUsd: isRecord(usage) ? numberField(usage.cost, "total") : 0, + }; +}; + +const emptyCategories = (): Record => + Object.fromEntries( + toolCategories.map((category) => [ + category, + { calls: 0, executionMs: 0, argumentChars: 0 }, + ]), + ) as Record; + +const addCategories = ( + left: Readonly>, + right: Readonly>, +): Record => + Object.fromEntries( + toolCategories.map((category) => [ + category, + { + calls: left[category].calls + right[category].calls, + executionMs: left[category].executionMs + right[category].executionMs, + argumentChars: + left[category].argumentChars + right[category].argumentChars, + }, + ]), + ) as Record; + +/** Net size from a browser result's read-back or definition, when it carries one. */ +export const netSizeOf = (output: unknown): NetSize | null => { + if (!isRecord(output)) return null; + const metadata = output.metadata; + const inner = output.output; + const definition = + (isRecord(metadata) ? metadata.readBack : undefined) ?? + (isRecord(inner) ? inner.definition : undefined); + if (!isRecord(definition) || !Array.isArray(definition.places)) return null; + const transitions = Array.isArray(definition.transitions) + ? definition.transitions + : []; + const arcs = transitions.reduce((total, transition) => { + if (!isRecord(transition)) return total; + const inputs = Array.isArray(transition.inputArcs) + ? transition.inputArcs.length + : 0; + const outputs = Array.isArray(transition.outputArcs) + ? transition.outputArcs.length + : 0; + return total + inputs + outputs; + }, 0); + return { + places: definition.places.length, + transitions: transitions.length, + arcs, + }; +}; + +const changedDocument = (output: unknown): boolean => { + if (!isRecord(output) || !isRecord(output.metadata)) return false; + const revision = output.metadata.documentRevision; + if (!isRecord(revision)) return false; + return ( + typeof revision.after === "string" && revision.after !== revision.before + ); +}; + +const skillResourcePath = (path: string): string => { + let decoded = path; + try { + decoded = decodeURIComponent(path); + } catch { + // Keep the raw path. + } + const match = /\/packaged-skills\/skill:([^/:]+):[^/]+\/(.+)$/u.exec(decoded); + return match?.[1] !== undefined && match[2] !== undefined + ? `${match[1]}/${match[2]}` + : path; +}; + +const userExcerptOf = (payload: Readonly>): string => { + const content = payload.content; + if (!Array.isArray(content)) return ""; + const text = petrinautUserMessageText( + content.map((part) => stringField(part, "text") ?? "").join(" "), + ) + .replace(/\s+/gu, " ") + .trim(); + return text.length > 160 ? `${text.slice(0, 157)}…` : text; +}; + +const distributionOf = (values: readonly number[]): Distribution => { + if (values.length === 0) return { median: null, p90: null, max: null }; + const sorted = [...values].sort((left, right) => left - right); + const rank = (fraction: number) => + sorted[ + Math.min(sorted.length - 1, Math.ceil(fraction * sorted.length) - 1) + ] ?? null; + return { median: rank(0.5), p90: rank(0.9), max: sorted.at(-1) ?? null }; +}; + +type TimelineEntry = + | { readonly at: number; readonly kind: "activation"; readonly name: string } + | { readonly at: number; readonly kind: "compaction" }; + +interface MutableToolCall { + toolCallId: string; + name: string; + argumentChars: number; + arguments: unknown; + at: number; + firstDeltaMs: number | null; + lastDeltaMs: number | null; + executionMs: number | null; + isError: boolean | null; + output: unknown; +} + +interface MutableStep { + turnId: string; + requestStartAt: number; + startedAt: number; + completedAt: number | null; + toolsCommittedAt: number | null; + firstTextAt: number | null; + reasoningSummaryMs: number; + textMs: number; + openBlocks: Map; + toolCalls: MutableToolCall[]; + usage: Usage; +} + +const buildExchange = ( + index: number, + submission: RunSubmission, + previous: RunSubmission | undefined, + events: readonly StreamEvent[], + chronology: LoggedSubmissionChronology | undefined, + netBefore: NetSize | null, +): { + exchange: Omit; + timeline: TimelineEntry[]; +} => { + const steps: MutableStep[] = []; + let boundary = submission.acceptedAt; + let settledAt = submission.settledAt; + let outcome: string | null = null; + let userExcerpt = ""; + let firstTextAt: number | null = null; + const callsById = new Map(); + const compactions: { tokensBefore: number | null }[] = []; + const timeline: TimelineEntry[] = []; + + for (const event of events) { + const { payload } = event; + const current = steps.at(-1); + const blockId = stringField(payload, "blockId") ?? ""; + switch (event.type) { + case "user_message": + if (userExcerpt === "") userExcerpt = userExcerptOf(payload); + break; + case "agent_start_run": + if (steps.length === 0) boundary = event.at; + break; + case "assistant_message_started": + steps.push({ + turnId: stringField(payload, "turnId") ?? "", + // A previous step that committed no tool results runs until this start. + requestStartAt: + current === undefined || current.toolsCommittedAt !== null + ? boundary + : event.at, + startedAt: event.at, + completedAt: null, + toolsCommittedAt: null, + firstTextAt: null, + reasoningSummaryMs: 0, + textMs: 0, + openBlocks: new Map(), + toolCalls: [], + usage: zeroUsage, + }); + break; + case "assistant_reasoning_started": + case "assistant_text_started": + current?.openBlocks.set(blockId, event.at); + if (event.type === "assistant_text_started" && current) { + current.firstTextAt ??= event.at; + firstTextAt ??= event.at; + } + break; + case "assistant_reasoning_completed": + case "assistant_text_completed": { + const startedAt = current?.openBlocks.get(blockId); + if (current && startedAt !== undefined) { + if (event.type === "assistant_text_completed") + current.textMs += event.at - startedAt; + else current.reasoningSummaryMs += event.at - startedAt; + } + break; + } + case "assistant_tool_call": { + const toolCallId = stringField(payload, "toolCallId") ?? ""; + const call: MutableToolCall = { + toolCallId, + name: stringField(payload, "name") ?? "?", + argumentChars: JSON.stringify(payload.arguments ?? {}).length, + arguments: payload.arguments, + at: event.at, + firstDeltaMs: null, + lastDeltaMs: null, + executionMs: null, + isError: null, + output: undefined, + }; + current?.toolCalls.push(call); + callsById.set(toolCallId, call); + break; + } + case "assistant_message_completed": + if (current) { + current.completedAt = event.at; + current.usage = usageOf(payload); + } + break; + case "tool_outcome": { + const call = callsById.get(stringField(payload, "toolCallId") ?? ""); + if (call) { + const durationMs = payload.durationMs; + call.executionMs = typeof durationMs === "number" ? durationMs : null; + call.isError = payload.isError === true; + call.output = payload.output; + } + break; + } + case "tool_results_committed": + if (current) current.toolsCommittedAt = event.at; + boundary = event.at; + break; + case "compaction": { + const { tokensBefore } = payload; + compactions.push({ + tokensBefore: typeof tokensBefore === "number" ? tokensBefore : null, + }); + timeline.push({ at: event.at, kind: "compaction" }); + break; + } + case "submission_settled": + settledAt ??= event.at; + outcome = stringField(payload, "outcome") ?? null; + break; + } + } + + const chronologyTurns = new Map( + (chronology?.turns ?? []).map((turn) => [turn.turnId, turn]), + ); + const exchangeEndAt = settledAt ?? events.at(-1)?.at ?? submission.acceptedAt; + const packetSteps: PacketStep[] = steps.map((step, stepIndex) => { + const cutOffAt = steps[stepIndex + 1]?.startedAt ?? exchangeEndAt; + const turn = chronologyTurns.get(step.turnId); + const deltas = new Map( + (turn?.toolCalls ?? []).map((call) => [call.toolCallId, call]), + ); + // Chronology offsets count from the turn's start; packet offsets from the step's request. + const turnStartOffsetMs = + turn?.startedAt === undefined + ? null + : Date.parse(turn.startedAt) - step.requestStartAt; + const streamedCalls = step.toolCalls.flatMap((call) => { + const streamed = deltas.get(call.toolCallId); + return streamed === undefined ? [] : [streamed]; + }); + const toolCalls: PacketToolCall[] = step.toolCalls.map((call) => { + const streamed = deltas.get(call.toolCallId); + return { + toolCallId: call.toolCallId, + name: call.name, + category: toolCategoryOf(call.name), + argumentChars: streamed?.argumentChars ?? call.argumentChars, + firstDeltaMs: + streamed === undefined || turnStartOffsetMs === null + ? null + : turnStartOffsetMs + streamed.firstDeltaMs, + lastDeltaMs: + streamed === undefined || turnStartOffsetMs === null + ? null + : turnStartOffsetMs + streamed.lastDeltaMs, + executionMs: call.executionMs, + isError: call.isError, + }; + }); + const completedAt = + step.completedAt ?? Math.max(step.requestStartAt, cutOffAt); + const durationMs = completedAt - step.requestStartAt; + const cutOffIn = + step.completedAt === null + ? ("message" as const) + : step.toolCalls.length > 0 && step.toolsCommittedAt === null + ? ("tools" as const) + : null; + const firstOutputOffsets = [ + ...(step.firstTextAt === null + ? [] + : [step.firstTextAt - step.requestStartAt]), + ...toolCalls.flatMap((call) => + call.firstDeltaMs === null ? [] : [call.firstDeltaMs], + ), + ]; + return { + turnId: step.turnId, + durationMs, + beforeOutputMs: Math.min(durationMs, ...firstOutputOffsets), + reasoningSummaryMs: step.reasoningSummaryMs, + textMs: step.textMs, + argumentStreamingMs: + turn === undefined + ? null + : streamedCalls.length === 0 + ? 0 + : Math.max(...streamedCalls.map((call) => call.lastDeltaMs)) - + Math.min(...streamedCalls.map((call) => call.firstDeltaMs)), + toolPhaseMs: + cutOffIn === "message" || step.toolCalls.length === 0 + ? 0 + : Math.max(0, (step.toolsCommittedAt ?? cutOffAt) - completedAt), + cutOffIn, + toolCalls, + usage: step.usage, + }; + }); + + const categories = emptyCategories(); + const skillActivations: string[] = []; + const skillResourceReads: string[] = []; + const ledger = { + commits: 0, + recorded: 0, + refused: 0, + recordsAdded: 0, + argumentChars: 0, + }; + let netChanges = 0; + let netAfter = netBefore; + const longToolCalls: { name: string; executionMs: number }[] = []; + const argumentCharsById = new Map( + packetSteps + .flatMap((step) => step.toolCalls) + .map((call) => [call.toolCallId, call.argumentChars]), + ); + for (const step of steps) { + for (const call of step.toolCalls) { + const argumentChars = + argumentCharsById.get(call.toolCallId) ?? call.argumentChars; + const category = toolCategoryOf(call.name); + categories[category] = { + calls: categories[category].calls + 1, + executionMs: categories[category].executionMs + (call.executionMs ?? 0), + argumentChars: categories[category].argumentChars + argumentChars, + }; + if (call.name === brunchTools.activateSkill) { + const name = stringField(call.arguments, "name") ?? "?"; + skillActivations.push(name); + timeline.push({ at: call.at, kind: "activation", name }); + } + if (call.name === brunchTools.readSkillResource) { + skillResourceReads.push( + skillResourcePath(stringField(call.arguments, "path") ?? "?"), + ); + } + if (call.name === brunchTools.ledgerCommit) { + ledger.commits += 1; + ledger.argumentChars += argumentChars; + if (stringField(call.output, "status") === "recorded") { + ledger.recorded += 1; + const ids = isRecord(call.output) ? call.output.ids : undefined; + ledger.recordsAdded += Array.isArray(ids) ? ids.length : 0; + } else if (call.isError !== null) { + ledger.refused += 1; + } + } + if (changedDocument(call.output)) netChanges += 1; + netAfter = netSizeOf(call.output) ?? netAfter; + if (call.executionMs !== null && call.executionMs >= longToolCallMs) { + longToolCalls.push({ name: call.name, executionMs: call.executionMs }); + } + } + } + + const sum = (pick: (step: PacketStep) => number) => + packetSteps.reduce((total, step) => total + pick(step), 0); + const modelMs = sum((step) => step.durationMs); + const toolPhaseMs = sum((step) => step.toolPhaseMs); + const approvalWaitMs = sum((step) => + step.toolCalls.some( + (call) => call.executionMs !== null && call.executionMs >= longToolCallMs, + ) || + (step.cutOffIn === "tools" && + step.toolPhaseMs >= longToolCallMs && + step.toolCalls.some((call) => call.executionMs === null)) + ? step.toolPhaseMs + : 0, + ); + const wallMs = settledAt === null ? null : settledAt - submission.acceptedAt; + const exchange = { + index, + submissionId: submission.submissionId, + outcome, + userExcerpt, + thinkTimeBeforeMs: + previous?.settledAt === null || previous === undefined + ? null + : submission.acceptedAt - previous.settledAt, + wallMs, + untilFirstTextMs: + firstTextAt === null ? null : firstTextAt - submission.acceptedAt, + modelMs, + beforeOutputMs: sum((step) => step.beforeOutputMs), + argumentStreamingMs: chronology + ? sum((step) => step.argumentStreamingMs ?? 0) + : null, + textMs: sum((step) => step.textMs), + toolPhaseMs, + otherMs: wallMs === null ? null : wallMs - modelMs - toolPhaseMs, + steps: packetSteps, + categories, + skillActivations, + skillResourceReads, + compactions, + ledger, + netChanges, + netAfter, + longToolCalls, + approvalWaitMs, + activeMs: wallMs === null ? null : wallMs - approvalWaitMs, + stepShape: packetSteps.map((step) => + step.toolCalls.length === 0 + ? "reply" + : [ + ...(step.textMs > 0 ? ["text"] : []), + ...toolCategories.filter((category) => + step.toolCalls.some((call) => call.category === category), + ), + ].join("+"), + ), + usage: packetSteps.reduce( + (total, step) => addUsage(total, step.usage), + zeroUsage, + ), + }; + return { exchange, timeline }; +}; + +/** A run's per-exchange turn-time and tool accounting, from its conversation stream. */ +export const buildRunPacket = (input: RunPacketInput): RunPacket => { + const eventsBySubmission = new Map(); + for (const event of input.events) { + const list = eventsBySubmission.get(event.submissionId) ?? []; + list.push(event); + eventsBySubmission.set(event.submissionId, list); + } + const exchanges: PacketExchange[] = []; + let netBefore: NetSize | null = null; + // Per skill: whether a compaction has come since its last activation. + const skillState = new Map(); + input.submissions.forEach((submission, index) => { + const events = [...(eventsBySubmission.get(submission.submissionId) ?? [])] + // Stable: events sharing a timestamp keep stream order. + .sort((left, right) => left.at - right.at); + const { exchange, timeline } = buildExchange( + index + 1, + submission, + input.submissions[index - 1], + events, + input.chronology.get(submission.submissionId), + netBefore, + ); + const skillReloads: { name: string; afterCompaction: boolean }[] = []; + for (const entry of timeline.sort((left, right) => left.at - right.at)) { + if (entry.kind === "compaction") { + for (const state of skillState.values()) state.compactedSince = true; + continue; + } + const state = skillState.get(entry.name); + if (state) { + skillReloads.push({ + name: entry.name, + afterCompaction: state.compactedSince, + }); + } + skillState.set(entry.name, { compactedSince: false }); + } + netBefore = exchange.netAfter; + exchanges.push({ ...exchange, skillReloads }); + }); + + const steps = exchanges.flatMap((exchange) => exchange.steps); + const categories = exchanges.reduce( + (totals, exchange) => addCategories(totals, exchange.categories), + emptyCategories(), + ); + const skillActivations: Record = {}; + for (const name of exchanges.flatMap((exchange) => exchange.skillActivations)) + skillActivations[name] = (skillActivations[name] ?? 0) + 1; + const total = (pick: (exchange: PacketExchange) => number) => + exchanges.reduce((sum, exchange) => sum + pick(exchange), 0); + const settled = exchanges.flatMap((exchange) => + exchange.wallMs === null || exchange.activeMs === null + ? [] + : [{ ...exchange, wallMs: exchange.wallMs, activeMs: exchange.activeMs }], + ); + const totalSettled = (pick: (exchange: (typeof settled)[number]) => number) => + settled.reduce((sum, exchange) => sum + pick(exchange), 0); + const hasChronology = input.chronology.size > 0; + + return { + version: 2, + label: input.label, + metadata: input.metadata, + sources: { chronology: hasChronology, skipped: input.skipped }, + summary: { + exchanges: exchanges.length, + incomplete: { + unsettledExchanges: exchanges.length - settled.length, + cutOffMessages: steps.filter((step) => step.cutOffIn === "message") + .length, + cutOffToolPhases: steps.filter((step) => step.cutOffIn === "tools") + .length, + }, + wallMs: totalSettled((exchange) => exchange.wallMs), + approvalWaitMs: totalSettled((exchange) => exchange.approvalWaitMs), + activeMs: totalSettled((exchange) => exchange.activeMs), + activePerExchangeMs: distributionOf( + exchanges.flatMap((exchange) => + exchange.activeMs === null ? [] : [exchange.activeMs], + ), + ), + untilFirstTextMs: distributionOf( + exchanges.flatMap((exchange) => + exchange.untilFirstTextMs === null ? [] : [exchange.untilFirstTextMs], + ), + ), + steps: steps.length, + stepsPerExchange: + exchanges.length === 0 ? null : steps.length / exchanges.length, + stepsByToolCount: { + none: steps.filter((step) => step.toolCalls.length === 0).length, + one: steps.filter((step) => step.toolCalls.length === 1).length, + several: steps.filter((step) => step.toolCalls.length > 1).length, + }, + modelMs: totalSettled((exchange) => exchange.modelMs), + beforeOutputMs: totalSettled((exchange) => exchange.beforeOutputMs), + argumentStreamingMs: hasChronology + ? totalSettled((exchange) => exchange.argumentStreamingMs ?? 0) + : null, + textMs: totalSettled((exchange) => exchange.textMs), + toolPhaseMs: totalSettled((exchange) => exchange.toolPhaseMs), + categories, + skillActivations, + skillReloads: total((exchange) => exchange.skillReloads.length), + skillReloadsAfterCompaction: total( + (exchange) => + exchange.skillReloads.filter((reload) => reload.afterCompaction) + .length, + ), + compactions: total((exchange) => exchange.compactions.length), + skillResourceReads: total( + (exchange) => exchange.skillResourceReads.length, + ), + ledgerCommits: total((exchange) => exchange.ledger.commits), + ledgerRefused: total((exchange) => exchange.ledger.refused), + ledgerArgumentChars: total((exchange) => exchange.ledger.argumentChars), + longToolCalls: total((exchange) => exchange.longToolCalls.length), + usage: exchanges.reduce( + (sum, exchange) => addUsage(sum, exchange.usage), + zeroUsage, + ), + }, + finalLedger: input.finalLedger, + finalNet: input.finalNet, + exchanges, + }; +}; diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/parse-chronology-log.test.ts b/apps/brunch-agent/evals/petrinaut/run-packet/parse-chronology-log.test.ts new file mode 100644 index 00000000000..2051c66951a --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/parse-chronology-log.test.ts @@ -0,0 +1,18 @@ +import { expect, test } from "vitest"; + +import { parseChronologyLog } from "./parse-chronology-log.ts"; + +test("chronology entries are read from coloured server-log lines and keyed by submission, counting the lines that don't parse", () => { + const entry = { submissionId: "sub-1", outcome: "completed", turns: [] }; + const log = [ + "\u001B[32minfo\u001B[39m: [brunch] listening", + `\u001B[32minfo\u001B[39m: [brunch] flue.submission chronology ${JSON.stringify(entry)}`, + "\u001B[32minfo\u001B[39m: [brunch] flue.submission chronology {truncated", + ].join("\n"); + + const { entries, skippedLines } = parseChronologyLog(log); + + expect([...entries.keys()]).toEqual(["sub-1"]); + expect(entries.get("sub-1")).toEqual(entry); + expect(skippedLines).toBe(1); +}); diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/parse-chronology-log.ts b/apps/brunch-agent/evals/petrinaut/run-packet/parse-chronology-log.ts new file mode 100644 index 00000000000..5a1f7d213ce --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/parse-chronology-log.ts @@ -0,0 +1,56 @@ +import type { SubmissionChronology } from "../../../src/server/observe-turn-chronology.ts"; + +type LoggedTurnChronology = Omit< + SubmissionChronology["turns"][number], + "startedAt" +> & { + /** Absent from logs written before turns carried their start time. */ + readonly startedAt?: string; +}; + +/** A chronology as read back from a server log, which may predate `startedAt`. */ +export type LoggedSubmissionChronology = Omit & { + readonly turns: readonly LoggedTurnChronology[]; +}; + +const chronologyMarker = "[brunch] flue.submission chronology "; + +/** + * The `flue.submission chronology` entries in a Brunch server log, keyed by + * submission, and how many marked lines could not be read as one. Lines carry + * winston's colour codes before the marker; the JSON after it is the logged + * entry. + */ +export const parseChronologyLog = ( + log: string, +): { + entries: ReadonlyMap; + skippedLines: number; +} => { + const entries = new Map(); + let skippedLines = 0; + for (const line of log.split("\n")) { + const markerAt = line.indexOf(chronologyMarker); + if (markerAt === -1) continue; + let entry: unknown; + try { + entry = JSON.parse(line.slice(markerAt + chronologyMarker.length)); + } catch { + skippedLines += 1; + continue; + } + if ( + typeof entry === "object" && + entry !== null && + "submissionId" in entry && + typeof entry.submissionId === "string" && + "turns" in entry && + Array.isArray(entry.turns) + ) { + entries.set(entry.submissionId, entry as LoggedSubmissionChronology); + } else { + skippedLines += 1; + } + } + return { entries, skippedLines }; +}; diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/read-run-directory.ts b/apps/brunch-agent/evals/petrinaut/run-packet/read-run-directory.ts new file mode 100644 index 00000000000..0f761e2bfe2 --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/read-run-directory.ts @@ -0,0 +1,209 @@ +import { existsSync } from "node:fs"; +import { readFile } from "node:fs/promises"; +import { basename, join } from "node:path"; +import { DatabaseSync } from "node:sqlite"; + +import { + foldCommits, + type LedgerHistoryMessage, +} from "@hashintel/brunch-agent"; + +import { netSizeOf } from "./build-run-packet.ts"; +import { parseChronologyLog } from "./parse-chronology-log.ts"; + +import type { + FinalLedgerSummary, + FinalNetSummary, + RunMetadata, + RunPacketInput, + RunSubmission, + StreamEvent, +} from "./build-run-packet.ts"; + +const isRecord = (value: unknown): value is Record => + typeof value === "object" && value !== null && !Array.isArray(value); + +const stringOrNull = (value: unknown): string | null => + typeof value === "string" ? value : null; + +const readJson = async (path: string): Promise => + existsSync(path) ? JSON.parse(await readFile(path, "utf8")) : undefined; + +/** + * Submissions and stream events from Flue's SQLite store. A batch too large + * for one row is stored across `flue_conversation_stream_batch_chunks`. + */ +const readConversationDb = ( + path: string, +): { + submissions: RunSubmission[]; + events: StreamEvent[]; + skippedBatches: number; + skippedEvents: number; +} => { + const database = new DatabaseSync(path, { readOnly: true }); + try { + const submissions = database + .prepare( + "select submission_id, accepted_at, settled_at from flue_agent_submissions where joined_into is null order by sequence", + ) + .all() + .map((row) => ({ + submissionId: String(row.submission_id), + acceptedAt: Number(row.accepted_at), + settledAt: row.settled_at === null ? null : Number(row.settled_at), + })); + const rows = database + .prepare( + `select batch.submission_id as submission_id, + coalesce( + (select group_concat(chunk.data, '' order by chunk.chunk_index) + from flue_conversation_stream_batch_chunks as chunk + where chunk.path = batch.path and chunk.seq = batch.seq), + batch.data) as data + from flue_conversation_stream_batches as batch + where batch.submission_id is not null + order by batch.path, batch.seq`, + ) + .all(); + const events: StreamEvent[] = []; + let skippedBatches = 0; + let skippedEvents = 0; + for (const row of rows) { + let parsed: unknown; + try { + parsed = JSON.parse(String(row.data)); + } catch { + skippedBatches += 1; + continue; + } + for (const event of Array.isArray(parsed) ? parsed : [parsed]) { + if ( + !isRecord(event) || + typeof event.type !== "string" || + typeof event.timestamp !== "string" + ) { + skippedEvents += 1; + continue; + } + events.push({ + type: event.type, + at: Date.parse(event.timestamp), + submissionId: String(row.submission_id), + payload: event, + }); + } + } + return { submissions, events, skippedBatches, skippedEvents }; + } finally { + database.close(); + } +}; + +const readMetadata = async (directory: string): Promise => { + const run = await readJson(join(directory, "run.json")); + if (!isRecord(run)) return null; + const persona = isRecord(run.personaAgent) ? run.personaAgent : {}; + const personaAgent = [ + persona.agent, + persona.personaModel, + persona.personaThinking, + ] + .filter((part): part is string => typeof part === "string") + .join(" / "); + return { + caseName: + typeof run.caseDirectory === "string" + ? basename(run.caseDirectory) + : null, + brunchModel: stringOrNull(run.brunchModel), + brunchThinking: stringOrNull(run.brunchThinking), + personaAgent: personaAgent === "" ? null : personaAgent, + guidanceSha256: isRecord(run.guidance) + ? stringOrNull(run.guidance.sha256) + : null, + createdAt: stringOrNull(run.createdAt), + }; +}; + +const finalNetOf = (net: unknown): FinalNetSummary | null => { + if (!isRecord(net) || !isRecord(net.sdcpn)) return null; + const sdcpn = net.sdcpn; + const size = netSizeOf({ output: { definition: sdcpn } }); + if (size === null) return null; + const count = (key: string) => { + const value = sdcpn[key]; + return Array.isArray(value) ? value.length : 0; + }; + const places = Array.isArray(sdcpn.places) ? sdcpn.places : []; + return { + ...size, + tokenTypes: count("types"), + colouredPlaces: places.filter( + (place) => isRecord(place) && typeof place.colorId === "string", + ).length, + parameters: count("parameters"), + differentialEquations: count("differentialEquations"), + metrics: count("metrics"), + scenarios: count("scenarios"), + }; +}; + +const isHistoryMessage = (message: unknown): message is LedgerHistoryMessage => + isRecord(message) && + typeof message.id === "string" && + typeof message.role === "string" && + Array.isArray(message.parts); + +const finalLedgerOf = (history: unknown): FinalLedgerSummary | null => { + if (!isRecord(history) || !Array.isArray(history.messages)) return null; + const { state, commits } = foldCommits({ + messages: history.messages.filter(isHistoryMessage), + }); + return { + acceptedCommits: commits.length, + entities: state.entities.length, + claims: state.claims.length, + obligations: state.obligations.length, + }; +}; + +/** + * Everything a packet reads from a persona run or turn-probe directory: + * `conversation.db` (required), and when present `dev-brunch-server.log`, + * `run.json`, `evidence/net.json`, and `evidence/snapshot.json` or + * `history.json` for the final Ledger. + */ +export const readRunDirectory = async ( + directory: string, +): Promise => { + const databasePath = join(directory, "conversation.db"); + if (!existsSync(databasePath)) { + throw new Error(`No conversation.db in ${directory}`); + } + const logPath = join(directory, "dev-brunch-server.log"); + const [metadata, log, net, snapshot, history] = await Promise.all([ + readMetadata(directory), + existsSync(logPath) ? readFile(logPath, "utf8") : Promise.resolve(""), + readJson(join(directory, "evidence", "net.json")), + readJson(join(directory, "evidence", "snapshot.json")), + readJson(join(directory, "history.json")), + ]); + const { submissions, events, skippedBatches, skippedEvents } = + readConversationDb(databasePath); + const { entries: chronology, skippedLines } = parseChronologyLog(log); + return { + label: basename(directory), + metadata, + submissions, + events, + chronology, + skipped: { + batches: skippedBatches, + events: skippedEvents, + chronologyLines: skippedLines, + }, + finalLedger: finalLedgerOf(snapshot ?? history), + finalNet: finalNetOf(net), + }; +}; diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/render-run-packet.ts b/apps/brunch-agent/evals/petrinaut/run-packet/render-run-packet.ts new file mode 100644 index 00000000000..71c4e00420f --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/render-run-packet.ts @@ -0,0 +1,213 @@ +import { toolCategories } from "./tool-categories.ts"; + +import type { + Distribution, + PacketExchange, + RunPacket, +} from "./build-run-packet.ts"; + +const seconds = (ms: number | null): string => + ms === null ? "–" : `${(ms / 1000).toFixed(ms < 10_000 ? 1 : 0)} s`; + +const share = (part: number | null, whole: number): string => + part === null || whole === 0 ? "–" : `${Math.round((100 * part) / whole)}%`; + +const distribution = ({ median, p90, max }: Distribution): string => + `median ${seconds(median)}, p90 ${seconds(p90)}, max ${seconds(max)}`; + +const table = ( + header: readonly string[], + rows: readonly (readonly string[])[], +) => + [ + `| ${header.join(" | ")} |`, + `| ${header.map(() => "---").join(" | ")} |`, + ...rows.map((row) => `| ${row.join(" | ")} |`), + ].join("\n"); + +const toolMix = (exchange: PacketExchange): string => + toolCategories + .flatMap((category) => + exchange.categories[category].calls === 0 + ? [] + : [`${category} ${exchange.categories[category].calls}`], + ) + .join(", "); + +const netCell = (exchange: PacketExchange): string => + exchange.netAfter === null + ? "–" + : `${exchange.netAfter.places}/${exchange.netAfter.transitions}/${exchange.netAfter.arcs}`; + +/** The packet as Markdown for reading beside the run's transcript. */ +export const renderRunPacket = (packet: RunPacket): string => { + const { metadata, summary } = packet; + const lines: string[] = [`# Run packet: ${packet.label}`, ""]; + if (metadata) { + lines.push( + `- Case: ${metadata.caseName ?? "–"}`, + `- Brunch: ${metadata.brunchModel ?? "default"} (${metadata.brunchThinking ?? "default"})`, + `- Persona: ${metadata.personaAgent ?? "–"}`, + `- Guidance: ${metadata.guidanceSha256?.slice(0, 12) ?? "–"}`, + `- Created: ${metadata.createdAt ?? "–"}`, + ); + } + if (!packet.sources.chronology) { + lines.push( + "- No server-log chronology: argument-streaming times are unknown, and time before output includes argument streaming for steps without text.", + ); + } + const { skipped } = packet.sources; + const { incomplete } = summary; + const gaps = [ + [ + incomplete.unsettledExchanges, + "never settled, left out of the time sums", + "exchange", + ], + [incomplete.cutOffMessages, "cut off before its message completed", "step"], + [ + incomplete.cutOffToolPhases, + "cut off before its tool results were committed", + "step", + ], + [skipped.batches, "unreadable, skipped", "stream batch"], + [skipped.events, "malformed, skipped", "stream event"], + [skipped.chronologyLines, "unreadable, skipped", "chronology line"], + ] as const; + const found = gaps.flatMap(([count, what, noun]) => + count === 0 ? [] : [`${count} ${noun}${count === 1 ? "" : "s"} ${what}`], + ); + if (found.length > 0) { + lines.push(`- Incomplete run: ${found.join("; ")}.`); + } + lines.push( + "", + "## Summary", + "", + `${summary.exchanges} exchanges, ${seconds(summary.activeMs)} of Brunch time (${seconds(summary.wallMs)} wall, less ${seconds(summary.approvalWaitMs)} waiting on approvals); ${summary.steps} model steps (${summary.stepsPerExchange?.toFixed(1) ?? "–"} per exchange: ${summary.stepsByToolCount.none} without tools, ${summary.stepsByToolCount.one} with one, ${summary.stepsByToolCount.several} with several).`, + "", + `- Per exchange: ${distribution(summary.activePerExchangeMs)}`, + `- Until first reply text (approval waits included): ${distribution(summary.untilFirstTextMs)}`, + "", + table( + ["Where the time goes", "Time", "Share of Brunch time"], + [ + [ + "Model steps", + seconds(summary.modelMs), + share(summary.modelMs, summary.activeMs), + ], + [ + "… before first output (hidden reasoning)", + seconds(summary.beforeOutputMs), + share(summary.beforeOutputMs, summary.activeMs), + ], + [ + "… tool-argument streaming", + seconds(summary.argumentStreamingMs), + share(summary.argumentStreamingMs, summary.activeMs), + ], + [ + "… reply text", + seconds(summary.textMs), + share(summary.textMs, summary.activeMs), + ], + [ + "Tool phases (execution, browser round trips)", + seconds(summary.toolPhaseMs - summary.approvalWaitMs), + share(summary.toolPhaseMs - summary.approvalWaitMs, summary.activeMs), + ], + ], + ), + "", + table( + [ + "Tool category", + "Calls", + "Execution (summed per call)", + "Argument chars", + ], + toolCategories + .filter((category) => summary.categories[category].calls > 0) + .map((category) => [ + category, + String(summary.categories[category].calls), + seconds(summary.categories[category].executionMs), + String(summary.categories[category].argumentChars), + ]), + ), + "", + `- Skill activations: ${ + Object.entries(summary.skillActivations) + .map(([name, count]) => `${name} ×${count}`) + .join(", ") || "none" + }; resource reads: ${summary.skillResourceReads}`, + `- Skill reloads: ${summary.skillReloads} (${summary.skillReloadsAfterCompaction} after a compaction); compactions: ${summary.compactions}`, + `- Ledger: ${summary.ledgerCommits} commits (${summary.ledgerRefused} refused), ${summary.ledgerArgumentChars} argument chars`, + `- Tool calls of ${seconds(30_000)} or more (likely approval waits): ${summary.longToolCalls}`, + `- Tokens: ${summary.usage.inputTokens} input, ${summary.usage.cacheReadTokens} cache read, ${summary.usage.cacheWriteTokens} cache write, ${summary.usage.outputTokens} output (${summary.usage.reasoningTokens} reasoning); USD ${summary.usage.costUsd.toFixed(2)}`, + ); + if (packet.finalLedger) { + const ledger = packet.finalLedger; + lines.push( + `- Final Ledger: ${ledger.acceptedCommits} accepted commits, ${ledger.entities} entities, ${ledger.claims} claims, ${ledger.obligations} obligations`, + ); + } + if (packet.finalNet) { + const net = packet.finalNet; + lines.push( + `- Final net: ${net.places} places, ${net.transitions} transitions, ${net.arcs} arcs; ${net.tokenTypes} token types, ${net.colouredPlaces} coloured places; ${net.parameters} parameters, ${net.differentialEquations} equations, ${net.metrics} metrics, ${net.scenarios} scenarios`, + ); + } + lines.push( + "", + "## Exchanges", + "", + table( + [ + "#", + "Active", + "First text", + "Steps", + "Before output", + "Args", + "Tools", + "Commit chars", + "Net P/T/A", + "Tool mix", + "Skills", + ], + packet.exchanges.map((exchange) => [ + exchange.outcome === "completed" + ? String(exchange.index) + : `${exchange.index} (${exchange.outcome ?? (exchange.wallMs === null ? "unsettled" : "no outcome")})`, + seconds(exchange.activeMs), + seconds(exchange.untilFirstTextMs), + String(exchange.steps.length), + seconds(exchange.beforeOutputMs), + seconds(exchange.argumentStreamingMs), + seconds(exchange.toolPhaseMs), + String(exchange.ledger.argumentChars), + netCell(exchange), + toolMix(exchange), + [ + ...exchange.compactions.map(() => "⟲ compaction"), + ...exchange.skillActivations, + ].join(", "), + ]), + ), + "", + "## Step shapes", + "", + ...packet.exchanges.map( + (exchange) => `${exchange.index}. ${exchange.stepShape.join(" → ")}`, + ), + "", + "## Scoring", + "", + `Against the case's \`key.md\`: not scored.`, + "", + ); + return lines.join("\n"); +}; diff --git a/apps/brunch-agent/evals/petrinaut/run-packet/tool-categories.ts b/apps/brunch-agent/evals/petrinaut/run-packet/tool-categories.ts new file mode 100644 index 00000000000..edf80fe954a --- /dev/null +++ b/apps/brunch-agent/evals/petrinaut/run-packet/tool-categories.ts @@ -0,0 +1,67 @@ +import { + brunchTools, + netReaderToolNames, + petrinautToolEffects, +} from "@hashintel/brunch-agent"; + +/** What a tool call spends a step on, for turn-time accounting. */ +export type ToolCategory = + | "ledger" + | "skills" + | "net-read" + | "compile-check" + | "layout" + | "net-change" + | "experiment" + | "other"; + +export const toolCategories: readonly ToolCategory[] = [ + "ledger", + "skills", + "net-read", + "compile-check", + "layout", + "net-change", + "experiment", + "other", +]; + +const netReadTools = new Set([ + "getLatestNetDefinition", + "readPetrinautDoc", + ...Object.values(netReaderToolNames), +]); + +const layoutTools = new Set([ + "applyAutoLayout", + "commitNodePositions", + "updatePlacePosition", + "updateTransitionPosition", + "updateComponentInstancePosition", +]); + +export const toolCategoryOf = (toolName: string): ToolCategory => { + if ( + toolName === brunchTools.ledgerCommit || + toolName === brunchTools.ledgerCompile + ) { + return "ledger"; + } + if ( + toolName === brunchTools.activateSkill || + toolName === brunchTools.readSkillResource + ) { + return "skills"; + } + if (netReadTools.has(toolName)) return "net-read"; + if (toolName === "getNetCompilationErrors") return "compile-check"; + if (layoutTools.has(toolName)) return "layout"; + if ( + toolName === brunchTools.draftPetrinautExperiment || + toolName === "createExperiment" + ) { + return "experiment"; + } + if (Object.hasOwn(petrinautToolEffects, toolName)) return "net-change"; + return "other"; +}; diff --git a/apps/brunch-agent/evals/petrinaut/turn-probe.ts b/apps/brunch-agent/evals/petrinaut/turn-probe.ts index 190b73ec34e..4330be8cde3 100644 --- a/apps/brunch-agent/evals/petrinaut/turn-probe.ts +++ b/apps/brunch-agent/evals/petrinaut/turn-probe.ts @@ -2,7 +2,7 @@ * One real Brunch turn through the built app, with no browser and no persona. * Sends a case's opening message in a conversation with no document binding, * so only the guidance and Ledger tools are mounted, then reports the reply, - * the tool calls and what the Ledger recorded. + * the tool calls and what the Ledger recorded, and writes a run packet. * * yarn workspace @apps/brunch-agent probe:turn [case] [provider/model] [thinking] * @@ -32,6 +32,7 @@ import { flueConversationIdFrom, } from "../../src/shared/conversation/identity.ts"; import { loadBuiltBrunchApplication } from "../../test/load-built-application.ts"; +import { writeRunPacket } from "./run-packet.ts"; const [ caseName = "pharma-cold-chain", @@ -61,6 +62,10 @@ const report = (text: string) => process.stdout.write(`${text}\n`); const application = await loadBuiltBrunchApplication(); const started = Date.now(); +const stop = new AbortController(); +const interrupt = () => stop.abort(new Error("Probe interrupted")); +const stopSignals = ["SIGINT", "SIGTERM", "SIGHUP"] as const; +for (const signal of stopSignals) process.once(signal, interrupt); try { const identity = { principalKey: "probe", @@ -76,7 +81,7 @@ try { }); await client.wait( await client.send({ message: { kind: "user", body: opening } }), - { signal: AbortSignal.timeout(600_000) }, + { signal: AbortSignal.any([AbortSignal.timeout(600_000), stop.signal]) }, ); const history = await client.history(); writeFileSync( @@ -117,4 +122,13 @@ try { report(`\nhistory: ${join(directory, "history.json")}`); } finally { await application.stop(); + for (const signal of stopSignals) process.removeListener(signal, interrupt); + // After the app stops, the conversation store is final. + await writeRunPacket(directory).then( + () => report(`packet: ${join(directory, "evidence", "packet.md")}`), + (error: unknown) => + process.stderr.write( + `Could not write the run packet (retry with \`yarn workspace @apps/brunch-agent packet:run ${directory}\`): ${String(error)}\n`, + ), + ); } diff --git a/apps/brunch-agent/package.json b/apps/brunch-agent/package.json index 3fc5d66e14f..873150d6ba0 100644 --- a/apps/brunch-agent/package.json +++ b/apps/brunch-agent/package.json @@ -16,6 +16,7 @@ "fix:eslint": "oxlint --fix --type-aware --type-check --report-unused-disable-directives-severity=error .", "lint:eslint": "oxlint --type-aware --type-check --report-unused-disable-directives-severity=error .", "lint:tsc": "tsgo --noEmit", + "packet:run": "node --experimental-strip-types --disable-warning=ExperimentalWarning evals/petrinaut/run-packet-cli.ts", "persona": "node --experimental-strip-types evals/petrinaut/persona/launch.ts", "petrinaut:dev": "vite dev --config petrinaut-local.vite.config.ts", "probe:turn": "node --experimental-strip-types evals/petrinaut/turn-probe.ts", diff --git a/apps/brunch-agent/src/server/observe-turn-chronology.ts b/apps/brunch-agent/src/server/observe-turn-chronology.ts index 747652f1fd7..8145b5c80d0 100644 --- a/apps/brunch-agent/src/server/observe-turn-chronology.ts +++ b/apps/brunch-agent/src/server/observe-turn-chronology.ts @@ -4,7 +4,8 @@ * observed progress) from slow generation (deltas still arriving); it does not * by itself separate provider failure from hidden reasoning. * - * Content policy: ids, counts and durations only — never argument text. + * Content policy: ids, counts, durations and turn start times only — never + * argument text. */ import type { @@ -45,6 +46,8 @@ type ResponseShape = { type TurnChronology = { readonly turnId: string; readonly purpose: string; + /** ISO-8601 time of `turn_start`, the origin of this turn's millisecond offsets. */ + readonly startedAt: string; /** Milliseconds from `turn_start` to the first model event; null when none arrived. */ readonly timeToFirstEventMs: number | null; /** Milliseconds from `turn_start` to `turn`, or to settlement when no `turn` arrived. */ @@ -108,6 +111,7 @@ const finishTurn = (turn: TurnState, terminalAt: number): TurnChronology => { return { turnId: turn.turnId, purpose: turn.purpose, + startedAt: new Date(turn.startedAt).toISOString(), timeToFirstEventMs: turn.firstEventAt === null ? null : turn.firstEventAt - turn.startedAt, durationMs: endedAt - turn.startedAt, diff --git a/apps/brunch-agent/src/server/turn-chronology.test.ts b/apps/brunch-agent/src/server/turn-chronology.test.ts index 536f61787f6..c4869ae3c60 100644 --- a/apps/brunch-agent/src/server/turn-chronology.test.ts +++ b/apps/brunch-agent/src/server/turn-chronology.test.ts @@ -96,6 +96,7 @@ test("one delta then silence reports the silence as last-delta to terminal and n { turnId: "t1", purpose: "prompt", + startedAt: at(0), timeToFirstEventMs: 1_200, durationMs: 21_700, terminal: "turn", @@ -128,6 +129,7 @@ test("one delta then silence reports the silence as last-delta to terminal and n { turnId: "t2", purpose: "prompt", + startedAt: at(22_000), timeToFirstEventMs: 100, durationMs: 30_100, terminal: "settlement", diff --git a/apps/petrinaut-website/.oxlintrc.json b/apps/petrinaut-website/.oxlintrc.json index e8ce8ff2601..27b6fdb27db 100644 --- a/apps/petrinaut-website/.oxlintrc.json +++ b/apps/petrinaut-website/.oxlintrc.json @@ -152,6 +152,14 @@ "no-throw-literal": "off", "prefer-promise-reject-errors": "off" }, + "overrides": [ + { + "files": ["src/main/app/plugins/brunch/**/*.{ts,tsx}"], + "rules": { + "react/memo-dependencies": "error" + } + } + ], "ignorePatterns": [ "dist/**", "styled-system/**", diff --git a/apps/petrinaut-website/react-plugin.ts b/apps/petrinaut-website/react-plugin.ts new file mode 100644 index 00000000000..cc128ed9851 --- /dev/null +++ b/apps/petrinaut-website/react-plugin.ts @@ -0,0 +1,21 @@ +import react from "@vitejs/plugin-react"; + +/** + * The React plugin with the React Compiler, shared by the app build and the + * unit tests so tests exercise the compiled components that ship. + */ +export const websiteReactPlugin = () => + react({ + // @hashintel/ds-components ships prebuilt jsx() calls; the compiler + // can't recognize ref forwarding in that form and bails with + // "Cannot access refs during render". Opt that package out. + exclude: [ + /[\\/]node_modules[\\/]/, + /[\\/]libs[\\/]@hashintel[\\/]ds-components[\\/]/, + ], + compiler: { + target: "19", + compilationMode: "infer", + panicThreshold: "all_errors", + }, + }); diff --git a/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.in-editor.test.tsx b/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.in-editor.test.tsx index 00059706064..0c9fda98542 100644 --- a/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.in-editor.test.tsx +++ b/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.in-editor.test.tsx @@ -354,7 +354,10 @@ test("refuses a static mutation on a read-only net before its host boundary runs }); test("starts Voice from the composer in the editor's panel and pauses it when the panel closes", async () => { - const events: string[] = []; + // Controls are spies, not closures over test state: the React Compiler + // hoists a closure the component doesn't capture out of the test's scope. + const endVoice = vi.fn(async () => {}); + const pauseVoice = vi.fn(); const VoiceMode = ({ registerVoiceModeControls, isAiAssistantOpen, @@ -363,12 +366,8 @@ test("starts Voice from the composer in the editor's panel and pauses it when th useEffect( () => registerVoiceModeControls({ - end: async () => { - events.push("end"); - }, - pause: () => { - events.push("pause"); - }, + end: endVoice, + pause: pauseVoice, reconnect: () => {}, resume: () => {}, setMicrophoneMuted: () => {}, @@ -396,7 +395,7 @@ test("starts Voice from the composer in the editor's panel and pauses it when th within(panel).getByRole("button", { name: "Start voice mode" }), ); expect(await screen.findByText("Voice voice open")).toBeTruthy(); - expect(events).toEqual([]); + expect(pauseVoice).not.toHaveBeenCalled(); act(() => screen.getByRole("button", { name: "Close AI assistant" }).click()); @@ -409,6 +408,7 @@ test("starts Voice from the composer in the editor's panel and pauses it when th ?.hasAttribute("inert"), ).toBe(true), ); - expect(events).toEqual(["pause"]); + expect(pauseVoice).toHaveBeenCalledOnce(); + expect(endVoice).not.toHaveBeenCalled(); expect(screen.getByText("Voice voice closed")).toBeTruthy(); }); diff --git a/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.tsx b/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.tsx index 65944e31f8e..0a0f73850fe 100644 --- a/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.tsx +++ b/apps/petrinaut-website/src/main/app/plugins/brunch/chat/ai-assistant-panel.tsx @@ -177,6 +177,13 @@ const voiceInputWithdrawn = (signal: AbortSignal | undefined): unknown => "AbortError", ); +const abortAndClear = (controllers: Set) => { + for (const controller of controllers) { + controller.abort(); + } + controllers.clear(); +}; + export const getVoiceToolCallIds = ( metadata: PetrinautAiMessage["metadata"], ): string[] => @@ -801,23 +808,9 @@ const ConversationAiAssistantPanel = ({ current === report ? null : current, ); }, []); - const abortAutomaticTools = () => { - for (const controller of automaticToolAbortsRef.current) { - controller.abort(); - } - automaticToolAbortsRef.current.clear(); - }; // Conversation switch remounts this panel via key; abort here, not only // when conversationId changes on the same instance. - useEffect( - () => () => { - for (const controller of automaticToolAbortsRef.current) { - controller.abort(); - } - automaticToolAbortsRef.current.clear(); - }, - [], - ); + useEffect(() => () => abortAndClear(automaticToolAbortsRef.current), []); const automaticToolExecutionTimersRef = useRef( new Map, string>(), ); @@ -1463,7 +1456,7 @@ const ConversationAiAssistantPanel = ({ submissionConversationIdRef.current = conversationId; followedMessagesRef.current = undefined; locallyStreamedToolCallsRef.current.clear(); - abortAutomaticTools(); + abortAndClear(automaticToolAbortsRef.current); explicitContinuationToolCallIdsRef.current.clear(); submissionGenerationRef.current += 1; stopRequestedRef.current = false; @@ -1935,7 +1928,7 @@ const ConversationAiAssistantPanel = ({ setStreamError(null); setStopped(false); stopRequestedRef.current = false; - abortAutomaticTools(); + abortAndClear(automaticToolAbortsRef.current); explicitContinuationToolCallIdsRef.current.clear(); submissionGenerationRef.current += 1; await submitMessage({ @@ -2053,7 +2046,7 @@ const ConversationAiAssistantPanel = ({ const generation = submissionGenerationRef.current; automaticToolTerminationRef.current = { generation, kind: "stopped" }; - abortAutomaticTools(); + abortAndClear(automaticToolAbortsRef.current); for (const controller of experimentControllersRef.current.values()) controller.abort(); stopRequestedRef.current = true; @@ -2313,7 +2306,7 @@ const ConversationAiAssistantPanel = ({ messages={aiAssistant.mapMessagesForDisplay?.(messages) ?? messages} renderSystemMessage={aiAssistant.renderSystemMessage} onClearMessages={() => { - abortAutomaticTools(); + abortAndClear(automaticToolAbortsRef.current); for (const controller of experimentControllersRef.current.values()) controller.abort(); experimentControllersRef.current.clear(); diff --git a/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.in-editor.test.tsx b/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.in-editor.test.tsx index 069d832d94b..ef0ce327067 100644 --- a/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.in-editor.test.tsx +++ b/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.in-editor.test.tsx @@ -6,7 +6,7 @@ import { render, screen, } from "@testing-library/react"; -import { use } from "react"; +import { use, useMemo } from "react"; import { afterEach, beforeAll, expect, test, vi } from "vitest"; import { createJsonDocHandle } from "@hashintel/petrinaut-core"; @@ -25,8 +25,11 @@ import { import { AiAssistantPanel } from "../chat/ai-assistant-panel"; import { brunchPlugin, BrunchAssistantContext } from "../editor-api-plugin"; import { + createBrunchMutationApprovalComponents, createBrunchMutationApprovalCoordinator, createBrunchMutationApprovalInteractiveTools, + useBrunchMutationApprovalTools, + type BrunchMutationApprovalCoordinator, } from "./brunch-mutation-approval"; import type { PetrinautAiMessage } from "../chat/ai-assistant-panel"; @@ -117,7 +120,10 @@ const openPendingApproval = async (plugins: readonly PetrinautPlugin[]) => { signal: new AbortController().signal, }); const aiAssistant: PetrinautAiAssistant = { - interactiveTools: createBrunchMutationApprovalInteractiveTools(coordinator), + interactiveTools: createBrunchMutationApprovalInteractiveTools( + createBrunchMutationApprovalComponents(coordinator), + coordinator.getPendingSnapshot(), + ), messages: [pendingRemoval], transport: { reconnectToStream: () => Promise.resolve(null), @@ -169,3 +175,84 @@ test("without the Brunch plugin, the approval says why it can't show and reports ).toBe("alert"); expect(screen.queryByRole("button", { name: "Allow" })).toBeNull(); }); + +// Two removals in one step, as Brunch sends them: the second asks for approval +// only after the first was allowed, while the stream is quiet. +const twoRemovals: PetrinautAiMessage = { + id: "assistant-2", + role: "assistant", + parts: ["remove-1", "remove-2"].map((toolCallId, index) => ({ + type: "dynamic-tool" as const, + toolName: "removePlace", + toolCallId, + state: "input-available" as const, + input: { placeId: `place-${index + 1}` }, + })), +}; + +const ApprovingEditor = ({ + coordinator, +}: { + coordinator: BrunchMutationApprovalCoordinator; +}) => { + const interactiveTools = useBrunchMutationApprovalTools(coordinator); + const aiAssistant = useMemo( + () => ({ + interactiveTools, + messages: [twoRemovals], + transport: { + reconnectToStream: () => Promise.resolve(null), + sendMessages: openStream, + }, + }), + [interactiveTools], + ); + const handle = useMemo( + () => createJsonDocHandle({ initial: { places: [], transitions: [] } }), + [], + ); + return ( + + + + ); +}; + +test("a removal that starts waiting after an earlier one was allowed still shows its approval", async () => { + vi.stubGlobal("ResizeObserver", NoopResizeObserver); + vi.stubGlobal("Worker", InProcessLspWorker); + const coordinator = createBrunchMutationApprovalCoordinator(); + coordinator.open(); + render( + + + , + ); + await act(async () => + fireEvent.click( + await screen.findByRole("button", { name: "Show AI assistant" }), + ), + ); + const request = (toolCallId: string) => + coordinator.request({ + toolCallId, + toolName: "removePlace", + signal: new AbortController().signal, + }); + + let first: ReturnType | undefined; + act(() => { + first = request("remove-1"); + }); + await act(async () => + fireEvent.click(await screen.findByRole("button", { name: "Allow" })), + ); + await expect(first).resolves.toEqual({ decision: "allow" }); + expect(screen.queryByRole("button", { name: "Allow" })).toBeNull(); + + act(() => { + void request("remove-2"); + }); + expect(await screen.findByRole("button", { name: "Allow" })).not.toBeNull(); + coordinator.close(); +}); diff --git a/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.test.tsx b/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.test.tsx index 0825035b600..78a741d8e8b 100644 --- a/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.test.tsx +++ b/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.test.tsx @@ -15,6 +15,7 @@ import { import { createBrunchMutationAdmission, + createBrunchMutationApprovalComponents, createBrunchMutationApprovalCoordinator, createBrunchMutationApprovalInteractiveTools, createBrunchMutationApprovalWidget, @@ -58,7 +59,10 @@ const netWith = (initial: SDCPNInput) => describe("Brunch destructive edit approval", () => { test("covers every destructive canonical tool and no constructive one", () => { const toolNames = createBrunchMutationApprovalInteractiveTools( - createBrunchMutationApprovalCoordinator(), + createBrunchMutationApprovalComponents( + createBrunchMutationApprovalCoordinator(), + ), + new Set(), ).map(({ toolName }) => toolName); expect(toolNames).toContain("deleteItemsByIds"); expect(toolNames).toContain("removeArc"); @@ -208,27 +212,30 @@ describe("Brunch destructive edit approval", () => { expect(coordinator.hasPending("delete-3")).toBe(true); }); - test("changes its snapshot for every pending call, including calls of the same tool", () => { + test("publishes a new pending snapshot for every change, including calls of the same tool", () => { const coordinator = createBrunchMutationApprovalCoordinator(); - const idle = coordinator.getVersion(); + const idle = coordinator.getPendingSnapshot(); void coordinator.request({ toolCallId: "remove-1", toolName: "removePlace", signal: new AbortController().signal, }); - const waiting = coordinator.getVersion(); + const waiting = coordinator.getPendingSnapshot(); expect(waiting).not.toBe(idle); - expect(coordinator.getVersion()).toBe(waiting); + expect([...waiting]).toEqual(["remove-1"]); + expect(coordinator.getPendingSnapshot()).toBe(waiting); void coordinator.request({ toolCallId: "remove-2", toolName: "removePlace", signal: new AbortController().signal, }); - const bothWaiting = coordinator.getVersion(); + const bothWaiting = coordinator.getPendingSnapshot(); expect(bothWaiting).not.toBe(waiting); + expect([...bothWaiting]).toEqual(["remove-1", "remove-2"]); coordinator.resolve("remove-1", "deny"); - expect(coordinator.getVersion()).not.toBe(bothWaiting); - expect(coordinator.hasPending("remove-2")).toBe(true); + expect([...coordinator.getPendingSnapshot()]).toEqual(["remove-2"]); + // An earlier snapshot never changes under a reader. + expect([...bothWaiting]).toEqual(["remove-1", "remove-2"]); coordinator.close(); }); diff --git a/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.tsx b/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.tsx index 7269f9ff63f..68d87c079f2 100644 --- a/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.tsx +++ b/apps/petrinaut-website/src/main/app/plugins/brunch/tools/brunch-mutation-approval.tsx @@ -1,3 +1,5 @@ +import { useMemo, useSyncExternalStore, type ComponentType } from "react"; + import { Button } from "@hashintel/ds-components"; import { css } from "@hashintel/ds-helpers/css"; import { @@ -46,8 +48,11 @@ export interface BrunchMutationApprovalCoordinator { * output round-trips, so its part alone cannot tell it from an executing call. */ approvalState(toolCallId: string): "awaiting" | "refused" | null; - /** Changes whenever a call enters or leaves the approval gate. */ - getVersion: () => number; + /** + * The calls waiting for a decision. A new set whenever a call enters or + * leaves the approval gate, so rendering from it re-renders on every change. + */ + getPendingSnapshot: () => ReadonlySet; subscribe: (listener: () => void) => () => void; /** Settles waiting approvals as stopped and refuses new ones until reopened. */ close(): void; @@ -66,9 +71,9 @@ export const createBrunchMutationApprovalCoordinator = const listeners = new Set<() => void>(); let alwaysAllow = false; let closed = false; - let version = 0; + let pendingSnapshot: ReadonlySet = new Set(); const notify = () => { - version += 1; + pendingSnapshot = new Set(pending.keys()); listeners.forEach((listener) => listener()); }; const settle = ( @@ -122,7 +127,7 @@ export const createBrunchMutationApprovalCoordinator = : refused.has(toolCallId) ? "refused" : null, - getVersion: () => version, + getPendingSnapshot: () => pendingSnapshot, subscribe: (listener) => { listeners.add(listener); return () => listeners.delete(listener); @@ -401,23 +406,77 @@ export const createBrunchMutationApprovalWidget = ( const passthrough = { parse: (value: unknown) => value }; -export const createBrunchMutationApprovalInteractiveTools = ( +/** + * Approval widgets for the calls in `pending`, a coordinator snapshot. Build + * them from each new snapshot: a widget shows only while its call waits. + */ +type ApprovalComponent = ComponentType< + PetrinautAiInteractiveToolWidgetProps +>; + +export type BrunchMutationApprovalComponents = Readonly< + Record +>; + +/** + * One approval widget per destructive tool. Create once per coordinator and + * reuse across snapshots, so a waiting approval re-renders instead of remounting. + */ +export const createBrunchMutationApprovalComponents = ( coordinator: BrunchMutationApprovalCoordinator, +): BrunchMutationApprovalComponents => + Object.fromEntries( + destructiveToolNames.map((toolName) => { + const Widget = createBrunchMutationApprovalWidget(coordinator, toolName); + const component: ApprovalComponent = (props) => ( + + {(api) => } + + ); + return [toolName, component]; + }), + ) as Record; + +/** + * Approval tools for the calls in `pending`, a coordinator snapshot. Build + * them from each new snapshot: a widget shows only while its call waits. + */ +export const createBrunchMutationApprovalInteractiveTools = ( + components: BrunchMutationApprovalComponents, + pending: ReadonlySet, ): readonly PetrinautAiInteractiveTool[] => - destructiveToolNames.map((toolName) => { - const Widget = createBrunchMutationApprovalWidget(coordinator, toolName); - return definePetrinautAiInteractiveTool({ + destructiveToolNames.map((toolName) => + definePetrinautAiInteractiveTool({ toolName, inputSchema: { parse: (value) => mutationActionInputSchemas[toolName].parse(value), }, outputSchema: passthrough, // Earlier rows of the same tool keep their normal presentation. - shouldHandle: ({ toolCallId }) => coordinator.hasPending(toolCallId), - component: (props) => ( - - {(api) => } - - ), - }); - }); + shouldHandle: ({ toolCallId }) => pending.has(toolCallId), + component: components[toolName], + }), + ); + +/** + * The approval tools for `coordinator`, rebuilt from each pending snapshot. A + * call that starts waiting while nothing else re-renders the transcript, such + * as the second removal of a step after the first was allowed, still shows. + */ +export const useBrunchMutationApprovalTools = ( + coordinator: BrunchMutationApprovalCoordinator, +): readonly PetrinautAiInteractiveTool[] => { + const pending = useSyncExternalStore( + coordinator.subscribe, + coordinator.getPendingSnapshot, + coordinator.getPendingSnapshot, + ); + const components = useMemo( + () => createBrunchMutationApprovalComponents(coordinator), + [coordinator], + ); + return useMemo( + () => createBrunchMutationApprovalInteractiveTools(components, pending), + [components, pending], + ); +}; diff --git a/apps/petrinaut-website/src/main/app/plugins/brunch/use-brunch-assistant.tsx b/apps/petrinaut-website/src/main/app/plugins/brunch/use-brunch-assistant.tsx index f61f27e620f..a2ea327c7db 100644 --- a/apps/petrinaut-website/src/main/app/plugins/brunch/use-brunch-assistant.tsx +++ b/apps/petrinaut-website/src/main/app/plugins/brunch/use-brunch-assistant.tsx @@ -64,7 +64,7 @@ import { BrunchExperimentFollowUp } from "./tools/brunch-experiment-follow-up"; import { createBrunchMutationAdmission, createBrunchMutationApprovalCoordinator, - createBrunchMutationApprovalInteractiveTools, + useBrunchMutationApprovalTools, } from "./tools/brunch-mutation-approval"; import { createCanonicalPetrinautHostTools, @@ -312,14 +312,19 @@ export const useBrunchAssistant = ({ : [], [wordsActive, conversationWords.entries], ); - // Each binding gets its own non-persisted approval authority. - const mutationApproval = useMemo( - () => ({ + // Each binding gets its own non-persisted approval authority. This is state + // reset on a new binding, not a memo keyed on it: the coordinator reads + // nothing from the binding, so the React Compiler would create it only once. + const [mutationApproval, setMutationApproval] = useState(() => ({ + binding: processAgentBinding, + coordinator: createBrunchMutationApprovalCoordinator(), + })); + if (mutationApproval.binding !== processAgentBinding) { + setMutationApproval({ binding: processAgentBinding, coordinator: createBrunchMutationApprovalCoordinator(), - }), - [processAgentBinding], - ); + }); + } // The panel stays mounted when the binding changes, so a replaced authority // must settle the approvals still waiting on it. useEffect(() => { @@ -327,28 +332,11 @@ export const useBrunchAssistant = ({ coordinator.open(); return () => coordinator.close(); }, [mutationApproval]); - const allMutationApprovalTools = useMemo( - () => - createBrunchMutationApprovalInteractiveTools( - mutationApproval.coordinator, - ), - [mutationApproval], - ); // A registered widget replaces the tool's row, so only calls still waiting - // for a decision render as approvals. Refresh the registry when call identities - // change; shouldHandle is the single gate, including for same-name calls. - const approvalVersion = useSyncExternalStore( - mutationApproval.coordinator.subscribe, - mutationApproval.coordinator.getVersion, - mutationApproval.coordinator.getVersion, + // for a decision render as approvals. + const mutationApprovalTools = useBrunchMutationApprovalTools( + mutationApproval.coordinator, ); - const mutationApprovalTools = useMemo( - () => ({ - version: approvalVersion, - tools: [...allMutationApprovalTools], - }), - [allMutationApprovalTools, approvalVersion], - ).tools; const processAgentSession = useProcessAgentSession({ binding: processAgentBinding, brunchSelected: selected, diff --git a/apps/petrinaut-website/vite.config.ts b/apps/petrinaut-website/vite.config.ts index a132fbed5e3..3d91292861c 100644 --- a/apps/petrinaut-website/vite.config.ts +++ b/apps/petrinaut-website/vite.config.ts @@ -1,7 +1,6 @@ import { fileURLToPath } from "node:url"; import { tanstackRouter } from "@tanstack/router-plugin/vite"; -import react from "@vitejs/plugin-react"; import { createServerAdapter } from "@whatwg-node/server"; import { defaultClientConditions, @@ -11,6 +10,7 @@ import { type Plugin, } from "vite"; +import { websiteReactPlugin } from "./react-plugin.ts"; import { routerCodegenConfig } from "./router-codegen-config.ts"; import type { IncomingMessage, ServerResponse } from "node:http"; @@ -121,20 +121,7 @@ export default defineConfig(({ command, mode }) => { plugins: [ petrinautApiDevPlugin(), tanstackRouter(routerCodegenConfig), - react({ - // @hashintel/ds-components ships prebuilt jsx() calls; the compiler - // can't recognize ref forwarding in that form and bails with - // "Cannot access refs during render". Opt that package out. - exclude: [ - /[\\/]node_modules[\\/]/, - /[\\/]libs[\\/]@hashintel[\\/]ds-components[\\/]/, - ], - compiler: { - target: "19", - compilationMode: "infer", - panicThreshold: "all_errors", - }, - }), + websiteReactPlugin(), ], }; }); diff --git a/apps/petrinaut-website/vitest.config.ts b/apps/petrinaut-website/vitest.config.ts index 998312f6524..0bc832c6304 100644 --- a/apps/petrinaut-website/vitest.config.ts +++ b/apps/petrinaut-website/vitest.config.ts @@ -1,6 +1,9 @@ import { configDefaults, defineConfig } from "vitest/config"; +import { websiteReactPlugin } from "./react-plugin.ts"; + export default defineConfig({ + plugins: [websiteReactPlugin()], test: { exclude: [...configDefaults.exclude, "src/**/*.integration.test.{ts,tsx}"], setupFiles: ["./vitest.setup.ts"],