diff --git a/README.md b/README.md index eb1b167..de6bfb8 100644 --- a/README.md +++ b/README.md @@ -69,20 +69,22 @@ A check that throws, hangs past `timeout`, or returns anything else fails itself closed; its siblings still run. Jev transport failures fail closed. Jev sees `{ input, output, expectedOutput, steps }`, and a case's string and -`ask` checks answer in one call. Configure it with -`judge: jev({ endpoint, model, apiKey, timeout })`; `endpoint` is -`"official"` (default), `"gateway"` or a URL. Without an `apiKey` the -key comes from the environment as `@corbits/system-one` reads it. +`ask` checks answer in one call. The model is not configured on `jev()`. +Build it with `judgeModel({ model, invokeStep, thresholds, escalate, timeout })` +from `@corbits/evals/workflow`. `model` is one workflow step (or a one-step +workflow). `invokeStep` runs that step; its input is `{ state, questions }` +and its output is `{ decisions }` or `{ fallback: true, reason }`. +`jev({ thresholds, escalate })` only sets bars and an escalation agent. Thresholds: a yes/no check passes at p(yes) >= `yes` and fails at p(yes) <= `no` (0.7 / 0.3); `ask` fails as unsure when its top probability is under `sure` (0.6). Retune them per judge with -`jev({ thresholds: { yes, no, sure } })` or per check as above. -Thresholds are tuned to a judge version, so pin `model` — e.g. -`jev({ model: "jev-1.13.0" })` — when they must stay stable. +`judgeModel({ thresholds: { yes, no, sure }, ... })` or per check as above. +Thresholds are tuned to a judge version, so pin the step agent's model +when they must stay stable. Escalation: `sure` doubles as the escalation gate. Give the judge an -agent — `jev({ escalate: agent })`, anything with +agent — `escalate` on `jev` or `judgeModel`, anything with `send(prompt)` returning `{ type: "reply", reply }` or `{ type: "suspended" }`, e.g. an `@intx/agent` agent from a `@corbits/*` adapter — and every unsure decision is re-asked with the exact answer @@ -104,7 +106,8 @@ space cannot absorb — those fail closed by design. definition is refused. Each step lands in `r.steps`. `output` picks the result from the finished run (default: the last listed step's output), `trigger` maps the input to the trigger payload. Without them, type the - run as `workflow(...)`. + run as `workflow(...)`. The same entry exports `judgeModel` for + string and `ask` scorers. A code-only eval needs no judge. `repeats` runs each case several times; a check passes only if every attempt passes. `concurrency` (default 4) bounds cases in flight; diff --git a/e2e/calibrate.e2e.test.ts b/e2e/calibrate.e2e.test.ts index 7f95eb6..86a7ce3 100644 --- a/e2e/calibrate.e2e.test.ts +++ b/e2e/calibrate.e2e.test.ts @@ -1,48 +1,43 @@ import { describe, expect, test } from "bun:test"; -import type { EvaluateDeps } from "@corbits/system-one"; +import { defineAgent } from "@intx/agent"; +import { step, type StepInvoker } from "@intx/workflow"; -import { ask, calibrate, jev } from "../src/index.js"; - -function stubDeps(fetch: EvaluateDeps["fetch"]): EvaluateDeps { - return { - fetch, - scheduler: { - setTimeout: ((fn: () => void, ms: number) => { - const t = setTimeout(fn, ms); - return () => clearTimeout(t); - }) as EvaluateDeps["scheduler"]["setTimeout"], - now: () => Date.now(), - }, - }; -} +import { ask, calibrate } from "../src/index.js"; +import { judgeModel } from "../src/workflow.js"; /** Answers yes/no questions with the input's probability; choice questions pick ready/revise by it. */ function stubCalibration(byInput: Record) { - const fetch = (async (_url: string, init: RequestInit) => { - const body = JSON.parse(String(init.body)) as { + const invokeStep: StepInvoker = async (request) => { + const body = request.input as { state: { input: string }; questions: Record; }; const p = byInput[body.state.input] ?? 0.5; - const answers = Object.fromEntries( - Object.entries(body.questions).map(([id, q]) => [ - id, - q.type === "choice" - ? { - type: "choice", - choice: p >= 0.5 ? "ready" : "revise", - probabilities: { ready: p, revise: Math.round((1 - p) * 100) / 100 }, - confidence: p, - } - : { type: "noul", noul: p }, - ]), + const decisions = Object.entries(body.questions).map(([id, q]) => + q.type === "choice" + ? { + id, + type: "choice", + choice: p >= 0.5 ? "ready" : "revise", + probabilities: { ready: p, revise: Math.round((1 - p) * 100) / 100 }, + confidence: p, + } + : { id, type: "noul", noul: p }, ); - return new Response(JSON.stringify({ model: "jev-test", answers }), { - status: 200, - headers: { "content-type": "application/json" }, - }); - }) as unknown as EvaluateDeps["fetch"]; - return jev({ apiKey: "test-key", deps: stubDeps(fetch) }); + return { output: { decisions } }; + }; + return judgeModel({ + model: step({ + agent: defineAgent({ + id: "jev", + systemPrompt: "judge", + tools: [], + capabilities: [], + inference: { sources: [{ provider: "stub", model: "stub" }] }, + }), + }), + invokeStep, + }); } describe("calibrate", () => { diff --git a/e2e/jev.e2e.test.ts b/e2e/jev.e2e.test.ts index a37ab9b..65328a2 100644 --- a/e2e/jev.e2e.test.ts +++ b/e2e/jev.e2e.test.ts @@ -2,7 +2,15 @@ import { describe, expect, test } from "bun:test"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import type { EvaluateDeps } from "@corbits/system-one"; +import { defineAgent } from "@intx/agent"; +import { + childWorkflow, + defineWorkflow, + map, + step, + type StepInvoker, + type WorkflowDefinition, +} from "@intx/workflow"; import { ask, @@ -15,33 +23,37 @@ import { type Outcome, type Report, } from "../src/index.js"; - -function stubDeps(fetch: EvaluateDeps["fetch"]): EvaluateDeps { - return { - fetch, - scheduler: { - setTimeout: ((fn: () => void, ms: number) => { - const t = setTimeout(fn, ms); - return () => clearTimeout(t); - }) as EvaluateDeps["scheduler"]["setTimeout"], - now: () => Date.now(), - }, - }; +import { judgeModel } from "../src/workflow.js"; + +function judgeAgent(sources: readonly { provider: string; model: string }[]) { + return defineAgent({ + id: "jev", + systemPrompt: "judge", + tools: [], + capabilities: [], + inference: { sources }, + }); } -/** Answers every question on a POST with `answer` and records the request bodies. */ -function stubJev(answer: unknown) { +/** Answers every question with `answer` and records the step request. */ +function stubJev( + answer: unknown, + sources: readonly { provider: string; model: string }[] = [{ provider: "stub", model: "stub" }], +) { const bodies: Record[] = []; - const fetch = (async (_url: string, init: RequestInit) => { - const body = JSON.parse(String(init.body)) as { questions: Record }; - bodies.push(body); - const answers = Object.fromEntries(Object.keys(body.questions).map((id) => [id, answer])); - return new Response(JSON.stringify({ model: "jev-test", answers }), { - status: 200, - headers: { "content-type": "application/json" }, - }); - }) as unknown as EvaluateDeps["fetch"]; - return { bodies, judge: jev({ apiKey: "test-key", deps: stubDeps(fetch) }) }; + const seen: { provider: string; model: string }[][] = []; + const invokeStep: StepInvoker = async (request) => { + const input = request.input as { questions: Record }; + bodies.push(input); + seen.push(request.agent.inference.sources.map((source) => ({ ...source }))); + const decisions = Object.keys(input.questions).map((id) => ({ id, ...(answer as object) })); + return { output: { decisions } }; + }; + return { + bodies, + seen, + judge: judgeModel({ model: step({ agent: judgeAgent(sources) }), invokeStep }), + }; } async function judged(check: Scorer, judge: Judge) { @@ -101,16 +113,20 @@ describe("sentence checks", () => { }); }); - test("transport failure fails closed", async () => { - const failing = (async () => { + test("a thrown invoke fails closed", async () => { + const failing: StepInvoker = async () => { throw new TypeError("boom"); - }) as unknown as EvaluateDeps["fetch"]; + }; const { outcome } = await judged( question, - jev({ apiKey: "test-key", deps: stubDeps(failing) }), + judgeModel({ + model: step({ agent: judgeAgent([{ provider: "stub", model: "stub" }]) }), + invokeStep: failing, + }), ); expect(outcome).toMatchObject({ pass: false, value: 0 }); expect(outcome?.explanation).toContain("jev fallback"); + expect(outcome?.explanation).toContain("boom"); }); }); @@ -138,10 +154,10 @@ describe("batching", () => { }); }); - test("a transport failure fails every Jev check in the batch", async () => { - const failing = (async () => { + test("a thrown invoke fails every Jev check in the batch", async () => { + const failing: StepInvoker = async () => { throw new TypeError("boom"); - }) as unknown as EvaluateDeps["fetch"]; + }; let report: Report; try { await runEvaluation({ @@ -149,7 +165,10 @@ describe("batching", () => { task: (input: string) => input, data: [{ input: "a" }], scorers: { ready: "Ready to send?", polite: "Is the tone polite?" }, - judge: jev({ apiKey: "test-key", deps: stubDeps(failing) }), + judge: judgeModel({ + model: step({ agent: judgeAgent([{ provider: "stub", model: "stub" }]) }), + invokeStep: failing, + }), }); throw new Error("expected failure"); } catch (cause) { @@ -180,24 +199,21 @@ describe("thresholds", () => { const question = "Is this reply ready to send to the customer?"; const noul = (p: number) => ({ type: "noul", noul: p }); - function flatFetch(answer: unknown) { - return (async (_url: string, init: RequestInit) => { - const body = JSON.parse(String(init.body)) as { questions: Record }; - const answers = Object.fromEntries(Object.keys(body.questions).map((id) => [id, answer])); - return new Response(JSON.stringify({ model: "jev-test", answers }), { - status: 200, - headers: { "content-type": "application/json" }, - }); - }) as unknown as EvaluateDeps["fetch"]; + function answering(answer: unknown): StepInvoker { + return async (request) => { + const questions = (request.input as { questions: Record }).questions; + const decisions = Object.keys(questions).map((id) => ({ id, ...(answer as object) })); + return { output: { decisions } }; + }; } test("judge thresholds retune every check's bars", async () => { const { judge } = stubJev(noul(0.75)); expect((await judged(question, judge)).outcome?.pass).toBe(true); - const strict = jev({ - apiKey: "test-key", + const strict = judgeModel({ + model: step({ agent: judgeAgent([{ provider: "stub", model: "stub" }]) }), thresholds: { yes: 0.8 }, - deps: stubDeps(flatFetch(noul(0.75))), + invokeStep: answering(noul(0.75)), }); expect((await judged(question, strict)).outcome).toMatchObject({ pass: false, @@ -545,3 +561,153 @@ describe("stored reports", () => { } }); }); + +describe("judgeModel", () => { + const agent = judgeAgent([{ provider: "stub", model: "stub" }]); + const invokeStep: StepInvoker = async () => ({ output: { decisions: [] } }); + + test("swapping inference sources does not change the judge payload", async () => { + const alpha = [{ provider: "alpha", model: "m1" }]; + const beta = [{ provider: "beta", model: "m2" }]; + const a = stubJev({ type: "noul", noul: 0.9 }, alpha); + const b = stubJev({ type: "noul", noul: 0.9 }, beta); + const question = "Is this reply ready to send to the customer?"; + await judged(question, a.judge); + await judged(question, b.judge); + expect(a.bodies).toHaveLength(1); + expect(a.bodies[0]).toEqual(b.bodies[0]); + expect(a.seen[0]).toEqual(alpha); + expect(b.seen[0]).toEqual(beta); + expect(a.seen[0]).not.toEqual(b.seen[0]); + }); + + test("string and ask checks throw before cases without judge.invoke", async () => { + await expect( + runEvaluation({ + name: "missing-judge", + task: (input: string) => input, + data: "/no/such/cases.jsonl", + scorers: { ready: "Ready?" }, + }), + ).rejects.toThrow("judge.invoke"); + await expect( + runEvaluation({ + name: "thresholds-only", + task: (input: string) => input, + data: "/no/such/cases.jsonl", + scorers: { ready: ask("Ready?", { options: { a: "a" }, pass: "a" }) }, + judge: jev({ thresholds: { yes: 0.9 } }), + }), + ).rejects.toThrow("judge.invoke"); + const report = await runEvaluation({ + name: "code-only", + task: (input: string) => input, + data: [{ input: "a", expectedOutput: "a" }], + }); + expect(report.ok).toBe(true); + }); + + test("rejects a model that is not one plain step", () => { + expect(() => judgeModel({ model: step({ agent, triggers: 2 }), invokeStep })).toThrow( + "triggers must be 1", + ); + expect(() => + judgeModel({ + model: step({ agent, input: { from: "steps.other.output" } }), + invokeStep, + }), + ).toThrow("must not set input"); + expect(() => + judgeModel({ + model: defineWorkflow({ + id: "custom-input", + steps: { only: step({ agent, input: { literal: "fixed" } }) }, + }), + invokeStep, + }), + ).toThrow("must not set input"); + expect(() => + judgeModel({ + model: step({ agent, inference: { literal: null } }), + invokeStep, + }), + ).toThrow("must not set inference"); + const two = defineWorkflow({ + id: "two", + trigger: { type: "manual" }, + steps: { + a: step({ agent }), + b: step({ agent, after: ["a"] }), + }, + }); + expect(() => judgeModel({ model: two, invokeStep })).toThrow("exactly one step"); + const mapped = defineWorkflow({ + id: "mapped", + steps: { + fan: map({ over: { from: "trigger.payload" }, step: step({ agent }) }), + }, + }); + expect(() => judgeModel({ model: mapped, invokeStep })).toThrow("must be a step"); + const child = defineWorkflow({ + id: "parent", + steps: { + inner: childWorkflow({ definition: defineWorkflow({ id: "inner", agent }) }), + }, + }); + expect(() => judgeModel({ model: child, invokeStep })).toThrow("must be a step"); + const empty = { + id: "empty", + triggers: [{ type: "manual" }], + steps: {}, + stepOrder: [], + } as WorkflowDefinition; + expect(() => judgeModel({ model: empty, invokeStep })).toThrow("no steps"); + const gated = { + id: "gated", + triggers: [{ type: "manual" }], + steps: { g: { id: "g", kind: "gate" } }, + stepOrder: ["g"], + } as unknown as WorkflowDefinition; + expect(() => judgeModel({ model: gated, invokeStep })).toThrow("must be a step"); + const twoTriggers = { + id: "twice", + triggers: [{ type: "manual" }, { type: "manual" }], + steps: { only: { id: "only", kind: "step", agent } }, + stepOrder: ["only"], + } as unknown as WorkflowDefinition; + expect(() => judgeModel({ model: twoTriggers, invokeStep })).toThrow("exactly one trigger"); + expect(() => + judgeModel({ + model: step({ agent }), + invokeStep: undefined as unknown as StepInvoker, + }), + ).toThrow("needs invokeStep"); + }); + + test("a one-step workflow is invoked once with the case payload", async () => { + const question = "Is this reply ready to send to the customer?"; + const payload = { + state: { input: "refund?", output: "reply to refund?", expectedOutput: "yes" }, + questions: { judged: { type: "noul", instructions: question } }, + }; + const forms = [ + step({ agent }), + defineWorkflow({ id: "plural", steps: { only: step({ agent }) } }), + defineWorkflow({ + id: "explicit", + steps: { only: step({ agent, input: { from: "trigger.payload" } }) }, + }), + defineWorkflow({ id: "singular", agent }), + ]; + for (const model of forms) { + const bodies: unknown[] = []; + const seen: StepInvoker = async (request) => { + bodies.push(request.input); + return { output: { decisions: [{ id: "judged", type: "noul", noul: 0.9 }] } }; + }; + const { outcome } = await judged(question, judgeModel({ model, invokeStep: seen })); + expect(outcome?.pass).toBe(true); + expect(bodies).toEqual([payload]); + } + }); +}); diff --git a/examples/ollama.ts b/examples/ollama.ts index 3db5511..f3738c4 100644 --- a/examples/ollama.ts +++ b/examples/ollama.ts @@ -1,12 +1,59 @@ -// Shared local-Ollama wiring for the examples. The dummy key works -// around @corbits/system-one fail-closing on keyless custom endpoints. -import { jev } from "../src/index.js"; +// Shared local-Ollama wiring for the examples. The judge is a workflow step; +// this file's invoker turns the step input into a Jev reply. No provider SDK. +import { defineAgent } from "@intx/agent"; +import { step } from "@intx/workflow"; -export const localJev = jev({ - endpoint: "http://localhost:11434/v1/systemone", - model: "clef-flash:9b", - apiKey: "ollama-local", +import { judgeModel } from "../src/workflow.js"; + +const JUDGE_URL = "http://localhost:11434/api/chat"; + +export const localJev = judgeModel({ + model: step({ + agent: defineAgent({ + id: "local-jev", + systemPrompt: "Answer the Jev questions in the input.", + tools: [], + capabilities: [], + inference: { sources: [{ provider: "ollama", model: "clef-flash:9b" }] }, + }), + }), timeout: "180s", + invokeStep: async ({ input, agent }) => { + const model = agent.inference.sources[0]?.model ?? "clef-flash:9b"; + const res = await fetch(JUDGE_URL, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + model, + stream: false, + messages: [ + { + role: "user", + content: `Reply with JSON {"decisions":[...]} for this Jev payload:\n${JSON.stringify(input)}`, + }, + ], + }), + }); + if (!res.ok) { + return { output: { fallback: true, reason: "http-error", detail: `HTTP ${res.status}` } }; + } + const data = (await res.json()) as { message?: { content?: unknown } }; + const content = data.message?.content; + if (typeof content !== "string") { + return { output: { fallback: true, reason: "parse-error", detail: "no reply text" } }; + } + try { + return { output: JSON.parse(content) as unknown }; + } catch (cause) { + return { + output: { + fallback: true, + reason: "parse-error", + detail: cause instanceof Error ? cause.message : String(cause), + }, + }; + } + }, }); export async function ollamaGenerate(model: string, prompt: string): Promise { diff --git a/src/calibrate.ts b/src/calibrate.ts index 6453f9d..f55211f 100644 --- a/src/calibrate.ts +++ b/src/calibrate.ts @@ -22,7 +22,7 @@ export type CalibrateOptions = { checks: Record; judge: Judge; concurrency?: number; - /** Bounds each case run. The Jev call is bounded by `judge.timeout`. */ + /** Bounds each case run. The judge step is bounded by its own timeout. */ timeout?: Duration; }; @@ -140,7 +140,6 @@ export async function calibrate( const rows: CalibrationRow[] = []; for (const threshold of GRID) { const judge: Judge = { - config: {}, thresholds: isQuestion ? { yes: base.yes, no: base.no, sure: threshold } : { yes: threshold, no: r3(1 - threshold), sure: base.sure }, diff --git a/src/evaluation.ts b/src/evaluation.ts index e88d040..26666aa 100644 --- a/src/evaluation.ts +++ b/src/evaluation.ts @@ -1,6 +1,6 @@ import { matches } from "./checks.js"; import { fromJsonl } from "./datasets.js"; -import { checkThresholds, jev, type Ask, type Judge, type Sentence } from "./jev.js"; +import { checkThresholds, type Ask, type Judge, type Sentence } from "./jev.js"; import { flips, fromReport, @@ -47,7 +47,7 @@ export type Evaluation = { timeout?: Duration; /** Stamped into the experiment, e.g. `{ model }`. */ about?: Record; - /** Jev for string and `ask` scorers. Defaults to `jev()`. */ + /** Jev for string and `ask` scorers. Those checks need `judge.invoke` from `judgeModel`. */ judge?: Judge; /** Skip the console summary. */ quiet?: boolean; @@ -271,11 +271,16 @@ export async function runEvaluation(spec: Evaluation): Promise let judge: Judge | undefined = spec.judge; const defined = spec.scorers ?? { matches: matches() as Scorer }; const { code, jev: jevChecks } = toCheckParts(name, defined); + if (Object.keys(jevChecks).length > 0 && judge?.invoke === undefined) { + throw new Error( + `evaluation "${name}": string and ask checks need judge.invoke from judgeModel`, + ); + } const checkNames = Object.keys(defined); const jevGroup = - Object.keys(jevChecks).length === 0 + Object.keys(jevChecks).length === 0 || judge === undefined ? undefined - : { judge: (judge ??= jev()), checks: jevChecks }; + : { judge, checks: jevChecks }; const allowances = toAllowances(name, spec.allow ?? {}, checkNames); const cases = named( name, diff --git a/src/jev.ts b/src/jev.ts index d07f4f4..107a081 100644 --- a/src/jev.ts +++ b/src/jev.ts @@ -1,29 +1,12 @@ -import { - evaluate, - type Decision, - type EndpointConfig, - type EvaluateConfig, - type EvaluateDeps, - type Question, -} from "@corbits/system-one"; - -import { toMs, type Duration } from "./timeout.js"; +import type { Decision, Question } from "@corbits/system-one"; + import type { Outcome, Run } from "./types.js"; export type JevOptions = { - /** "official" (default), "gateway", or the URL of a custom endpoint. */ - endpoint?: string; - model?: string; - /** Defaults to the endpoint's environment key, as `@corbits/system-one` reads it. */ - apiKey?: string; - /** Bounds one Jev call. `@corbits/system-one` defaults to 1500ms. */ - timeout?: Duration; /** Pass bars for every Jev check through this judge. */ thresholds?: Thresholds; /** Consulted when a decision lands unsure. `sure` doubles as the escalation gate. */ escalate?: Escalation; - /** Replaces `fetch` and the scheduler, e.g. to stub Jev in tests. */ - deps?: EvaluateDeps; }; /** Yes/no passes at p >= `yes` and fails at p <= `no`; `ask` is unsure under `sure`. */ @@ -34,11 +17,23 @@ export type Resolved = { yes: number; no: number; sure: number }; /** The one documented set of numbers, tuned per workload with `calibrate`. */ export const DEFAULTS: Resolved = { yes: 0.7, no: 0.3, sure: 0.6 }; +/** What one judge call sends. Questions are id-keyed; boolean checks are `noul`. */ +export type JudgePayload = { + readonly state: Record; + readonly questions: Readonly< + Record + >; +}; + export type Judge = { - readonly config: EvaluateConfig; readonly thresholds?: Resolved; readonly escalate?: Escalation; - readonly deps?: EvaluateDeps; + /** + * One call per case attempt. `judgeModel` sets this; `jev()` alone does not, + * so it is not a model. + * Resolves to `{ decisions }` or `{ fallback: true, reason, detail? }`. + */ + readonly invoke?: (payload: JudgePayload) => Promise; }; /** @@ -68,27 +63,12 @@ function resolve(label: string, raw: Thresholds): Resolved { return out; } -function toEndpoint(endpoint?: string, model?: string): EndpointConfig | undefined { - const named = model === undefined ? {} : { model }; - if (endpoint === undefined) - return model === undefined ? undefined : { kind: "official", ...named }; - if (endpoint === "official" || endpoint === "gateway") return { kind: endpoint, ...named }; - return { kind: "custom", url: endpoint, ...named }; -} - export function jev(options: JevOptions = {}): Judge { - const config: EvaluateConfig = {}; - const endpoint = toEndpoint(options.endpoint, options.model); - if (endpoint !== undefined) config.endpoint = endpoint; - if (options.apiKey !== undefined) config.apiKey = options.apiKey; - if (options.timeout !== undefined) config.timeoutMs = toMs(options.timeout, "jev timeout"); const thresholds = options.thresholds === undefined ? undefined : resolve("thresholds", options.thresholds); return { - config, ...(thresholds === undefined ? {} : { thresholds }), ...(options.escalate === undefined ? {} : { escalate: options.escalate }), - ...(options.deps === undefined ? {} : { deps: options.deps }), }; } @@ -333,6 +313,59 @@ function toState(run: Run): Record { return state; } +function wireQuestions(questions: readonly Question[]): JudgePayload["questions"] { + const wire: Record< + string, + { type: string; instructions: Question["instructions"]; criteria?: unknown } + > = {}; + for (const question of questions) { + const mapped: { type: string; instructions: Question["instructions"]; criteria?: unknown } = { + type: question.type === "boolean" ? "noul" : question.type, + instructions: question.instructions, + }; + if ("criteria" in question && question.criteria !== undefined) + mapped.criteria = question.criteria; + wire[question.id] = mapped; + } + return wire; +} + +type JudgeReply = { fallback: true; reason: string; detail?: string } | { decisions: Decision[] }; + +function readReply(out: unknown): JudgeReply | undefined { + if (typeof out !== "object" || out === null) return undefined; + const record = out as Record; + if (record["fallback"] === true) { + const reason = typeof record["reason"] === "string" ? record["reason"] : "parse-error"; + const detail = typeof record["detail"] === "string" ? record["detail"] : undefined; + return { fallback: true, reason, ...(detail === undefined ? {} : { detail }) }; + } + if (Array.isArray(record["decisions"])) return { decisions: record["decisions"] as Decision[] }; + return undefined; +} + +/** Fail closed: a thrown invoke is a transport fallback, same as a typed one. */ +async function callJudge(judge: Judge, questions: Question[], run: Run): Promise { + if (judge.invoke === undefined) throw new Error("jev: judge has no invoke"); + let out: unknown; + try { + out = await judge.invoke({ state: toState(run), questions: wireQuestions(questions) }); + } catch (cause) { + return { + fallback: true, + reason: "network", + detail: cause instanceof Error ? cause.message : String(cause), + }; + } + return ( + readReply(out) ?? { + fallback: true, + reason: "parse-error", + detail: "judge output was not decisions or a fallback", + } + ); +} + /** * Answer every check in one Jev call, keyed by check name. Fail-closed: * transport fallbacks, thrown errors and missing decisions all fail — a @@ -346,16 +379,13 @@ export async function askJevBatch( ): Promise> { const names = Object.keys(checks); const questions = names.map((id) => toQuestion(id, checks[id] as Ask | Sentence | string)); - let out; + let out: JudgeReply; try { - out = await evaluate( - { state: toState(run), questions, config: judge.config }, - judge.deps === undefined ? undefined : { deps: judge.deps }, - ); + out = await callJudge(judge, questions, run); } catch (cause) { return failedAll(names, `jev threw: ${cause instanceof Error ? cause.message : String(cause)}`); } - if (out.fallback) { + if ("fallback" in out) { return failedAll(names, `jev fallback (${out.reason}): ${out.detail ?? "no detail"}`); } const decisions = new Map(out.decisions.map((decision) => [decision.id, decision])); @@ -387,15 +417,13 @@ export async function decideBatch( run: Run, ): Promise> { const names = Object.keys(checks); - const out = await evaluate( - { - state: toState(run), - questions: names.map((id) => toQuestion(id, checks[id] as Ask | Sentence | string)), - config: judge.config, - }, - judge.deps === undefined ? undefined : { deps: judge.deps }, + const out = await callJudge( + judge, + names.map((id) => toQuestion(id, checks[id] as Ask | Sentence | string)), + run, ); - if (out.fallback) throw new Error(`jev fallback (${out.reason}): ${out.detail ?? "no detail"}`); + if ("fallback" in out) + throw new Error(`jev fallback (${out.reason}): ${out.detail ?? "no detail"}`); return Object.fromEntries(out.decisions.map((decision) => [decision.id, decision])); } diff --git a/src/workflow.ts b/src/workflow.ts index 55e5c20..05d5a22 100644 --- a/src/workflow.ts +++ b/src/workflow.ts @@ -1,13 +1,17 @@ import { + defineWorkflow, runLocal, type RunLocalOptions, type RunResult, type StepInvoker, + type StepPrimitive, type WorkflowAuthorizeFn, type WorkflowDefinition, } from "@intx/workflow"; import { TRACE, type Traced } from "./runs.js"; +import { jev, type Escalation, type Judge, type Thresholds } from "./jev.js"; +import { toMs, type Duration } from "./timeout.js"; import type { Step } from "./types.js"; const allowAll: WorkflowAuthorizeFn = async () => ({ @@ -82,3 +86,128 @@ export function workflow( }, }; } + +export type JudgeModelOptions = { + /** A step, or a workflow whose only primitive is that step. */ + model: StepPrimitive | WorkflowDefinition; + /** Runs the judge step. Required: a definition alone cannot call a model. */ + invokeStep: StepInvoker; + thresholds?: Thresholds; + escalate?: Escalation; + /** Bounds the judge step. */ + timeout?: Duration; +}; + +function isWorkflow(model: StepPrimitive | WorkflowDefinition): model is WorkflowDefinition { + return "steps" in model && "stepOrder" in model; +} + +/** + * `defineWorkflow` stamps a lone step that omits `input` with this selector. + * It is the same binding `judgeModel` forces, so it is not a custom input. + */ +function isStampedTriggerInput(input: NonNullable): boolean { + return "from" in input && input.from === "trigger.payload" && Object.keys(input).length === 1; +} + +function assertJudgeStep(step: StepPrimitive): void { + if (step.kind !== "step") throw new Error("judgeModel: the only primitive must be a step"); + if (step.input !== undefined && !isStampedTriggerInput(step.input)) { + throw new Error("judgeModel: step must not set input"); + } + if (step.inference !== undefined) throw new Error("judgeModel: step must not set inference"); + if (step.triggers !== undefined && step.triggers !== 1) { + throw new Error("judgeModel: step triggers must be 1"); + } +} + +function takeJudgeStep(model: StepPrimitive | WorkflowDefinition): StepPrimitive { + if (!isWorkflow(model)) { + if (!("kind" in model) || model.kind !== "step") { + throw new Error("judgeModel: model must be a step or a one-step workflow"); + } + assertJudgeStep(model); + return model; + } + if (model.triggers.length !== 1) + throw new Error("judgeModel: workflow must have exactly one trigger"); + if (model.stepOrder.length === 0) throw new Error("judgeModel: workflow has no steps"); + if (model.stepOrder.length !== 1) + throw new Error("judgeModel: workflow must have exactly one step"); + const id = model.stepOrder[0] as string; + const primitive = model.steps[id]; + if (primitive === undefined || primitive.kind !== "step") { + throw new Error("judgeModel: the only primitive must be a step"); + } + assertJudgeStep(primitive); + return primitive; +} + +/** + * A Jev judge whose model is one workflow step. The step input is forced to + * the trigger payload `{ state, questions }`, so the agent's inference sources + * never enter the request. One `invokeStep` call per case attempt. The step + * output is the judge reply: `{ decisions }` or `{ fallback: true, reason }`. + */ +export function judgeModel(options: JudgeModelOptions): Judge { + if (options.invokeStep === undefined) throw new Error("judgeModel needs invokeStep"); + const primitive = takeJudgeStep(options.model); + const timeoutMs = + options.timeout === undefined ? undefined : toMs(options.timeout, "judgeModel timeout"); + const forced: StepPrimitive = { + ...primitive, + id: "", + input: { from: "trigger.payload" }, + ...(timeoutMs === undefined ? {} : { timeout: timeoutMs }), + }; + const definition = defineWorkflow({ + id: "jev-judge", + trigger: { type: "manual" }, + steps: { judge: forced }, + }); + const base = jev({ + ...(options.thresholds === undefined ? {} : { thresholds: options.thresholds }), + ...(options.escalate === undefined ? {} : { escalate: options.escalate }), + }); + return { + ...base, + invoke: async (payload) => { + let calls = 0; + const run = runLocal(definition, { + authorize: allowAll, + hasUpstreamSignalResolver: true, + triggerPayload: payload, + invokeStep: async (request) => { + calls += 1; + if (calls > 1) throw new Error("judgeModel: more than one invoke"); + try { + const result = await options.invokeStep(request); + if ("suspend" in result) { + return { + output: { fallback: true, reason: "network", detail: "judge step suspended" }, + }; + } + return result; + } catch (cause) { + return { + output: { + fallback: true, + reason: "network", + detail: cause instanceof Error ? cause.message : String(cause), + }, + }; + } + }, + }); + const finished = await run.complete; + if (finished.terminalStatus !== "completed") { + return { + fallback: true, + reason: "network", + detail: `judge step ended ${finished.terminalStatus}`, + }; + } + return finished.outputs["judge"]; + }, + }; +}