From ecfff3018bfa1774c365a2446c3a520afd2f562a Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 01:02:48 +0300 Subject: [PATCH 1/4] feat(observe): run code evaluators on new traces The daemon now runs enabled `code` evaluator rules on each trace whose root observation ends: filter + sample_rate decide, the rule source runs in the isolate sandbox (timeoutMs 2000), and the result is written as a `source: 'online'` score named after the rule. Ingest only hands the worker the ended roots' trace ids; matching and runs happen off the request, one at a time, with a bounded in-memory queue (drops are counted). Without isolated-vm, code rules are skipped with one warning. Adds POST /api/v1/evaluators/:id/test, a dry run on one trace that writes no score, and 'online' to the score sources. Co-Authored-By: Claude Opus 5.5 --- docs/observability.md | 54 +++- llms-full.txt | 54 +++- src/observe/daemon.ts | 10 + src/observe/envelope.ts | 4 +- src/observe/evaluatorWorker.test.ts | 359 ++++++++++++++++++++++++++ src/observe/evaluatorWorker.ts | 303 ++++++++++++++++++++++ src/observe/routes/evaluators.test.ts | 73 +++++- src/observe/routes/evaluators.ts | 23 ++ src/observe/routes/httpKit.ts | 3 + src/observe/routes/ingest.ts | 16 +- 10 files changed, 887 insertions(+), 12 deletions(-) create mode 100644 src/observe/evaluatorWorker.test.ts create mode 100644 src/observe/evaluatorWorker.ts diff --git a/docs/observability.md b/docs/observability.md index 747e3722a..61aae04ae 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -589,10 +589,10 @@ Retention never deletes datasets or examples. An example keeps its 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 only stores -rules for now: it does not compile a `source` or call a model yet. Running -rules on new traces comes separately. The routes take the same token as -ingest: +`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 +token as ingest: | Route | Answer | | --- | --- | @@ -618,6 +618,52 @@ An unknown field, an unknown `kind` or a `config` that does not match the `404` with the same code. `updated_at` moves on every `PUT`. The body may be up to 128 KiB. Retention never deletes rules. +### Online evaluators + +When a trace's root observation ends, the daemon runs every enabled `code` +rule whose `filter` matches the root (`agent_id`, `status`, and +`name_contains` on its name). A rule runs when `Math.random() < sample_rate`, +so `0` never runs and `1` always does. Each run writes one score with +`source: "online"`, `name` = the rule's name, the root's `agent_id`, and no +`obs_id`. A trace that already has an online score with that name is not scored +again. + +The rule's `source` runs as `()(trace)` in the isolate sandbox +(`sandbox({ driver: 'isolate', timeoutMs: 2000 })`). `trace` is a plain JSON +object: + +```json +{ "trace_id": "", "agent_id": "", "name": "", + "status": "ok", "start": 0, "end": 0, "duration_ms": 0, + "input": "", "output": "", "error_message": null, + "observations": [] } +``` + +`observations` holds the trace's rows, oldest first, capped at 200. The +function returns a finite number (the score's `value`) or +`{ "value": , "comment"?: "" }` (the comment is cut at +1000 characters). Anything else scores `value: null` with comment +`invalid_result`. A run that fails scores `value: null` with the sandbox error +kind as the comment: `timeout`, `oom`, or `driver`. An error your code throws +carries no kind, so it reports as `driver`. + +The isolate driver needs the optional `isolated-vm` package +(`npm install isolated-vm`). On Node 20 and later the daemon also needs +`NODE_OPTIONS=--no-node-snapshot`. Without them, code rules are skipped and +the daemon warns once. + +Rules run off the ingest request, one trace at a time, so ingest stays fast +even when an evaluator is slow or fails. Up to 1000 traces wait in memory; +past that, new traces are dropped (and counted). Waiting traces are lost when +the daemon restarts, so those traces never get online scores. + +`POST /api/v1/evaluators/:id/test` `{ "trace_id": "" }` dry-runs a +rule on one trace and answers `200 { "result": { "value", "comment" } }`. It +ignores `enabled`, `filter` and `sample_rate`, and it writes no score. An +unknown rule or trace is a `404`, and a bad body is a `400`. A `judge` rule is a +`409` (`judge evaluators run through Agent Forge`). Without the isolate it +answers `{ "result": { "value": null, "comment": "isolate_unavailable" } }`. + ### 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 44088a452..32a895f4c 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14483,10 +14483,10 @@ Retention never deletes datasets or examples. An example keeps its 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 only stores -rules for now: it does not compile a `source` or call a model yet. Running -rules on new traces comes separately. The routes take the same token as -ingest: +`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 +token as ingest: | Route | Answer | | --- | --- | @@ -14512,6 +14512,52 @@ An unknown field, an unknown `kind` or a `config` that does not match the `404` with the same code. `updated_at` moves on every `PUT`. The body may be up to 128 KiB. Retention never deletes rules. +### Online evaluators + +When a trace's root observation ends, the daemon runs every enabled `code` +rule whose `filter` matches the root (`agent_id`, `status`, and +`name_contains` on its name). A rule runs when `Math.random() < sample_rate`, +so `0` never runs and `1` always does. Each run writes one score with +`source: "online"`, `name` = the rule's name, the root's `agent_id`, and no +`obs_id`. A trace that already has an online score with that name is not scored +again. + +The rule's `source` runs as `()(trace)` in the isolate sandbox +(`sandbox({ driver: 'isolate', timeoutMs: 2000 })`). `trace` is a plain JSON +object: + +```json +{ "trace_id": "", "agent_id": "", "name": "", + "status": "ok", "start": 0, "end": 0, "duration_ms": 0, + "input": "", "output": "", "error_message": null, + "observations": [] } +``` + +`observations` holds the trace's rows, oldest first, capped at 200. The +function returns a finite number (the score's `value`) or +`{ "value": , "comment"?: "" }` (the comment is cut at +1000 characters). Anything else scores `value: null` with comment +`invalid_result`. A run that fails scores `value: null` with the sandbox error +kind as the comment: `timeout`, `oom`, or `driver`. An error your code throws +carries no kind, so it reports as `driver`. + +The isolate driver needs the optional `isolated-vm` package +(`npm install isolated-vm`). On Node 20 and later the daemon also needs +`NODE_OPTIONS=--no-node-snapshot`. Without them, code rules are skipped and +the daemon warns once. + +Rules run off the ingest request, one trace at a time, so ingest stays fast +even when an evaluator is slow or fails. Up to 1000 traces wait in memory; +past that, new traces are dropped (and counted). Waiting traces are lost when +the daemon restarts, so those traces never get online scores. + +`POST /api/v1/evaluators/:id/test` `{ "trace_id": "" }` dry-runs a +rule on one trace and answers `200 { "result": { "value", "comment" } }`. It +ignores `enabled`, `filter` and `sample_rate`, and it writes no score. An +unknown rule or trace is a `404`, and a bad body is a `400`. A `judge` rule is a +`409` (`judge evaluators run through Agent Forge`). Without the isolate it +answers `{ "result": { "value": null, "comment": "isolate_unavailable" } }`. + ### Annotation queues An annotation queue is a named list of traces for a person to review one by diff --git a/src/observe/daemon.ts b/src/observe/daemon.ts index 2f0647fec..06afe10f7 100644 --- a/src/observe/daemon.ts +++ b/src/observe/daemon.ts @@ -27,6 +27,7 @@ import { OBSERVE_DEFAULT_PORT, OBSERVE_SERVICE } from './envelope'; import { BodyTooLargeError, matchRoute, sendJson, type RouteContext } from './routes/index'; import { createAlerter, resolveAlertConfig, type AlertConfig } from './alerts'; import { startStatusSweep, type StatusSweep } from './sweep'; +import { startEvaluatorWorker, type EvaluatorWorker, type EvaluatorWorkerOptions } from './evaluatorWorker'; import { openDb, verifyToken } from './store'; /** `~/.lousho/observe.db`. */ @@ -91,6 +92,8 @@ export interface ObserveDaemonOptions { * their agent goes down. */ sessionIdleMinutes?: number; + /** Seams of the online-evaluator worker (tests inject a fake sandbox); defaults run code rules in the isolate. */ + evaluators?: EvaluatorWorkerOptions; } /** `lousho observe start`'s default retention window: finished traces older than 30 days are pruned daily. */ @@ -115,6 +118,8 @@ export interface ObserveDaemon { alerts: AlertConfig; /** The running status sweep (`run()` forces a pass; `stop()` is called by `close()`). */ sweep: StatusSweep; + /** The online-evaluator worker: scores each new trace whose root ends (`stop()` is called by `close()`). */ + evaluators: EvaluatorWorker; close(): Promise; } @@ -176,6 +181,7 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr const alertConfig = resolveAlertConfig(dbPath, options.webhooks ?? []); const db = options.db ?? openDb(dbPath); const requireToken = options.requireToken === true || !isLoopbackHost(host); + const evaluators = startEvaluatorWorker(db, options.evaluators); const onRequest = async (req: http.IncomingMessage, res: http.ServerResponse): Promise => { const { pathname, searchParams } = new URL(req.url ?? '/', 'http://localhost'); @@ -195,6 +201,7 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr res, params: match.params, query: searchParams, + evaluators, readBody(limitBytes: number) { bodyRead ??= readBodyCapped(req, limitBytes); return bodyRead; @@ -232,6 +239,7 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr // A failed bind (EADDRINUSE and friends) must not leak the db handle the // daemon opened above - the CLI exits on the error, but tests and embedded // callers would keep the WAL lock forever. + evaluators.stop(); if (!options.db) db.close(); throw error; } @@ -270,9 +278,11 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr db, alerts: alertConfig, sweep, + evaluators, ...(pidPath !== undefined ? { pidPath } : {}), close: async () => { sweep.stop(); + evaluators.stop(); if (pidPath !== undefined) { try { fs.rmSync(pidPath, { force: true }); diff --git a/src/observe/envelope.ts b/src/observe/envelope.ts index d73698509..fab967d5d 100644 --- a/src/observe/envelope.ts +++ b/src/observe/envelope.ts @@ -44,8 +44,8 @@ export const OBSERVE_MAX_BATCH_EVENTS = 1000; const OBSERVATION_KINDS = ['span', 'generation', 'tool', 'event'] as const; export type ObservationKind = (typeof OBSERVATION_KINDS)[number]; -/** A score's origin: an eval assertion, an llmJudge verdict, or human feedback. */ -export const SCORE_SOURCES = ['eval', 'judge', 'human'] as const; +/** A score's origin: an eval assertion, an llmJudge verdict, human feedback, or an online evaluator rule. */ +export const SCORE_SOURCES = ['eval', 'judge', 'human', 'online'] as const; /** `agent.heartbeat` - one row of the fleet registry, refreshed on the reporter's interval. */ const agentHeartbeatData = z.object({ diff --git a/src/observe/evaluatorWorker.test.ts b/src/observe/evaluatorWorker.test.ts new file mode 100644 index 000000000..10696f3eb --- /dev/null +++ b/src/observe/evaluatorWorker.test.ts @@ -0,0 +1,359 @@ +/** + * The online-evaluator worker: code rules score new traces whose root ended, + * through an injected fake sandbox (the real isolate only runs in the compat + * CI job; see the last block). Also pins that ingest stays 2xx and fast when + * the worker or its sandbox fails. + */ +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 { + rootsEndedIn, + startEvaluatorWorker, + toEvaluatorResult, + type EvaluatorWorker, + type EvaluatorWorkerOptions, +} from './evaluatorWorker'; +import { startObserveDaemon, type ObserveDaemon } from './daemon'; +import { ingestBatch, listScores, openDb, type ScoreRow } from './store'; +import type { IngestEvent } from './envelope'; +import { SandboxError } from '../sandbox'; +import type { SqlDatabase } from '../storage/sqlite/driver'; +import { ingestRoutes } from './routes/ingest'; + +const T0 = 1_700_000_000_000; + +const dbs: SqlDatabase[] = []; +const workers: EvaluatorWorker[] = []; +const daemons: ObserveDaemon[] = []; +afterEach(async () => { + while (workers.length > 0) workers.pop()?.stop(); + while (daemons.length > 0) await daemons.pop()?.close(); + while (dbs.length > 0) dbs.pop()?.close(); +}); + +function newDb(): SqlDatabase { + const db = openDb(':memory:'); + dbs.push(db); + return db; +} + +const rootEnd = (traceId: string, over: Record = {}): IngestEvent => + ({ + type: 'observation', + data: { + obs_id: `${traceId}-root`, + trace_id: traceId, + agent_id: 'support', + kind: 'span', + name: 'invoke_agent refund-bot', + start: T0, + end: T0 + 250, + status: 'ok', + input: 'hi', + output: 'hello', + ...over, + }, + }) as IngestEvent; + +const child = (traceId: string): IngestEvent => + ({ + type: 'observation', + data: { + obs_id: `${traceId}-tool`, + trace_id: traceId, + parent_obs_id: `${traceId}-root`, + agent_id: 'support', + kind: 'tool', + name: 'execute_tool lookup', + start: T0 + 10, + end: T0 + 20, + status: 'error', + attrs: { error: 'lookup failed' }, + }, + }) as IngestEvent; + +const codeRule = (over: Partial = {}): EvaluatorRuleFields => + ({ name: 'len', enabled: true, filter: {}, sample_rate: 1, kind: 'code', config: { source: '(t) => 1' }, ...over }) as EvaluatorRuleFields; + +interface Harness { + db: SqlDatabase; + worker: EvaluatorWorker; + runs: Array<{ code: string; trace: Record }>; + warnings: string[]; +} + +/** A worker over a fresh db whose sandbox calls `impl` instead of an isolate. */ +function harness(impl: (code: string, trace: Record) => unknown = () => 1, opts: EvaluatorWorkerOptions = {}): Harness { + const db = newDb(); + const runs: Harness['runs'] = []; + const warnings: string[] = []; + const worker = startEvaluatorWorker(db, { + isolateAvailable: () => true, + warn: (message) => warnings.push(message), + sandbox: (options) => { + expect(options).toEqual({ driver: 'isolate', timeoutMs: 2000 }); + return { + driver: 'fake', + run: async (code, callOpts) => { + const trace = callOpts?.args?.[0] as Record; + runs.push({ code, trace }); + return impl(code, trace); + }, + }; + }, + ...opts, + }); + workers.push(worker); + return { db, worker, runs, warnings }; +} + +async function ingestAndWait(h: Harness, batch: IngestEvent[]): Promise { + ingestBatch(h.db, batch); + h.worker.notifyRootsEnded(rootsEndedIn(batch)); + await h.worker.idle(); +} + +const onlineScores = (db: SqlDatabase, traceId: string): ScoreRow[] => + listScores(db, { trace_id: traceId, source: 'online', limit: 100 }).scores; + +describe('evaluator worker', () => { + it('scores a matching trace once, as an online score named after the rule', async () => { + const h = harness(() => ({ value: 0.75, comment: 'ok-ish' })); + createEvaluatorRule(h.db, codeRule({ filter: { agent_id: 'support', status: 'ok', name_contains: 'refund' } })); + await ingestAndWait(h, [rootEnd('t1'), child('t1')]); + + expect(h.runs).toHaveLength(1); + expect(h.runs[0].code).toBe('(t) => 1'); + expect(h.runs[0].trace).toMatchObject({ + trace_id: 't1', + agent_id: 'support', + name: 'invoke_agent refund-bot', + status: 'ok', + start: T0, + end: T0 + 250, + duration_ms: 250, + input: 'hi', + output: 'hello', + error_message: 'lookup failed', + }); + expect((h.runs[0].trace.observations as unknown[]).length).toBe(2); + expect(onlineScores(h.db, 't1')).toEqual([ + expect.objectContaining({ name: 'len', value: 0.75, comment: 'ok-ish', source: 'online', agent_id: 'support', obs_id: null }), + ]); + expect(h.worker.stats()).toMatchObject({ ran: 1, queued: 0, dropped: 0, skipped: 0 }); + + // The same root ending again does not score twice. + await ingestAndWait(h, [rootEnd('t1')]); + expect(onlineScores(h.db, 't1')).toHaveLength(1); + }); + + it('never scores a trace a filter key rules out', async () => { + for (const filter of [{ agent_id: 'billing' }, { status: 'error' as const }, { name_contains: 'onboarding' }]) { + const h = harness(); + createEvaluatorRule(h.db, codeRule({ filter })); + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.runs).toEqual([]); + expect(onlineScores(h.db, 't1')).toEqual([]); + } + }); + + it('ignores child ends, open roots, disabled rules, judge rules and sample_rate 0', async () => { + const h = harness(); + createEvaluatorRule(h.db, codeRule({ name: 'off', enabled: false })); + createEvaluatorRule(h.db, codeRule({ name: 'never', sample_rate: 0 })); + createEvaluatorRule(h.db, { name: 'judge', enabled: true, filter: {}, sample_rate: 1, kind: 'judge', config: { model: 'm', rubric: 'r' } }); + expect(rootsEndedIn([child('t1'), rootEnd('t2', { end: undefined })])).toEqual([]); + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.runs).toEqual([]); + expect(onlineScores(h.db, 't1')).toEqual([]); + expect(h.worker.stats()).toMatchObject({ ran: 0, skipped: 0 }); + }); + + it('samples with random() < sample_rate', async () => { + const draws = [0.2, 0.6]; + const h = harness(() => 1, { random: () => draws.shift() ?? 1 }); + createEvaluatorRule(h.db, codeRule({ sample_rate: 0.5 })); + await ingestAndWait(h, [rootEnd('t1')]); + await ingestAndWait(h, [rootEnd('t2')]); + expect(onlineScores(h.db, 't1')).toHaveLength(1); + expect(onlineScores(h.db, 't2')).toEqual([]); + }); + + it('turns a throwing evaluator into value null with the error kind', async () => { + const h = harness((code) => { + if (code.includes('slow')) throw new SandboxError('too slow', 'timeout', 'isolate'); + throw new RangeError('boom'); + }); + createEvaluatorRule(h.db, codeRule({ name: 'throws' })); + createEvaluatorRule(h.db, codeRule({ name: 'slow', config: { source: '(t) => slow()' } })); + await ingestAndWait(h, [rootEnd('t1')]); + const byName = Object.fromEntries(onlineScores(h.db, 't1').map((s) => [s.name, s])); + expect(byName.throws).toMatchObject({ value: null, comment: 'driver' }); + expect(byName.slow).toMatchObject({ value: null, comment: 'timeout' }); + }); + + it('normalizes what an evaluator returns', () => { + expect(toEvaluatorResult(1)).toEqual({ value: 1, comment: null }); + expect(toEvaluatorResult({ value: null, comment: 7 })).toEqual({ value: null, comment: '7' }); + expect(toEvaluatorResult({ value: 2, comment: 'x'.repeat(2000) }).comment).toHaveLength(1000); + for (const bad of [Number.NaN, Infinity, 'one', null, undefined, {}, { value: '1' }, [1]]) { + expect(toEvaluatorResult(bad)).toEqual({ value: null, comment: 'invalid_result' }); + } + }); + + it('skips code rules without the isolate and warns once', async () => { + const h = harness(() => 1, { isolateAvailable: () => false }); + createEvaluatorRule(h.db, codeRule()); + await ingestAndWait(h, [rootEnd('t1')]); + await ingestAndWait(h, [rootEnd('t2')]); + expect(h.runs).toEqual([]); + expect(onlineScores(h.db, 't1')).toEqual([]); + expect(h.worker.stats().skipped).toBe(2); + expect(h.warnings).toHaveLength(1); + expect(h.warnings[0]).toMatch(/isolated-vm/); + expect(h.warnings[0]).toMatch(/--no-node-snapshot/); + }); + + it('probes for isolated-vm by default (runs with it, skips without)', async () => { + const h = harness(() => 1, { isolateAvailable: undefined }); + createEvaluatorRule(h.db, codeRule()); + await ingestAndWait(h, [rootEnd('t1')]); + expect(h.worker.stats()).toMatchObject(hasIsolatedVm ? { ran: 1, skipped: 0 } : { ran: 0, skipped: 1 }); + expect(h.warnings).toHaveLength(hasIsolatedVm ? 0 : 1); + }); + + it('drops and counts traces past maxQueue', async () => { + const h = harness(() => 1, { maxQueue: 2 }); + createEvaluatorRule(h.db, codeRule()); + ingestBatch(h.db, [rootEnd('t1'), rootEnd('t2'), rootEnd('t3')]); + h.worker.notifyRootsEnded(['t1', 't2', 't3']); + expect(h.worker.stats()).toMatchObject({ queued: 2, dropped: 1 }); + await h.worker.idle(); + expect(h.worker.stats()).toMatchObject({ queued: 0, dropped: 1, ran: 2 }); + expect(onlineScores(h.db, 't3')).toEqual([]); + }); + + it('runs off the caller: notifyRootsEnded returns before any rule runs', () => { + const h = harness(); + createEvaluatorRule(h.db, codeRule()); + ingestBatch(h.db, [rootEnd('t1')]); + h.worker.notifyRootsEnded(['t1']); + expect(h.runs).toEqual([]); + }); + + it('stop() drops the queue and an in-flight run writes nothing', async () => { + let release: () => void = () => undefined; + const h = harness(() => new Promise((resolve) => (release = () => resolve(1)))); + createEvaluatorRule(h.db, codeRule()); + ingestBatch(h.db, [rootEnd('t1'), rootEnd('t2')]); + h.worker.notifyRootsEnded(['t1', 't2']); + await new Promise((resolve) => setTimeout(resolve, 10)); + h.worker.stop(); + release(); + await h.worker.idle(); + expect(onlineScores(h.db, 't1')).toEqual([]); + expect(h.worker.stats()).toMatchObject({ queued: 0, ran: 0 }); + h.worker.notifyRootsEnded(['t2']); + expect(h.worker.stats().queued).toBe(0); + }); +}); + +describe('ingest with the evaluator worker', () => { + it('still answers 202 when the worker throws', async () => { + const db = newDb(); + let status = 0; + const res = { + writeHead(code: number) { + status = code; + return res; + }, + end() {}, + } as unknown as http.ServerResponse; + const throwing = { + notifyRootsEnded() { + throw new Error('worker down'); + }, + } as unknown as EvaluatorWorker; + await ingestRoutes[0].handle({ + db, + dbPath: ':memory:', + req: {} as http.IncomingMessage, + res, + params: {}, + query: new URLSearchParams(), + readBody: () => Promise.resolve(JSON.stringify({ batch: [rootEnd('t1')] })), + evaluators: throwing, + }); + expect(status).toBe(202); + }); + + it('a daemon acks fast while a failing sandbox scores the trace null', async () => { + const daemon = await startObserveDaemon({ + dbPath: ':memory:', + port: 0, + sweepIntervalMs: 0, + evaluators: { + isolateAvailable: () => true, + sandbox: () => ({ + driver: 'fake', + run: async () => { + await new Promise((resolve) => setTimeout(resolve, 200)); + throw new SandboxError('driver gone', 'driver', 'isolate'); + }, + }), + }, + }); + daemons.push(daemon); + createEvaluatorRule(daemon.db, codeRule()); + const started = Date.now(); + const res = await fetch(`http://127.0.0.1:${daemon.port}/api/v1/ingest`, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify({ batch: [rootEnd('t1')] }), + }); + expect(res.status).toBe(202); + expect(Date.now() - started).toBeLessThan(200); + await daemon.evaluators.idle(); + expect(onlineScores(daemon.db, 't1')).toEqual([expect.objectContaining({ value: null, comment: 'driver' })]); + }); +}); + +/** The real isolate (compat CI only: isolated-vm installed and NODE_OPTIONS=--no-node-snapshot). */ +function probeIsolatedVm(): { ok: boolean } { + const needsFlag = Number(process.versions.node.split('.')[0]) >= 20; + const hasFlag = `${process.env.NODE_OPTIONS ?? ''} ${process.execArgv.join(' ')}`.includes('--no-node-snapshot'); + if (needsFlag && !hasFlag) return { ok: false }; + try { + createRequire(import.meta.url)('isolated-vm'); + return { ok: true }; + } catch { + return { ok: false }; + } +} +const hasIsolatedVm = probeIsolatedVm().ok; + +describe.runIf(hasIsolatedVm)('evaluator worker (real isolate)', () => { + it('scores a matching trace from the rule source', async () => { + const db = newDb(); + const worker = startEvaluatorWorker(db); + workers.push(worker); + createEvaluatorRule(db, codeRule({ config: { source: '(trace) => ({ value: trace.duration_ms / 1000, comment: trace.name })' } })); + ingestBatch(db, [rootEnd('t1')]); + worker.notifyRootsEnded(['t1']); + await worker.idle(); + expect(onlineScores(db, 't1')).toEqual([expect.objectContaining({ value: 0.25, comment: 'invoke_agent refund-bot' })]); + }); + + it("reports a user-code throw as 'driver' (the isolate passes it through without a kind)", async () => { + const db = newDb(); + const worker = startEvaluatorWorker(db); + workers.push(worker); + createEvaluatorRule(db, codeRule({ config: { source: '() => { throw new Error("boom") }' } })); + ingestBatch(db, [rootEnd('t1')]); + worker.notifyRootsEnded(['t1']); + await worker.idle(); + expect(onlineScores(db, 't1')).toEqual([expect.objectContaining({ value: null, comment: 'driver' })]); + }); +}); diff --git a/src/observe/evaluatorWorker.ts b/src/observe/evaluatorWorker.ts new file mode 100644 index 000000000..76a1451ef --- /dev/null +++ b/src/observe/evaluatorWorker.ts @@ -0,0 +1,303 @@ +/** + * Online evaluators, the running half: when a trace's root observation ends, + * every enabled rule whose `filter` matches (and whose `sample_rate` draw + * passes) scores the trace once, as a `source: 'online'` score named after + * the rule. + * + * - Ingest only hands over trace ids ({@link EvaluatorWorker.notifyRootsEnded}): + * synchronous, O(batch), never throws. Rule matching, trace loading and the + * sandbox runs happen later, off the request, one trace at a time. + * - The pending queue is in memory and bounded (`maxQueue`, default 1000); + * past it a trace drops and counts in `stats().dropped`. Pending traces are + * lost on a daemon restart. + * - `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. + */ + +import { randomUUID } from 'node:crypto'; +import * as path from 'node:path'; +import { createRequire } from 'node:module'; + +import { sandbox as realSandbox, sandboxErrorKind, type Sandbox, type SandboxOptions } from '../sandbox'; +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'; + +/** What one evaluator run produced: the score's value and comment. */ +export interface EvaluatorResult { + value: number | null; + comment: string | null; +} + +/** The worker's counters since it started. */ +export interface EvaluatorWorkerStats { + /** Traces waiting to be evaluated. */ + queued: number; + /** Traces dropped because the queue was full. */ + dropped: number; + /** Rule runs that wrote a score. */ + ran: number; + /** Rule runs skipped because no runner could take them (no isolate). */ + skipped: number; +} + +/** Options of {@link startEvaluatorWorker}; every one is a test seam with a production default. */ +export interface EvaluatorWorkerOptions { + /** Builds the box a code rule runs in; default the SDK's `sandbox`. */ + sandbox?: (options: SandboxOptions) => Sandbox; + /** The sampling draw in `[0, 1)`; default `Math.random`. */ + random?: () => number; + /** Where warnings go; default `console.warn`. */ + warn?: (message: string) => void; + /** Most traces waiting at once; default 1000. */ + maxQueue?: number; + /** Whether the isolate driver can run here; default a cached probe for `isolated-vm`. */ + isolateAvailable?: () => boolean; +} + +/** The running worker the daemon owns. */ +export interface EvaluatorWorker { + /** Queue these traces (their root ended). Synchronous, never throws, never waits. */ + notifyRootsEnded(traceIds: readonly string[]): void; + /** Run `rule` on `traceId` once, ignoring `enabled`, `filter` and `sample_rate`; writes nothing. */ + runOnce(rule: EvaluatorRuleRow, traceId: string): Promise; + /** Resolves once the queue is empty and nothing is running. */ + idle(): Promise; + stats(): EvaluatorWorkerStats; + /** Drop the queue and stop taking traces; an in-flight run writes nothing. */ + stop(): void; +} + +/** The default bound on traces waiting to be evaluated. */ +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; + +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 + * not load the native module. + */ +function probeIsolate(): boolean { + const needsFlag = Number(process.versions.node.split('.')[0]) >= 20; + const hasFlag = `${process.env.NODE_OPTIONS ?? ''} ${process.execArgv.join(' ')}`.includes('--no-node-snapshot'); + if (needsFlag && !hasFlag) return false; + try { + createRequire(path.join(moduleDir(), 'noop.js')).resolve('isolated-vm'); + return true; + } catch { + return false; + } +} + +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. */ +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), + }; +} + +/** Whether every key the filter sets matches the trace's root. */ +function matchesFilter(filter: EvaluatorRuleFilter, trace: EvaluatorTrace): boolean { + if (filter.agent_id !== undefined && trace.agent_id !== filter.agent_id) return false; + if (filter.status !== undefined && trace.status !== filter.status) return false; + if (filter.name_contains !== undefined && !(trace.name ?? '').includes(filter.name_contains)) return false; + return true; +} + +/** An evaluator's return value as a score: a finite number, or `{ value: number | null, comment? }`. */ +export function toEvaluatorResult(raw: unknown): EvaluatorResult { + if (typeof raw === 'number') return Number.isFinite(raw) ? { value: raw, comment: null } : invalidResult(); + if (raw === null || typeof raw !== 'object' || !('value' in raw)) return invalidResult(); + const { value, comment } = raw as { value: unknown; comment?: unknown }; + if (value !== null && !(typeof value === 'number' && Number.isFinite(value))) return invalidResult(); + return { value, comment: comment === undefined || comment === null ? null : String(comment).slice(0, MAX_COMMENT) }; +} + +const invalidResult = (): EvaluatorResult => ({ value: null, comment: 'invalid_result' }); + +/** The trace ids of root observations this batch ended, each once. */ +export function rootsEndedIn(batch: readonly IngestEvent[]): string[] { + const ids = new Set(); + for (const event of batch) { + if (event.type === 'observation' && event.data.parent_obs_id === undefined && event.data.end !== undefined) { + ids.add(event.data.trace_id); + } + } + return [...ids]; +} + +/** Start the worker over `db`. */ +export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptions = {}): EvaluatorWorker { + const makeSandbox = opts.sandbox ?? realSandbox; + const random = opts.random ?? Math.random; + const warn = opts.warn ?? ((message: string) => console.warn(message)); + const maxQueue = opts.maxQueue ?? EVALUATOR_QUEUE_DEFAULT; + const isolateAvailable = opts.isolateAvailable ?? defaultIsolateAvailable; + const counts = { dropped: 0, ran: 0, skipped: 0 }; + const pending: string[] = []; + let stopped = false; + let warned = false; + let draining: Promise | undefined; + + const runCode = async (rule: EvaluatorRuleRow, trace: EvaluatorTrace): Promise => { + if (rule.kind !== 'code') return { value: null, comment: 'unsupported_kind' }; + try { + const box = makeSandbox({ driver: 'isolate', timeoutMs: CODE_TIMEOUT_MS }); + return toEvaluatorResult(await box.run(rule.config.source, { args: [trace] })); + } catch (error) { + return { value: null, comment: sandboxErrorKind(error) ?? 'driver' }; + } + }; + + /** 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 => { + ingestBatch(db, [ + { + type: 'score', + data: { + score_id: randomUUID(), + trace_id: trace.trace_id, + name: rule.name, + value: result.value ?? undefined, + source: 'online', + comment: result.comment ?? undefined, + agent_id: trace.agent_id ?? undefined, + }, + }, + ]); + }; + + const skipCode = (): void => { + counts.skipped++; + if (warned) return; + warned = true; + warn(ISOLATE_WARNING); + }; + + 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 (!isolateAvailable()) return skipCode(); + const result = await runner(trace); + if (stopped) return; + writeScore(rule, trace, result); + counts.ran++; + }; + + const evaluateTrace = async (traceId: string): Promise => { + const rules = listEvaluatorRules(db, { limit: RULES_PAGE, offset: 0, enabled: true }).evaluators; + if (rules.length === 0) return; + const trace = evaluatorTrace(db, traceId); + if (trace === undefined) return; + for (const rule of rules) { + if (stopped) return; + await evaluateRule(rule, trace); + } + }; + + const drain = async (): Promise => { + await new Promise((resolve) => setImmediate(resolve)); + while (!stopped && pending.length > 0) { + const traceId = pending.shift() as string; + try { + await evaluateTrace(traceId); + } catch (error) { + if (!stopped) warn(`[lousho observe] evaluating trace '${traceId}' failed: ${error instanceof Error ? error.message : String(error)}`); + } + } + }; + + const schedule = (): void => { + if (draining !== undefined || stopped) return; + draining = drain().finally(() => { + draining = undefined; + if (pending.length > 0) schedule(); + }); + }; + + return { + notifyRootsEnded(traceIds) { + if (stopped) return; + for (const traceId of traceIds) { + if (pending.length >= maxQueue) counts.dropped++; + else pending.push(traceId); + } + schedule(); + }, + async runOnce(rule, traceId) { + const trace = evaluatorTrace(db, traceId); + if (trace === undefined) return { value: null, comment: 'trace_missing' }; + const runner = runnerFor(rule); + if (runner === undefined) return { value: null, comment: 'unsupported_kind' }; + if (!isolateAvailable()) return { value: null, comment: 'isolate_unavailable' }; + return runner(trace); + }, + async idle() { + while (draining !== undefined) await draining; + }, + stats: () => ({ queued: pending.length, ...counts }), + stop() { + stopped = true; + pending.length = 0; + }, + }; +} diff --git a/src/observe/routes/evaluators.test.ts b/src/observe/routes/evaluators.test.ts index d26d5ea46..9fb8cb78e 100644 --- a/src/observe/routes/evaluators.test.ts +++ b/src/observe/routes/evaluators.test.ts @@ -6,6 +6,8 @@ import { describe, expect, it } from 'vitest'; import type * as http from 'node:http'; import { openDb } from '../store'; import type { EvaluatorRuleRow } from '../evaluators'; +import { startEvaluatorWorker, type EvaluatorWorker } from '../evaluatorWorker'; +import { ingestBatch, listScores } from '../store'; import type { SqlDatabase } from '../../storage/sqlite/driver'; // Through the registry, production's load order (see scores.test.ts). import { matchRoute } from './index'; @@ -20,7 +22,8 @@ async function call( database: SqlDatabase, method: 'GET' | 'POST' | 'PUT' | 'DELETE', path: string, - body?: unknown + body?: unknown, + evaluators?: EvaluatorWorker ): Promise { const captured: Captured = { status: 0, body: undefined }; const res = { @@ -44,6 +47,7 @@ async function call( params: matched.params, query: new URLSearchParams(search), readBody: () => Promise.resolve(text), + evaluators, }); return captured; } @@ -215,3 +219,70 @@ describe('evaluator rules API', () => { }); }); }); + +describe('POST /api/v1/evaluators/:id/test', () => { + const seedTrace = (database: SqlDatabase): void => { + 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' }, + }, + ]); + }; + const fakeWorker = (database: SqlDatabase, run: (code: string, args: unknown[]) => unknown, available = true) => + startEvaluatorWorker(database, { + isolateAvailable: () => available, + sandbox: () => ({ driver: 'fake', run: async (code, opts) => run(code, opts?.args ?? []) }), + }); + + it('dry-runs the rule on the trace, ignoring enabled/filter/sample_rate, and writes no score', async () => { + await withDb(async (database) => { + seedTrace(database); + const created = rule( + await create(database, { ...codeRule, enabled: false, sample_rate: 0, filter: { agent_id: 'other' } }) + ); + const seen: unknown[] = []; + const worker = fakeWorker(database, (code, args) => { + seen.push(code, args[0]); + return { value: 0.5, comment: 'half' }; + }); + const tested = await call(database, 'POST', `/api/v1/evaluators/${created.rule_id}/test`, { trace_id: 't1' }, worker); + expect(tested.status).toBe(200); + expect(tested.body).toEqual({ result: { value: 0.5, comment: 'half' } }); + expect(seen[0]).toBe('(trace) => 1'); + expect(seen[1]).toMatchObject({ trace_id: 't1', agent_id: 'ag', duration_ms: 10, status: 'ok' }); + expect(listScores(database, { trace_id: 't1', limit: 10 }).scores).toEqual([]); + }); + }); + + it('answers isolate_unavailable without the isolate', async () => { + await withDb(async (database) => { + seedTrace(database); + const created = rule(await create(database, codeRule)); + const worker = fakeWorker(database, () => 1, false); + const tested = await call(database, 'POST', `/api/v1/evaluators/${created.rule_id}/test`, { trace_id: 't1' }, worker); + expect(tested.body).toEqual({ result: { value: null, comment: 'isolate_unavailable' } }); + }); + }); + + it('404s an unknown rule or trace, 400s a bad body, 409s a judge rule', async () => { + await withDb(async (database) => { + seedTrace(database); + const worker = fakeWorker(database, () => 1); + const code = rule(await create(database, codeRule)); + const judge = rule(await create(database, judgeRule)); + const unknownRule = await call(database, 'POST', '/api/v1/evaluators/nope/test', { trace_id: 't1' }, worker); + expect(unknownRule.status).toBe(404); + const unknownTrace = await call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, { trace_id: 'nope' }, worker); + expect(unknownTrace.status).toBe(404); + expect(unknownTrace.body).toMatchObject(invalid); + await expect(call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, {}, worker)).rejects.toMatchObject(invalid); + await expect( + call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, { trace_id: 't1', x: 1 }, worker) + ).rejects.toMatchObject(invalid); + const judged = await call(database, 'POST', `/api/v1/evaluators/${judge.rule_id}/test`, { trace_id: 't1' }, worker); + expect(judged.status).toBe(409); + expect(judged.body).toEqual({ error: 'judge evaluators run through Agent Forge', code: 'LOUSHO_REQUEST_INVALID' }); + }); + }); +}); diff --git a/src/observe/routes/evaluators.ts b/src/observe/routes/evaluators.ts index ee62fb7da..847299ea6 100644 --- a/src/observe/routes/evaluators.ts +++ b/src/observe/routes/evaluators.ts @@ -11,6 +11,9 @@ * - `PUT /api/v1/evaluators/:id` any subset of the create fields; the merged * rule is validated as a whole -> `{ evaluator }`. * - `DELETE /api/v1/evaluators/:id` -> `204`. + * - `POST /api/v1/evaluators/:id/test` `{ trace_id }` -> `{ result: { value, comment } }`: + * 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. * * Auth is the daemon's global API-token check; nothing per-route. */ @@ -27,6 +30,7 @@ import { updateEvaluatorRule, type EvaluatorRuleFields, } from '../evaluators'; +import { traceDetail } from '../store'; import { intParam, requireJsonBody, sendJson, type Route } from './httpKit'; const RULE_BODY_LIMIT = 128 * 1024; @@ -61,6 +65,8 @@ const createBody = z const updateBody = z.object(fieldShape).partial().strict(); +const testBody = z.object({ trace_id: nonEmpty }).strict(); + 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(), @@ -151,6 +157,23 @@ export const evaluatorsRoutes: readonly Route[] = [ sendJson(res, 200, { evaluator: updated }); }, }, + { + method: 'POST', + path: '/api/v1/evaluators/:id/test', + handle: async ({ db, res, params, readBody, evaluators }) => { + const { trace_id } = requireJsonBody(testBody, await readBody(RULE_BODY_LIMIT), 'evaluator test'); + const rule = getEvaluatorRule(db, params.id); + if (rule === undefined) return notFound(res, params.id); + if (rule.kind === 'judge') { + return sendJson(res, 409, { error: 'judge evaluators run through Agent Forge', code: 'LOUSHO_REQUEST_INVALID' }); + } + if (traceDetail(db, trace_id) === undefined) { + return sendJson(res, 404, { error: `No trace '${trace_id}'.`, code: 'LOUSHO_REQUEST_INVALID' }); + } + if (evaluators === undefined) throw new Error('the evaluator worker is not running'); + sendJson(res, 200, { result: await evaluators.runOnce(rule, trace_id) }); + }, + }, { method: 'DELETE', path: '/api/v1/evaluators/:id', diff --git a/src/observe/routes/httpKit.ts b/src/observe/routes/httpKit.ts index d76dc0ff3..b874db17c 100644 --- a/src/observe/routes/httpKit.ts +++ b/src/observe/routes/httpKit.ts @@ -11,6 +11,7 @@ import { z } from 'zod'; import { SDKError } from '../../execution/errors'; import type { SqlDatabase } from '../../storage/sqlite/driver'; import { sendJson } from '../../server/sendJson'; +import type { EvaluatorWorker } from '../evaluatorWorker'; export { sendJson }; @@ -29,6 +30,8 @@ export interface RouteContext { * `LOUSHO_REQUEST_INVALID` on a second call. */ readBody(limitBytes: number): Promise; + /** The daemon's online-evaluator worker (absent in route tests that do not need it). */ + evaluators?: EvaluatorWorker; } export interface Route { diff --git a/src/observe/routes/ingest.ts b/src/observe/routes/ingest.ts index b5e5b7d28..9cda88923 100644 --- a/src/observe/routes/ingest.ts +++ b/src/observe/routes/ingest.ts @@ -3,13 +3,26 @@ * collector implements later. Validates the envelope (`400` names every bad * `batch[i]` path), enforces the ~5MB body cap (`413` via the bounded * `readBody`), then runs the whole batch - upserts plus rollup recompute - in - * one transaction and acks `202` fast. + * one transaction and acks `202` fast. The trace ids of roots the batch + * ended go to the evaluator worker, which never fails or delays the ack. */ import { OBSERVE_MAX_BODY_BYTES, parseIngestEnvelope } from '../envelope'; +import { rootsEndedIn, type EvaluatorWorker } from '../evaluatorWorker'; +import type { IngestEvent } from '../envelope'; import { ingestBatch } from '../store'; import { sendJson, parseJsonBody, type Route } from './httpKit'; +/** Hand the worker the batch's ended roots; a worker fault never reaches the response. */ +function notifyEvaluators(worker: EvaluatorWorker | undefined, batch: readonly IngestEvent[]): void { + if (worker === undefined) return; + try { + worker.notifyRootsEnded(rootsEndedIn(batch)); + } catch { + // online evaluation is best effort; ingest must stay infallible + } +} + export const ingestRoutes: readonly Route[] = [ { method: 'POST', @@ -21,6 +34,7 @@ export const ingestRoutes: readonly Route[] = [ return sendJson(ctx.res, 400, { error: `Invalid ingest envelope: ${parsed.issues.join('; ')}`, code: 'LOUSHO_REQUEST_INVALID' }); } const result = ingestBatch(ctx.db, parsed.envelope.batch); + notifyEvaluators(ctx.evaluators, parsed.envelope.batch); sendJson(ctx.res, 202, { received: result.received }); }, }, From 693e5a069a9a335d6e612fdae1eaf20bd7a94070 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 01:07:34 +0300 Subject: [PATCH 2/4] fix(observe): serialize evaluator dry runs, 404 a rootless trace - runOnce now returns undefined for a trace with no root observation (read once); POST /api/v1/evaluators/:id/test answers 404 for it. - Dry runs share one promise chain, so parallel /test calls never start parallel isolates. - The daemon ack test's fake sandbox waits 1000 ms, well past its 200 ms bound. - Studio's Scores screen lists the 'online' source. Co-Authored-By: Claude Opus 5.5 --- .../src/screens/scores/scoresData.ts | 2 +- docs/observability.md | 4 ++- llms-full.txt | 4 ++- src/observe/evaluatorWorker.test.ts | 23 ++++++++++++++- src/observe/evaluatorWorker.ts | 29 +++++++++++++------ src/observe/routes/evaluators.test.ts | 8 +++++ src/observe/routes/evaluators.ts | 10 +++---- 7 files changed, 62 insertions(+), 18 deletions(-) diff --git a/apps/agent-forge/src/screens/scores/scoresData.ts b/apps/agent-forge/src/screens/scores/scoresData.ts index 6b237fd74..680b30083 100644 --- a/apps/agent-forge/src/screens/scores/scoresData.ts +++ b/apps/agent-forge/src/screens/scores/scoresData.ts @@ -13,7 +13,7 @@ import type { TimeSeries } from '../../ui/TimeSeriesChart'; import { formatTimeRange, parseTimeRange, toWindow, type TimeRange } from '../../ui/TimeRangePicker'; /** The score sources the daemon accepts for `source=`. */ -export const SCORE_SOURCES = ['eval', 'judge', 'human'] as const; +export const SCORE_SOURCES = ['eval', 'judge', 'human', 'online'] as const; const DEFAULT_RANGE: TimeRange = { preset: '30d' }; const HOUR = 3_600_000; diff --git a/docs/observability.md b/docs/observability.md index 61aae04ae..4b043b3f5 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -660,7 +660,9 @@ the daemon restarts, so those traces never get online scores. `POST /api/v1/evaluators/:id/test` `{ "trace_id": "" }` dry-runs a rule on one trace and answers `200 { "result": { "value", "comment" } }`. It ignores `enabled`, `filter` and `sample_rate`, and it writes no score. An -unknown rule or trace is a `404`, and a bad body is a `400`. A `judge` rule is a +unknown rule is a `404`, and so is a trace that does not exist or has no root +observation. A bad body is a `400`. Dry runs go one at a time: a second +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" } }`. diff --git a/llms-full.txt b/llms-full.txt index 32a895f4c..f58759406 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -14554,7 +14554,9 @@ the daemon restarts, so those traces never get online scores. `POST /api/v1/evaluators/:id/test` `{ "trace_id": "" }` dry-runs a rule on one trace and answers `200 { "result": { "value", "comment" } }`. It ignores `enabled`, `filter` and `sample_rate`, and it writes no score. An -unknown rule or trace is a `404`, and a bad body is a `400`. A `judge` rule is a +unknown rule is a `404`, and so is a trace that does not exist or has no root +observation. A bad body is a `400`. Dry runs go one at a time: a second +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" } }`. diff --git a/src/observe/evaluatorWorker.test.ts b/src/observe/evaluatorWorker.test.ts index 10696f3eb..191744bb5 100644 --- a/src/observe/evaluatorWorker.test.ts +++ b/src/observe/evaluatorWorker.test.ts @@ -224,6 +224,27 @@ describe('evaluator worker', () => { expect(h.warnings).toHaveLength(hasIsolatedVm ? 0 : 1); }); + it('runOnce serializes dry runs and answers undefined for a trace with no root', async () => { + let active = 0; + let peak = 0; + const h = harness(async () => { + active++; + peak = Math.max(peak, active); + await new Promise((resolve) => setTimeout(resolve, 20)); + active--; + return 1; + }); + const rule = createEvaluatorRule(h.db, codeRule()); + if (rule === undefined) throw new Error('rule not created'); + ingestBatch(h.db, [rootEnd('t1'), child('orphan')]); + const results = await Promise.all([h.worker.runOnce(rule, 't1'), h.worker.runOnce(rule, 't1'), h.worker.runOnce(rule, 't1')]); + expect(results).toEqual([1, 1, 1].map((value) => ({ value, comment: null }))); + expect(peak).toBe(1); + await expect(h.worker.runOnce(rule, 'orphan')).resolves.toBeUndefined(); + await expect(h.worker.runOnce(rule, 'nope')).resolves.toBeUndefined(); + expect(onlineScores(h.db, 't1')).toEqual([]); + }); + it('drops and counts traces past maxQueue', async () => { const h = harness(() => 1, { maxQueue: 2 }); createEvaluatorRule(h.db, codeRule()); @@ -299,7 +320,7 @@ describe('ingest with the evaluator worker', () => { sandbox: () => ({ driver: 'fake', run: async () => { - await new Promise((resolve) => setTimeout(resolve, 200)); + await new Promise((resolve) => setTimeout(resolve, 1000)); throw new SandboxError('driver gone', 'driver', 'isolate'); }, }), diff --git a/src/observe/evaluatorWorker.ts b/src/observe/evaluatorWorker.ts index 76a1451ef..a4da74e78 100644 --- a/src/observe/evaluatorWorker.ts +++ b/src/observe/evaluatorWorker.ts @@ -63,8 +63,13 @@ export interface EvaluatorWorkerOptions { export interface EvaluatorWorker { /** Queue these traces (their root ended). Synchronous, never throws, never waits. */ notifyRootsEnded(traceIds: readonly string[]): void; - /** Run `rule` on `traceId` once, ignoring `enabled`, `filter` and `sample_rate`; writes nothing. */ - runOnce(rule: EvaluatorRuleRow, traceId: string): Promise; + /** + * Run `rule` on `traceId` once, ignoring `enabled`, `filter` and + * `sample_rate`; writes nothing. `undefined` when the trace has no root + * observation (or does not exist). Dry runs are serialized: a second call + * waits for the one in flight, so parallel calls never run parallel isolates. + */ + runOnce(rule: EvaluatorRuleRow, traceId: string): Promise; /** Resolves once the queue is empty and nothing is running. */ idle(): Promise; stats(): EvaluatorWorkerStats; @@ -190,6 +195,8 @@ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptio let stopped = false; let warned = false; let draining: Promise | undefined; + /** The tail of the dry-run lane: `runOnce` calls run one after another. */ + let dryRuns: Promise = Promise.resolve(); const runCode = async (rule: EvaluatorRuleRow, trace: EvaluatorTrace): Promise => { if (rule.kind !== 'code') return { value: null, comment: 'unsupported_kind' }; @@ -283,13 +290,17 @@ export function startEvaluatorWorker(db: SqlDatabase, opts: EvaluatorWorkerOptio } schedule(); }, - async runOnce(rule, traceId) { - const trace = evaluatorTrace(db, traceId); - if (trace === undefined) return { value: null, comment: 'trace_missing' }; - const runner = runnerFor(rule); - if (runner === undefined) return { value: null, comment: 'unsupported_kind' }; - if (!isolateAvailable()) return { value: null, comment: 'isolate_unavailable' }; - return runner(trace); + runOnce(rule, traceId) { + 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 (!isolateAvailable()) return { value: null, comment: 'isolate_unavailable' }; + return runner(trace); + }); + dryRuns = run.catch(() => undefined); + return run; }, async idle() { while (draining !== undefined) await draining; diff --git a/src/observe/routes/evaluators.test.ts b/src/observe/routes/evaluators.test.ts index 9fb8cb78e..d7879d5fa 100644 --- a/src/observe/routes/evaluators.test.ts +++ b/src/observe/routes/evaluators.test.ts @@ -276,6 +276,14 @@ describe('POST /api/v1/evaluators/:id/test', () => { const unknownTrace = await call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, { trace_id: 'nope' }, worker); expect(unknownTrace.status).toBe(404); expect(unknownTrace.body).toMatchObject(invalid); + ingestBatch(database, [ + { + type: 'observation', + data: { obs_id: 'kid', trace_id: 't-orphan', parent_obs_id: 'gone', kind: 'tool', start: 1, end: 2 }, + }, + ]); + const rootless = await call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, { trace_id: 't-orphan' }, worker); + expect(rootless.status).toBe(404); await expect(call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, {}, worker)).rejects.toMatchObject(invalid); await expect( call(database, 'POST', `/api/v1/evaluators/${code.rule_id}/test`, { trace_id: 't1', x: 1 }, worker) diff --git a/src/observe/routes/evaluators.ts b/src/observe/routes/evaluators.ts index 847299ea6..4f0fe3798 100644 --- a/src/observe/routes/evaluators.ts +++ b/src/observe/routes/evaluators.ts @@ -30,7 +30,6 @@ import { updateEvaluatorRule, type EvaluatorRuleFields, } from '../evaluators'; -import { traceDetail } from '../store'; import { intParam, requireJsonBody, sendJson, type Route } from './httpKit'; const RULE_BODY_LIMIT = 128 * 1024; @@ -167,11 +166,12 @@ export const evaluatorsRoutes: readonly Route[] = [ if (rule.kind === 'judge') { return sendJson(res, 409, { error: 'judge evaluators run through Agent Forge', code: 'LOUSHO_REQUEST_INVALID' }); } - if (traceDetail(db, trace_id) === undefined) { - return sendJson(res, 404, { error: `No trace '${trace_id}'.`, code: 'LOUSHO_REQUEST_INVALID' }); - } if (evaluators === undefined) throw new Error('the evaluator worker is not running'); - sendJson(res, 200, { result: await evaluators.runOnce(rule, trace_id) }); + const result = await evaluators.runOnce(rule, trace_id); + if (result === undefined) { + return sendJson(res, 404, { error: `No trace '${trace_id}' with a root observation.`, code: 'LOUSHO_REQUEST_INVALID' }); + } + sendJson(res, 200, { result }); }, }, { From 1c284f8331427d38609f36a8b0ca021afbe1fc8b Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 01:31:48 +0300 Subject: [PATCH 3/4] fix(observe): make the route context's evaluator worker required The daemon always hands routes its worker, so the optional field only created an unreachable `throw new Error(...)` in the /test route (flagged by plainErrors.test.ts). RouteContext.evaluators is now required; route tests pass a real worker over their db. Co-Authored-By: Claude Opus 5.5 --- src/observe/routes/datasets.test.ts | 2 ++ src/observe/routes/evaluators.test.ts | 2 +- src/observe/routes/evaluators.ts | 1 - src/observe/routes/httpKit.ts | 4 ++-- src/observe/routes/index.test.ts | 3 +++ src/observe/routes/ingest.ts | 3 +-- src/observe/routes/queues.test.ts | 2 ++ src/observe/routes/scores.test.ts | 3 +++ 8 files changed, 14 insertions(+), 6 deletions(-) diff --git a/src/observe/routes/datasets.test.ts b/src/observe/routes/datasets.test.ts index 3a1484962..2a3b20f9f 100644 --- a/src/observe/routes/datasets.test.ts +++ b/src/observe/routes/datasets.test.ts @@ -3,6 +3,7 @@ * from a trace's root observation), delete examples, and retention leaving * datasets alone. */ +import { startEvaluatorWorker } from '../evaluatorWorker'; import { describe, expect, it } from 'vitest'; import type * as http from 'node:http'; import type { IngestEvent } from '../envelope'; @@ -43,6 +44,7 @@ async function call( await matched.route.handle({ db: database, dbPath: ':memory:', + evaluators: startEvaluatorWorker(database), req: {} as http.IncomingMessage, res, params: matched.params, diff --git a/src/observe/routes/evaluators.test.ts b/src/observe/routes/evaluators.test.ts index d7879d5fa..0ff20ed01 100644 --- a/src/observe/routes/evaluators.test.ts +++ b/src/observe/routes/evaluators.test.ts @@ -47,7 +47,7 @@ async function call( params: matched.params, query: new URLSearchParams(search), readBody: () => Promise.resolve(text), - evaluators, + evaluators: evaluators ?? startEvaluatorWorker(database), }); return captured; } diff --git a/src/observe/routes/evaluators.ts b/src/observe/routes/evaluators.ts index 4f0fe3798..0ded5bdcb 100644 --- a/src/observe/routes/evaluators.ts +++ b/src/observe/routes/evaluators.ts @@ -166,7 +166,6 @@ export const evaluatorsRoutes: readonly Route[] = [ if (rule.kind === 'judge') { return sendJson(res, 409, { error: 'judge evaluators run through Agent Forge', code: 'LOUSHO_REQUEST_INVALID' }); } - if (evaluators === undefined) throw new Error('the evaluator worker is not running'); const result = await evaluators.runOnce(rule, trace_id); if (result === undefined) { return sendJson(res, 404, { error: `No trace '${trace_id}' with a root observation.`, code: 'LOUSHO_REQUEST_INVALID' }); diff --git a/src/observe/routes/httpKit.ts b/src/observe/routes/httpKit.ts index b874db17c..821b65e82 100644 --- a/src/observe/routes/httpKit.ts +++ b/src/observe/routes/httpKit.ts @@ -30,8 +30,8 @@ export interface RouteContext { * `LOUSHO_REQUEST_INVALID` on a second call. */ readBody(limitBytes: number): Promise; - /** The daemon's online-evaluator worker (absent in route tests that do not need it). */ - evaluators?: EvaluatorWorker; + /** The daemon's online-evaluator worker (ingest notifies it; the evaluator `/test` route dry-runs through it). */ + evaluators: EvaluatorWorker; } export interface Route { diff --git a/src/observe/routes/index.test.ts b/src/observe/routes/index.test.ts index c8de51ea7..645c93caf 100644 --- a/src/observe/routes/index.test.ts +++ b/src/observe/routes/index.test.ts @@ -3,6 +3,7 @@ * `:param` segments capture one decoded segment, and method, segment count * or a literal miss return `undefined`. */ +import { startEvaluatorWorker } from '../evaluatorWorker'; import { describe, expect, it } from 'vitest'; import { openDb } from '../store'; import { matchRoute, type RouteContext } from './index'; @@ -48,6 +49,7 @@ describe('GET /api/v1/traces query params', () => { void hit.route.handle({ db, dbPath: ':memory:', + evaluators: startEvaluatorWorker(db), req: {} as RouteContext['req'], res, params: hit.params, @@ -102,6 +104,7 @@ describe('sessions API', () => { void hit.route.handle({ db, dbPath: ':memory:', + evaluators: startEvaluatorWorker(db), req: {} as RouteContext['req'], res, params: hit.params, diff --git a/src/observe/routes/ingest.ts b/src/observe/routes/ingest.ts index 9cda88923..989836ffb 100644 --- a/src/observe/routes/ingest.ts +++ b/src/observe/routes/ingest.ts @@ -14,8 +14,7 @@ import { ingestBatch } from '../store'; import { sendJson, parseJsonBody, type Route } from './httpKit'; /** Hand the worker the batch's ended roots; a worker fault never reaches the response. */ -function notifyEvaluators(worker: EvaluatorWorker | undefined, batch: readonly IngestEvent[]): void { - if (worker === undefined) return; +function notifyEvaluators(worker: EvaluatorWorker, batch: readonly IngestEvent[]): void { try { worker.notifyRootsEnded(rootsEndedIn(batch)); } catch { diff --git a/src/observe/routes/queues.test.ts b/src/observe/routes/queues.test.ts index 5dd93242c..cb8cc195c 100644 --- a/src/observe/routes/queues.test.ts +++ b/src/observe/routes/queues.test.ts @@ -2,6 +2,7 @@ * Annotation queues API: create/list/get/delete queues, add trace ids, walk * them with `next`, and complete items into human score rows. */ +import { startEvaluatorWorker } from '../evaluatorWorker'; import { describe, expect, it } from 'vitest'; import type * as http from 'node:http'; import type { IngestEvent } from '../envelope'; @@ -42,6 +43,7 @@ async function call( await matched.route.handle({ db: database, dbPath: ':memory:', + evaluators: startEvaluatorWorker(database), req: {} as http.IncomingMessage, res, params: matched.params, diff --git a/src/observe/routes/scores.test.ts b/src/observe/routes/scores.test.ts index d388276dc..cc369ab58 100644 --- a/src/observe/routes/scores.test.ts +++ b/src/observe/routes/scores.test.ts @@ -3,6 +3,7 @@ * prefix, agent, source, time range) and the `aggregate=eval` pass-rate * rollup the Studio eval-history view reads. */ +import { startEvaluatorWorker } from '../evaluatorWorker'; import { describe, expect, it } from 'vitest'; import type * as http from 'node:http'; import type { IngestEvent } from '../envelope'; @@ -46,6 +47,7 @@ function get(database: SqlDatabase, query: string): Captured { matched.route.handle({ db: database, dbPath: ':memory:', + evaluators: startEvaluatorWorker(database), req: {} as http.IncomingMessage, res, params: matched.params, @@ -189,6 +191,7 @@ async function post(database: SqlDatabase, body: unknown): Promise { await matched.route.handle({ db: database, dbPath: ':memory:', + evaluators: startEvaluatorWorker(database), req: {} as http.IncomingMessage, res, params: matched.params, From 682da7f3da6673111b5c27dd1b92f6d69e08dda0 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 29 Oct 2026 01:43:53 +0300 Subject: [PATCH 4/4] ci: raise the pack-smoke entry budget to 940 CI counted 853 entries with the online evaluator modules; keep ~10% headroom. Co-Authored-By: Claude Opus 5.5 --- scripts/pack-smoke.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/scripts/pack-smoke.ts b/scripts/pack-smoke.ts index 65cd4d8a9..c10526f20 100644 --- a/scripts/pack-smoke.ts +++ b/scripts/pack-smoke.ts @@ -67,8 +67,10 @@ const REPO_ROOT = path.resolve(__dirname, '..'); * (2026-10-24); packed budget raised to 5 MiB (~10% headroom). * CI measured 16,796,541 bytes (~16.0 MiB) unpacked on 2026-10-29 (Studio * revamp dist plus docs); unpacked budget raised to 18 MiB (~10% headroom). + * CI counted 853 entries on 2026-10-29 (online evaluators); entry budget + * raised to 940 (~10% headroom). */ -const MAX_ENTRIES = 850; +const MAX_ENTRIES = 940; const MAX_UNPACKED_BYTES = 18 * 1024 * 1024; const MAX_PACKED_BYTES = 5 * 1024 * 1024;