diff --git a/.aiwg/reports/g-icm-01-interface-inventory.json b/.aiwg/reports/g-icm-01-interface-inventory.json index 306d3e788..50a052798 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": "260beb4a954eca7a71019b89f8e09df8c65c10bb", - "commit_tree_sha": "1b692f290d11ae470cb2009156266d6de039ebe7", - "inventory_input_sha256": "8091d93e47a45cd4cb2d4c38cab6448af9186743c84d73406ed2ea2a269fc51e" + "commit_sha": "b244872f43e6c98363d864305f5fcd594d9d1e31", + "commit_tree_sha": "4a0008a80fb82dc38ad719f30b452c969b09acb6", + "inventory_input_sha256": "3e1807639308f135d8310debd30c1836fc8b33b25e377199752884e9b051645c" }, "public_route_allowlist": [ "GET /agentmemory/livez" diff --git a/CHANGELOG.md b/CHANGELOG.md index 336e7d937..4432e4c25 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ Backlog-closure release: live-stream disk bounding, deployment watchdog v2, and ### 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. +- **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. A 10-minute cooldown between rotations keeps the publish path from thrashing when the engine re-materializes its full in-memory stream into the fresh file (observed on a live deployment). 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. diff --git a/ci/r13-test-manifest.json b/ci/r13-test-manifest.json index 9618f7fb6..003010a3f 100644 --- a/ci/r13-test-manifest.json +++ b/ci/r13-test-manifest.json @@ -1,5 +1,5 @@ { "count": 180, "sha256": "e8489acf3396ad149ca9342214e4865124021fd1e05d37ff548e63dcf2f6a242", - "content_sha256": "480974ca30f7632c133fd0fcd7aff9bb03b99c04783d7f86a005134fb4b5e87e" + "content_sha256": "d814f7af2764ee51260a113cdce86f535b3bc7ca1074bafe1b9cdebc9bef7596" } diff --git a/docs/upstream-sync.md b/docs/upstream-sync.md index 01cbacfc7..4f5153a35 100644 --- a/docs/upstream-sync.md +++ b/docs/upstream-sync.md @@ -161,7 +161,7 @@ main sha. the same tree must hash identically): ```sh - npm ci && npm run build + npm ci --legacy-peer-deps --no-audit --no-fund && 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 diff --git a/src/state/live-stream-rotation.ts b/src/state/live-stream-rotation.ts index 03b415735..1a2f75f78 100644 --- a/src/state/live-stream-rotation.ts +++ b/src/state/live-stream-rotation.ts @@ -36,8 +36,18 @@ export interface LiveStreamRotationOptions { filePath?: string; maxBytes?: number; env?: NodeJS.ProcessEnv; + nowMs?: number; } +// Minimum interval between two rotations. Without it a rotation can thrash: +// the engine keeps the full stream in memory, so its first save after a +// rotation re-materializes the entire history into the fresh file — if that +// still exceeds the cap (it does until the engine restarts or trims), every +// subsequent publish would rename and rewrite ~cap bytes forever. One +// rotation per cooldown window bounds disk without the storm. +export const ROTATION_COOLDOWN_MS = 10 * 60_000; +let lastRotationAtMs = 0; + // 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 @@ -78,10 +88,14 @@ export function rotateLiveStreamIfOversized( } if (size <= maxBytes) return false; + const nowMs = options.nowMs ?? Date.now(); + if (nowMs - lastRotationAtMs < ROTATION_COOLDOWN_MS) return false; + try { const previousPath = `${filePath}.prev`; if (existsSync(previousPath)) rmSync(previousPath); renameSync(filePath, previousPath); + lastRotationAtMs = nowMs; return true; } catch (error) { logger.warn("live stream rotation failed; keeping oversized stream", { diff --git a/test/live-stream-rotation.test.ts b/test/live-stream-rotation.test.ts index 7e2551926..91a32cc3a 100644 --- a/test/live-stream-rotation.test.ts +++ b/test/live-stream-rotation.test.ts @@ -1,4 +1,4 @@ -import { describe, it, expect, afterEach } from "vitest"; +import { describe, it, expect, afterEach, beforeEach } from "vitest"; import { existsSync, mkdirSync, @@ -11,6 +11,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { DEFAULT_LIVE_STREAM_MAX_BYTES, + ROTATION_COOLDOWN_MS, resolveLiveStreamMaxBytes, rotateLiveStreamIfOversized, viewerLiveStreamPath, @@ -71,6 +72,12 @@ describe("viewerLiveStreamPath", () => { }); describe("rotateLiveStreamIfOversized", () => { + let clock = 1_700_000_000_000; + + beforeEach(() => { + clock += ROTATION_COOLDOWN_MS * 2; + }); + function seedStream(dataDir: string, bytes: number): string { const store = join(dataDir, "data", "stream_store"); mkdirSync(store, { recursive: true }); @@ -83,7 +90,7 @@ describe("rotateLiveStreamIfOversized", () => { const dataDir = tempDataDir(); const filePath = seedStream(dataDir, 64); - const rotated = rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 }); + const rotated = rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock }); expect(rotated).toBe(true); expect(existsSync(filePath)).toBe(false); @@ -94,7 +101,7 @@ describe("rotateLiveStreamIfOversized", () => { it("starts fresh so the next append recreates the stream file", () => { const dataDir = tempDataDir(); const filePath = seedStream(dataDir, 64); - rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 }); + rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock }); writeFileSync(filePath, Buffer.alloc(8, 0x62)); @@ -106,7 +113,7 @@ describe("rotateLiveStreamIfOversized", () => { const dataDir = tempDataDir(); const filePath = seedStream(dataDir, 16); - expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(false); + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock })).toBe(false); expect(existsSync(filePath)).toBe(true); expect(existsSync(`${filePath}.prev`)).toBe(false); }); @@ -116,14 +123,14 @@ describe("rotateLiveStreamIfOversized", () => { const filePath = seedStream(dataDir, 64); writeFileSync(`${filePath}.prev`, Buffer.alloc(4, 0x01)); - expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(true); + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock })).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); + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock })).toBe(false); }); it("swallows rotation errors and keeps the oversized file", () => { @@ -134,7 +141,7 @@ describe("rotateLiveStreamIfOversized", () => { mkdirSync(`${filePath}.prev`); writeFileSync(join(`${filePath}.prev`, "occupied"), "x"); - expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32 })).toBe(false); + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock })).toBe(false); expect(existsSync(filePath)).toBe(true); }); @@ -144,6 +151,7 @@ describe("rotateLiveStreamIfOversized", () => { const rotated = rotateLiveStreamIfOversized({ dataDir, + nowMs: clock, env: { AGENTMEMORY_LIVE_STREAM_MAX_BYTES: "16" }, }); @@ -155,7 +163,34 @@ describe("rotateLiveStreamIfOversized", () => { const dataDir = tempDataDir(); const filePath = seedStream(dataDir, 64); - expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 0 })).toBe(false); + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 0, nowMs: clock })).toBe(false); expect(existsSync(`${filePath}.prev`)).toBe(false); }); + + it("suppresses a second rotation within the cooldown window", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock })).toBe(true); + writeFileSync(filePath, Buffer.alloc(64, 0x63)); + + expect( + rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock + ROTATION_COOLDOWN_MS - 1 }), + ).toBe(false); + expect(existsSync(`${filePath}.prev`)).toBe(true); + }); + + it("rotates again once the cooldown window has passed", () => { + const dataDir = tempDataDir(); + const filePath = seedStream(dataDir, 64); + + expect(rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock })).toBe(true); + writeFileSync(filePath, Buffer.alloc(64, 0x63)); + + expect( + rotateLiveStreamIfOversized({ dataDir, maxBytes: 32, nowMs: clock + ROTATION_COOLDOWN_MS }), + ).toBe(true); + expect(existsSync(filePath)).toBe(false); + expect(readFileSync(`${filePath}.prev`).length).toBe(64); + }); });