Files
rohitg00--agentmemory/test/replay-import-key.test.ts
Rohit Ghumare 6cc9b9f0fe fix: env hydration, indexing, consolidation lifecycle, connector activation, hardening (#1136)
* fix: env hydration, indexing, consolidation, connectors, hardening

- config: hydrate ~/.agentmemory/.env into process.env at boot so all modules see it
- search: shared indexRecords() so export-import and replay populate BM25 and vector (#1072)
- snapshot: wire the periodic timer (#1006), clamp non-positive intervals, add a reentrancy guard
- schema: CJK-aware jaccard dedup plus exact-match fallback for short memories
- embeddings: shared resolveDimensions() so openrouter stops hardcoding 1536 (#1002)
- viewer: buffer request bodies before decoding to fix multibyte corruption (#930)
- providers: retry 429/503 with Retry-After under a total-elapsed budget cap
- consolidation: fire on session stop (#1087), gate keyless installs, debounce the per-turn stop hook, drop the client-side double-fire
- evict: bound stale-session recovery to one consolidation pass
- api/patterns: bound session fan-out (#1100)
- connect: write a memory-usage guideline into each hook-less agent's native rules file (12 agents, doc-verified paths, --no-guidelines opt-out)
- graph: import graphify's graph.json via mem::graph::import-graphify + POST /agentmemory/graph/import-graphify; shared persistGraphDelta with endpoint remap so merged nodes never leave dangling or duplicate edges
- fs-watcher: stat roots before fs.watch so missing roots fail deterministically on Node 24+
- test: regression tests for every fix

* fix: address review findings on import, debounce, and connect paths

- guidelines: refuse to touch files with a lone or reversed marker pair
- export-import/replay: indexing after committed writes is best-effort,
  logged instead of failing the import; flatten the nested runChunked so
  replace-mode deletes stay bounded to one chunk
- graph: persist the snapshot when merge-only batches mutate cached
  topNodes/topEdges entries
- graph-import: async fs, typeof validation on path/cwd; REST handler
  whitelists the payload and 400s non-string values
- fetch: cancel discarded response bodies before retrying
- events: serialize the consolidation cooldown check so concurrent stops
  cannot both pass the read-check-write window
- evict: gate recovered-session consolidation on isConsolidationEnabled
  and mirror the stop path's force flag
- search: rebuild indexes per session chunk to bound peak memory
- test: regression coverage for each (malformed markers, concurrent
  stops, snapshot persistence, AMBIGUOUS/default mappings, env isolation)
2026-08-02 11:16:30 +01:00

249 lines
8.1 KiB
TypeScript

import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
import { mkdtempSync, writeFileSync, rmSync } from "node:fs";
import { tmpdir } from "node:os";
import { join } from "node:path";
vi.mock("../src/logger.js", () => ({
logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
}));
import { registerReplayFunctions } from "../src/functions/replay.js";
import { KV } from "../src/state/schema.js";
import {
getSearchIndex,
setVectorIndex,
setEmbeddingProvider,
} from "../src/functions/search.js";
import { VectorIndex } from "../src/state/vector-index.js";
import type { EmbeddingProvider } from "../src/types.js";
function mockKV() {
const store = new Map<string, Map<string, unknown>>();
const setCalls: Array<{ scope: string; key: string | undefined; value: any }> = [];
return {
get: async <T>(scope: string, key: string): Promise<T | null> =>
(store.get(scope)?.get(key) as T) ?? null,
set: async <T>(scope: string, key: string, value: T): Promise<T> => {
setCalls.push({ scope, key, value });
if (!store.has(scope)) store.set(scope, new Map());
// Mirror the engine: a state::set with key=undefined fails. We
// surface this via setCalls so the test can assert key !== undefined.
if (key === undefined) {
throw new Error("missing field `key`");
}
store.get(scope)!.set(key, value);
return value;
},
delete: async (scope: string, key: string) => {
store.get(scope)?.delete(key);
},
list: async <T>(scope: string): Promise<T[]> =>
Array.from(store.get(scope)?.values() ?? []) as T[],
getSetCalls: () => setCalls,
};
}
function mockSdk(kv: ReturnType<typeof mockKV>) {
const fns = new Map<string, Function>();
return {
registerFunction: (id: string, handler: Function) => fns.set(id, handler),
registerTrigger: () => {},
trigger: async (
idOrInput: string | { function_id: string; payload?: unknown },
data?: unknown,
) => {
const id =
typeof idOrInput === "string" ? idOrInput : idOrInput.function_id;
const payload =
typeof idOrInput === "string" ? data : (idOrInput as any).payload;
const fn = fns.get(id);
if (!fn) return { success: true };
return fn(payload);
},
_kv: kv,
} as any;
}
describe("import-jsonl re-key on parsed.sessionId (#775)", () => {
let tmpRoot: string;
beforeEach(() => {
tmpRoot = mkdtempSync(join(tmpdir(), "replay-import-key-"));
});
function writeFixture(sessionId: string, ts = "2026-04-17T10:00:00.000Z") {
const dir = join(tmpRoot, "proj");
rmSync(dir, { recursive: true, force: true });
require("node:fs").mkdirSync(dir, { recursive: true });
const lines = [
JSON.stringify({
type: "user",
uuid: "u1",
sessionId,
timestamp: ts,
cwd: tmpRoot,
message: {
role: "user",
content: [{ type: "text", text: "hello" }],
},
}),
JSON.stringify({
type: "assistant",
uuid: "a1",
sessionId,
timestamp: ts,
message: {
role: "assistant",
content: [{ type: "text", text: "world" }],
},
}),
];
writeFileSync(join(dir, `${sessionId}.jsonl`), lines.join("\n") + "\n");
}
it("re-imports a session whose stored row is missing the `id` field without aborting the batch", async () => {
writeFixture("sess-no-id");
const kv = mockKV();
const sdk = mockSdk(kv);
registerReplayFunctions(sdk, kv as never);
// Seed an existing session row that is MISSING `id` — the
// pre-fix code would re-key on `existing.id` (undefined) and
// throw `missing field \`key\``, aborting the whole import.
await kv.set(KV.sessions, "sess-no-id", {
project: "proj",
cwd: tmpRoot,
startedAt: "2026-04-17T09:00:00Z",
endedAt: "2026-04-17T09:30:00Z",
status: "completed",
observationCount: 2,
tags: [],
});
const result = (await sdk.trigger("mem::replay::import-jsonl", {
path: tmpRoot,
})) as { success: boolean; imported?: number; error?: string };
expect(result.success).toBe(true);
expect(result.imported).toBe(1);
const undefinedKeyWrites = kv
.getSetCalls()
.filter((c) => c.scope === KV.sessions && c.key === undefined);
expect(undefinedKeyWrites.length).toBe(0);
const sessionWrites = kv
.getSetCalls()
.filter((c) => c.scope === KV.sessions && c.key === "sess-no-id");
expect(sessionWrites.length).toBeGreaterThan(0);
// The handler also backfills the missing id field so future reads
// are well-formed.
expect((sessionWrites.at(-1)!.value as any).id).toBe("sess-no-id");
});
it("fresh import (no existing row) still writes session keyed by parsed.sessionId", async () => {
writeFixture("sess-fresh");
const kv = mockKV();
const sdk = mockSdk(kv);
registerReplayFunctions(sdk, kv as never);
const result = (await sdk.trigger("mem::replay::import-jsonl", {
path: tmpRoot,
})) as { success: boolean; imported?: number };
expect(result.success).toBe(true);
expect(result.imported).toBe(1);
const sessionWrites = kv
.getSetCalls()
.filter((c) => c.scope === KV.sessions && c.key === "sess-fresh");
expect(sessionWrites.length).toBe(1);
});
});
describe("import-jsonl indexes observations into BM25 AND vector", () => {
const mockEmbedder: EmbeddingProvider = {
name: "test",
dimensions: 3,
embed: async () => new Float32Array([0.1, 0.2, 0.3]),
embedBatch: async (texts: string[]) =>
texts.map(() => new Float32Array([0.1, 0.2, 0.3])),
};
let tmpRoot: string;
let vectorIndex: VectorIndex;
beforeEach(() => {
tmpRoot = mkdtempSync(join(tmpdir(), "replay-import-index-"));
getSearchIndex().clear();
vectorIndex = new VectorIndex();
setVectorIndex(vectorIndex);
setEmbeddingProvider(mockEmbedder);
});
afterEach(() => {
setVectorIndex(null);
setEmbeddingProvider(null);
getSearchIndex().clear();
});
function writeFixture(sessionId: string) {
const dir = join(tmpRoot, "proj");
require("node:fs").mkdirSync(dir, { recursive: true });
const ts = "2026-04-17T10:00:00.000Z";
const lines = [
JSON.stringify({
type: "user",
uuid: "u1",
sessionId,
timestamp: ts,
cwd: tmpRoot,
message: { role: "user", content: [{ type: "text", text: "how do I fix the flaky retry" }] },
}),
JSON.stringify({
type: "assistant",
uuid: "a1",
sessionId,
timestamp: ts,
message: { role: "assistant", content: [{ type: "text", text: "await the fetch before asserting" }] },
}),
];
writeFileSync(join(dir, `${sessionId}.jsonl`), lines.join("\n") + "\n");
}
it("populates the vector index (regression: replay used to BM25-add only, leaving imports unsearchable by meaning)", async () => {
writeFixture("sess-index");
const kv = mockKV();
const sdk = mockSdk(kv);
registerReplayFunctions(sdk, kv as never);
expect(vectorIndex.size).toBe(0);
expect(getSearchIndex().size).toBe(0);
const result = (await sdk.trigger("mem::replay::import-jsonl", {
path: tmpRoot,
})) as { success: boolean; imported?: number };
expect(result.success).toBe(true);
// Both lanes must be populated. The old code left vectorIndex at 0.
expect(vectorIndex.size).toBeGreaterThan(0);
expect(getSearchIndex().size).toBeGreaterThan(0);
expect(vectorIndex.size).toBe(getSearchIndex().size);
});
it("skips the vector lane cleanly when no embedding provider is configured (keyless install)", async () => {
setVectorIndex(null);
setEmbeddingProvider(null);
writeFixture("sess-keyless");
const kv = mockKV();
const sdk = mockSdk(kv);
registerReplayFunctions(sdk, kv as never);
const result = (await sdk.trigger("mem::replay::import-jsonl", {
path: tmpRoot,
})) as { success: boolean };
// BM25 still works; no crash from the absent vector index.
expect(result.success).toBe(true);
expect(getSearchIndex().size).toBeGreaterThan(0);
});
});