diff --git a/README.md b/README.md index 50da473..0210ce3 100644 --- a/README.md +++ b/README.md @@ -47,19 +47,19 @@ returned. `unsure (p=…)` in between. `{ question, yes, no }` sets its own bars. - **Helpers**: -| helper | passes when | -| ----------------------------- | ----------------------------------------------------------------------------------------------------------- | -| `matches(field?)` | output deep-equals expectedOutput, or the same field (key or accessor) of each | -| `includes()` | both strings, and output contains expectedOutput after NFKC, trim, and whitespace collapse; value is 1 or 0 | -| `fitsSchema(schema)` | output satisfies a Standard Schema (arktype, zod, valibot) | -| `mentions(words)` | output contains every word, case-insensitive; value is the fraction found | -| `avoids(words)` | output contains none of the words | -| `ranSteps(names)` | `tool` and `step` names match `names`, order ignored; value is their F1 | -| `stepsAtMost(n)` | the run took at most `n` steps | -| `under({ ms })` | the run finished within `ms` | -| `ask(q, { options, pass })` | Jev picks an option named in `pass` | -| `ask(q, { options, passAt })` | Jev picks an option; its credit (0..1) is the value and passes at `passAt` | -| `ask(q, { levels, pass })` | Jev grades at level `pass` or above (levels lowest first) | +| helper | passes when | +| ----------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `matches(field?)` | output deep-equals expectedOutput, or the same field (key or accessor) of each | +| `includes()` | both strings, and output contains expectedOutput after NFKC, trim, and whitespace collapse; value is 1 or 0 | +| `fitsSchema(schema)` | output satisfies a Standard Schema (arktype, zod, valibot) | +| `mentions(words)` | output contains every word, case-insensitive; value is the fraction found | +| `avoids(words)` | output contains none of the words | +| `ranSteps(names)` | `tool` and `step` names match `names`, order ignored; value is their F1. Agent trajectory comes from `stream()`. A reply-only send without `stream` does not satisfy `ranSteps([])` | +| `stepsAtMost(n)` | the run took at most `n` steps | +| `under({ ms })` | the run finished within `ms` | +| `ask(q, { options, pass })` | Jev picks an option named in `pass` | +| `ask(q, { options, passAt })` | Jev picks an option; its credit (0..1) is the value and passes at `passAt` | +| `ask(q, { levels, pass })` | Jev grades at level `pass` or above (levels lowest first) | `mentions` and `avoids` also take `(r) => words` for per-case words. `mentions` and `ranSteps` take `{ atLeast }` to pass on partial credit. @@ -100,7 +100,12 @@ space cannot absorb — those fail closed by design. - **Function**: `(input, signal) => output`. `signal` aborts on `timeout`. - **Agent**: an `@intx/agent` Agent, detected by `send`. Input is sent - as a message; output is the reply. + as a message; output is the reply. When the agent has `stream()`, steps + are recorded from that stream (`inference.done` as `llm`, `tool.start` / + `tool.done` as `tool`). A policy deny emits `tool.done` without + `tool.start`; the tool is still recorded when `inference.tool_call.end` + or the assistant `tool_call` block names it. A reply-only send without + `stream` leaves `steps` empty, which does not satisfy `ranSteps([])`. - **Workflow**: `workflow(definition, { invokeStep })` from `@corbits/evals/workflow`. A definition cannot run locally without a step invoker, and the root never loads `@intx/workflow`, so a bare diff --git a/e2e/agent-trajectory.e2e.test.ts b/e2e/agent-trajectory.e2e.test.ts new file mode 100644 index 0000000..4d2449f --- /dev/null +++ b/e2e/agent-trajectory.e2e.test.ts @@ -0,0 +1,412 @@ +import { describe, expect, test } from "bun:test"; + +import { EvalFailure, ranSteps, runEvaluation } from "../src/index.js"; +import { toTraced, type AgentLike } from "../src/runs.js"; + +type Event = { type: string; data?: unknown }; +type SendResult = + | { type: "reply"; reply: string; turn?: { model?: string } } + | { type: "suspended" }; + +function createStub(options: { + stream: boolean; + onSend: (input: string, emit: (event: Event) => void, signal: AbortSignal) => Promise; +}): { agent: AgentLike; maxConsumers: () => number } { + let consumers = 0; + let maxConsumers = 0; + let buffer: Event[] = [{ type: "reactor.start" }]; + const waiters: Array<(result: IteratorResult) => void> = []; + + function emit(event: Event): void { + const waiter = waiters.shift(); + if (waiter !== undefined) waiter({ value: event, done: false }); + else buffer.push(event); + } + + const agent: AgentLike = { + send: (content, opts) => + options.onSend(content, emit, opts?.signal ?? new AbortController().signal), + }; + if (options.stream) { + agent.stream = () => ({ + [Symbol.asyncIterator](): AsyncIterator { + consumers += 1; + maxConsumers = Math.max(maxConsumers, consumers); + if (consumers > 1) throw new Error("two consumers"); + return { + next(): Promise> { + const queued = buffer.shift(); + if (queued !== undefined) return Promise.resolve({ value: queued, done: false }); + return new Promise((resolve) => { + waiters.push(resolve); + }); + }, + async return(): Promise> { + consumers -= 1; + const waiter = waiters.shift(); + if (waiter !== undefined) waiter({ value: undefined, done: true }); + return { value: undefined, done: true }; + }, + }; + }, + }); + } + return { agent, maxConsumers: () => maxConsumers }; +} + +async function failure(pending: Promise): Promise { + try { + await pending; + } catch (cause) { + if (cause instanceof EvalFailure) return cause; + throw cause; + } + throw new Error("expected runEvaluation to fail"); +} + +describe("agent trajectory", () => { + test("a watched tool cycle records llm and tool steps", async () => { + const { agent } = createStub({ + stream: true, + onSend: async (input, emit) => { + emit({ + type: "inference.done", + data: { + turn: { model: "stub", content: [{ type: "text", text: "looking" }] }, + }, + }); + emit({ + type: "inference.tool_call.end", + data: { callId: "c1", name: "search", arguments: { q: input } }, + }); + emit({ + type: "tool.start", + data: { call: { id: "c1", name: "search", arguments: { q: input } } }, + }); + emit({ type: "tool.update", data: { callId: "c1", partial: "x" } }); + emit({ type: "inference.text.delta", data: { token: "nope" } }); + emit({ + type: "tool.done", + data: { result: { callId: "orphan", content: "no start" } }, + }); + emit({ + type: "tool.done", + data: { result: { callId: "c1", content: "found" } }, + }); + emit({ + type: "inference.done", + data: { + turn: { + model: "stub", + content: [ + { type: "text", text: "done " }, + { type: "thinking", text: "hidden" }, + { type: "text", text: "now" }, + ], + }, + }, + }); + return { type: "reply", reply: "done now", turn: { model: "stub" } }; + }, + }); + const report = await runEvaluation({ + name: "tool-cycle", + task: agent, + data: [{ input: "order" }], + scorers: { ran: ranSteps(["search"]), none: ranSteps([]) }, + allow: { none: 1 }, + }); + expect(report.cases[0]?.steps).toEqual([ + { name: "stub", kind: "llm", output: "looking" }, + { name: "search", kind: "tool", input: { q: "order" }, output: "found" }, + { name: "stub", kind: "llm", output: "done now" }, + ]); + expect(report.cases[0]?.checks["ran"]?.pass).toBe(true); + expect(report.cases[0]?.checks["none"]?.pass).toBe(false); + }); + + test("a watched reply with no inference or tool is one llm step", async () => { + const { agent } = createStub({ + stream: true, + onSend: async () => ({ type: "reply", reply: "hello", turn: { model: "stub-model" } }), + }); + const report = await runEvaluation({ + name: "reply-only-watched", + task: agent, + data: [{ input: "hi" }], + scorers: { none: ranSteps([]), search: ranSteps(["search"]) }, + allow: { search: 1 }, + }); + expect(report.cases[0]?.steps).toEqual([{ name: "stub-model", kind: "llm", output: "hello" }]); + expect(report.cases[0]?.checks["none"]?.pass).toBe(true); + expect(report.cases[0]?.checks["search"]?.pass).toBe(false); + }); + + test("a watched reply without a model name is assistant", async () => { + const { agent } = createStub({ + stream: true, + onSend: async () => ({ type: "reply", reply: "hello" }), + }); + const report = await runEvaluation({ + name: "reply-only-assistant", + task: agent, + data: [{ input: "hi" }], + scorers: { none: ranSteps([]) }, + }); + expect(report.cases[0]?.steps).toEqual([{ name: "assistant", kind: "llm", output: "hello" }]); + }); + + test("send without stream leaves steps empty and ranSteps([]) fails", async () => { + const { agent } = createStub({ + stream: false, + onSend: async () => ({ type: "reply", reply: "hello", turn: { model: "stub" } }), + }); + const err = await failure( + runEvaluation({ + name: "send-only", + task: agent, + data: [{ input: "hi" }], + scorers: { none: ranSteps([]) }, + }), + ); + expect(err.report.cases[0]?.steps).toEqual([]); + expect(err.report.cases[0]?.checks["none"]).toMatchObject({ + pass: false, + value: 0, + explanation: "an empty step list is not an observed empty tool path", + }); + }); + + test("a suspended send still fails the case", async () => { + const { agent } = createStub({ + stream: false, + onSend: async () => ({ type: "suspended" }), + }); + const err = await failure( + runEvaluation({ + name: "suspended", + task: agent, + data: [{ input: "hi", expectedOutput: "hi" }], + }), + ); + expect(err.report.cases[0]?.error).toBe('agent did not reply (got "suspended")'); + }); + + test("overlapping cases on one agent do not mix steps", async () => { + const { agent, maxConsumers } = createStub({ + stream: true, + onSend: async (input, emit) => { + await new Promise((resolve) => setTimeout(resolve, 20)); + emit({ + type: "inference.done", + data: { turn: { model: "stub", content: [{ type: "text", text: input }] } }, + }); + emit({ + type: "tool.start", + data: { call: { id: input, name: input, arguments: { q: input } } }, + }); + emit({ + type: "tool.done", + data: { result: { callId: input, content: input } }, + }); + return { type: "reply", reply: input, turn: { model: "stub" } }; + }, + }); + const report = await runEvaluation({ + name: "overlap", + task: agent, + concurrency: 2, + data: [ + { name: "a", input: "search" }, + { name: "b", input: "lookup" }, + ], + scorers: { ran: ranSteps([]) }, + allow: { ran: 2 }, + }); + const byName = Object.fromEntries( + report.cases.map((c) => [c.name, c.steps.map((s) => s.name)]), + ); + expect(byName).toEqual({ + a: ["stub", "search"], + b: ["stub", "lookup"], + }); + expect(maxConsumers()).toBe(1); + }); + + test("a reply inference.done emitted after send still lands", async () => { + const { agent } = createStub({ + stream: true, + onSend: async (_input, emit) => { + emit({ + type: "tool.start", + data: { call: { id: "c1", name: "search", arguments: { q: "late" } } }, + }); + emit({ + type: "tool.done", + data: { result: { callId: "c1", content: "found" } }, + }); + setTimeout(() => { + emit({ + type: "inference.done", + data: { + turn: { model: "reply-model", content: [{ type: "text", text: "from-stream" }] }, + }, + }); + }, 0); + return { type: "reply", reply: "from-send", turn: { model: "send-model" } }; + }, + }); + const report = await runEvaluation({ + name: "late-inference", + task: agent, + data: [{ input: "late" }], + scorers: { ran: ranSteps(["search"]) }, + }); + expect(report.cases[0]?.steps).toEqual([ + { name: "search", kind: "tool", input: { q: "late" }, output: "found" }, + { name: "reply-model", kind: "llm", output: "from-stream" }, + ]); + }); + + test("an aborted waiter does not send", async () => { + let sends = 0; + let releaseA: () => void = () => {}; + const gate = new Promise((resolve) => { + releaseA = resolve; + }); + const { agent } = createStub({ + stream: true, + onSend: async () => { + sends += 1; + await gate; + return { type: "reply", reply: "a", turn: { model: "stub" } }; + }, + }); + const run = toTraced(agent); + const inFlight = new AbortController(); + const queued = new AbortController(); + const pendingA = run("a", inFlight.signal); + await new Promise((resolve) => setTimeout(resolve, 10)); + const pendingB = run("b", queued.signal); + await new Promise((resolve) => setTimeout(resolve, 10)); + queued.abort(); + releaseA(); + await pendingA; + await expect(pendingB).rejects.toThrow("run aborted before send"); + expect(sends).toBe(1); + }); + + test("an aborted turn's late tool does not land on the next case", async () => { + const { agent } = createStub({ + stream: true, + onSend: (input, emit, signal) => { + if (input !== "slow") { + return new Promise((resolve) => { + setTimeout(() => { + resolve({ type: "reply", reply: "found", turn: { model: "stub" } }); + }, 80); + }); + } + return new Promise((_resolve, reject) => { + const abort = (): void => { + reject(signal.reason instanceof Error ? signal.reason : new Error("aborted")); + setTimeout(() => { + emit({ + type: "tool.start", + data: { call: { id: "late", name: "search", arguments: { q: "slow" } } }, + }); + emit({ type: "connector.reply", data: { content: "late" } }); + }, 30); + }; + if (signal.aborted) abort(); + else signal.addEventListener("abort", abort, { once: true }); + }); + }, + }); + const err = await failure( + runEvaluation({ + name: "abort-then-lookup", + task: agent, + concurrency: 1, + timeout: 200, + data: [ + { name: "slow", input: "slow" }, + { name: "lookup", input: "lookup" }, + ], + scorers: { none: ranSteps([]), search: ranSteps(["search"]) }, + }), + ); + const lookup = err.report.cases.find((c) => c.name === "lookup"); + expect(err.report.cases[0]?.error).toContain("timed out"); + expect(lookup?.error).toBeUndefined(); + expect(lookup?.steps).toEqual([{ name: "stub", kind: "llm", output: "found" }]); + expect(lookup?.checks["none"]?.pass).toBe(true); + expect(lookup?.checks["search"]?.pass).toBe(false); + }); + + test("a policy deny records the tool name without tool.start", async () => { + const { agent } = createStub({ + stream: true, + onSend: async (input, emit) => { + if (input === "from-block") { + emit({ + type: "inference.done", + data: { + turn: { + model: "stub", + content: [ + { type: "text", text: "no" }, + { type: "tool_call", id: "c1", name: "search", arguments: { q: input } }, + ], + }, + }, + }); + } else { + emit({ + type: "inference.done", + data: { turn: { model: "stub", content: [{ type: "text", text: "no" }] } }, + }); + emit({ + type: "inference.tool_call.end", + data: { callId: "c1", name: "search", arguments: { q: input } }, + }); + emit({ + type: "inference.tool_call.end", + data: { callId: "empty", name: "", arguments: {} }, + }); + } + emit({ + type: "tool.done", + data: { result: { callId: "nameless", content: "no name" } }, + }); + emit({ + type: "tool.done", + data: { result: { callId: "empty", content: "skip" } }, + }); + emit({ + type: "tool.done", + data: { result: { callId: "c1", content: "blocked", isError: true } }, + }); + return { type: "reply", reply: "no", turn: { model: "stub" } }; + }, + }); + const report = await runEvaluation({ + name: "policy-deny", + task: agent, + data: [ + { name: "from-end", input: "from-end" }, + { name: "from-block", input: "from-block" }, + ], + scorers: { ran: ranSteps(["search"]), none: ranSteps([]) }, + allow: { none: 2 }, + }); + for (const evalCase of report.cases) { + expect(evalCase.steps).toEqual([ + { name: "stub", kind: "llm", output: "no" }, + { name: "search", kind: "tool", input: { q: evalCase.input }, output: "blocked" }, + ]); + expect(evalCase.checks["ran"]?.pass).toBe(true); + expect(evalCase.checks["none"]?.pass).toBe(false); + } + }); +}); diff --git a/src/checks.ts b/src/checks.ts index 613a289..c784fe7 100644 --- a/src/checks.ts +++ b/src/checks.ts @@ -181,6 +181,13 @@ export function ranSteps(names: readonly string[], options: AtLeast = {}): Check const actual = stepNames(run).sort(); const explanation = `expected [${expected.join(", ")}], got [${actual.join(", ")}]`; if (expected.length === 0) { + if (run.steps.length === 0) { + return { + pass: false, + value: 0, + explanation: "an empty step list is not an observed empty tool path", + }; + } return { pass: actual.length === 0, value: actual.length === 0 ? 1 : 0, explanation }; } const counts = new Map(); diff --git a/src/runner.ts b/src/runner.ts index 798eff8..7bd55d9 100644 --- a/src/runner.ts +++ b/src/runner.ts @@ -47,7 +47,14 @@ export async function runOnce( }, timeoutMs); }); try { - const trace = await Promise.race([spec.run(evalCase.input, controller.signal), expiry]); + const pending = spec.run(evalCase.input, controller.signal); + // Timeout settles this race while an aborted turn may still be draining. + // Observe that later rejection so it is not unhandled. + void pending.then( + () => {}, + () => {}, + ); + const trace = await Promise.race([pending, expiry]); const run: Run = { input: evalCase.input, output: trace.output, diff --git a/src/runs.ts b/src/runs.ts index f446d98..435f34b 100644 --- a/src/runs.ts +++ b/src/runs.ts @@ -5,12 +5,16 @@ export type Trace = { output: O; steps: Step[] }; export type RunFn = (input: I, signal: AbortSignal) => O | Promise; +type StreamEvent = { type: string; data?: unknown }; + +type AgentSendResult = + | { type: "reply"; reply: string; turn?: { model?: string } } + | { type: "suspended" }; + /** Structural `@intx/agent` Agent, so the root never imports it. */ export type AgentLike = { - send( - content: string, - opts?: { signal?: AbortSignal }, - ): Promise<{ type: "reply"; reply: string } | { type: "suspended" }>; + send(content: string, opts?: { signal?: AbortSignal }): Promise; + stream?(): AsyncIterable; }; export const TRACE: unique symbol = Symbol("@corbits/evals trace"); @@ -28,8 +32,310 @@ function isWorkflowDefinition(value: object): boolean { return "stepOrder" in value && "steps" in value && "triggers" in value; } +const tails = new WeakMap>(); + +function withAgentLock(agent: object, fn: () => Promise): Promise { + const prev = tails.get(agent) ?? Promise.resolve(); + let release: (() => void) | undefined; + const gate = new Promise((resolve) => { + release = resolve; + }); + tails.set( + agent, + prev.then( + () => gate, + () => gate, + ), + ); + return prev + .then( + () => fn(), + () => fn(), + ) + .finally(() => { + release?.(); + }); +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null; +} + +function joinedText(turn: unknown): { name: string; output: string } { + if (!isRecord(turn)) return { name: "assistant", output: "" }; + const name = typeof turn.model === "string" && turn.model.length > 0 ? turn.model : "assistant"; + if (!Array.isArray(turn.content)) return { name, output: "" }; + const parts: string[] = []; + for (const block of turn.content) { + if (isRecord(block) && block.type === "text" && typeof block.text === "string") { + parts.push(block.text); + } + } + return { name, output: parts.join("") }; +} + +type KnownCall = { name: string; input: unknown; step?: Step }; + +function rememberCall( + tools: Map, + id: string, + name: string, + input: unknown, +): void { + if (name.length === 0 || tools.has(id)) return; + tools.set(id, { name, input }); +} + +/** Names a blocked call can carry without `tool.start`. */ +function rememberTurnCalls(turn: unknown, tools: Map): void { + if (!isRecord(turn) || !Array.isArray(turn.content)) return; + for (const block of turn.content) { + if (!isRecord(block) || block.type !== "tool_call") continue; + if (typeof block.id !== "string" || typeof block.name !== "string") continue; + rememberCall(tools, block.id, block.name, block.arguments); + } +} + +function toolName(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined; +} + +function recordEvent(event: StreamEvent, steps: Step[], tools: Map): boolean { + if (event.type === "inference.tool_call.end") { + const data = isRecord(event.data) ? event.data : undefined; + if (data === undefined || typeof data.callId !== "string") return false; + const name = toolName(data.name); + if (name === undefined) return false; + rememberCall(tools, data.callId, name, data.arguments); + return false; + } + if (event.type === "inference.done") { + const turn = isRecord(event.data) ? event.data.turn : undefined; + rememberTurnCalls(turn, tools); + const text = joinedText(turn); + steps.push({ name: text.name, kind: "llm", output: text.output }); + return true; + } + if (event.type === "tool.start") { + const call = isRecord(event.data) ? event.data.call : undefined; + if (!isRecord(call)) return false; + const id = typeof call.id === "string" ? call.id : undefined; + const known = id === undefined ? undefined : tools.get(id); + const name = toolName(call.name) ?? known?.name; + if (name === undefined) return false; + if (known?.step !== undefined) return true; + const step: Step = { name, kind: "tool", input: call.arguments ?? known?.input }; + steps.push(step); + if (id !== undefined) tools.set(id, { name, input: step.input, step }); + return true; + } + if (event.type === "tool.done") { + const result = isRecord(event.data) ? event.data.result : undefined; + if (!isRecord(result) || typeof result.callId !== "string") return false; + const known = tools.get(result.callId); + if (known === undefined) return false; + let step = known.step; + if (step === undefined) { + step = { name: known.name, kind: "tool", input: known.input }; + steps.push(step); + known.step = step; + } + step.output = "content" in result ? result.content : result; + return true; + } + return false; +} + +type Raced = { ready: true; result: T } | { ready: false }; + +function raceReady(pending: Promise): Promise> { + return new Promise((resolve, reject) => { + let settled = false; + pending.then( + (result) => { + settled = true; + resolve({ ready: true, result }); + }, + (error: unknown) => { + settled = true; + reject(error); + }, + ); + queueMicrotask(() => { + if (!settled) resolve({ ready: false }); + }); + }); +} + +/** Drop events already queued at subscribe. The returned read is the first live one. */ +async function discardBuffered( + iterator: AsyncIterator, +): Promise<{ pending: Promise> }> { + let pending = iterator.next(); + for (;;) { + const raced = await raceReady(pending); + if (!raced.ready) return { pending }; + if (raced.result.done === true) return { pending: Promise.resolve(raced.result) }; + pending = iterator.next(); + } +} + +async function takeReady( + pending: Promise>, +): Promise | undefined> { + const raced = await raceReady(pending); + if (!raced.ready) return undefined; + return raced.result; +} + +/** Quiet once a macrotask passes with no event, so a post-send `setTimeout(0)` is kept. */ +async function nextEventOrIdle( + pending: Promise>, +): Promise<{ idle: true } | { idle: false; result: IteratorResult }> { + let timer: ReturnType | undefined; + const idle = new Promise<{ idle: true }>((resolve) => { + timer = setTimeout(() => { + resolve({ idle: true }); + }, 0); + }); + try { + return await Promise.race([pending.then((result) => ({ idle: false as const, result })), idle]); + } finally { + clearTimeout(timer); + } +} + +/** The reactor settles the in-flight send on these, then stops this turn. */ +function isTurnEnd(event: StreamEvent): boolean { + if ( + event.type === "connector.reply" || + event.type === "reactor.gate.blocked" || + event.type === "reactor.done" + ) { + return true; + } + return event.type === "reactor.error" && isRecord(event.data) && event.data.fatal === true; +} +async function collect( + iterator: AsyncIterator, + first: Promise>, + steps: Step[], + send: Promise, + signal: AbortSignal, +): Promise { + let saw = false; + const tools = new Map(); + let sendSettled = false; + let sendFailed = false; + const mark = send.then( + () => { + sendSettled = true; + }, + () => { + sendSettled = true; + sendFailed = true; + }, + ); + const note = (event: StreamEvent): void => { + if (recordEvent(event, steps, tools)) saw = true; + }; + let next = first; + while (!sendSettled) { + const winner = await Promise.race([ + next.then((result) => ({ kind: "event" as const, result })), + mark.then(() => ({ kind: "send" as const })), + ]); + if (winner.kind === "send") break; + if (winner.result.done === true) break; + note(winner.result.value); + next = iterator.next(); + } + // Abort rejects `send` while the reactor keeps emitting. Hold this + // subscription until the turn ends so a late tool cannot be recorded + // on the next case, where it would also suppress that case's model step. + let holdForTurn = sendFailed && signal.aborted; + for (;;) { + if (holdForTurn) { + const result = await next; + if (result.done === true) break; + note(result.value); + next = iterator.next(); + if (isTurnEnd(result.value)) holdForTurn = false; + continue; + } + const winner = await nextEventOrIdle(next); + if (winner.idle) { + const ready = await takeReady(next); + if (ready === undefined || ready.done === true) break; + note(ready.value); + next = iterator.next(); + continue; + } + if (winner.result.done === true) break; + note(winner.result.value); + next = iterator.next(); + } + void next.then( + () => {}, + () => {}, + ); + return saw; +} + +function replyModel(res: AgentSendResult): string { + if (res.type !== "reply") return "assistant"; + const model = res.turn?.model; + return typeof model === "string" && model.length > 0 ? model : "assistant"; +} + +async function sendAgent( + agent: AgentLike, + input: string, + signal: AbortSignal, +): Promise> { + if (agent.stream === undefined) { + const res = await agent.send(input, { signal }); + if (res.type !== "reply") throw new Error(`agent did not reply (got "${res.type}")`); + return { output: res.reply as O, steps: [] }; + } + const stream = agent.stream; + return withAgentLock(agent, async () => { + if (signal.aborted) throw new Error("run aborted before send"); + const iterator = stream()[Symbol.asyncIterator](); + const steps: Step[] = []; + try { + const first = await discardBuffered(iterator); + let settled: { ok: true; res: AgentSendResult } | { ok: false; error: unknown } | undefined; + const pending = agent.send(input, { signal }); + const send = pending.then( + (res) => { + settled = { ok: true, res }; + }, + (error: unknown) => { + settled = { ok: false, error }; + }, + ); + const saw = await collect(iterator, first.pending, steps, pending, signal); + await send; + if (settled === undefined || !settled.ok) { + throw settled === undefined ? new Error("agent send did not settle") : settled.error; + } + if (settled.res.type !== "reply") { + throw new Error(`agent did not reply (got "${settled.res.type}")`); + } + if (!saw) { + steps.push({ name: replyModel(settled.res), kind: "llm", output: settled.res.reply }); + } + return { output: settled.res.reply as O, steps }; + } finally { + await iterator.return?.(); + } + }); +} + /** - * Agents run with no recorded steps. A raw workflow definition is refused: + * Agents record steps from `stream()` when it exists. Without it, a reply + * is returned with no steps. A raw workflow definition is refused: * running it needs a step invoker, which `workflow()` takes. */ export function toTraced( @@ -47,9 +353,7 @@ export function toTraced( return async (input, signal) => { if (typeof input !== "string") throw new Error("agent input must be a string"); if (signal.aborted) throw new Error("run aborted before send"); - const res = await run.send(input, { signal }); - if (res.type !== "reply") throw new Error(`agent did not reply (got "${res.type}")`); - return { output: res.reply as O, steps: [] }; + return sendAgent(run, input, signal); }; } if (isWorkflowDefinition(run)) {