From 3f3bb88c6b5e545fcc8cbff216fc93772d21c4d3 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 01:26:26 +0300 Subject: [PATCH 1/4] feat(studio): LLM-judge evaluators The observe daemon holds no provider keys, so a matching judge rule now queues an in-memory job (bounded at 500, pending + leased). Agent Forge's server pulls them (POST /api/v1/evaluators/judge-jobs/claim, leased, expired leases requeue at the front), runs each judge as one direct provider call with the playground's provider resolution (no spans, so no observe feedback loop), and posts the result back (POST /api/v1/evaluators/judge-jobs/:job_id/result -> online score). Studio also handles test-on-trace for judge rules and forwards code rules. The evaluator trace builder moves to src/observe/evaluatorTrace.ts and is exported from the traces subpath so Studio builds the identical object. Co-Authored-By: Claude Opus 5.5 --- api/traces.api.md | 75 ++++ .../server/__tests__/judge.test.ts | 281 +++++++++++++++ apps/agent-forge/server/app.ts | 7 + apps/agent-forge/server/index.ts | 15 +- apps/agent-forge/server/judge.ts | 332 ++++++++++++++++++ apps/agent-forge/server/playground.ts | 2 +- docs/observability.md | 44 ++- llms-full.txt | 44 ++- src/observe/evaluatorTrace.ts | 68 ++++ src/observe/evaluatorWorker.test.ts | 112 +++++- src/observe/evaluatorWorker.ts | 130 +++---- src/observe/judgeJobs.ts | 171 +++++++++ src/observe/routes/evaluators.test.ts | 86 +++++ src/observe/routes/evaluators.ts | 50 ++- src/observe/store.ts | 23 ++ src/traces/index.ts | 4 + 16 files changed, 1372 insertions(+), 72 deletions(-) create mode 100644 apps/agent-forge/server/__tests__/judge.test.ts create mode 100644 apps/agent-forge/server/judge.ts create mode 100644 src/observe/evaluatorTrace.ts create mode 100644 src/observe/judgeJobs.ts diff --git a/api/traces.api.md b/api/traces.api.md index 850160d3..bb30b28b 100644 --- a/api/traces.api.md +++ b/api/traces.api.md @@ -7,6 +7,33 @@ // @public export const DEFAULT_TRACE_DIR = ".lousho/traces"; +// @public +export interface EvaluatorTrace { + // (undocumented) + agent_id: string | null; + // (undocumented) + duration_ms: number | null; + // (undocumented) + end: number | null; + error_message: string | null; + // (undocumented) + input: string | null; + // (undocumented) + name: string | null; + observations: ObservationRow[]; + // (undocumented) + output: string | null; + // (undocumented) + start: number; + // (undocumented) + status: string | null; + // (undocumented) + 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 +54,54 @@ export interface ListTracesOptions { limit?: number; } +// @public +export interface ObservationRow { + // (undocumented) + agent_id: string | null; + // (undocumented) + attrs: Record | null; + // (undocumented) + cost_partial: boolean | null; + // (undocumented) + cost_usd: number | null; + // (undocumented) + end: number | null; + // (undocumented) + input: string | null; + // (undocumented) + kind: string; + // (undocumented) + model: string | null; + // (undocumented) + name: string | null; + // (undocumented) + obs_id: string; + // (undocumented) + output: string | null; + // (undocumented) + parent_obs_id: string | null; + // (undocumented) + session_id: string | null; + // (undocumented) + start: number; + // (undocumented) + status: string | null; + // (undocumented) + trace_id: string; + // (undocumented) + truncated: boolean; + // (undocumented) + usage_cached_tokens: number | null; + // (undocumented) + usage_estimated: boolean | null; + // (undocumented) + usage_input_tokens: number | null; + // (undocumented) + usage_output_tokens: number | null; + // (undocumented) + usage_reasoning_tokens: number | null; +} + // @public export function observeExporter(options?: ObserveExporterOptions): ObserveReporter; 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 00000000..03c0a1a3 --- /dev/null +++ b/apps/agent-forge/server/__tests__/judge.test.ts @@ -0,0 +1,281 @@ +/** + * 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 } 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: { warn?: (m: string) => void } = {}): JudgeLoop => { + const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), 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), 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('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); + 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 1d87866c..03e22cc0 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 2c1ddee9..bafdb5d3 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,13 @@ 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`, `fetch`, `judge`, `warn`). + */ + judge?: false | Partial>; } async function listen(server: http.Server, port: number, host: string): Promise { @@ -114,6 +122,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, baseDir }); 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 +133,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()))); + }, }; } diff --git a/apps/agent-forge/server/judge.ts b/apps/agent-forge/server/judge.ts new file mode 100644 index 00000000..fb1a20e0 --- /dev/null +++ b/apps/agent-forge/server/judge.ts @@ -0,0 +1,332 @@ +/** + * 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()` (with `LOUSHO_OBSERVE_TOKEN` as the bearer + * when set), run them one at a time, post each result. 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}. */ +export type JudgeFn = (rule: JudgeRule, trace: EvaluatorTrace) => Promise; + +export interface JudgeLoopOptions { + secretsStore: SecretsStore; + /** The project dir; the default judge makes no reporter, so nothing is written under it. */ + baseDir: string; + /** Pause between ticks; default 2000 ms. */ + intervalMs?: 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 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) => { + 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, + }); + 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 }; +} + +/** `judge` never throws (a throw becomes a `judge_error` result) and never returns what the daemon would 400. */ +async function runJudge(judge: JudgeFn, rule: JudgeRule, trace: EvaluatorTrace): Promise { + try { + return acceptable(await judge(rule, trace)); + } catch (error) { + return judgeError(error); + } +} + +/** 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)); + let stopped = false; + let down = false; + let inFlight: Promise | undefined; + + /** A daemon answer: `true` when it was a 2xx; the first failure of an outage is logged. */ + const healthy = (response: globalThis.Response | undefined): response is globalThis.Response => { + if (response?.ok) { + down = false; + return true; + } + if (!down) { + down = true; + const why = response === undefined ? 'unreachable' : `HTTP ${response.status}`; + warn(`[lousho studio] judge evaluators paused: the observe daemon at ${observeTarget()} is ${why}; retrying.`); + } + return false; + }; + + const runTick = async (): Promise => { + const claimed = await callDaemon(fetchFn, '/api/v1/evaluators/judge-jobs/claim', { max: CLAIM_MAX }); + if (!healthy(claimed)) return; + const { jobs } = (await claimed.json()) as { jobs: ClaimedJob[] }; + for (const job of jobs) { + if (stopped) return; + const result = await runJudge(judge, job.rule, job.trace); + const body = { value: result.value, ...(result.comment === null ? {} : { comment: result.comment }) }; + const posted = await callDaemon(fetchFn, `/api/v1/evaluators/judge-jobs/${encodeURIComponent(job.job_id)}/result`, body); + // 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; +} + +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())); +} + +const invalid = (res: Response, status: number, error: string): void => { + res.status(status).json({ error, code: 'LOUSHO_REQUEST_INVALID' }); +}; + +/** 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: JudgeFn, fetchFn: typeof fetch): Promise { + const traceId: unknown = (req.body as { trace_id?: unknown } | undefined)?.trace_id; + if (typeof traceId !== 'string' || traceId === '') { + return invalid(res, 400, 'Invalid evaluator test body: trace_id: must be a non-empty string.'); + } + 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 runJudge(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 judge = options.judge ?? modelJudge(options.secretsStore); + 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 (id === '.' || 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 b8682ecf..0596482e 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,48 @@ 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"?: "" }` | `201 { "score_id" }`, and the job leaves the queue. 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 and runs them one at a time. +A daemon that is down only pauses it, and it logs once per outage. + +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 f5875940..251582b0 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,48 @@ 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"?: "" }` | `201 { "score_id" }`, and the job leaves the queue. 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 and runs them one at a time. +A daemon that is down only pauses it, and it logs once per outage. + +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 00000000..7de568bc --- /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 191744bb..12cb835f 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 a4da74e7..e5d4f96f 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 00000000..f3d1dd4a --- /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 0ff20ed0..0913688b 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. + expect((await result({ value: 1, comment: 'x'.repeat(1000) })).status).toBe(201); + worker.stop(); + }); + }); +}); diff --git a/src/observe/routes/evaluators.ts b/src/observe/routes/evaluators.ts index 0ded5bdc..a609e62e 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 }` + * -> `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).optional() }) + .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 1a6da484..9c52d0c5 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 856f9582..7b72f28b 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'; From 974b21d2d538e68832824aa3e555e2846ba5bb3c Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 01:39:34 +0300 Subject: [PATCH 2/4] fix(studio): bound judge calls by timeout and lease deadline Each judge call now gets an AbortSignal and a cap (judgeTimeoutMs, default 20 s, cut to what is left of the job's lease minus a 1 s post margin); a claimed job whose local lease deadline has passed is skipped and left to requeue. The claim sends lease_ms explicitly. A non-JSON 2xx claim answer goes through the once-per-outage warning. The test-on-trace route rejects unsafe ids and caps its dry run the same way. baseDir is dropped from the loop options; LOUSHO_STUDIO_JUDGE=0 turns the loop off and the e2e server sets it. The daemon accepts a null result comment. Docs note the timeout and that a renamed rule scores under its claim-time name. Co-Authored-By: Claude Opus 5.5 --- apps/agent-forge/playwright.config.ts | 2 + .../server/__tests__/judge.test.ts | 82 ++++++++++- apps/agent-forge/server/index.ts | 11 +- apps/agent-forge/server/judge.ts | 129 +++++++++++++----- docs/observability.md | 13 +- llms-full.txt | 13 +- src/observe/routes/evaluators.test.ts | 4 +- src/observe/routes/evaluators.ts | 4 +- 8 files changed, 207 insertions(+), 51 deletions(-) diff --git a/apps/agent-forge/playwright.config.ts b/apps/agent-forge/playwright.config.ts index e9e83eff..63308416 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 index 03c0a1a3..23da560e 100644 --- a/apps/agent-forge/server/__tests__/judge.test.ts +++ b/apps/agent-forge/server/__tests__/judge.test.ts @@ -15,7 +15,7 @@ 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 } from '../judge'; +import { judgePrompt, parseJudgeReply, startJudgeLoop, type JudgeFn, type JudgeLoop, type JudgeLoopOptions } from '../judge'; import { RunManager } from '../runRegistry'; import { FileCheckpointStore } from '../checkpointStore'; import { FileApprovalStore } from '../approvalStore'; @@ -71,8 +71,8 @@ describe('judge loop and test-on-trace', () => { const onlineScores = (traceId: string) => listScores(daemon.db, { trace_id: traceId, source: 'online', limit: 10 }).scores; - const loop = (judge: JudgeFn, extra: { warn?: (m: string) => void } = {}): JudgeLoop => { - const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), baseDir, intervalMs: 60_000, judge, ...extra }); + const loop = (judge: JudgeFn, extra: Partial = {}): JudgeLoop => { + const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), intervalMs: 60_000, judge, ...extra }); loops.push(started); return started; }; @@ -112,7 +112,7 @@ describe('judge loop and test-on-trace', () => { const prevKey = process.env.OPENAI_API_KEY; delete process.env.OPENAI_API_KEY; try { - const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), baseDir, intervalMs: 60_000 }); + const started = startJudgeLoop({ secretsStore: new SecretsStore(baseDir), intervalMs: 60_000 }); loops.push(started); await started.tick(); } finally { @@ -158,6 +158,74 @@ describe('judge loop and test-on-trace', () => { 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)); @@ -222,6 +290,12 @@ describe('judge loop and test-on-trace', () => { 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); diff --git a/apps/agent-forge/server/index.ts b/apps/agent-forge/server/index.ts index bafdb5d3..724931b3 100644 --- a/apps/agent-forge/server/index.ts +++ b/apps/agent-forge/server/index.ts @@ -64,9 +64,11 @@ export interface StartStudioServerOptions { * 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`, `fetch`, `judge`, `warn`). + * 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>; + judge?: false | Partial>; } async function listen(server: http.Server, port: number, host: string): Promise { @@ -123,7 +125,7 @@ export async function startStudioServer( await listen(server, port, host); const judgeOption = options.judge ?? (options.observe === false ? false : {}); - const judgeLoop = judgeOption === false ? undefined : startJudgeLoop({ ...judgeOption, secretsStore, baseDir }); + 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; @@ -162,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 index fb1a20e0..5a737036 100644 --- a/apps/agent-forge/server/judge.ts +++ b/apps/agent-forge/server/judge.ts @@ -8,9 +8,11 @@ * 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()` (with `LOUSHO_OBSERVE_TOKEN` as the bearer - * when set), run them one at a time, post each result. Ticks never - * overlap. A daemon that is down or answers non-2xx only skips the tick, + * 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 @@ -44,15 +46,23 @@ export interface JudgeResult { comment: string | null; } -/** Runs one judge rule on one trace; the default is {@link modelJudge}. */ -export type JudgeFn = (rule: JudgeRule, trace: EvaluatorTrace) => Promise; +/** + * 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; - /** The project dir; the default judge makes no reporter, so nothing is written under it. */ - baseDir: string; /** 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`. */ @@ -68,6 +78,10 @@ export interface JudgeLoop { } 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; @@ -146,7 +160,7 @@ const judgeError = (error: unknown): JudgeResult => ({ /** 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) => { + return async (rule, trace, signal) => { try { const spec: AgentSpec = { name: 'judge', prompt: '', provider: providerOf(rule.config.model) }; const { provider } = buildAgentFromSpec(spec, 'judge', undefined, { secretsStore }); @@ -156,6 +170,7 @@ function modelJudge(secretsStore: SecretsStore): JudgeFn { { role: 'user', content: judgePrompt(trace) }, ], temperature: 0, + signal, }); return parseJudgeReply(reply.text); } catch (error) { @@ -172,12 +187,31 @@ function acceptable(result: JudgeResult): JudgeResult { return { value, comment: typeof comment === 'string' ? comment.slice(0, MAX_COMMENT) : null }; } -/** `judge` never throws (a throw becomes a `judge_error` result) and never returns what the daemon would 400. */ -async function runJudge(judge: JudgeFn, rule: JudgeRule, trace: EvaluatorTrace): Promise { +/** + * 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 judge(rule, trace)); + return acceptable(await Promise.race([run, timedOut])); } catch (error) { return judgeError(error); + } finally { + clearTimeout(timer); } } @@ -212,33 +246,56 @@ 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; - /** A daemon answer: `true` when it was a 2xx; the first failure of an outage is logged. */ + /** 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; } - if (!down) { - down = true; - const why = response === undefined ? 'unreachable' : `HTTP ${response.status}`; - warn(`[lousho studio] judge evaluators paused: the observe daemon at ${observeTarget()} is ${why}; retrying.`); - } + 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 => { - const claimed = await callDaemon(fetchFn, '/api/v1/evaluators/judge-jobs/claim', { max: CLAIM_MAX }); - if (!healthy(claimed)) return; - const { jobs } = (await claimed.json()) as { jobs: ClaimedJob[] }; - for (const job of jobs) { - if (stopped) return; - const result = await runJudge(judge, job.rule, job.trace); - const body = { value: result.value, ...(result.comment === null ? {} : { comment: result.comment }) }; - const posted = await callDaemon(fetchFn, `/api/v1/evaluators/judge-jobs/${encodeURIComponent(job.job_id)}/result`, body); + // 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; } @@ -269,6 +326,8 @@ export interface JudgeTestRouteOptions { 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.'; @@ -281,15 +340,21 @@ async function passThrough(res: Response, upstream: globalThis.Response): Promis 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: JudgeFn, fetchFn: typeof fetch): Promise { +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 === '') { - return invalid(res, 400, 'Invalid evaluator test body: trace_id: must be a non-empty string.'); + 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 }); @@ -297,7 +362,7 @@ async function testJudgeRule(req: Request, res: Response, rule: JudgeRule, judge 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 runJudge(judge, rule, trace) }); + res.status(200).json({ result: await judge(rule, trace) }); } /** @@ -310,12 +375,14 @@ async function testJudgeRule(req: Request, res: Response, rule: JudgeRule, judge * is a 502. */ export function registerJudgeTestRoute(app: Express, options: JudgeTestRouteOptions): void { - const judge = options.judge ?? modelJudge(options.secretsStore); + 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 (id === '.' || id === '..') return invalid(res, 400, 'Invalid evaluator rule 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 }); diff --git a/docs/observability.md b/docs/observability.md index f0180363..8d634d48 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -697,10 +697,15 @@ 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"?: "" }` | `201 { "score_id" }`, and the job leaves the queue. 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 and runs them one at a time. -A daemon that is down only pauses it, and it logs once per outage. +| `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 diff --git a/llms-full.txt b/llms-full.txt index 251582b0..68f69b2d 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14591,10 +14591,15 @@ 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"?: "" }` | `201 { "score_id" }`, and the job leaves the queue. 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 and runs them one at a time. -A daemon that is down only pauses it, and it logs once per outage. +| `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 diff --git a/src/observe/routes/evaluators.test.ts b/src/observe/routes/evaluators.test.ts index 0913688b..fc3c8567 100644 --- a/src/observe/routes/evaluators.test.ts +++ b/src/observe/routes/evaluators.test.ts @@ -374,8 +374,8 @@ describe('judge-job routes', () => { } 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. - expect((await result({ value: 1, comment: 'x'.repeat(1000) })).status).toBe(201); + // 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 a609e62e..5faaeefc 100644 --- a/src/observe/routes/evaluators.ts +++ b/src/observe/routes/evaluators.ts @@ -18,7 +18,7 @@ * -> `{ 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 }` + * - `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. * @@ -81,7 +81,7 @@ const claimBody = z .strict(); const resultBody = z - .object({ value: z.number().finite().min(0).max(1).nullable(), comment: z.string().max(1000).optional() }) + .object({ value: z.number().finite().min(0).max(1).nullable(), comment: z.string().max(1000).nullish() }) .strict(); const JUDGE_BODY_LIMIT = 16 * 1024; From 9eccc2245d7f8f3dee2654581b170ea41b3c812c Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 02:06:44 +0300 Subject: [PATCH 3/4] chore: regenerate api/traces.api.md Co-Authored-By: Claude Opus 5.5 --- api/traces.api.md | 31 ------------------------------- 1 file changed, 31 deletions(-) diff --git a/api/traces.api.md b/api/traces.api.md index bb30b28b..af7f76ed 100644 --- a/api/traces.api.md +++ b/api/traces.api.md @@ -9,25 +9,16 @@ export const DEFAULT_TRACE_DIR = ".lousho/traces"; // @public export interface EvaluatorTrace { - // (undocumented) agent_id: string | null; - // (undocumented) duration_ms: number | null; - // (undocumented) end: number | null; error_message: string | null; - // (undocumented) input: string | null; - // (undocumented) name: string | null; observations: ObservationRow[]; - // (undocumented) output: string | null; - // (undocumented) start: number; - // (undocumented) status: string | null; - // (undocumented) trace_id: string; } @@ -56,49 +47,27 @@ export interface ListTracesOptions { // @public export interface ObservationRow { - // (undocumented) agent_id: string | null; - // (undocumented) attrs: Record | null; - // (undocumented) cost_partial: boolean | null; - // (undocumented) cost_usd: number | null; - // (undocumented) end: number | null; - // (undocumented) input: string | null; - // (undocumented) kind: string; - // (undocumented) model: string | null; - // (undocumented) name: string | null; - // (undocumented) obs_id: string; - // (undocumented) output: string | null; - // (undocumented) parent_obs_id: string | null; - // (undocumented) session_id: string | null; - // (undocumented) start: number; - // (undocumented) status: string | null; - // (undocumented) trace_id: string; - // (undocumented) truncated: boolean; - // (undocumented) usage_cached_tokens: number | null; - // (undocumented) usage_estimated: boolean | null; - // (undocumented) usage_input_tokens: number | null; - // (undocumented) usage_output_tokens: number | null; - // (undocumented) usage_reasoning_tokens: number | null; } From 4addc7317bb39af20cd08345433e3ca61c2610ec Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 02:19:40 +0300 Subject: [PATCH 4/4] test(studio): cover the queues screen in the revamp reload suite Co-Authored-By: Claude Opus 5.5 --- apps/agent-forge/e2e/revamp.spec.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/apps/agent-forge/e2e/revamp.spec.ts b/apps/agent-forge/e2e/revamp.spec.ts index ad450883..c8a6da1b 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 ]);