diff --git a/.aiwg/reports/g-icm-01-interface-inventory.json b/.aiwg/reports/g-icm-01-interface-inventory.json index 1ff4eb402..4a2322d6f 100644 --- a/.aiwg/reports/g-icm-01-interface-inventory.json +++ b/.aiwg/reports/g-icm-01-interface-inventory.json @@ -3,9 +3,9 @@ "control_id": "G-ICM-01", "project_id": "github.com/chronodeai/agentmemory", "source_identity": { - "commit_sha": "e241ebb778a9d32a7726ed80b6359fa474139307", - "commit_tree_sha": "bbc5df4d5bfc44e47227e3096455df47a4d64f92", - "inventory_input_sha256": "fd4bf3dffd45c890813b20b16efa791cd6b72f055f65198e8489faffbbe55be8" + "commit_sha": "facabd34882d1e66a5ec3d4821eecde235fed1b1", + "commit_tree_sha": "b58a99a88f05e094b42be7c52e72784d87cc4e6d", + "inventory_input_sha256": "2e776824dd49040af5451aab80bbbe2e2f0dcab5aab982653711558baa74625b" }, "public_route_allowlist": [ "GET /agentmemory/livez" @@ -2270,12 +2270,12 @@ "source": "src/functions/compress-file.ts:148" }, { - "surface_id": "PROVIDER:ATTEMPT:src/functions/compress.ts:111:describeImage", + "surface_id": "PROVIDER:ATTEMPT:src/functions/compress.ts:112:describeImage", "purpose": "vision", "kind": "invocation", "method": "describeImage", "receiver": "provider", - "source": "src/functions/compress.ts:111" + "source": "src/functions/compress.ts:112" }, { "surface_id": "PROVIDER:ATTEMPT:src/functions/consolidate.ts:176:compress", @@ -10819,12 +10819,12 @@ ] }, { - "surface_id": "PROVIDER:ATTEMPT:src/functions/compress.ts:111:describeImage", + "surface_id": "PROVIDER:ATTEMPT:src/functions/compress.ts:112:describeImage", "purpose": "vision", "kind": "invocation", "method": "describeImage", "receiver": "provider", - "source": "src/functions/compress.ts:111", + "source": "src/functions/compress.ts:112", "type": "provider-attempt", "auth_control": "processing-policy", "control_ids": [ diff --git a/CHANGELOG.md b/CHANGELOG.md index ed5392686..336e7d937 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,20 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## [Unreleased] +## [0.9.30-chronode.5] — 2026-08-25 + +Backlog-closure release: live-stream disk bounding, deployment watchdog v2, and a written upstream-sync process. + +### Added + +- **Viewer live-stream size cap with rotation** (`fix/stream`). The iii-stream file adapter persists the viewer live stream (`mem-live`/`viewer`) as one append-only `.bin` that grew unbounded — 130 MB observed on the long-running deployment. Every publish path into that stream (`mem::observe` raw and compressed viewer events, `mem::compress`, and the session-activity trigger) now checks the persisted file first and, past `AGENTMEMORY_LIVE_STREAM_MAX_BYTES` (default `33554432`; `0` opts out), rotates it once to `.prev` (overwriting the previous generation) so the engine's next append starts fresh. Rotation is best-effort by contract: failures cost at most one warn line and never block or fail an observation. The engine opens its store by path per append (verified: no long-lived fd), so no restart is required for rotation to take effect. +- **Upstream sync playbook** (`docs/upstream-sync.md`) capturing the staged-train process proven on this deployment: pinned-refspec ref refreshes (plain `git fetch origin` does not move `origin/main` here), four-bucket classification of `..upstream/main` (pull-first bugfixes / decompose-and-port / re-port-through-bundle-pipeline / skip-unused-host-adapters), the non-negotiable gates (hook bundles as committed build outputs, the `ci/r13-test-manifest.json` recompute algorithm, evidence inventory refresh, skills gen/check, tool-count consistency surfaces), the full pre-push verification battery including canonical R13 under dummy secrets with `RUN_HF_SMOKE=1`, direct-to-main PR policy (stacked-base deletion auto-closes stacked PRs), and the local deploy recipe including the `lib/node_modules` anchor-`package.json` pitfall for `npm install --omit=dev`. +- **Sync position helper** (`scripts/sync/upstream-status.sh`): prints the deployed release dir, local main sha, upstream/main sha, ahead/behind of both versus the deployed sha, and the 17 newest upstream commit subjects. + +### Deployment (not repo code) + +- Watchdog v2 (`~/.agentmemory/bin/watchdog.sh`): alongside the existing livez check it now curls the viewer health surface (`http://127.0.0.1:3113/health`); when livez answers but the viewer does not, one `"viewer down"` line lands in `~/Library/Logs/Agentmemory/watchdog.log` instead of kickstarting healthy engine state; a livez failure that issues a kickstart logs `"engine restarted at "`. The launchd stderr spool is bounded: past 52428800 bytes it keeps the newest 20000 lines via tail-then-mv. + ## [0.9.30-chronode.4] — 2026-08-25 Upstream sync: port of rohitg00/agentmemory `8c90741` — migration from the deprecated `@xenova/transformers` (silently broken on Node 22+) to `@huggingface/transformers` v4, the renamed continuation of the same Apache-2.0 codebase. diff --git a/ci/r13-test-manifest.json b/ci/r13-test-manifest.json index 4a376626e..13e5b68b8 100644 --- a/ci/r13-test-manifest.json +++ b/ci/r13-test-manifest.json @@ -1,5 +1,5 @@ { - "count": 179, - "sha256": "820931c466afa768c5277d46fc9e638fece5d60cfd58552d0ca07b8a0c1ff522", - "content_sha256": "533be650099bf10f780e09ead8e0a90b53e1bf917b316bf22c3d447f83d659e9" + "count": 180, + "sha256": "e8489acf3396ad149ca9342214e4865124021fd1e05d37ff548e63dcf2f6a242", + "content_sha256": "43402e473a95ebcfa5a7a27c1a7cf0f0a056ffdd9b95ed6fd5f8ade4ddcc2b10" } diff --git a/docs/upstream-sync.md b/docs/upstream-sync.md new file mode 100644 index 000000000..01cbacfc7 --- /dev/null +++ b/docs/upstream-sync.md @@ -0,0 +1,233 @@ +# Upstream Sync Playbook + +How this fork pulls changes from [`rohitg00/agentmemory`](https://github.com/rohitg00/agentmemory) +(`upstream`) into [`ChronodeAi/agentmemory`](https://github.com/ChronodeAi/agentmemory) +(`origin`) using the staged-train process. Every step below has been executed +end-to-end on this machine; follow it in order and do not skip the gates. + +Companion tooling: [`scripts/sync/upstream-status.sh`](../scripts/sync/upstream-status.sh) +prints the current sync position at any point in the process. + +## 0. Refresh refs (explicit refspecs) + +Both remotes pin their refspec. A plain `git fetch origin` does NOT update +`refs/remotes/origin/main` on this checkout — always fetch with explicit +refspecs: + +```sh +git fetch origin refs/heads/main:refs/remotes/origin/main +git fetch upstream refs/heads/main:refs/remotes/upstream/main +``` + +Work on a throwaway branch off `origin/main`, never directly on `main`. + +## 1. Classify the delta + +List what upstream has that the deployed release does not: + +```sh +git log ..upstream/main --oneline +``` + +Sort every commit into exactly one bucket before touching code: + +| Bucket | Criteria | Action | +| --- | --- | --- | +| **pull-first bugfixes** | Small, self-contained fixes in code paths the fork tracks verbatim | Port first, one commit each, minimal adaptation | +| **decompose-and-port** | Upstream changes bundling several unrelated concerns | Split into fork-sized commits; port piece by piece; drop pieces that collide with fork-only subsystems | +| **re-port-through-bundle-pipeline** | Anything whose real artifact is generated output (dist bundles, plugin sidecars, skill manifests) | Never hand-edit artifacts. Apply the source change, run `npm run build`, commit the regenerated bundles | +| **skip-unused-host-adapters** | Deploy-platform adapters the fork does not run (`deploy/coolify`, `deploy/fly`, `deploy/railway`, `deploy/render`, docker-compose variants) | Skip; note the skip in the train summary | + +Record the classification (bucket per upstream sha) in the PR description so +the next train starts from a written baseline instead of archaeology. + +## 2. Non-negotiable gates + +All of these run against the finished tree before anything is pushed. + +### Hook bundles are committed build outputs + +`dist/*.mjs` and `plugin/` hook sidecars are build products. After ANY change +they depend on (including version constants): + +```sh +npm run build +git add dist plugin # commit the regenerated bundles together with the source +``` + +Hand-edited bundles drift silently and ship stale behavior — always rebuild. + +### `ci/r13-test-manifest.json` recompute algorithm + +Adding, removing, or renaming a file under `test/**.test.ts` invalidates the +frozen manifest. Recompute it with the same algorithm `scripts/r13/run.mjs` +uses (sorted tracked test paths; `count`; `sha256` over the newline-joined +path list plus trailing newline; `content_sha256` over `PATH\0\0` + +file bytes per path): + +```sh +node -e ' +import("node:fs").then(async ({readdirSync, readFileSync, writeFileSync}) => { + const {createHash} = await import("node:crypto"); + const {join, relative} = await import("node:path"); + const root = process.cwd(); + const files = []; + const visit = (d) => { for (const e of readdirSync(d,{withFileTypes:true})) { + const p=join(d,e.name); + if(e.isDirectory()) visit(p); + else if(e.isFile()&&e.name.endsWith(".test.ts")) files.push(relative(root,p).replaceAll("\\\\","/")); + } }; + visit(join(root,"test")); + const tests = files.sort(); + const sha = (v)=>createHash("sha256").update(v).digest("hex"); + const content=createHash("sha256"); + for (const p of tests){ content.update(`PATH\0${p}\0`); content.update(readFileSync(join(root,p))); } + writeFileSync(join(root,"ci/r13-test-manifest.json"), + JSON.stringify({count:tests.length, sha256:sha(`${tests.join("\n")}\n`), content_sha256:content.digest("hex")},null,2)+"\n"); +})' +``` + +Committing tests without this recompute fails the canonical R13 preflight +with `test manifest mismatch` / `test content mismatch`. + +### Evidence inventory refresh + +```sh +npm run evidence:interfaces # regenerates .aiwg/reports/g-icm-01-interface-inventory.json +npm run evidence:interfaces:test # unit-checks the generator itself +``` + +Commit the refreshed inventory JSON in the same train. + +### Skills generation and check + +```sh +npm run skills:gen +npm run skills:check +``` + +Commit regenerated skill outputs; `skills:check` must be green at push time. + +### Tool-count consistency surfaces + +Any change that adds/removes MCP tools or REST endpoints MUST touch all +surfaces listed in [`AGENTS.md`](../AGENTS.md) ("Consistency Rules"): +tools-registry, server switch, REST registration, index registration + +endpoint log count, `test/mcp-standalone.test.ts` assertion, README counts, +`plugin/.claude-plugin/plugin.json` description, `plugin/plugin.json` / +`plugin/.mcp.copilot.json`, and the evidence inventory. The interface +inventory gate catches drift, but fix the sources rather than suppressing. + +## 3. Verification battery (pre-push) + +Run locally, in this order; do not push until every line passes: + +1. `npx vitest run` — zero failures (known base file-level artifacts are + exempt; new failures are not). +2. `npx tsc --noEmit 2>&1 | grep "^src/" | sort` — identical to the frozen + baseline set: `transport.ts` socket-union family, `tools-registry.ts` + pattern/items TS2353 ×3+1, `coding-memory.ts` TS6133. Any addition or + removal is a regression. +3. Full canonical R13 with dummy secrets and HF smoke enabled: + + ```sh + AGENTMEMORY_SECRET=dummy \ + AGENTMEMORY_PROJECT_CAPABILITY_SECRET=dummy \ + RUN_HF_SMOKE=1 \ + node scripts/r13/run.mjs + ``` + + Must end with `R-13 passed`. This is the battery; `npm test` alone is not + a substitute for a release train. +4. `npm run evidence:interfaces:test` — pass. +5. `npm run skills:check` — green. + +## 4. PR flow + +- PRs target `origin/main` DIRECTLY. Do not introduce stacked bases: deleting + a stacked base branch auto-closes every PR stacked on it (learned the hard + way on this repo). +- One PR per train. Push, wait ~30 s, then + `gh pr checks --watch --repo ChronodeAi/agentmemory`. On failure read + `gh run view --log-failed`, fix, push again — repeat until green. +- Merge with `gh pr merge --merge --delete-branch`. + +## 5. Deploy recipe (local production) + +Executed after the train's PR merges; `sha12` = first 12 chars of the new +main sha. + +1. **Build and prove reproducibility** (double-pack: two consecutive packs of + the same tree must hash identically): + + ```sh + npm ci && npm run build + npm pack >/dev/null && cp agentmemory-*.tgz /tmp/pack-a.tgz && rm agentmemory-*.tgz + npm pack >/dev/null && cp agentmemory-*.tgz /tmp/pack-b.tgz && rm agentmemory-*.tgz + shasum -a 256 /tmp/pack-a.tgz /tmp/pack-b.tgz # hashes must match + TGZ=/tmp/pack-a.tgz + ``` + +2. **Stage the release dir** `releases/merge-` under + `~/Library/Application Support/Agentmemory/releases/`: + + ```sh + REL="$HOME/Library/Application Support/Agentmemory/releases/merge-" + mkdir -p "$REL/lib/node_modules/@agentmemory/agentmemory" "$REL/bin" + cd "$REL/lib/node_modules/@agentmemory/agentmemory" + # Anchor package.json pitfall: `npm install ./x.tgz` walks UP looking for + # the nearest package.json and can unpack into a parent directory. An + # empty dir has none, so seed one first (or install while already inside + # the package dir like prior releases did). + echo '{"name":"anchor","private":true,"version":"0.0.0"}' > package.json + npm install "$TGZ" --omit=dev + ``` + + Result must mirror the previous release: the package contents land in + `lib/node_modules/@agentmemory/agentmemory/` (dist/, plugin/, iii-config.yaml…). + +3. **Bin symlink**: + + ```sh + ln -sf ../lib/node_modules/@agentmemory/agentmemory/dist/cli.mjs "$REL/bin/agentmemory" + ``` + +4. **Receipt JSON** mirroring `receipts/activation-05b08ffc.json` + (`~/Library/Application Support/Agentmemory/receipts/activation-.json`): + + ```json + { + "schema_version": 1, + "kind": "agentmemory_local_activation_receipt", + "activated_at": "", + "source": { "repository": "ChronodeAi/agentmemory", "commit": "" }, + "package": { "sha256": "", "reproducible": true }, + "supersedes": { "release_dir": "merge-", "version": "" } + } + ``` + +5. **Flip and restart**: + + ```sh + ln -sfn "$REL" "$HOME/Library/Application Support/Agentmemory/current" + launchctl print "gui/$(id -u)/com.chronode.agentmemory" >/dev/null 2>&1 \ + || launchctl bootstrap "gui/$(id -u)" ~/Library/LaunchAgents/com.chronode.agentmemory.plist + launchctl kickstart -k "gui/$(id -u)/com.chronode.agentmemory" + sleep 15 + ``` + +6. **Verify**: + + ```sh + curl -s -H "Authorization: Bearer $(cat ~/.agentmemory/secret)" \ + http://127.0.0.1:3113/health | python3 -c 'import json,sys; d=json.load(sys.stdin); print(d["build"]["backend"], d["status"])' + # expect: agentmemory- healthy + + CLAUDE_PLUGIN_ROOT="$HOME/Library/Application Support/Agentmemory/current/plugin" \ + echo '{}' | node "$HOME/Library/Application Support/Agentmemory/current/plugin/scripts/post-tool-use.mjs" + # then confirm the round-trip landed in the daemon stderr log + tail -n 20 ~/Library/Logs/Agentmemory/agentmemory.stderr.log + ``` + +7. Update the tier-1 decision record (`decisions` table via `dsh-sor.sh`) so + the deployment is queryable truth, not session folklore. diff --git a/package-lock.json b/package-lock.json index 66ce6a27e..5bd4f02e5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@agentmemory/agentmemory", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@agentmemory/agentmemory", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "license": "Apache-2.0", "dependencies": { "@anthropic-ai/claude-agent-sdk": "^0.3.142", diff --git a/package.json b/package.json index f684e50a8..33c894e34 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@agentmemory/agentmemory", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "description": "Persistent memory for AI coding agents, powered by iii-engine's three primitives", "type": "module", "main": "dist/index.mjs", diff --git a/packages/mcp/package.json b/packages/mcp/package.json index fc2860e28..ff919b619 100644 --- a/packages/mcp/package.json +++ b/packages/mcp/package.json @@ -1,6 +1,6 @@ { "name": "@agentmemory/mcp", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "description": "Standalone MCP server for agentmemory — thin shim that re-exposes @agentmemory/agentmemory's MCP entrypoint", "type": "module", "bin": { @@ -28,7 +28,7 @@ "homepage": "https://github.com/rohitg00/agentmemory#readme", "bugs": "https://github.com/rohitg00/agentmemory/issues", "dependencies": { - "@agentmemory/agentmemory": "0.9.30-chronode.4" + "@agentmemory/agentmemory": "0.9.30-chronode.5" }, "publishConfig": { "access": "public", diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json index 2ea59f72a..39dec430f 100644 --- a/plugin/.claude-plugin/plugin.json +++ b/plugin/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentmemory", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "description": "Persistent memory for AI coding agents -- captures tool usage, compresses via LLM, injects context into future sessions. 12 hooks, 60 MCP tools, 17 skills, real-time viewer.", "author": { "name": "Rohit Ghumare", diff --git a/plugin/.codex-plugin/plugin.json b/plugin/.codex-plugin/plugin.json index fd8db19b6..1bf014c32 100644 --- a/plugin/.codex-plugin/plugin.json +++ b/plugin/.codex-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentmemory", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "description": "Persistent memory for AI coding agents -- captures tool usage, compresses via LLM, injects context into future sessions. 11 hooks, 60 MCP tools, 17 skills, real-time viewer.", "author": { "name": "Rohit Ghumare", diff --git a/plugin/plugin.json b/plugin/plugin.json index f22d540bd..3130d3d8d 100644 --- a/plugin/plugin.json +++ b/plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "agentmemory", - "version": "0.9.30-chronode.4", + "version": "0.9.30-chronode.5", "description": "Persistent memory for AI coding agents -- captures tool usage, compresses via LLM, injects context into future sessions. 12 hooks, 60 MCP tools, 17 skills, real-time viewer.", "author": { "name": "Rohit Ghumare", diff --git a/plugin/scripts/standalone.mjs b/plugin/scripts/standalone.mjs index f9de0a1ce..5a7a2cb84 100755 --- a/plugin/scripts/standalone.mjs +++ b/plugin/scripts/standalone.mjs @@ -1761,7 +1761,7 @@ function getAllTools() { } //#endregion //#region src/version.ts -const VERSION = "0.9.30-chronode.4"; +const VERSION = "0.9.30-chronode.5"; process.env["AGENTMEMORY_BUILD_ID"]; process.env["AGENTMEMORY_VIEWER_BUILD_ID"]; //#endregion diff --git a/plugin/skills/agentmemory-config/REFERENCE.md b/plugin/skills/agentmemory-config/REFERENCE.md index a0daf9dcb..4a51bd0b8 100644 --- a/plugin/skills/agentmemory-config/REFERENCE.md +++ b/plugin/skills/agentmemory-config/REFERENCE.md @@ -3,7 +3,7 @@ Generated by scanning `src/` for `AGENTMEMORY_*` usage. Do not edit the block below by hand; run `npm run skills:gen` after adding or removing a variable. Internal markers ending in two underscores are excluded. -Configuration is read from the environment and from `~/.agentmemory/.env` (no `export` prefix). 67 recognized variables: +Configuration is read from the environment and from `~/.agentmemory/.env` (no `export` prefix). 68 recognized variables: - `AGENTMEMORY_ADMIN_SECRET` - `AGENTMEMORY_ADMIN_SECRET_FILE` @@ -37,6 +37,7 @@ Configuration is read from the environment and from `~/.agentmemory/.env` (no `e - `AGENTMEMORY_IMAGE_STORE_MAX_BYTES` - `AGENTMEMORY_INJECT_CONTEXT` - `AGENTMEMORY_LAUNCHD_LABELS` +- `AGENTMEMORY_LIVE_STREAM_MAX_BYTES` - `AGENTMEMORY_LLM_TIMEOUT_MS` - `AGENTMEMORY_LOCAL_PROCESSING` - `AGENTMEMORY_MCP_BLOCK` diff --git a/scripts/sync/upstream-status.sh b/scripts/sync/upstream-status.sh new file mode 100755 index 000000000..d119aeb85 --- /dev/null +++ b/scripts/sync/upstream-status.sh @@ -0,0 +1,54 @@ +#!/bin/bash +# Print the current upstream-sync position: deployed release, local main, +# upstream/main, divergence vs the deployed sha, and the newest upstream +# commit subjects. Read-only; run from anywhere inside the repo. +set -euo pipefail + +root=$(git rev-parse --show-toplevel) +cd "$root" + +app_dir="$HOME/Library/Application Support/Agentmemory" +current_link="$app_dir/current" + +deployed_release="unknown" +deployed_sha12="unknown" +if [ -L "$current_link" ]; then + deployed_release=$(basename "$(readlink "$current_link")") + case "$deployed_release" in + merge-*) + deployed_sha12="${deployed_release#merge-}" + ;; + esac +fi + +main_sha=$(git rev-parse --verify --quiet refs/heads/main || git rev-parse origin/main) + +upstream_sha=unknown +if git show-ref --verify --quiet refs/remotes/upstream/main; then + upstream_sha=$(git rev-parse upstream/main) +else + echo "upstream/main missing — fetch with: git fetch upstream refs/heads/main:refs/remotes/upstream/main" >&2 +fi + +echo "deployed release dir: $deployed_release" +echo "current main sha: $main_sha" +echo "upstream/main sha: $upstream_sha" + +deployed_full="" +if [ "$deployed_sha12" != "unknown" ]; then + deployed_full=$(git rev-parse --verify --quiet "${deployed_sha12}" || true) + if [ -n "$deployed_full" ]; then + read -r behind ahead <<<"$(git rev-list --left-right --count "refs/heads/main...$deployed_full")" + echo "main vs deployed: ahead $ahead, behind $behind (deployed $deployed_full)" + read -r uback uahead <<<"$(git rev-list --left-right --count "upstream/main...$deployed_full")" + echo "upstream vs deployed: ahead $uahead, behind $uback" + else + echo "main vs deployed: deployed sha $deployed_sha12 not found in this clone" + fi +fi + +if [ "$upstream_sha" != "unknown" ]; then + echo + echo "17 newest upstream commits:" + git log upstream/main -n 17 --format=' %h %s' +fi diff --git a/src/functions/compress.ts b/src/functions/compress.ts index 9040f3efe..876cdfe0e 100644 --- a/src/functions/compress.ts +++ b/src/functions/compress.ts @@ -9,6 +9,7 @@ import type { } from "../types.js"; import { KV, STREAM } from "../state/schema.js"; import { StateKV } from "../state/kv.js"; +import { rotateLiveStreamIfOversized } from "../state/live-stream-rotation.js"; import { COMPRESSION_SYSTEM, buildCompressionPrompt, @@ -210,6 +211,7 @@ export function registerCompressFunction( scheduleIndexSave(); } + rotateLiveStreamIfOversized(); const streamResults = await Promise.allSettled([ sdk.trigger({ function_id: "stream::set", diff --git a/src/functions/observe.ts b/src/functions/observe.ts index 0890a7730..78a57651a 100644 --- a/src/functions/observe.ts +++ b/src/functions/observe.ts @@ -17,6 +17,7 @@ const TOOL_HOOKS = new Set([ ]); import { KV, STREAM, generateId } from "../state/schema.js"; import { StateKV } from "../state/kv.js"; +import { rotateLiveStreamIfOversized } from "../state/live-stream-rotation.js"; import { stripPrivateData } from "./privacy.js"; import { DedupMap } from "./dedup.js"; import { withKeyedLock } from "../state/keyed-mutex.js"; @@ -570,6 +571,7 @@ export function registerObserveFunction( }, }); + rotateLiveStreamIfOversized(); await sdk.trigger({ function_id: "stream::send", payload: { @@ -687,6 +689,7 @@ export function registerObserveFunction( data: { type: "compressed", observation: synthetic }, }, }); + rotateLiveStreamIfOversized(); await sdk.trigger({ function_id: "stream::set", payload: { diff --git a/src/state/live-stream-rotation.ts b/src/state/live-stream-rotation.ts new file mode 100644 index 000000000..233a3e8c7 --- /dev/null +++ b/src/state/live-stream-rotation.ts @@ -0,0 +1,92 @@ +import { existsSync, renameSync, rmSync, statSync } from "node:fs"; +import { join } from "node:path"; +import { resolveDataDir } from "../data-dir.js"; +import { logger } from "../logger.js"; +import { STREAM } from "./schema.js"; + +// The viewer live stream (`stream::send`/`stream::set` with group +// STREAM.viewerGroup) persists through iii-stream's file-backed adapter as one +// append-only .bin per stream group and grows without bound. This module caps +// that one file; per-session streams stay small and are not rotated. +export const DEFAULT_LIVE_STREAM_MAX_BYTES = 33_554_432; + +export function resolveLiveStreamMaxBytes( + env: NodeJS.ProcessEnv = process.env, +): number { + const raw = Number(env.AGENTMEMORY_LIVE_STREAM_MAX_BYTES); + if (!Number.isFinite(raw)) return DEFAULT_LIVE_STREAM_MAX_BYTES; + if (raw === 0) return 0; + if (raw < 0) return DEFAULT_LIVE_STREAM_MAX_BYTES; + return Math.floor(raw); +} + +export function viewerLiveStreamPath(dataDir: string = resolveDataDir()): string { + return join( + dataDir, + "stream_store", + `stream%3A${STREAM.name}%3A${STREAM.viewerGroup}.bin`, + ); +} + +export interface LiveStreamRotationOptions { + dataDir?: string; + filePath?: string; + maxBytes?: number; + env?: NodeJS.ProcessEnv; +} + +// Rotates the viewer live stream when its persisted file has outgrown +// AGENTMEMORY_LIVE_STREAM_MAX_BYTES: the current file is renamed to +// `.prev` (overwriting any previous generation) so the engine's next +// append starts a fresh file. The engine opens the store by path per append +// (no long-lived fd), so no engine restart or cooperation is needed. +// +// Best-effort by contract: every failure — missing file, stat error, rename +// race — degrades to "keep appending to the oversized file" and at most one +// warn line, because blocking or failing an observation publish to save disk +// would be the wrong trade. +// +// Under vitest the default path would point at the operator's real store +// (tests share HOME and no engine owns the file there), so default-path +// rotation is disabled in test workers; callers that inject filePath or +// maxBytes bypass the guard so the rotation logic itself stays testable. +export function rotateLiveStreamIfOversized( + options: LiveStreamRotationOptions = {}, +): boolean { + const injected = + options.filePath !== undefined || + options.maxBytes !== undefined || + options.dataDir !== undefined; + if (!injected && process.env.VITEST) return false; + + const maxBytes = options.maxBytes ?? resolveLiveStreamMaxBytes(options.env); + if (maxBytes <= 0) return false; + + const filePath = options.filePath ?? viewerLiveStreamPath(options.dataDir); + + let size: number; + try { + size = statSync(filePath).size; + } catch { + // No persisted stream yet (first boot, fresh data dir) — nothing to + // rotate and nothing to warn about; the engine creates the file on the + // next append. + return false; + } + if (size <= maxBytes) return false; + + try { + const previousPath = `${filePath}.prev`; + if (existsSync(previousPath)) rmSync(previousPath); + renameSync(filePath, previousPath); + return true; + } catch (error) { + logger.warn("live stream rotation failed; keeping oversized stream", { + filePath, + sizeBytes: size, + maxBytes, + error: error instanceof Error ? error.message : String(error), + }); + return false; + } +} diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 3732bf8b3..9bd91715c 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -2,6 +2,7 @@ import { TriggerAction, type ISdk } from "iii-sdk"; import type { CompressedObservation, HookPayload, Session } from "../types.js"; import { KV, STREAM, generateId } from "../state/schema.js"; import { StateKV } from "../state/kv.js"; +import { rotateLiveStreamIfOversized } from "../state/live-stream-rotation.js"; import { isReflectEnabled } from "../functions/slots.js"; import { getAgentId, @@ -998,6 +999,7 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { const newCount = payload.new_value?.observationCount ?? 0; if (newCount <= oldCount) return { skipped: true }; + rotateLiveStreamIfOversized(); await sdk.trigger({ function_id: "stream::send", payload: { diff --git a/src/version.ts b/src/version.ts index 610a477d4..a372c1e2b 100644 --- a/src/version.ts +++ b/src/version.ts @@ -1,4 +1,4 @@ -export const VERSION = "0.9.30-chronode.4"; +export const VERSION = "0.9.30-chronode.5"; export const EXPORT_FORMAT_VERSION = "0.9.28" as const; export const API_CONTRACT_VERSION = 1; export const BACKEND_BUILD_ID = diff --git a/test/live-stream-rotation.test.ts b/test/live-stream-rotation.test.ts new file mode 100644 index 000000000..fad96c781 --- /dev/null +++ b/test/live-stream-rotation.test.ts @@ -0,0 +1,159 @@ +import { describe, it, expect, afterEach } from "vitest"; +import { + existsSync, + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { + DEFAULT_LIVE_STREAM_MAX_BYTES, + resolveLiveStreamMaxBytes, + rotateLiveStreamIfOversized, + viewerLiveStreamPath, +} from "../src/state/live-stream-rotation.js"; + +const VIEWER_STREAM_NAME = "stream%3Amem-live%3Aviewer.bin"; + +let cleanup: Array<() => void> = []; +function tempDataDir(): string { + const dir = mkdtempSync(join(tmpdir(), "agentmemory-live-stream-")); + cleanup.push(() => rmSync(dir, { recursive: true, force: true })); + return dir; +} + +afterEach(() => { + for (const done of cleanup) done(); + cleanup = []; +}); + +describe("DEFAULT_LIVE_STREAM_MAX_BYTES", () => { + it("caps the viewer stream at 32 MiB", () => { + expect(DEFAULT_LIVE_STREAM_MAX_BYTES).toBe(33554432); + }); +}); + +describe("resolveLiveStreamMaxBytes", () => { + it("defaults when unset", () => { + expect(resolveLiveStreamMaxBytes({})).toBe(DEFAULT_LIVE_STREAM_MAX_BYTES); + }); + + it("honors a positive override", () => { + expect(resolveLiveStreamMaxBytes({ AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "4096" })).toBe(4096); + }); + + it("treats zero as an explicit opt-out", () => { + expect(resolveLiveStreamMaxBytes({ AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "0" })).toBe(0); + }); + + it("falls back to the default for invalid values", () => { + expect( + resolveLiveStreamMaxBytes({ AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "not-a-number" }), + ).toBe(DEFAULT_LIVE_STREAM_MAX_BYTES); + expect( + resolveLiveStreamMaxBytes({ AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "-5" }), + ).toBe(DEFAULT_LIVE_STREAM_MAX_BYTES); + expect( + resolveLiveStreamMaxBytes({ AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "1.5" }), + ).toBe(1); + }); +}); + +describe("viewerLiveStreamPath", () => { + it("mirrors the iii-stream file-backed adapter layout", () => { + expect(viewerLiveStreamPath("/data")).toBe(join("/data", "stream_store", VIEWER_STREAM_NAME)); + }); +}); + +describe("rotateLiveStreamIfOversized", () => { + function seedStream(dataDir: string, bytes: number): string { + const store = join(dataDir, "stream_store"); + mkdirSync(store, { recursive: true }); + const filePath = join(store, VIEWER_STREAM_NAME); + writeFileSync(filePath, Buffer.alloc(bytes, 0x61)); + return filePath; + } + + it("rotates once to .prev when the injected cap is tiny", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + + const rotated = rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 }); + + expect(rotated).toBe(true); + expect(existsSync(filePath)).toBe(false); + const previous = readFileSync(`${filePath}.prev`); + expect(previous.length).toBe(64); + }); + + it("starts fresh so the next append recreates the stream file", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 }); + + writeFileSync(filePath, Buffer.alloc(8, 0x62)); + + expect(readFileSync(filePath).length).toBe(8); + expect(existsSync(`${filePath}.prev`)).toBe(true); + }); + + it("keeps the file when it is within the cap", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 16); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(false); + expect(existsSync(filePath)).toBe(true); + expect(existsSync(`${filePath}.prev`)).toBe(false); + }); + + it("overwrites a previous generation instead of failing", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + writeFileSync(`${filePath}.prev`, Buffer.alloc(4, 0x01)); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(true); + expect(readFileSync(`${filePath}.prev`).length).toBe(64); + }); + + it("is a no-op when no stream file exists yet", () => { + const dataDir = tempDataDir(); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(false); + }); + + it("swallows rotation errors and keeps the oversized file", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + // A directory at .prev makes rmSync fail without recursive, which + // forces the rename path into its best-effort catch. + mkdirSync(`${filePath}.prev`); + writeFileSync(join(`${filePath}.prev`, "occupied"), "x"); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(false); + expect(existsSync(filePath)).toBe(true); + }); + + it("triggers from a tiny injected env cap without explicit maxBytes", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 48); + + const rotated = rotateLiveStreamIfOversized({ + dataDir, + env: { AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "16" }, + }); + + expect(rotated).toBe(true); + expect(existsSync(`${filePath}.prev`)).toBe(true); + }); + + it("does nothing at cap zero (explicit opt-out)", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 0 })).toBe(false); + expect(existsSync(`${filePath}.prev`)).toBe(false); + }); +});