feat(observe): run code evaluators on new traces

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;

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,54 @@ 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 `(<source>)(trace)` in the isolate sandbox
+(`sandbox({ driver: 'isolate', timeoutMs: 2000 })`). `trace` is a plain JSON
+object:
+
+```json
+{ "trace_id": "<trace-id>", "agent_id": "<agent-id>", "name": "<root name>",
+  "status": "ok", "start": 0, "end": 0, "duration_ms": 0,
+  "input": "<root input>", "output": "<root 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": <number | null>, "comment"?: "<text>" }` (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": "<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 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" } }`.
+
 ### Annotation queues
 
 An annotation queue is a named list of traces for a person to review one by

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,54 @@ 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 `(<source>)(trace)` in the isolate sandbox
+(`sandbox({ driver: 'isolate', timeoutMs: 2000 })`). `trace` is a plain JSON
+object:
+
+```json
+{ "trace_id": "<trace-id>", "agent_id": "<agent-id>", "name": "<root name>",
+  "status": "ok", "start": 0, "end": 0, "duration_ms": 0,
+  "input": "<root input>", "output": "<root 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": <number | null>, "comment"?: "<text>" }` (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": "<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 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" } }`.
+
 ### Annotation queues
 
 An annotation queue is a named list of traces for a person to review one by

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;
 

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<void>;
 }
 
@@ -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<void> => {
     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 });

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({