fix(observe): bound eval flush waits so a down daemon never slows a case

api/traces.api.md

@@ -62,6 +62,7 @@ export interface ObserveReporter extends TraceExporter {
     readonly agentId: string;
     close(): Promise<void>;
     readonly dropped: number;
+    flush(timeoutMs?: number): Promise<void>;
     promptUse?: PromptUseProvider;
     readonly sessionId: string;
 }

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

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

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);

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<IngestEvent, { type: 'observation' }>).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<Response>(() => {})));
+    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)', () => {

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<ObserveReporter>();
+
+/** Resolves when `work` settles or after `ms`, whichever is first; `false` on the timeout. Never rejects. */
+async function within(work: Promise<unknown>, ms: number): Promise<boolean> {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const timeout = new Promise<false>((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<T>(work: (parentSpanId: string | undefined) => Promise<T>): Promise<T>;
-  /** 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<void>;
 }
 
 /**
@@ -113,25 +132,33 @@ export function evalCaseObservation(options: {
   return {
     async run<T>(work: (parentSpanId: string | undefined) => Promise<T>): Promise<T> {
       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<void> {
       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);
     },
   };
 }

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;
 }
 

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<void>;
   /** Deregister, flush once, and release timers/hooks. Idempotent. */
   close(): Promise<void>;
 }
@@ -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,
   };
 }