diff --git a/api/traces.api.md b/api/traces.api.md index 333f433a4..850160d36 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; } diff --git a/docs/evals.md b/docs/evals.md index 7286eb55d..49803fec2 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 0e8269d98..f4b583bb4 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/defineEval.ts b/src/evals/defineEval.ts index 429cbc31f..0ac8a879f 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 944f78237..211ce016d 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); @@ -135,6 +135,46 @@ 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); + }); + + 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 b6e34c3c2..a19cb6959 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 { /** @@ -79,8 +98,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,25 +132,33 @@ export function evalCaseObservation(options: { return { async run(work: (parentSpanId: string | undefined) => Promise): Promise { if (!reporter) return work(undefined); - return 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 - ); + // 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. + try { + 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 { + if (!unresponsive.has(reporter) && !(await within(reporter.flush(EVAL_FLUSH_TIMEOUT_MS), EVAL_FLUSH_TIMEOUT_MS))) unresponsive.add(reporter); + } }, - 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 }); + 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); }, }; } diff --git a/src/evals/trajectory.ts b/src/evals/trajectory.ts index 7930853be..239e9ac61 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 e607ef793..95bec46c3 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, }; }