diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 000000000..b96cb9a6c --- /dev/null +++ b/.dockerignore @@ -0,0 +1,10 @@ +node_modules +.git +dist +eval +benchmark +test +.claude +.serena +docs +*.log diff --git a/deploy/README.md b/deploy/README.md index 91aa199e2..a84f653aa 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -1,15 +1,10 @@ # One-click deploy templates Stand up agentmemory on managed infrastructure without rolling your own -Docker host. Each template ships a self-contained Dockerfile that pulls -`@agentmemory/agentmemory` from npm at build time and copies the iii -engine binary in from the official `iiidev/iii` image — no pre-built -agentmemory image required. Storage mounts at `/data`; an HMAC secret -is generated by the first-boot entrypoint and persisted to the volume. -The entrypoint overwrites the npm-bundled iii config with a -deploy-tuned one that binds `0.0.0.0` and uses absolute `/data` paths, -then drops privileges from `root` to `node` via `gosu` before -exec'ing the agentmemory CLI. +Docker host. Each template ships a self-contained Dockerfile. Fly, Render, and Coolify +install `@agentmemory/agentmemory` from npm at build time. Railway builds it +from this repo's source instead (see `deploy/railway/README.md`). All four +copy the iii engine binary in from the `iiidev/iii` image. | Platform | Pitch | Cost floor | |----------|-------|------------| @@ -95,6 +90,7 @@ agentmemory worker reg : 2.0 s healthcheck passes : ~9-10 s ``` -Every template's health-check `grace_period` (or compose +Railway's `healthcheckTimeout` is 60 s (the BM25 startup backfill needs it). +Every other template's health-check `grace_period` (or compose `start_period`) is set to 30 s for a 3x safety margin. Tune lower once you've measured your own platform's image-pull characteristics. diff --git a/deploy/railway/Dockerfile b/deploy/railway/Dockerfile index e09469988..eb884dc24 100644 --- a/deploy/railway/Dockerfile +++ b/deploy/railway/Dockerfile @@ -2,9 +2,29 @@ ARG III_VERSION=0.11.2 FROM iiidev/iii:${III_VERSION} AS iii-image +# Build the package from this repo rather than pulling it from the registry, so +# fixes that live on a fork actually reach the deployed container. `npm pack` +# produces the same tarball shape the registry would serve, which keeps the +# install layout below identical to the previous registry install. +FROM node:22-slim AS builder +WORKDIR /build +# No lockfile: .gitignore:23 excludes it by repo policy, so it is absent from +# any git-based build context and `npm ci` cannot run. This trades build +# reproducibility away, which is the policy's cost, not a choice made here. +COPY package.json ./ +# node:22-slim ships npm 10.9.x, whose arborist fails this dependency tree with +# "Cannot read properties of null (reading 'edgesOut')" when resolving without a +# lockfile. Reproduced on a clean git-only context; npm 11 resolves it cleanly. +RUN npm install -g npm@11.19.0 --no-fund --no-audit \ + && npm install --no-fund --no-audit +COPY . . +# npm pack does not create --pack-destination, so make it first. +RUN mkdir -p /out \ + && npm run build \ + && npm pack --pack-destination /out + FROM node:22-slim -ARG AGENTMEMORY_VERSION=0.9.29 ARG III_VERSION=0.11.2 ARG III_SDK_VERSION=0.11.2 @@ -14,21 +34,44 @@ RUN apt-get update \ COPY --from=iii-image /app/iii /usr/local/bin/iii -# Install agentmemory into a dedicated prefix so the local package.json's -# `overrides` field pins iii-sdk down to match the engine (agentmemory's -# caret range `^0.11.2` otherwise resolves to 0.11.6, the version that -# requires the new sandbox-everything worker model the agentmemory CLI -# is not refactored for yet). `npm install -g` ignores overrides, hence -# the local prefix. +# Install into a dedicated prefix. The path below is load-bearing: entrypoint.sh +# writes the iii worker config to +# /opt/agentmemory/node_modules/@agentmemory/agentmemory/dist/iii-config.yaml +# under `set -eu`, so moving this prefix kills the container at boot rather than +# at build time. Installing the packed tarball keeps npm placing the package at +# node_modules//, which reproduces that exact path. WORKDIR /opt/agentmemory -RUN printf '{"name":"agentmemory-deploy","version":"1.0.0","private":true,"overrides":{"iii-sdk":"%s"}}\n' "${III_SDK_VERSION}" > package.json \ - && npm install "@agentmemory/agentmemory@${AGENTMEMORY_VERSION}" --omit=optional --no-fund --no-audit \ +COPY --from=builder /out/*.tgz /tmp/agentmemory.tgz +# The runtime install resolves the tarball's dependencies against the registry, +# so the repo's `overrides` never reach it on their own. These are CVE pins +# (added by 91c78e7, "clear all 8 audit vulns"). Read them out of the packed +# tarball rather than retyping them, so a new pin in package.json cannot ship +# unpinned here without anyone noticing. +RUN OVERRIDES="$(tar -xzOf /tmp/agentmemory.tgz package/package.json | node -p 'JSON.stringify((JSON.parse(require("fs").readFileSync(0,"utf8")).overrides)||{})')" \ + && printf '{"name":"agentmemory-deploy","version":"1.0.0","private":true,"overrides":%s}\n' "$OVERRIDES" > package.json \ + && npm install /tmp/agentmemory.tgz --omit=optional --no-fund --no-audit \ + && rm -f /tmp/agentmemory.tgz \ && ln -s /opt/agentmemory/node_modules/.bin/agentmemory /usr/local/bin/agentmemory +# The previous Dockerfile pinned iii-sdk through an `overrides` block to stop a +# caret range resolving past the engine. Both this repo and the published package +# now pin iii-sdk exactly, so the override is redundant. Assert the resolved +# version instead of carrying a workaround for a condition that may not hold: a +# mismatch fails the build here rather than at runtime in production. +RUN RESOLVED="$(node -p "require('/opt/agentmemory/node_modules/iii-sdk/package.json').version")" \ + && echo "resolved iii-sdk: ${RESOLVED} (expected ${III_SDK_VERSION})" \ + && [ "${RESOLVED}" = "${III_SDK_VERSION}" ] + +# Prove the build produced a usable install at the path the entrypoint expects. +# Without this the image can ship a broken layout that only fails at boot, where +# failure costs a restart-budget entry instead of a build. +RUN test -f /opt/agentmemory/node_modules/@agentmemory/agentmemory/dist/iii-config.yaml \ + && test -x /usr/local/bin/agentmemory + ENV AGENTMEMORY_III_VERSION=${III_VERSION} \ TINI_SUBREAPER=1 -COPY --chmod=0755 entrypoint.sh /usr/local/bin/agentmemory-entrypoint.sh +COPY --chmod=0755 deploy/railway/entrypoint.sh /usr/local/bin/agentmemory-entrypoint.sh EXPOSE 3111 diff --git a/deploy/railway/README.md b/deploy/railway/README.md index 9aad4fb32..9ec128ee3 100644 --- a/deploy/railway/README.md +++ b/deploy/railway/README.md @@ -65,7 +65,9 @@ For an authenticated call, your client must send `Authorization: Bearer ## Viewer access (port 3113 stays internal) Railway only exposes the single public port from your service's -`PORT` env var (which we map to 3111). The viewer stays bound to +`PORT` env var. The container always serves on 3111 (the app reads +`III_REST_PORT`, never `PORT`), so the service's target port must be set to +3111 in the Railway dashboard. The viewer stays bound to localhost inside the container. `railway ssh` is an interactive shell only — it does not support `-L`-style port forwarding, so reach the viewer with one of the following. @@ -90,6 +92,39 @@ expose that port through a second Railway TCP Proxy, then use a native `ssh -L 3113:localhost:3113 -p ` from your laptop. This is the heavier path; option A is what most users will want. +## Self-healing + +Railway only queries `healthcheckPath` at deploy time and only restarts on a +process **exit**, so a container that stops serving without exiting is invisible +to the platform. Every wedge observed so far has the same shape: the iii engine +process dies and the node process keeps running, reconnecting forever. The +engine owns the REST listener, the stream port, and the state store, so nothing +can be served once it is gone. + +| Variable | Default | Effect | +|---|---|---| +| `AGENTMEMORY_EXIT_ON_ENGINE_DEATH` | **on** | Exits non-zero when the engine dies more than 5s after it was spawned, so the platform restarts the container. Set `0` to disable. | +| `AGENTMEMORY_HEALTH_ESCALATE` | unset (**off**) | Exits after 10 consecutive failures of the 30s KV probe. | + +`AGENTMEMORY_EXIT_ON_ENGINE_DEATH` is on by default because it acts on a +process-exit event rather than a probe: there is no threshold to tune and no +false positive to trade against. Deaths within the first 5 seconds keep the +startup-failure path, which reports a clearer message. + +`AGENTMEMORY_HEALTH_ESCALATE` is a probe and stays off. Do not enable it until an +external uptime check exists against `/agentmemory/livez`. It forces process +exits and `restartPolicyMaxRetries` is 10, so a wedge recurring on every boot +reaches a stopped deployment in under an hour with nothing notifying you. +Truthy spellings are `1`, `true`, `TRUE`. **`yes` and `on` are rejected.** + +`restartPolicyType` is `ALWAYS` rather than `ON_FAILURE`: the SIGTERM shutdown +path ends in `process.exit(0)`, which `ON_FAILURE` reads as success and would not +restart. + +An earlier revision of this file documented an in-container shell watchdog +(`AGENTMEMORY_WATCHDOG*`). That was removed on 2026-08-26 in favour of the +engine-exit handler above plus external uptime monitoring. + ## Rotate the HMAC secret ```bash @@ -132,5 +167,5 @@ See for the current rate card. or use the dashboard's manual snapshot feature. - The Dockerfile builds on Railway's builder on every deploy. First deploy is ~2 minutes; cached layers make subsequent rebuilds quick. - Pin `AGENTMEMORY_VERSION` / `III_VERSION` build args in the + Pin the `III_VERSION` build arg in the service's *Variables* tab to lock a specific release. diff --git a/deploy/railway/entrypoint.sh b/deploy/railway/entrypoint.sh index ffdd63339..7cde41a29 100755 --- a/deploy/railway/entrypoint.sh +++ b/deploy/railway/entrypoint.sh @@ -67,7 +67,9 @@ workers: file_path: /data/stream_store - name: iii-observability config: - enabled: true + # false, unlike the other three deploy targets: the in-memory OTEL + # exporter drove heap growth that crashed the container (2026-08-23). + enabled: false service_name: agentmemory exporter: memory sampling_ratio: 1.0 diff --git a/deploy/railway/railway.json b/deploy/railway/railway.json index 43f52173b..427298885 100644 --- a/deploy/railway/railway.json +++ b/deploy/railway/railway.json @@ -7,8 +7,8 @@ "deploy": { "numReplicas": 1, "healthcheckPath": "/agentmemory/livez", - "healthcheckTimeout": 30, - "restartPolicyType": "ON_FAILURE", + "healthcheckTimeout": 60, + "restartPolicyType": "ALWAYS", "restartPolicyMaxRetries": 10, "requiredMountPath": "/data" } diff --git a/docs/plans/2026-08-25-001-infra-agentmemory-hang-resilience-plan.md b/docs/plans/2026-08-25-001-infra-agentmemory-hang-resilience-plan.md new file mode 100644 index 000000000..7e67c8373 --- /dev/null +++ b/docs/plans/2026-08-25-001-infra-agentmemory-hang-resilience-plan.md @@ -0,0 +1,773 @@ +--- +artifact_contract: ce-unified-plan/v1 +artifact_readiness: implementation-ready +product_contract_source: ce-plan-bootstrap +execution: code +created: 2026-08-25 +depth: deep +branch: fix/1223-heap-severity-denominator +--- + +# agentmemory hang resilience + +## Goal Capsule + +**Objective.** A wedged agentmemory backend recovers without a human, and a human +finds out that it happened. + +**Means.** Two independent detection layers plus a notification layer, and the +deploy-path change that lets the in-process layer actually ship (KTD1, KTD6). + +**Authority hierarchy.** Product Contract outranks Planning Contract. A user +instruction in session outranks both. Where this plan and +`~/.claude/rules/destructive-commands.md` disagree about running a deploy, the +rule wins: draft, show target, confirm. + +**Stop conditions.** Stop and ask before any `railway up`, before any `railway +redeploy`, and before opening the upstream PR. Stop if U1's source build produces +a `dist/` that fails U1's parity check. + +**Execution profile.** Verification-first. Every unit that changes runtime +behaviour carries a check that fails if the behaviour breaks. + +**Tail ownership.** `ce-work` implements. Multi-lens subagent review follows, per +the user's instruction and `~/.claude/rules/code-review-methodology.md`. + +--- + +## Product Contract + +### Summary + +On 2026-08-25 the agentmemory Railway service stopped serving while its process +stayed alive. Railway reported `Online` for about 8 hours. This was the second +occurrence. Nothing detected it and nothing recovered it. This plan makes the +failure self-healing and visible. + +### Problem Frame + +Two platform mechanisms should have caught it. Neither can, and the gap between +them is exactly where this failure lives. + +- `healthcheckPath: /agentmemory/livez` gates a **new deploy** only. Railway's + docs state: "Railway does not monitor the healthcheck endpoint after the + deployment has gone live." +- `restartPolicyType: ON_FAILURE` fires when the process **exits**. This process + hangs and never exits. + +**CORRECTED 2026-08-26: the root cause below is wrong, and the correction is +load-bearing for every unit in this plan.** This is not an application hang. The +**iii engine process dies and the node process keeps running.** Verified by +`railway ssh` into a wedged container: `/proc` held only `tini` and +`node /usr/local/bin/agentmemory`, with **no `iii` process at all**; +`/proc/net/tcp` had one listener, `127.0.0.1:3113` (the viewer, owned by node), +while 3111, 3112 and 49134 were all unbound; `/data/state_store.db` mtime equalled +`last_ok` exactly. + +The 3111 REST listener lives **inside the engine**, so it dies first, with the +engine, not last. That inverts F1 and F2 below. The `state::set` timeout in the +table is a **consequence** of the engine already being gone: an invocation with +no peer to answer it, timing out at the engine's own 180000 ms setting. The +engine died at 05:59:11Z, where the `ws` Sender errors are. + +A dead engine is therefore invisible to the platform. Worse, the application +already collects the signal that would identify it and then discards it: +`src/health/monitor.ts` probes `kv.set`/`kv.get` every 30 seconds and measures +event-loop lag, writes the snapshot to KV, and returns. `src/health/thresholds.ts` +never reads `kvConnectivity`, and nothing acts on `status: "critical"`. + +The observed failure sequence, from the Railway logs: + +| Time (UTC) | Event | +|---|---| +| 08-24 20:45:00 | Deploy `a201e22d` succeeds | +| 08-24 20:45:46 | `[iii] Reconnecting` attempt 1 (starts at boot, unrelated to the fault) | +| 08-25 05:59:11 | `ws` Sender errors in `node_modules/ws/lib/sender.js` | +| 08-25 06:05:10 | `error Compression failed: "Invocation timeout after 180000ms: state::set"` | +| 08-25 06:08:19 | Last application log line, then silence | + +Memory sat at 663.9 MB against an 8192 MB limit with CPU idle, so this was a hang +and not an out-of-memory event. + +### Requirements + +- **R1.** A process that stops serving on 3111 must be restarted without human + action. +- **R2.** The restart mechanism must not exhaust `restartPolicyMaxRetries: 10` + and leave the deployment permanently `CRASHED`. That outcome is strictly worse + than the bug it replaces. +- **R3.** An outage, and a restart loop, must reach a human without a human + polling for it. +- **R4.** Source-level fixes held in this fork must reach production. Today they + cannot. +- **R5.** The health monitor must act on the failure signal it already collects, + rather than recording it and returning. +- **R6.** No detection path may depend on `/agentmemory/health` while upstream + issue #1223 is unfixed in production, because that route returns 503 on a + healthy process. +- **R7.** The source-level fixes must be offered upstream to + `rohitg00/agentmemory`, not held only in the fork. + +### Actors + +- **A1. The container process.** `agentmemory` under `tini`, PID equal to the + entrypoint shell's PID after `exec`. +- **A2. The Railway platform.** Owns deploy-time healthchecks and the + `ON_FAILURE` restart policy. Sees process exits, not hangs. +- **A3. The operator.** Currently the only detector. This plan removes that role + from the recovery path and leaves it in the notification path. +- **A4. MCP clients.** Claude Code, Codex, and OpenCode. All read + `AGENTMEMORY_URL=http://127.0.0.1:8899`, so they degrade and recover together. + +### Key Flows + +- **F1. Listener death.** The 3111 listener disappears while the process lives. + The out-of-process watchdog detects it and forces an exit. Railway restarts. +- **F2. KV stall with a live loop.** `state::set` begins timing out while the + event loop still runs. The in-process monitor detects it and exits before the + listener dies. This is the path the real outage took at 06:05Z. +- **F3. Restart loop.** A deterministic wedge recurs on every boot. The + once-per-boot cap bounds it, and the notification layer surfaces it. + +### Acceptance Examples + +- **AE1.** Given a live container, when the 3111 listener stops answering, then + the container exits and restarts within roughly 4 minutes, and + `/agentmemory/livez` returns 200 without operator action. Covers R1, F1. +- **AE2.** Given a live node process, when the KV probe fails on **10** + consecutive 30-second collections, then the process exits. The threshold is 10, + not 3: `src/index.ts` documents `state::set` exceeding the SDK's 30 s timeout + under sustained hook load, and a 5 s probe cannot tell a slow store from a dead + one. An earlier draft said "before the HTTP listener dies" -- **that is wrong**, + the listener belongs to the engine and is already gone. Covers R5, F2. +- **AE3.** Given a wedge that recurs on every boot, when the watchdog acts, then + it acts at most once per container lifetime, and watchdog action alone never + drives the deployment to `CRASHED`. Covers R2, F3. +- **AE4.** Given a source-built image, when it is deployed, then + `grep -rl heapSizeLimit /dist/` returns at least one file. + **`dist/` is bundled, not per-module** — `tsdown` emits hash-suffixed chunks + (`src-CzgoepGU.mjs`, `index.mjs`) and there is no `dist/health/` directory, so + the assertion must be a recursive grep over `dist/` and never a fixed module + path. Verified against the live container: `heapSizeLimit` appears **nowhere** + in the deployed `dist/`, which is the current failing state this proves out of. + Covers R4. +- **AE5.** Given the backend is down, when the uptime check next runs, then a + notification reaches the operator without the operator looking. Covers R3. + +### Success Criteria + +- No operator action is required to recover from a hang. +- No hang exceeds roughly 5 minutes of downtime. +- Every recovery event and every restart loop produces a notification. +- The four health commits on this branch run in production. + +### Scope Boundaries + +**In scope.** The Railway deploy target. The entrypoint watchdog. The in-process +health escalation. The Dockerfile source build. External uptime monitoring. An +upstream PR. + +**Out of scope.** +- Fixing the `iii` engine's `state::set` hang at its source. The cause sits in a + third-party engine, the payoff is uncertain, and every layer here makes the + hang survivable regardless. +- Restructuring the health module's architecture. +- `deploy/fly/`, `deploy/render/`, and `deploy/coolify/`. Each carries the same + gap in its own `entrypoint.sh`. None is deployed, so none can be verified, and + unverified copies drift. Recorded as follow-up per the user's decision. +- The MCP 7-tool degradation. It is a symptom of backend downtime, not a separate + defect. It resolves when the backend stays up, and it needs an MCP server + restart per host because clients fetch `tools/list` once at connect. + +### Dependencies + +- Railway CLI, authenticated, linked at `~/rw-agentmemory`. +- `railway ssh` access for in-container verification. +- An uptime-check provider for U5. +- A GitHub account able to open a PR against `rohitg00/agentmemory`. The fork + `inix-x/agentmemory` already exists as `origin`. + +### Outstanding Questions + +- **Q1 (deferred, with a resolution path).** The running image is 0.9.28 while + the Dockerfile ARG pins 0.9.29, so a rebuild on the current Dockerfile would + bump the app version alongside any entrypoint change. The user parked this + decision. **U1 dissolves it:** once the image builds from the repo, the app + version stops being an npm pin and becomes the repo state, so there is no + separate bump to decide. If U1 is dropped or deferred, Q1 becomes blocking + again and must return to the user. Do not decide it silently. + +### Sources + +- Railway docs, healthchecks: "Railway does not monitor the healthcheck endpoint + after the deployment has gone live." Retrieved via context7, 2026-08-25. +- Upstream issue #1223, heap severity denominator. Fixed locally on this branch, + absent from the deployed package. +- Live container inspection via `railway ssh`, 2026-08-25. + +--- + +## Planning Contract + +### Key Technical Decisions + +- **KTD1. RETIRED 2026-08-26. The out-of-process shell watchdog was removed.** + The original reasoning was that two layers cover each other's blind spots. Two + things retired it. + + First, the root cause turned out to be **engine process death**, not an + application hang, and `src/cli.ts` already registers `child.on("exit")` on the + detached engine carrying the exit code, the signal, and its dying stderr. That + is a direct event, detected in milliseconds, with no probe, no polling, and no + false-positive threshold. It replaces ~93 lines of shell with a handler that + already existed. + + Second, U5 external monitoring is mandatory before either killer may be enabled + (KTD5), so it exists either way. `livez` is unauthenticated and static, so an + external check sees exactly what an in-container `curl` sees, **and** survives a + whole-container wedge the in-container subshell cannot. Railway exposes + `deploymentRestart` on its public API. + + What is lost, stated rather than glossed: an external monitor needs a Railway + token held at the provider, `deploymentRestart` takes a per-deploy id so it is a + small script rather than a bare URL, and an egress failure between monitor and + Railway can cause a restart the in-container watchdog would not have. + Governs R1, R5. + +- **KTD2. Probe `/agentmemory/livez`, never `/agentmemory/health`.** `livez` is + unauthenticated (`src/triggers/api.ts:180-192` registers `api::liveness` with no + `checkAuth`, verified by plain curl returning 200) and returns a static payload. + `health` returns 503 on a healthy process in production because #1223 is + unfixed there. Governs R6. + +- **KTD3. Hardcode 3111 in the watchdog. Never use `$PORT`.** The application + never reads `PORT`; `entrypoint.sh` writes `port: 3111` into the iii-http worker + config directly. `PORT` happens to equal 3111 today, verified in-container. If + Railway ever injected a different value, `${PORT:-3111}` would poll a dead port, + never arm, and fail silently. The literal is the more correct choice here. + +- **KTD4. SIGTERM first, and know that the flush it buys never happens.** An + earlier draft justified the grace period as protecting `/data/state_store.db` + from corruption. **Two separate findings retire that reasoning.** First, the + engine is a detached process in its own session that owns the store, so a + single-PID SIGTERM cannot reach it and it dies by namespace teardown either + way. Second, `shutdown` awaits `indexPersistence.save()` -> `kv.set` -> + `state::set`, which parks for the engine's 180 s invocation timeout against a + dead engine, while the grace is 15 s. The hard exit always wins, so the flush + never completes in the failure this targets. + + The grace stays for the case where the engine is alive and only node is being + restarted, where the flush does complete. It is not a corruption guard, and + the plan should not claim it is. Separately, Railway already sends SIGKILL + after 0 s of draining on every deploy, so the store has survived mid-write + kills routinely. Governs R2. + +- **KTD5. Ship the notification layer first. The once-per-boot cap is weaker than + it looks.** An earlier draft claimed the cap "bounds a deterministic wedge to + one restart per deploy." **That is wrong.** The watchdog's `exit 0` / `exit 1` + are the *subshell's* status, not the container's, and by the time either runs + PID 1 has already been signalled. The cap bounds one kill per **container**, + and containers are unbounded up to `restartPolicyMaxRetries`. Every restart is + a fresh container with a fresh watchdog. + + Measured cycle times, not assumed: the shell watchdog loops in roughly 3.5 to 4 + minutes, reaching `CRASHED` in about 35 to 40 minutes; the in-process layer + loops in about 5.3 minutes, reaching it in roughly 53 minutes. The `armed` + guard and the `until curl` loop only close the never-healthy-at-boot case, not + wedge-after-boot. + + So U5 is not a nicety that makes enabling tidier. It is the only thing that + makes a restart loop visible before the budget is gone, which is why it gates + U6. Governs R2, R3. + +- **KTD6. Build from source, and accept that it is not reproducible.** An earlier + draft of this decision claimed the repo carries a `package-lock.json` so + `npm ci` would be reproducible. **That was wrong.** The lockfile exists only in + a working tree: `.gitignore:23` excludes it under the comment "Lock files — + never commit". It is absent from `git archive HEAD`, so `COPY package-lock.json` + fails every git-based build. Verified by building from a git-only context. + + The build therefore runs `npm install` with no lockfile, which floats every + caret dependency. That is the cost of the repo's lockfile policy, not a choice + made here, and it applies equally to the registry install running today. State + it plainly rather than claiming a reproducibility this build does not have. + + Two further constraints found by building it rather than reasoning about it: + `node:22-slim` ships npm 10.9.x, whose arborist fails this tree without a + lockfile (`Cannot read properties of null (reading 'edgesOut')`), so the builder + upgrades to npm 11 first; and `npm pack` ships only the `files:` allowlist, so + the runtime install resolves dependencies against the registry and the repo's + `overrides` must be carried into the runtime manifest explicitly or the CVE pins + from `91c78e7` never reach the container. Governs R4. + +- **KTD7. Evaluate and count before the persist, never after.** In + `collectHealth`, the KV probe is raced against a 5-second timeout, but the + subsequent `await kv.set(KV.health, "latest", snapshot)` at `monitor.ts:94` is + **not raced**. During the real outage `state::set` was timing out at 180000 ms. + An escalation counter placed after that `await` would never increment during the + exact failure it targets, because the function would be parked on the persist + while the 30-second interval spawned more hung collections. The counter lives in + memory and is incremented before the persist. Governs R5. + +- **KTD8. The `overrides` block in the Dockerfile is vestigial.** Its comment + describes agentmemory resolving `iii-sdk` through a caret range. Both the + published 0.9.28 and the repo's 0.9.29 pin `iii-sdk` at exactly `0.11.2`, + verified in-container and in `package.json`. U1 should drop the workaround, and + must verify the resolved version after the build rather than assume it. + +### Assumptions + +- ~~The `[iii] Reconnecting` stream is background noise.~~ **RETRACTED + 2026-08-26. This was arithmetically false, and believing it produced the wrong + root cause above.** Port 49134 is the *engine's* port, so `ECONNREFUSED` there + means the engine is gone. iii-sdk caps one reconnect attempt at + `maxDelayMs 30000 x jitter 1.3` = **39 s**. Deployment `a201e22d` booted + 08-24T20:45:00Z and logged attempt **807** at 08-25T12:39:06Z: 57,200 s / 807 = + **70.9 s per attempt**, above the ceiling, so the stream cannot have begun at + boot. At the measured 30.6 s/attempt the onset is ~05:48Z, within 11 minutes of + the `ws` Sender errors. + + **The reconnect counter dates the outage.** Divide elapsed seconds by the + attempt number and compare against the 39 s ceiling. It is the earliest and + cheapest signal in the log buffer. +- `tini` exits non-zero when its child dies by signal, so Railway classifies the + exit as a failure and `ON_FAILURE` restarts. **U2 must verify this**, because + a clean exit 0 would not trigger a restart and the watchdog would be inert. +- A hang can leave the event loop running. Grounded: the application logged a + completed smart search at 06:08:19Z, three minutes after the `state::set` + timeout. This is what makes the in-process layer worth having. + +### Implementation Constraints + +- Stay on `fix/1223-heap-severity-denominator`. No branch switch, no worktree, + per `~/.claude/rules/no-branch-switching.md`. +- `deploy/railway/entrypoint.sh` and `deploy/railway/railway.json` are currently + modified and uncommitted, and those edits are what production runs. U0 commits + that state before anything stacks on it. +- The entrypoint is POSIX `sh`, not bash. `node:22-slim` provides `dash`. +- Never print `AGENTMEMORY_SECRET` or any `/data/.hmac` content. Handle by path + and by metadata, per `~/.claude/rules/triggr-credential-handling.md`. + +### Sequencing + +U0 gates everything. U5 gates the enablement of U2 and U4 per KTD5. U1 gates U3 +and U4 reaching production. U8 is independent and may run any time after U3/U4. + +**U1 and U2 both edit `deploy/railway/entrypoint.sh`, so they must be +serialized.** Run U1 first: it may relocate the install root, and U2's watchdog is +inserted relative to the final `exec` line that U1 may rewrite. Running them as +parallel branches means whichever is written second clobbers the first. + +``` +U0 (commit + backup) + | + +--> U1 (source build, edits entrypoint.sh) + | + +--> U2 (watchdog, edits entrypoint.sh, shipped DISABLED) --+ + | | + +--> U3 --> U4 -------------------------------------------- + + | | + | U5 --> U6 (enable) --> U7 (verify) + | + +--> U8 (upstream PR) +``` + +--- + +## Implementation Units + +### U0. Commit the deployed state and back up the volume + +**Goal.** Establish a known-good rollback point before anything changes. + +**Requirements.** R2. + +**Files.** `deploy/railway/entrypoint.sh`, `deploy/railway/railway.json`. + +**Approach.** Both files are modified and uncommitted, and both are what +production runs. Commit them as-is on the current branch so later changes are +separable. Then back up `/data` using the command in `deploy/railway/README.md`. +Record the current deployment id and image digest so the rollback target is +written down rather than remembered. + +**Test Scenarios.** +- `git status --short -- deploy/railway/` reports clean after the commit. +- The backup archive exists locally and is non-empty. +- The recorded digest matches `railway deployment list --json` for the live + deployment. + +**Verification.** `git log -1 --stat -- deploy/railway/` shows both files. The +archive's size is within an order of magnitude of the volume's 0.8 GB. + +--- + +### U1. Build the image from repo source + +**Goal.** Make fork-level source changes reach production. Today they cannot. + +**Requirements.** R4. Dissolves Q1. + +**Files.** `deploy/railway/Dockerfile`, `deploy/railway/entrypoint.sh`, +`.dockerignore` (new). Plus one **Railway service setting**, which is not a file. + +**`rootDirectory` is NOT in `railway.json`.** Verified: the file's keys are +`$schema`, `build{builder,dockerfilePath}`, `deploy{...}`, and neither +`serviceManifest` nor `fileServiceManifest` carries `rootDirectory`, yet the +deployment meta reports `rootDirectory: deploy/railway`. It is a dashboard/API +service setting. **Editing `railway.json` will not change the build context.** If +this unit only edits files, the context stays `deploy/railway/`, `src/` never +uploads, and U1 silently no-ops while appearing to succeed. Change the service +setting to the repo root and repoint `dockerfilePath` accordingly. + +**Approach.** The current Dockerfile has two `COPY` lines, neither touching +`src/`, and no build step. It installs the published package from npm, which is +why `/opt/agentmemory/src` does not exist in the image and the deployed version +is 0.9.28. Change the build context to the repo root, copy the manifest and +lockfile, run `npm ci`, copy the source, run `npm run build` (`tsdown`), and point +the `agentmemory` bin at the built `dist/cli.mjs`. Prefer a multi-stage build so +dev dependencies do not ship. Add `.dockerignore` to keep `node_modules`, +`.git`, `dist`, and `eval/` out of the upload. Drop the vestigial `overrides` +block per KTD8, then verify the resolved `iii-sdk` rather than assuming. + +Note that `dist/` is gitignored, so the image must build it and must never expect +a prebuilt copy. + +**The entrypoint is hard-coupled to the npm install layout, and this is the way +U1 fails at boot.** `entrypoint.sh` runs under `set -eu` and does: + +```sh +III_CONFIG="/opt/agentmemory/node_modules/@agentmemory/agentmemory/dist/iii-config.yaml" +cat > "$III_CONFIG" <<'EOF' +``` + +If U1 moves the install root, that `cat >` writes into a directory that no longer +exists, `set -e` aborts, and the container dies **before the app starts**. Railway +reads that as a failure and retries, so a botched U1 burns all ten retries on the +first deploy and lands in `CRASHED` — precisely the R2 outcome this plan exists to +prevent. `npm run build` is `tsdown && cp iii-config.yaml dist/`, so the config +always lands at `/dist/iii-config.yaml`. + +Choose one deliberately and write down which: **either** preserve the exact +`/opt/agentmemory/node_modules/@agentmemory/agentmemory/` prefix, **or** update +`III_CONFIG` to the new root. The same applies to the final +`exec gosu "$RUN_AS" agentmemory "$@"`, which resolves `agentmemory` on PATH +through the Dockerfile's symlink into `node_modules/.bin/`. U1 must recreate an +equivalent or that line fails too. + +**Test Scenarios.** +- The image builds from a clean context. +- `node -e "require('/opt/agentmemory/.../package.json').version"` inside the + built image reports 0.9.29, matching the repo. +- `grep -rl heapSizeLimit /dist/` inside the image returns at least + one file. This is AE4. Use a recursive grep: `dist/` is bundled and + `dist/health/thresholds.js` does not exist. +- **The container boots at all.** `III_CONFIG` resolves and `agentmemory` is on + PATH. Test this before any deploy, because failure here is a boot loop that + eats the retry budget. +- The resolved `iii-sdk` is 0.11.2. +- `/agentmemory/livez` returns 200 from the built image. + +**Verification.** Build locally first. Only then consider a deploy, which is +gated by U7 and by the destructive-commands confirmation. + +--- + +### U2. REMOVED — out-of-process watchdog + +**Deleted 2026-08-26**, superseded by the engine-exit handler (U9) and U5. See +KTD1 for the reasoning. Removed: the ~93-line watchdog block from +`deploy/railway/entrypoint.sh`, four `AGENTMEMORY_WATCHDOG*` variables, their +validation guards, and the tests that exercised them. `curl` **stays** in the +image: `deploy/railway/README.md` documents `railway ssh` plus +`curl http://localhost:3113` as the in-container viewer check, and that path was +used for the 2026-08-26 forensics that corrected the root cause. + +### U9. Exit when the engine dies + +**Goal.** Turn the actual failure into a process exit, at the place that already +detects it. + +**Requirements.** R1, R2. Replaces F1. + +**Files.** `src/cli.ts`. + +**Approach.** The engine is spawned detached and already has a `child.on("exit")` +handler that captures the exit code, the signal, and up to 16KB of stderr, then +logs under `vlog` and returns. Past a 60-second startup grace, report the death on +`console.error` with that stderr and exit non-zero. Deaths inside the grace keep +the existing path, which renders a better startup message. + +Default **on**, unlike the probe-based layers: this acts on a process-exit event, +so there is no threshold to tune and no false positive to trade against. The +engine owns the REST listener, the stream port, and the state store, so nothing +can be served once it is gone. Opt out with `AGENTMEMORY_EXIT_ON_ENGINE_DEATH=0`. + +**Test Scenarios.** Death inside the grace keeps the startup path. Death after it +exits non-zero. The opt-out suppresses the exit. The stderr reaches the log. + +**Verification.** `npm test`, plus a container run that kills the engine and +observes the exit code. + +### U2-original (superseded, kept for the record) + +**Goal.** Convert a hang into an exit so the existing restart policy can act. + +**Requirements.** R1, R2. Implements F1. + +**Files.** `deploy/railway/entrypoint.sh`. + +**Approach.** Insert a backgrounded POSIX `sh` loop immediately before the final +`exec gosu`. Capture `$$` first: `exec` preserves the PID, so the shell's PID is +the PID the application will hold. The loop waits for a first successful `livez` +before arming, so a slow BM25 startup backfill can never trigger it, and a boot +that never succeeds is left to Railway's deploy-time healthcheck. After arming, it +polls every 60 seconds; on `WATCHDOG_FAILS` consecutive failures it sends SIGTERM, +waits `WATCHDOG_GRACE` seconds, then sends SIGKILL only if the process survives. +It then exits, which enforces the once-per-boot cap in KTD5 structurally rather +than by a flag. + +**Ship it disabled.** Default `AGENTMEMORY_WATCHDOG=0` in this unit. U6 flips it +after U5 exists. + +The loop's logic was drafted and exercised during planning, and all five +behaviours passed: never arms against a server that was never up; does not fire +at 2 of 3; fires at exactly 3 of 3; escalates SIGTERM to SIGKILL when SIGTERM is +ignored; a short blip resets the counter without a kill. `ce-work` should re-run +these rather than trust the record. + +**Test Scenarios.** +- Never arms when the endpoint has never returned 200. +- Does not fire at `WATCHDOG_FAILS - 1`. +- Fires at exactly `WATCHDOG_FAILS`. +- Escalates to SIGKILL when the target ignores SIGTERM. +- A recovery shorter than the threshold resets the counter and no kill occurs. +- **`AGENTMEMORY_WATCHDOG=0` produces no watchdog process at all.** +- **`tini` exits non-zero when its child is signalled**, so `ON_FAILURE` fires. + This is the Planning Contract assumption that must be proven, not assumed. A + clean exit 0 would make the whole unit inert. + +**Verification.** `sh -n` for syntax, then the behavioural harness against a +throwaway HTTP server. Confirm the exit code with a container-local run. + +--- + +### U3. Evaluate `kvConnectivity` in the health thresholds + +**Goal.** Stop discarding the signal that identifies this exact failure. + +**Requirements.** R5. Implements part of F2. + +**Files.** `src/health/thresholds.ts`, `test/health-thresholds.test.ts` +(**existing, append only**). + +**`test/health-thresholds.test.ts` ALREADY EXISTS** — 174 lines, 9 tests, +carrying the regression coverage for issue #158 and for the #1223 denominator +work in commit `44e5372`. **Append to it. Never overwrite it.** It already +defines a `snap()` fixture helper whose default includes `kvConnectivity`, so +reuse that helper rather than introducing a second fixture. + +An earlier draft of this plan claimed no health test file existed. That claim +came from a `find` whose output was swallowed by the `rtk` wrapper, and acting on +it destroyed the file (recovered from `HEAD`). Read a negative search result +through `rtk proxy ` before trusting it. + +**Approach.** `collectHealth` populates `snapshot.kvConnectivity` with `status`, +`latencyMs`, and an optional `error`, using a 5-second race. `evaluateHealth` +checks `connectionState`, `eventLoopLagMs`, `cpu`, and `memory`, and never reads +`kvConnectivity`. Add it: `status === "error"` is critical, with an alert string +matching the existing `snake_case_value` convention used by +`connection_disconnected` and `event_loop_lag_critical_NNNms`. Follow the existing +alert vocabulary rather than inventing a new shape. + +**Test Scenarios.** +- A snapshot with `kvConnectivity.status === "error"` evaluates to `critical` and + carries a `kv_*` alert. +- A snapshot with `status === "ok"` does not add an alert. +- An absent or malformed `kvConnectivity` does not throw and does not evaluate to + critical. The field is optional in older persisted snapshots, so this is a real + compatibility path, not a hypothetical. +- The existing heap, CPU, event-loop, and connection assertions still pass. + +**Verification.** `npm test`, scoped to the health tests. + +--- + +### U4. Escalate sustained critical health to a process exit + +**Goal.** Let the in-process layer act on what it detects, catching a wedge while +the event loop still runs. + +**Requirements.** R1, R2, R5. Implements F2. + +**Files.** `src/health/monitor.ts`, `test/health-monitor.test.ts` (new file). +`test/health-thresholds.test.ts` **already exists** — see U3. An earlier draft +claimed no health test file existed; that claim was wrong and acting on it +destroyed the file once already. + +**Approach.** `registerHealthMonitor` currently ends `collectHealth` with a +persist and a return. Add an in-memory consecutive-critical counter. **Increment +and evaluate it before the `await kv.set(KV.health, "latest", snapshot)` call at +`monitor.ts:94`, never after** — that persist is not raced against any timeout, +and during the real outage `state::set` hung for 180 seconds, so a counter behind +it would never advance during the failure it exists to catch (KTD7). On reaching +the threshold, log a clear reason and exit so `ON_FAILURE` restarts. + +Reuse the existing SIGTERM `shutdown` path at `src/index.ts:629-630` rather than +calling `process.exit` directly, so the KV store flushes. Fall back to a hard +exit only if shutdown does not complete within a grace window. + +Gate the whole behaviour behind an environment flag, default **off**, matching +U2's disabled-by-default posture and KTD5. + +**Test Scenarios.** +- N consecutive critical snapshots trigger exactly one escalation. +- A single healthy snapshot between criticals resets the counter. +- With the flag off, no escalation occurs regardless of snapshot status. +- The counter advances when the persist is slow or rejects. This is the + regression test for KTD7 and the one most likely to be got wrong. +- Escalation runs the graceful shutdown path, not a bare `process.exit`. + +**Verification.** `npm test`. Assert on the escalation decision, not on a real +process exit. + +--- + +### U5. External uptime monitoring + +**Goal.** Make an outage and a restart loop reach a human without polling. + +**Requirements.** R3. Prerequisite for U6 per KTD5. + +**Files.** None in this repo. Configuration lives with the provider. Record the +setup in `deploy/railway/README.md`. + +**Approach.** Point an uptime check at +`https://agentmemory-production-0c12.up.railway.app/agentmemory/livez` on roughly +a 5-minute interval, alerting to a channel the user actually reads. The route +needs no auth, verified. This is the only layer that catches a restart loop +burning the retry budget, which is precisely the failure mode the other two +layers can create. + +**Test Scenarios.** +- The check reports healthy against the live service. +- A deliberate outage window produces an alert. +- The alert reaches the intended channel, not only the provider's dashboard. + +**Verification.** Confirm one real alert end to end. An untested alert path is +not a notification layer. + +--- + +### U6. Enable the watchdog and the escalation + +**Goal.** Turn on the killers, once notification exists. + +**Requirements.** R2, R3. + +**Files.** `deploy/railway/entrypoint.sh` default, or a Railway service variable. + +**Approach.** Flip `AGENTMEMORY_WATCHDOG` and U4's flag to on. Prefer Railway +service variables over a Dockerfile default so a rollback needs no rebuild. Do +not perform this unit until U5 has produced one verified alert. + +**Test Scenarios.** +- Both layers report armed in the deploy logs. +- A synthetic wedge in a throwaway environment produces exactly one restart. + +**Verification.** Read the deploy logs for both arming lines. + +--- + +### U7. Deploy and verify end to end + +**Goal.** Prove the whole chain on the live service. + +**Requirements.** All. + +**Files.** None. + +**Approach.** `railway up` from the repo root. **Confirm with the user first**, +per `~/.claude/rules/destructive-commands.md`: show the target project, +environment, service, and the rollback digest from U0. Watch the deployment to a +terminal state; never call it good on `railway status` alone, which reported +`Online` throughout the outage. + +**Test Scenarios.** +- AE4: `heapSizeLimit` is present in the deployed `dist/health/thresholds.js`. +- `/agentmemory/livez` returns 200 externally. +- The viewer at `http://127.0.0.1:8899/agentmemory/viewer` returns 200. +- `_proxy/status` shows a small `seconds_since_last_ok`. +- `memory_recall` through MCP returns real observations. +- Both watchdog layers log that they armed. + +**Verification.** Run every check above. Per +`~/.claude/rules/verification-no-dismiss.md`, trace any non-green item to ground +before declaring the deploy verified. + +--- + +### U8. Upstream PR to `rohitg00/agentmemory` + +**Goal.** Offer the fixes upstream instead of holding them only in the fork. + +**Requirements.** R7. + +**Files.** The U3 and U4 changes, plus the four existing health commits on this +branch. + +**Approach.** `origin` is `inix-x/agentmemory` and `upstream` is +`rohitg00/agentmemory`. Open a PR carrying the `kvConnectivity` evaluation and the +escalation. Reference issue #1223 for the heap work already committed here. +Confirm with the user before opening, per the destructive-commands rule: a PR is +outward-facing. + +**Test Scenarios.** Upstream CI passes. The PR description states the failure +mode, the evidence, and the mechanism. + +**Verification.** The PR URL is surfaced to the user. + +--- + +## Verification Contract + +Repo-specific commands: + +```bash +npm test # vitest, excludes test/integration.test.ts +npm run test:all # includes integration +npm run build # tsdown -> dist/ +sh -n deploy/railway/entrypoint.sh +``` + +Live-service checks: + +```bash +curl -sS -o /dev/null -w '%{http_code}\n' \ + https://agentmemory-production-0c12.up.railway.app/agentmemory/livez +curl -sS http://127.0.0.1:8899/_proxy/status +# Recursive: dist/ is BUNDLED, so dist/health/thresholds.js does not exist. +cd ~/rw-agentmemory && railway ssh --service agentmemory -- \ + "grep -rl heapSizeLimit /opt/agentmemory/node_modules/@agentmemory/agentmemory/dist/ | wc -l" +``` + +Quality gates: no unit is done while its own tests fail. No deploy is verified +while any check above is non-green or untraced. + +--- + +## Definition of Done + +**Global.** +- A hang recovers with no operator action, within roughly 5 minutes. +- A restart loop produces a notification rather than silence. +- The deployed image contains this branch's health fixes, proven by AE4. +- The upstream PR is open. +- No dead-end or experimental code remains in the diff. A long autonomous run + accumulates abandoned attempts; declaring done requires removing them. + +**Per unit.** Every Test Scenario passes, and every unit that changes runtime +behaviour leaves behind a check that fails if that behaviour breaks. + +**Explicitly not done** if: `railway status` alone is used as proof, the +notification path is configured but never fired once, or Q1 is resolved silently +in the event U1 is dropped. diff --git a/src/cli.ts b/src/cli.ts index 8bbdda932..6474a4294 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -54,6 +54,7 @@ import { type ConnectManifest, type RemoveOptions, } from "./cli/remove-plan.js"; +import { createEngineLogForwarder } from "./cli/engine-log.js"; import { renderSplash } from "./cli/splash.js"; import { isFirstRun, readPrefs, resetPrefs, writePrefs } from "./cli/preferences.js"; import { runOnboarding } from "./cli/onboarding.js"; @@ -917,6 +918,28 @@ type StartupFailure = { let startupFailure: StartupFailure | null = null; +// Off by default. The engine's own stdout is the only place a frame's +// `function_id` exists, but an unbounded log path here once wrote 137 GB +// (issue #519), so the forwarder is opt-in and hard-capped. +const ENGINE_LOG_ENABLED = + process.env["AGENTMEMORY_ENGINE_LOG"] === "1" || + process.env["AGENTMEMORY_ENGINE_LOG"] === "true"; + +function attachEngineLog( + stream: NodeJS.ReadableStream | null, + prefix: string, +): void { + if (!stream) return; + const forwarder = createEngineLogForwarder({ + prefix, + write: (line) => { + process.stderr.write(`${line}\n`); + }, + }); + stream.on("data", (chunk: Buffer) => forwarder.push(chunk)); + stream.on("end", () => forwarder.flush()); +} + // Spawn a background engine and collect any startup stderr for a short // window. The process is unref'd so the CLI parent can exit cleanly; we // only care about stderr that shows up BEFORE the health check succeeds, @@ -929,13 +952,18 @@ function spawnEngineBackground( vlog(`spawn: ${bin} ${spawnArgs.join(" ")}`); const child = spawn(bin, spawnArgs, { detached: true, - stdio: ["ignore", "ignore", "pipe"], + stdio: ["ignore", ENGINE_LOG_ENABLED ? "pipe" : "ignore", "pipe"], windowsHide: true, }); const isDocker = label.includes("Docker"); if (!isDocker && typeof child.pid === "number") { writeEnginePidfile(child.pid); } + if (ENGINE_LOG_ENABLED) { + attachEngineLog(child.stdout, "[engine]"); + attachEngineLog(child.stderr, "[engine:err]"); + } + const spawnedAt = Date.now(); const stderrChunks: Buffer[] = []; let stderrBytes = 0; const MAX_STDERR_CAPTURE = 16 * 1024; @@ -965,12 +993,44 @@ function spawnEngineBackground( } if (!isDocker) clearEnginePidfile(); clearEngineState(); + + // The engine owns the REST listener, the stream port and the state store. + // When it dies the surviving node process cannot serve anything, but it + // stays up reconnecting forever, so the platform sees a healthy container + // and no HTTP. That is the shape of every wedge observed so far. + // + // Report it and exit, so a supervisor can restart the whole container. + // This is a process-exit event, not a heuristic probe, so there is no + // false-positive to tune. Death during startup keeps the old path: the + // startup code below reads `startupFailure` and renders a better message. + const engineRanFor = Date.now() - spawnedAt; + if (engineRanFor > ENGINE_STARTUP_GRACE_MS) { + console.error( + `[agentmemory] engine exited after ${Math.round(engineRanFor / 1000)}s ` + + `(code=${code} signal=${signal}); nothing can be served without it, exiting`, + ); + if (stderr.trim()) console.error(`[agentmemory] engine stderr:\n${stderr}`); + if (process.env["AGENTMEMORY_EXIT_ON_ENGINE_DEATH"] !== "0") { + process.exit(1); + } + } } }); child.unref(); return child; } +// Engine deaths inside this window are treated as startup failures, which the +// startup path reports with a better message. Later deaths mean a running +// deployment lost its engine, which is fatal to serving. +// +// Keep this SHORT. It is a hole in the only mechanism that catches engine death: +// an engine that dies inside the window after startup already completed leaves +// the container up and serving nothing, which is the original bug. A failed +// spawn surfaces within a second or two, so a few seconds is all the startup +// path needs. +const ENGINE_STARTUP_GRACE_MS = 5_000; + function startIiiBin(iiiBin: string, configPath: string): boolean { const s = p.spinner(); s.start(`Starting iii-engine: ${iiiBin}`); diff --git a/src/cli/engine-log.ts b/src/cli/engine-log.ts new file mode 100644 index 000000000..2dab376e0 --- /dev/null +++ b/src/cli/engine-log.ts @@ -0,0 +1,112 @@ +import { StringDecoder } from "node:string_decoder"; + +export const ENGINE_LOG_MAX_TOTAL_BYTES = 32 * 1024 * 1024; +export const ENGINE_LOG_MAX_BYTES_PER_SECOND = 64 * 1024; +export const ENGINE_LOG_MAX_LINE_BYTES = 8 * 1024; +export const ENGINE_LOG_WINDOW_MS = 1000; + +export interface EngineLogForwarder { + push(chunk: Buffer): void; + flush(): void; +} + +export interface EngineLogForwarderOptions { + prefix: string; + write: (line: string) => void; + maxTotalBytes?: number; + maxBytesPerSecond?: number; + maxLineBytes?: number; + windowMs?: number; + now?: () => number; +} + +export function createEngineLogForwarder( + options: EngineLogForwarderOptions, +): EngineLogForwarder { + const { prefix, write } = options; + const maxTotalBytes = options.maxTotalBytes ?? ENGINE_LOG_MAX_TOTAL_BYTES; + const maxBytesPerSecond = + options.maxBytesPerSecond ?? ENGINE_LOG_MAX_BYTES_PER_SECOND; + const maxLineBytes = options.maxLineBytes ?? ENGINE_LOG_MAX_LINE_BYTES; + const windowMs = options.windowMs ?? ENGINE_LOG_WINDOW_MS; + const now = options.now ?? Date.now; + + const decoder = new StringDecoder("utf8"); + let pending = ""; + let totalBytes = 0; + let windowStart = now(); + let windowBytes = 0; + let droppedBytes = 0; + let stopped = false; + + function emitLine(line: string, exemptFromRateCap = false): void { + if (stopped) return; + const cost = Buffer.byteLength(line) + 1; + if (totalBytes + cost > maxTotalBytes) { + stopped = true; + write( + `${prefix} log forwarding stopped: ${maxTotalBytes} byte ceiling reached`, + ); + return; + } + if (!exemptFromRateCap && windowBytes + cost > maxBytesPerSecond) { + droppedBytes += cost; + return; + } + totalBytes += cost; + windowBytes += cost; + write(line); + } + + // The suppression notice is exempt from the rate cap so silent loss is + // impossible: it is one short line per window, still charged against + // both counters, so the lifetime ceiling still bounds it. + function reportDropped(): void { + if (droppedBytes === 0) return; + const dropped = droppedBytes; + droppedBytes = 0; + emitLine( + `${prefix} dropped ${dropped} bytes (rate cap ${maxBytesPerSecond} bytes/s)`, + true, + ); + } + + function rollWindow(): void { + if (now() - windowStart < windowMs) return; + windowStart = now(); + windowBytes = 0; + reportDropped(); + } + + function emitRecord(record: string): void { + emitLine(`${prefix} ${record.endsWith("\r") ? record.slice(0, -1) : record}`); + } + + return { + push(chunk: Buffer): void { + if (stopped) return; + rollWindow(); + pending += decoder.write(chunk); + let newline = pending.indexOf("\n"); + while (newline !== -1) { + emitRecord(pending.slice(0, newline)); + pending = pending.slice(newline + 1); + newline = pending.indexOf("\n"); + } + if (pending.length >= maxLineBytes) { + emitRecord(pending); + pending = ""; + } + }, + flush(): void { + if (stopped) return; + rollWindow(); + pending += decoder.end(); + if (pending.length > 0) { + emitRecord(pending); + pending = ""; + } + reportDropped(); + }, + }; +} diff --git a/src/health/monitor.ts b/src/health/monitor.ts index 953c94a0f..6985be3e4 100644 --- a/src/health/monitor.ts +++ b/src/health/monitor.ts @@ -1,13 +1,90 @@ +import { getHeapStatistics } from "node:v8"; import type { ISdk } from "iii-sdk"; import type { HealthSnapshot } from "../types.js"; import type { StateKV } from "../state/kv.js"; import { KV } from "../state/schema.js"; import { evaluateHealth } from "./thresholds.js"; +export interface EscalationState { + /** Consecutive snapshots whose KV probe failed. Reset by any healthy probe. */ + consecutiveStalls: number; + /** Set once a healthy KV probe has been seen. Nothing escalates before then. */ + armed: boolean; + escalated: boolean; +} + +/** + * Advance the stall counter and report whether the process should exit. + * + * Gated on the KV probe specifically, NOT on `snapshot.status`. `evaluateHealth` + * raises `critical` from five independent conditions (connection, KV, event-loop + * lag, CPU, memory), so gating on the aggregate would let a CPU spike during a + * consolidation pass kill a process that is not wedged at all. + * + * `armed` mirrors the shell watchdog's "wait for a first success" rule. Without + * it, a store that is already stalled at boot escalates on the first minute of + * every container life, which spends the platform's restart budget in minutes + * and ends at a stopped deployment. + * + * Call this BEFORE persisting the snapshot. The persist is not raced against any + * timeout, and a stalled store parks it for the engine's full invocation timeout, + * so a counter behind it would never advance during the failure it exists to + * catch. + */ +export function bumpEscalation( + snapshot: HealthSnapshot, + state: EscalationState, + threshold: number, +): boolean { + const stalled = snapshot.kvConnectivity?.status === "error"; + if (!stalled) { + state.armed = true; + state.consecutiveStalls = 0; + return false; + } + if (!state.armed) return false; + state.consecutiveStalls += 1; + if (state.escalated || state.consecutiveStalls < threshold) return false; + state.escalated = true; + return true; +} + export function registerHealthMonitor( sdk: ISdk, kv: StateKV, ): { stop: () => void } { + const escalationState: EscalationState = { + consecutiveStalls: 0, + armed: false, + escalated: false, + }; + // Default off. Enabling this arms an automatic process-killer, so it stays + // opt-in until external uptime monitoring exists to make a restart loop + // visible. The threshold spans 5 minutes at the 30s collection interval, + // deliberately wider than the write bursts this codebase already documents + // as able to exceed the engine's own 30s timeout (see src/index.ts). + const escalateEnabled = process.env["AGENTMEMORY_HEALTH_ESCALATE"] === "1" || + process.env["AGENTMEMORY_HEALTH_ESCALATE"]?.toLowerCase() === "true"; + const escalateAfter = 10; + + function escalate(alerts: string[]): void { + console.error( + `[agentmemory] health escalation: KV unreachable for ${escalateAfter} consecutive checks (${alerts.join(", ")}); exiting so the platform restarts`, + ); + // Prefer the registered SIGTERM shutdown so the index flushes. That handler + // ends in process.exit(0), so railway.json uses restartPolicyType ALWAYS — + // ON_FAILURE would read a graceful exit as success and never restart. + process.kill(process.pid, "SIGTERM"); + // NOT unref'd. If shutdown parks on the same stalled store, this is the only + // thing that still reaches an exit. + setTimeout(() => { + console.error( + "[agentmemory] health escalation: graceful shutdown did not finish; forcing exit", + ); + process.exit(1); + }, 15_000); + } + let connectionState = "connected"; let prevCpuUsage = process.cpuUsage(); let prevCpuTime = Date.now(); @@ -37,11 +114,21 @@ export function registerHealthMonitor( const eventLoopLagMs = performance.now() - startMark; let workers: HealthSnapshot["workers"] = []; + // Raced, like the KV probe below. Unraced, this inherits the engine's + // invocation timeout (180s), so a dead engine parks the whole collection + // here -- upstream of the probe and of the escalation decision -- and + // detection slips from minutes to tens of minutes. + const WORKERS_PROBE_TIMEOUT = 5000; try { - const result = await sdk.trigger< - unknown, - { workers?: HealthSnapshot["workers"] } - >({ function_id: "engine::workers::list", payload: {} }); + const result = await Promise.race([ + sdk.trigger({ + function_id: "engine::workers::list", + payload: {}, + }), + new Promise((_, reject) => + setTimeout(() => reject(new Error("timeout")), WORKERS_PROBE_TIMEOUT), + ), + ]); if (result?.workers) workers = result.workers; } catch {} @@ -71,6 +158,7 @@ export function registerHealthMonitor( heapTotal: mem.heapTotal, rss: mem.rss, external: mem.external, + heapSizeLimit: getHeapStatistics().heap_size_limit, }, cpu: { userMicros: currentCpu.user, @@ -89,6 +177,11 @@ export function registerHealthMonitor( snapshot.alerts = evaluated.alerts; snapshot.notes = evaluated.notes; + // Decide before the persist below, which races no timeout. + if (escalateEnabled && bumpEscalation(snapshot, escalationState, escalateAfter)) { + escalate(snapshot.alerts); + } + await kv.set(KV.health, "latest", snapshot).catch(() => {}); return snapshot; } diff --git a/src/health/thresholds.ts b/src/health/thresholds.ts index 7279afad1..cf5bd1108 100644 --- a/src/health/thresholds.ts +++ b/src/health/thresholds.ts @@ -30,6 +30,12 @@ export function evaluateHealth( let critical = false; let degraded = false; + // NOTE: unreachable in production today. iii-sdk's setConnectionState only + // assigns a private field and emits nothing, so the "connection_state" + // listener in monitor.ts never fires and connectionState stays "connected" + // for the life of the process. Kept because the field is part of the + // snapshot contract and a future SDK may emit it; do not rely on it as a + // liveness signal until it does. if ( snapshot.connectionState === "disconnected" || snapshot.connectionState === "failed" @@ -41,6 +47,18 @@ export function evaluateHealth( degraded = true; } + // The KV probe in collectHealth is the only check that exercises the state + // store end to end (set then get, raced against a 5s timeout). A store that + // stops answering takes the HTTP workers down with it, so a failed probe is + // the earliest reliable signal of that failure and belongs at critical. + // kvConnectivity is optional on the snapshot, and older persisted snapshots + // predate it, so an absent or malformed value must read as "no signal" + // rather than as a failure. + if (snapshot.kvConnectivity?.status === "error") { + alerts.push("kv_probe_failed"); + critical = true; + } + if (snapshot.eventLoopLagMs > cfg.eventLoopLagCriticalMs) { alerts.push( `event_loop_lag_critical_${Math.round(snapshot.eventLoopLagMs)}ms`, @@ -59,10 +77,15 @@ export function evaluateHealth( degraded = true; } + // heapTotal is what V8 has committed so far, not what it may grow to, and V8 + // sizes it to demand — so a healthy busy process sits near 100% of it + // permanently. Measure against heap_size_limit when the snapshot carries it. + const heapCeiling = + snapshot.memory.heapSizeLimit && snapshot.memory.heapSizeLimit > 0 + ? snapshot.memory.heapSizeLimit + : snapshot.memory.heapTotal; const memPercent = - snapshot.memory.heapTotal > 0 - ? (snapshot.memory.heapUsed / snapshot.memory.heapTotal) * 100 - : 0; + heapCeiling > 0 ? (snapshot.memory.heapUsed / heapCeiling) * 100 : 0; const rss = snapshot.memory.rss ?? 0; const rssAboveFloor = rss >= cfg.memoryRssFloorBytes; const memMb = Math.round(rss / (1024 * 1024)); diff --git a/src/mcp/server.ts b/src/mcp/server.ts index ef26427aa..88142ea5d 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -5,8 +5,7 @@ import type { SessionSummary, Memory, Session, - GraphNode, - GraphEdge, + GraphSnapshot, } from "../types.js"; import { getVisibleTools } from "./tools-registry.js"; import { timingSafeCompare } from "../auth.js"; @@ -1479,14 +1478,11 @@ export function registerMcpEndpoints( if (uri === "agentmemory://graph/stats") { try { - const nodes = await kv.list(KV.graphNodes); - const edges = await kv.list(KV.graphEdges); - const nodesByType: Record = {}; - for (const n of nodes) - nodesByType[n.type] = (nodesByType[n.type] || 0) + 1; - const edgesByType: Record = {}; - for (const e of edges) - edgesByType[e.type] = (edgesByType[e.type] || 0) + 1; + const snapshot = await kv.get( + KV.graphSnapshot, + "current", + ); + const stats = snapshot?.stats; return { status_code: 200, body: { @@ -1495,10 +1491,11 @@ export function registerMcpEndpoints( uri, mimeType: "application/json", text: JSON.stringify({ - totalNodes: nodes.length, - totalEdges: edges.length, - nodesByType, - edgesByType, + totalNodes: stats?.totalNodes ?? 0, + totalEdges: stats?.totalEdges ?? 0, + nodesByType: stats?.nodesByType ?? {}, + edgesByType: stats?.edgesByType ?? {}, + ...(snapshot ? {} : { pending: true }), }), }, ], diff --git a/src/providers/resilient.ts b/src/providers/resilient.ts index 95ece40c9..dd6c4dc22 100644 --- a/src/providers/resilient.ts +++ b/src/providers/resilient.ts @@ -1,25 +1,88 @@ import type { MemoryProvider, CircuitBreakerState } from "../types.js"; import { CircuitBreaker } from "./circuit-breaker.js"; +import { getEnvVar } from "../config.js"; + +const DEFAULT_MAX_CONCURRENT = 4; + +/** + * Bounds how many calls are inside the provider at once. + * + * release() hands its slot directly to the next waiter rather than + * decrementing and letting waiters race for the opening, so the limit holds + * under a burst. + */ +class Semaphore { + private active = 0; + private waiting: Array<() => void> = []; + + constructor(private readonly limit: number) {} + + async acquire(): Promise { + if (this.active < this.limit) { + this.active++; + return; + } + await new Promise((resolve) => this.waiting.push(resolve)); + } + + release(): void { + const next = this.waiting.shift(); + if (next) next(); + else this.active--; + } +} + +// 429 is backpressure, not a fault. Counting it opened the breaker on a healthy +// provider and failed every compression for the recovery window. +function isRateLimited(err: unknown): boolean { + const message = err instanceof Error ? err.message : String(err); + return /\b429\b|rate.?limit|too many (concurrent )?requests/i.test(message); +} + +// Option is the test seam, env is the operator knob on Railway, constant is the +// default. getEnvVar rather than process.env so ~/.agentmemory/.env is honoured, +// which is how every sibling provider reads config. +function resolveMaxConcurrent(configured: number | undefined): number { + const candidate = + configured ?? + Number(getEnvVar("AGENTMEMORY_MAX_PROVIDER_CONCURRENCY") ?? NaN); + if (!Number.isFinite(candidate)) return DEFAULT_MAX_CONCURRENT; + const whole = Math.floor(candidate); + return whole >= 1 ? whole : DEFAULT_MAX_CONCURRENT; +} export class ResilientProvider implements MemoryProvider { - private breaker = new CircuitBreaker(); + private breaker: CircuitBreaker; + private gate: Semaphore; name: string; - constructor(private inner: MemoryProvider) { + constructor( + private inner: MemoryProvider, + options: { maxConcurrent?: number } = {}, + ) { + this.breaker = new CircuitBreaker(); + this.gate = new Semaphore(resolveMaxConcurrent(options.maxConcurrent)); this.name = `resilient(${inner.name})`; } private async call(fn: () => Promise): Promise { + // Checked before queueing. An open breaker should fail fast rather than + // occupy a slot that a call with a chance of succeeding could use. if (!this.breaker.isAllowed) { throw new Error("circuit_breaker_open"); } + await this.gate.acquire(); try { const result = await fn(); this.breaker.recordSuccess(); return result; } catch (err) { - this.breaker.recordFailure(); + if (!isRateLimited(err)) this.breaker.recordFailure(); throw err; + } finally { + // In `finally` so a throw cannot leak the slot and deadlock every call + // queued behind it. + this.gate.release(); } } diff --git a/src/state/index-persistence.ts b/src/state/index-persistence.ts index 6df0e2fda..bfe97a72c 100644 --- a/src/state/index-persistence.ts +++ b/src/state/index-persistence.ts @@ -24,6 +24,27 @@ type IndexShardManifest = { chars: number; }; +// Suffix for the reclaim ledger that sits beside each manifest. +const GC_LEDGER_SUFFIX = ":gc"; + +// Stands in for a manifest written before generations were recorded, whose +// `generation` field is absent. Cannot collide with createIndexGeneration(). +const PRE_LEDGER_GENERATION = "pre-ledger"; + +// Every generation whose shards may still be on disk, live one included. The +// manifest alone cannot answer that: it names only the generation that is +// current, so a generation stranded by a failed read, a throw after commit, or +// a kill mid-cleanup becomes unreachable the moment the next manifest replaces +// it. Nothing else enumerates shard scopes — StateKV lists keys within a scope, +// not scopes by prefix — so what is not recorded here can never be found again. +type IndexGcLedger = { + v: 1; + generations: Array<{ + generation: string; + shards: Array<{ scope: string; key: string }>; + }>; +}; + type IndexPersistenceOptions = { shardChars?: number; createGeneration?: () => string; @@ -67,7 +88,8 @@ function isValidShardDescriptor( export class IndexPersistence { private timer: ReturnType | null = null; - private lastFailureLogAt = 0; + private lastFailureLogAt = new Map(); + private queue: Promise = Promise.resolve(); constructor( private kv: StateKV, @@ -83,7 +105,7 @@ export class IndexPersistence { // under sustained iii-engine write timeouts (issue #204). Funnel // rejections through logFailure() instead. this.timer = setTimeout(() => { - this.save().catch((err) => this.logFailure(err)); + this.save().catch((err) => this.logFailure("index", err)); }, DEBOUNCE_MS); } @@ -92,13 +114,52 @@ export class IndexPersistence { clearTimeout(this.timer); this.timer = null; } + return this.enqueue(() => this.runSave()); + } + + /** + * Serialise everything that read-modify-writes the gc ledger. + * + * Two saves genuinely overlap here: scheduleSave fires save() unawaited from + * a timer, flushIndexSave awaits save() on every delete path + * (src/functions/search.ts), and stop() clears the timer without awaiting a + * save already running. Overlapping saves drop each other's ledger entries, + * and worse, a save that publishes second reclaims the shards the first is + * still writing — leaving the first to publish a manifest naming data that + * is already gone, which fails the next load closed. + * + * Queue, never coalesce. flushIndexSave awaits this to make a delete + * durable, so handing back an in-flight promise that started before the + * delete would report success for a snapshot that does not contain it. The + * cost is real: a delete-path flush waits out the save ahead of it. + */ + private enqueue(work: () => Promise): Promise { + // runSave never rejects and the tail below always fulfils, so the queue + // cannot be left rejected and needs no rejection handler here. + const run = this.queue.then(work); + this.queue = run.then( + () => undefined, + () => undefined, + ); + return run; + } + + private async runSave(): Promise { + // Each index fails on its own. One try around both would let a BM25 + // failure stop the vector index persisting at all, and a lost vector index + // is never rebuilt: both rebuild triggers key on the BM25 size + // (src/index.ts, src/functions/search.ts). try { await this.saveBm25Index(this.bm25.serialize()); - if (this.vector) { + } catch (err) { + this.logFailure("BM25", err); + } + if (this.vector) { + try { await this.saveVectorIndex(this.vector.serialize()); + } catch (err) { + this.logFailure("vector", err); } - } catch (err) { - this.logFailure(err); } } @@ -129,16 +190,19 @@ export class IndexPersistence { } } - private logFailure(err: unknown): void { + private logFailure(index: string, err: unknown): void { const now = Date.now(); // Throttle: persistence failures under load arrive in bursts // (iii-engine queue pressure). Logging every debounce flush adds - // noise without information. - if (now - this.lastFailureLogAt < FAILURE_LOG_THROTTLE_MS) return; - this.lastFailureLogAt = now; + // noise without information. Throttled PER INDEX, so a vector failure + // right after a BM25 one is not swallowed — the two fail independently + // now, and at 3am you need to know which one stopped persisting. + const lastAt = this.lastFailureLogAt.get(index) ?? 0; + if (now - lastAt < FAILURE_LOG_THROTTLE_MS) return; + this.lastFailureLogAt.set(index, now); const code = (err as { code?: string })?.code; const message = err instanceof Error ? err.message : String(err); - logger.warn("index persistence: failed to save BM25/vector index", { + logger.warn(`index persistence: failed to save ${index} index`, { code, message, hint: @@ -192,6 +256,16 @@ export class IndexPersistence { chunks.push(chunk); } + // Record the generation BEFORE the first shard write. A kill anywhere from + // here to the manifest publish would otherwise leave shards on disk that + // nothing references and nothing can enumerate. + const tracked = await this.trackGeneration( + manifestKey, + generation, + shards, + previous, + ); + const writeResults = await Promise.allSettled( shards.map(async (shard, index) => { const chunk = chunks[index] ?? ""; @@ -211,7 +285,14 @@ export class IndexPersistence { (result): result is PromiseRejectedResult => result.status === "rejected", ); if (failedWrite) { - await this.deleteShards(shards, "shard_write_rollback"); + const allGone = await this.deleteShards(shards, "shard_write_rollback"); + // Drop the entry only if every shard actually went. A shard that survived + // its delete is unnameable once its entry is gone, which is the contract + // this file states for reclaim: a delete failure costs a retry, never a + // stranded generation. + if (tracked && allGone) { + await this.untrackGeneration(manifestKey, generation); + } throw failedWrite.reason; } @@ -249,23 +330,270 @@ export class IndexPersistence { error: errorMessage(err), }); } else { - await this.deleteShards(shards, "manifest_publish_rollback"); + const allGone = await this.deleteShards( + shards, + "manifest_publish_rollback", + ); + if (tracked && allGone) { + await this.untrackGeneration(manifestKey, generation); + } } throw err; } await this.deleteKey(KV.bm25Index, legacyKey, "legacy_cleanup"); - if (previous?.v === 1 && Array.isArray(previous.shards)) { - const currentShardIds = new Set( - shards.map((shard) => `${shard.scope}\0${shard.key}`), + // Only reclaim when this generation is actually in the ledger. Reclaiming + // against a ledger that does not list us would treat live shards as dead. + if (tracked) { + await this.reclaimGenerations(manifestKey, generation); + } else if (previous?.v === 1 && Array.isArray(previous.shards)) { + // The ledger was unusable this cycle, so nothing above will ever revisit + // `previous`. Fall back to the pre-ledger cleanup: the manifest just + // published supersedes it, and saves are serialised, so its shards are + // dead. Without this, a run of unusable cycles orphans one generation + // each — strictly worse than the code this replaced, which always had + // this path. + const liveIds = new Set( + shards.map((shard) => `${shard.scope}\u0000${shard.key}`), + ); + await this.deleteShards( + previous.shards.filter( + (shard) => + isValidShardDescriptor(shard) && + !liveIds.has(`${shard.scope}\u0000${shard.key}`), + ), + "previous_generation_cleanup", + ); + } + } + + /** Drop a generation's entry after its shards have been rolled back. */ + private async untrackGeneration( + manifestKey: string, + generation: string, + ): Promise { + try { + const ledger = await this.readLedger(manifestKey); + const generations = ledger.generations.filter( + (entry) => entry?.generation !== generation, ); - for (const shard of previous.shards) { - if (currentShardIds.has(`${shard.scope}\0${shard.key}`)) continue; - await this.deleteShards([shard], "previous_generation_cleanup"); + if (generations.length === ledger.generations.length) return; + await this.kv.set(KV.bm25Index, this.gcKey(manifestKey), { + v: 1, + generations, + }); + } catch { + // Best effort. A surviving entry costs a retry on the next reclaim, and + // this runs while a save is already failing — never make that worse. + } + } + + private gcKey(manifestKey: string): string { + return `${manifestKey}${GC_LEDGER_SUFFIX}`; + } + + private async readLedger(manifestKey: string): Promise { + // Throws rather than returning a blank ledger, because a blank one would + // be written straight back over whatever is stored. Callers catch it and + // skip tracking for that cycle; they must never treat it as "no ledger". + const stored = await this.kv.get( + KV.bm25Index, + this.gcKey(manifestKey), + ); + if (stored == null) return { v: 1, generations: [] }; + // Present but unrecognised: a newer version, or a rollback to this build + // after one that wrote a different shape. Overwriting drops every + // generation it tracked, so leave it exactly where it is. + if (stored.v !== 1 || !Array.isArray(stored.generations)) { + throw new Error( + `index gc ledger ${this.gcKey(manifestKey)} has an unrecognised shape ` + + `(v=${String((stored as { v?: unknown }).v)}); refusing to overwrite it`, + ); + } + return stored; + } + + private async trackGeneration( + manifestKey: string, + generation: string, + shards: IndexShardManifest["shards"], + previous: IndexShardManifest | null, + ): Promise { + try { + // The WHOLE body runs under this guard, not just the ledger read. A + // malformed ledger entry or manifest shard would otherwise throw a + // TypeError out of saveShardedIndex before any shard write, leaving one + // throttled log line per 60s as the only trace while BM25 stopped + // persisting for good. isValidShardDescriptor exists in this file + // because a per-shard-malformed manifest is already its threat model. + return await this.recordGeneration( + manifestKey, + generation, + shards, + previous, + ); + } catch (err) { + // A state::get brownout is the exact condition this bug appears under, + // so this path is not rare. Aborting the save here would stop persisting + // the index at all, which is worse than the leak being fixed. Writing a + // fresh ledger would drop every generation the stored one lists. Do + // neither: let the shards and manifest land, skip this cycle's tracking + // and its reclaim, and leak at most one generation instead of one per + // brownout. + // + // That leak can OUTLIVE the unreadable window. This generation publishes + // untracked; the next save re-seeds it from `previous`, but that read is + // itself `.catch(() => null)`, so if it also fails the generation is in + // no ledger and reclaim only ever looks at what precedes the live entry. + // With no scope enumeration in StateKV, nothing can find it again. + // Throttled for the same reason every other failure log here is: an + // unusable ledger stays unusable, so this fires on every debounce. + const throttleKey = `gc:${manifestKey}`; + const now = Date.now(); + if (now - (this.lastFailureLogAt.get(throttleKey) ?? 0) >= FAILURE_LOG_THROTTLE_MS) { + this.lastFailureLogAt.set(throttleKey, now); + logger.warn( + "index persistence: gc ledger unavailable, skipping reclaim", + { manifestKey, message: errorMessage(err) }, + ); } + return false; } } + private async recordGeneration( + manifestKey: string, + generation: string, + shards: IndexShardManifest["shards"], + previous: IndexShardManifest | null, + ): Promise { + const ledger = await this.readLedger(manifestKey); + // Optional-chained so one malformed entry costs that entry, not the whole + // cycle's tracking. reclaimGenerations tolerates them the same way. + const known = new Set(ledger.generations.map((entry) => entry?.generation)); + + // Seed the generation the pre-ledger code left live, so upgrading does not + // strand it. Only reachable while `previous` is readable, which is exactly + // the case the old cleanup already handled. + // + // `generation` is optional on the manifest and older stores really do omit + // it, so fall back to a sentinel rather than skipping the seed. The + // sentinel cannot collide with createIndexGeneration()'s `idx_` ids, and + // since every manifest written from here on carries a generation, the + // seeded entry always ends up preceding a live one and gets reclaimed. + if (previous?.v === 1 && Array.isArray(previous.shards)) { + const previousGeneration = previous.generation ?? PRE_LEDGER_GENERATION; + if (!known.has(previousGeneration)) { + ledger.generations.push({ + generation: previousGeneration, + shards: previous.shards + .filter(isValidShardDescriptor) + .map(({ scope, key }) => ({ scope, key })), + }); + known.add(previousGeneration); + } + } + + if (!known.has(generation)) { + ledger.generations.push({ + generation, + shards: shards.map(({ scope, key }) => ({ scope, key })), + }); + } + + await this.kv.set( + KV.bm25Index, + this.gcKey(manifestKey), + ledger, + ); + return true; + } + + /** + * Delete every tracked generation except the live one, then rewrite the + * ledger with whatever survived. A shard whose delete failed stays listed and + * is retried on the next save or load, so a delete failure costs a retry + * instead of stranding the generation for good. + */ + private async reclaimGenerations( + manifestKey: string, + liveGeneration: string | undefined, + ): Promise { + if (!liveGeneration) return; + const ledger = await this.readLedger(manifestKey).catch(() => null); + if (!ledger) return; + + // Reclaim strictly what precedes the live generation in the ledger, which + // trackGeneration appends to in creation order. Anything at or after the + // live entry is either live or a save still in flight: setIndexPersistence + // runs before load() in src/index.ts, so a request arriving during boot can + // have a save writing shards while this reclaim runs, and deleting those + // would publish a manifest whose shards are already half gone. A generation + // stranded after the live one is not lost, only deferred — it becomes + // reclaimable as soon as a newer generation is published. + const liveIndex = ledger.generations.findIndex( + (entry) => entry?.generation === liveGeneration, + ); + // Live generation untracked (a pre-ledger store, or a manifest written + // before this shipped). Nothing can be classified as superseded, so leave + // every entry alone rather than guess. + if (liveIndex < 0) return; + + const reclaimedPaths: string[] = []; + let failed = 0; + const survivors: IndexGcLedger["generations"] = []; + for (const [index, entry] of ledger.generations.entries()) { + // A malformed entry is kept, never iterated. This also runs on the load + // path, where a throw would null the loaded index and trigger the + // full-corpus rebuild. A GC step must not be able to fail a load. + if (index >= liveIndex || !entry || !Array.isArray(entry.shards)) { + survivors.push(entry); + continue; + } + const stranded: IndexGcLedger["generations"][number]["shards"] = []; + for (const shard of entry.shards) { + try { + await this.kv.delete(shard.scope, shard.key); + reclaimedPaths.push(statePath(shard.scope, shard.key)); + } catch { + failed += 1; + stranded.push(shard); + } + } + if (stranded.length > 0) { + survivors.push({ generation: entry.generation, shards: stranded }); + } + } + + // One audit row for the sweep, not one per shard. src/functions/audit.ts + // sets the policy: automatic bulk sweeps emit a single row listing every + // removed id, because per-item rows flood the log. A reclaim on a badly + // leaked store is well over a thousand shards, and this runs during boot. + if (reclaimedPaths.length > 0) { + await this.auditIndexPersistence("delete", reclaimedPaths, { + manifestKey, + reason: "generation_reclaim", + liveGeneration, + // `evicted` is the field name src/functions/audit.ts specifies for a + // sweep; retention.ts is the reference shape. `failed` surfaces deletes + // that will be retried, which would otherwise vanish silently. + evicted: reclaimedPaths.length, + failed, + }); + } + + // A successful delete is the only thing that shrinks the ledger, so with + // none there is nothing to rewrite. Without this, every boot writes the + // ledger back unchanged. + if (reclaimedPaths.length === 0) return; + await this.kv + .set(KV.bm25Index, this.gcKey(manifestKey), { + v: 1, + generations: survivors, + }) + .catch(() => undefined); + } + private async auditIndexPersistence( action: string, targetIds: string[], @@ -280,35 +608,41 @@ export class IndexPersistence { ); } + /** Reports whether the delete landed, so the reclaim path can retry the rest. */ private async deleteKey( scope: string, key: string, reason: string, - ): Promise { - let result = "deleted"; + ): Promise { + let ok = true; let error: string | undefined; try { await this.kv.delete(scope, key); } catch (err) { - result = "failed"; + ok = false; error = errorMessage(err); } await this.auditIndexPersistence("delete", [statePath(scope, key)], { scope, key, reason, - result, + result: ok ? "deleted" : "failed", error, }); + return ok; } private async deleteShards( shards: IndexShardManifest["shards"], reason: string, - ): Promise { + ): Promise { + let allGone = true; for (const shard of shards) { - await this.deleteKey(shard.scope, shard.key, reason); + if (!(await this.deleteKey(shard.scope, shard.key, reason))) { + allGone = false; + } } + return allGone; } private async isManifestPublished( @@ -368,7 +702,20 @@ export class IndexPersistence { manifest.value != null && typeof manifest.value === "object" ) { - return this.loadManifestData(manifest.value, label); + const data = await this.loadManifestData(manifest.value, label); + // Boot is the only point that sees a generation stranded by a kill: the + // save path only ever inspects its own predecessor. Reclaim once the live + // generation has actually loaded — a failed load must not authorise + // deleting anything. + if (data !== null) { + // Through the same queue as save(), or this sweep's ledger rewrite + // clobbers a concurrent save's entry. Never allowed to fail the load. + const live = manifest.value.generation; + await this + .enqueue(() => this.reclaimGenerations(manifestKey, live)) + .catch(() => undefined); + } + return data; } const legacy = await this.readIndexValue( diff --git a/src/types.ts b/src/types.ts index 1118b3f99..d8543c13a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -226,6 +226,7 @@ export interface HealthSnapshot { heapTotal: number; rss: number; external: number; + heapSizeLimit?: number; }; cpu: { userMicros: number; systemMicros: number; percent: number }; eventLoopLagMs: number; diff --git a/src/viewer/index.html b/src/viewer/index.html index de47f4f5f..c532a76e9 100644 --- a/src/viewer/index.html +++ b/src/viewer/index.html @@ -1210,7 +1210,7 @@

