From 3005b67f22abd9cbbb719680e637aebb84b88053 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 14:16:14 +0800 Subject: [PATCH 01/15] docs(deploy): correct the store figure cited for the cold-start budget Two pushed commits on this branch state that the deployment carries "1798 files and roughly 350 MB" under /data: 54225a0 build(railway): give a cold engine room to accept the worker 960c8ad build(railway): raise the healthcheck timeout for a cold store That is wrong by about 7x. Measured from the container on 2026-08-27 at 05:39Z, /data held 2,421 files across 2,363 MB. A pushed commit message cannot be edited, so the corrected figure goes into deploy/README.md, which both of those bodies already send the reader to for the fresh-store baseline. The figure was already refutable from this branch's own log when it was written. 6a27bca, a day earlier, records the same store at 1013 MB. Nothing else in those two bodies depended on it. The 63 to 67 seconds from container start to "Worker registered" against a 60 second window was timed from the container logs, not derived from store size, and the conclusion (raise healthcheckTimeout to Railway's default of 300) is unchanged. The cold read argument is stronger at 2,363 MB than at 350 MB, but this commit claims only what was measured. The same paragraph in deploy/README.md still described healthcheckTimeout as 60 s, which 54225a0 had already changed to 300 in deploy/railway/railway.json. Correcting one wrong number there while leaving a second one two lines below would defeat the point, so both land together. --- deploy/README.md | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/deploy/README.md b/deploy/README.md index a84f653aa..94f6dfecb 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -90,7 +90,14 @@ agentmemory worker reg : 2.0 s healthcheck passes : ~9-10 s ``` -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. +**Those numbers are for a fresh store.** Startup cost scales with what is +under `/data`: the worker reads the store before `/agentmemory/livez` has a +route to answer on, so the healthcheck window has to cover that read. On a +store of 2,363 MB across 2,421 files, the worker took 63 to 67 s to register. + +Railway's `healthcheckTimeout` is set to 300 s in +`deploy/railway/railway.json`, which is Railway's own default, so a cold +start on a grown store has room. Every other template's health-check +`grace_period` (or compose `start_period`) is set to 30 s, a 3x margin on +the fresh-store figure. Tune against your own store rather than against the +table above. From 562d61ec8c708aca8c3f08cbb722b7b6aaff5bf8 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 14:21:10 +0800 Subject: [PATCH 02/15] docs(deploy): decouple the store size from the startup timing 3005b67 wrote the two figures as one observation: "On a store of 2,363 MB across 2,421 files, the worker took 63 to 67 s to register." That pairing was never measured. The timing was taken during the healthcheck failures behind 960c8ad and 54225a0, in the early hours of 08-27. The store figure was measured at 05:39Z, hours after the last of those failures. 6a27bca records the same store at 1013 MB the previous day, so it more than doubled across that window. Both numbers are point samples of a quantity that was moving between them, and the store size at the moment of the timing was never read. State them as two separate observations and say the store was still growing, so a reader cannot take the size as the condition the timing was measured under. This is the defect the correction in 3005b67 exists to prevent, reached by a different route: a composite claim is as discountable as a wrong one. --- deploy/README.md | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/deploy/README.md b/deploy/README.md index 94f6dfecb..d2ef7141b 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -92,8 +92,11 @@ healthcheck passes : ~9-10 s **Those numbers are for a fresh store.** Startup cost scales with what is under `/data`: the worker reads the store before `/agentmemory/livez` has a -route to answer on, so the healthcheck window has to cover that read. On a -store of 2,363 MB across 2,421 files, the worker took 63 to 67 s to register. +route to answer on, so the healthcheck window has to cover that read. On +agentmemory's own production deployment, container start to `Worker +registered` has been measured at 63 to 67 s. That store was measured +separately, at 2,363 MB across 2,421 files on 2026-08-27, and it was growing +across the same period, so the two are not a matched pair. Railway's `healthcheckTimeout` is set to 300 s in `deploy/railway/railway.json`, which is Railway's own default, so a cold From be82d39dc21397e7b47f9337b0f701ed2d272246 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 17:39:21 +0800 Subject: [PATCH 03/15] perf(graph): extract only observations captured since the last extract Claude Code's per-turn Stop hook POSTs /agentmemory/session/end, so event::session::stopped runs on every agent turn, not once per session. It handed mem::graph-extract the session's entire compressed observation list each time. Downstream, persistGraphDelta takes the MERGE path for every node and edge it has already seen (kv.get name-index + kv.get nodes + kv.set per node, and the same three per edge), so turn N re-merged everything from turns 1..N-1. Measured on the new harness: eight turns of one observation each dispatched 36 observations before this change (1+2+...+8) and 8 after. The growth was quadratic in turn count, and per #843 nothing written through kv.set is ever evicted from the iii engine, so that was quadratic permanent heap walking the engine toward its cap. The session record now carries a watermark. Two optional fields rather than one, because a bare max-timestamp watermark is not safe here: observe.ts dispatches mem::compress fire-and-forget and compress.ts stamps the capture time, not the write time, so a slow compression lands an older timestamp after a newer one was already extracted. The count is the tripwire. When the number of observations newer than the watermark does not exactly account for the growth since the watermark was recorded, something arrived out of order (or was evicted) and the whole session is re-sent. Losing a memory is worse than re-merging one. Both fields are optional, so existing session records stay valid and simply take the full-extract path once. The watermark advances only after mem::graph-extract's dispatch is accepted, so a hand-off that never left is retried on the next turn. fireVoid now resolves to a dispatch-success boolean to make that observable; the other callers ignore the value exactly as before. Limitations: - Completion is not observable through TriggerAction.Void(). An extract whose dispatch is accepted but which then fails downstream leaves its delta out of the graph until someone POSTs /agentmemory/graph/build. Advancing on completion would mean awaiting LLM work inside the stop handler, which is the latency this trigger exists to avoid. - How often the out-of-order fallback fires in a real deployment is not measured. Compression latency scales with payload size, so timestamp inversions may be common rather than rare. When the fallback fires it does exactly what the pre-fix code did on every turn, so this is never worse than the status quo, but the saving on a busy session is unquantified. - Interleaving is handled by construction, not by a lock: the (timestamp, count) pair is only ever written as a matched snapshot from a single run, and a pair that does not reconcile degrades to a full extract. A stale pair costs a re-merge; it cannot skip an observation. Ten tests added. Five fail against unmodified source, including the 36-vs-8 dispatch count. The five that pass by construction cover the fallback branch, whose correct behaviour is the pre-fix behaviour; they exist as mutation coverage and kill removing the count tripwire, making the comparison inclusive, dropping the empty-batch guard, and skipping instead of falling back. --- src/triggers/events.ts | 57 ++++- src/types.ts | 9 + test/graph-extract-incremental.test.ts | 296 +++++++++++++++++++++++++ 3 files changed, 358 insertions(+), 4 deletions(-) create mode 100644 test/graph-extract-incremental.test.ts diff --git a/src/triggers/events.ts b/src/triggers/events.ts index bbf15db33..6d9d1fe5d 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -95,15 +95,20 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { sdk.registerFunction("event::session::stopped", async (data: { sessionId: string; skipConsolidation?: boolean }) => { const summary = await sdk.trigger({ function_id: "mem::summarize", payload: data }); + // Resolves true when the trigger was accepted, false when the dispatch + // itself failed. Callers that need to know (graph-extract's watermark) + // read it; the rest ignore it exactly as before. const fireVoid = (function_id: string, payload: unknown) => sdk .trigger({ function_id, payload, action: TriggerAction.Void() }) - .catch((err) => + .then(() => true) + .catch((err) => { logger.warn(function_id + " trigger failed", { sessionId: data.sessionId, error: err instanceof Error ? err.message : String(err), - }), - ); + }); + return false; + }); if (isReflectEnabled()) { fireVoid("mem::slot-reflect", { sessionId: data.sessionId }); } @@ -114,7 +119,51 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { ); const compressed = observations.filter((o) => o.title); if (compressed.length > 0) { - fireVoid("mem::graph-extract", { observations: compressed }); + // /session/end is posted by the per-turn Stop hook, so this handler + // runs every agent turn. Re-sending the whole session each time makes + // persistGraphDelta re-merge turns 1..N-1 on turn N — quadratic engine + // calls, and per #843 every kv.set stays resident in the engine, so + // that is quadratic permanent heap. Send only what landed since the + // last extract. + // + // The count is what makes the timestamp watermark safe. mem::compress + // is dispatched fire-and-forget (observe.ts) and stamps the capture + // time, not the write time, so a slow compression can land an OLDER + // timestamp after a newer one was already extracted. When the number + // of observations newer than the watermark does not exactly account + // for the growth since the watermark was recorded, something arrived + // out of order (or was evicted) and we re-send the whole session + // rather than skip it. Missing a memory is worse than re-merging one. + const session = await kv + .get(KV.sessions, data.sessionId) + .catch(() => null); + const at = session?.graphExtractedAt; + const count = session?.graphExtractedCount; + let batch = compressed; + if (typeof at === "string" && typeof count === "number") { + const fresh = compressed.filter((o) => o.timestamp > at); + if (fresh.length === compressed.length - count) batch = fresh; + } + if (batch.length > 0) { + const newest = compressed.reduce( + (max, o) => (o.timestamp > max ? o.timestamp : max), + "", + ); + // Advance only after the dispatch is accepted, so a hand-off that + // never left retries on the next turn. Completion is unobservable + // through TriggerAction.Void(); an extract that fails downstream + // leaves its delta out of the graph until POST /agentmemory/graph/build. + if (await fireVoid("mem::graph-extract", { observations: batch })) { + await kv.update(KV.sessions, data.sessionId, [ + { type: "set", path: "graphExtractedAt", value: newest }, + { + type: "set", + path: "graphExtractedCount", + value: compressed.length, + }, + ]); + } + } } } catch (err) { logger.warn("graph-extract trigger failed", { diff --git a/src/types.ts b/src/types.ts index d8543c13a..8221e8328 100644 --- a/src/types.ts +++ b/src/types.ts @@ -12,6 +12,15 @@ export interface Session { summary?: string; commitShas?: string[]; agentId?: string; + // Incremental graph-extract watermark. Written as a matched pair from a + // single observation-list snapshot: `graphExtractedAt` is the newest + // observation timestamp handed to mem::graph-extract, `graphExtractedCount` + // is how many compressed observations existed at that moment. The count is + // the tripwire for out-of-order arrivals — mem::compress is dispatched + // fire-and-forget, so observations do not land in KV in timestamp order. + // Absent on session records written before this field existed. + graphExtractedAt?: string; + graphExtractedCount?: number; } export interface CommitLink { diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts new file mode 100644 index 000000000..f2cd07e2d --- /dev/null +++ b/test/graph-extract-incremental.test.ts @@ -0,0 +1,296 @@ +import { describe, it, expect, beforeEach, vi } from "vitest"; +import type { CompressedObservation } from "../src/types.js"; + +vi.mock("../src/logger.js", () => ({ + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() }, +})); + +vi.mock("../src/config.js", () => ({ + getAgentId: vi.fn(() => undefined), + isConsolidationEnabled: vi.fn(() => false), + getConsolidationCooldownMs: vi.fn(() => 300000), +})); + +vi.mock("../src/functions/slots.js", () => ({ + isReflectEnabled: vi.fn(() => false), +})); + +import { registerEventTriggers } from "../src/triggers/events.js"; + +// /agentmemory/session/end is posted by Claude Code's per-turn Stop hook, so +// event::session::stopped runs on EVERY agent turn — not once per session. +// It used to hand mem::graph-extract the session's entire observation list +// each time, and persistGraphDelta takes the 3-call MERGE path for every node +// and edge it has already seen, so turn N re-merged turns 1..N-1. Engine +// invocations were quadratic in turn count, and per #843 nothing written via +// kv.set is ever evicted from the iii engine, so that was quadratic permanent +// heap. These tests pin the extract to the observations captured since the +// last successful extract. + +const SID = "ses_1"; + +function obs(id: string, timestamp: string): CompressedObservation { + return { + id, + sessionId: SID, + timestamp, + type: "conversation", + title: id, + facts: [], + narrative: id, + concepts: [], + files: [], + importance: 0.5, + }; +} + +// A KV that persists writes, so the watermark survives between simulated +// per-turn stops the way the real session record does. +function persistentKV() { + const store = new Map>(); + const scope = (s: string) => { + if (!store.has(s)) store.set(s, new Map()); + return store.get(s)!; + }; + return { + store, + scope, + get: vi.fn(async (s: string, k: string) => scope(s).get(k) ?? null), + set: vi.fn(async (s: string, k: string, v: unknown) => { + scope(s).set(k, v); + return v; + }), + delete: vi.fn(async (s: string, k: string) => { + scope(s).delete(k); + }), + update: vi.fn( + async ( + s: string, + k: string, + ops: Array<{ type: string; path: string; value?: unknown }>, + ) => { + const cur = (scope(s).get(k) ?? {}) as Record; + for (const op of ops) if (op.type === "set") cur[op.path] = op.value; + scope(s).set(k, cur); + return cur; + }, + ), + list: vi.fn(async (s: string) => [...scope(s).values()]), + }; +} + +type StoppedHandler = (data: { + sessionId: string; + skipConsolidation?: boolean; +}) => Promise; + +function mockSdk(opts?: { rejectGraphExtract?: () => boolean }) { + const handlers = new Map(); + const trigger = vi.fn( + async (input: { function_id: string; payload?: unknown }) => { + if ( + input.function_id === "mem::graph-extract" && + opts?.rejectGraphExtract?.() + ) { + throw new Error("dispatch refused"); + } + if (input.function_id === "mem::summarize") { + return { summary: "s", sessionId: SID }; + } + return { ok: true }; + }, + ); + return { + sdk: { + registerFunction: (id: string, h: StoppedHandler) => handlers.set(id, h), + registerTrigger: () => {}, + trigger, + }, + handlers, + trigger, + }; +} + +// Every batch handed to mem::graph-extract, as arrays of observation ids. +function batches(trigger: ReturnType): string[][] { + return trigger.mock.calls + .filter((c) => (c[0] as { function_id: string }).function_id === "mem::graph-extract") + .map((c) => + ( + (c[0] as { payload: { observations: CompressedObservation[] } }).payload + .observations ?? [] + ).map((o) => o.id), + ); +} + +function harness(opts?: { rejectGraphExtract?: () => boolean }) { + const kv = persistentKV(); + const { sdk, handlers, trigger } = mockSdk(opts); + registerEventTriggers(sdk as never, kv as never); + const stopped = handlers.get("event::session::stopped")!; + kv.scope("mem:sessions").set(SID, { + id: SID, + project: "p", + cwd: "/p", + startedAt: "2026-01-01T00:00:00.000Z", + status: "active", + observationCount: 0, + }); + const land = (...os: CompressedObservation[]) => { + for (const o of os) kv.scope(`mem:obs:${SID}`).set(o.id, o); + }; + const stop = () => stopped({ sessionId: SID }); + const session = () => + kv.scope("mem:sessions").get(SID) as Record; + return { kv, trigger, stop, land, session }; +} + +describe("event::session::stopped graph-extract is incremental", () => { + beforeEach(() => vi.clearAllMocks()); + + it("hands each turn only the observations captured since the last extract", async () => { + const h = harness(); + + h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); + await h.stop(); + h.land(obs("c", "2026-01-01T00:00:03.000Z")); + await h.stop(); + h.land(obs("d", "2026-01-01T00:00:04.000Z")); + await h.stop(); + + expect(batches(h.trigger)).toEqual([["a", "b"], ["c"], ["d"]]); + }); + + it("keeps total dispatched observations linear in turn count, not quadratic", async () => { + // Pre-fix, 8 turns of one observation each dispatched 1+2+…+8 = 36 + // observations (and 36 merge passes downstream). Incremental sends 8. + const h = harness(); + for (let i = 1; i <= 8; i++) { + h.land(obs(`o${i}`, `2026-01-01T00:00:0${i}.000Z`)); + await h.stop(); + } + const dispatched = batches(h.trigger).reduce((n, b) => n + b.length, 0); + expect(dispatched).toBe(8); + }); + + it("does not dispatch graph-extract at all when no new observations landed", async () => { + const h = harness(); + h.land(obs("a", "2026-01-01T00:00:01.000Z")); + await h.stop(); + await h.stop(); + await h.stop(); + + expect(batches(h.trigger)).toEqual([["a"]]); + }); + + it("records the watermark as a matched (timestamp, count) pair on the session", async () => { + const h = harness(); + h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); + await h.stop(); + + expect(h.session()).toMatchObject({ + graphExtractedAt: "2026-01-01T00:00:02.000Z", + graphExtractedCount: 2, + }); + }); +}); + +// The fallback branch. Most of these assert the PRE-FIX behaviour (extract +// everything) on purpose, so they pass against unmodified source by +// construction — their job is to kill mutations that would make the watermark +// silently skip an observation. +describe("graph-extract watermark never skips an observation", () => { + beforeEach(() => vi.clearAllMocks()); + + it("extracts the whole session on the first stop, when no watermark exists", async () => { + const h = harness(); + h.land( + obs("a", "2026-01-01T00:00:01.000Z"), + obs("b", "2026-01-01T00:00:02.000Z"), + obs("c", "2026-01-01T00:00:03.000Z"), + ); + await h.stop(); + expect(batches(h.trigger)).toEqual([["a", "b", "c"]]); + }); + + it("re-extracts everything when an observation lands out of order below the watermark", async () => { + // mem::compress is fire-and-forget, so a slow compression writes an older + // timestamp after a newer one was already extracted. Without the count + // tripwire that observation would never reach the graph. + const h = harness(); + h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("c", "2026-01-01T00:00:03.000Z")); + await h.stop(); + + h.land(obs("b", "2026-01-01T00:00:02.000Z"), obs("d", "2026-01-01T00:00:04.000Z")); + await h.stop(); + + expect(batches(h.trigger)[1]).toEqual(["a", "c", "b", "d"]); + }); + + it("re-extracts everything when two observations share the watermark timestamp", async () => { + const h = harness(); + h.land(obs("a", "2026-01-01T00:00:01.000Z")); + await h.stop(); + h.land(obs("b", "2026-01-01T00:00:01.000Z")); + await h.stop(); + + expect(batches(h.trigger)[1]).toEqual(["a", "b"]); + }); + + it("falls back to the whole session when the watermark is half-written", async () => { + const h = harness(); + h.kv.scope("mem:sessions").set(SID, { + id: SID, + project: "p", + cwd: "/p", + startedAt: "2026-01-01T00:00:00.000Z", + status: "active", + observationCount: 0, + graphExtractedAt: "2026-01-01T00:00:01.000Z", + // graphExtractedCount deliberately missing + }); + h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); + await h.stop(); + + expect(batches(h.trigger)).toEqual([["a", "b"]]); + }); + + it("falls back to the whole session when the stored pair is inconsistent", async () => { + // Interleaved stops only ever write (at, count) as a matched snapshot from + // one run, but a torn or stale pair must still degrade to a full extract + // rather than skip anything. + const h = harness(); + h.kv.scope("mem:sessions").set(SID, { + id: SID, + project: "p", + cwd: "/p", + startedAt: "2026-01-01T00:00:00.000Z", + status: "active", + observationCount: 0, + graphExtractedAt: "2026-01-01T00:00:03.000Z", + graphExtractedCount: 99, + }); + h.land( + obs("a", "2026-01-01T00:00:01.000Z"), + obs("b", "2026-01-01T00:00:04.000Z"), + ); + await h.stop(); + + expect(batches(h.trigger)).toEqual([["a", "b"]]); + }); + + it("leaves the watermark unset when the extract dispatch fails, so the next stop retries", async () => { + let refuse = true; + const h = harness({ rejectGraphExtract: () => refuse }); + + h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); + await h.stop(); + expect(h.session().graphExtractedAt).toBeUndefined(); + + refuse = false; + await h.stop(); + + expect(batches(h.trigger)[1]).toEqual(["a", "b"]); + expect(h.session()).toMatchObject({ graphExtractedCount: 2 }); + }); +}); From b92a512a34df2afa7652b7b4af41c769877d7a70 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 17:52:07 +0800 Subject: [PATCH 04/15] fix(graph): fingerprint the extracted set so a swap cannot skip an observation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the previous commit found a hole in its tripwire. Comparing the number of observations at or below the watermark proves the cardinality is unchanged, not that they are the same observations. A deletion and a late arrival in the same window cancel out: snapshot {o1(T1)..o5(T5)} extracted, watermark (T5, 5); then evict removes o3 and a slow compression lands o2b(T2b + obs.reduce((sum, o) => (sum + Date.parse(o.timestamp)) % OBS_DIGEST_MOD, obs.length); + async function consolidationDueUnserialized(kv: StateKV): Promise { const cooldownMs = getConsolidationCooldownMs(); if (cooldownMs <= 0) return true; // debounce disabled @@ -126,23 +139,24 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { // that is quadratic permanent heap. Send only what landed since the // last extract. // - // The count is what makes the timestamp watermark safe. mem::compress + // The digest is what makes the timestamp watermark safe. mem::compress // is dispatched fire-and-forget (observe.ts) and stamps the capture // time, not the write time, so a slow compression can land an OLDER - // timestamp after a newer one was already extracted. When the number - // of observations newer than the watermark does not exactly account - // for the growth since the watermark was recorded, something arrived - // out of order (or was evicted) and we re-send the whole session - // rather than skip it. Missing a memory is worse than re-merging one. + // timestamp after a newer one was already extracted; evict can also + // remove one at any point. Whenever the already-extracted half no + // longer fingerprints the same, we re-send the whole session rather + // than skip it. Missing a memory is worse than re-merging one. const session = await kv .get(KV.sessions, data.sessionId) .catch(() => null); const at = session?.graphExtractedAt; - const count = session?.graphExtractedCount; + const mark = session?.graphExtractedDigest; let batch = compressed; - if (typeof at === "string" && typeof count === "number") { - const fresh = compressed.filter((o) => o.timestamp > at); - if (fresh.length === compressed.length - count) batch = fresh; + if (typeof at === "string" && typeof mark === "number") { + const seen = compressed.filter((o) => o.timestamp <= at); + if (observationDigest(seen) === mark) { + batch = compressed.filter((o) => o.timestamp > at); + } } if (batch.length > 0) { const newest = compressed.reduce( @@ -158,8 +172,8 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { { type: "set", path: "graphExtractedAt", value: newest }, { type: "set", - path: "graphExtractedCount", - value: compressed.length, + path: "graphExtractedDigest", + value: observationDigest(compressed), }, ]); } diff --git a/src/types.ts b/src/types.ts index 8221e8328..35acc593f 100644 --- a/src/types.ts +++ b/src/types.ts @@ -14,13 +14,14 @@ export interface Session { agentId?: string; // Incremental graph-extract watermark. Written as a matched pair from a // single observation-list snapshot: `graphExtractedAt` is the newest - // observation timestamp handed to mem::graph-extract, `graphExtractedCount` - // is how many compressed observations existed at that moment. The count is - // the tripwire for out-of-order arrivals — mem::compress is dispatched - // fire-and-forget, so observations do not land in KV in timestamp order. - // Absent on session records written before this field existed. + // observation timestamp handed to mem::graph-extract, `graphExtractedDigest` + // fingerprints the observations at or below it. The digest is the tripwire — + // mem::compress is dispatched fire-and-forget so observations do not land in + // KV in timestamp order, and evict can delete one at any time; either way + // the already-extracted set changes and the whole session is re-sent rather + // than skipping anything. Absent on records written before this existed. graphExtractedAt?: string; - graphExtractedCount?: number; + graphExtractedDigest?: number; } export interface CommitLink { diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index f2cd07e2d..8acd04937 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -139,10 +139,11 @@ function harness(opts?: { rejectGraphExtract?: () => boolean }) { const land = (...os: CompressedObservation[]) => { for (const o of os) kv.scope(`mem:obs:${SID}`).set(o.id, o); }; + const drop = (id: string) => kv.scope(`mem:obs:${SID}`).delete(id); const stop = () => stopped({ sessionId: SID }); const session = () => kv.scope("mem:sessions").get(SID) as Record; - return { kv, trigger, stop, land, session }; + return { kv, trigger, stop, land, drop, session }; } describe("event::session::stopped graph-extract is incremental", () => { @@ -183,14 +184,14 @@ describe("event::session::stopped graph-extract is incremental", () => { expect(batches(h.trigger)).toEqual([["a"]]); }); - it("records the watermark as a matched (timestamp, count) pair on the session", async () => { + it("records the watermark as a matched (timestamp, digest) pair on the session", async () => { const h = harness(); h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); await h.stop(); expect(h.session()).toMatchObject({ graphExtractedAt: "2026-01-01T00:00:02.000Z", - graphExtractedCount: 2, + graphExtractedDigest: expect.any(Number), }); }); }); @@ -247,7 +248,7 @@ describe("graph-extract watermark never skips an observation", () => { status: "active", observationCount: 0, graphExtractedAt: "2026-01-01T00:00:01.000Z", - // graphExtractedCount deliberately missing + // graphExtractedDigest deliberately missing }); h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); await h.stop(); @@ -256,9 +257,9 @@ describe("graph-extract watermark never skips an observation", () => { }); it("falls back to the whole session when the stored pair is inconsistent", async () => { - // Interleaved stops only ever write (at, count) as a matched snapshot from - // one run, but a torn or stale pair must still degrade to a full extract - // rather than skip anything. + // Interleaved stops only ever write (at, digest) as a matched snapshot + // from one run, but a torn or stale pair must still degrade to a full + // extract rather than skip anything. const h = harness(); h.kv.scope("mem:sessions").set(SID, { id: SID, @@ -268,7 +269,7 @@ describe("graph-extract watermark never skips an observation", () => { status: "active", observationCount: 0, graphExtractedAt: "2026-01-01T00:00:03.000Z", - graphExtractedCount: 99, + graphExtractedDigest: 123456, }); h.land( obs("a", "2026-01-01T00:00:01.000Z"), @@ -279,6 +280,42 @@ describe("graph-extract watermark never skips an observation", () => { expect(batches(h.trigger)).toEqual([["a", "b"]]); }); + it("re-extracts everything when a deletion and a late arrival cancel out", async () => { + // evict's per-project cap (evict.ts) is age- and status-independent, so it + // can delete an observation from the session that is still being appended + // to. If a late compression lands in the same window, a count-based + // tripwire nets to zero and the late observation would never be extracted. + const h = harness(); + h.land( + obs("a", "2026-01-01T00:00:01.000Z"), + obs("c", "2026-01-01T00:00:03.000Z"), + obs("e", "2026-01-01T00:00:05.000Z"), + ); + await h.stop(); + + h.drop("c"); // evicted + h.land(obs("b", "2026-01-01T00:00:02.000Z")); // compressed late + + await h.stop(); + + expect(batches(h.trigger)[1]).toEqual(["a", "e", "b"]); + }); + + it("re-extracts when a late observation carries a timestamp the digest sums to zero", async () => { + // observe.ts only validates that a timestamp is a string, so a clock-skewed + // client can send the epoch. Date.parse("1970-01-01T00:00:00.000Z") is 0 and + // adds nothing to the digest sum — the set size folded into the seed is what + // keeps that observation visible. + const h = harness(); + h.land(obs("x", "2026-01-01T00:00:01.000Z")); + await h.stop(); + + h.land(obs("y", "1970-01-01T00:00:00.000Z")); + await h.stop(); + + expect(batches(h.trigger)[1]).toEqual(["x", "y"]); + }); + it("leaves the watermark unset when the extract dispatch fails, so the next stop retries", async () => { let refuse = true; const h = harness({ rejectGraphExtract: () => refuse }); @@ -291,6 +328,8 @@ describe("graph-extract watermark never skips an observation", () => { await h.stop(); expect(batches(h.trigger)[1]).toEqual(["a", "b"]); - expect(h.session()).toMatchObject({ graphExtractedCount: 2 }); + expect(h.session()).toMatchObject({ + graphExtractedDigest: expect.any(Number), + }); }); }); From 2138d2f28f77b55cbb603ef725f3eb9da091df40 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 17:59:01 +0800 Subject: [PATCH 05/15] docs(graph): record what the smaller extract batch does and does not change MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review asked the question the batch-content tests cannot answer: the tests mock sdk.trigger, so mem::graph-extract never runs and they prove only that the payload shrank, not that the resulting graph is the same. Traced it. extractGraphHeuristics (graph.ts:457) loops per observation and every link() call joins nodes built from that one observation's own files and concepts. Its nodeByKey and edgeByPair maps are batch-scoped dedup, and persistGraphDelta performs the same dedup across batches through graphNameIndex and graphEdgeKey, with mergeNode and mergeEdge unioning sourceObservationIds. So on the default path the delta produces the identical node and edge set. That is what makes the previous "perf" framing accurate rather than a quiet behaviour change. The opt-in LLM pass is different and worth naming. isGraphExtractionEnabled requires GRAPH_EXTRACTION_ENABLED=true and is off by default; when on, buildGraphExtractionPrompt builds ONE prompt from the whole array, so a per-turn delta of a few observations gives the model less co-occurrence to work with than a whole-session batch did. Cross-turn relations it used to propose will not be proposed. Bounded batches are already the norm on that path, api::graph-build feeds it 25 observations at a time (api.ts:1616), so this narrows an existing property rather than introducing a new limit. Also corrects the residual stated in the previous commit, which was narrower than a sum actually guarantees. The digest is blind to any change to the already-extracted set whose timestamp total is congruent mod 2^32, not only to a single swap 49.7 days apart to the millisecond — deleting two and adding two with a coinciding total would also pass. Still an arithmetic coincidence at millisecond resolution, but the bound is a coinciding sum, not a coinciding pair. One further effect worth knowing, in the direction of better: mergeNode and mergeEdge union the WHOLE batch's obsIds into every node and edge they touch. Sending the entire session every turn therefore accumulated every observation id in the session into sourceObservationIds on every node it merged. The delta attributes only the observations that were actually in the batch, so provenance gets narrower and those arrays stop growing per turn. Comment only, no behaviour change. --- src/triggers/events.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 15cc3972f..9660e566c 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -163,6 +163,16 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { (max, o) => (o.timestamp > max ? o.timestamp : max), "", ); + // What the smaller batch costs, for the next person to read this: + // extractGraphHeuristics (graph.ts) loops per observation and only + // ever links nodes drawn from one observation's own files/concepts, + // and persistGraphDelta dedupes across batches through the name and + // edge indexes, so the heuristic graph is identical either way. The + // opt-in LLM pass (GRAPH_EXTRACTION_ENABLED, off by default) builds + // one prompt from the array, so it sees less co-occurrence per call + // than a whole-session batch did — bounded batches are already the + // norm there, api::graph-build feeds it 25 at a time. + // // Advance only after the dispatch is accepted, so a hand-off that // never left retries on the next turn. Completion is unobservable // through TriggerAction.Void(); an extract that fails downstream From 8d074bc6f537464f1294470eb419d0e6941778d0 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:30:21 +0800 Subject: [PATCH 06/15] fix(graph): fingerprint observation ids, not timestamps The digest that guards the incremental watermark summed Date.parse of each timestamp, seeded with the set size. That collides on exactly the case it was written to catch: evict drops one observation while a late compression lands another carrying the SAME millisecond. Size is unchanged, the sum is unchanged, the digest matches, the incremental path is taken, and the late observation is never handed to mem::graph-extract. On a review of the watermark logic a skip is otherwise impossible, so the whole correctness argument rested on this digest. Fingerprint the sorted observation ids through the existing fingerprintId() helper instead. Ids differ whatever the timestamps do, and order-independence is now structural rather than an arithmetic property a test has to assert. graphExtractedDigest becomes a string. The pinned test previously used two different milliseconds, which is why it passed. Moved to the same millisecond; it fails against the arithmetic digest with the second stop dispatching nothing at all. --- src/triggers/events.ts | 26 ++++++++++-------------- src/types.ts | 2 +- test/graph-extract-incremental.test.ts | 28 +++++++------------------- 3 files changed, 19 insertions(+), 37 deletions(-) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 9660e566c..47520b7cb 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -1,6 +1,6 @@ import { TriggerAction, type ISdk } from "iii-sdk"; import type { CompressedObservation, HookPayload, Session } from "../types.js"; -import { KV, STREAM } from "../state/schema.js"; +import { KV, STREAM, fingerprintId } from "../state/schema.js"; import { StateKV } from "../state/kv.js"; import { isReflectEnabled } from "../functions/slots.js"; import { @@ -14,18 +14,14 @@ import { logger } from "../logger.js"; // the per-turn session-stop fan-out. const CONSOLIDATION_MARKER_KEY = "consolidation:lastRun"; -// Order-independent fingerprint of an observation set, used to tell whether -// the already-extracted half of a session still looks the way it did at the -// last graph extract. Counting is not enough: evict's per-project cap +// Order-independent fingerprint of an observation set: tells whether the +// already-extracted half of a session still looks the way it did at the last +// graph extract. Over ids, not counts or timestamps — evict's per-project cap // (evict.ts, age- and status-independent) can delete an observation from the -// live session in the same window a late compression lands another, and the -// two cancel out. Seeded with the set size so a pure size change always shows, -// and kept under 2^32 so every intermediate stays an exact integer no matter -// how large the session gets. An unparseable timestamp yields NaN, which never -// compares equal, so the session falls back to a full extract. -const OBS_DIGEST_MOD = 0x1_0000_0000; -const observationDigest = (obs: CompressedObservation[]): number => - obs.reduce((sum, o) => (sum + Date.parse(o.timestamp)) % OBS_DIGEST_MOD, obs.length); +// live session in the same window a late compression lands another, and if the +// two share a millisecond only the ids tell the sets apart. +const observationFingerprint = (obs: CompressedObservation[]): string => + fingerprintId("gx", obs.map((o) => o.id).sort().join(",")); async function consolidationDueUnserialized(kv: StateKV): Promise { const cooldownMs = getConsolidationCooldownMs(); @@ -152,9 +148,9 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { const at = session?.graphExtractedAt; const mark = session?.graphExtractedDigest; let batch = compressed; - if (typeof at === "string" && typeof mark === "number") { + if (typeof at === "string" && typeof mark === "string") { const seen = compressed.filter((o) => o.timestamp <= at); - if (observationDigest(seen) === mark) { + if (observationFingerprint(seen) === mark) { batch = compressed.filter((o) => o.timestamp > at); } } @@ -183,7 +179,7 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { { type: "set", path: "graphExtractedDigest", - value: observationDigest(compressed), + value: observationFingerprint(compressed), }, ]); } diff --git a/src/types.ts b/src/types.ts index 35acc593f..e4357c477 100644 --- a/src/types.ts +++ b/src/types.ts @@ -21,7 +21,7 @@ export interface Session { // the already-extracted set changes and the whole session is re-sent rather // than skipping anything. Absent on records written before this existed. graphExtractedAt?: string; - graphExtractedDigest?: number; + graphExtractedDigest?: string; } export interface CommitLink { diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index 8acd04937..e712ecd64 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -191,7 +191,7 @@ describe("event::session::stopped graph-extract is incremental", () => { expect(h.session()).toMatchObject({ graphExtractedAt: "2026-01-01T00:00:02.000Z", - graphExtractedDigest: expect.any(Number), + graphExtractedDigest: expect.any(String), }); }); }); @@ -280,11 +280,12 @@ describe("graph-extract watermark never skips an observation", () => { expect(batches(h.trigger)).toEqual([["a", "b"]]); }); - it("re-extracts everything when a deletion and a late arrival cancel out", async () => { + it("re-extracts everything when a deletion and a late arrival share a millisecond", async () => { // evict's per-project cap (evict.ts) is age- and status-independent, so it // can delete an observation from the session that is still being appended - // to. If a late compression lands in the same window, a count-based - // tripwire nets to zero and the late observation would never be extracted. + // to. When a late compression lands in the same window AND carries the + // same timestamp, size and any timestamp-derived checksum both net to + // zero — only the observation ids tell the two sets apart. const h = harness(); h.land( obs("a", "2026-01-01T00:00:01.000Z"), @@ -294,28 +295,13 @@ describe("graph-extract watermark never skips an observation", () => { await h.stop(); h.drop("c"); // evicted - h.land(obs("b", "2026-01-01T00:00:02.000Z")); // compressed late + h.land(obs("b", "2026-01-01T00:00:03.000Z")); // compressed late, same ms await h.stop(); expect(batches(h.trigger)[1]).toEqual(["a", "e", "b"]); }); - it("re-extracts when a late observation carries a timestamp the digest sums to zero", async () => { - // observe.ts only validates that a timestamp is a string, so a clock-skewed - // client can send the epoch. Date.parse("1970-01-01T00:00:00.000Z") is 0 and - // adds nothing to the digest sum — the set size folded into the seed is what - // keeps that observation visible. - const h = harness(); - h.land(obs("x", "2026-01-01T00:00:01.000Z")); - await h.stop(); - - h.land(obs("y", "1970-01-01T00:00:00.000Z")); - await h.stop(); - - expect(batches(h.trigger)[1]).toEqual(["x", "y"]); - }); - it("leaves the watermark unset when the extract dispatch fails, so the next stop retries", async () => { let refuse = true; const h = harness({ rejectGraphExtract: () => refuse }); @@ -329,7 +315,7 @@ describe("graph-extract watermark never skips an observation", () => { expect(batches(h.trigger)[1]).toEqual(["a", "b"]); expect(h.session()).toMatchObject({ - graphExtractedDigest: expect.any(Number), + graphExtractedDigest: expect.any(String), }); }); }); From 32c9c932811f64f72076ae505c04d638644f695a Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:31:15 +0800 Subject: [PATCH 07/15] fix(graph): derive the whole watermark pair from the dispatched batch `newest` reduced over the full compressed list and the digest fingerprinted the full compressed list, while the batch handed to mem::graph-extract was only the tail. Those are equal today because the batch is always `timestamp > at`, so max(batch) === max(compressed). They stop being equal the moment anyone caps this batch, and the file's own comment cites api::graph-build capping at 25 as precedent. A cap would make `newest` advance past observations that were never dispatched, which is the exact silent skip the digest exists to prevent, and would leave the digest covering observations above the watermark so the pair could never match again. Reduce over `batch` and fingerprint everything at or below the resulting watermark. Both fields now come off one predicate, so the invariant holds locally instead of resting on a max-equality argument in prose. --- src/triggers/events.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 47520b7cb..458b262df 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -155,10 +155,14 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { } } if (batch.length > 0) { - const newest = compressed.reduce( + // Both halves of the pair come off the batch actually dispatched, so + // capping the batch (api::graph-build already feeds 25 at a time) + // cannot advance the watermark past an observation nobody sent. + const newest = batch.reduce( (max, o) => (o.timestamp > max ? o.timestamp : max), "", ); + const extracted = compressed.filter((o) => o.timestamp <= newest); // What the smaller batch costs, for the next person to read this: // extractGraphHeuristics (graph.ts) loops per observation and only // ever links nodes drawn from one observation's own files/concepts, @@ -179,7 +183,7 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { { type: "set", path: "graphExtractedDigest", - value: observationFingerprint(compressed), + value: observationFingerprint(extracted), }, ]); } From 2f059ac57f5aad3d1cef0b9e823762fde7bdbb1f Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:32:22 +0800 Subject: [PATCH 08/15] feat(graph): log when the incremental extract falls back to a full one Neither branch logged anything, so a session whose watermark keeps going stale re-sends its whole observation list every turn and looks identical to a healthy one from the outside. One line on the mismatch path, carrying the two set sizes so the cause (a deletion, a late arrival, a half-written pair) is readable from the log rather than guessed. --- src/triggers/events.ts | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 458b262df..66f3674c8 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -152,6 +152,14 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { const seen = compressed.filter((o) => o.timestamp <= at); if (observationFingerprint(seen) === mark) { batch = compressed.filter((o) => o.timestamp > at); + } else { + // Otherwise the fallback is silent: a session stuck re-extracting + // itself every turn looks exactly like a healthy one. + logger.info("graph-extract watermark stale, re-extracting session", { + sessionId: data.sessionId, + atOrBelow: seen.length, + total: compressed.length, + }); } } if (batch.length > 0) { From de82c4ddecc937852991752174256ff023cc6624 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:33:19 +0800 Subject: [PATCH 09/15] docs(graph): say what the smaller batch actually changes The comment claimed the heuristic graph is "identical either way" and hung that on persistGraphDelta's dedupe. The dedupe argument is right about the node and edge SETS and wrong about provenance: mergeNode and mergeEdge union the whole batch's obsIds (graph.ts), so a whole-session batch stamped every node and edge with every observation id in the session. The union cited as proof of equivalence is where the divergence lives. Post-change provenance is per-batch, which is narrower and more accurate, and it is read by graph retrieval and by supersede-staling. Net an improvement, but it ships on the default path and the comment should say so rather than assert equivalence. Also drops the GRAPH_EXTRACTION_ENABLED framing. That flag gates only the LLM pass; the heuristic pass that produces this behaviour always runs. --- src/triggers/events.ts | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 66f3674c8..0de43b665 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -171,15 +171,13 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { "", ); const extracted = compressed.filter((o) => o.timestamp <= newest); - // What the smaller batch costs, for the next person to read this: - // extractGraphHeuristics (graph.ts) loops per observation and only - // ever links nodes drawn from one observation's own files/concepts, - // and persistGraphDelta dedupes across batches through the name and - // edge indexes, so the heuristic graph is identical either way. The - // opt-in LLM pass (GRAPH_EXTRACTION_ENABLED, off by default) builds - // one prompt from the array, so it sees less co-occurrence per call - // than a whole-session batch did — bounded batches are already the - // norm there, api::graph-build feeds it 25 at a time. + // The node and edge SETS are unchanged: extractGraphHeuristics links + // only within a single observation, and persistGraphDelta dedupes + // across batches through the name and edge-key indexes. Provenance + // does change — mergeNode/mergeEdge union the whole batch's obsIds, + // so a whole-session batch stamped every node and edge with every + // observation id in the session. Narrower is more accurate, and + // graph retrieval and supersede-staling both read it. // // Advance only after the dispatch is accepted, so a hand-off that // never left retries on the next turn. Completion is unobservable From 67519a70d26829afc680788bb3c36d215d37198c Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:36:40 +0800 Subject: [PATCH 10/15] refactor(graph): drop the scaffolding the watermark did not need MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three things that carried no weight: `typeof mark === "string"` was a no-op. The comparison on the next line already yields false for an absent or legacy-numeric digest, which is the same fallback the guard produced. fireVoid grew a boolean return for one of its four callers. Calling sdk.trigger directly for mem::graph-extract lets a rejected dispatch throw into the try/catch that was already there, which logs the same message and skips the watermark write the same way, and restores fireVoid to what the other three callers use. The "mem::compress is fire-and-forget so writes are not in timestamp order" paragraph was duplicated verbatim in types.ts and events.ts. types.ts now points at events.ts. Four test cases went with it. Each asserted a mutation another case already kills: the linear-vs-quadratic count is subsumed by the exact per-turn batch assertion, the matched-pair case by the same (a wrong watermark changes the next turn's batch), the synthetic torn pair by the realistic out-of-order case, and the half-written pair killed nothing at all — it passed with the guard present or absent. The fake KV's set/delete were never invoked. --- src/triggers/events.ts | 45 ++++++++-------- src/types.ts | 11 ++-- test/graph-extract-incremental.test.ts | 72 -------------------------- test/graph-heuristic-extract.test.ts | 2 +- 4 files changed, 26 insertions(+), 104 deletions(-) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index 0de43b665..dfc3bcbcb 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -104,20 +104,15 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { sdk.registerFunction("event::session::stopped", async (data: { sessionId: string; skipConsolidation?: boolean }) => { const summary = await sdk.trigger({ function_id: "mem::summarize", payload: data }); - // Resolves true when the trigger was accepted, false when the dispatch - // itself failed. Callers that need to know (graph-extract's watermark) - // read it; the rest ignore it exactly as before. const fireVoid = (function_id: string, payload: unknown) => sdk .trigger({ function_id, payload, action: TriggerAction.Void() }) - .then(() => true) - .catch((err) => { + .catch((err) => logger.warn(function_id + " trigger failed", { sessionId: data.sessionId, error: err instanceof Error ? err.message : String(err), - }); - return false; - }); + }), + ); if (isReflectEnabled()) { fireVoid("mem::slot-reflect", { sessionId: data.sessionId }); } @@ -148,7 +143,7 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { const at = session?.graphExtractedAt; const mark = session?.graphExtractedDigest; let batch = compressed; - if (typeof at === "string" && typeof mark === "string") { + if (typeof at === "string") { const seen = compressed.filter((o) => o.timestamp <= at); if (observationFingerprint(seen) === mark) { batch = compressed.filter((o) => o.timestamp > at); @@ -179,20 +174,24 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { // observation id in the session. Narrower is more accurate, and // graph retrieval and supersede-staling both read it. // - // Advance only after the dispatch is accepted, so a hand-off that - // never left retries on the next turn. Completion is unobservable - // through TriggerAction.Void(); an extract that fails downstream - // leaves its delta out of the graph until POST /agentmemory/graph/build. - if (await fireVoid("mem::graph-extract", { observations: batch })) { - await kv.update(KV.sessions, data.sessionId, [ - { type: "set", path: "graphExtractedAt", value: newest }, - { - type: "set", - path: "graphExtractedDigest", - value: observationFingerprint(extracted), - }, - ]); - } + // A dispatch that throws skips the watermark write (the catch below + // logs it) and retries on the next turn. Accepted is not done — + // completion is unobservable through TriggerAction.Void(), so an + // extract that fails downstream leaves its delta out of the graph + // until POST /agentmemory/graph/build. + await sdk.trigger({ + function_id: "mem::graph-extract", + payload: { observations: batch }, + action: TriggerAction.Void(), + }); + await kv.update(KV.sessions, data.sessionId, [ + { type: "set", path: "graphExtractedAt", value: newest }, + { + type: "set", + path: "graphExtractedDigest", + value: observationFingerprint(extracted), + }, + ]); } } } catch (err) { diff --git a/src/types.ts b/src/types.ts index e4357c477..11dc4f82a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -12,14 +12,9 @@ export interface Session { summary?: string; commitShas?: string[]; agentId?: string; - // Incremental graph-extract watermark. Written as a matched pair from a - // single observation-list snapshot: `graphExtractedAt` is the newest - // observation timestamp handed to mem::graph-extract, `graphExtractedDigest` - // fingerprints the observations at or below it. The digest is the tripwire — - // mem::compress is dispatched fire-and-forget so observations do not land in - // KV in timestamp order, and evict can delete one at any time; either way - // the already-extracted set changes and the whole session is re-sent rather - // than skipping anything. Absent on records written before this existed. + // Matched incremental graph-extract watermark, written together by + // event::session::stopped (triggers/events.ts) — see there for why the + // digest is needed. Absent on records written before this existed. graphExtractedAt?: string; graphExtractedDigest?: string; } diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index e712ecd64..3d65d17da 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -56,13 +56,6 @@ function persistentKV() { store, scope, get: vi.fn(async (s: string, k: string) => scope(s).get(k) ?? null), - set: vi.fn(async (s: string, k: string, v: unknown) => { - scope(s).set(k, v); - return v; - }), - delete: vi.fn(async (s: string, k: string) => { - scope(s).delete(k); - }), update: vi.fn( async ( s: string, @@ -162,18 +155,6 @@ describe("event::session::stopped graph-extract is incremental", () => { expect(batches(h.trigger)).toEqual([["a", "b"], ["c"], ["d"]]); }); - it("keeps total dispatched observations linear in turn count, not quadratic", async () => { - // Pre-fix, 8 turns of one observation each dispatched 1+2+…+8 = 36 - // observations (and 36 merge passes downstream). Incremental sends 8. - const h = harness(); - for (let i = 1; i <= 8; i++) { - h.land(obs(`o${i}`, `2026-01-01T00:00:0${i}.000Z`)); - await h.stop(); - } - const dispatched = batches(h.trigger).reduce((n, b) => n + b.length, 0); - expect(dispatched).toBe(8); - }); - it("does not dispatch graph-extract at all when no new observations landed", async () => { const h = harness(); h.land(obs("a", "2026-01-01T00:00:01.000Z")); @@ -183,17 +164,6 @@ describe("event::session::stopped graph-extract is incremental", () => { expect(batches(h.trigger)).toEqual([["a"]]); }); - - it("records the watermark as a matched (timestamp, digest) pair on the session", async () => { - const h = harness(); - h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); - await h.stop(); - - expect(h.session()).toMatchObject({ - graphExtractedAt: "2026-01-01T00:00:02.000Z", - graphExtractedDigest: expect.any(String), - }); - }); }); // The fallback branch. Most of these assert the PRE-FIX behaviour (extract @@ -238,48 +208,6 @@ describe("graph-extract watermark never skips an observation", () => { expect(batches(h.trigger)[1]).toEqual(["a", "b"]); }); - it("falls back to the whole session when the watermark is half-written", async () => { - const h = harness(); - h.kv.scope("mem:sessions").set(SID, { - id: SID, - project: "p", - cwd: "/p", - startedAt: "2026-01-01T00:00:00.000Z", - status: "active", - observationCount: 0, - graphExtractedAt: "2026-01-01T00:00:01.000Z", - // graphExtractedDigest deliberately missing - }); - h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); - await h.stop(); - - expect(batches(h.trigger)).toEqual([["a", "b"]]); - }); - - it("falls back to the whole session when the stored pair is inconsistent", async () => { - // Interleaved stops only ever write (at, digest) as a matched snapshot - // from one run, but a torn or stale pair must still degrade to a full - // extract rather than skip anything. - const h = harness(); - h.kv.scope("mem:sessions").set(SID, { - id: SID, - project: "p", - cwd: "/p", - startedAt: "2026-01-01T00:00:00.000Z", - status: "active", - observationCount: 0, - graphExtractedAt: "2026-01-01T00:00:03.000Z", - graphExtractedDigest: 123456, - }); - h.land( - obs("a", "2026-01-01T00:00:01.000Z"), - obs("b", "2026-01-01T00:00:04.000Z"), - ); - await h.stop(); - - expect(batches(h.trigger)).toEqual([["a", "b"]]); - }); - it("re-extracts everything when a deletion and a late arrival share a millisecond", async () => { // evict's per-project cap (evict.ts) is age- and status-independent, so it // can delete an observation from the session that is still being appended diff --git a/test/graph-heuristic-extract.test.ts b/test/graph-heuristic-extract.test.ts index f3bfec805..0564bbc32 100644 --- a/test/graph-heuristic-extract.test.ts +++ b/test/graph-heuristic-extract.test.ts @@ -96,7 +96,7 @@ describe("keyless graph extraction wiring", () => { const events = readFileSync("src/triggers/events.ts", "utf-8"); const stopped = events.slice(events.indexOf("event::session::stopped")); const gate = stopped.indexOf("isGraphExtractionEnabled()"); - const fire = stopped.indexOf('fireVoid("mem::graph-extract"'); + const fire = stopped.indexOf('"mem::graph-extract"'); expect(fire).toBeGreaterThan(-1); expect(gate === -1 || gate > fire).toBe(true); }); From 43f44e589946b3bfb44885787e224584ffd182b8 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:40:17 +0800 Subject: [PATCH 11/15] test(graph): pin the graph shape the smaller batch is supposed to preserve Nothing verified the claim the whole change rests on. The incremental suite sets concepts and files to [] and mocks sdk.trigger, so mem::graph-extract never runs and the graph is never built. Call extractGraphHeuristics directly with populated concepts and files and compare one batch of two observations against two batches of one, merged by (type, name) and edge endpoints the way persistGraphDelta merges them. Node ids are random, so the comparison is by name throughout. This is what would catch a future cross-observation link, which is the only way splitting the batch could change the set. Also pins order-independence of the fingerprint: the same observation set listed back in a different order must stay on the incremental path. Kills the mutation that drops the sort. --- test/graph-extract-incremental.test.ts | 15 ++++++++++ test/graph-heuristic-extract.test.ts | 41 ++++++++++++++++++++++++++ 2 files changed, 56 insertions(+) diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index 3d65d17da..3cef57fea 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -164,6 +164,21 @@ describe("event::session::stopped graph-extract is incremental", () => { expect(batches(h.trigger)).toEqual([["a"]]); }); + + it("stays incremental when kv.list returns the same set in a different order", async () => { + // Listing order is the store's, not ours. The same observation set coming + // back rearranged is not a change and must not force a full re-extract. + const h = harness(); + h.land(obs("a", "2026-01-01T00:00:01.000Z"), obs("b", "2026-01-01T00:00:02.000Z")); + await h.stop(); + + h.drop("a"); + h.land(obs("a", "2026-01-01T00:00:01.000Z")); // same observation, now last + h.land(obs("c", "2026-01-01T00:00:03.000Z")); + await h.stop(); + + expect(batches(h.trigger)[1]).toEqual(["c"]); + }); }); // The fallback branch. Most of these assert the PRE-FIX behaviour (extract diff --git a/test/graph-heuristic-extract.test.ts b/test/graph-heuristic-extract.test.ts index 0564bbc32..0d0593858 100644 --- a/test/graph-heuristic-extract.test.ts +++ b/test/graph-heuristic-extract.test.ts @@ -73,6 +73,47 @@ describe("extractGraphHeuristics", () => { expect(edges.length).toBeLessThanOrEqual(12); }); + // event::session::stopped now sends one batch per turn instead of the whole + // session. persistGraphDelta merges batches by (type, name) and by edge key, + // so splitting a batch must change neither set. Node ids are random, so + // compare by name. + function shape(results: Array>) { + const prov = new Map>(); + const edges = new Set(); + for (const r of results) { + const named = new Map(r.nodes.map((n) => [n.id, `${n.type}:${n.name}`])); + for (const n of r.nodes) { + const key = `${n.type}:${n.name}`; + const ids = prov.get(key) ?? new Set(); + for (const id of n.sourceObservationIds) ids.add(id); + prov.set(key, ids); + } + for (const e of r.edges) { + const pair = [named.get(e.sourceNodeId)!, named.get(e.targetNodeId)!]; + edges.add(pair.sort().join("|")); + } + } + return { + nodes: [...prov] + .map(([key, ids]) => `${key}=${[...ids].sort().join(",")}`) + .sort(), + edges: [...edges].sort(), + }; + } + + it("builds the same graph whether observations arrive together or one batch at a time", () => { + // One shared entity so the merge and provenance union are exercised, and + // disjoint ones on either side so any link drawn ACROSS observations + // (the only way splitting could change the set) shows up as an edge the + // split calls cannot produce. + const o1 = obs("o1", ["src/auth.ts", "src/db.ts"], ["authentication"]); + const o2 = obs("o2", ["src/db.ts", "src/cache.ts"], ["caching"]); + + expect( + shape([extractGraphHeuristics([o1]), extractGraphHeuristics([o2])]), + ).toEqual(shape([extractGraphHeuristics([o1, o2])])); + }); + it("never emits self edges or duplicate pairs", () => { const { edges } = extractGraphHeuristics([ obs("o1", ["a.ts"], ["a"]), From 554d8e5f6fc1e6395993b4ab14b5a2c4effdd491 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:43:54 +0800 Subject: [PATCH 12/15] test(graph): pin the fallback log to the fallback The mutation run showed the `typeof at === "string"` guard could be removed with every test still green: with no watermark the timestamp filter yields an empty set, that fingerprints to something no absent digest equals, and the full extract happens anyway. The guard's one remaining job is keeping the stale-watermark log off every session's first turn, and nothing checked that. Assert the log is silent on the first stop and carries the two set sizes on the fallback path. Also the only coverage the log line has. --- test/graph-extract-incremental.test.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index 3cef57fea..a1968075f 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -16,6 +16,9 @@ vi.mock("../src/functions/slots.js", () => ({ })); import { registerEventTriggers } from "../src/triggers/events.js"; +import { logger } from "../src/logger.js"; + +const STALE = "graph-extract watermark stale, re-extracting session"; // /agentmemory/session/end is posted by Claude Code's per-turn Stop hook, so // event::session::stopped runs on EVERY agent turn — not once per session. @@ -197,6 +200,8 @@ describe("graph-extract watermark never skips an observation", () => { ); await h.stop(); expect(batches(h.trigger)).toEqual([["a", "b", "c"]]); + // A session that has never been extracted is not a stale watermark. + expect(logger.info).not.toHaveBeenCalled(); }); it("re-extracts everything when an observation lands out of order below the watermark", async () => { @@ -243,6 +248,11 @@ describe("graph-extract watermark never skips an observation", () => { await h.stop(); expect(batches(h.trigger)[1]).toEqual(["a", "e", "b"]); + expect(logger.info).toHaveBeenCalledWith(STALE, { + sessionId: SID, + atOrBelow: 3, + total: 3, + }); }); it("leaves the watermark unset when the extract dispatch fails, so the next stop retries", async () => { From ed3cf971f398693555fc4d16c954577bbf241b76 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 19:50:32 +0800 Subject: [PATCH 13/15] test(graph): cover the legacy record every install hits on upgrade MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit graphExtractedDigest is new, so on the first stop after this ships every existing session record has graphExtractedAt and no digest. That branch is traversed once by 100% of real sessions and nothing exercised it — a permissiveness mutation (`!mark || fingerprint(seen) === mark`, the "do not punish upgraders with one re-extract" shortcut) passes the entire 1720-test suite untouched, while silently dropping every observation at or below the stale watermark. Assert the whole session goes out when the digest is absent and the observations straddle the watermark. Kills that mutation. --- test/graph-extract-incremental.test.ts | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index a1968075f..ac1750d74 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -204,6 +204,31 @@ describe("graph-extract watermark never skips an observation", () => { expect(logger.info).not.toHaveBeenCalled(); }); + it("re-extracts the whole session when the record predates the digest", async () => { + // The path every deployed session takes exactly once, on the first stop + // after this ships: a session record carrying graphExtractedAt and no + // digest at all. An untrustworthy watermark costs one re-extract; + // trusting it costs every observation at or below it. + const h = harness(); + h.kv.scope("mem:sessions").set(SID, { + id: SID, + project: "p", + cwd: "/p", + startedAt: "2026-01-01T00:00:00.000Z", + status: "active", + observationCount: 0, + graphExtractedAt: "2026-01-01T00:00:02.000Z", + // graphExtractedDigest absent — written by a build that had none + }); + h.land( + obs("a", "2026-01-01T00:00:01.000Z"), + obs("b", "2026-01-01T00:00:03.000Z"), + ); + await h.stop(); + + expect(batches(h.trigger)).toEqual([["a", "b"]]); + }); + it("re-extracts everything when an observation lands out of order below the watermark", async () => { // mem::compress is fire-and-forget, so a slow compression writes an older // timestamp after a newer one was already extracted. Without the count From 7cd5577996d694dbb597ab236704f22d537c3715 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 22:07:51 +0800 Subject: [PATCH 14/15] refactor(graph): drop the derived set that always equalled its source `extracted` filtered `compressed` down to observations at or below `newest`, and returned `compressed` on every reachable path. On the fallback branch `batch` IS `compressed`, so `newest` is its maximum. On the incremental branch `batch` is the tail above `at` and non-empty, so `newest` is the maximum of the whole list either way. Verified algebraically and by exhaustive enumeration. It was scaffolding for a batch cap that does not exist. Without it, a future cap makes the next turn fingerprint miss and the session falls back to a full re-extract, which the log added alongside it now makes visible rather than silent. Nothing is skipped either way, and `newest` reducing over `batch` is what actually closes that hazard. The comment claimed both halves of the watermark pair came off the dispatched batch. Only `newest` did; the fingerprint came off `compressed` through a filter that excluded nothing. Says what the code does now, and drops two paragraphs that restated what the tests already execute. --- src/triggers/events.ts | 24 ++++++++++-------------- 1 file changed, 10 insertions(+), 14 deletions(-) diff --git a/src/triggers/events.ts b/src/triggers/events.ts index dfc3bcbcb..3d0503add 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -158,25 +158,21 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { } } if (batch.length > 0) { - // Both halves of the pair come off the batch actually dispatched, so - // capping the batch (api::graph-build already feeds 25 at a time) - // cannot advance the watermark past an observation nobody sent. + // Off the dispatched batch, so a future cap cannot advance the + // watermark past an observation nobody sent. const newest = batch.reduce( (max, o) => (o.timestamp > max ? o.timestamp : max), "", ); - const extracted = compressed.filter((o) => o.timestamp <= newest); - // The node and edge SETS are unchanged: extractGraphHeuristics links - // only within a single observation, and persistGraphDelta dedupes - // across batches through the name and edge-key indexes. Provenance - // does change — mergeNode/mergeEdge union the whole batch's obsIds, + // Same node and edge sets either way — pinned by + // graph-heuristic-extract.test.ts. Provenance narrows, which is the + // real change: mergeNode/mergeEdge union the whole batch's obsIds, // so a whole-session batch stamped every node and edge with every - // observation id in the session. Narrower is more accurate, and - // graph retrieval and supersede-staling both read it. + // observation id in the session. // - // A dispatch that throws skips the watermark write (the catch below - // logs it) and retries on the next turn. Accepted is not done — - // completion is unobservable through TriggerAction.Void(), so an + // Accepted is not done. A throw skips the watermark write and + // retries next turn (pinned by graph-extract-incremental.test.ts), + // but completion is unobservable through TriggerAction.Void(), so an // extract that fails downstream leaves its delta out of the graph // until POST /agentmemory/graph/build. await sdk.trigger({ @@ -189,7 +185,7 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { { type: "set", path: "graphExtractedDigest", - value: observationFingerprint(extracted), + value: observationFingerprint(compressed), }, ]); } From 83313c05c10fd72cc69bcd372a46f9edd5b06337 Mon Sep 17 00:00:00 2001 From: Omar Gerardo Date: Thu, 27 Aug 2026 22:08:06 +0800 Subject: [PATCH 15/15] docs(test): point at the rationale instead of keeping a third copy The quadratic-heap explanation lived in three files. An earlier commit replaced the copy in types.ts with a pointer to events.ts and left the one in this test header untouched. --- test/graph-extract-incremental.test.ts | 11 ++--------- 1 file changed, 2 insertions(+), 9 deletions(-) diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts index ac1750d74..29738ed3f 100644 --- a/test/graph-extract-incremental.test.ts +++ b/test/graph-extract-incremental.test.ts @@ -20,15 +20,8 @@ import { logger } from "../src/logger.js"; const STALE = "graph-extract watermark stale, re-extracting session"; -// /agentmemory/session/end is posted by Claude Code's per-turn Stop hook, so -// event::session::stopped runs on EVERY agent turn — not once per session. -// It used to hand mem::graph-extract the session's entire observation list -// each time, and persistGraphDelta takes the 3-call MERGE path for every node -// and edge it has already seen, so turn N re-merged turns 1..N-1. Engine -// invocations were quadratic in turn count, and per #843 nothing written via -// kv.set is ever evicted from the iii engine, so that was quadratic permanent -// heap. These tests pin the extract to the observations captured since the -// last successful extract. +// These tests pin the extract to the observations captured since the last +// successful one. Why that matters is in src/triggers/events.ts. const SID = "ses_1";