From f06361e1823a9c8bfe08cd3f541b0a7c4dbaf5a5 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 18:55:36 +0300 Subject: [PATCH 1/3] fix(observe): flush eval spans so score rows link to a real trace The eval span went to a batched reporter that was never flushed before the vitest worker exited, so score rows pointed at a trace the daemon never saw. Add ObserveReporter.flush(), await it after the eval span settles, and make emit() return its promise so spans land before scores. Co-Authored-By: Claude Sonnet 5.5 --- src/evals/defineEval.ts | 2 +- src/evals/observe.test.ts | 17 +++++++++++++++++ src/evals/observe.ts | 19 +++++++++++++------ src/evals/trajectory.ts | 2 +- src/observe/reporter.ts | 9 +++++++++ 5 files changed, 41 insertions(+), 8 deletions(-) diff --git a/src/evals/defineEval.ts b/src/evals/defineEval.ts index 429cbc31..0ac8a879 100644 --- a/src/evals/defineEval.ts +++ b/src/evals/defineEval.ts @@ -308,7 +308,7 @@ function defineClassicEval(config: EvalConfig): void { usage: result.usage, file, }; - observation.emit(evalResult); + await observation.emit(evalResult); return evalResult; }); expect(evalResult.assertions[0].score).toBeGreaterThanOrEqual(threshold); diff --git a/src/evals/observe.test.ts b/src/evals/observe.test.ts index 944f7823..951c0feb 100644 --- a/src/evals/observe.test.ts +++ b/src/evals/observe.test.ts @@ -135,6 +135,23 @@ describe('evalCaseObservation', () => { } expect(scores[0].data.name).toBe('refund eval/completed()'); }); + + it('posts the eval span before the score rows, with the same trace_id', async () => { + const { fetchFn, calls } = noopFetch(); + vi.stubGlobal('fetch', fetchFn); + const observation = evalCaseObservation({ evalName: 'e', agent: { name: 'eval-flush-scored' }, env: { LOUSHO_OBSERVE: 'http://x' } }); + await observation.run(async () => {}); + await observation.emit(result()); + const spanBatch = calls.findIndex((c) => c.batch.some((e) => e.type === 'observation')); + const scoreBatch = calls.findIndex((c) => c.batch.some((e) => e.type === 'score')); + expect(spanBatch).toBeGreaterThanOrEqual(0); + expect(spanBatch).toBeLessThan(scoreBatch); + const span = calls[spanBatch].batch.find((e) => e.type === 'observation' && e.data.name === 'eval e'); + expect(span).toBeDefined(); + const scores = scoreEvents(calls); + expect(scores).toHaveLength(2); + for (const event of scores) expect(event.data.trace_id).toBe((span as Extract).data.obs_id); + }); }); describe('runTrajectoryCase emission (integration)', () => { diff --git a/src/evals/observe.ts b/src/evals/observe.ts index b6e34c3c..306d4589 100644 --- a/src/evals/observe.ts +++ b/src/evals/observe.ts @@ -79,8 +79,8 @@ export interface EvalCaseObservation { * putting it in the same trace); a plain `work(undefined)` call when off. */ run(work: (parentSpanId: string | undefined) => Promise): Promise; - /** POSTs the case's score events (fire-and-forget); a no-op when observation is off or the case recorded no assertions. */ - emit(result: EvalResult): void; + /** POSTs the case's score events; resolves (never rejects) once they are sent or dropped. A no-op when observation is off or the case recorded no assertions. */ + emit(result: EvalResult): Promise; } /** @@ -113,7 +113,9 @@ export function evalCaseObservation(options: { return { async run(work: (parentSpanId: string | undefined) => Promise): Promise { if (!reporter) return work(undefined); - return withSpan( + // The eval span ends into the reporter's batched outbox; a vitest worker + // can exit before the interval flush, leaving the scores' trace_id dangling. + const settled = withSpan( reporter, `eval ${options.evalName}`, { @@ -126,12 +128,17 @@ export function evalCaseObservation(options: { }, options.parentSpanId ); + try { + return await settled; + } finally { + await reporter.flush(); + } }, - emit(result: EvalResult): void { + async emit(result: EvalResult): Promise { if (!reporter || traceId === undefined || result.assertions.length === 0) return; const events = scoreEventsOf(result, { traceId, ...(scoredName !== undefined ? { agentId: reporter.agentId } : {}) }); - // Fire-and-forget: emitObserveEvents drops-and-counts, it never throws into the eval. - void emitObserveEvents(events, { env }); + // emitObserveEvents drops-and-counts, it never throws into the eval. + await emitObserveEvents(events, { env }); }, }; } diff --git a/src/evals/trajectory.ts b/src/evals/trajectory.ts index 7930853b..239e9ac6 100644 --- a/src/evals/trajectory.ts +++ b/src/evals/trajectory.ts @@ -388,7 +388,7 @@ export async function runTrajectoryCase( } const base = { name: spec.name, case: label, tags: spec.tags ?? [], file }; const result = context.toResult(base, Date.now() - started, error); - observation.emit(result); + await observation.emit(result); return result; } diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index e607ef79..95bec46c 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -101,6 +101,14 @@ export interface ObserveReporter extends TraceExporter { * on the shared reporter. */ promptUse?: PromptUseProvider; + /** + * Send everything pending now, batch after batch, without closing. Resolves + * once the outbox is drained or a batch failed (it is then requeued for the + * normal retry); never rejects. Callers that must see their spans land + * before the process may exit (eval workers) await this. + * @param timeoutMs Per-request timeout; default the reporter's own. + */ + flush(timeoutMs?: number): Promise; /** Deregister, flush once, and release timers/hooks. Idempotent. */ close(): Promise; } @@ -606,6 +614,7 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe // The root ended: forget every span of this trace so the map does not grow forever. for (const [id, tid] of targets) if (tid === traceId) targets.delete(id); }, + flush: (timeoutMs) => flush(timeoutMs ?? requestTimeoutMs), close, }; } From 4f1cf7219ecdc5cdb879bae50257e96938cc0e4d Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 19:00:38 +0300 Subject: [PATCH 2/3] chore(api): regenerate traces.api.md for ObserveReporter.flush Co-Authored-By: Claude Sonnet 5.5 --- api/traces.api.md | 1 + 1 file changed, 1 insertion(+) diff --git a/api/traces.api.md b/api/traces.api.md index 333f433a..850160d3 100644 --- a/api/traces.api.md +++ b/api/traces.api.md @@ -62,6 +62,7 @@ export interface ObserveReporter extends TraceExporter { readonly agentId: string; close(): Promise; readonly dropped: number; + flush(timeoutMs?: number): Promise; promptUse?: PromptUseProvider; readonly sessionId: string; } From 9d1d3c41867898ad2673d2b9d9e6240a4beba63e Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 19:04:05 +0300 Subject: [PATCH 3/3] fix(observe): bound eval flush waits so a down daemon never slows a case Cap the per-case span flush and score POST at 1s, and skip both for the rest of the run once the collector stops answering. Flush in a finally that also covers a synchronous throw from withSpan. Document the behaviour in docs/evals.md. Co-Authored-By: Claude Sonnet 5.5 --- docs/evals.md | 2 ++ llms-full.txt | 2 ++ src/evals/observe.test.ts | 25 ++++++++++++++++- src/evals/observe.ts | 56 ++++++++++++++++++++++++++------------- 4 files changed, 66 insertions(+), 19 deletions(-) diff --git a/docs/evals.md b/docs/evals.md index 7286eb55..49803fec 100644 --- a/docs/evals.md +++ b/docs/evals.md @@ -279,6 +279,8 @@ the `LOUSHO_EVAL_RESULTS` environment variable, which `lousho eval` sets and reads back. This is more robust than a custom vitest reporter: it works across vitest versions and worker pools, and needs no module loaded from your project. +With `LOUSHO_OBSERVE` set (or `lousho eval --observe`), each case's `eval` span and score rows are flushed to the collector when the case finishes, each wait capped at 1 second; a collector that is down or silent never fails or noticeably slows a case. + ### Time limits An agent run on a real model takes far longer than vitest's 5 s default test diff --git a/llms-full.txt b/llms-full.txt index 0e8269d9..f4b583bb 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -9984,6 +9984,8 @@ the `LOUSHO_EVAL_RESULTS` environment variable, which `lousho eval` sets and reads back. This is more robust than a custom vitest reporter: it works across vitest versions and worker pools, and needs no module loaded from your project. +With `LOUSHO_OBSERVE` set (or `lousho eval --observe`), each case's `eval` span and score rows are flushed to the collector when the case finishes, each wait capped at 1 second; a collector that is down or silent never fails or noticeably slows a case. + ### Time limits An agent run on a real model takes far longer than vitest's 5 s default test diff --git a/src/evals/observe.test.ts b/src/evals/observe.test.ts index 951c0feb..211ce016 100644 --- a/src/evals/observe.test.ts +++ b/src/evals/observe.test.ts @@ -125,7 +125,7 @@ describe('evalCaseObservation', () => { spanId = parentSpanId; }); expect(spanId).toEqual(expect.any(String)); - observation.emit(result()); + await observation.emit(result()); await vi.waitFor(() => expect(calls.length).toBeGreaterThan(0)); const scores = scoreEvents(calls); expect(scores).toHaveLength(2); @@ -152,6 +152,29 @@ describe('evalCaseObservation', () => { expect(scores).toHaveLength(2); for (const event of scores) expect(event.data.trace_id).toBe((span as Extract).data.obs_id); }); + + it('adds at most ~2.5s per case when the daemon accepts but never answers, then skips the waits', async () => { + vi.stubGlobal('fetch', vi.fn(() => new Promise(() => {}))); + const observation = () => evalCaseObservation({ evalName: 'e', agent: { name: 'eval-blackhole' }, env: { LOUSHO_OBSERVE: 'http://x' } }); + const first = observation(); + const started = Date.now(); + await expect(first.run(async () => 'ok')).resolves.toBe('ok'); + await first.emit(result()); + expect(Date.now() - started).toBeLessThan(2500); + const second = observation(); + const again = Date.now(); + await second.run(async () => 'ok'); + await second.emit(result()); + expect(Date.now() - again).toBeLessThan(200); + }); + + it('lets the case pass and resolves emit when every request is refused', async () => { + vi.stubGlobal('fetch', vi.fn(async () => Promise.reject(new Error('connect ECONNREFUSED')))); + vi.spyOn(console, 'warn').mockImplementation(() => undefined); + const observation = evalCaseObservation({ evalName: 'e', agent: { name: 'eval-refused' }, env: { LOUSHO_OBSERVE: 'http://x' } }); + await expect(observation.run(async () => 'ok')).resolves.toBe('ok'); + await expect(observation.emit(result())).resolves.toBeUndefined(); + }); }); describe('runTrajectoryCase emission (integration)', () => { diff --git a/src/evals/observe.ts b/src/evals/observe.ts index 306d4589..a19cb695 100644 --- a/src/evals/observe.ts +++ b/src/evals/observe.ts @@ -18,7 +18,7 @@ import { createAgentConfigOf, type SimpleAgent } from '../createAgent'; import { withSpan } from '../execution/tracing'; import { emitObserveEvents, observeTargetFromEnv } from '../observe/emit'; -import { observeExporterFromEnv } from '../observe/reporter'; +import { observeExporterFromEnv, type ObserveReporter } from '../observe/reporter'; import type { IngestEvent } from '../observe/envelope'; import { newId } from '../utils/id'; import type { EvalResult } from './evalResult'; @@ -71,6 +71,25 @@ export function evalObservationEnabled(env: NodeJS.ProcessEnv | undefined = type return observeTargetFromEnv(env) !== undefined; } +/** Cap on each per-case network wait (span flush, score POST): a down daemon must never slow or fail a case. */ +const EVAL_FLUSH_TIMEOUT_MS = 1000; + +/** Reporters whose daemon stopped answering in time; later cases skip the waits (the events still queue and retry on the reporter's own cadence). */ +const unresponsive = new WeakSet(); + +/** Resolves when `work` settles or after `ms`, whichever is first; `false` on the timeout. Never rejects. */ +async function within(work: Promise, ms: number): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(false), ms); + }); + try { + return await Promise.race([work.then(() => true, () => true), timeout]); + } finally { + clearTimeout(timer); + } +} + /** One eval case's observation: the wrapper span to parent exercised runs under, and the score emit. */ export interface EvalCaseObservation { /** @@ -115,30 +134,31 @@ export function evalCaseObservation(options: { if (!reporter) return work(undefined); // The eval span ends into the reporter's batched outbox; a vitest worker // can exit before the interval flush, leaving the scores' trace_id dangling. - const settled = withSpan( - reporter, - `eval ${options.evalName}`, - { - 'lousho.eval.name': options.evalName, - ...(options.caseLabel !== undefined ? { 'lousho.eval.case': options.caseLabel } : {}), - }, - (span) => { - traceId = span.id; - return work(span.id); - }, - options.parentSpanId - ); try { - return await settled; + return await withSpan( + reporter, + `eval ${options.evalName}`, + { + 'lousho.eval.name': options.evalName, + ...(options.caseLabel !== undefined ? { 'lousho.eval.case': options.caseLabel } : {}), + }, + (span) => { + traceId = span.id; + return work(span.id); + }, + options.parentSpanId + ); } finally { - await reporter.flush(); + if (!unresponsive.has(reporter) && !(await within(reporter.flush(EVAL_FLUSH_TIMEOUT_MS), EVAL_FLUSH_TIMEOUT_MS))) unresponsive.add(reporter); } }, async emit(result: EvalResult): Promise { if (!reporter || traceId === undefined || result.assertions.length === 0) return; const events = scoreEventsOf(result, { traceId, ...(scoredName !== undefined ? { agentId: reporter.agentId } : {}) }); - // emitObserveEvents drops-and-counts, it never throws into the eval. - await emitObserveEvents(events, { env }); + if (unresponsive.has(reporter)) return; + // emitObserveEvents drops-and-counts, it never throws into the eval; a daemon that does not answer in time is skipped for the rest of the run. + const sent = emitObserveEvents(events, { env, timeoutMs: EVAL_FLUSH_TIMEOUT_MS }); + if (!(await within(sent, EVAL_FLUSH_TIMEOUT_MS)) || (await sent) === 0) unresponsive.add(reporter); }, }; }