agentmemory

return 'CPU ' + (m[1] === 'critical' ? 'critically high' : 'elevated') + ': ' + m[2] + '%.'; if ((m = /^event_loop_lag_(warn|critical)_(\d+)ms$/.exec(f))) return 'Event loop ' + (m[1] === 'critical' ? 'severely delayed' : 'delayed') + ': ' + m[2] + ' ms behind. The worker is busy or blocked.'; - if (f === 'connection_reconnecting') + if (f === 'kv_probe_failed') return 'State store unreachable: the health probe could not read back what it wrote. Search and recall are likely stalled.'; if (f === 'connection_reconnecting') return 'Engine connection lost — reconnecting.'; if ((m = /^connection_(.+)$/.exec(f))) return 'Engine connection state: ' + m[1] + '.'; @@ -1579,10 +1579,16 @@

agentmemory

var heapUsed = Math.round((snap.memory.heapUsed || 0) / 1024 / 1024); var heapTotal = Math.round((snap.memory.heapTotal || 0) / 1024 / 1024); var rss = Math.round((snap.memory.rss || 0) / 1024 / 1024); - var heapPct = heapTotal > 0 ? Math.round((heapUsed / heapTotal) * 100) : 0; + var limitBytes = snap.memory.heapSizeLimit || 0; + var ceilingBytes = limitBytes > 0 ? limitBytes : (snap.memory.heapTotal || 0); + var heapCeiling = Math.round(ceilingBytes / 1024 / 1024); + // Percentage off raw bytes so the gauge agrees with evaluateHealth, + // which never sees the MB rounding used for the label. + var heapPercent = ceilingBytes > 0 ? ((snap.memory.heapUsed || 0) / ceilingBytes) * 100 : 0; + var heapPct = Math.round(heapPercent); var rssAboveFloor = rss >= 512; - var heapColor = (heapPct > 80 && rssAboveFloor) ? 'var(--red)' : (heapPct > 60 && rssAboveFloor) ? 'var(--yellow)' : 'var(--green)'; - html += '
Heap
' + heapUsed + ' / ' + heapTotal + ' MB
'; + var heapColor = (heapPercent > 80 && rssAboveFloor) ? 'var(--red)' : (heapPercent > 60 && rssAboveFloor) ? 'var(--yellow)' : 'var(--green)'; + html += '
Heap
' + heapUsed + ' / ' + heapCeiling + ' MB
'; html += '
RSS
' + rss + ' MB
'; if (snap.memory.external) { var ext = Math.round(snap.memory.external / 1024 / 1024); diff --git a/test/cli-engine-startup.test.ts b/test/cli-engine-startup.test.ts new file mode 100644 index 000000000..82199a623 --- /dev/null +++ b/test/cli-engine-startup.test.ts @@ -0,0 +1,69 @@ +import { readFileSync } from "node:fs"; +import { describe, expect, it } from "vitest"; +import { ENGINE_LOG_MAX_BYTES_PER_SECOND } from "../src/cli/engine-log.js"; + +describe("engine spawn stdio and death detection", () => { + const source = readFileSync("src/cli.ts", "utf8"); + const spawnStart = source.indexOf("function spawnEngineBackground"); + const spawnEnd = source.indexOf("const ENGINE_STARTUP_GRACE_MS", spawnStart); + const spawnBody = source.slice(spawnStart, spawnEnd); + + it("forwards engine stdout only behind the opt-in gate", () => { + expect(spawnBody).toContain( + 'stdio: ["ignore", ENGINE_LOG_ENABLED ? "pipe" : "ignore", "pipe"]', + ); + expect(spawnBody).not.toContain('stdio: ["ignore", "ignore", "pipe"]'); + expect(spawnBody).toContain( + 'if (ENGINE_LOG_ENABLED) {\n attachEngineLog(child.stdout, "[engine]");\n attachEngineLog(child.stderr, "[engine:err]");\n }', + ); + + expect(source).toContain( + 'const ENGINE_LOG_ENABLED =\n process.env["AGENTMEMORY_ENGINE_LOG"] === "1" ||\n process.env["AGENTMEMORY_ENGINE_LOG"] === "true"', + ); + expect(source).toContain('stream.on("end", () => forwarder.flush())'); + }); + + // Engine-death detection is the only thing that turns a dead engine into a + // container restart. Log forwarding shares the same spawn call and the same + // stderr stream, so these assertions exist to fail if forwarding is ever + // allowed to gate, replace, or reorder any part of it. + it("keeps engine-death detection ungated by the log forwarder", () => { + expect(spawnBody).toContain("const spawnedAt = Date.now()"); + expect(spawnBody).toContain("const stderrChunks: Buffer[] = []"); + expect(spawnBody).toContain( + 'child.stderr?.on("data", (chunk: Buffer) => {\n if (stderrBytes >= MAX_STDERR_CAPTURE) return;', + ); + expect(spawnBody).toContain("const engineRanFor = Date.now() - spawnedAt"); + expect(spawnBody).toContain("if (engineRanFor > ENGINE_STARTUP_GRACE_MS)"); + expect(spawnBody).toContain( + 'if (process.env["AGENTMEMORY_EXIT_ON_ENGINE_DEATH"] !== "0") {\n process.exit(1);', + ); + + // The capture listener and the exit handler must sit outside the gate. + const gateStart = spawnBody.indexOf("if (ENGINE_LOG_ENABLED) {"); + const gateEnd = spawnBody.indexOf("}", spawnBody.indexOf("[engine:err]")); + const gateBlock = spawnBody.slice(gateStart, gateEnd); + expect(gateBlock).not.toContain("stderrChunks"); + expect(gateBlock).not.toContain("process.exit(1)"); + expect(spawnBody.indexOf('child.on("exit"')).toBeGreaterThan(gateEnd); + }); + + // The forwarder and the death report share process.stderr, and process.exit(1) + // discards pending async writes on a pipe, so the rate cap is what bounds how + // much forwarder output can ever sit ahead of the death report inside one + // second. Measured non-displacing at 64 KiB against a no-forwarder control; + // raising this cap voids that measurement, so it is pinned here rather than + // left to the structural assertions above, which would all still pass. + it("keeps the rate cap low enough not to crowd out the death report", () => { + expect(ENGINE_LOG_MAX_BYTES_PER_SECOND).toBeLessThanOrEqual(64 * 1024); + }); + + it("still captures the dying engine's stderr for the death report", () => { + expect(spawnBody).toContain( + 'const stderr = Buffer.concat(stderrChunks).toString("utf-8")', + ); + expect(spawnBody).toContain( + "if (stderr.trim()) console.error(`[agentmemory] engine stderr:\\n${stderr}`)", + ); + }); +}); diff --git a/test/deploy-entrypoint-drift.test.ts b/test/deploy-entrypoint-drift.test.ts new file mode 100644 index 000000000..0fb221503 --- /dev/null +++ b/test/deploy-entrypoint-drift.test.ts @@ -0,0 +1,68 @@ +import { describe, it, expect } from "vitest"; +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; + +// The four deploy targets ship copies of one entrypoint, not a shared file: +// three of them build from their own directory and Docker cannot COPY from +// outside the build context. Copies drift silently -- the iii-observability flag +// sat at `true` in three files and `false` in a fourth with nothing recording +// that a decision had been made. +// +// This guard is deliberately not line-indexed. An earlier version pinned line +// numbers and broke on any insertion. +const read = (t: string) => + readFileSync( + fileURLToPath(new URL(`../deploy/${t}/entrypoint.sh`, import.meta.url)), + "utf8", + ); + +const TARGETS = ["railway", "fly", "render", "coolify"] as const; +const files = Object.fromEntries(TARGETS.map((t) => [t, read(t)])) as Record< + (typeof TARGETS)[number], + string +>; + +/** Body with comments and blank lines dropped, so a comment cannot mask drift. */ +const code = (s: string) => + s + .split("\n") + .map((l) => l.trim()) + .filter((l) => l.length > 0 && !l.startsWith("#")) + .join("\n"); + +describe("deploy entrypoint drift", () => { + it("coolify and render remain byte-identical", () => { + expect(files.coolify).toBe(files.render); + }); + + // Railway is the only target that disables the in-memory OTEL exporter. The + // reason lives in deploy/railway/entrypoint.sh next to the value, so deleting + // this test cannot lose it. + it("only railway disables iii-observability", () => { + expect(files.railway).toMatch(/enabled: false/); + for (const t of ["fly", "render", "coolify"] as const) { + expect(files[t]).toMatch(/enabled: true/); + expect(files[t]).not.toMatch(/enabled: false/); + } + }); + + it("railway carries its reason next to the value", () => { + expect(files.railway).toMatch(/2026-08-23[\s\S]{0,120}enabled: false/); + }); + + // The real guard: with the observability line normalised away, railway's + // executable body must still equal render's. Any other divergence fails here, + // wherever it is inserted. + it("railway's executable body matches render apart from that one flag", () => { + const norm = (s: string) => code(s).replace(/enabled: (true|false)/, "enabled: X"); + expect(norm(files.railway)).toBe(norm(files.render)); + }); + + // Fly inserts its own block BEFORE the final exec rather than appending after + // it, so render's body up to that exec is the shared part. + it("fly shares render's body up to the final exec", () => { + const renderBody = code(files.render).split("\n").slice(0, -1); + const flyBody = code(files.fly).split("\n").slice(0, renderBody.length); + expect(flyBody).toEqual(renderBody); + }); +}); diff --git a/test/engine-log.test.ts b/test/engine-log.test.ts new file mode 100644 index 000000000..6b839e148 --- /dev/null +++ b/test/engine-log.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, it } from "vitest"; +import { + ENGINE_LOG_MAX_BYTES_PER_SECOND, + ENGINE_LOG_MAX_TOTAL_BYTES, + createEngineLogForwarder, +} from "../src/cli/engine-log.js"; + +function harness(overrides: Record = {}) { + const lines: string[] = []; + let clock = 0; + const forwarder = createEngineLogForwarder({ + prefix: "[engine]", + write: (line) => lines.push(line), + now: () => clock, + ...overrides, + }); + return { + lines, + forwarder, + advance(ms: number) { + clock += ms; + }, + push(text: string) { + forwarder.push(Buffer.from(text, "utf8")); + }, + }; +} + +describe("engine log forwarder", () => { + it("prefixes each complete line and holds partial lines back", () => { + const h = harness(); + + h.push("registered worker\nWorker unregis"); + expect(h.lines).toEqual(["[engine] registered worker"]); + + h.push("tered\n"); + expect(h.lines).toEqual([ + "[engine] registered worker", + "[engine] Worker unregistered", + ]); + }); + + it("strips carriage returns and decodes utf8 split across chunks", () => { + const h = harness(); + const snowman = Buffer.from("☃", "utf8"); + + h.push("crlf line\r\n"); + h.forwarder.push(snowman.subarray(0, 1)); + h.forwarder.push(snowman.subarray(1)); + h.push("\n"); + + expect(h.lines).toEqual(["[engine] crlf line", "[engine] ☃"]); + }); + + it("flushes a trailing partial line when the stream ends", () => { + const h = harness(); + + h.push("panic: engine died"); + expect(h.lines).toEqual([]); + + h.forwarder.flush(); + expect(h.lines).toEqual(["[engine] panic: engine died"]); + }); + + it("stops permanently once the lifetime ceiling is reached", () => { + const h = harness({ maxTotalBytes: 64, maxBytesPerSecond: 1024 * 1024 }); + + for (let i = 0; i < 200; i += 1) h.push(`line ${i}\n`); + + const terminal = h.lines.filter((l) => l.includes("ceiling reached")); + expect(terminal).toHaveLength(1); + expect(terminal[0]).toBe( + "[engine] log forwarding stopped: 64 byte ceiling reached", + ); + expect(h.lines[h.lines.length - 1]).toBe(terminal[0]); + + const forwarded = h.lines.filter((l) => !l.includes("ceiling reached")); + const forwardedBytes = forwarded.reduce( + (sum, l) => sum + Buffer.byteLength(l) + 1, + 0, + ); + expect(forwardedBytes).toBeLessThanOrEqual(64); + + const before = h.lines.length; + h.push("still chatty\n"); + h.advance(10_000); + h.push("still chatty\n"); + h.forwarder.flush(); + expect(h.lines).toHaveLength(before); + }); + + it("stays silent for the rest of a single chunk that trips the ceiling", () => { + const h = harness({ maxTotalBytes: 64, maxBytesPerSecond: 1024 * 1024 }); + + h.push(Array.from({ length: 50 }, (_, i) => `line ${i}`).join("\n") + "\n"); + + expect(h.lines.filter((l) => l.includes("ceiling reached"))).toHaveLength(1); + expect(h.lines[h.lines.length - 1]).toBe( + "[engine] log forwarding stopped: 64 byte ceiling reached", + ); + }); + + it("drops over the rate cap and reports the drop once per window", () => { + const h = harness({ maxBytesPerSecond: 64, maxTotalBytes: 1024 * 1024 }); + const a = "a".repeat(30); + + h.push(`${a}\n`); + h.push(`${"b".repeat(30)}\n`); + h.push(`${"c".repeat(30)}\n`); + expect(h.lines).toEqual([`[engine] ${a}`]); + + h.advance(1000); + h.push(`${"d".repeat(30)}\n`); + expect(h.lines).toEqual([ + `[engine] ${a}`, + "[engine] dropped 80 bytes (rate cap 64 bytes/s)", + ]); + }); + + it("never suppresses silently, even with the window budget exhausted", () => { + const h = harness({ maxBytesPerSecond: 64, maxTotalBytes: 1024 * 1024 }); + + for (let i = 0; i < 20; i += 1) h.push(`${"a".repeat(30)}\n`); + h.forwarder.flush(); + + expect(h.lines).toEqual([ + `[engine] ${"a".repeat(30)}`, + "[engine] dropped 760 bytes (rate cap 64 bytes/s)", + ]); + }); + + it("charges its exempt notices against the lifetime ceiling", () => { + const h = harness({ maxBytesPerSecond: 8, maxTotalBytes: 200 }); + + for (let round = 0; round < 40; round += 1) { + h.push(`${"a".repeat(30)}\n`); + h.advance(1000); + } + + const forwardedBytes = h.lines + .filter((l) => !l.includes("ceiling reached")) + .reduce((sum, l) => sum + Buffer.byteLength(l) + 1, 0); + expect(forwardedBytes).toBeLessThanOrEqual(200); + expect(h.lines.filter((l) => l.includes("ceiling reached"))).toHaveLength(1); + }); + + it("force-flushes a newline-free stream instead of buffering it forever", () => { + const h = harness({ maxLineBytes: 16 }); + + h.push("0123456789abcdefghij"); + + expect(h.lines).toEqual(["[engine] 0123456789abcdefghij"]); + }); + + it("ships ceilings that bound a firehose without operator action", () => { + expect(ENGINE_LOG_MAX_TOTAL_BYTES).toBe(32 * 1024 * 1024); + expect(ENGINE_LOG_MAX_BYTES_PER_SECOND).toBe(64 * 1024); + }); +}); diff --git a/test/health-monitor.test.ts b/test/health-monitor.test.ts new file mode 100644 index 000000000..14b42e0f6 --- /dev/null +++ b/test/health-monitor.test.ts @@ -0,0 +1,142 @@ +import { describe, it, expect } from "vitest"; +import { readFileSync } from "node:fs"; +import { fileURLToPath } from "node:url"; +import type { HealthSnapshot } from "../src/types.js"; +import { bumpEscalation, type EscalationState } from "../src/health/monitor.js"; + +function snap(over: Partial = {}): HealthSnapshot { + return { + connectionState: "connected", + workers: [], + memory: { heapUsed: 0, heapTotal: 1, rss: 0, external: 0 }, + cpu: { userMicros: 0, systemMicros: 0, percent: 0 }, + eventLoopLagMs: 0, + uptimeSeconds: 600, + kvConnectivity: { status: "ok", latencyMs: 1 }, + status: "healthy", + alerts: [], + ...over, + }; +} + +const stalled = () => snap({ kvConnectivity: { status: "error", error: "kv_probe_failed" } }); + +function fresh(): EscalationState { + return { consecutiveStalls: 0, armed: false, escalated: false }; +} + +/** Arm the state the way a healthy first collection would. */ +function armed(): EscalationState { + const s = fresh(); + bumpEscalation(snap(), s, 3); + return s; +} + +describe("bumpEscalation arming", () => { + // Mirrors the shell watchdog's "wait for a first success" rule. Without it a + // store that is already stalled at boot escalates on the first minute of every + // container life, spending the platform's restart budget in minutes. + it("never escalates when the store is stalled from the very first check", () => { + const state = fresh(); + for (let i = 0; i < 50; i++) { + expect(bumpEscalation(stalled(), state, 3)).toBe(false); + } + expect(state.armed).toBe(false); + expect(state.consecutiveStalls).toBe(0); + }); + + it("arms on the first healthy probe", () => { + const state = fresh(); + bumpEscalation(snap(), state, 3); + expect(state.armed).toBe(true); + }); +}); + +describe("bumpEscalation gating", () => { + it("escalates on the Nth consecutive stall, not before", () => { + const state = armed(); + expect(bumpEscalation(stalled(), state, 3)).toBe(false); + expect(bumpEscalation(stalled(), state, 3)).toBe(false); + expect(bumpEscalation(stalled(), state, 3)).toBe(true); + }); + + it("resets the counter on a healthy probe", () => { + const state = armed(); + bumpEscalation(stalled(), state, 3); + bumpEscalation(stalled(), state, 3); + bumpEscalation(snap(), state, 3); + expect(state.consecutiveStalls).toBe(0); + expect(bumpEscalation(stalled(), state, 3)).toBe(false); + }); + + it("escalates at most once per process", () => { + const state = armed(); + let fired = 0; + for (let i = 0; i < 20; i++) { + if (bumpEscalation(stalled(), state, 1)) fired++; + } + expect(fired).toBe(1); + }); + + // The gate is the KV probe, not snapshot.status. evaluateHealth raises + // `critical` from five independent conditions, so gating on the aggregate + // would let a CPU spike during a consolidation pass kill a healthy process. + it.each([ + ["cpu", { cpu: { userMicros: 0, systemMicros: 0, percent: 99 } }], + ["event loop lag", { eventLoopLagMs: 5000 }], + ["connection", { connectionState: "disconnected" }], + ["memory", { memory: { heapUsed: 99, heapTotal: 100, rss: 99, external: 0 } }], + ])("never escalates on a %s critical while the KV probe is healthy", (_label, over) => { + const state = armed(); + for (let i = 0; i < 50; i++) { + expect( + bumpEscalation(snap({ ...over, status: "critical" } as Partial), state, 3), + ).toBe(false); + } + expect(state.consecutiveStalls).toBe(0); + }); + + it("treats an absent kvConnectivity as no signal, not as a stall", () => { + const state = armed(); + const s = snap(); + delete s.kvConnectivity; + for (let i = 0; i < 50; i++) { + expect(bumpEscalation(s, state, 3)).toBe(false); + } + expect(state.consecutiveStalls).toBe(0); + }); +}); + +// KTD7. The escalation decision must precede the snapshot persist. `kv.set` +// routes through `sdk.trigger`, whose invocationTimeoutMs is 180000 +// (src/index.ts), so a stalled store parks the persist for three minutes. A +// counter placed behind it would never advance during the exact failure it +// exists to catch. +// +// Asserted on source order because no test constructs registerHealthMonitor, so +// the call site is otherwise executed by nothing. Verified discriminating: this +// passes on the real tree and fails on a tree with the two lines swapped. +// Matches the precedent in test/cli-second-instance-guard.test.ts. +describe("collectHealth statement order", () => { + const src = readFileSync( + fileURLToPath(new URL("../src/health/monitor.ts", import.meta.url)), + "utf8", + ); + + it("decides escalation before the un-raced snapshot persist", () => { + const decide = src.indexOf("bumpEscalation(snapshot, escalationState"); + const persist = src.indexOf('kv.set(KV.health, "latest"'); + expect(decide).toBeGreaterThan(-1); + expect(persist).toBeGreaterThan(-1); + expect(decide).toBeLessThan(persist); + }); + + // The workers probe sits upstream of both. Unraced it inherits the engine's + // 180s invocation timeout and delays detection by tens of minutes. + it("races the workers probe so it cannot park the collection", () => { + const workers = src.indexOf('function_id: "engine::workers::list"'); + expect(workers).toBeGreaterThan(-1); + const before = src.slice(Math.max(0, workers - 400), workers); + expect(before).toContain("Promise.race"); + }); +}); diff --git a/test/health-thresholds.test.ts b/test/health-thresholds.test.ts index 6f918a917..bae18455e 100644 --- a/test/health-thresholds.test.ts +++ b/test/health-thresholds.test.ts @@ -95,3 +95,125 @@ describe("evaluateHealth memory severity", () => { expect(strict.status).toBe("healthy"); }); }); + +describe("evaluateHealth memory severity — denominator", () => { + const LIMIT_6192MB = 6192 * 1024 * 1024; + + it("stays healthy when a busy process fills its committed heap but sits far below the V8 limit", () => { + // Captured from a live deployment: this process reported + // memory_critical_97% while using 6% of the heap it may grow to. + const s = snap({ + memory: { + heapUsed: 390_771_624, + heapTotal: 404_930_560, + rss: 584_327_168, + external: 6_566_392, + heapSizeLimit: LIMIT_6192MB, + }, + }); + + const { status, alerts } = evaluateHealth(s); + + expect(status).toBe("healthy"); + expect(alerts.find((a) => a.startsWith("memory_critical_"))).toBeUndefined(); + expect(alerts.find((a) => a.startsWith("memory_warn_"))).toBeUndefined(); + }); + + it("reports the percentage against the limit, not against the committed heap", () => { + const s = snap({ + memory: { + heapUsed: 390_771_624, + heapTotal: 404_930_560, + rss: 584_327_168, + external: 0, + heapSizeLimit: LIMIT_6192MB, + }, + }); + + // Widen the band so the alert fires as a warn either way and the rendered + // percentage is the only thing under test. + const { alerts } = evaluateHealth(s, { + memoryWarnPercent: 1, + memoryCriticalPercent: 99, + }); + const warn = alerts.find((a) => a.startsWith("memory_warn_")); + + // 390771624 / 6492782592 = 6%, not 390771624 / 404930560 = 97%. + expect(warn).toBe("memory_warn_6%_rss557mb"); + }); + + it("still goes critical when the heap genuinely approaches the V8 limit", () => { + const s = snap({ + memory: { + heapUsed: 6_200_000_000, + heapTotal: 6_300_000_000, + rss: 6_800_000_000, + external: 0, + heapSizeLimit: LIMIT_6192MB, + }, + }); + + const { status, alerts } = evaluateHealth(s); + + expect(status).toBe("critical"); + expect(alerts.some((a) => a.startsWith("memory_critical_"))).toBe(true); + }); + + it("falls back to the committed heap when heapSizeLimit is absent", () => { + const s = snap({ + memory: { + heapUsed: 970 * 1024 * 1024, + heapTotal: 1000 * 1024 * 1024, + rss: 1100 * 1024 * 1024, + external: 0, + }, + }); + + expect(evaluateHealth(s).status).toBe("critical"); + }); +}); + +describe("evaluateHealth KV connectivity", () => { + it("goes critical when the KV probe fails", () => { + const s = snap({ + kvConnectivity: { status: "error", error: "kv_probe_failed", latencyMs: 5000 }, + }); + const { status, alerts } = evaluateHealth(s); + expect(status).toBe("critical"); + expect(alerts).toContain("kv_probe_failed"); + }); + + it("adds no KV alert when the probe succeeds", () => { + const { status, alerts } = evaluateHealth(snap()); + expect(status).toBe("healthy"); + expect(alerts.find((a) => a.startsWith("kv_"))).toBeUndefined(); + }); + + // kvConnectivity is optional on HealthSnapshot, and snapshots persisted before + // the field existed still come back from KV, so absence must read as "no + // signal" rather than as a failure. + it("stays healthy when kvConnectivity is absent", () => { + const s = snap(); + delete s.kvConnectivity; + const { status, alerts } = evaluateHealth(s); + expect(status).toBe("healthy"); + expect(alerts.find((a) => a.startsWith("kv_"))).toBeUndefined(); + }); + + it("does not throw or alert on a malformed kvConnectivity", () => { + const s = snap({ kvConnectivity: { status: undefined as unknown as string } }); + expect(() => evaluateHealth(s)).not.toThrow(); + expect(evaluateHealth(s).alerts.find((a) => a.startsWith("kv_"))).toBeUndefined(); + }); + + it("reports the KV alert alongside other critical signals", () => { + const s = snap({ + kvConnectivity: { status: "error", error: "kv_probe_failed" }, + eventLoopLagMs: 900, + }); + const { status, alerts } = evaluateHealth(s); + expect(status).toBe("critical"); + expect(alerts).toContain("kv_probe_failed"); + expect(alerts.find((a) => a.startsWith("event_loop_lag_critical_"))).toBeDefined(); + }); +}); diff --git a/test/index-persistence.test.ts b/test/index-persistence.test.ts index 929791657..3dff05409 100644 --- a/test/index-persistence.test.ts +++ b/test/index-persistence.test.ts @@ -563,6 +563,383 @@ describe("IndexPersistence", () => { expect(loaded.bm25!.search("alpha").length).toBe(0); }); + it("reclaims the previous generation when the previous manifest read fails (#1115)", async () => { + const previous = makeBm25("obs_old", "alpha previous snapshot"); + await new IndexPersistence(kv as never, previous, null, { + shardChars: 80, + createGeneration: () => "gen_old", + }).save(); + const oldShardScope = "mem:index:bm25:bm25:gen_old:00000"; + await expect(kv.get(oldShardScope, "data")).resolves.not.toBeNull(); + + // The manifest read that opens saveShardedIndex times out. It used to be + // swallowed into `previous = null`, which skipped the cleanup guard and + // stranded gen_old's shards with nothing left to ever revisit them. + const readFailsKv = { + ...kv, + get: vi.fn(async (scope: string, key: string): Promise => { + if (scope === BM25_SCOPE && key === BM25_MANIFEST_KEY) { + throw new Error("Invocation timeout after 180000ms: state::get"); + } + return kv.get(scope, key); + }), + }; + + const next = makeBm25("obs_new", "bravo new snapshot"); + await new IndexPersistence(readFailsKv as never, next, null, { + shardChars: 80, + createGeneration: () => "gen_new", + }).save(); + + const manifest = await getBm25Manifest(kv); + expect(manifest.generation).toBe("gen_new"); + await expect(kv.get(oldShardScope, "data")).resolves.toBeNull(); + }); + + it("reclaims a generation stranded by a failed cleanup on the next load (#1115)", async () => { + const previous = makeBm25("obs_old", "alpha previous snapshot"); + await new IndexPersistence(kv as never, previous, null, { + shardChars: 80, + createGeneration: () => "gen_old", + }).save(); + const oldShardScope = "mem:index:bm25:bm25:gen_old:00000"; + + const cleanupKv = { + ...kv, + delete: vi.fn(async () => { + throw new Error("cleanup failed"); + }), + }; + const next = makeBm25("obs_new", "bravo new snapshot"); + await new IndexPersistence(cleanupKv as never, next, null, { + shardChars: 80, + createGeneration: () => "gen_new", + }).save(); + + // Cleanup failed, so gen_old is still on disk. Nothing in the save path + // will revisit it — a later save only ever inspects its own predecessor. + await expect(kv.get(oldShardScope, "data")).resolves.not.toBeNull(); + + const loaded = await new IndexPersistence( + kv as never, + new SearchIndex(), + null, + ).load(); + expect(loaded.bm25!.search("bravo").length).toBe(1); + await expect(kv.get(oldShardScope, "data")).resolves.toBeNull(); + }); + + it("reclaims the previous vector generation when the vector manifest read fails (#1115)", async () => { + await new IndexPersistence( + kv as never, + makeBm25("obs_old", "alpha previous snapshot"), + makeVector("obs_old"), + { shardChars: 80, createGeneration: () => "gen_old" }, + ).save(); + const oldVectorScope = "mem:index:bm25:vectors:gen_old:00000"; + await expect(kv.get(oldVectorScope, "data")).resolves.not.toBeNull(); + + const readFailsKv = { + ...kv, + get: vi.fn(async (scope: string, key: string): Promise => { + if (scope === BM25_SCOPE && key === VECTOR_MANIFEST_KEY) { + throw new Error("Invocation timeout after 180000ms: state::get"); + } + return kv.get(scope, key); + }), + }; + + await new IndexPersistence( + readFailsKv as never, + makeBm25("obs_new", "bravo new snapshot"), + makeVector("obs_new"), + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + await expect(kv.get(oldVectorScope, "data")).resolves.toBeNull(); + }); + + it("leaves a generation recorded after the live one untouched on load (#1115)", async () => { + await new IndexPersistence( + kv as never, + makeBm25("obs_old", "alpha previous snapshot"), + null, + { shardChars: 80, createGeneration: () => "gen_live" }, + ).save(); + + // A concurrent save has recorded its generation and is mid-write, but has + // not published its manifest yet. setIndexPersistence runs before load() + // in src/index.ts, so a request arriving during boot produces exactly this. + const inflightScope = "mem:index:bm25:bm25:gen_inflight:00000"; + await kv.set(inflightScope, "data", "partial shard"); + const gcKey = `${BM25_MANIFEST_KEY}:gc`; + const ledger = await kv.get<{ + v: 1; + generations: Array<{ + generation: string; + shards: Array<{ scope: string; key: string }>; + }>; + }>(BM25_SCOPE, gcKey); + ledger!.generations.push({ + generation: "gen_inflight", + shards: [{ scope: inflightScope, key: "data" }], + }); + await kv.set(BM25_SCOPE, gcKey, ledger); + + await new IndexPersistence(kv as never, new SearchIndex(), null).load(); + + // Deleting it would leave the in-flight save publishing a manifest whose + // shards are already gone, which fails closed on the next load. + await expect(kv.get(inflightScope, "data")).resolves.toBe("partial shard"); + }); + + it("reclaims a pre-ledger manifest that carries no generation (#1115)", async () => { + const legacyScope = "mem:index:bm25:bm25:gen_legacy:00000"; + await kv.set(legacyScope, "data", "legacy shard"); + await kv.set(BM25_SCOPE, BM25_MANIFEST_KEY, { + v: 1, + shards: [{ scope: legacyScope, key: "data", chars: 12 }], + chars: 12, + }); + + await new IndexPersistence( + kv as never, + makeBm25("obs_new", "bravo new snapshot"), + null, + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + await expect(kv.get(legacyScope, "data")).resolves.toBeNull(); + }); + + it("keeps a published manifest whole when two saves overlap (#1115)", async () => { + // scheduleSave fires save() unawaited from a timer while flushIndexSave + // awaits save() on every delete path, so two saves on ONE instance is the + // normal shape, not a contrivance. Without the queue, whichever published + // second reclaimed the other's shards and left it naming data already gone. + vi.useRealTimers(); + const store = new Map>(); + const slowKv = { + get: async (scope: string, key: string): Promise => + (store.get(scope)?.get(key) as T) ?? null, + set: async (scope: string, key: string, data: T): Promise => { + // Stall one of the first generation's shard writes so the second save + // overtakes it. + if (scope.includes(":gen_0:") && scope.endsWith("00005")) { + await new Promise((resolve) => setTimeout(resolve, 200)); + } + if (!store.has(scope)) store.set(scope, new Map()); + store.get(scope)!.set(key, data); + return data; + }, + delete: async (scope: string, key: string): Promise => { + store.get(scope)?.delete(key); + }, + list: async (scope: string): Promise => + Array.from((store.get(scope)?.values() ?? []) as Iterable), + }; + + const bm25 = new SearchIndex(); + for (let i = 0; i < 30; i++) { + bm25.add( + makeObs({ id: `obs_${i}`, title: `lorem ipsum dolor sit amet ${i}` }), + ); + } + let generation = 0; + const persistence = new IndexPersistence(slowKv as never, bm25, null, { + shardChars: 400, + createGeneration: () => `gen_${generation++}`, + }); + + // Let the first save get into its shard writes before the second starts, + // so the second is the one that publishes. + const first = persistence.save(); + await new Promise((resolve) => setTimeout(resolve, 20)); + const second = persistence.save(); + await Promise.all([first, second]); + + const manifest = await slowKv.get( + BM25_SCOPE, + BM25_MANIFEST_KEY, + ); + const missing: string[] = []; + for (const shard of manifest!.shards) { + if ((await slowKv.get(shard.scope, shard.key)) === null) { + missing.push(shard.scope); + } + } + expect(missing).toEqual([]); + + const loaded = await new IndexPersistence( + slowKv as never, + new SearchIndex(), + null, + ).load(); + expect(loaded.bm25).not.toBeNull(); + }); + + it("still persists the index when the gc ledger is unreadable (#1115)", async () => { + // state::get timing out is the condition this bug appears under. Aborting + // the save there would stop persisting entirely, which is worse than the + // leak: a lost BM25 index costs a full-corpus rebuild. + const failingKv = { + ...kv, + get: vi.fn(async (scope: string, key: string): Promise => { + if (scope === BM25_SCOPE && key === `${BM25_MANIFEST_KEY}:gc`) { + throw new Error("Invocation timeout after 180000ms: state::get"); + } + return kv.get(scope, key); + }), + }; + + await new IndexPersistence( + failingKv as never, + makeBm25("obs_new", "bravo new snapshot"), + null, + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + const manifest = await getBm25Manifest(kv); + expect(manifest.generation).toBe("gen_new"); + await expect( + kv.get(manifest.shards[0].scope, manifest.shards[0].key), + ).resolves.not.toBeNull(); + }); + + it("still persists when the gc ledger holds a malformed entry (#1115)", async () => { + // A null entry used to throw a TypeError out of saveShardedIndex before any + // shard write, leaving one throttled log line per 60s while BM25 silently + // stopped persisting for good. + await kv.set(BM25_SCOPE, `${BM25_MANIFEST_KEY}:gc`, { + v: 1, + generations: [null], + }); + + await new IndexPersistence( + kv as never, + makeBm25("obs_new", "bravo new snapshot"), + null, + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + const manifest = await getBm25Manifest(kv); + expect(manifest.generation).toBe("gen_new"); + await expect( + kv.get(manifest.shards[0].scope, manifest.shards[0].key), + ).resolves.not.toBeNull(); + }); + + it("still reclaims when the gc ledger holds a malformed entry (#1115)", async () => { + // Publishing is not enough. A malformed entry that silently disabled + // reclaim would leak a generation every cycle while this test stayed green. + await new IndexPersistence(kv as never, makeBm25("obs_old", "alpha"), null, { + shardChars: 80, + createGeneration: () => "gen_old", + }).save(); + const oldShardScope = "mem:index:bm25:bm25:gen_old:00000"; + + const gcKey = `${BM25_MANIFEST_KEY}:gc`; + const ledger = await kv.get<{ v: 1; generations: unknown[] }>( + BM25_SCOPE, + gcKey, + ); + ledger!.generations.unshift(null); + await kv.set(BM25_SCOPE, gcKey, ledger); + + await new IndexPersistence(kv as never, makeBm25("obs_new", "bravo"), null, { + shardChars: 80, + createGeneration: () => "gen_new", + }).save(); + + await expect(kv.get(oldShardScope, "data")).resolves.toBeNull(); + }); + + it("falls back to previous-generation cleanup when the ledger is unusable (#1115)", async () => { + // Measured regression guard: with no fallback, an unusable ledger leaked + // one generation per cycle — strictly worse than the code this replaced, + // which always had this path. Without this test the `else if` can be + // deleted with the whole suite green. + await new IndexPersistence(kv as never, makeBm25("obs_old", "alpha"), null, { + shardChars: 80, + createGeneration: () => "gen_old", + }).save(); + const oldShardScope = "mem:index:bm25:bm25:gen_old:00000"; + await expect(kv.get(oldShardScope, "data")).resolves.not.toBeNull(); + + // Ledger read fails, manifest read succeeds. Both are state::get with + // independent timeouts, so this split is ordinary, not contrived. + const ledgerFailsKv = { + ...kv, + get: vi.fn(async (scope: string, key: string): Promise => { + if (scope === BM25_SCOPE && key === `${BM25_MANIFEST_KEY}:gc`) { + throw new Error("Invocation timeout after 180000ms: state::get"); + } + return kv.get(scope, key); + }), + }; + + await new IndexPersistence( + ledgerFailsKv as never, + makeBm25("obs_new", "bravo"), + null, + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + const manifest = await getBm25Manifest(kv); + expect(manifest.generation).toBe("gen_new"); + await expect(kv.get(oldShardScope, "data")).resolves.toBeNull(); + }); + + it("keeps the ledger entry when a rollback delete fails (#1115)", async () => { + // Shards that survived their rollback delete must keep the entry that + // names them, or nothing can ever reclaim them. + let shardWrites = 0; + const failingKv = { + ...kv, + set: vi.fn(async (scope: string, key: string, data: T): Promise => { + if (scope.includes(":gen_new:")) { + shardWrites += 1; + if (shardWrites === 2) throw new Error("shard write failed"); + } + return kv.set(scope, key, data); + }), + delete: vi.fn(async () => { + throw new Error("rollback delete failed"); + }), + }; + + await new IndexPersistence( + failingKv as never, + makeBm25("obs_new", "bravo new snapshot"), + null, + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + const ledger = await kv.get<{ + v: 1; + generations: Array<{ generation: string }>; + }>(BM25_SCOPE, `${BM25_MANIFEST_KEY}:gc`); + expect(ledger!.generations.map((entry) => entry.generation)).toContain( + "gen_new", + ); + }); + + it("refuses to overwrite a gc ledger it does not recognise (#1115)", async () => { + const gcKey = `${BM25_MANIFEST_KEY}:gc`; + const future = { v: 2, generations: [], writtenByANewerBuild: true }; + await kv.set(BM25_SCOPE, gcKey, future); + + await new IndexPersistence( + kv as never, + makeBm25("obs_new", "bravo new snapshot"), + null, + { shardChars: 80, createGeneration: () => "gen_new" }, + ).save(); + + // Rewriting it as a v1 ledger would drop every generation it tracked. + await expect(kv.get(BM25_SCOPE, gcKey)).resolves.toEqual(future); + }); + it("keeps the previous vector generation when vector save fails after BM25 publish", async () => { const previousBm25 = makeBm25("obs_old", "alpha previous snapshot"); const previousVector = makeVector("obs_old"); diff --git a/test/resilient-provider.test.ts b/test/resilient-provider.test.ts new file mode 100644 index 000000000..4f7b028a1 --- /dev/null +++ b/test/resilient-provider.test.ts @@ -0,0 +1,103 @@ +import { describe, it, expect } from "vitest"; +import { ResilientProvider } from "../src/providers/resilient.js"; +import type { MemoryProvider } from "../src/types.js"; + +// Tracks how many calls are inside the provider at once, which is the thing the +// upstream API actually rejects. Counting total calls would not discriminate. +function countingProvider( + behaviour: (n: number) => Promise = async () => "ok", +): MemoryProvider & { peak: number; calls: number } { + let inFlight = 0; + const state = { + name: "counting", + peak: 0, + calls: 0, + async compress(): Promise { + inFlight++; + state.calls++; + if (inFlight > state.peak) state.peak = inFlight; + try { + return await behaviour(state.calls); + } finally { + inFlight--; + } + }, + async summarize(): Promise { + return state.compress(); + }, + }; + return state; +} + +function rateLimited(): Error { + return new Error('OpenAI API error (429): {"error":"too many concurrent requests"}'); +} + +describe("ResilientProvider concurrency", () => { + it("never runs more than the configured number of calls at once", async () => { + const inner = countingProvider( + () => new Promise((resolve) => setTimeout(() => resolve("ok"), 5)), + ); + const provider = new ResilientProvider(inner, { maxConcurrent: 3 }); + + const results = await Promise.all( + Array.from({ length: 24 }, () => provider.compress("sys", "user")), + ); + + expect(inner.peak).toBeLessThanOrEqual(3); + // Bounding must not drop work: every call still ran and still resolved. + expect(inner.calls).toBe(24); + expect(results.every((r) => r === "ok")).toBe(true); + }); + + it("releases its slot when a call throws", async () => { + const inner = countingProvider(async (n) => { + if (n <= 2) throw new Error("boom"); + return "ok"; + }); + // maxConcurrent 1 is load-bearing here: at the default of 4 a leaked slot + // would not deadlock, and the test would pass despite the bug. + const provider = new ResilientProvider(inner, { maxConcurrent: 1 }); + + const settled = await Promise.allSettled( + Array.from({ length: 5 }, () => provider.compress("sys", "user")), + ); + + expect(settled).toHaveLength(5); + expect(settled.filter((s) => s.status === "fulfilled")).toHaveLength(3); + }); +}); + +describe("ResilientProvider rate limiting", () => { + it("does not open the breaker on 429s", async () => { + const inner = countingProvider(async (n) => { + if (n <= 5) throw rateLimited(); + return "ok"; + }); + const provider = new ResilientProvider(inner); + + // Five, against a default failure threshold of three. + for (let i = 0; i < 5; i++) { + await provider.compress("sys", "user").catch(() => undefined); + } + + expect(provider.circuitState.state).toBe("closed"); + await expect(provider.compress("sys", "user")).resolves.toBe("ok"); + }); + + it("still opens the breaker on genuine failures", async () => { + const inner = countingProvider(async () => { + throw new Error("upstream exploded"); + }); + const provider = new ResilientProvider(inner); + + for (let i = 0; i < 3; i++) { + await provider.compress("sys", "user").catch(() => undefined); + } + + expect(provider.circuitState.state).toBe("open"); + await expect(provider.compress("sys", "user")).rejects.toThrow( + "circuit_breaker_open", + ); + }); +}); diff --git a/test/viewer-heap-gauge.test.ts b/test/viewer-heap-gauge.test.ts new file mode 100644 index 000000000..b99ee4907 --- /dev/null +++ b/test/viewer-heap-gauge.test.ts @@ -0,0 +1,44 @@ +import { describe, it, expect } from "vitest"; +import { readFileSync } from "node:fs"; + +// The dashboard used to recompute heapUsed / heapTotal client-side, which is +// the same over-reporting the health thresholds had: heapTotal is what V8 has +// committed, not what it may grow to. The gauge now measures against +// heapSizeLimit, and it divides raw byte values so it cannot disagree with +// evaluateHealth over a rounding step. +// +// Asserting on the emitted source rather than running the gauge follows +// viewer-graph-cooldown and viewer-memories-sort: the viewer ships as one +// HTML file with inline JS, so there is no module to import and execute. +describe("viewer heap gauge", () => { + const viewer = readFileSync("src/viewer/index.html", "utf-8"); + + it("measures against the V8 heap limit, falling back to heapTotal", () => { + expect(viewer).toMatch(/limitBytes\s*=\s*snap\.memory\.heapSizeLimit\s*\|\|\s*0/); + expect(viewer).toMatch( + /ceilingBytes\s*=\s*limitBytes\s*>\s*0\s*\?\s*limitBytes\s*:\s*\(snap\.memory\.heapTotal\s*\|\|\s*0\)/, + ); + }); + + it("computes the percentage from raw bytes, not the MB-rounded label values", () => { + expect(viewer).toMatch( + /heapPercent\s*=\s*ceilingBytes\s*>\s*0\s*\?\s*\(\(snap\.memory\.heapUsed\s*\|\|\s*0\)\s*\/\s*ceilingBytes\)\s*\*\s*100/, + ); + // The old form divided two already-rounded MB numbers. + expect(viewer).not.toMatch(/heapPct\s*=\s*heapTotal\s*>\s*0\s*\?\s*Math\.round\(\(heapUsed\s*\/\s*heapTotal\)/); + }); + + it("picks the gauge colour on the unrounded percentage", () => { + // Rounding first put the gauge a whole point out of step with + // evaluateHealth, which compares the raw value: at 80.4% health warns + // while a rounded 80 left the bar on the lower colour. + expect(viewer).toMatch(/heapPct\s*=\s*Math\.round\(heapPercent\)/); + expect(viewer).toMatch(/heapColor\s*=\s*\(heapPercent\s*>\s*80\s*&&\s*rssAboveFloor\)/); + expect(viewer).toMatch(/\(heapPercent\s*>\s*60\s*&&\s*rssAboveFloor\)/); + }); + + it("rounds only for the displayed label", () => { + expect(viewer).toMatch(/heapCeiling\s*=\s*Math\.round\(ceilingBytes\s*\/\s*1024\s*\/\s*1024\)/); + expect(viewer).toMatch(/\+\s*heapUsed\s*\+\s*' \/ '\s*\+\s*heapCeiling\s*\+\s*' MB/); + }); +});