From 17678de903962e271c15b5e143b7efa4e96a8aab Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:11:56 +0300 Subject: [PATCH 1/7] fix(observe): keep the loopback content default to the observe daemon With LOUSHO_OBSERVE on loopback, createAgent turned captureContent on for every exporter, so trace files, OTLP/vendor exporters and the built server's OTel export also received message and tool content. The default now strips the gen_ai content from every non-observe exporter part; an explicit captureContent: true or the OTel content env var still reaches all. Co-Authored-By: Claude Opus 5.5 --- docs/observability.md | 10 ++-- src/createAgent.ts | 19 ++++--- src/createAgentTracing.test.ts | 91 +++++++++++++++++++++++++++------- src/observe/emit.ts | 4 +- src/observe/reporter.ts | 38 ++++++++++++++ 5 files changed, 133 insertions(+), 29 deletions(-) diff --git a/docs/observability.md b/docs/observability.md index 1eb7722a..347fd012 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -240,9 +240,13 @@ The exception is a local collector: when `LOUSHO_OBSERVE` points at loopback (`1`/`true`, or a URL whose host is `127.0.0.1`, `localhost` or `::1`), `createAgent` defaults `captureContent` to `true`, so the daemon's trace view shows the prompt, the model output and the tool arguments and results that -explain a failure. The data stays in the daemon's db on your machine. Turn it -off with `captureContent: false`; a collector beyond loopback keeps the -opt-in default. The default applies to agents built with `createAgent`. +explain a failure. Only the daemon gets that content, and it stays in the +daemon's db on your machine: the agent's own `exporter` (`fileTraceExporter()` +files, an OTLP or vendor exporter, the built server's OTel export) still sees +every span without it, unless the agent sets `captureContent: true` or the +OTel content variable below is `true`. Turn it off with `captureContent: false`; +a collector beyond loopback keeps the opt-in default. The default applies to +agents built with `createAgent`. `redactContent` keeps only the legacy `input`/`prompt`/`args`/`result` attributes off; it does not scrub `gen_ai.*` message content. On a loopback diff --git a/src/createAgent.ts b/src/createAgent.ts index 909148a1..a2e76d3d 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -21,11 +21,11 @@ import { streamResumed, throwingRun, type AgentRun } from './execution/agentRun' import type { AgentEvent } from './execution/agentEvents'; import type { TraceExporter } from './execution/tracing'; import { defaultCaptureContent } from './observe/emit'; -import { withObserveExporter, type PromptUseProvider } from './observe/reporter'; +import { capturedContentForObserveOnly, withObserveExporter, type PromptUseProvider } from './observe/reporter'; import { parsePromptRef, registryPrompt } from './observe/promptClient'; import { assertModelSettings, mergeModelSettings } from './execution/modelSettings'; import { assertKnownCreateAgentKeys } from './createAgentKeys'; -import { DEFAULT_AGENT_NAME } from './execution/genAiSpans'; +import { DEFAULT_AGENT_NAME, resolveCaptureContent } from './execution/genAiSpans'; import { LLMProvider, LLMProviderRegistry } from './providers/llm'; import { ToolRegistry } from './tools/ToolRegistry'; import { ToolDescriptor } from './types'; @@ -411,7 +411,8 @@ export interface CreateAgentBase // Every run of this agent (send, stream, sessions, resume, approvals) is traced. // `LOUSHO_OBSERVE` fans the config's exporter out with the collector reporter; // unset means the env half is a no-op (see src/observe/reporter.ts). + // Content defaults on when the env collector is this machine's own (the trace + // view cannot explain a failure without it); an explicit setting always wins. + const captureContent = defaultCaptureContent(config.captureContent); + // That default is for the local daemon only: the config's own exporters + // (trace files, OTLP, a vendor) keep the content-free default unless the + // agent or the OTel content env var asked for content. + const contentForObserveOnly = config.captureContent === undefined && captureContent === true && !resolveCaptureContent(undefined); const exporter = withObserveExporter( - config.exporter, + contentForObserveOnly ? capturedContentForObserveOnly(config.exporter) : config.exporter, agentName, resolvedInstructions.promptUse !== undefined ? { promptUse: resolvedInstructions.promptUse } : {} ); - // Content defaults on when the env collector is this machine's own (the trace - // view cannot explain a failure without it); an explicit setting always wins. - const captureContent = defaultCaptureContent(config.captureContent); const tracing: Pick = { ...(exporter && { exporter }), ...(captureContent !== undefined && { captureContent }), diff --git a/src/createAgentTracing.test.ts b/src/createAgentTracing.test.ts index 179a3317..31263d3c 100644 --- a/src/createAgentTracing.test.ts +++ b/src/createAgentTracing.test.ts @@ -11,6 +11,7 @@ import { z } from 'zod'; import { createAgent } from './createAgent'; import { PropagatingToolError } from './execution/AgentExecutor'; import { createOtelTraceExporter } from './execution/otel'; +import { observeExporterFromEnv } from './observe/reporter'; import { fileTraceExporter } from './traces/fileTraceExporter'; import type { TraceLine } from './traces/format'; import { withSpan, type Span, type TraceExporter } from './execution/tracing'; @@ -337,53 +338,105 @@ describe('captureContent default on a local collector', () => { vi.unstubAllEnvs(); vi.unstubAllGlobals(); }); + /** The ingest bodies the env reporter POSTed; each test gets a fresh reporter through a unique agent name. */ + let posted: string[] = []; + let names = 0; const observe = (value: string) => { + posted = []; vi.stubEnv('LOUSHO_OBSERVE', value); - vi.stubGlobal('fetch', async () => new Response('{}', { status: 200 })); + vi.stubGlobal('fetch', async (_url: string, init: RequestInit) => { + posted.push(String(init.body)); + return new Response('{}', { status: 200 }); + }); + }; + const daemonText = async (name: string) => { + await observeExporterFromEnv({ agentName: name })?.flush(); + return posted.join(''); }; const chatInput = async (options: { captureContent?: boolean; redactContent?: boolean } = {}) => { const exporter = recording(); - await createAgent({ provider: mockModel(['Hi.']), exporter, ...options }).send('my key is sk-secret123'); - return exporter.ended.find((span) => op(span) === 'chat')!; + const name = `content-default-${++names}`; + await createAgent({ name, provider: mockModel(['Hi.']), exporter, ...options }).send('my key is sk-secret123'); + return { chat: exporter.ended.find((span) => op(span) === 'chat')!, daemon: await daemonText(name) }; }; - it('captures content by default when LOUSHO_OBSERVE points at loopback', async () => { + it('captures content for the loopback daemon only, never for the config exporter', async () => { + observe('1'); + const { chat, daemon } = await chatInput(); + expect(chat.attributes['gen_ai.input.messages']).toBeUndefined(); + expect(chat.attributes['gen_ai.output.messages']).toBeUndefined(); + expect(chat.attributes['gen_ai.request.model']).toBeDefined(); + expect(daemon).toContain('my key is'); + expect(daemon).toContain('Hi.'); + }); + + it('keeps content out of fileTraceExporter() files while the loopback daemon gets it', async () => { + observe('1'); + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lousho-traces-')); + try { + const name = `content-default-${++names}`; + await createAgent({ name, provider: mockModel(['Hi.']), exporter: fileTraceExporter({ dir }) }).send('my key is sk-secret123'); + const files = fs + .readdirSync(dir) + .flatMap((day) => fs.readdirSync(path.join(dir, day)).map((file) => fs.readFileSync(path.join(dir, day, file), 'utf8'))) + .join(''); + expect(files.length).toBeGreaterThan(0); + expect(files).not.toContain('sk-secret123'); + expect(await daemonText(name)).toContain('sk-secret123'); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + + it('captureContent: true still reaches every exporter', async () => { observe('1'); - const chat = await chatInput(); + const { chat, daemon } = await chatInput({ captureContent: true }); expect(chat.attributes['gen_ai.input.messages']).toContain('my key is'); - expect(chat.attributes['gen_ai.output.messages']).toContain('Hi.'); + expect(daemon).toContain('my key is'); + }); + + it('the OTel content env var still reaches every exporter', async () => { + observe('1'); + vi.stubEnv('OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT', 'true'); + expect((await chatInput()).chat.attributes['gen_ai.input.messages']).toContain('my key is'); }); it('does not capture by default for a non-loopback collector', async () => { observe('http://collector.test:7777'); - expect((await chatInput()).attributes['gen_ai.input.messages']).toBeUndefined(); + const { chat, daemon } = await chatInput(); + expect(chat.attributes['gen_ai.input.messages']).toBeUndefined(); + expect(daemon).not.toContain('my key is'); }); it('does not capture by default when LOUSHO_OBSERVE is unset', async () => { vi.stubEnv('LOUSHO_OBSERVE', ''); - expect((await chatInput()).attributes['gen_ai.input.messages']).toBeUndefined(); + const exporter = recording(); + await createAgent({ provider: mockModel(['Hi.']), exporter }).send('my key is sk-secret123'); + expect(exporter.ended.find((span) => op(span) === 'chat')!.attributes['gen_ai.input.messages']).toBeUndefined(); }); it('captureContent: false wins over the loopback default', async () => { observe('1'); - expect((await chatInput({ captureContent: false })).attributes['gen_ai.input.messages']).toBeUndefined(); + const { chat, daemon } = await chatInput({ captureContent: false }); + expect(chat.attributes['gen_ai.input.messages']).toBeUndefined(); + expect(daemon).not.toContain('my key is'); }); it('redactContent: false records the deprecated prompt next to the loopback-captured content', async () => { observe('1'); - const chat = await chatInput({ redactContent: false }); - expect(chat.attributes['gen_ai.input.messages']).toContain('my key is'); + const { chat, daemon } = await chatInput({ redactContent: false }); expect(chat.attributes.prompt).toBeDefined(); + expect(daemon).toContain('my key is'); }); - it('redactContent: true keeps the deprecated prompt off while gen_ai content is still captured', async () => { + it('redactContent: true keeps the deprecated prompt off while the daemon still gets gen_ai content', async () => { observe('1'); - const chat = await chatInput({ redactContent: true }); - expect(chat.attributes['gen_ai.input.messages']).toContain('my key is'); + const { chat, daemon } = await chatInput({ redactContent: true }); expect(chat.attributes.prompt).toBeUndefined(); + expect(daemon).toContain('my key is'); }); - it('resume() captures content by default on loopback', async () => { + it('resume() captures content for the loopback daemon by default', async () => { observe('1'); const store = memoryStore(); let crashed = false; @@ -399,10 +452,12 @@ describe('captureContent default on a local collector', () => { return 'ok'; }, }); - const crashing = createAgent({ provider: mockModel([{ toolCalls: [{ name: 'flaky', id: 'c1' }] }]), tools: [flaky], store }); + const name = `content-default-${++names}`; + const crashing = createAgent({ name, provider: mockModel([{ toolCalls: [{ name: 'flaky', id: 'c1' }] }]), tools: [flaky], store }); await expect(crashing.send('Go', { sessionId: 'job' })).rejects.toThrow('process died'); const exporter = recording(); - await createAgent({ provider: mockModel(['Done.']), tools: [flaky], store, exporter }).resume('job'); - expect(exporter.ended.find((span) => op(span) === 'chat')!.attributes['gen_ai.output.messages']).toContain('Done.'); + await createAgent({ name, provider: mockModel(['Done.']), tools: [flaky], store, exporter }).resume('job'); + expect(exporter.ended.find((span) => op(span) === 'chat')!.attributes['gen_ai.output.messages']).toBeUndefined(); + expect(await daemonText(name)).toContain('Done.'); }); }); diff --git a/src/observe/emit.ts b/src/observe/emit.ts index b45f5580..f6e84727 100644 --- a/src/observe/emit.ts +++ b/src/observe/emit.ts @@ -93,7 +93,9 @@ export function isLoopbackTarget(target: ObserveTarget): boolean { /** * The `captureContent` an agent runs with: the explicit setting when there is * one, else on when the env collector is loopback, else `undefined` (the - * executor then follows the OTel content env var). + * executor then follows the OTel content env var). The loopback default is + * meant for the daemon alone: createAgent strips that content from its other + * exporters (see `capturedContentForObserveOnly`). */ export function defaultCaptureContent(explicit: boolean | undefined, env?: NodeJS.ProcessEnv): boolean | undefined { if (explicit !== undefined) return explicit; diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index 0f17ccde..711648ce 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -781,6 +781,44 @@ function exporterParts(explicit: TraceExporter): readonly TraceExporter[] { return composedParts.get(explicit) ?? [explicit]; } +/** The opt-in `gen_ai.*` content attributes `captureContent` records. */ +const CAPTURED_CONTENT_KEYS = [ + GenAiAttr.SYSTEM_INSTRUCTIONS, + GenAiAttr.INPUT_MESSAGES, + GenAiAttr.OUTPUT_MESSAGES, + GenAiAttr.TOOL_CALL_ARGUMENTS, + GenAiAttr.TOOL_CALL_RESULT, +] as const; + +/** `span` without the captured content (a copy; `span` itself is untouched, and returned as-is when it holds none). */ +function withoutCapturedContent(span: Span): Span { + if (!CAPTURED_CONTENT_KEYS.some((key) => key in span.attributes)) return span; + const attributes = { ...span.attributes }; + for (const key of CAPTURED_CONTENT_KEYS) delete attributes[key]; + return { ...span, attributes }; +} + +/** + * `exporter` with the captured `gen_ai.*` content kept for its `LOUSHO_OBSERVE` + * reporters only: every other part (a `fileTraceExporter()`, an OTLP or vendor + * exporter) sees each span without it. createAgent applies this when content + * capture is on only because the env collector is loopback - that default is + * for the local daemon, not for every exporter the agent has. + */ +export function capturedContentForObserveOnly(exporter: TraceExporter | undefined): TraceExporter | undefined { + if (!exporter) return undefined; + return fanOutExporters( + exporterParts(exporter).map((part): TraceExporter => + envReporterSet.has(part) + ? part + : { + onSpanStart: (span) => part.onSpanStart(withoutCapturedContent(span)), + onSpanEnd: (span) => part.onSpanEnd(withoutCapturedContent(span)), + } + ) + ); +} + /** * The exporter a run reports with: the caller's, the env's, or both fanned * out. When `explicit` already carries an env reporter - a delegated From bb3da245107d69378c3a53a1687d3dcd2e81afb1 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:16:53 +0300 Subject: [PATCH 2/7] fix(evals): queue eval scores on the reporter outbox and let the slow-daemon latch expire One flush slower than 1s latched the memoized reporter as unresponsive for good, and every later case's scores were never sent; scores posted during an outage were dropped. Scores now queue on the reporter's outbox (after the span, retried with it, covered by the exit and crash flushes), each case still waits at most 1s, and a timed-out wait skips the waits for 30s only. An explicit reporter flush() no longer sits out the retry backoff. Co-Authored-By: Claude Opus 5.5 --- docs/evals.md | 2 +- src/evals/observe.test.ts | 73 ++++++++++++++++++++++++++++++++++++ src/evals/observe.ts | 53 +++++++++++++++++--------- src/observe/reporter.test.ts | 27 ++++++++++++- src/observe/reporter.ts | 41 +++++++++++++++----- 5 files changed, 167 insertions(+), 29 deletions(-) diff --git a/docs/evals.md b/docs/evals.md index 49803fec..aced862e 100644 --- a/docs/evals.md +++ b/docs/evals.md @@ -279,7 +279,7 @@ 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. +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. The score rows ride the same queue as the spans: they go out after the case's span and, while the collector is down, are kept and retried with it (and sent on exit or crash) instead of being dropped. When a wait runs out, the cases of the next 30 seconds skip the wait (their rows still queue), then the next case tries again. ### Time limits diff --git a/src/evals/observe.test.ts b/src/evals/observe.test.ts index 211ce016..cef0c6c6 100644 --- a/src/evals/observe.test.ts +++ b/src/evals/observe.test.ts @@ -168,6 +168,79 @@ describe('evalCaseObservation', () => { expect(Date.now() - again).toBeLessThan(200); }); + /** + * A daemon whose behaviour the test switches: `hang` accepts and never answers + * (until the request's abort, like real fetch, or a restart), `refuse` fails at once. + */ + const switchableDaemon = () => { + const hung: Array<() => void> = []; + const state = { + mode: 'ok' as 'ok' | 'hang' | 'refuse', + calls: [] as { url: string; batch: IngestEvent[] }[], + /** The daemon restarts healthy: the connections it held open are reset. */ + restart() { + state.mode = 'ok'; + for (const reset of hung.splice(0)) reset(); + }, + }; + vi.stubGlobal( + 'fetch', + vi.fn(async (input: string | URL | Request, init?: RequestInit) => { + if (state.mode === 'refuse') throw new TypeError('connect ECONNREFUSED'); + if (state.mode === 'hang') { + return new Promise((_, reject) => { + init?.signal?.addEventListener('abort', () => reject(init.signal!.reason)); + hung.push(() => reject(new TypeError('socket hang up'))); + }); + } + state.calls.push({ url: String(input), batch: (JSON.parse(String(init?.body)) as { batch: IngestEvent[] }).batch }); + return new Response('{}', { status: 202 }); + }) + ); + vi.spyOn(console, 'warn').mockImplementation(() => undefined); + return state; + }; + + it("sends later cases' scores once a slow flush is past: the skip-the-waits latch expires", async () => { + const daemon = switchableDaemon(); + let clock = 1_000_000; + const observation = () => + evalCaseObservation({ evalName: 'e', agent: { name: 'eval-slow-then-healthy' }, env: { LOUSHO_OBSERVE: 'http://x' }, now: () => clock }); + daemon.mode = 'hang'; + const first = observation(); + await first.run(async () => 'ok'); + await first.emit(result()); + daemon.restart(); + clock += 31_000; + const second = observation(); + await second.run(async () => 'ok'); + await second.emit(result()); + const scores = scoreEvents(daemon.calls); + expect(scores).toHaveLength(4); // the slow case's scores were queued, not lost + expect(new Set(scores.map((e) => e.data.trace_id)).size).toBe(2); + }); + + it('sends the scores of a case run during a daemon outage once the daemon is back', async () => { + const daemon = switchableDaemon(); + const observation = () => evalCaseObservation({ evalName: 'e', agent: { name: 'eval-outage' }, env: { LOUSHO_OBSERVE: 'http://x' } }); + daemon.mode = 'refuse'; + const first = observation(); + await first.run(async () => 'ok'); + await first.emit(result()); + expect(daemon.calls).toHaveLength(0); + daemon.mode = 'ok'; + const second = observation(); + await second.run(async () => 'ok'); + await second.emit(result()); + const events = daemon.calls.flatMap((c) => c.batch); + expect(scoreEvents(daemon.calls)).toHaveLength(4); + // Each case's eval span lands before its own scores. + const spanAt = (name: string) => events.findIndex((e) => e.type === 'observation' && e.data.name === name && e.data.end !== undefined); + const firstScoreAt = events.findIndex((e) => e.type === 'score'); + expect(spanAt('eval e')).toBeGreaterThanOrEqual(0); + expect(spanAt('eval e')).toBeLessThan(firstScoreAt); + }); + 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); diff --git a/src/evals/observe.ts b/src/evals/observe.ts index a19cb695..3125d74c 100644 --- a/src/evals/observe.ts +++ b/src/evals/observe.ts @@ -4,21 +4,22 @@ * When `LOUSHO_OBSERVE` is set (or `lousho eval --observe` set it inside the * vitest worker), each eval case runs inside an `eval ` span reported * to the same collector reporter the exercised agent uses, and the case's - * assertions POST as `score` ingest events. The span's id becomes the + * assertions are queued as `score` ingest events on that reporter's outbox - + * after the span, with its retry, buffer cap and exit and crash flushes. The span's id becomes the * `trace_id` of both the score rows and the exercised run's spans, so a * failing score reaches the run that produced it in one click - the eval * harness's own judgement context sits in the same trace as the root span. * * Emission lives here rather than in recorder.ts because it needs the span, * not just the result. Everything degrades silently: `LOUSHO_OBSERVE` unset - * means no span and no POSTs, and a dead daemon is counted as dropped, never - * thrown into the eval (see src/observe/emit.ts). + * means no span and no POSTs, and a dead daemon is retried or counted as + * dropped, never thrown into the eval (see src/observe/reporter.ts). */ import { createAgentConfigOf, type SimpleAgent } from '../createAgent'; import { withSpan } from '../execution/tracing'; -import { emitObserveEvents, observeTargetFromEnv } from '../observe/emit'; -import { observeExporterFromEnv, type ObserveReporter } from '../observe/reporter'; +import { observeTargetFromEnv } from '../observe/emit'; +import { observeExporterFromEnv, queueObserveEvents, type ObserveReporter } from '../observe/reporter'; import type { IngestEvent } from '../observe/envelope'; import { newId } from '../utils/id'; import type { EvalResult } from './evalResult'; @@ -71,11 +72,26 @@ 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. */ +/** Cap on each per-case network wait (span flush, score flush): 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(); +/** How long cases skip the waits after a daemon did not answer in time - then the next case tries again. */ +const UNRESPONSIVE_SKIP_MS = 30_000; + +/** + * Reporters whose daemon stopped answering in time, with when that skip ends. + * Meanwhile cases do not wait on a flush; their spans and scores still queue + * on the reporter and go out on its own retry cadence. + */ +const unresponsiveUntil = new WeakMap(); + +/** Flush `reporter` for at most EVAL_FLUSH_TIMEOUT_MS, unless it is in a skip window; a flush that runs out starts one. */ +async function boundedFlush(reporter: ObserveReporter, now: () => number): Promise { + if (now() < (unresponsiveUntil.get(reporter) ?? 0)) return; + if (!(await within(reporter.flush(EVAL_FLUSH_TIMEOUT_MS), EVAL_FLUSH_TIMEOUT_MS))) { + unresponsiveUntil.set(reporter, now() + UNRESPONSIVE_SKIP_MS); + } +} /** Resolves when `work` settles or after `ms`, whichever is first; `false` on the timeout. Never rejects. */ async function within(work: Promise, ms: number): Promise { @@ -98,7 +114,7 @@ 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; resolves (never rejects) once they are sent or dropped. A no-op when observation is off or the case recorded no assertions. */ + /** Queues the case's score events and flushes them (bounded); resolves, never rejects. A no-op when observation is off or the case recorded no assertions. */ emit(result: EvalResult): Promise; } @@ -106,8 +122,10 @@ export interface EvalCaseObservation { * The observation context of one eval case. When `LOUSHO_OBSERVE` resolves a * target, `run` wraps the case in an `eval ` span on the scored * agent's own env reporter (same memoized instance -> the run's spans nest - * into this trace); `emit` posts one `score` event per assertion with that - * span's id as `trace_id` and the scored agent's `agent_id`. + * into this trace); `emit` queues one `score` event per assertion on that + * reporter, with the span's id as `trace_id` and the scored agent's `agent_id`. + * Each wait for the daemon is capped at 1s; one that runs out skips the waits + * of the cases in the next 30s. */ export function evalCaseObservation(options: { /** The eval's name (the score rows' `/...` prefix). */ @@ -120,7 +138,10 @@ export function evalCaseObservation(options: { parentSpanId?: string; /** Env override (tests); default `process.env`. */ env?: NodeJS.ProcessEnv; + /** Clock for the unresponsive-daemon skip window (tests); default `Date.now`. */ + now?: () => number; }): EvalCaseObservation { + const now = options.now ?? Date.now; const env = options.env ?? (typeof process !== 'undefined' ? process.env : undefined); const scoredName = scoredAgentName(options.agent); // The scored agent's reporter is the memoized env reporter under the same @@ -149,16 +170,14 @@ export function evalCaseObservation(options: { options.parentSpanId ); } finally { - if (!unresponsive.has(reporter) && !(await within(reporter.flush(EVAL_FLUSH_TIMEOUT_MS), EVAL_FLUSH_TIMEOUT_MS))) unresponsive.add(reporter); + await boundedFlush(reporter, now); } }, async emit(result: EvalResult): Promise { if (!reporter || traceId === undefined || result.assertions.length === 0) return; - const events = scoreEventsOf(result, { traceId, ...(scoredName !== undefined ? { agentId: reporter.agentId } : {}) }); - 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); + // On the reporter's outbox the scores go out after the eval span and retry with it. + queueObserveEvents(reporter, scoreEventsOf(result, { traceId, ...(scoredName !== undefined ? { agentId: reporter.agentId } : {}) })); + await boundedFlush(reporter, now); }, }; } diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts index 2207918e..3e653f63 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -16,7 +16,7 @@ import { GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/sem import type { Span, TraceExporter } from '../execution/tracing'; import { blockingPost } from './blockingPost'; import type { IngestEvent } from './envelope'; -import { fanOutExporters, observeExporter, observeExporterFromEnv, withObserveExporter, type ObserveReporter } from './reporter'; +import { fanOutExporters, observeExporter, observeExporterFromEnv, queueObserveEvents, withObserveExporter, type ObserveReporter } from './reporter'; interface PostedBatch { url: string; @@ -319,6 +319,31 @@ describe('observeExporter', () => { expect(r.dropped).toBe(0); }); + it('flush() sends at once even while backing off, and a failure there is requeued', async () => { + const { fetchFn, posted } = scriptedFetch(['fail', 'fail', 'ok']); + const { r } = reporter({ fetchFn, flushIntervalMs: 1000 }); + emit(r, ['a']); + await vi.advanceTimersByTimeAsync(1000); // rejected; the interval waits until t=2000 + await r.flush(); // explicit: no backoff wait, fails again, requeued + expect(posted).toHaveLength(2); + await r.flush(); + expect(posted).toHaveLength(3); + expect(posted[2]!.events.map((e) => e.type === 'observation' && e.data.obs_id)).toEqual(['a']); + expect(r.dropped).toBe(0); + }); + + it('queueObserveEvents() puts non-span events behind the spans queued before them, retried with them', async () => { + const { fetchFn, posted } = scriptedFetch(['fail', 'ok']); + const { r } = reporter({ fetchFn }); + emit(r, ['span-1']); + const score: IngestEvent = { type: 'score', data: { score_id: 'sc', trace_id: 'span-1', name: 'e/a', value: 1, source: 'eval', ts: 1 } }; + expect(queueObserveEvents(r, [score])).toBe(true); + await r.flush(); + await r.flush(); + expect(posted[1]!.events.map((e) => e.type)).toEqual(['observation', 'score']); + expect(queueObserveEvents({} as ObserveReporter, [score])).toBe(false); + }); + it('close() makes one final attempt despite the backoff and counts leftovers as dropped', async () => { const { fetchFn, posted } = scriptedFetch(['fail']); const { r } = reporter({ fetchFn, flushIntervalMs: 1000 }); diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index 711648ce..d73e913d 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -111,10 +111,11 @@ export interface ObserveReporter extends TraceExporter { */ 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. + * Send everything pending now, batch after batch, without closing - even + * while backing off from a failed send. 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; @@ -483,21 +484,22 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe * Regular flushes give each POST its own timeout; the final attempt shares * one deadline. */ - const send = async (timeoutMs: number, final: boolean): Promise => { + const send = async (timeoutMs: number, final: boolean, force: boolean): Promise => { const deadline = Date.now() + timeoutMs; for (;;) { - const batch = outbox.take(final); + const batch = outbox.take(final || force); const cut = outbox.size > 0; // read before the await: events queued meanwhile wait for their own trigger if (batch.length === 0) return; if (!(await sendBatch(batch, final ? Math.max(1, deadline - Date.now()) : timeoutMs, final))) return; if (!cut) return; } }; - const flush = (timeoutMs: number, final = false, starting?: () => void): Promise => { + /** `force` skips the backoff (an explicit flush(); the final attempt always does). */ + const flush = (timeoutMs: number, final = false, starting?: () => void, force = false): Promise => { const run = chain .then(() => { starting?.(); - return send(timeoutMs, final); + return send(timeoutMs, final, force); }) .catch(() => undefined); chain = run; @@ -629,7 +631,7 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe } } - return { + const reporter: ObserveReporter = { get dropped() { return droppedCount; }, @@ -666,9 +668,28 @@ 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), + flush: (timeoutMs) => flush(timeoutMs ?? requestTimeoutMs, false, undefined, true), close, }; + queues.set(reporter, (events) => { + for (const event of events) enqueue(event); + }); + return reporter; +} + +/** Each reporter's own enqueue, for {@link queueObserveEvents}. */ +const queues = new WeakMap void>(); + +/** + * Queue ingest events that are not spans (eval scores) on `reporter`'s outbox: + * they go out after everything queued before them, and share its retry with + * backoff, buffer cap and exit and crash flushes. `false` when `reporter` was + * not made by {@link observeExporter}. + */ +export function queueObserveEvents(reporter: ObserveReporter, events: readonly IngestEvent[]): boolean { + const queue = queues.get(reporter); + queue?.(events); + return queue !== undefined; } /* ------------------------------------------------------- env wiring + fanout */ From fdefc4e2a49642a4718bb22a3413f93ceab5e08c Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:20:06 +0300 Subject: [PATCH 3/7] fix(observe): cap the reporter outbox by bytes and bound the crash flush The outbox was capped by event count only: with content capture on, 5000 generation events can hold around a gigabyte while the daemon is down, and the crash flush copied all of it into the worker before its 1s wait. The outbox now also evicts its oldest past 32MB, a crash sends only the newest 16MB (the rest counted as dropped), and sizes use Buffer.byteLength where there is one instead of an encoded copy per event. Co-Authored-By: Claude Opus 5.5 --- docs/observability.md | 9 +-- src/observe/__fixtures__/crashAfterRun.mts | 15 ++++- src/observe/reporter.test.ts | 35 ++++++++++- src/observe/reporter.ts | 69 +++++++++++++++++----- 4 files changed, 105 insertions(+), 23 deletions(-) diff --git a/docs/observability.md b/docs/observability.md index 347fd012..a6f769de 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -438,16 +438,17 @@ down, the failed batch is requeued and retried with exponential backoff (1 s, backlog drains in batches one right after another, each at most 1000 events and 4 MB of JSON (under the daemon's per-request caps of 1000 events and 5 MB). Only these are dropped and counted in the reporter's `dropped`: -events evicted past the buffer cap (`maxBuffered`, default 5000, oldest -first); events still unsent after the exit flush's final attempt; a single +events evicted past the buffer cap (`maxBuffered`, default 5000 events, or +32 MB of serialized events, whichever comes first; oldest first); events +past the newest 16 MB when a crash flushes; events still unsent after the exit flush's final attempt; a single event too large for any request (over 5 MB serialized, for example with content capture on); and a batch the daemon refuses outright (`400`/`413`), which is never retried. A process that dies of an uncaught exception or an unhandled rejection still reports its last run: the reporter watches `uncaughtExceptionMonitor` and -sends what it holds, oldest first, holding up the crash for up to 1 s per -reporter. Node's error output and exit code stay as they were, and a script +sends what it holds (the newest 16 MB of it), oldest first, holding up the +crash for up to 1 s per reporter. Node's error output and exit code stay as they were, and a script that survives the throw (its own `uncaughtException` handler, or a capture callback such as the REPL's) keeps reporting. An explicit `process.exit()` or a SIGKILL still loses what is buffered. diff --git a/src/observe/__fixtures__/crashAfterRun.mts b/src/observe/__fixtures__/crashAfterRun.mts index ae553b3a..8de5ff19 100644 --- a/src/observe/__fixtures__/crashAfterRun.mts +++ b/src/observe/__fixtures__/crashAfterRun.mts @@ -11,7 +11,11 @@ * - inflight: flushSize 1, so the span's batch is already in a POST the * collector holds open when a timer callback throws * - backlog: 1500 spans pending (past the daemon's 1000-event batch cap), then a top-level throw - * - bigbacklog: 30 spans of ~200KB input (~6MB, past the byte budget), then a top-level throw + * - bigbacklog: 30 spans of ~100KB input (~12MB with their start events, input riding twice: + * past the 4MB batch budget, under the crash flush's 16MB), then a top-level throw + * - hugebacklog: 200 spans of ~200KB input (~160MB with their start events, past the + * outbox and crash budgets), then a top-level throw; stamps `thrown-at=` + * and `exit-at=` (ms) on stderr * - handled: the script has its own 'uncaughtException' listener, so it * survives the throw, finishes a second span and closes normally * - capture: the same through process.setUncaughtExceptionCaptureCallback (REPL, domains) @@ -27,7 +31,7 @@ const reporter = observeExporter({ project: 'crash-fixture', agentName: 'crash-fixture', heartbeatIntervalMs: 0, - ...(mode === 'inflight' ? { flushSize: 1 } : mode === 'backlog' || mode === 'bigbacklog' ? { flushSize: 10_000 } : {}), + ...(mode === 'inflight' ? { flushSize: 1 } : mode === 'backlog' || mode === 'bigbacklog' || mode === 'hugebacklog' ? { flushSize: 10_000 } : {}), }); const finish = (id: string, attributes: Record = {}): void => { const span = { id, name: 'invoke_agent crash-fixture', attributes, startTime: Date.now() }; @@ -35,7 +39,12 @@ const finish = (id: string, attributes: Record = {}): void => { reporter.onSpanEnd({ ...span, endTime: Date.now() }); }; finish('crash-span'); -if (mode === 'bigbacklog') for (let i = 0; i < 30; i++) finish(`big-${i}`, { 'gen_ai.input.messages': 'x'.repeat(200 * 1024) }); +if (mode === 'bigbacklog') for (let i = 0; i < 30; i++) finish(`big-${i}`, { 'gen_ai.input.messages': 'x'.repeat(100 * 1024) }); +if (mode === 'hugebacklog') { + for (let i = 0; i < 200; i++) finish(`huge-${i}`, { 'gen_ai.input.messages': 'x'.repeat(200 * 1024) }); + process.on('exit', () => process.stderr.write(`exit-at=${Date.now()}\n`)); + process.stderr.write(`thrown-at=${Date.now()}\n`); +} if (mode === 'backlog') for (let i = 1; i < 1500; i++) finish(`crash-span-${i}`); if (mode === 'reject') { diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts index 3e653f63..b3277fb6 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -433,6 +433,22 @@ describe('observeExporter', () => { expect(r.dropped).toBe(0); }); + it('evicts the oldest events past the 32MB outbox byte budget and counts them', async () => { + vi.spyOn(console, 'warn').mockImplementation(() => undefined); + let down = true; + const { fetchFn, calls } = sizedFetch(() => (down ? 500 : 202)); + const { r } = reporter({ fetchFn, flushSize: 10_000 }); + // ~2MB per event (the input rides twice: input + attrs) x 24 = ~48MB, well under the 5000-event cap. + for (let i = 0; i < 24; i++) r.onSpanEnd(bigSpan(`big${i}`, 1024 * 1024)); + expect(r.dropped).toBeGreaterThanOrEqual(7); + expect(r.dropped).toBeLessThanOrEqual(9); + const kept = 24 - r.dropped; + down = false; + await r.flush(); + await r.flush(); + expect(calls.flatMap((c) => c.ids)).toEqual(Array.from({ length: kept }, (_, i) => `big${24 - kept + i}`)); + }); + it('drops and counts one event too large for any request, and keeps delivering', async () => { const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); const { fetchFn, calls } = sizedFetch(); @@ -669,7 +685,7 @@ describe('observeExporter', () => { describe('crash flush', () => { const fixture = path.join(__dirname, '__fixtures__', 'crashAfterRun.mts'); const repoRoot = path.resolve(__dirname, '..', '..'); - type Mode = 'throw' | 'reject' | 'inflight' | 'handled' | 'capture' | 'backlog' | 'bigbacklog'; + type Mode = 'throw' | 'reject' | 'inflight' | 'handled' | 'capture' | 'backlog' | 'bigbacklog' | 'hugebacklog'; /** * Run the fixture against an in-test collector; resolves with its exit and @@ -736,11 +752,26 @@ describe('crash flush', () => { it('cuts a crash backlog of large events at the byte budget', async () => { const { code, batches } = await crash('bigbacklog'); expect(code).toBe(1); - expect(batches.length).toBeGreaterThanOrEqual(2); // 30 x 200KB = ~6MB + expect(batches.length).toBeGreaterThanOrEqual(3); // 31 runs x (start + end) x 2 copies of ~100KB = ~12MB for (const batch of batches) expect(Buffer.byteLength(JSON.stringify({ batch }))).toBeLessThanOrEqual(4 * 1024 * 1024); expect(batches.flatMap(obsIds)).toEqual(['crash-span', ...Array.from({ length: 30 }, (_, i) => `big-${i}`)]); }, 30_000); + it('sends only the newest 16MB of a crash backlog and stays within its time bound', async () => { + const { code, stderr, batches } = await crash('hugebacklog'); + expect(code).toBe(1); + // The fixture stamps the moment it throws and the reporter's crash flush has its 1s cap (worker start-up included). + const thrownAt = Number(/thrown-at=(\d+)/.exec(stderr)?.[1]); + const exitedAt = Number(/exit-at=(\d+)/.exec(stderr)?.[1]); + expect(exitedAt - thrownAt).toBeLessThan(1500); + const bytes = batches.reduce((sum, batch) => sum + Buffer.byteLength(JSON.stringify({ batch })), 0); + expect(bytes).toBeLessThanOrEqual(16 * 1024 * 1024 + 64 * 1024); + const ids = batches.flatMap(obsIds); + expect(ids.length).toBeGreaterThan(0); + expect(ids.at(-1)).toBe('huge-199'); // newest first to be kept + expect(batches.flat().map((e) => e.type)).toContain('agent.deregister'); + }, 30_000); + it('resends the batch a POST still in flight was carrying when the process died', async () => { const { code, batches } = await crash('inflight', true); expect(code).toBe(1); diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index d73e913d..b6c63cee 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -29,7 +29,8 @@ * SIGKILL loses the run's unsent child spans; its root start was sent eagerly. * - The daemon being down requeues the failed batch and retries with * exponential backoff (1s base, 30s cap); only events evicted past the - * buffer cap, or still unsent after close()'s final attempt, drop into a + * buffer cap (5000 events or 32MB, whichever comes first), events past the + * crash flush's 16MB, or still unsent after close()'s final attempt, drop into a * `dropped` counter. It warns ONCE and never blocks or throws into a run. * Timers are unref'd so observation never keeps a process alive. * - Its own HTTP calls are plain `fetch`, outside any span: instrumenting @@ -73,7 +74,7 @@ export interface ObserveExporterOptions { flushIntervalMs?: number; /** Heartbeat cadence (default 60000ms); `0` disables heartbeats. */ heartbeatIntervalMs?: number; - /** Buffer cap; past it the oldest events are evicted into `dropped` (default 5000, floor 1). */ + /** Buffer cap in events; past it - or past the 32MB byte budget - the oldest events are evicted into `dropped` (default 5000, floor 1). */ maxBuffered?: number; /** Per-request timeout (default 5000ms); close() aborts such a request and caps its final attempt at ~2000ms. */ requestTimeoutMs?: number; @@ -131,6 +132,18 @@ const DEFAULT_REQUEST_TIMEOUT_MS = 5000; const EXIT_FLUSH_TIMEOUT_MS = 2000; /** How long a crash may block the dying process to deliver what is pending. */ const CRASH_FLUSH_TIMEOUT_MS = 1000; +/** + * Outbox byte budget: with content capture on, each generation event carries + * the conversation (twice: input and attrs), so an event-count cap alone could + * hold gigabytes while the daemon is down. Past it the oldest are evicted. + */ +const MAX_BUFFERED_BYTES = 32 * 1024 * 1024; +/** + * What a crash flush sends at most, newest first kept: the bodies are copied + * to the worker synchronously before the 1s wait starts, so the copy is + * bounded too, and more would not get through in 1s anyway. + */ +const CRASH_FLUSH_MAX_BYTES = 16 * 1024 * 1024; const RETRY_BASE_MS = 1000; const INSTANCE_KEY = Symbol.for('lousho.observeInstance'); @@ -173,10 +186,16 @@ interface OutboxEntry { const utf8 = new TextEncoder(); +/** UTF-8 length of `text`: Buffer.byteLength where there is a Buffer (no copy), else an encoded copy's length. */ +const byteLengthOf: (text: string) => number = + typeof Buffer !== 'undefined' && typeof Buffer.byteLength === 'function' + ? (text) => Buffer.byteLength(text, 'utf8') + : (text) => utf8.encode(text).length; + /** An event as an outbox entry; `undefined` when its body alone would pass the daemon's hard cap. */ function entryOf(event: IngestEvent): OutboxEntry | undefined { const json = JSON.stringify(event); - const bytes = utf8.encode(json).length; + const bytes = byteLengthOf(json); return ENVELOPE_BYTES + bytes > OBSERVE_MAX_BODY_BYTES ? undefined : { json, bytes }; } @@ -208,21 +227,34 @@ function batchesOf(entries: readonly OutboxEntry[]): OutboxEntry[][] { return batches; } +/** The newest of `entries` that fit in `budget` bytes, and how many older ones did not. */ +function newestWithin(entries: readonly OutboxEntry[], budget: number): { kept: OutboxEntry[]; cut: number } { + let start = entries.length; + for (let total = 0; start > 0 && total + entries[start - 1].bytes <= budget; start--) total += entries[start - 1].bytes; + return { kept: entries.slice(start), cut: start }; +} + /** - * The reporter's pending events: FIFO, capped at `cap` (the oldest are - * evicted first, reported through `onEvict`), and gated by a retry backoff + * The reporter's pending events: FIFO, capped at `cap` events and `byteCap` + * bytes (the oldest are evicted first, the newest always stays, reported + * through `onEvict`), and gated by a retry backoff * after a failed send. Every send path - size, interval, exit - goes through * `take` / `delivered` / `failed`, so a failed batch is never lost while * there is room to keep it. A batch is cut by {@link batchEnd}, oldest first. */ -function createOutbox(cap: number, now: () => number, onEvict: (count: number) => void) { +function createOutbox(cap: number, byteCap: number, now: () => number, onEvict: (count: number) => void) { let items: OutboxEntry[] = []; + let bytes = 0; let failures = 0; let retryAt = 0; + const sizeOf = (entries: readonly OutboxEntry[]): number => entries.reduce((sum, e) => sum + e.bytes, 0); const trim = (): void => { - const over = items.length - cap; - if (over <= 0) return; + let over = Math.max(0, items.length - cap); + let left = bytes - sizeOf(items.slice(0, over)); + for (; left > byteCap && over < items.length - 1; over++) left -= items[over].bytes; + if (over === 0) return; items.splice(0, over); + bytes = left; onEvict(over); }; return { @@ -238,15 +270,20 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) = const entry = entryOf(event); if (!entry) return false; items.push(entry); + bytes += entry.bytes; trim(); return true; }, /** The oldest batch, for one send - or nothing while backing off, unless `force` (the final attempt). */ take(force = false): OutboxEntry[] { - return force || now() >= retryAt ? items.splice(0, batchEnd(items, 0)) : []; + if (!force && now() < retryAt) return []; + const batch = items.splice(0, batchEnd(items, 0)); + bytes -= sizeOf(batch); + return batch; }, /** Everything pending, uncut (the crash flush; counting what a failed final attempt leaves). */ drain(): OutboxEntry[] { + bytes = 0; return items.splice(0, items.length); }, delivered(): void { @@ -256,6 +293,7 @@ function createOutbox(cap: number, now: () => number, onEvict: (count: number) = /** Put a failed batch back in front of anything queued since, trim to the cap, and back off. */ failed(batch: OutboxEntry[]): void { items = batch.concat(items); // no spread: keeps call arguments bounded whatever the batch size + bytes += sizeOf(batch); trim(); retryAt = now() + retryDelayMs(failures++); }, @@ -403,10 +441,10 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe ); }; - const outbox = createOutbox(maxBuffered, now, (count) => { + const outbox = createOutbox(maxBuffered, MAX_BUFFERED_BYTES, now, (count) => { droppedCount += count; warnOnce( - `buffer is full (${maxBuffered} events); dropping the oldest`, + `buffer is full (${maxBuffered} events or ${MAX_BUFFERED_BYTES / (1024 * 1024)}MB); dropping the oldest`, new SDKError('observe buffer full', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }) ); }); @@ -583,13 +621,16 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe /** * The process is dying of an uncaught exception or unhandled rejection: * Node exits as soon as the monitor returns, so no async send can finish. - * Send everything pending, oldest first - the batch a regular POST was + * Send what is pending, oldest first - the batch a regular POST was * carrying included (re-sent events upsert by key) - as successive POSTs, - * cut like any other batch, that together block for at most 1s. + * cut like any other batch, that together block for at most 1s. Only the + * newest CRASH_FLUSH_MAX_BYTES go; older events are counted as dropped. */ const crashFlush = (): void => { if (!shutdown()) return; - const batches = batchesOf(inFlight.concat(outbox.drain())); + const { kept, cut } = newestWithin(inFlight.concat(outbox.drain()), CRASH_FLUSH_MAX_BYTES); + droppedCount += cut; + const batches = batchesOf(kept); const delivered = blockingPost(`${url}/api/v1/ingest`, headers, batches.map(bodyOf), CRASH_FLUSH_TIMEOUT_MS); batches.forEach((batch, i) => { if (!delivered[i]) droppedCount += batch.length; From dc03038e5e7ee3dc3f9b8244e21332ef86579766 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:21:04 +0300 Subject: [PATCH 4/7] docs(changelog): record the observability fix round under Unreleased Adds entries for span status ok, requeue/backoff with batch and byte caps and the crash flush, eval span and score flushing, strict LOUSHO_OBSERVE parsing, the loopback content default (daemon only) and redactContent's scope, per-db pidfiles, port-aware hints and /observe/__target, and the Studio shell; fixes the prompt registry entry: no instructions fails closed, with a 30s retry window. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 33e403ff..dbabf700 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,10 +15,16 @@ opt-out. - The `lousho observe` daemon: a local collector (loopback, `http://127.0.0.1:7777`, SQLite at `~/.lousho/observe.db`) that gathers every agent's traces, heartbeats and eval scores when `LOUSHO_OBSERVE` is set — no per-agent code changes. `lousho observe start|status|stop|rebuild-rollups|token create`; a second `start` on a busy port attaches when it is already the observe daemon. See [Observability](docs/observability.md#the-observe-daemon). - Agent Forge gains Observe tabs — Traces, Insights, Monitoring, Scores — reading the daemon through a token-gated `/observe/**` proxy on the studio server. -- Prompt registry: `createAgent({ prompt: 'name@version' })` or `LOUSHO_PROMPT` resolves versioned instructions from the daemon at build (embedded `instructions` are the fallback); every run records the version used. `lousho prompt push|list|get` manages it. +- Prompt registry: `createAgent({ prompt: 'name@version' })` or `LOUSHO_PROMPT` resolves versioned instructions from the daemon at build; every run records the version used. `lousho prompt push|list|get` manages it. When the daemon is unreachable, embedded `instructions` are used; with no `instructions` the agent fails closed: every run rejects with `LOUSHO_PROMPT_UNRESOLVED` instead of running without a system prompt. A failed fetch is not cached for good: runs in the next 30 seconds resolve at once, then the next run fetches again (waiting up to 2.5 seconds). - Context hub: `hub()` / `hubTools()` publish and query namespaced shared memory scoped `private` (the writing agent) or `fleet` (every agent on the daemon). - Fleet monitoring: heartbeat-derived agent status, `agent.deregister` on graceful exit, a crashed agent's open sessions and unfinished observations flip to `interrupted`, and webhook alerts (`--webhook`, `alerts.json`) on down transitions and error-rate thresholds. -- Eval scores report to the daemon: every `AssertionResult` lands as a score row on the exercised run's trace (`source: eval` or `judge`); `lousho eval --observe` forces emission. +- Eval scores report to the daemon: every `AssertionResult` lands as a score row on the exercised run's trace (`source: eval` or `judge`); `lousho eval --observe` forces emission. Each case's `eval` span and score rows are flushed when the case finishes, each wait capped at 1 second; the scores share the reporter's queue, so they go out after the span and are retried while the daemon is down. A wait that runs out skips the waits of the next 30 seconds of cases. +- Reporting to the daemon: spans that finished without an error report status `ok`. A down daemon's batches are requeued and retried with exponential backoff (1 s up to 30 s); batches are capped at 1000 events and 4 MB, and a batch the daemon refuses (`400`/`413`) is dropped and counted in `dropped`. The queue holds at most 5000 events or 32 MB, oldest evicted first. A process dying of an uncaught exception or unhandled rejection sends the newest 16 MB it holds, blocking for at most 1 second; a graceful exit makes one final attempt capped at about 2 seconds. A run's root span is reported when the run starts, so a run whose process is killed is still listed and turns `interrupted`. +- `LOUSHO_OBSERVE` accepts `1`/`true`/`on`/`yes`, an `http(s)://` URL, or `0`/`false`/`off`/`no`/`none`/`disabled`/empty (any case). Any other value warns once naming `LOUSHO_OBSERVE` and disables reporting, instead of being posted to as a URL. +- With `LOUSHO_OBSERVE` pointing at a loopback daemon, `createAgent` captures message and tool content for the daemon by default (`captureContent: false` turns it off). Only the daemon gets it: the agent's own `exporter` (trace files, OTLP or vendor exporters, the built server's OTel export) still records no content unless the agent sets `captureContent: true`. `redactContent` does not scrub this `gen_ai.*` content; it only governs the legacy `input`/`prompt`/`args`/`result` attributes. See [Observability](docs/observability.md#message-and-argument-content-is-opt-in). +- Each observe daemon writes its own pidfile beside its db (`foo.db` gives `foo.pid`), so two dbs in one folder never share one. `lousho observe start` exits 1 when the port already serves another db, and `stop` refuses to signal when the daemon on the pidfile's port reports another db or answers as a different service. A daemon started with `--db foo.db` by an earlier build of this feature wrote the shared pidfile and is not found by the new `stop`: stop it before updating. +- `lousho observe start` and Studio print the `LOUSHO_OBSERVE` value for the port and host the daemon really uses (`LOUSHO_OBSERVE=http://127.0.0.1:7788` on a non-default port), and the Studio proxy answers `/observe/__target` with the daemon URL reporters should use. +- Agent Forge navigates with hash routes (`#/traces/`: a trace has a link of its own, and back and forward work), has a left navigation instead of the Build/Observe toggle, and shows the Build controls only on Agents. - `sandbox({ driver: 'isolate' })` runs tool code in an `isolated-vm` V8 isolate (optional peer `isolated-vm`); `defineTool({ sandbox })` opts a tool in, timeouts and OOM map to `sandbox.error` on the tool span. `'docker'` and `'e2b'` driver names are reserved for later drivers. ## [1.0.0-alpha.29] - 2026-10-21 From cf26e6fdb110b32352e56852a1ba3e3600df5c59 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:26:21 +0300 Subject: [PATCH 5/7] fix(observe): final-review minors (IPv6 hints, drop warnings, router #, /health db, heartbeat slack, docs) - M1: bracket IPv6 hosts in the reporter hint, start/attach logs and the health probe. - M2: drops (buffer full, oversized, 400/413) warn once on their own flag without promising a retry, so a drop warning no longer hides a later outage warning. - M3: Studio navigate('#/x') no longer yields '##/x'. - M4: /health names the db path only to callers on this machine. - M9: a same-host+pid instance counts as replaced only after one interval plus the 10s freshness slack, so live twins started together are not interrupted. - M5-M8: errors.md (every run rejects while unresolved), cli.md (stop refuses a port answering as another service), observability.md (SSH tunnels count as loopback; the retry fetch can hold one run up to 2.5s). Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 2 +- .../src/router/__tests__/router.test.ts | 8 +++++ apps/agent-forge/src/router/router.ts | 5 ++-- docs/cli.md | 5 ++-- docs/errors.md | 7 +++-- docs/observability.md | 13 +++++--- src/cli/observe.test.ts | 5 ++++ src/cli/observe.ts | 7 +++-- src/observe/daemon.test.ts | 30 +++++++++++++++++++ src/observe/daemon.ts | 7 ++++- src/observe/reporter.test.ts | 16 ++++++++++ src/observe/reporter.ts | 26 +++++++++------- src/observe/routes/status.ts | 26 +++++++++++++--- src/observe/store.test.ts | 25 ++++++++++++++-- src/observe/store.ts | 10 ++++--- 15 files changed, 156 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dbabf700..b0dc7b4d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,7 +22,7 @@ opt-out. - Reporting to the daemon: spans that finished without an error report status `ok`. A down daemon's batches are requeued and retried with exponential backoff (1 s up to 30 s); batches are capped at 1000 events and 4 MB, and a batch the daemon refuses (`400`/`413`) is dropped and counted in `dropped`. The queue holds at most 5000 events or 32 MB, oldest evicted first. A process dying of an uncaught exception or unhandled rejection sends the newest 16 MB it holds, blocking for at most 1 second; a graceful exit makes one final attempt capped at about 2 seconds. A run's root span is reported when the run starts, so a run whose process is killed is still listed and turns `interrupted`. - `LOUSHO_OBSERVE` accepts `1`/`true`/`on`/`yes`, an `http(s)://` URL, or `0`/`false`/`off`/`no`/`none`/`disabled`/empty (any case). Any other value warns once naming `LOUSHO_OBSERVE` and disables reporting, instead of being posted to as a URL. - With `LOUSHO_OBSERVE` pointing at a loopback daemon, `createAgent` captures message and tool content for the daemon by default (`captureContent: false` turns it off). Only the daemon gets it: the agent's own `exporter` (trace files, OTLP or vendor exporters, the built server's OTel export) still records no content unless the agent sets `captureContent: true`. `redactContent` does not scrub this `gen_ai.*` content; it only governs the legacy `input`/`prompt`/`args`/`result` attributes. See [Observability](docs/observability.md#message-and-argument-content-is-opt-in). -- Each observe daemon writes its own pidfile beside its db (`foo.db` gives `foo.pid`), so two dbs in one folder never share one. `lousho observe start` exits 1 when the port already serves another db, and `stop` refuses to signal when the daemon on the pidfile's port reports another db or answers as a different service. A daemon started with `--db foo.db` by an earlier build of this feature wrote the shared pidfile and is not found by the new `stop`: stop it before updating. +- Each observe daemon writes its own pidfile beside its db (`foo.db` gives `foo.pid`), so two dbs in one folder never share one. `lousho observe start` exits 1 when the port already serves another db, and `stop` refuses to signal when the daemon on the pidfile's port reports another db or answers as a different service. `/health` names the db it serves only to callers on the same machine. A daemon started with `--db foo.db` by an earlier build of this feature wrote the shared pidfile and is not found by the new `stop`: stop it before updating. - `lousho observe start` and Studio print the `LOUSHO_OBSERVE` value for the port and host the daemon really uses (`LOUSHO_OBSERVE=http://127.0.0.1:7788` on a non-default port), and the Studio proxy answers `/observe/__target` with the daemon URL reporters should use. - Agent Forge navigates with hash routes (`#/traces/`: a trace has a link of its own, and back and forward work), has a left navigation instead of the Build/Observe toggle, and shows the Build controls only on Agents. - `sandbox({ driver: 'isolate' })` runs tool code in an `isolated-vm` V8 isolate (optional peer `isolated-vm`); `defineTool({ sandbox })` opts a tool in, timeouts and OOM map to `sandbox.error` on the tool span. `'docker'` and `'e2b'` driver names are reserved for later drivers. diff --git a/apps/agent-forge/src/router/__tests__/router.test.ts b/apps/agent-forge/src/router/__tests__/router.test.ts index 95e0fb90..58dd7f9d 100644 --- a/apps/agent-forge/src/router/__tests__/router.test.ts +++ b/apps/agent-forge/src/router/__tests__/router.test.ts @@ -61,6 +61,14 @@ describe('navigate', () => { expect(window.location.hash).toBe('#/traces?status=error'); expect(window.location.search).toBe('?token=secret'); }); + it('takes a target that already starts with # (no ##)', async () => { + navigate('#/traces/t1'); + await tick(); + expect(window.location.hash).toBe('#/traces/t1'); + navigate('#agents'); + await tick(); + expect(window.location.hash).toBe('#/agents'); + }); it('replace does not add a history entry', async () => { const before = window.history.length; navigate('/a'); diff --git a/apps/agent-forge/src/router/router.ts b/apps/agent-forge/src/router/router.ts index da9b9277..0d1ad406 100644 --- a/apps/agent-forge/src/router/router.ts +++ b/apps/agent-forge/src/router/router.ts @@ -46,9 +46,10 @@ export function matchRoute(hash: string, patterns: string[]): Route | null { return null; } -/** Go to `to` (`/traces?status=error`). `replace` swaps the current history entry. Fires `hashchange` once. */ +/** Go to `to` (`/traces?status=error`, a leading `#` allowed). `replace` swaps the current history entry. Fires `hashchange` once. */ export function navigate(to: string, opts?: { replace?: boolean }): void { - const hash = `#${to.startsWith('/') ? to : `/${to}`}`; + const path = to.startsWith('#') ? to.slice(1) : to; + const hash = `#${path.startsWith('/') ? path : `/${path}`}`; if (window.location.hash === hash) return; if (opts?.replace) window.location.replace(hash); else window.location.hash = hash; diff --git a/docs/cli.md b/docs/cli.md index 2e076040..b7e41227 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -287,8 +287,9 @@ hooks. Each daemon writes its own `.pid` beside the db (`observe.db` gives `observe.pid`), so two dbs in one folder never share a pidfile. `start` exits 1 when the port already serves a different db, and `stop` refuses to signal when the daemon at the pidfile's port reports another -db, and falls back to signalling when it cannot verify (unreachable, or an old -one-line pidfile). See +db or the port answers as a different service, and falls back to signalling +when it cannot verify (unreachable, or an old one-line pidfile). `/health` +tells which db a daemon serves only to callers on the same machine. See [The observe daemon](./observability.md#the-observe-daemon). ## `lousho prompt` diff --git a/docs/errors.md b/docs/errors.md index acb14a45..637981f7 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -187,8 +187,11 @@ names the option and `error.cause` is what the function threw. **Means:** `createAgent({ prompt: 'name@version' })` (or `LOUSHO_PROMPT`) has no `instructions` fallback and the prompt could not be fetched: the observe daemon is unreachable, answered 404 for that name and version, or `LOUSHO_OBSERVE` is -not set. The first `send()` rejects and `stream()` ends with an `error` event; -the model is never called, so the agent never runs without a system prompt. The +not set. Every run rejects while the prompt stays unresolved (`send()` rejects, +`stream()` ends with an `error` event): the runs inside the 30-second retry +window at once, and each later run after its retry fetch fails while the +daemon stays down. The model is never called, so the agent never runs without +a system prompt. The message names the prompt ref. **Fix:** start the daemon (`lousho observe`) and set `LOUSHO_OBSERVE` to its URL, diff --git a/docs/observability.md b/docs/observability.md index a6f769de..e14aae48 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -237,7 +237,9 @@ the `gen_ai.*` content attributes are **never recorded by default**. Opt in per agent with `createAgent({ exporter, captureContent: true })`. The exception is a local collector: when `LOUSHO_OBSERVE` points at loopback -(`1`/`true`, or a URL whose host is `127.0.0.1`, `localhost` or `::1`), +(`1`/`true`, or a URL whose host is `127.0.0.1`, `localhost` or `::1`; +a `localhost` port that an SSH tunnel forwards to a remote daemon counts as +loopback too, so set `captureContent: false` if that daemon must not get content), `createAgent` defaults `captureContent` to `true`, so the daemon's trace view shows the prompt, the model output and the tool arguments and results that explain a failure. Only the daemon gets that content, and it stays in the @@ -425,7 +427,8 @@ The daemon also serves two stores agents can use at runtime: `instructions`, the run fails with `LOUSHO_PROMPT_UNRESOLVED`). A failed fetch is not cached for good: runs in the next 30 seconds resolve at once (using `instructions`, or failing), then the next run retries it, so a - daemon that was briefly down at build recovers without a restart. Every run logs the exact version + daemon that was briefly down at build recovers without a restart. That one + run waits for the retry fetch, up to 2.5 seconds. Every run logs the exact version used. `lousho prompt push` publishes a new version. - **Context hub.** `hub()` / `hubTools()` let agents publish and query @@ -470,8 +473,10 @@ several workers died), the sweep marks the quiet process's open runs `interrupted` once it has missed about two heartbeats (the `stale` mark), judged on that process's own heartbeat interval and on the daemon's clock. A restart on the same host with the same pid (a container's pid 1) marks them -sooner: at the new process's second heartbeat, if the old process has been -silent since the new one appeared. With no process of the agent left, the +sooner: at the first heartbeat of the new process (after its first) that +comes more than one heartbeat interval plus 10 seconds after the old +process's last, if the old process has been silent since the new one +appeared. With no process of the agent left, the agent goes `down` and its open runs are marked as above. A process that exits gracefully mid-run (SIGINT, SIGTERM) marks its own open runs `interrupted` right away. An agent with heartbeats turned off (`heartbeatIntervalMs: 0`) is never swept, so its diff --git a/src/cli/observe.test.ts b/src/cli/observe.test.ts index 7b23d36b..ab12b053 100644 --- a/src/cli/observe.test.ts +++ b/src/cli/observe.test.ts @@ -52,6 +52,11 @@ describe('reporterHint', () => { expect(reporterHint('127.0.0.1', 7790)).not.toContain('LOUSHO_OBSERVE=1'); expect(reporterHint('localhost', 7777)).toContain('LOUSHO_OBSERVE=http://localhost:7777'); }); + + it('brackets an IPv6 host', () => { + expect(reporterHint('::1', 7777)).toContain('LOUSHO_OBSERVE=http://[::1]:7777 '); + expect(reporterHint('[::1]', 7790)).toContain('LOUSHO_OBSERVE=http://[::1]:7790 '); + }); }); describe('lousho observe', () => { diff --git a/src/cli/observe.ts b/src/cli/observe.ts index 94c7c199..49b27d03 100644 --- a/src/cli/observe.ts +++ b/src/cli/observe.ts @@ -24,6 +24,7 @@ import { sameDbPath, probeHealth, startObserveDaemon, + urlHost, type HealthProbe, } from '../observe/daemon'; import { createToken, hasTokens, openDb, rebuildRollups } from '../observe/store'; @@ -115,7 +116,7 @@ function envPort(): number { /** The banner line telling agents how to reach this daemon: `=1` only reaches the default loopback port. */ export function reporterHint(host: string, port: number): string { - const value = host === '127.0.0.1' && port === OBSERVE_DEFAULT_PORT ? '1' : `http://${host}:${port}`; + const value = host === '127.0.0.1' && port === OBSERVE_DEFAULT_PORT ? '1' : `http://${urlHost(host)}:${port}`; return `lousho observe: agents emit with LOUSHO_OBSERVE=${value} (no auth on loopback).`; } @@ -136,7 +137,7 @@ async function reportPortInUse(port: number, host: string, dbPath: string): Prom ); return 1; } - console.log(`lousho observe: already running on http://${host}:${port} (service ${OBSERVE_SERVICE}, version ${health.version ?? 'unknown'}).`); + console.log(`lousho observe: already running on http://${urlHost(host)}:${port} (service ${OBSERVE_SERVICE}, version ${health.version ?? 'unknown'}).`); return 0; } @@ -162,7 +163,7 @@ async function runStart(args: string[]): Promise { throw error; } - console.log(`lousho observe: ${OBSERVE_SERVICE} v${VERSION} listening on http://${host}:${daemon.port}`); + console.log(`lousho observe: ${OBSERVE_SERVICE} v${VERSION} listening on http://${urlHost(host)}:${daemon.port}`); console.log(`lousho observe: db ${dbPath}`); if (daemon.requireToken) { console.log( diff --git a/src/observe/daemon.test.ts b/src/observe/daemon.test.ts index 3f36bfb0..141c25b9 100644 --- a/src/observe/daemon.test.ts +++ b/src/observe/daemon.test.ts @@ -5,10 +5,13 @@ * port-collision detection. */ import { afterEach, describe, expect, it } from 'vitest'; +import type * as http from 'node:http'; import * as nodePath from 'node:path'; import { defaultPidPath, isObserveDaemon, probeHealth, sameDbPath, startObserveDaemon, type ObserveDaemon } from './daemon'; import { OBSERVE_SERVICE } from './envelope'; import { createToken, openDb } from './store'; +import type { RouteContext } from './routes/httpKit'; +import { statusRoutes } from './routes/status'; const T0 = 1_700_000_000_000; @@ -76,6 +79,33 @@ describe('per-db pidfile and db identity', () => { }); }); +describe('/health db path exposure', () => { + /** Call the /health handler as if from `remote`, connected to the daemon's `local` address. */ + const healthFrom = (remote: string, local: string): Record => { + const health = statusRoutes.find((route) => route.path === '/health')!; + let body = ''; + const res = { writeHead: () => res, end: (text: string) => (body = text) } as unknown as http.ServerResponse; + const req = { socket: { remoteAddress: remote, localAddress: local } } as unknown as http.IncomingMessage; + void health.handle({ req, res, dbPath: '/home/someone/.lousho/observe.db' } as RouteContext); + return JSON.parse(body) as Record; + }; + + it('tells only a caller on this machine which db it serves', () => { + expect(healthFrom('127.0.0.1', '127.0.0.1').db).toBe(nodePath.resolve('/home/someone/.lousho/observe.db')); + expect(healthFrom('::1', '::1').db).toBeDefined(); + expect(healthFrom('::ffff:127.0.0.1', '::ffff:127.0.0.1').db).toBeDefined(); + // the CLI probing a daemon bound to this machine's LAN address connects from that same address + expect(healthFrom('192.168.1.5', '192.168.1.5').db).toBeDefined(); + }); + + it('keeps the db path (home dir, username) from a remote caller, marker intact', () => { + const body = healthFrom('192.168.1.20', '192.168.1.5'); + expect(body.db).toBeUndefined(); + expect(body.service).toBe(OBSERVE_SERVICE); + expect(body.ok).toBe(true); + }); +}); + describe('POST /api/v1/ingest', () => { it('accepts a batch and serves it back through the read endpoints', async () => { const daemon = await start(); diff --git a/src/observe/daemon.ts b/src/observe/daemon.ts index 2b563e7c..1c782882 100644 --- a/src/observe/daemon.ts +++ b/src/observe/daemon.ts @@ -262,6 +262,11 @@ export async function startObserveDaemon(options: ObserveDaemonOptions = {}): Pr }; } +/** `host` as a URL authority host: an IPv6 address (`::1`) is bracketed (`[::1]`), anything else is itself. */ +export function urlHost(host: string): string { + return host.includes(':') && !host.startsWith('[') ? `[${host}]` : host; +} + export interface HealthProbe { service: string; ok: boolean; @@ -276,7 +281,7 @@ export interface HealthProbe { */ export async function probeHealth(port: number, host = '127.0.0.1', fetchFn: typeof fetch = fetch): Promise { try { - const res = await fetchFn(`http://${host}:${port}/health`, { signal: AbortSignal.timeout(2000) }); + const res = await fetchFn(`http://${urlHost(host)}:${port}/health`, { signal: AbortSignal.timeout(2000) }); if (!res.ok) return undefined; return (await res.json()) as HealthProbe; } catch { diff --git a/src/observe/reporter.test.ts b/src/observe/reporter.test.ts index b3277fb6..20b3d672 100644 --- a/src/observe/reporter.test.ts +++ b/src/observe/reporter.test.ts @@ -495,6 +495,22 @@ describe('observeExporter', () => { expect(r.dropped).toBe(2); // the refused batch, not requeued expect(String(warn.mock.calls[0][0])).toContain('refused a batch of 2 events (413)'); }); + + it('warns about a drop without promising a retry, and still warns about a later outage', async () => { + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined); + const { fetchFn } = sizedFetch((n) => (n === 1 ? 413 : 503)); + const { r } = reporter({ fetchFn, flushSize: 1000 }); + r.onSpanEnd(makeSpan({ id: 'refused' })); + await r.flush(); // 413: dropped + r.onSpanEnd(makeSpan({ id: 'later' })); + await r.flush(); // 503: an outage, requeued + expect(warn).toHaveBeenCalledTimes(2); + const [drop, outage] = warn.mock.calls.map((call) => String(call[0])); + expect(drop).toContain('refused a batch'); + expect(drop).not.toContain('retrying'); + expect(outage).toContain('could not reach the observe daemon'); + expect(outage).toContain('retrying with backoff'); + }); }); it('clamps maxBuffered to 1 so the deregister survives', async () => { diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index b6c63cee..c51fe33f 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -31,7 +31,7 @@ * exponential backoff (1s base, 30s cap); only events evicted past the * buffer cap (5000 events or 32MB, whichever comes first), events past the * crash flush's 16MB, or still unsent after close()'s final attempt, drop into a - * `dropped` counter. It warns ONCE and never blocks or throws into a run. + * `dropped` counter. It warns ONCE per kind (an outage, a drop) and never blocks or throws into a run. * Timers are unref'd so observation never keeps a process alive. * - Its own HTTP calls are plain `fetch`, outside any span: instrumenting * them would recurse. @@ -427,32 +427,36 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe let droppedCount = 0; let closed = false; - let warned = false; + let warnedOutage = false; + let warnedDrop = false; const headers: Record = { 'content-type': 'application/json' }; if (token) headers.authorization = `Bearer ${token}`; - const warnOnce = (what: string, error: unknown) => { - if (warned) return; - warned = true; + // Outages and drops warn once each: a drop warning must not hide a later outage. + const warnOutageOnce = (what: string, error: unknown) => { + if (warnedOutage) return; + warnedOutage = true; const reason = error instanceof Error ? error.message : String(error); console.warn( `[lousho] observe: ${what} (${reason}); retrying with backoff, events past the buffer cap are counted in 'dropped' and the run is unaffected.` ); }; + const warnDropOnce = (what: string) => { + if (warnedDrop) return; + warnedDrop = true; + console.warn(`[lousho] observe: ${what}; dropped events are counted in 'dropped' and the run is unaffected.`); + }; const outbox = createOutbox(maxBuffered, MAX_BUFFERED_BYTES, now, (count) => { droppedCount += count; - warnOnce( - `buffer is full (${maxBuffered} events or ${MAX_BUFFERED_BYTES / (1024 * 1024)}MB); dropping the oldest`, - new SDKError('observe buffer full', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false }) - ); + warnDropOnce(`buffer is full (${maxBuffered} events or ${MAX_BUFFERED_BYTES / (1024 * 1024)}MB); dropping the oldest`); }); /** Count events the daemon will never take - too large, or a batch it refused - and warn once. */ const discard = (count: number, why: string): void => { droppedCount += count; - warnOnce(why, new SDKError('observe event rejected', 'LOUSHO_REMOTE_REQUEST_FAILED', { appendHelp: false })); + warnDropOnce(why); }; const push = (event: IngestEvent) => { @@ -508,7 +512,7 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe } catch (error) { // Never slow or fail a run: requeue for a later retry, drop-and-count only as the last resort. // close() aborting a regular flush is not an outage: its batch rides along in the final attempt. - if (final || !closing.signal.aborted) warnOnce(`could not reach the observe daemon at ${url}`, error); + if (final || !closing.signal.aborted) warnOutageOnce(`could not reach the observe daemon at ${url}`, error); if (final) droppedCount += batch.length + outbox.drain().length; else outbox.failed(batch); return false; diff --git a/src/observe/routes/status.ts b/src/observe/routes/status.ts index 32a84df8..eeedc572 100644 --- a/src/observe/routes/status.ts +++ b/src/observe/routes/status.ts @@ -2,22 +2,40 @@ * Health and status. `/health` is the port-collision marker: `lousho observe * start` probes it on EADDRINUSE, and the `service` field is what tells a * running collector apart from anything else that grabbed the port - it must - * stay unauthenticated for attach detection to work. `/api/v1/status` is the + * stay unauthenticated for attach detection to work. Its `db` path (a home + * dir, so a username) is told only to callers on this machine (isLocalCaller). + * `/api/v1/status` is the * row-count surface `lousho observe status` prints. */ import * as path from 'node:path'; +import type * as http from 'node:http'; +import { isLoopbackHost } from '../../server/hostCheck'; import { VERSION } from '../../version'; import { observeCounts } from '../store'; import { sendJson, type Route } from './httpKit'; +const unmapped = (address: string | undefined): string | undefined => address?.replace(/^::ffff:/i, ''); + +/** + * Whether `req` comes from this machine: a loopback peer, or one connected + * from the very address it reached (the CLI probing a daemon bound to this + * machine's LAN address). Anyone else is told the service marker only. + */ +export function isLocalCaller(req: http.IncomingMessage): boolean { + const remote = unmapped(req.socket.remoteAddress); + if (remote === undefined) return false; + return isLoopbackHost(remote) || remote === unmapped(req.socket.localAddress); +} + export const statusRoutes: readonly Route[] = [ { method: 'GET', path: '/health', - handle: ({ res, dbPath }) => { - // `db` lets `start`/`stop` tell which database a port or pid serves. - sendJson(res, 200, { service: 'lousho-observe', ok: true, version: VERSION, db: dbPath === ':memory:' ? dbPath : path.resolve(dbPath) }); + handle: ({ req, res, dbPath }) => { + // `db` lets `start`/`stop` (run on this machine) tell which database a port or pid serves. + const db = dbPath === ':memory:' ? dbPath : path.resolve(dbPath); + sendJson(res, 200, { service: 'lousho-observe', ok: true, version: VERSION, ...(isLocalCaller(req) ? { db } : {}) }); }, }, { diff --git a/src/observe/store.test.ts b/src/observe/store.test.ts index b7a6049a..98a3868e 100644 --- a/src/observe/store.test.ts +++ b/src/observe/store.test.ts @@ -447,7 +447,7 @@ describe('agent instances (one agent_id, several processes over time)', () => { } }); - it('a restart on the same host and pid interrupts the silent old instance at the new one\'s second heartbeat', () => { + it('a restart on the same host and pid interrupts the silent old instance once it missed a beat (interval + slack)', () => { const database = db(); try { const c1 = { host: 'c1', pid: 1 }; // a container restarted: pid 1 again @@ -456,7 +456,10 @@ describe('agent instances (one agent_id, several processes over time)', () => { // one heartbeat proves nothing: a live twin in another pid namespace looks the same expect(traceStatus(database, 'runA')).toBe('running'); ingestBatch(database, [beat('B', T0 + 61_000, c1)], T0 + 61_000); - // A stayed silent since before B was first seen: it is the process B replaced + // 61s since A's last beat is still a live gap (beats ride the 5s flush tick) + expect(traceStatus(database, 'runA')).toBe('running'); + ingestBatch(database, [beat('B', T0 + 121_000, c1)], T0 + 121_000); + // A stayed silent since before B was first seen, past interval + slack: it is the process B replaced expect(traceStatus(database, 'runA')).toBe('interrupted'); expect(traceStatus(database, 'runB')).toBe('running'); expect(instances(database)).toEqual(['B']); @@ -483,6 +486,24 @@ describe('agent instances (one agent_id, several processes over time)', () => { } }); + it('twins started 300ms apart never interrupt each other when one beat lands a flush tick late', () => { + const database = db(); + try { + const pid1 = { host: 'node-1', pid: 1 }; + ingestBatch(database, [beat('A', T0, pid1), openRoot('runA', 'A')], T0); + ingestBatch(database, [beat('B', T0 + 300, pid1), openRoot('runB', 'B')], T0 + 300); + for (let t = 60_000; t <= 600_000; t += 60_000) { + // B's beat lands first; A's rides its next 5s flush tick, ~4s late + ingestBatch(database, [beat('B', T0 + t + 300, pid1)], T0 + t + 300); + ingestBatch(database, [beat('A', T0 + t + 4_300, pid1)], T0 + t + 4_300); + } + expect([traceStatus(database, 'runA'), traceStatus(database, 'runB')]).toEqual(['running', 'running']); + expect(instances(database)).toEqual(['A', 'B']); + } finally { + database.close(); + } + }); + it("a dead instance's delayed final heartbeat landing after its same-pid successor started cannot interrupt the successor", () => { const database = db(); try { diff --git a/src/observe/store.ts b/src/observe/store.ts index 3e8c59ae..3b5a519c 100644 --- a/src/observe/store.ts +++ b/src/observe/store.ts @@ -202,8 +202,10 @@ interface InstanceBeat { /** * Same-host+pid instances of `agentId` that went silent before `instanceId` - * was first seen and have missed at least one of their own beats since - - * the process `instanceId` replaced. A live twin (another pid namespace: + * was first seen and have missed at least one of their own beats since + * (one interval plus FRESHNESS_SLACK_MS: beats ride the reporter's 5s flush + * tick, so live gaps run a few seconds past the interval) - the process + * `instanceId` replaced. A live twin (another pid namespace: * `--network host` replicas, hostNetwork pods) keeps heartbeating, so it is * never among them. */ @@ -213,7 +215,7 @@ const REPLACED_INSTANCES = ` WHERE old.agent_id = ? AND old.instance_id != cur.instance_id AND old.hostname = cur.hostname AND old.pid = cur.pid AND old.last_heartbeat < cur.first_seen - AND ? - old.last_heartbeat > COALESCE(old.heartbeat_interval_ms, ?) + AND ? - old.last_heartbeat > COALESCE(old.heartbeat_interval_ms, ?) + ? `; /** @@ -232,7 +234,7 @@ function recordInstance(db: SqlDatabase, agentId: string, beat: InstanceBeat, no db.prepare(INSTANCE_UPSERT).run(agentId, beat.instanceId, beat.hostname ?? null, beat.pid ?? null, beat.intervalMs ?? null, now, now); if (!seen || beat.hostname === undefined || beat.pid === undefined) return; const fallback = beat.intervalMs ?? HEARTBEAT_INTERVAL_DEFAULT_MS; - for (const row of db.prepare(REPLACED_INSTANCES).all(beat.instanceId, agentId, now, fallback)) { + for (const row of db.prepare(REPLACED_INSTANCES).all(beat.instanceId, agentId, now, fallback, FRESHNESS_SLACK_MS)) { interruptInstance(db, agentId, String(row.instance_id), now); } } From cf85c616ff920c58e5256bfbabb827339c5612d5 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:28:09 +0300 Subject: [PATCH 6/7] docs(llms): regenerate llms-full.txt for the final-review fixes Co-Authored-By: Claude Opus 5.5 --- llms-full.txt | 46 ++++++++++++++++++++++++++++++---------------- 1 file changed, 30 insertions(+), 16 deletions(-) diff --git a/llms-full.txt b/llms-full.txt index 879ae805..4f8a80c2 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -5164,8 +5164,9 @@ hooks. Each daemon writes its own `.pid` beside the db (`observe.db` gives `observe.pid`), so two dbs in one folder never share a pidfile. `start` exits 1 when the port already serves a different db, and `stop` refuses to signal when the daemon at the pidfile's port reports another -db, and falls back to signalling when it cannot verify (unreachable, or an old -one-line pidfile). See +db or the port answers as a different service, and falls back to signalling +when it cannot verify (unreachable, or an old one-line pidfile). `/health` +tells which db a daemon serves only to callers on the same machine. See [The observe daemon](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md#the-observe-daemon). ## `lousho prompt` @@ -8496,8 +8497,11 @@ names the option and `error.cause` is what the function threw. **Means:** `createAgent({ prompt: 'name@version' })` (or `LOUSHO_PROMPT`) has no `instructions` fallback and the prompt could not be fetched: the observe daemon is unreachable, answered 404 for that name and version, or `LOUSHO_OBSERVE` is -not set. The first `send()` rejects and `stream()` ends with an `error` event; -the model is never called, so the agent never runs without a system prompt. The +not set. Every run rejects while the prompt stays unresolved (`send()` rejects, +`stream()` ends with an `error` event): the runs inside the 30-second retry +window at once, and each later run after its retry fetch fails while the +daemon stays down. The model is never called, so the agent never runs without +a system prompt. The message names the prompt ref. **Fix:** start the daemon (`lousho observe`) and set `LOUSHO_OBSERVE` to its URL, @@ -9984,7 +9988,7 @@ 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. +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. The score rows ride the same queue as the spans: they go out after the case's span and, while the collector is down, are kept and retried with it (and sent on exit or crash) instead of being dropped. When a wait runs out, the cases of the next 30 seconds skip the wait (their rows still queue), then the next case tries again. ### Time limits @@ -13921,12 +13925,18 @@ the `gen_ai.*` content attributes are **never recorded by default**. Opt in per agent with `createAgent({ exporter, captureContent: true })`. The exception is a local collector: when `LOUSHO_OBSERVE` points at loopback -(`1`/`true`, or a URL whose host is `127.0.0.1`, `localhost` or `::1`), +(`1`/`true`, or a URL whose host is `127.0.0.1`, `localhost` or `::1`; +a `localhost` port that an SSH tunnel forwards to a remote daemon counts as +loopback too, so set `captureContent: false` if that daemon must not get content), `createAgent` defaults `captureContent` to `true`, so the daemon's trace view shows the prompt, the model output and the tool arguments and results that -explain a failure. The data stays in the daemon's db on your machine. Turn it -off with `captureContent: false`; a collector beyond loopback keeps the -opt-in default. The default applies to agents built with `createAgent`. +explain a failure. Only the daemon gets that content, and it stays in the +daemon's db on your machine: the agent's own `exporter` (`fileTraceExporter()` +files, an OTLP or vendor exporter, the built server's OTel export) still sees +every span without it, unless the agent sets `captureContent: true` or the +OTel content variable below is `true`. Turn it off with `captureContent: false`; +a collector beyond loopback keeps the opt-in default. The default applies to +agents built with `createAgent`. `redactContent` keeps only the legacy `input`/`prompt`/`args`/`result` attributes off; it does not scrub `gen_ai.*` message content. On a loopback @@ -14105,7 +14115,8 @@ The daemon also serves two stores agents can use at runtime: `instructions`, the run fails with `LOUSHO_PROMPT_UNRESOLVED`). A failed fetch is not cached for good: runs in the next 30 seconds resolve at once (using `instructions`, or failing), then the next run retries it, so a - daemon that was briefly down at build recovers without a restart. Every run logs the exact version + daemon that was briefly down at build recovers without a restart. That one + run waits for the retry fetch, up to 2.5 seconds. Every run logs the exact version used. `lousho prompt push` publishes a new version. - **Context hub.** `hub()` / `hubTools()` let agents publish and query @@ -14118,16 +14129,17 @@ down, the failed batch is requeued and retried with exponential backoff (1 s, backlog drains in batches one right after another, each at most 1000 events and 4 MB of JSON (under the daemon's per-request caps of 1000 events and 5 MB). Only these are dropped and counted in the reporter's `dropped`: -events evicted past the buffer cap (`maxBuffered`, default 5000, oldest -first); events still unsent after the exit flush's final attempt; a single +events evicted past the buffer cap (`maxBuffered`, default 5000 events, or +32 MB of serialized events, whichever comes first; oldest first); events +past the newest 16 MB when a crash flushes; events still unsent after the exit flush's final attempt; a single event too large for any request (over 5 MB serialized, for example with content capture on); and a batch the daemon refuses outright (`400`/`413`), which is never retried. A process that dies of an uncaught exception or an unhandled rejection still reports its last run: the reporter watches `uncaughtExceptionMonitor` and -sends what it holds, oldest first, holding up the crash for up to 1 s per -reporter. Node's error output and exit code stay as they were, and a script +sends what it holds (the newest 16 MB of it), oldest first, holding up the +crash for up to 1 s per reporter. Node's error output and exit code stay as they were, and a script that survives the throw (its own `uncaughtException` handler, or a capture callback such as the REPL's) keeps reporting. An explicit `process.exit()` or a SIGKILL still loses what is buffered. @@ -14149,8 +14161,10 @@ several workers died), the sweep marks the quiet process's open runs `interrupted` once it has missed about two heartbeats (the `stale` mark), judged on that process's own heartbeat interval and on the daemon's clock. A restart on the same host with the same pid (a container's pid 1) marks them -sooner: at the new process's second heartbeat, if the old process has been -silent since the new one appeared. With no process of the agent left, the +sooner: at the first heartbeat of the new process (after its first) that +comes more than one heartbeat interval plus 10 seconds after the old +process's last, if the old process has been silent since the new one +appeared. With no process of the agent left, the agent goes `down` and its open runs are marked as above. A process that exits gracefully mid-run (SIGINT, SIGTERM) marks its own open runs `interrupted` right away. An agent with heartbeats turned off (`heartbeatIntervalMs: 0`) is never swept, so its From 2af948c348a76e63354a9813553041ece179e9da Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Sat, 24 Oct 2026 20:33:35 +0300 Subject: [PATCH 7/7] refactor(observe): decide the observe-only content default inside the helper Moves the "capture is on only because of the loopback default" check out of createAgent into capturedContentForObserveOnly, so createAgent stays under the fallow complexity limits; isLocalCaller is no longer exported. Co-Authored-By: Claude Opus 5.5 --- src/createAgent.ts | 5 ++--- src/observe/reporter.ts | 25 +++++++++++++++++-------- src/observe/routes/status.ts | 2 +- 3 files changed, 20 insertions(+), 12 deletions(-) diff --git a/src/createAgent.ts b/src/createAgent.ts index a2e76d3d..8802ac23 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -25,7 +25,7 @@ import { capturedContentForObserveOnly, withObserveExporter, type PromptUseProvi import { parsePromptRef, registryPrompt } from './observe/promptClient'; import { assertModelSettings, mergeModelSettings } from './execution/modelSettings'; import { assertKnownCreateAgentKeys } from './createAgentKeys'; -import { DEFAULT_AGENT_NAME, resolveCaptureContent } from './execution/genAiSpans'; +import { DEFAULT_AGENT_NAME } from './execution/genAiSpans'; import { LLMProvider, LLMProviderRegistry } from './providers/llm'; import { ToolRegistry } from './tools/ToolRegistry'; import { ToolDescriptor } from './types'; @@ -1074,9 +1074,8 @@ export function createAgent // That default is for the local daemon only: the config's own exporters // (trace files, OTLP, a vendor) keep the content-free default unless the // agent or the OTel content env var asked for content. - const contentForObserveOnly = config.captureContent === undefined && captureContent === true && !resolveCaptureContent(undefined); const exporter = withObserveExporter( - contentForObserveOnly ? capturedContentForObserveOnly(config.exporter) : config.exporter, + capturedContentForObserveOnly(config.exporter, { explicit: config.captureContent, captureContent }), agentName, resolvedInstructions.promptUse !== undefined ? { promptUse: resolvedInstructions.promptUse } : {} ); diff --git a/src/observe/reporter.ts b/src/observe/reporter.ts index c51fe33f..5c2f5b5c 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -41,7 +41,7 @@ * `gen_ai.conversation.id` when the run has a real session. */ -import { GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/semconv'; +import { CAPTURE_CONTENT_ENV, GenAiAttr, GenAiOperation, LegacyAttr, SdkAttr } from '../execution/semconv'; import { SDKError } from '../execution/errors'; import type { Span, TraceExporter } from '../execution/tracing'; import { VERSION } from '../version'; @@ -864,15 +864,24 @@ function withoutCapturedContent(span: Span): Span { return { ...span, attributes }; } +/** The OTel content env var asks for content on every exporter. */ +const otelContentEnvOn = (): boolean => + typeof process !== 'undefined' && process.env?.[CAPTURE_CONTENT_ENV]?.toLowerCase() === 'true'; + /** - * `exporter` with the captured `gen_ai.*` content kept for its `LOUSHO_OBSERVE` - * reporters only: every other part (a `fileTraceExporter()`, an OTLP or vendor - * exporter) sees each span without it. createAgent applies this when content - * capture is on only because the env collector is loopback - that default is - * for the local daemon, not for every exporter the agent has. + * The config exporter an agent reports with, given its content settings: when + * capture is on only because the env collector is loopback (`captureContent` + * true but no `explicit` setting and no OTel content env var), the captured + * `gen_ai.*` content is kept for the `LOUSHO_OBSERVE` reporters alone and every + * other part (a `fileTraceExporter()`, an OTLP or vendor exporter) sees each + * span without it - that default is for the local daemon, not for every + * exporter the agent has. Otherwise `exporter` itself. */ -export function capturedContentForObserveOnly(exporter: TraceExporter | undefined): TraceExporter | undefined { - if (!exporter) return undefined; +export function capturedContentForObserveOnly( + exporter: TraceExporter | undefined, + content: { explicit: boolean | undefined; captureContent: boolean | undefined } +): TraceExporter | undefined { + if (!exporter || content.explicit !== undefined || content.captureContent !== true || otelContentEnvOn()) return exporter; return fanOutExporters( exporterParts(exporter).map((part): TraceExporter => envReporterSet.has(part) diff --git a/src/observe/routes/status.ts b/src/observe/routes/status.ts index eeedc572..b83defd5 100644 --- a/src/observe/routes/status.ts +++ b/src/observe/routes/status.ts @@ -22,7 +22,7 @@ const unmapped = (address: string | undefined): string | undefined => address?.r * from the very address it reached (the CLI probing a daemon bound to this * machine's LAN address). Anyone else is told the service marker only. */ -export function isLocalCaller(req: http.IncomingMessage): boolean { +function isLocalCaller(req: http.IncomingMessage): boolean { const remote = unmapped(req.socket.remoteAddress); if (remote === undefined) return false; return isLoopbackHost(remote) || remote === unmapped(req.socket.localAddress);