diff --git a/plugin/scripts/stop.mjs b/plugin/scripts/stop.mjs index 03d30c64f..7c4b21fc6 100755 --- a/plugin/scripts/stop.mjs +++ b/plugin/scripts/stop.mjs @@ -23,22 +23,16 @@ async function main() { } if (isSdkChildContext(data)) return; const sessionId = data.session_id || data.sessionId || "unknown"; - fetch(`${REST_URL}/agentmemory/summarize`, { - method: "POST", - headers: authHeaders(), - body: JSON.stringify({ sessionId }), - signal: AbortSignal.timeout(12e4) - }).catch(() => {}); fetch(`${REST_URL}/agentmemory/session/end`, { method: "POST", headers: authHeaders(), body: JSON.stringify({ sessionId }), signal: AbortSignal.timeout(5e3) }).catch(() => {}); - setTimeout(() => process.exit(0), 1500).unref(); + setTimeout(() => process.exit(0), 500).unref(); } main(); - //#endregion -export { }; +export {}; + //# sourceMappingURL=stop.mjs.map \ No newline at end of file diff --git a/src/functions/input-fingerprint.ts b/src/functions/input-fingerprint.ts new file mode 100644 index 000000000..b87c608f8 --- /dev/null +++ b/src/functions/input-fingerprint.ts @@ -0,0 +1,14 @@ +import { createHash } from "node:crypto"; +import type { CompressedObservation } from "../types.js"; + +// Stable fingerprint of an exact observation set, used to skip LLM re-runs +// (summarize, graph extraction) when the input is unchanged. Count alone is +// not enough: evict/auto-forget can delete observations, so the set can +// change while the count returns to a previous value. +export function computeInputFingerprint( + observations: CompressedObservation[], +): string { + const h = createHash("sha256"); + for (const o of observations) h.update(`${o.id} ${o.timestamp} `); + return h.digest("hex").slice(0, 32); +} diff --git a/src/functions/summarize.ts b/src/functions/summarize.ts index 4c501ca8c..fc2dbc86b 100644 --- a/src/functions/summarize.ts +++ b/src/functions/summarize.ts @@ -7,6 +7,7 @@ import type { } from "../types.js"; import { KV } from "../state/schema.js"; import { StateKV } from "../state/kv.js"; +import { withKeyedLock } from "../state/keyed-mutex.js"; import { SUMMARY_SYSTEM, buildSummaryPrompt, @@ -14,6 +15,7 @@ import { buildReducePrompt, } from "../prompts/summary.js"; import { getXmlTag, getXmlChildren } from "../prompts/xml.js"; +import { computeInputFingerprint } from "./input-fingerprint.js"; import { SummaryOutputSchema } from "../eval/schemas.js"; import { validateOutput } from "../eval/validator.js"; import { scoreSummary } from "../eval/quality.js"; @@ -50,6 +52,44 @@ function getChunkConcurrency(): number { return Number.isFinite(n) && n > 0 ? n : CHUNK_CONCURRENCY_DEFAULT; } +// Upper bound on how long one summarize run may hold the per-session lock. +// Some providers (anthropic, agent-sdk) have no request timeout of their +// own; without a bound a hung LLM call would block every later summarize +// for that session forever. Generous enough for chunked map-reduce runs. +const LOCK_TIMEOUT_MS_DEFAULT = 300_000; + +function getLockTimeoutMs(): number { + const raw = process.env.SUMMARIZE_LOCK_TIMEOUT_MS; + if (!raw) return LOCK_TIMEOUT_MS_DEFAULT; + const n = parseInt(raw, 10); + return Number.isFinite(n) && n > 0 ? n : LOCK_TIMEOUT_MS_DEFAULT; +} + +// Resolves with a timeout marker instead of rejecting so the caller gets a +// normal { success: false } result. The underlying run keeps going in the +// background, but the keyed lock is released. +function raceLockTimeout( + work: Promise, + ms: number, + sessionId: string, +): Promise { + let timer: ReturnType | undefined; + const timeout = new Promise<{ success: false; error: string }>((resolve) => { + timer = setTimeout(() => { + logger.warn("Summarize lock timeout — releasing session lock", { + sessionId, + timeoutMs: ms, + }); + resolve({ success: false, error: "summarize_timeout" }); + }, ms); + timer.unref?.(); + }); + return Promise.race([ + work.finally(() => clearTimeout(timer)), + timeout, + ]); +} + // One chunk call with retry-once. Returns null when both attempts fail — // whether by parse failure, provider 4xx (content rejected by upstream // filters), or transient network/5xx errors that didn't recover on retry. @@ -240,159 +280,188 @@ export function registerSummarizeFunction( } const sessionId = data.sessionId.trim(); - const session = await kv.get(KV.sessions, sessionId); - if (!session) { - logger.warn("Session not found for summarize", { - sessionId, - }); - return { success: false, error: "session_not_found" }; - } + // Serialize per-session runs: the stop path and the explicit API can + // race, and a second concurrent run would redo the same LLM work. + // See the up-to-date short-circuit below for the sequential case. + const runSummarize = async () => { + const session = await kv.get(KV.sessions, sessionId); + if (!session) { + logger.warn("Session not found for summarize", { + sessionId, + }); + return { success: false, error: "session_not_found" }; + } - const observations = await kv.list( - KV.observations(sessionId), - ); - const compressed = observations.filter((o) => o.title); + const observations = await kv.list( + KV.observations(sessionId), + ); + const compressed = observations.filter((o) => o.title); - if (compressed.length === 0) { - logger.info("No observations to summarize", { - sessionId, - }); - return { success: false, error: "no_observations" }; - } + if (compressed.length === 0) { + logger.info("No observations to summarize", { + sessionId, + }); + return { success: false, error: "no_observations" }; + } - if (provider.name === "noop") { - logger.info("Summarize skipped — no LLM provider configured", { + // An unchanged input fingerprint means the LLM input would be + // identical to the previous run. Skipping avoids re-summarizing + // long-lived sessions (e.g. heartbeat loops) from scratch on every + // stop when nothing new happened. Summaries written before this + // field existed have no fingerprint and never match, so they + // re-summarize once and pick one up. + const inputFingerprint = computeInputFingerprint(compressed); + const existing = await kv.get( + KV.summaries, sessionId, - }); - return { - success: false, - error: "no_provider", - reason: - "No LLM provider key set; Summarize is a no-op. Set ANTHROPIC_API_KEY (or GEMINI/OPENROUTER/MINIMAX) in ~/.agentmemory/.env to enable.", - }; - } + ); + if (existing && existing.inputFingerprint === inputFingerprint) { + logger.info("Summary already up to date — skipping", { + sessionId, + observationCount: compressed.length, + }); + return { success: true, summary: existing, skipped: true }; + } - try { - // #783: chunk-level produceSummaryXml retries internally, but - // the final merge used to parse once and bail. Wrap the - // produce-and-parse pair in the same 2-attempt loop so a - // markdown-wrapped or otherwise wrapped response gets a - // second roll-of-the-dice instead of dropping the summary. - let summary: SessionSummary | null = null; - let response = ""; - let mode = "single"; - let chunks = 1; - for (let attempt = 1; attempt <= 2; attempt++) { - const produced = await produceSummaryXml( - provider, - compressed, + if (provider.name === "noop") { + logger.info("Summarize skipped — no LLM provider configured", { sessionId, - session.project, - ); - response = produced.response; - mode = produced.mode; - chunks = produced.chunks; - if (!response || !response.trim()) { - logger.warn("Empty provider response on summarize", { + }); + return { + success: false, + error: "no_provider", + reason: + "No LLM provider key set; Summarize is a no-op. Set ANTHROPIC_API_KEY (or GEMINI/OPENROUTER/MINIMAX) in ~/.agentmemory/.env to enable.", + }; + } + + try { + // #783: chunk-level produceSummaryXml retries internally, but + // the final merge used to parse once and bail. Wrap the + // produce-and-parse pair in the same 2-attempt loop so a + // markdown-wrapped or otherwise wrapped response gets a + // second roll-of-the-dice instead of dropping the summary. + let summary: SessionSummary | null = null; + let response = ""; + let mode = "single"; + let chunks = 1; + for (let attempt = 1; attempt <= 2; attempt++) { + const produced = await produceSummaryXml( + provider, + compressed, sessionId, - provider: provider.name, - mode, - chunks, - observationCount: compressed.length, - attempt, - }); - continue; + session.project, + ); + response = produced.response; + mode = produced.mode; + chunks = produced.chunks; + if (!response || !response.trim()) { + logger.warn("Empty provider response on summarize", { + sessionId, + provider: provider.name, + mode, + chunks, + observationCount: compressed.length, + attempt, + }); + continue; + } + summary = parseSummaryXml( + response, + sessionId, + session.project, + compressed.length, + ); + if (summary) break; + logger.warn("Failed to parse summary XML", { sessionId, attempt }); } - summary = parseSummaryXml( - response, - sessionId, - session.project, - compressed.length, + + if (!response || !response.trim()) { + const latencyMs = Date.now() - startMs; + if (metricsStore) { + await metricsStore.record("mem::summarize", latencyMs, false); + } + return { success: false, error: "empty_provider_response" }; + } + + if (!summary) { + const latencyMs = Date.now() - startMs; + if (metricsStore) { + await metricsStore.record("mem::summarize", latencyMs, false); + } + return { success: false, error: "parse_failed" }; + } + + const summaryForValidation = { + title: summary.title, + narrative: summary.narrative, + keyDecisions: summary.keyDecisions, + filesModified: summary.filesModified, + concepts: summary.concepts, + }; + const validation = validateOutput( + SummaryOutputSchema, + summaryForValidation, + "mem::summarize", ); - if (summary) break; - logger.warn("Failed to parse summary XML", { sessionId, attempt }); - } - if (!response || !response.trim()) { - const latencyMs = Date.now() - startMs; - if (metricsStore) { - await metricsStore.record("mem::summarize", latencyMs, false); + if (!validation.valid) { + const latencyMs = Date.now() - startMs; + if (metricsStore) { + await metricsStore.record("mem::summarize", latencyMs, false); + } + logger.warn("Summary validation failed", { + sessionId, + errors: validation.result.errors, + }); + return { success: false, error: "validation_failed" }; } - return { success: false, error: "empty_provider_response" }; - } - if (!summary) { + const qualityScore = scoreSummary(summaryForValidation); + + summary.inputFingerprint = inputFingerprint; + await kv.set(KV.summaries, sessionId, summary); + await safeAudit(kv, "compress", "mem::summarize", [sessionId], { + title: summary.title, + observationCount: compressed.length, + }); + const latencyMs = Date.now() - startMs; if (metricsStore) { - await metricsStore.record("mem::summarize", latencyMs, false); + await metricsStore.record( + "mem::summarize", + latencyMs, + true, + qualityScore, + ); } - return { success: false, error: "parse_failed" }; - } - const summaryForValidation = { - title: summary.title, - narrative: summary.narrative, - keyDecisions: summary.keyDecisions, - filesModified: summary.filesModified, - concepts: summary.concepts, - }; - const validation = validateOutput( - SummaryOutputSchema, - summaryForValidation, - "mem::summarize", - ); + logger.info("Session summarized", { + sessionId, + title: summary.title, + decisions: summary.keyDecisions.length, + qualityScore, + valid: validation.valid, + }); - if (!validation.valid) { + return { success: true, summary, qualityScore }; + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); const latencyMs = Date.now() - startMs; if (metricsStore) { await metricsStore.record("mem::summarize", latencyMs, false); } - logger.warn("Summary validation failed", { + logger.error("Summarize failed", { sessionId, - errors: validation.result.errors, + error: msg, }); - return { success: false, error: "validation_failed" }; + return { success: false, error: msg }; } + }; - const qualityScore = scoreSummary(summaryForValidation); - - await kv.set(KV.summaries, sessionId, summary); - await safeAudit(kv, "compress", "mem::summarize", [sessionId], { - title: summary.title, - observationCount: compressed.length, - }); - - const latencyMs = Date.now() - startMs; - if (metricsStore) { - await metricsStore.record( - "mem::summarize", - latencyMs, - true, - qualityScore, - ); - } - - logger.info("Session summarized", { - sessionId, - title: summary.title, - decisions: summary.keyDecisions.length, - qualityScore, - valid: validation.valid, - }); - - return { success: true, summary, qualityScore }; - } catch (err) { - const msg = err instanceof Error ? err.message : String(err); - const latencyMs = Date.now() - startMs; - if (metricsStore) { - await metricsStore.record("mem::summarize", latencyMs, false); - } - logger.error("Summarize failed", { - sessionId, - error: msg, - }); - return { success: false, error: msg }; - } + return withKeyedLock(`summarize:${sessionId}`, () => + raceLockTimeout(runSummarize(), getLockTimeoutMs(), sessionId), + ); }, ); } diff --git a/src/hooks/stop.ts b/src/hooks/stop.ts index 4103796e0..42c761466 100644 --- a/src/hooks/stop.ts +++ b/src/hooks/stop.ts @@ -39,13 +39,9 @@ async function main() { const sessionId = ((data.session_id || data.sessionId) as string) || "unknown"; - fetch(`${REST_URL}/agentmemory/summarize`, { - method: "POST", - headers: authHeaders(), - body: JSON.stringify({ sessionId }), - signal: AbortSignal.timeout(120000), - }).catch(() => {}); - + // Summarize is NOT called directly here: /agentmemory/session/end fans out + // event::session::stopped, whose handler already runs mem::summarize. + // Calling both used to double every summarize LLM run. fetch(`${REST_URL}/agentmemory/session/end`, { method: "POST", headers: authHeaders(), @@ -53,7 +49,7 @@ async function main() { signal: AbortSignal.timeout(5000), }).catch(() => {}); - setTimeout(() => process.exit(0), 1500).unref(); + setTimeout(() => process.exit(0), 500).unref(); } main(); diff --git a/src/state/schema.ts b/src/state/schema.ts index cb29d41ad..75c756605 100644 --- a/src/state/schema.ts +++ b/src/state/schema.ts @@ -22,6 +22,10 @@ export const KV = { // Single fixed key ("current") so writes are read-modify-write under // the same keyed mutex as graph-extract. graphSnapshot: "mem:graph:snapshot", + // Per-session fingerprint of the observation set last sent to + // mem::graph-extract on session stop; used to skip re-extraction when + // the session stops again without new observations. + graphExtractState: "mem:graph:extract-state", // #814 v2: targeted-lookup indexes so graph-extract never enumerates // the full nodes/edges scope. Each entry is a single small kv.get, // bounded payload — works at 75K+ nodes where kv.list would block diff --git a/src/triggers/events.ts b/src/triggers/events.ts index e38b58db4..001c35b14 100644 --- a/src/triggers/events.ts +++ b/src/triggers/events.ts @@ -4,6 +4,7 @@ import { KV, STREAM } from "../state/schema.js"; import { StateKV } from "../state/kv.js"; import { isReflectEnabled } from "../functions/slots.js"; import { isGraphExtractionEnabled } from "../config.js"; +import { computeInputFingerprint } from "../functions/input-fingerprint.js"; import { logger } from "../logger.js"; export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { @@ -67,11 +68,32 @@ export function registerEventTriggers(sdk: ISdk, kv: StateKV): void { ); const compressed = observations.filter((o) => o.title); if (compressed.length > 0) { - sdk.trigger({ - function_id: "mem::graph-extract", - payload: { observations: compressed }, - action: TriggerAction.Void(), - }); + // Sessions that stop repeatedly without new observations (e.g. + // heartbeat loops) would re-extract the identical set every time; + // skip when the fingerprint of the set is unchanged. + const fingerprint = computeInputFingerprint(compressed); + const prev = await kv + .get<{ fingerprint: string }>( + KV.graphExtractState, + data.sessionId, + ) + .catch(() => null); + if (prev && prev.fingerprint === fingerprint) { + logger.info("Graph extraction skipped — input unchanged", { + sessionId: data.sessionId, + observationCount: compressed.length, + }); + } else { + await kv.set(KV.graphExtractState, data.sessionId, { + fingerprint, + at: new Date().toISOString(), + }); + sdk.trigger({ + function_id: "mem::graph-extract", + payload: { observations: compressed }, + action: TriggerAction.Void(), + }); + } } } catch (err) { logger.warn("graph-extract trigger failed", { diff --git a/src/types.ts b/src/types.ts index 6797dfaf9..5cf5d9d2d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -114,6 +114,10 @@ export interface SessionSummary { filesModified: string[]; concepts: string[]; observationCount: number; + // Fingerprint of the exact observation set this summary was built from; + // used by mem::summarize to skip re-runs with identical input. Absent on + // summaries written by older versions. + inputFingerprint?: string; } export type HookType = diff --git a/test/session-end-triggers-graph.test.ts b/test/session-end-triggers-graph.test.ts index 22a780432..869a2856e 100644 --- a/test/session-end-triggers-graph.test.ts +++ b/test/session-end-triggers-graph.test.ts @@ -91,3 +91,28 @@ describe("agentmemory status no longer depends on /export (#666)", () => { expect(cli).toMatch(/memoriesRes\?\.latestCount\s*\?\?\s*memoriesRes\?\.total/); }); }); + +// Sessions that stop repeatedly without new observations (e.g. heartbeat +// loops) used to re-extract the identical observation set on every stop. +// The handler now fingerprints the set and skips when unchanged. +describe("event::session::stopped graph-extract dedup", () => { + const events = readFileSync("src/triggers/events.ts", "utf-8"); + + it("fingerprints the compressed set before triggering mem::graph-extract", () => { + expect(events).toMatch( + /const fingerprint = computeInputFingerprint\(compressed\);[\s\S]*?function_id:\s*"mem::graph-extract"/, + ); + }); + + it("skips extraction when the stored fingerprint matches", () => { + expect(events).toMatch( + /prev\.fingerprint === fingerprint[\s\S]*?Graph extraction skipped/, + ); + }); + + it("persists the fingerprint under KV.graphExtractState before triggering", () => { + expect(events).toMatch( + /kv\.set\(KV\.graphExtractState,\s*data\.sessionId,\s*\{\s*fingerprint,[\s\S]*?\}\);[\s\S]*?function_id:\s*"mem::graph-extract"/, + ); + }); +}); diff --git a/test/summarize.test.ts b/test/summarize.test.ts index 4723da878..64e16f5e1 100644 --- a/test/summarize.test.ts +++ b/test/summarize.test.ts @@ -480,3 +480,117 @@ describe("mem::summarize chunking", () => { expect(result.error).toBe("parse_failed"); }); }); + +describe("mem::summarize dedup (double-trigger guard)", () => { + it("skips the LLM when observation count is unchanged since last summary", async () => { + const provider = makeProvider([summaryXml({ title: "First run" })]); + const { handler } = await setupHandler({ + sessionId: "ses_dedup", + obsCount: 10, + provider, + }); + + const first: any = await handler({ sessionId: "ses_dedup" }); + expect(first.success).toBe(true); + expect(provider.calls).toHaveLength(1); + + const second: any = await handler({ sessionId: "ses_dedup" }); + expect(second.success).toBe(true); + expect(second.skipped).toBe(true); + expect(second.summary.title).toBe("First run"); + expect(provider.calls).toHaveLength(1); // no second LLM run + }); + + it("collapses concurrent duplicate calls into a single LLM run", async () => { + const provider = makeProvider([summaryXml({ title: "Only run" })]); + const { handler } = await setupHandler({ + sessionId: "ses_race", + obsCount: 10, + provider, + }); + + const [a, b]: any[] = await Promise.all([ + handler({ sessionId: "ses_race" }), + handler({ sessionId: "ses_race" }), + ]); + + expect(a.success).toBe(true); + expect(b.success).toBe(true); + expect(provider.calls).toHaveLength(1); + }); + + it("re-summarizes when the observation set changed but the count did not", async () => { + const provider = makeProvider([ + summaryXml({ title: "First run" }), + summaryXml({ title: "Second run" }), + ]); + const { handler, kv } = await setupHandler({ + sessionId: "ses_swap", + obsCount: 10, + provider, + }); + + await handler({ sessionId: "ses_swap" }); + + // evict/auto-forget can delete observations; a new one can then bring + // the count back to its previous value with different content. + await kv.delete("obs:ses_swap", "obs_0"); + const extra = makeObs(10, "ses_swap"); + await kv.set("obs:ses_swap", extra.id, extra); + + const result: any = await handler({ sessionId: "ses_swap" }); + expect(result.success).toBe(true); + expect(result.skipped).toBeUndefined(); + expect(result.summary.title).toBe("Second run"); + expect(provider.calls).toHaveLength(2); + }); + + it("releases the session lock when a run exceeds SUMMARIZE_LOCK_TIMEOUT_MS", async () => { + process.env.SUMMARIZE_LOCK_TIMEOUT_MS = "50"; + try { + const provider: MemoryProvider = { + name: "hung", + compress: async () => "", + summarize: () => new Promise(() => {}), // never settles + }; + const { handler } = await setupHandler({ + sessionId: "ses_hung", + obsCount: 10, + provider, + }); + + const result: any = await handler({ sessionId: "ses_hung" }); + expect(result.success).toBe(false); + expect(result.error).toBe("summarize_timeout"); + + // The lock must be free again: a follow-up call proceeds instead of + // queueing behind the hung run. + const again: any = await handler({ sessionId: "ses_hung" }); + expect(again.error).toBe("summarize_timeout"); + } finally { + delete process.env.SUMMARIZE_LOCK_TIMEOUT_MS; + } + }); + + it("re-summarizes when new observations arrived since last summary", async () => { + const provider = makeProvider([ + summaryXml({ title: "First run" }), + summaryXml({ title: "Second run" }), + ]); + const { handler, kv } = await setupHandler({ + sessionId: "ses_grow", + obsCount: 10, + provider, + }); + + await handler({ sessionId: "ses_grow" }); + const extra = makeObs(10, "ses_grow"); + await kv.set("obs:ses_grow", extra.id, extra); + + const result: any = await handler({ sessionId: "ses_grow" }); + expect(result.success).toBe(true); + expect(result.skipped).toBeUndefined(); + expect(result.summary.title).toBe("Second run"); + expect(provider.calls).toHaveLength(2); + }); +});