fix(stream): rate-limit live-stream rotations to prevent disk thrash

.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"

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 `<path>.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 `<path>.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 `<deployed-sha>..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 <tgz> --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.
 

ci/r13-test-manifest.json

@@ -1,5 +1,5 @@
 {
   "count": 180,
   "sha256": "e8489acf3396ad149ca9342214e4865124021fd1e05d37ff548e63dcf2f6a242",
-  "content_sha256": "480974ca30f7632c133fd0fcd7aff9bb03b99c04783d7f86a005134fb4b5e87e"
+  "content_sha256": "d814f7af2764ee51260a113cdce86f535b3bc7ca1074bafe1b9cdebc9bef7596"
 }

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

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
 // `<path>.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", {

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);
+  });
 });