diff --git a/CHANGELOG.md b/CHANGELOG.md index 33e403ff..b0dc7b4d 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. `/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. ## [1.0.0-alpha.29] - 2026-10-21 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/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/docs/observability.md b/docs/observability.md index 1eb7722a..e14aae48 100644 --- a/docs/observability.md +++ b/docs/observability.md @@ -237,12 +237,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 @@ -421,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 @@ -434,16 +441,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. @@ -465,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/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 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/createAgent.ts b/src/createAgent.ts index 909148a1..8802ac23 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -21,7 +21,7 @@ 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'; @@ -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 exporter = withObserveExporter( - config.exporter, + capturedContentForObserveOnly(config.exporter, { explicit: config.captureContent, captureContent }), 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/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/__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/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/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.test.ts b/src/observe/reporter.test.ts index 2207918e..20b3d672 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 }); @@ -408,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(); @@ -454,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 () => { @@ -644,7 +701,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 @@ -711,11 +768,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 0f17ccde..5c2f5b5c 100644 --- a/src/observe/reporter.ts +++ b/src/observe/reporter.ts @@ -29,8 +29,9 @@ * 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 - * `dropped` counter. It warns ONCE and never blocks or throws into a run. + * 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 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. @@ -40,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'; @@ -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; @@ -111,10 +112,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; @@ -130,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'); @@ -172,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 }; } @@ -207,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 { @@ -237,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 { @@ -255,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++); }, @@ -388,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, now, (count) => { + const outbox = createOutbox(maxBuffered, MAX_BUFFERED_BYTES, now, (count) => { droppedCount += count; - warnOnce( - `buffer is full (${maxBuffered} events); 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) => { @@ -469,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; @@ -483,21 +526,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; @@ -581,13 +625,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; @@ -629,7 +676,7 @@ export function observeExporter(options: ObserveExporterOptions = {}): ObserveRe } } - return { + const reporter: ObserveReporter = { get dropped() { return droppedCount; }, @@ -666,9 +713,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 */ @@ -781,6 +847,53 @@ 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 }; +} + +/** The OTel content env var asks for content on every exporter. */ +const otelContentEnvOn = (): boolean => + typeof process !== 'undefined' && process.env?.[CAPTURE_CONTENT_ENV]?.toLowerCase() === 'true'; + +/** + * 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, + 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) + ? 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 diff --git a/src/observe/routes/status.ts b/src/observe/routes/status.ts index 32a84df8..b83defd5 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. + */ +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); } }