diff --git a/deploy/README.md b/deploy/README.md
index a84f653aa..d2ef7141b 100644
--- a/deploy/README.md
+++ b/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.
diff --git a/src/triggers/events.ts b/src/triggers/events.ts
index bbf15db33..3d0503add 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,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", {
diff --git a/src/types.ts b/src/types.ts
index d8543c13a..11dc4f82a 100644
--- a/src/types.ts
+++ b/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 {
diff --git a/test/graph-extract-incremental.test.ts b/test/graph-extract-incremental.test.ts
new file mode 100644
index 000000000..29738ed3f
--- /dev/null
+++ b/test/graph-extract-incremental.test.ts
@@ -0,0 +1,292 @@
+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";
+import { logger } from "../src/logger.js";
+
+const STALE = "graph-extract watermark stale, re-extracting session";
+
+// 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";
+
+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<string, Map<string, unknown>>();
+  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),
+    update: vi.fn(
+      async (
+        s: string,
+        k: string,
+        ops: Array<{ type: string; path: string; value?: unknown }>,
+      ) => {
+        const cur = (scope(s).get(k) ?? {}) as Record<string, unknown>;
+        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<unknown>;
+
+function mockSdk(opts?: { rejectGraphExtract?: () => boolean }) {
+  const handlers = new Map<string, StoppedHandler>();
+  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<typeof vi.fn>): 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 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<string, unknown>;
+  return { kv, trigger, stop, land, drop, 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("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("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
+// 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"]]);
+    // A session that has never been extracted is not a stale watermark.
+    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
+    // 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("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. 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"),
+      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:03.000Z")); // compressed late, same ms
+
+    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 () => {
+    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({
+      graphExtractedDigest: expect.any(String),
+    });
+  });
+});
diff --git a/test/graph-heuristic-extract.test.ts b/test/graph-heuristic-extract.test.ts
index f3bfec805..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<ReturnType<typeof extractGraphHeuristics>>) {
+    const prov = new Map<string, Set<string>>();
+    const edges = new Set<string>();
+    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<string>();
+        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"]),
@@ -96,7 +137,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);
   });