deploy: incremental graph-extract (upstream #1268)

deploy/README.md

@@ -90,7 +90,17 @@ 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
+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
+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.

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,6 +14,15 @@ 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: 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 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<boolean> {
   const cooldownMs = getConsolidationCooldownMs();
   if (cooldownMs <= 0) return true; // debounce disabled
@@ -114,7 +123,72 @@ 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 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; 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<Session>(KV.sessions, data.sessionId)
+          .catch(() => null);
+        const at = session?.graphExtractedAt;
+        const mark = session?.graphExtractedDigest;
+        let batch = compressed;
+        if (typeof at === "string") {
+          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) {
+          // 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),
+            "",
+          );
+          // 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.
+          //
+          // 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({
+            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(compressed),
+            },
+          ]);
+        }
       }
     } catch (err) {
       logger.warn("graph-extract trigger failed", {

src/types.ts

@@ -12,6 +12,11 @@ export interface Session {
   summary?: string;
   commitShas?: string[];
   agentId?: string;
+  // 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;
 }
 
 export interface CommitLink {