diff --git a/api/traces.api.md b/api/traces.api.md index 850160d36..af7f76ed9 100644 --- a/api/traces.api.md +++ b/api/traces.api.md @@ -7,6 +7,24 @@ // @public export const DEFAULT_TRACE_DIR = ".lousho/traces"; +// @public +export interface EvaluatorTrace { + agent_id: string | null; + duration_ms: number | null; + end: number | null; + error_message: string | null; + input: string | null; + name: string | null; + observations: ObservationRow[]; + output: string | null; + start: number; + status: string | null; + trace_id: string; +} + +// @public +export function evaluatorTraceFrom(traceId: string, observations: readonly ObservationRow[]): EvaluatorTrace | undefined; + // @public export function fanOutExporters(exporters: readonly (TraceExporter | undefined | null)[]): TraceExporter | undefined; @@ -27,6 +45,32 @@ export interface ListTracesOptions { limit?: number; } +// @public +export interface ObservationRow { + agent_id: string | null; + attrs: Record | null; + cost_partial: boolean | null; + cost_usd: number | null; + end: number | null; + input: string | null; + kind: string; + model: string | null; + name: string | null; + obs_id: string; + output: string | null; + parent_obs_id: string | null; + session_id: string | null; + start: number; + status: string | null; + trace_id: string; + truncated: boolean; + usage_cached_tokens: number | null; + usage_estimated: boolean | null; + usage_input_tokens: number | null; + usage_output_tokens: number | null; + usage_reasoning_tokens: number | null; +} + // @public export function observeExporter(options?: ObserveExporterOptions): ObserveReporter; diff --git a/apps/agent-forge/e2e/revamp.spec.ts b/apps/agent-forge/e2e/revamp.spec.ts index ad4508830..c8a6da1b8 100644 --- a/apps/agent-forge/e2e/revamp.spec.ts +++ b/apps/agent-forge/e2e/revamp.spec.ts @@ -125,6 +125,8 @@ const FILTERS: Record = { scores: '?eval=e2e-eval&agent=orders-bot&source=eval&range=7d&open=e2e-eval', // The only URL state Datasets keeps is the server-side page (`offset`). datasets: '?offset=50', + // Queues pages server-side the same way (`?offset=`). + queues: '?offset=50', evals: '', prompts: '?diff=1,2', hub: '?agent=triage-bot&scope=fleet', @@ -195,6 +197,9 @@ const CONTROLS: Record Promise> = { '/evals/datasets/:name': async (main) => { await expectPreviousPage(main); }, + '/evals/queues': async (main) => { + await expectPreviousPage(main); + }, '/prompts/:name': async (main) => { await expect(main.getByRole('combobox', { name: 'Compare version A' })).toHaveValue('1'); await expect(main.getByRole('combobox', { name: 'Compare version B' })).toHaveValue('2'); @@ -218,6 +223,7 @@ const NO_CONTROLS = new Set([ '/threads/:sessionId', // a thread's turns; the list's filters only ride along in the URL '/evals/:section', // evaluators / annotation placeholder screen '/evals/:section/:id', // evaluators / annotation placeholder screen + '/evals/queues/:id', // one queue's review pane; no URL-backed controls beyond the id '/prompts', // the registry list has no filters '/settings', // no URL state ]); diff --git a/apps/agent-forge/playwright.config.ts b/apps/agent-forge/playwright.config.ts index e9e83eff6..633084168 100644 --- a/apps/agent-forge/playwright.config.ts +++ b/apps/agent-forge/playwright.config.ts @@ -54,6 +54,8 @@ export default defineConfig({ BASE_DIR: baseDir, LOUSHO_STUDIO_TOKEN: E2E_STUDIO_TOKEN, LOUSHO_OBSERVE_URL: E2E_OBSERVE_URL, + // No daemon here: keep the LLM-judge loop (server/judge.ts) from polling it. + LOUSHO_STUDIO_JUDGE: '0', }, url: `http://127.0.0.1:${port}/health`, reuseExistingServer: false, diff --git a/apps/agent-forge/server/__tests__/judge.test.ts b/apps/agent-forge/server/__tests__/judge.test.ts new file mode 100644 index 000000000..23da560ee --- /dev/null +++ b/apps/agent-forge/server/__tests__/judge.test.ts @@ -0,0 +1,355 @@ +/** + * LLM-judge evaluators, the Studio half (judge.ts), against a real + * in-process observe daemon: the loop pulls judge jobs, runs a mock judge + * and posts the score back; a daemon that is down never breaks the loop; + * jobs wait in the daemon while no loop runs; test-on-trace runs a judge + * rule here as a dry run. Plus the judge-reply parser. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import * as http from 'node:http'; +import { once } from 'node:events'; +import request from 'supertest'; +import { startObserveDaemon, type ObserveDaemon } from '../../../../src/observe/daemon'; +import { ingestBatch, listScores } from '../../../../src/observe/store'; +import { createApp } from '../app'; +import { judgePrompt, parseJudgeReply, startJudgeLoop, type JudgeFn, type JudgeLoop, type JudgeLoopOptions } from '../judge'; +import { RunManager } from '../runRegistry'; +import { FileCheckpointStore } from '../checkpointStore'; +import { FileApprovalStore } from '../approvalStore'; +import { SecretsStore } from '../secretsStore'; +import { createFsAgentStore } from '../../src/persistence/fsAgentStore'; + +const T0 = 1_700_000_000_000; + +describe('judge loop and test-on-trace', () => { + let baseDir: string; + let daemon: ObserveDaemon; + let daemonUrl: string; + const loops: JudgeLoop[] = []; + const prevUrl = process.env.LOUSHO_OBSERVE_URL; + + beforeEach(async () => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-judge-test-')); + daemon = await startObserveDaemon({ dbPath: ':memory:', port: 0, sweepIntervalMs: 0, evaluators: { isolateAvailable: () => true } }); + daemonUrl = `http://127.0.0.1:${daemon.port}`; + process.env.LOUSHO_OBSERVE_URL = daemonUrl; + }); + + afterEach(async () => { + while (loops.length > 0) loops.pop()?.stop(); + await daemon.close(); + if (prevUrl === undefined) delete process.env.LOUSHO_OBSERVE_URL; + else process.env.LOUSHO_OBSERVE_URL = prevUrl; + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + const seedTrace = (traceId: string): void => { + ingestBatch(daemon.db, [ + { + type: 'observation', + data: { obs_id: `${traceId}-root`, trace_id: traceId, agent_id: 'support', kind: 'span', name: 'invoke_agent refund-bot', start: T0, end: T0 + 50, status: 'ok', input: 'refund?', output: 'done' }, + }, + ]); + }; + + /** A judge rule created over HTTP, then one trace whose root ended (so the daemon queues a job). */ + async function judgeRuleWithJob(traceId = 't1'): Promise { + const created = await fetch(`${daemonUrl}/api/v1/evaluators`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ name: 'helpful', kind: 'judge', config: { model: 'mock', rubric: 'Is the refund handled?' } }), + }); + expect(created.status).toBe(201); + seedTrace(traceId); + daemon.evaluators.notifyRootsEnded([traceId]); + await daemon.evaluators.idle(); + return ((await created.json()) as { evaluator: { rule_id: string } }).evaluator.rule_id; + } + + const onlineScores = (traceId: string) => listScores(daemon.db, { trace_id: traceId, source: 'online', limit: 10 }).scores; + + const loop = (judge: JudgeFn, extra: Partial = {}): JudgeLoop => { + const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), intervalMs: 60_000, judge, ...extra }); + loops.push(started); + return started; + }; + + it('pulls a judge job, runs the mock judge and writes its score as an online score', async () => { + await judgeRuleWithJob(); + const seen: unknown[] = []; + const judge: JudgeFn = async (rule, trace) => { + seen.push(rule.config, trace); + return { value: 0.8, comment: 'refund handled' }; + }; + await loop(judge).tick(); + expect(seen[0]).toEqual({ model: 'mock', rubric: 'Is the refund handled?' }); + expect(seen[1]).toMatchObject({ trace_id: 't1', agent_id: 'support', input: 'refund?', output: 'done', duration_ms: 50 }); + expect(onlineScores('t1')).toEqual([expect.objectContaining({ name: 'helpful', value: 0.8, comment: 'refund handled', agent_id: 'support' })]); + expect(daemon.evaluators.stats()).toMatchObject({ judge_queued: 0, judge_leased: 0, judge_scored: 1 }); + }); + + it('a judge that throws scores null with a judge_error comment', async () => { + await judgeRuleWithJob(); + await loop(async () => { + throw new Error('no key for openai'); + }).tick(); + expect(onlineScores('t1')).toEqual([expect.objectContaining({ value: null, comment: 'judge_error: no key for openai' })]); + }); + + it('the default judge turns a missing provider key into a judge_error score', async () => { + const created = await fetch(`${daemonUrl}/api/v1/evaluators`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ name: 'keyless', kind: 'judge', config: { model: 'openai/gpt-4o-mini', rubric: 'r' } }), + }); + expect(created.status).toBe(201); + seedTrace('t1'); + daemon.evaluators.notifyRootsEnded(['t1']); + await daemon.evaluators.idle(); + const prevKey = process.env.OPENAI_API_KEY; + delete process.env.OPENAI_API_KEY; + try { + const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), intervalMs: 60_000 }); + loops.push(started); + await started.tick(); + } finally { + if (prevKey !== undefined) process.env.OPENAI_API_KEY = prevKey; + } + const [score] = onlineScores('t1'); + expect(score).toMatchObject({ name: 'keyless', value: null }); + expect(score.comment).toMatch(/^judge_error: /); + expect((score.comment ?? '').length).toBeLessThanOrEqual('judge_error: '.length + 200); + }); + + it('survives a daemon that is down: no throw, one warning per outage, and it resumes', async () => { + await judgeRuleWithJob(); + // A closed port: bind one, take its number, close it. + const closed = http.createServer(); + closed.listen(0, '127.0.0.1'); + await once(closed, 'listening'); + const closedUrl = `http://127.0.0.1:${(closed.address() as { port: number }).port}`; + await new Promise((resolve) => closed.close(() => resolve())); + + const warnings: string[] = []; + const started = loop(async () => ({ value: 1, comment: null }), { warn: (m) => warnings.push(m) }); + process.env.LOUSHO_OBSERVE_URL = closedUrl; + await started.tick(); + await started.tick(); + expect(warnings).toHaveLength(1); + expect(warnings[0]).toContain('unreachable'); + expect(onlineScores('t1')).toEqual([]); + + process.env.LOUSHO_OBSERVE_URL = daemonUrl; + await started.tick(); + expect(onlineScores('t1')).toHaveLength(1); + }); + + it('normalizes a judge result the daemon would refuse (out of range, long comment)', async () => { + await judgeRuleWithJob('t1'); + await loop(async () => ({ value: 7, comment: 'way out' })).tick(); + expect(onlineScores('t1')).toEqual([expect.objectContaining({ value: null, comment: 'invalid_judge_output' })]); + seedTrace('t2'); + daemon.evaluators.notifyRootsEnded(['t2']); + await daemon.evaluators.idle(); + await loop(async () => ({ value: 0.5, comment: 'y'.repeat(5000) })).tick(); + expect(onlineScores('t2')).toEqual([expect.objectContaining({ value: 0.5, comment: 'y'.repeat(1000) })]); + }); + + it('aborts a judge call at judgeTimeoutMs and scores it judge_error, even when the judge ignores the signal', async () => { + await judgeRuleWithJob(); + let aborted: unknown; + const started = loop( + (_rule, _trace, signal) => + new Promise(() => { + signal.addEventListener('abort', () => (aborted = signal.reason)); + }), + { judgeTimeoutMs: 100 } + ); + const begun = Date.now(); + await started.tick(); + expect(Date.now() - begun).toBeLessThan(2000); + expect(aborted).toBeInstanceOf(Error); + expect(onlineScores('t1')).toEqual([expect.objectContaining({ value: null, comment: 'judge_error: timed out after 100 ms' })]); + }); + + it('does not run a claimed job once its lease has run out locally (it expires and requeues)', async () => { + await judgeRuleWithJob('t1'); + seedTrace('t2'); + daemon.evaluators.notifyRootsEnded(['t2']); + await daemon.evaluators.idle(); + const clock = { now: Date.now() }; + const judged: string[] = []; + const started = loop( + async (_rule, trace) => { + judged.push(trace.trace_id); + clock.now += 60_000; // this call used up the whole lease + return { value: 1, comment: null }; + }, + { leaseMs: 60_000, now: () => clock.now } + ); + await started.tick(); + expect(judged).toEqual(['t1']); + // t1's result posted (the daemon's own lease is still live); t2 was skipped, still leased until it expires. + expect(daemon.evaluators.stats()).toMatchObject({ judge_scored: 1, judge_leased: 1, judge_queued: 0 }); + expect(onlineScores('t2')).toEqual([]); + }); + + it('caps a judge call at what is left of the lease', async () => { + await judgeRuleWithJob(); + const clock = { now: Date.now() }; + let signalled: AbortSignal | undefined; + const started = loop( + (_rule, _trace, signal) => { + signalled = signal; + return new Promise(() => undefined); + }, + // A 1150 ms lease minus the 1000 ms post margin leaves 150 ms (the clock stands still). + { leaseMs: 1150, judgeTimeoutMs: 20_000, now: () => clock.now } + ); + const begun = Date.now(); + await started.tick(); + expect(Date.now() - begun).toBeLessThan(2000); + expect(signalled?.aborted).toBe(true); + expect(onlineScores('t1')).toEqual([expect.objectContaining({ comment: 'judge_error: timed out after 150 ms' })]); + }); + + it('warns once per outage for a 2xx claim answer that is not JSON', async () => { + const warnings: string[] = []; + const notJson = (async () => new Response('proxy', { status: 200, headers: { 'content-type': 'text/html' } })) as typeof fetch; + const started = loop(async () => ({ value: 1, comment: null }), { fetch: notJson, warn: (m) => warnings.push(m) }); + await started.tick(); + await started.tick(); + expect(warnings).toHaveLength(1); + expect(warnings[0]).toContain('without a jobs array'); + }); + + it('without a loop the job stays queued (nothing claims it)', async () => { + await judgeRuleWithJob(); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(daemon.evaluators.stats()).toMatchObject({ judge_queued: 1, judge_leased: 0, judge_scored: 0 }); + expect(onlineScores('t1')).toEqual([]); + }); + + describe('POST /observe/api/v1/evaluators/:id/test', () => { + const appWith = (judge: JudgeFn) => { + const agentStore = createFsAgentStore(baseDir); + const runManager = new RunManager({ + baseDir, + checkpointStore: new FileCheckpointStore(baseDir), + approvalStore: new FileApprovalStore(baseDir), + loadSpec: (id) => agentStore.load(id), + saveSpec: (id, spec) => agentStore.save(id, spec), + }); + return createApp({ agentStore, runManager, baseDir, judge }); + }; + + it('dry-runs a judge rule on the trace with the judge and writes no score', async () => { + const ruleId = await judgeRuleWithJob(); + const seen: unknown[] = []; + const app = appWith(async (_rule, trace) => { + seen.push(trace); + return { value: 0.8, comment: 'looks right' }; + }); + const res = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({ trace_id: 't1' }); + expect(res.status).toBe(200); + expect(res.body).toEqual({ result: { value: 0.8, comment: 'looks right' } }); + expect(seen[0]).toMatchObject({ trace_id: 't1', name: 'invoke_agent refund-bot', status: 'ok' }); + expect(onlineScores('t1')).toEqual([]); + // The dry run left the queued job alone. + expect(daemon.evaluators.stats()).toMatchObject({ judge_queued: 1 }); + }); + + it('forwards a code rule to the daemon unchanged', async () => { + seedTrace('t1'); + const created = await fetch(`${daemonUrl}/api/v1/evaluators`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ name: 'len', kind: 'code', config: { source: '(t) => 1' } }), + }); + const ruleId = ((await created.json()) as { evaluator: { rule_id: string } }).evaluator.rule_id; + const app = appWith(async () => { + throw new Error('the judge must not run for a code rule'); + }); + // The daemon's own validation answers (the body goes through as sent). + const bad = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({ trace_id: 't1', extra: 1 }); + expect(bad.status).toBe(400); + const missing = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({ trace_id: 'nope' }); + expect(missing.status).toBe(404); + }); + + it('passes through 404s for an unknown rule or trace, 400s a bad body, 502s a daemon that is down', async () => { + const ruleId = await judgeRuleWithJob(); + const app = appWith(async () => ({ value: 1, comment: null })); + const unknownRule = await request(app).post('/observe/api/v1/evaluators/nope/test').send({ trace_id: 't1' }); + expect(unknownRule.status).toBe(404); + expect(unknownRule.body).toMatchObject({ code: 'LOUSHO_REQUEST_INVALID' }); + const unknownTrace = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({ trace_id: 'nope' }); + expect(unknownTrace.status).toBe(404); + const badBody = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({}); + expect(badBody.status).toBe(400); + for (const trace_id of ['.', '..', 'a/b', 'a\\b']) { + const unsafe = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({ trace_id }); + expect(unsafe.status).toBe(400); + } + const encodedSlash = await request(app).post('/observe/api/v1/evaluators/a%2Fb/test').send({ trace_id: 't1' }); + expect(encodedSlash.status).toBe(400); + process.env.LOUSHO_OBSERVE_URL = 'http://127.0.0.1:1'; + const down = await request(app).post(`/observe/api/v1/evaluators/${ruleId}/test`).send({ trace_id: 't1' }); + expect(down.status).toBe(502); + }); + }); +}); + +describe('parseJudgeReply', () => { + it('reads a valid JSON reply', () => { + expect(parseJudgeReply('{"score": 0.8, "comment": "good"}')).toEqual({ value: 0.8, comment: 'good' }); + expect(parseJudgeReply('{"score": 1}')).toEqual({ value: 1, comment: null }); + }); + + it('finds the first JSON object inside prose and code fences', () => { + expect(parseJudgeReply('Sure! Here is my grade:\n```json\n{"score": 0, "comment": "wrong {refund}"}\n```\nThanks.')).toEqual({ + value: 0, + comment: 'wrong {refund}', + }); + expect(parseJudgeReply('I think {this} is fine. {"score": 0.5, "comment": "ok"}')).toEqual({ value: 0.5, comment: 'ok' }); + }); + + it('rejects an out-of-range or non-numeric score', () => { + for (const reply of ['{"score": 1.5}', '{"score": -0.1}', '{"score": "0.5"}', '{"comment": "no score"}']) { + expect(parseJudgeReply(reply)).toEqual({ value: null, comment: 'invalid_judge_output' }); + } + }); + + it('rejects garbage', () => { + for (const reply of ['', 'no json here', '{not json', '[0.5]', '{"score": 0.5']) { + expect(parseJudgeReply(reply)).toEqual({ value: null, comment: 'invalid_judge_output' }); + } + }); +}); + +describe('judgePrompt', () => { + it('names each field and truncates long ones to ~8 KB', () => { + const prompt = judgePrompt({ + trace_id: 't', + agent_id: null, + name: 'run', + status: 'error', + start: 0, + end: null, + duration_ms: null, + input: 'x'.repeat(20_000), + output: null, + error_message: 'boom', + observations: [], + }); + expect(prompt).toContain('Name: run'); + expect(prompt).toContain('Status: error'); + expect(prompt).toContain('Output:\n(none)'); + expect(prompt).toContain('Error:\nboom'); + expect(prompt).toContain('[truncated]'); + expect(prompt.length).toBeLessThan(9000); + }); +}); diff --git a/apps/agent-forge/server/app.ts b/apps/agent-forge/server/app.ts index 1d87866ca..03e22cc0f 100644 --- a/apps/agent-forge/server/app.ts +++ b/apps/agent-forge/server/app.ts @@ -19,6 +19,7 @@ * GET /runs/compare?a=&b= -> RunComparisonPayload (compareTrajectories) * GET /observe/** -> proxied to the `lousho observe` daemon's read API (observeProxy.ts) * POST /observe/api/v1/scores -> proxied too (the proxy's write allowlist: human feedback scores) + * POST /observe/api/v1/evaluators/:id/test -> a judge rule's dry run here, a code rule forwarded (judge.ts) * POST /playground/run -> body: PlaygroundRunRequest; streams SSE PlaygroundEvents (playground.ts) * * `WS /agents/:id/stream` is wired separately in wsServer.ts (attached to @@ -63,6 +64,7 @@ import { DEPLOY_ADAPTERS, isDeployAdapter, runDeploy } from './deployRunner'; import { checkApiAccess, checkHostAndOrigin, type StudioAccessOptions } from './auth'; import { registerObserveProxy } from './observeProxy'; import { registerPlaygroundRoute, type PlaygroundOptions } from './playground'; +import { registerJudgeTestRoute, type JudgeFn } from './judge'; export interface CreateAppOptions { agentStore: AgentStore; @@ -94,6 +96,8 @@ export interface CreateAppOptions { access?: StudioAccessOptions; /** Observe reporter overrides for `POST /playground/run` (tests); see playground.ts. */ playgroundObserve?: PlaygroundOptions['observe']; + /** The judge `POST /observe/api/v1/evaluators/:id/test` runs for a judge rule (tests); default the real model call (judge.ts). */ + judge?: JudgeFn; } /** Path prefixes of the API (everything except `/health` and the static client). */ @@ -616,6 +620,7 @@ export function createApp({ staticDir, access, playgroundObserve, + judge, }: CreateAppOptions): Express { const app = express(); // No CORS middleware - the client is same-origin (served by @@ -677,6 +682,8 @@ export function createApp({ // daemon (see observeProxy.ts), plus its small write allowlist (feedback // scores; bodies come from the express.json above) - before the SPA // fallback so a proxied GET never gets index.html. + // Test-on-trace for judge rules runs here (the daemon has no provider keys); before the proxy. + registerJudgeTestRoute(app, { secretsStore: secrets, ...(judge ? { judge } : {}) }); registerObserveProxy(app); registerStaticClient(app, staticDir); diff --git a/apps/agent-forge/server/index.ts b/apps/agent-forge/server/index.ts index 2c1ddee90..724931b30 100644 --- a/apps/agent-forge/server/index.ts +++ b/apps/agent-forge/server/index.ts @@ -17,6 +17,7 @@ import { FileCheckpointStore } from './checkpointStore'; import { FileApprovalStore } from './approvalStore'; import { SecretsStore } from './secretsStore'; import { SettingsStore } from './settingsStore'; +import { startJudgeLoop, type JudgeLoopOptions } from './judge'; import { STUDIO_ALLOWED_ORIGINS_ENV, STUDIO_TOKEN_ENV, allowedHostsFor, mintStudioToken } from './auth'; export interface StudioServerHandle { @@ -59,6 +60,15 @@ export interface StartStudioServerOptions { * test that must not reach a daemon). */ observe?: boolean; + /** + * The LLM-judge loop (judge.ts) that pulls judge-evaluator jobs from the + * observe daemon and runs them with this server's provider keys. On by + * default unless `observe` is `false`; `false` turns it off, an object + * overrides its options (`intervalMs`, `leaseMs`, `judgeTimeoutMs`, + * `fetch`, `judge`, `warn`). Launched directly, `LOUSHO_STUDIO_JUDGE=0` + * turns it off (the e2e server does). + */ + judge?: false | Partial>; } async function listen(server: http.Server, port: number, host: string): Promise { @@ -114,6 +124,8 @@ export async function startStudioServer( attachWebSocketServer(server, runManager, access); await listen(server, port, host); + const judgeOption = options.judge ?? (options.observe === false ? false : {}); + const judgeLoop = judgeOption === false ? undefined : startJudgeLoop({ ...judgeOption, secretsStore }); const boundPort = (server.address() as { port: number }).port; const urlHost = host === '0.0.0.0' || host === '::' ? '127.0.0.1' : host.includes(':') ? `[${host}]` : host; @@ -123,7 +135,10 @@ export async function startStudioServer( runManager, token, url: `http://${urlHost}:${boundPort}/?token=${encodeURIComponent(token)}`, - close: () => new Promise((resolve, reject) => server.close((err) => (err ? reject(err) : resolve()))), + close: () => { + judgeLoop?.stop(); + return new Promise((resolve, reject) => server.close((err) => (err ? reject(err) : resolve()))); + }, }; } @@ -149,7 +164,8 @@ if (isMainModule) { // default `apps/agent-forge/dist` lookup in prod mode. const staticDir = process.env.NO_STATIC ? false : (process.env.STATIC_DIR ?? undefined); - startStudioServer({ port, host, baseDir, staticDir }) + const judge = process.env.LOUSHO_STUDIO_JUDGE === '0' ? false : undefined; + startStudioServer({ port, host, baseDir, staticDir, judge }) .then((handle) => { console.log(`[lousho studio] API server listening on http://${host ?? '127.0.0.1'}:${handle.port}`); // `lousho studio` mints the token itself and prints this URL; only print it when launched directly. diff --git a/apps/agent-forge/server/judge.ts b/apps/agent-forge/server/judge.ts new file mode 100644 index 000000000..5a7370361 --- /dev/null +++ b/apps/agent-forge/server/judge.ts @@ -0,0 +1,399 @@ +/** + * LLM-judge online evaluators, the Studio half. The observe daemon holds no + * provider keys, so it queues a job per (judge rule, trace) and this server + * pulls them (`POST /api/v1/evaluators/judge-jobs/claim`), runs each with + * the keys it already has (secretsStore.ts: a stored key beats the env var), + * and posts the score back (`POST .../judge-jobs/:job_id/result`). Judges + * therefore run only while Studio is up; until then the jobs wait in the + * daemon's bounded queue (500), and a job whose lease runs out is requeued. + * + * - {@link startJudgeLoop}: every `intervalMs` (default 2000) claim up to 5 + * jobs from `observeTarget()` for `leaseMs` (default 60 s; with + * `LOUSHO_OBSERVE_TOKEN` as the bearer when set), run them one at a time, + * post each result. Each judge call is capped at `judgeTimeoutMs` (default + * 20 s) and at what is left of the lease; a job whose lease has run out + * locally is not run (it expires and requeues). Ticks never overlap. A daemon that is down or answers non-2xx only skips the tick, + * logged once per outage. + * - The judge itself ({@link modelJudge}) is one `provider.generate()` call, + * with the provider resolved exactly as the playground resolves it + * (`buildAgentFromSpec`). It goes straight to the provider, not through + * AgentExecutor, so it makes no spans and reports nothing to the daemon: a + * judge trace would itself trigger evaluators. + * - {@link registerJudgeTestRoute}: `POST /observe/api/v1/evaluators/:id/test` + * handled here (the daemon answers 409 for a judge rule): a judge rule is + * dry-run on the trace with the same judge; a code rule is forwarded to the + * daemon unchanged. + */ +import type { Express, Request, Response } from 'express'; +import type { AgentSpec } from '@lousho/build-ai-agent'; +import { evaluatorTraceFrom, type EvaluatorTrace, type ObservationRow } from '@lousho/build-ai-agent/traces'; +import { buildAgentFromSpec } from './buildAgent'; +import { observeTarget } from './observeProxy'; +import { providerOf } from './playground'; +import type { SecretsStore } from './secretsStore'; + +/** A judge rule as the daemon serves it (`EvaluatorRuleRow` with `kind: 'judge'`). */ +export interface JudgeRule { + rule_id: string; + name: string; + kind: 'judge'; + config: { model: string; rubric: string }; +} + +/** What a judge run produced: the score's value in [0, 1] (or null) and its comment. */ +export interface JudgeResult { + value: number | null; + comment: string | null; +} + +/** + * Runs one judge rule on one trace; the default is {@link modelJudge}. + * `signal` aborts when the call's time is up (the result is then a + * `judge_error` whether or not the judge honours it). + */ +export type JudgeFn = (rule: JudgeRule, trace: EvaluatorTrace, signal: AbortSignal) => Promise; + +export interface JudgeLoopOptions { + secretsStore: SecretsStore; + /** Pause between ticks; default 2000 ms. */ + intervalMs?: number; + /** The lease asked for on each claim (`lease_ms`); default 60000 ms. */ + leaseMs?: number; + /** Cap on one judge call (also cut to what is left of the job's lease); default 20000 ms. */ + judgeTimeoutMs?: number; + /** The clock lease deadlines run on (tests); default `Date.now`. */ + now?: () => number; + /** The fetch daemon calls go through (tests). */ + fetch?: typeof fetch; + /** The judge; default {@link modelJudge} over `secretsStore`. */ + judge?: JudgeFn; + /** Where the once-per-outage note goes; default `console.warn`. */ + warn?: (message: string) => void; +} + +export interface JudgeLoop { + stop(): void; + /** Run one tick now, or wait for the one in flight (tests). */ + tick(): Promise; +} + +const CLAIM_MAX = 5; +const LEASE_MS_DEFAULT = 60_000; +const JUDGE_TIMEOUT_MS_DEFAULT = 20_000; +/** Time kept back from a lease to post the result before it runs out. */ +const POST_MARGIN_MS = 1000; +const DAEMON_TIMEOUT_MS = 8000; +const FIELD_LIMIT = 8 * 1024; +const MAX_COMMENT = 1000; +const MAX_ERROR = 200; +const INVALID_OUTPUT: JudgeResult = { value: null, comment: 'invalid_judge_output' }; + +const GRADER_INSTRUCTION = + 'You are an evaluator grading one run of an AI agent against a rubric. Read the rubric and the run, then reply with ' + + 'only a JSON object: {"score": , "comment": ""}.\n\nRubric:\n'; + +const clip = (text: string | null): string => + text === null ? '(none)' : text.length > FIELD_LIMIT ? `${text.slice(0, FIELD_LIMIT)}... [truncated]` : text; + +/** The user message a judge sees: the trace's name, status, input, output and error. */ +export function judgePrompt(trace: EvaluatorTrace): string { + return [ + `Name: ${clip(trace.name)}`, + `Status: ${clip(trace.status)}`, + `Input:\n${clip(trace.input)}`, + `Output:\n${clip(trace.output)}`, + `Error:\n${clip(trace.error_message)}`, + ].join('\n\n'); +} + +/** How many `{` (and, per `{`, how many `}`) the reply scan tries before it gives up on a long prose reply. */ +const MAX_BRACES = 50; + +/** A JSON object, or `undefined` when `slice` is not one. */ +function objectOf(slice: string): Record | undefined { + try { + const parsed: unknown = JSON.parse(slice); + return parsed !== null && typeof parsed === 'object' && !Array.isArray(parsed) ? (parsed as Record) : undefined; + } catch { + return undefined; + } +} + +/** Indexes of `char` in `text` from `from` on, at most {@link MAX_BRACES}. */ +function indexesOf(text: string, char: string, from = 0): number[] { + const found: number[] = []; + for (let i = text.indexOf(char, from); i !== -1 && found.length < MAX_BRACES; i = text.indexOf(char, i + 1)) found.push(i); + return found; +} + +/** + * The first JSON object in `text`: from the earliest `{`, the shortest + * `{...}` that parses (so braces inside strings or prose are tolerated). + */ +function firstJsonObject(text: string): Record | undefined { + for (const start of indexesOf(text, '{')) { + for (const end of indexesOf(text, '}', start)) { + const object = objectOf(text.slice(start, end + 1)); + if (object !== undefined) return object; + } + } + return undefined; +} + +/** + * A judge model's reply as a result: the first JSON object in it, which must + * carry a `score` in [0, 1] (`comment` optional). Anything else is + * `{ value: null, comment: 'invalid_judge_output' }`. + */ +export function parseJudgeReply(text: string): JudgeResult { + const reply = firstJsonObject(text); + const score = reply?.score; + if (typeof score !== 'number' || !Number.isFinite(score) || score < 0 || score > 1) return INVALID_OUTPUT; + const comment = reply?.comment; + return { value: score, comment: typeof comment === 'string' ? comment.slice(0, MAX_COMMENT) : null }; +} + +const judgeError = (error: unknown): JudgeResult => ({ + value: null, + comment: `judge_error: ${(error instanceof Error ? error.message : String(error)).slice(0, MAX_ERROR)}`, +}); + +/** The real judge: one model call with the rule's `provider/model`, resolved like the playground's. */ +function modelJudge(secretsStore: SecretsStore): JudgeFn { + return async (rule, trace, signal) => { + try { + const spec: AgentSpec = { name: 'judge', prompt: '', provider: providerOf(rule.config.model) }; + const { provider } = buildAgentFromSpec(spec, 'judge', undefined, { secretsStore }); + const reply = await provider.generate({ + messages: [ + { role: 'system', content: GRADER_INSTRUCTION + rule.config.rubric }, + { role: 'user', content: judgePrompt(trace) }, + ], + temperature: 0, + signal, + }); + return parseJudgeReply(reply.text); + } catch (error) { + // Unknown provider, missing key, or the call failed. + return judgeError(error); + } + }; +} + +/** A result the daemon accepts: a value in [0, 1] or null, a comment of at most 1000 chars. */ +function acceptable(result: JudgeResult): JudgeResult { + const { value, comment } = result; + if (value !== null && !(typeof value === 'number' && Number.isFinite(value) && value >= 0 && value <= 1)) return INVALID_OUTPUT; + return { value, comment: typeof comment === 'string' ? comment.slice(0, MAX_COMMENT) : null }; +} + +/** + * One judge call bounded by `timeoutMs` (its signal aborts then, and a judge + * that ignores the signal is no longer waited for). Never throws: a throw or + * the timeout becomes a `judge_error` result; never returns what the daemon + * would 400. + */ +async function runJudge(judge: JudgeFn, rule: JudgeRule, trace: EvaluatorTrace, timeoutMs: number): Promise { + const controller = new AbortController(); + let timer: ReturnType | undefined; + const timedOut = new Promise((_resolve, reject) => { + timer = setTimeout(() => { + const error = new Error(`timed out after ${timeoutMs} ms`); + controller.abort(error); + reject(error); + }, timeoutMs); + }); + const run = Promise.resolve().then(() => judge(rule, trace, controller.signal)); + // A judge that settles after the timeout must not surface as an unhandled rejection. + run.catch(() => undefined); + try { + return acceptable(await Promise.race([run, timedOut])); + } catch (error) { + return judgeError(error); + } finally { + clearTimeout(timer); + } +} + +/** Headers for a server-side daemon call: JSON, plus the bearer `LOUSHO_OBSERVE_TOKEN` names. */ +function daemonHeaders(): Record { + const token = process.env.LOUSHO_OBSERVE_TOKEN; + return { accept: 'application/json', 'content-type': 'application/json', ...(token ? { authorization: `Bearer ${token}` } : {}) }; +} + +/** POST/GET the daemon at `path`; the response, or `undefined` when it is unreachable. */ +async function callDaemon(fetchFn: typeof fetch, path: string, body?: unknown): Promise { + try { + return await fetchFn(`${observeTarget()}${path}`, { + method: body === undefined ? 'GET' : 'POST', + headers: daemonHeaders(), + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + signal: AbortSignal.timeout(DAEMON_TIMEOUT_MS), + }); + } catch { + return undefined; + } +} + +interface ClaimedJob { + job_id: string; + rule: JudgeRule; + trace: EvaluatorTrace; +} + +/** Start pulling judge jobs from the daemon; `stop()` on server close. */ +export function startJudgeLoop(options: JudgeLoopOptions): JudgeLoop { + const fetchFn = options.fetch ?? fetch; + const judge = options.judge ?? modelJudge(options.secretsStore); + const warn = options.warn ?? ((message: string) => console.warn(message)); + const now = options.now ?? Date.now; + const leaseMs = options.leaseMs ?? LEASE_MS_DEFAULT; + const judgeTimeoutMs = options.judgeTimeoutMs ?? JUDGE_TIMEOUT_MS_DEFAULT; + let stopped = false; + let down = false; + let inFlight: Promise | undefined; + + /** The first failure of an outage is logged; the rest stay quiet until a healthy answer. */ + const outage = (why: string): void => { + if (down) return; + down = true; + warn(`[lousho studio] judge evaluators paused: the observe daemon at ${observeTarget()} ${why}; retrying.`); + }; + + /** A daemon answer: `true` when it was a 2xx. */ + const healthy = (response: globalThis.Response | undefined): response is globalThis.Response => { + if (response?.ok) { + down = false; + return true; + } + outage(response === undefined ? 'is unreachable' : `answered HTTP ${response.status}`); + return false; + }; + + /** The claimed jobs, or `undefined` (an outage) when the daemon did not answer with them. */ + const claim = async (): Promise => { + const claimed = await callDaemon(fetchFn, '/api/v1/evaluators/judge-jobs/claim', { max: CLAIM_MAX, lease_ms: leaseMs }); + if (claimed === undefined || !claimed.ok) { + healthy(claimed); + return undefined; + } + const body = (await claimed.json().catch(() => undefined)) as { jobs?: unknown } | undefined; + if (!Array.isArray(body?.jobs)) { + outage('answered a claim without a jobs array'); + return undefined; + } + down = false; + return body.jobs as ClaimedJob[]; + }; + + const runTick = async (): Promise => { + // Measured before the claim, so the local deadline never trails the daemon's. + const deadline = now() + leaseMs; + const jobs = await claim(); + for (const job of jobs ?? []) { + // What is left of the lease after keeping time back to post; a job out of time is left to expire and requeue. + const budget = deadline - now() - POST_MARGIN_MS; + if (stopped || budget <= 0) return; + const result = await runJudge(judge, job.rule, job.trace, Math.min(judgeTimeoutMs, budget)); + const posted = await callDaemon(fetchFn, `/api/v1/evaluators/judge-jobs/${encodeURIComponent(job.job_id)}/result`, result); + // A 404 is a lease that ran out (the job was requeued): not an outage. + if (posted?.status !== 404 && !healthy(posted)) return; + } + }; + + const tick = (): Promise => { + if (stopped) return Promise.resolve(); + inFlight ??= runTick() + .catch((error: unknown) => warn(`[lousho studio] judge tick failed: ${error instanceof Error ? error.message : String(error)}`)) + .finally(() => (inFlight = undefined)); + return inFlight; + }; + + const timer = setInterval(() => void tick(), options.intervalMs ?? 2000); + timer.unref(); + return { + stop() { + stopped = true; + clearInterval(timer); + }, + tick, + }; +} + +export interface JudgeTestRouteOptions { + secretsStore: SecretsStore; + /** The judge; default {@link modelJudge} over `secretsStore`. */ + judge?: JudgeFn; + /** The fetch daemon calls go through (tests). */ + fetch?: typeof fetch; + /** Cap on the dry run's judge call; default 20000 ms. */ + judgeTimeoutMs?: number; +} + +const UNREACHABLE = 'The observe daemon is unreachable - run `lousho observe start` to collect agent data.'; + +/** Answer with the daemon's response as it is (status, content type, body). */ +async function passThrough(res: Response, upstream: globalThis.Response): Promise { + res.status(upstream.status); + const contentType = upstream.headers.get('content-type'); + if (contentType) res.setHeader('content-type', contentType); + res.send(Buffer.from(await upstream.arrayBuffer())); +} + +/** An id the daemon would read as something other than one plain path segment. */ +const unsafeId = (id: string): boolean => id === '.' || id === '..' || /[/\\]/.test(id); + +const invalid = (res: Response, status: number, error: string): void => { + res.status(status).json({ error, code: 'LOUSHO_REQUEST_INVALID' }); +}; + +/** A judge already bounded by its timeout (see {@link runJudge}). */ +type BoundJudge = (rule: JudgeRule, trace: EvaluatorTrace) => Promise; + +/** The judge dry run: the trace from the daemon, built as the daemon builds it, then the judge. */ +async function testJudgeRule(req: Request, res: Response, rule: JudgeRule, judge: BoundJudge, fetchFn: typeof fetch): Promise { + const traceId: unknown = (req.body as { trace_id?: unknown } | undefined)?.trace_id; + if (typeof traceId !== 'string' || traceId === '' || unsafeId(traceId)) { + return invalid(res, 400, 'Invalid evaluator test body: trace_id: must be a non-empty trace id.'); + } + const detail = await callDaemon(fetchFn, `/api/v1/traces/${encodeURIComponent(traceId)}`); + if (detail === undefined) return void res.status(502).json({ error: UNREACHABLE }); + if (!detail.ok) return passThrough(res, detail); + const { observations } = (await detail.json()) as { observations: ObservationRow[] }; + const trace = evaluatorTraceFrom(traceId, observations); + if (trace === undefined) return invalid(res, 404, `No trace '${traceId}' with a root observation.`); + res.status(200).json({ result: await judge(rule, trace) }); +} + +/** + * `POST /observe/api/v1/evaluators/:id/test` `{ trace_id }`, registered + * before the observe proxy (and behind the Studio's API auth like every + * `/observe` path). The rule comes from the daemon: a `judge` rule runs here + * as a dry run (`200 { result: { value, comment } }`, no score written); any + * other rule is forwarded to the daemon's own `/test` unchanged. The + * daemon's 404s (unknown rule or trace) pass through; a daemon that is down + * is a 502. + */ +export function registerJudgeTestRoute(app: Express, options: JudgeTestRouteOptions): void { + const judgeFn = options.judge ?? modelJudge(options.secretsStore); + const timeoutMs = options.judgeTimeoutMs ?? JUDGE_TIMEOUT_MS_DEFAULT; + const judge: BoundJudge = (rule, trace) => runJudge(judgeFn, rule, trace, timeoutMs); + const fetchFn = options.fetch ?? fetch; + app.post('/observe/api/v1/evaluators/:id/test', async (req, res, next) => { + try { + const id = String(req.params.id); + if (unsafeId(id)) return invalid(res, 400, 'Invalid evaluator rule id.'); + const rulePath = `/api/v1/evaluators/${encodeURIComponent(id)}`; + const found = await callDaemon(fetchFn, rulePath); + if (found === undefined) return void res.status(502).json({ error: UNREACHABLE }); + if (!found.ok) return passThrough(res, found); + const { evaluator } = (await found.json()) as { evaluator: JudgeRule | { kind: string } }; + if (evaluator.kind === 'judge') return testJudgeRule(req, res, evaluator as JudgeRule, judge, fetchFn); + const forwarded = await callDaemon(fetchFn, `${rulePath}/test`, req.body ?? {}); + if (forwarded === undefined) return void res.status(502).json({ error: UNREACHABLE }); + return passThrough(res, forwarded); + } catch (error) { + next(error); + } + }); +} diff --git a/apps/agent-forge/server/playground.ts b/apps/agent-forge/server/playground.ts index b8682ecf3..0596482e0 100644 --- a/apps/agent-forge/server/playground.ts +++ b/apps/agent-forge/server/playground.ts @@ -88,7 +88,7 @@ export function substituteVariables(content: string, variables: Record number | { value, comment }`, and a `judge` rule holds a model and a rubric to grade by. The daemon runs `code` rules on new traces (see [Online evaluators](#online-evaluators)); `judge` -rules are stored here and run through Agent Forge. The routes take the same +rules run through Agent Forge (see [Judge rules](#judge-rules-through-agent-forge)). The routes take the same token as ingest: | Route | Answer | @@ -666,6 +666,53 @@ request waits until the one in flight finishes. A `judge` rule is a `409` (`judge evaluators run through Agent Forge`). Without the isolate it answers `{ "result": { "value": null, "comment": "isolate_unavailable" } }`. +#### Judge rules (through Agent Forge) + +The daemon holds no provider keys, so it cannot run a `judge` rule itself. +When a `judge` rule matches a new trace (same `filter`, `sample_rate` and +"already scored" check as code rules), the daemon queues a job for that rule +and trace. Agent Forge (`lousho studio`) pulls the jobs, runs each judge with +the provider keys it already has (a key saved in Studio settings wins over the +provider's env var, as in the playground), and posts the result back. The +daemon writes it as the same `source: "online"` score. So judges run only +while Studio is running. Until then the jobs wait. + +- At most 500 jobs wait at once (pending plus claimed). Past that, new jobs + are dropped and counted. Jobs live in memory, so a daemon restart loses + them. +- The judge is one model call to the rule's `model` (`provider/model`, or + `mock`). The system prompt is a fixed grading instruction plus the rule's + `rubric`. The user message holds the trace's name, status, input, output and + error, each cut at about 8 KB. The model must reply with a JSON object + `{ "score": <0..1>, "comment": "" }`. Agent Forge reads the + first JSON object in the reply. A reply with no valid score scores + `value: null` with comment `invalid_judge_output`. A provider error (a + missing key, for example) scores `value: null` with comment + `judge_error: `. +- The judge call makes no trace of its own, so it never triggers evaluators. + +The two endpoints Agent Forge uses (they take the same token as ingest, which +Agent Forge sends from `LOUSHO_OBSERVE_TOKEN` when it is set): + +| Route | Answer | +| --- | --- | +| `POST /api/v1/evaluators/judge-jobs/claim` `{ "max"?: 1..20 (5), "lease_ms"?: 1000..300000 (60000) }` | `200 { "jobs": [{ "job_id", "rule", "trace" }] }` (an empty array when none wait). Each job is leased for `lease_ms`. A lease that runs out puts the job back at the front of the queue. A job whose rule was deleted, disabled or is no longer `judge` is discarded at claim time. `trace` is the same object a code rule receives. | +| `POST /api/v1/evaluators/judge-jobs/:job_id/result` `{ "value": <0..1 or null>, "comment"?: "" or null }` | `201 { "score_id" }`, and the job leaves the queue. The score is named after the rule as it was at claim time, so a rule renamed while its job is leased scores under its old name. An unknown job, or one whose lease ran out, is a `404`. A bad body is a `400`. | + +Agent Forge claims up to 5 jobs every 2 seconds, with a 60-second lease, and +runs them one at a time. Each judge call is cut off after 20 seconds, or +sooner if less of the lease is left (a call cut off scores `value: null` with +comment `judge_error: timed out after ms`). A claimed job whose lease has +run out before its turn is not run: it expires and is claimed again later. A +daemon that is down only pauses the loop, and it logs once per outage. The +dry run below has the same 20-second cap. + +To dry-run a judge rule, use Agent Forge: `POST /observe/api/v1/evaluators/:id/test` +`{ "trace_id" }` on the Studio server runs the judge on that trace and answers +`200 { "result": { "value", "comment" } }` without writing a score. For a +`code` rule it forwards the request to the daemon unchanged. The daemon's own +`/test` still answers `409` for a judge rule. + ### Annotation queues An annotation queue is a named list of traces for a person to review one by diff --git a/llms-full.txt b/llms-full.txt index f58759406..68f69b2d0 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14485,7 +14485,7 @@ An evaluator rule says which new traces to score and how. A `code` rule holds the source of a function `(trace) => number | { value, comment }`, and a `judge` rule holds a model and a rubric to grade by. The daemon runs `code` rules on new traces (see [Online evaluators](#online-evaluators)); `judge` -rules are stored here and run through Agent Forge. The routes take the same +rules run through Agent Forge (see [Judge rules](#judge-rules-through-agent-forge)). The routes take the same token as ingest: | Route | Answer | @@ -14560,6 +14560,53 @@ request waits until the one in flight finishes. A `judge` rule is a `409` (`judge evaluators run through Agent Forge`). Without the isolate it answers `{ "result": { "value": null, "comment": "isolate_unavailable" } }`. +#### Judge rules (through Agent Forge) + +The daemon holds no provider keys, so it cannot run a `judge` rule itself. +When a `judge` rule matches a new trace (same `filter`, `sample_rate` and +"already scored" check as code rules), the daemon queues a job for that rule +and trace. Agent Forge (`lousho studio`) pulls the jobs, runs each judge with +the provider keys it already has (a key saved in Studio settings wins over the +provider's env var, as in the playground), and posts the result back. The +daemon writes it as the same `source: "online"` score. So judges run only +while Studio is running. Until then the jobs wait. + +- At most 500 jobs wait at once (pending plus claimed). Past that, new jobs + are dropped and counted. Jobs live in memory, so a daemon restart loses + them. +- The judge is one model call to the rule's `model` (`provider/model`, or + `mock`). The system prompt is a fixed grading instruction plus the rule's + `rubric`. The user message holds the trace's name, status, input, output and + error, each cut at about 8 KB. The model must reply with a JSON object + `{ "score": <0..1>, "comment": "" }`. Agent Forge reads the + first JSON object in the reply. A reply with no valid score scores + `value: null` with comment `invalid_judge_output`. A provider error (a + missing key, for example) scores `value: null` with comment + `judge_error: `. +- The judge call makes no trace of its own, so it never triggers evaluators. + +The two endpoints Agent Forge uses (they take the same token as ingest, which +Agent Forge sends from `LOUSHO_OBSERVE_TOKEN` when it is set): + +| Route | Answer | +| --- | --- | +| `POST /api/v1/evaluators/judge-jobs/claim` `{ "max"?: 1..20 (5), "lease_ms"?: 1000..300000 (60000) }` | `200 { "jobs": [{ "job_id", "rule", "trace" }] }` (an empty array when none wait). Each job is leased for `lease_ms`. A lease that runs out puts the job back at the front of the queue. A job whose rule was deleted, disabled or is no longer `judge` is discarded at claim time. `trace` is the same object a code rule receives. | +| `POST /api/v1/evaluators/judge-jobs/:job_id/result` `{ "value": <0..1 or null>, "comment"?: "" or null }` | `201 { "score_id" }`, and the job leaves the queue. The score is named after the rule as it was at claim time, so a rule renamed while its job is leased scores under its old name. An unknown job, or one whose lease ran out, is a `404`. A bad body is a `400`. | + +Agent Forge claims up to 5 jobs every 2 seconds, with a 60-second lease, and +runs them one at a time. Each judge call is cut off after 20 seconds, or +sooner if less of the lease is left (a call cut off scores `value: null` with +comment `judge_error: timed out after ms`). A claimed job whose lease has +run out before its turn is not run: it expires and is claimed again later. A +daemon that is down only pauses the loop, and it logs once per outage. The +dry run below has the same 20-second cap. + +To dry-run a judge rule, use Agent Forge: `POST /observe/api/v1/evaluators/:id/test` +`{ "trace_id" }` on the Studio server runs the judge on that trace and answers +`200 { "result": { "value", "comment" } }` without writing a score. For a +`code` rule it forwards the request to the daemon unchanged. The daemon's own +`/test` still answers `409` for a judge rule. + ### Annotation queues An annotation queue is a named list of traces for a person to review one by diff --git a/src/observe/evaluatorTrace.ts b/src/observe/evaluatorTrace.ts new file mode 100644 index 000000000..7de568bc4 --- /dev/null +++ b/src/observe/evaluatorTrace.ts @@ -0,0 +1,68 @@ +/** + * The trace object an online evaluator receives, built from a trace's + * observation rows (`GET /api/v1/traces/:id`'s `observations`). The daemon's + * worker builds it for code rules and for the judge jobs it hands out; Agent + * Forge builds the same object from the trace detail for a judge dry run, so + * both sides share this one function. Pure: no node builtins, no db. + */ + +import type { ObservationRow } from './store'; + +/** Most observations handed to an evaluator (the oldest are kept). */ +const MAX_OBSERVATIONS = 200; + +/** What an online evaluator receives: the root observation's fields plus the trace's first observations. */ +export interface EvaluatorTrace { + /** The trace id. */ + trace_id: string; + /** The root's agent. */ + agent_id: string | null; + /** The root's name. */ + name: string | null; + /** The root's status. */ + status: string | null; + /** The root's start, epoch ms. */ + start: number; + /** The root's end, epoch ms. */ + end: number | null; + /** `end - start`, or `null` while open. */ + duration_ms: number | null; + /** The root's captured input. */ + input: string | null; + /** The root's captured output. */ + output: string | null; + /** The first error observation's `attrs.error`, stringified when it is not a string. */ + error_message: string | null; + /** The first 200 observations by start. */ + observations: ObservationRow[]; +} + +function errorMessageOf(observations: readonly ObservationRow[]): string | null { + const failed = observations.find((o) => o.status === 'error' && o.attrs?.error !== undefined && o.attrs.error !== null); + if (failed === undefined) return null; + const error = failed.attrs?.error; + return typeof error === 'string' ? error : JSON.stringify(error); +} + +/** + * The evaluator's view of one trace from its observation rows (ordered by + * start, as the trace detail returns them), or `undefined` when no row is a + * root (`parent_obs_id === null`). + */ +export function evaluatorTraceFrom(traceId: string, observations: readonly ObservationRow[]): EvaluatorTrace | undefined { + const root = observations.find((o) => o.parent_obs_id === null); + if (root === undefined) return undefined; + return { + trace_id: traceId, + agent_id: root.agent_id, + name: root.name, + status: root.status, + start: root.start, + end: root.end, + duration_ms: root.end === null ? null : root.end - root.start, + input: root.input, + output: root.output, + error_message: errorMessageOf(observations), + observations: observations.slice(0, MAX_OBSERVATIONS), + }; +} diff --git a/src/observe/evaluatorWorker.test.ts b/src/observe/evaluatorWorker.test.ts index 191744bb5..12cb835f0 100644 --- a/src/observe/evaluatorWorker.test.ts +++ b/src/observe/evaluatorWorker.test.ts @@ -7,7 +7,7 @@ import { createRequire } from 'node:module'; import type * as http from 'node:http'; import { afterEach, describe, expect, it } from 'vitest'; -import { createEvaluatorRule, type EvaluatorRuleFields } from './evaluators'; +import { createEvaluatorRule, deleteEvaluatorRule, updateEvaluatorRule, type EvaluatorRuleFields } from './evaluators'; import { rootsEndedIn, startEvaluatorWorker, @@ -159,7 +159,7 @@ describe('evaluator worker', () => { } }); - it('ignores child ends, open roots, disabled rules, judge rules and sample_rate 0', async () => { + it('ignores child ends, open roots, disabled rules and sample_rate 0; never runs a judge rule itself', async () => { const h = harness(); createEvaluatorRule(h.db, codeRule({ name: 'off', enabled: false })); createEvaluatorRule(h.db, codeRule({ name: 'never', sample_rate: 0 })); @@ -168,7 +168,7 @@ describe('evaluator worker', () => { await ingestAndWait(h, [rootEnd('t1')]); expect(h.runs).toEqual([]); expect(onlineScores(h.db, 't1')).toEqual([]); - expect(h.worker.stats()).toMatchObject({ ran: 0, skipped: 0 }); + expect(h.worker.stats()).toMatchObject({ ran: 0, skipped: 0, judge_queued: 1 }); }); it('samples with random() < sample_rate', async () => { @@ -281,6 +281,112 @@ describe('evaluator worker', () => { }); }); +const judgeRule = (over: Partial = {}): EvaluatorRuleFields => + ({ name: 'helpful', enabled: true, filter: {}, sample_rate: 1, kind: 'judge', config: { model: 'mock', rubric: 'Is it helpful?' }, ...over }) as EvaluatorRuleFields; + +describe('judge jobs', () => { + /** A harness whose judge leases run on a clock the test moves. */ + function judgeHarness(opts: EvaluatorWorkerOptions = {}): Harness & { clock: { now: number } } { + const clock = { now: T0 }; + return { ...harness(() => 1, { now: () => clock.now, ...opts }), clock }; + } + + it('a matching judge rule queues one job and writes no score until a result', async () => { + const h = judgeHarness(); + const rule = createEvaluatorRule(h.db, judgeRule({ filter: { agent_id: 'support' } })); + createEvaluatorRule(h.db, judgeRule({ name: 'other-agent', filter: { agent_id: 'billing' } })); + await ingestAndWait(h, [rootEnd('t1'), child('t1')]); + // The same root ending again does not queue a second job. + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.runs).toEqual([]); + expect(onlineScores(h.db, 't1')).toEqual([]); + expect(h.worker.stats()).toMatchObject({ judge_queued: 1, judge_leased: 0, ran: 0 }); + + const [job, ...rest] = h.worker.claimJudgeJobs(5, 60_000); + expect(rest).toEqual([]); + expect(job.rule).toEqual(rule); + expect(job.trace).toMatchObject({ trace_id: 't1', agent_id: 'support', status: 'ok', input: 'hi', output: 'hello', error_message: 'lookup failed' }); + expect(job.trace.observations).toHaveLength(2); + expect(h.worker.stats()).toMatchObject({ judge_queued: 0, judge_leased: 1 }); + // Leased, so still deduped. + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.worker.stats()).toMatchObject({ judge_queued: 0, judge_leased: 1 }); + + const scoreId = h.worker.completeJudgeJob(job.job_id, { value: 0.8, comment: 'clear answer' }); + expect(typeof scoreId).toBe('string'); + expect(onlineScores(h.db, 't1')).toEqual([ + expect.objectContaining({ score_id: scoreId, name: 'helpful', value: 0.8, comment: 'clear answer', source: 'online', agent_id: 'support' }), + ]); + expect(h.worker.stats()).toMatchObject({ judge_queued: 0, judge_leased: 0, judge_scored: 1 }); + // Scored now: a new root end queues nothing, and the job is gone. + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.worker.stats().judge_queued).toBe(0); + expect(h.worker.completeJudgeJob(job.job_id, { value: 1, comment: null })).toBeUndefined(); + }); + + it('an expired lease goes back to the front of the queue and its late result is refused', async () => { + const h = judgeHarness(); + createEvaluatorRule(h.db, judgeRule()); + await ingestAndWait(h, [rootEnd('t1'), rootEnd('t2')]); + await ingestAndWait(h, [rootEnd('t3')]); + const first = h.worker.claimJudgeJobs(1, 1000); + expect(first.map((j) => j.trace.trace_id)).toEqual(['t1']); + h.clock.now += 500; + expect(h.worker.claimJudgeJobs(1, 1000).map((j) => j.trace.trace_id)).toEqual(['t2']); + h.clock.now += 600; // t1's lease is over, t2's is not + expect(h.worker.stats()).toMatchObject({ judge_queued: 2, judge_leased: 1 }); + expect(h.worker.completeJudgeJob(first[0].job_id, { value: 1, comment: null })).toBeUndefined(); + const again = h.worker.claimJudgeJobs(5, 1000); + expect(again.map((j) => j.trace.trace_id)).toEqual(['t1', 't3']); + expect(again[0].job_id).toBe(first[0].job_id); + expect(h.worker.completeJudgeJob(first[0].job_id, { value: null, comment: 'invalid_judge_output' })).toEqual(expect.any(String)); + expect(onlineScores(h.db, 't1')).toEqual([expect.objectContaining({ value: null, comment: 'invalid_judge_output' })]); + }); + + it('drops and counts a job past maxJudgeJobs (pending + leased)', async () => { + const h = judgeHarness({ maxJudgeJobs: 2 }); + createEvaluatorRule(h.db, judgeRule()); + await ingestAndWait(h, [rootEnd('t1')]); + h.worker.claimJudgeJobs(1, 60_000); + await ingestAndWait(h, [rootEnd('t2'), rootEnd('t3')]); + expect(h.worker.stats()).toMatchObject({ judge_queued: 1, judge_leased: 1, judge_dropped: 1 }); + }); + + it('holds 500 jobs by default: the 501st drops and counts', async () => { + const h = judgeHarness(); + createEvaluatorRule(h.db, judgeRule()); + await ingestAndWait(h, Array.from({ length: 501 }, (_, i) => rootEnd(`t${i}`))); + expect(h.worker.stats()).toMatchObject({ judge_queued: 500, judge_dropped: 1 }); + }); + + it('discards a job at claim time whose rule was deleted, disabled or switched to code', async () => { + const h = judgeHarness(); + const gone = createEvaluatorRule(h.db, judgeRule({ name: 'gone' })); + const off = createEvaluatorRule(h.db, judgeRule({ name: 'off' })); + const switched = createEvaluatorRule(h.db, judgeRule({ name: 'switched' })); + const kept = createEvaluatorRule(h.db, judgeRule({ name: 'kept' })); + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.worker.stats().judge_queued).toBe(4); + deleteEvaluatorRule(h.db, gone?.rule_id ?? ''); + updateEvaluatorRule(h.db, off?.rule_id ?? '', judgeRule({ name: 'off', enabled: false })); + updateEvaluatorRule(h.db, switched?.rule_id ?? '', codeRule({ name: 'switched' })); + const jobs = h.worker.claimJudgeJobs(5, 60_000); + expect(jobs.map((j) => j.rule.rule_id)).toEqual([kept?.rule_id]); + expect(h.worker.stats()).toMatchObject({ judge_queued: 0, judge_leased: 1, judge_discarded: 3 }); + }); + + it('stop() clears judge jobs and refuses claims and results', async () => { + const h = judgeHarness(); + createEvaluatorRule(h.db, judgeRule()); + await ingestAndWait(h, [rootEnd('t1'), rootEnd('t2')]); + const [job] = h.worker.claimJudgeJobs(1, 60_000); + h.worker.stop(); + expect(h.worker.claimJudgeJobs(5, 60_000)).toEqual([]); + expect(h.worker.completeJudgeJob(job.job_id, { value: 1, comment: null })).toBeUndefined(); + expect(h.worker.stats()).toMatchObject({ judge_queued: 0, judge_leased: 0 }); + }); +}); + describe('ingest with the evaluator worker', () => { it('still answers 202 when the worker throws', async () => { const db = newDb(); diff --git a/src/observe/evaluatorWorker.ts b/src/observe/evaluatorWorker.ts index a4da74e78..e5d4f96fa 100644 --- a/src/observe/evaluatorWorker.ts +++ b/src/observe/evaluatorWorker.ts @@ -13,7 +13,11 @@ * - `code` rules run their `config.source` as `(source)(trace)` in * `sandbox({ driver: 'isolate', timeoutMs: 2000 })`. Without a loadable * `isolated-vm` (or on Node >= 20 without `--no-node-snapshot`) they are - * skipped with one warning per worker. `judge` rules have no runner here. + * skipped with one warning per worker. + * - `judge` rules do not run here (the daemon holds no provider keys): a + * match queues a judge job (judgeJobs.ts, bounded at 500) that Agent Forge + * claims, runs with its stored keys, and answers with a result, which the + * worker writes as the same `online` score. */ import { randomUUID } from 'node:crypto'; @@ -24,8 +28,10 @@ import { sandbox as realSandbox, sandboxErrorKind, type Sandbox, type SandboxOpt import type { SqlDatabase } from '../storage/sqlite/driver'; import { moduleDir } from '../utils/moduleDir'; import type { IngestEvent } from './envelope'; -import { listEvaluatorRules, type EvaluatorRuleFilter, type EvaluatorRuleRow } from './evaluators'; -import { ingestBatch, listScores, traceDetail, type ObservationRow } from './store'; +import { getEvaluatorRule, listEvaluatorRules, type EvaluatorRuleFilter, type EvaluatorRuleRow } from './evaluators'; +import { evaluatorTraceFrom, type EvaluatorTrace } from './evaluatorTrace'; +import { createJudgeJobs, type ClaimedJudgeJob, type JudgeJobStats, type JudgeScore } from './judgeJobs'; +import { ingestBatch, listScores, traceDetail } from './store'; /** What one evaluator run produced: the score's value and comment. */ export interface EvaluatorResult { @@ -43,6 +49,16 @@ export interface EvaluatorWorkerStats { ran: number; /** Rule runs skipped because no runner could take them (no isolate). */ skipped: number; + /** Judge jobs waiting for Agent Forge to claim them. */ + judge_queued: number; + /** Judge jobs claimed and not yet answered (lease running). */ + judge_leased: number; + /** Judge jobs dropped because the judge queue (500) was full. */ + judge_dropped: number; + /** Judge jobs discarded at claim time (rule deleted, disabled or no longer `judge`; trace gone). */ + judge_discarded: number; + /** Judge results written as scores. */ + judge_scored: number; } /** Options of {@link startEvaluatorWorker}; every one is a test seam with a production default. */ @@ -57,6 +73,10 @@ export interface EvaluatorWorkerOptions { maxQueue?: number; /** Whether the isolate driver can run here; default a cached probe for `isolated-vm`. */ isolateAvailable?: () => boolean; + /** Most judge jobs pending + leased at once; default 500. */ + maxJudgeJobs?: number; + /** The clock judge-job leases run on; default `Date.now`. */ + now?: () => number; } /** The running worker the daemon owns. */ @@ -73,6 +93,10 @@ export interface EvaluatorWorker { /** Resolves once the queue is empty and nothing is running. */ idle(): Promise; stats(): EvaluatorWorkerStats; + /** Lease up to `max` judge jobs for `leaseMs` (expired leases requeue first; stale jobs are discarded). */ + claimJudgeJobs(max: number, leaseMs: number): ClaimedJudgeJob[]; + /** Write a claimed judge job's result as its score; the score id, or `undefined` for an unknown or expired job. */ + completeJudgeJob(jobId: string, result: EvaluatorResult): string | undefined; /** Drop the queue and stop taking traces; an in-flight run writes nothing. */ stop(): void; } @@ -82,8 +106,6 @@ const EVALUATOR_QUEUE_DEFAULT = 1000; /** A code rule's wall-clock cap. */ const CODE_TIMEOUT_MS = 2000; -/** Most observations handed to an evaluator (the oldest are kept). */ -const MAX_OBSERVATIONS = 200; const MAX_COMMENT = 1000; const RULES_PAGE = 1000; @@ -91,21 +113,6 @@ const ISOLATE_WARNING = "[lousho observe] code evaluators are skipped: the 'isolate' sandbox needs the optional isolated-vm package " + '(npm install isolated-vm) and, on Node >= 20, NODE_OPTIONS=--no-node-snapshot.'; -/** What a code evaluator receives. */ -interface EvaluatorTrace { - trace_id: string; - agent_id: string | null; - name: string | null; - status: string | null; - start: number; - end: number | null; - duration_ms: number | null; - input: string | null; - output: string | null; - error_message: string | null; - observations: ObservationRow[]; -} - /** * Whether `isolated-vm` resolves beside the SDK and this process can load it * (Node >= 20 crashes on load without `--no-node-snapshot`). Resolving does @@ -126,31 +133,10 @@ function probeIsolate(): boolean { let isolateProbe: boolean | undefined; const defaultIsolateAvailable = (): boolean => (isolateProbe ??= probeIsolate()); -function errorMessageOf(observations: readonly ObservationRow[]): string | null { - const failed = observations.find((o) => o.status === 'error' && o.attrs?.error !== undefined && o.attrs.error !== null); - if (failed === undefined) return null; - const error = failed.attrs?.error; - return typeof error === 'string' ? error : JSON.stringify(error); -} - -/** The trace as an evaluator sees it, or `undefined` when it has no root observation. */ +/** The trace as an evaluator sees it, or `undefined` when it does not exist or has no root observation. */ function evaluatorTrace(db: SqlDatabase, traceId: string): EvaluatorTrace | undefined { const detail = traceDetail(db, traceId); - const root = detail?.observations.find((o) => o.parent_obs_id === null); - if (detail === undefined || root === undefined) return undefined; - return { - trace_id: traceId, - agent_id: root.agent_id, - name: root.name, - status: root.status, - start: root.start, - end: root.end, - duration_ms: root.end === null ? null : root.end - root.start, - input: root.input, - output: root.output, - error_message: errorMessageOf(detail.observations), - observations: detail.observations.slice(0, MAX_OBSERVATIONS), - }; + return detail === undefined ? undefined : evaluatorTraceFrom(traceId, detail.observations); } /** Whether every key the filter sets matches the trace's root. */ @@ -183,6 +169,15 @@ export function rootsEndedIn(batch: readonly IngestEvent[]): string[] { return [...ids]; } +/** The judge queue's counters under their `judge_` stats names. */ +const judgeStats = (stats: JudgeJobStats): Pick => ({ + judge_queued: stats.queued, + judge_leased: stats.leased, + judge_dropped: stats.dropped, + judge_discarded: stats.discarded, + judge_scored: stats.scored, +}); + /** Start the worker over `db`. */ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptions = {}): EvaluatorWorker { const makeSandbox = opts.sandbox ?? realSandbox; @@ -208,30 +203,36 @@ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptio } }; - /** The runner for a rule's kind; `judge` has none yet (Agent Forge runs those). */ - const runnerFor = (rule: EvaluatorRuleRow): ((trace: EvaluatorTrace) => Promise) | undefined => - rule.kind === 'code' ? (trace) => runCode(rule, trace) : undefined; - const alreadyScored = (rule: EvaluatorRuleRow, traceId: string): boolean => listScores(db, { trace_id: traceId, name: rule.name, source: 'online', limit: 1 }).scores.length > 0; - const writeScore = (rule: EvaluatorRuleRow, trace: EvaluatorTrace, result: EvaluatorResult): void => { + const writeScore = (score: JudgeScore): string => { + const score_id = randomUUID(); ingestBatch(db, [ { type: 'score', data: { - score_id: randomUUID(), - trace_id: trace.trace_id, - name: rule.name, - value: result.value ?? undefined, + score_id, + trace_id: score.trace_id, + name: score.rule_name, + value: score.value ?? undefined, source: 'online', - comment: result.comment ?? undefined, - agent_id: trace.agent_id ?? undefined, + comment: score.comment ?? undefined, + agent_id: score.agent_id ?? undefined, }, }, ]); + return score_id; }; + const judgeJobs = createJudgeJobs({ + getRule: (ruleId) => getEvaluatorRule(db, ruleId), + loadTrace: (traceId) => evaluatorTrace(db, traceId), + writeScore, + now: opts.now, + maxJobs: opts.maxJudgeJobs, + }); + const skipCode = (): void => { counts.skipped++; if (warned) return; @@ -240,13 +241,14 @@ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptio }; const evaluateRule = async (rule: EvaluatorRuleRow, trace: EvaluatorTrace): Promise => { - const runner = runnerFor(rule); - if (runner === undefined || !matchesFilter(rule.filter, trace) || !(random() < rule.sample_rate)) return; - if (alreadyScored(rule, trace.trace_id)) return; + if (!matchesFilter(rule.filter, trace) || !(random() < rule.sample_rate)) return; + if (alreadyScored(rule, trace.trace_id) || judgeJobs.has(rule.rule_id, trace.trace_id)) return; + // The daemon cannot run a judge: Agent Forge claims the job (judgeJobs.ts). + if (rule.kind === 'judge') return void judgeJobs.enqueue(rule.rule_id, trace.trace_id); if (!isolateAvailable()) return skipCode(); - const result = await runner(trace); + const result = await runCode(rule, trace); if (stopped) return; - writeScore(rule, trace, result); + writeScore({ rule_name: rule.name, trace_id: trace.trace_id, agent_id: trace.agent_id, ...result }); counts.ran++; }; @@ -294,10 +296,9 @@ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptio const run = dryRuns.then(async (): Promise => { const trace = evaluatorTrace(db, traceId); if (trace === undefined) return undefined; - const runner = runnerFor(rule); - if (runner === undefined) return { value: null, comment: 'unsupported_kind' }; + if (rule.kind !== 'code') return { value: null, comment: 'unsupported_kind' }; if (!isolateAvailable()) return { value: null, comment: 'isolate_unavailable' }; - return runner(trace); + return runCode(rule, trace); }); dryRuns = run.catch(() => undefined); return run; @@ -305,10 +306,13 @@ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptio async idle() { while (draining !== undefined) await draining; }, - stats: () => ({ queued: pending.length, ...counts }), + stats: () => ({ queued: pending.length, ...counts, ...judgeStats(judgeJobs.stats()) }), + claimJudgeJobs: (max, leaseMs) => (stopped ? [] : judgeJobs.claim(max, leaseMs)), + completeJudgeJob: (jobId, result) => (stopped ? undefined : judgeJobs.complete(jobId, result)), stop() { stopped = true; pending.length = 0; + judgeJobs.clear(); }, }; } diff --git a/src/observe/judgeJobs.ts b/src/observe/judgeJobs.ts new file mode 100644 index 000000000..f3d1dd4aa --- /dev/null +++ b/src/observe/judgeJobs.ts @@ -0,0 +1,171 @@ +/** + * The judge-job queue: LLM-judge evaluator runs the daemon cannot do itself + * (it holds no provider keys), waiting for Agent Forge to pull them. + * + * - A job is `(rule, trace)`; at most one per pair is queued or leased. + * - Bounded (`maxJobs`, default 500, pending + leased together); past it a + * job drops and counts in `stats().dropped`. + * - `claim` leases jobs from the front for `leaseMs`. The rule is re-read + * and the trace built at claim time: a job whose rule was deleted, + * disabled or is no longer `judge` (or whose trace is gone) is discarded. + * A lease that expires puts its job back at the front of the queue. + * - `complete` takes a leased, unexpired job's result, writes the score and + * forgets the job; anything else is `undefined` (the route's 404). + * - In memory: pending and leased jobs are lost on a daemon restart. + */ + +import { randomUUID } from 'node:crypto'; + +import type { EvaluatorRuleRow } from './evaluators'; +import type { EvaluatorTrace } from './evaluatorTrace'; + +/** The default bound on judge jobs (pending + leased). */ +const JUDGE_JOBS_DEFAULT = 500; + +/** A claimed job: what Agent Forge needs to run the judge. */ +export interface ClaimedJudgeJob { + job_id: string; + rule: EvaluatorRuleRow; + trace: EvaluatorTrace; +} + +/** The judge queue's counters (`queued`/`leased` are current sizes, the rest totals). */ +export interface JudgeJobStats { + queued: number; + leased: number; + dropped: number; + discarded: number; + scored: number; +} + +/** A judge result as the score it becomes. */ +export interface JudgeScore { + rule_name: string; + trace_id: string; + agent_id: string | null; + value: number | null; + comment: string | null; +} + +/** What the queue needs from the worker; `now` and `maxJobs` are test seams. */ +export interface JudgeJobsDeps { + getRule(ruleId: string): EvaluatorRuleRow | undefined; + loadTrace(traceId: string): EvaluatorTrace | undefined; + /** Writes the score and returns its id. */ + writeScore(score: JudgeScore): string; + now?: () => number; + maxJobs?: number; +} + +export interface JudgeJobs { + /** Whether a job for this pair is queued or leased. */ + has(ruleId: string, traceId: string): boolean; + /** Queue a job; `false` (and counted) when the queue is full. */ + enqueue(ruleId: string, traceId: string): boolean; + claim(max: number, leaseMs: number): ClaimedJudgeJob[]; + /** The new score's id, or `undefined` for an unknown, unclaimed or expired job. */ + complete(jobId: string, result: { value: number | null; comment: string | null }): string | undefined; + stats(): JudgeJobStats; + clear(): void; +} + +interface Job { + job_id: string; + rule_id: string; + trace_id: string; + enqueued_at: number; +} + +interface Lease { + job: Job; + until: number; + rule_name: string; + agent_id: string | null; +} + +const pairKey = (ruleId: string, traceId: string): string => `${ruleId}\u0000${traceId}`; + +/** A judge-job queue over the worker's rule, trace and score access. */ +export function createJudgeJobs(deps: JudgeJobsDeps): JudgeJobs { + const now = deps.now ?? Date.now; + const maxJobs = deps.maxJobs ?? JUDGE_JOBS_DEFAULT; + const pending: Job[] = []; + const leased = new Map(); + const pairs = new Set(); + const counts = { dropped: 0, discarded: 0, scored: 0 }; + + const forget = (job: Job): void => { + pairs.delete(pairKey(job.rule_id, job.trace_id)); + }; + + /** Expired leases go back to the front, oldest job first. */ + const reclaim = (): void => { + const at = now(); + const expired = [...leased.values()].filter((lease) => lease.until <= at).map((lease) => lease.job); + if (expired.length === 0) return; + for (const job of expired) leased.delete(job.job_id); + expired.sort((a, b) => a.enqueued_at - b.enqueued_at); + pending.unshift(...expired); + }; + + /** The job's live rule and fresh trace, or `undefined` when it can no longer run. */ + const resolve = (job: Job): { rule: EvaluatorRuleRow; trace: EvaluatorTrace } | undefined => { + const rule = deps.getRule(job.rule_id); + if (rule === undefined || !rule.enabled || rule.kind !== 'judge') return undefined; + const trace = deps.loadTrace(job.trace_id); + return trace === undefined ? undefined : { rule, trace }; + }; + + return { + has: (ruleId, traceId) => pairs.has(pairKey(ruleId, traceId)), + enqueue(ruleId, traceId) { + if (pending.length + leased.size >= maxJobs) { + counts.dropped++; + return false; + } + pending.push({ job_id: randomUUID(), rule_id: ruleId, trace_id: traceId, enqueued_at: now() }); + pairs.add(pairKey(ruleId, traceId)); + return true; + }, + claim(max, leaseMs) { + reclaim(); + const claimed: ClaimedJudgeJob[] = []; + while (claimed.length < max && pending.length > 0) { + const job = pending.shift() as Job; + const live = resolve(job); + if (live === undefined) { + forget(job); + counts.discarded++; + continue; + } + leased.set(job.job_id, { job, until: now() + leaseMs, rule_name: live.rule.name, agent_id: live.trace.agent_id }); + claimed.push({ job_id: job.job_id, ...live }); + } + return claimed; + }, + complete(jobId, result) { + reclaim(); + const lease = leased.get(jobId); + if (lease === undefined) return undefined; + leased.delete(jobId); + forget(lease.job); + const scoreId = deps.writeScore({ + rule_name: lease.rule_name, + trace_id: lease.job.trace_id, + agent_id: lease.agent_id, + ...result, + }); + counts.scored++; + return scoreId; + }, + stats() { + reclaim(); + return { queued: pending.length, leased: leased.size, ...counts }; + }, + clear() { + pending.length = 0; + leased.clear(); + pairs.clear(); + }, + }; +} diff --git a/src/observe/routes/evaluators.test.ts b/src/observe/routes/evaluators.test.ts index 0ff20ed01..fc3c85671 100644 --- a/src/observe/routes/evaluators.test.ts +++ b/src/observe/routes/evaluators.test.ts @@ -294,3 +294,89 @@ describe('POST /api/v1/evaluators/:id/test', () => { }); }); }); + +describe('judge-job routes', () => { + /** A worker with one queued judge job for trace t1. */ + async function queued(database: SqlDatabase, clock: { now: number }): Promise { + ingestBatch(database, [ + { + type: 'observation', + data: { obs_id: 'root', trace_id: 't1', agent_id: 'ag', kind: 'span', name: 'invoke_agent a', start: 1, end: 11, status: 'ok' }, + }, + ]); + await create(database, judgeRule); + const worker = startEvaluatorWorker(database, { isolateAvailable: () => true, now: () => clock.now }); + worker.notifyRootsEnded(['t1']); + await worker.idle(); + return worker; + } + + it('route order: judge-jobs never reads as a rule id', () => { + expect(matchRoute('POST', '/api/v1/evaluators/judge-jobs/claim')?.route.path).toBe('/api/v1/evaluators/judge-jobs/claim'); + expect(matchRoute('POST', '/api/v1/evaluators/judge-jobs/j1/result')?.params).toEqual({ job_id: 'j1' }); + }); + + it('claims with defaults (empty body), posts a result as an online score, then 404s the spent job', async () => { + await withDb(async (database) => { + const clock = { now: 1000 }; + const worker = await queued(database, clock); + const claimed = await call(database, 'POST', '/api/v1/evaluators/judge-jobs/claim', '', worker); + expect(claimed.status).toBe(200); + const jobs = (claimed.body as { jobs: Array<{ job_id: string; rule: EvaluatorRuleRow; trace: { trace_id: string } }> }).jobs; + expect(jobs).toHaveLength(1); + expect(jobs[0].rule).toMatchObject({ name: 'helpful', kind: 'judge' }); + expect(jobs[0].trace).toMatchObject({ trace_id: 't1', agent_id: 'ag', duration_ms: 10 }); + // Leased for the default 60 s: nothing more to claim. + clock.now += 59_000; + expect((await call(database, 'POST', '/api/v1/evaluators/judge-jobs/claim', { max: 20 }, worker)).body).toEqual({ jobs: [] }); + + const path = `/api/v1/evaluators/judge-jobs/${jobs[0].job_id}/result`; + const posted = await call(database, 'POST', path, { value: 0.8, comment: 'good' }, worker); + expect(posted.status).toBe(201); + const { score_id } = posted.body as { score_id: string }; + expect(listScores(database, { trace_id: 't1', source: 'online', limit: 10 }).scores).toEqual([ + expect.objectContaining({ score_id, name: 'helpful', value: 0.8, comment: 'good' }), + ]); + const again = await call(database, 'POST', path, { value: 0.8 }, worker); + expect(again.status).toBe(404); + expect(again.body).toMatchObject(invalid); + worker.stop(); + }); + }); + + it('honours lease_ms: an expired job 404s its result and is claimable again', async () => { + await withDb(async (database) => { + const clock = { now: 1000 }; + const worker = await queued(database, clock); + const first = (await call(database, 'POST', '/api/v1/evaluators/judge-jobs/claim', { max: 1, lease_ms: 1000 }, worker)).body as { + jobs: Array<{ job_id: string }>; + }; + clock.now += 1000; + const late = await call(database, 'POST', `/api/v1/evaluators/judge-jobs/${first.jobs[0].job_id}/result`, { value: null }, worker); + expect(late.status).toBe(404); + const second = (await call(database, 'POST', '/api/v1/evaluators/judge-jobs/claim', {}, worker)).body as { jobs: Array<{ job_id: string }> }; + expect(second.jobs.map((j) => j.job_id)).toEqual([first.jobs[0].job_id]); + worker.stop(); + }); + }); + + it('400s bad claim and result bodies; 404s an unknown job', async () => { + await withDb(async (database) => { + const worker = await queued(database, { now: 1000 }); + const claim = (body: unknown) => call(database, 'POST', '/api/v1/evaluators/judge-jobs/claim', body, worker); + for (const body of [{ max: 0 }, { max: 21 }, { max: 1.5 }, { lease_ms: 999 }, { lease_ms: 300_001 }, { other: 1 }, 'nope']) { + await expect(claim(body)).rejects.toMatchObject(invalid); + } + const [job] = worker.claimJudgeJobs(1, 60_000); + const result = (body: unknown) => call(database, 'POST', `/api/v1/evaluators/judge-jobs/${job.job_id}/result`, body, worker); + for (const body of [{}, { value: 1.01 }, { value: -0.1 }, { value: '0.5' }, { value: 0.5, comment: 'x'.repeat(1001) }, { value: 0.5, extra: 1 }]) { + await expect(result(body)).rejects.toMatchObject(invalid); + } + const unknown = await call(database, 'POST', '/api/v1/evaluators/judge-jobs/nope/result', { value: 1 }, worker); + expect(unknown.status).toBe(404); + // The leased job survived the bad bodies; a null comment is accepted. + expect((await result({ value: 1, comment: null })).status).toBe(201); + worker.stop(); + }); + }); +}); diff --git a/src/observe/routes/evaluators.ts b/src/observe/routes/evaluators.ts index 0ded5bdcb..5faaeefc0 100644 --- a/src/observe/routes/evaluators.ts +++ b/src/observe/routes/evaluators.ts @@ -1,7 +1,6 @@ /** * Online evaluator rules (LangSmith "online evaluators", the config half): - * which traces to score and how. Rules are stored only; running them on new - * traces is a separate piece. + * which traces to score and how (evaluatorWorker.ts runs them on new traces). * * - `GET /api/v1/evaluators?limit&offset&kind&enabled` -> * `{ evaluators, total, limit, offset }`, oldest first (the runner's order). @@ -15,6 +14,14 @@ * a dry run of the rule on one trace, ignoring `enabled`, `filter` and * `sample_rate`; no score is written. A `judge` rule is a 409. * + * - `POST /api/v1/evaluators/judge-jobs/claim` `{ max?: 1..20 (5), lease_ms?: 1000..300000 (60000) }` + * -> `{ jobs: [{ job_id, rule, trace }] }` (possibly empty): leases queued + * judge jobs to Agent Forge, which runs them with its provider keys + * (judgeJobs.ts). An expired lease requeues the job. + * - `POST /api/v1/evaluators/judge-jobs/:job_id/result` `{ value: number in [0, 1] | null, comment?: string <= 1000 | null }` + * -> `201 { score_id }`, the `online` score; an unknown or expired job is a 404. + * These two sit before the `:id` routes so `judge-jobs` never reads as a rule id. + * * Auth is the daemon's global API-token check; nothing per-route. */ @@ -66,6 +73,19 @@ const updateBody = z.object(fieldShape).partial().strict(); const testBody = z.object({ trace_id: nonEmpty }).strict(); +const claimBody = z + .object({ + max: z.number().int().min(1).max(20).default(5), + lease_ms: z.number().int().min(1000).max(300_000).default(60_000), + }) + .strict(); + +const resultBody = z + .object({ value: z.number().finite().min(0).max(1).nullable(), comment: z.string().max(1000).nullish() }) + .strict(); + +const JUDGE_BODY_LIMIT = 16 * 1024; + const configSchemas = { code: z.object({ source: z.string().min(1).max(65_536) }).strict(), judge: z.object({ model: z.string().min(1).max(200), rubric: z.string().min(1).max(20_000) }).strict(), @@ -112,7 +132,33 @@ function listFilters(query: URLSearchParams): { kind?: 'code' | 'judge'; enabled }; } +/** Judge-job routes; listed before the `:id` routes. */ +const judgeJobRoutes: readonly Route[] = [ + { + method: 'POST', + path: '/api/v1/evaluators/judge-jobs/claim', + handle: async ({ res, readBody, evaluators }) => { + const text = await readBody(JUDGE_BODY_LIMIT); + const { max, lease_ms } = requireJsonBody(claimBody, text.trim() === '' ? '{}' : text, 'judge-job claim'); + sendJson(res, 200, { jobs: evaluators.claimJudgeJobs(max, lease_ms) }); + }, + }, + { + method: 'POST', + path: '/api/v1/evaluators/judge-jobs/:job_id/result', + handle: async ({ res, params, readBody, evaluators }) => { + const { value, comment } = requireJsonBody(resultBody, await readBody(JUDGE_BODY_LIMIT), 'judge-job result'); + const score_id = evaluators.completeJudgeJob(params.job_id, { value, comment: comment ?? null }); + if (score_id === undefined) { + return sendJson(res, 404, { error: `No leased judge job '${params.job_id}' (unknown, or its lease expired).`, code: 'LOUSHO_REQUEST_INVALID' }); + } + sendJson(res, 201, { score_id }); + }, + }, +]; + export const evaluatorsRoutes: readonly Route[] = [ + ...judgeJobRoutes, { method: 'GET', path: '/api/v1/evaluators', diff --git a/src/observe/store.ts b/src/observe/store.ts index 1a6da4847..9c52d0c59 100644 --- a/src/observe/store.ts +++ b/src/observe/store.ts @@ -937,28 +937,51 @@ export function listTraces(db: SqlDatabase, filter: TraceFilter): { traces: Trac return { traces: rows.map(traceRow), total }; } +/** One observation (span) as the daemon stores and serves it (`GET /api/v1/traces/:id`'s `observations`). */ export interface ObservationRow { + /** The observation (span) id. */ obs_id: string; + /** The trace it belongs to. */ trace_id: string; + /** The reporting session, when known. */ session_id: string | null; + /** The agent that reported it. */ agent_id: string | null; + /** The parent observation; `null` for a root. */ parent_obs_id: string | null; + /** `span`, `generation`, `tool`, ... */ kind: string; + /** The span name (e.g. `invoke_agent `). */ name: string | null; + /** Start, epoch ms. */ start: number; + /** End, epoch ms; `null` while open. */ end: number | null; + /** `ok`, `error`, `interrupted`, ...; `null` while open. */ status: string | null; + /** Captured input text, when content capture is on. */ input: string | null; + /** Captured output text, when content capture is on. */ output: string | null; + /** The model a generation called. */ model: string | null; + /** Input tokens of a generation. */ usage_input_tokens: number | null; + /** Output tokens of a generation. */ usage_output_tokens: number | null; + /** Cached input tokens of a generation. */ usage_cached_tokens: number | null; + /** Reasoning tokens of a generation. */ usage_reasoning_tokens: number | null; + /** Cost in USD, when priced. */ cost_usd: number | null; + /** The cost misses some priced parts. */ cost_partial: boolean | null; + /** The token counts are estimates. */ usage_estimated: boolean | null; + /** Some content was cut to fit the size caps. */ truncated: boolean; + /** The remaining span attributes. */ attrs: Record | null; } diff --git a/src/traces/index.ts b/src/traces/index.ts index 856f95827..7b72f28b8 100644 --- a/src/traces/index.ts +++ b/src/traces/index.ts @@ -16,3 +16,7 @@ export { type ObserveExporterOptions, type ObserveReporter, } from '../observe/reporter'; +// The trace object online evaluators receive, shared with Agent Forge's +// judge runs (pure: no node builtins). +export { evaluatorTraceFrom, type EvaluatorTrace } from '../observe/evaluatorTrace'; +export type { ObservationRow } from '../observe/store';