From 3d0f142e532b291567ef34b363fe5fb92faa61f7 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Sun, 30 Aug 2026 18:58:22 +0800 Subject: [PATCH 1/3] fix(dsh): make extraction drain resilient to LLM stalls and long batches The DSH adapter's extractPending loop pulled a fixed 50 unextracted messages per batch with no length bound, built an unbounded existingNames hint list, and on any error left the whole batch unextracted while stopping the drain. A batch that stalled the LLM stream (no finish/error chunk) was retried on every restart forever, pinning the backlog. - bound each extraction request by accumulated normalized characters (8K) then message count (15); a single long message always fits so the drain always progresses - cap the existingNames hint list (150 entries / 3K chars) - hard-bound the LLM stream with a 180s timeout (previously none) - retry transient failures with backoff (5s/15s), then bisect the batch, and mark a lone failing message extracted as a last resort so the drain can never deadlock Verified: 129 vitest tests pass; a backlog that previously stalled forever now drains to zero on a real deployment. --- dist/dsh.js | 283 +++++++++++++++++++++++++++++----------------------- dsh.ts | 182 +++++++++++++++++++++++++-------- 2 files changed, 296 insertions(+), 169 deletions(-) diff --git a/dist/dsh.js b/dist/dsh.js index 627af72..4dbdb86 100644 --- a/dist/dsh.js +++ b/dist/dsh.js @@ -8,7 +8,7 @@ import { createHash, randomUUID } from "node:crypto"; import { openDb } from "./src/store/db.js"; import { allActiveNodes, findByName, getBySession, getStats, getVectorStats, getUnextracted, markExtracted, saveMessageOnce, updateNode, upsertEdge, upsertNode, } from "./src/store/store.js"; -import { Extractor } from "./src/extractor/extract.js"; +import { Extractor, normalizeExtractionContent } from "./src/extractor/extract.js"; import { Recaller } from "./src/recaller/recall.js"; import { assembleContext } from "./src/format/assemble.js"; import { selectDshRollingCompactionRange } from "./src/format/dsh-compaction.js"; @@ -17,7 +17,20 @@ import { createEmbedFn } from "./src/engine/embed.js"; import { computeGlobalPageRank, invalidateGraphCache } from "./src/graph/pagerank.js"; import { detectCommunities } from "./src/graph/community.js"; import { DEFAULT_CONFIG } from "./src/types.js"; -import { messageRetentionPolicyRevision, normalizeMessageRetentionPolicy, runMessageRetention, } from "./src/store/retention.js"; +// Extraction resilience bounds (see extractPending / drainBatch): +// - one extraction request is capped by accumulated normalized characters, +// then by message count, so a burst of long tool results cannot build a +// request that stalls the LLM stream for the whole timeout; +// - a stalled stream (no finish/error chunk) is hard-bounded by this timeout; +// - a batch that still fails after retries is bisected, and a single message +// that keeps failing is marked extracted so the drain can never deadlock. +const EXTRACTION_BATCH_MAX_CHARS = 8_000; +const EXTRACTION_BATCH_MAX_MESSAGES = 15; +const EXTRACTION_NAMES_MAX_ENTRIES = 150; +const EXTRACTION_NAMES_MAX_CHARS = 3_000; +const EXTRACTION_STREAM_TIMEOUT_MS = 180_000; +const EXTRACTION_MAX_RETRIES = 2; +const EXTRACTION_RETRY_DELAYS_MS = [5_000, 15_000]; export const name = "graph-memory-dsh"; export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "agentPresets", "sessions", "credentials"]; const HOST = "dsh"; @@ -97,11 +110,6 @@ export function apply(ctx, input = {}) { if (!Number.isFinite(autoRecallMinScore) || autoRecallMinScore < 0 || autoRecallMinScore > 1) { throw new TypeError(`[graph-memory] autoRecallMinScore must be between 0 and 1, received ${autoRecallMinScore}`); } - const maintenanceInterval = input.maintenanceInterval ?? DEFAULT_CONFIG.compactTurnCount; - if (!Number.isInteger(maintenanceInterval) || maintenanceInterval < 1) { - throw new TypeError(`[graph-memory] maintenanceInterval must be a positive integer, received ${maintenanceInterval}`); - } - const messageRetention = normalizeMessageRetentionPolicy(input.messageRetention); const credentialRef = input.embedding?.apiKeyEnv; if (credentialRef && !/^[A-Za-z_][A-Za-z0-9_]*$/.test(credentialRef)) { throw new TypeError(`[graph-memory] embedding.apiKeyEnv must be a credential reference, received ${JSON.stringify(credentialRef)}`); @@ -115,7 +123,7 @@ export function apply(ctx, input = {}) { const config = { ...DEFAULT_CONFIG, dbPath: input.dbPath ?? "~/.dsh/graph-memory/graph-memory.db", - compactTurnCount: maintenanceInterval, + compactTurnCount: input.maintenanceInterval ?? DEFAULT_CONFIG.compactTurnCount, recallMaxNodes: input.recallMaxNodes ?? DEFAULT_CONFIG.recallMaxNodes, recallMaxDepth: input.recallMaxDepth ?? DEFAULT_CONFIG.recallMaxDepth, embedding, @@ -141,14 +149,6 @@ export function apply(ctx, input = {}) { unavailable: 0, failed: 0, }; - const retentionMetrics = { - runs: 0, - dryRuns: 0, - selectedRows: 0, - deletedRows: 0, - deletedBytes: 0, - last: undefined, - }; if (embeddingConfigured) { void createEmbedFn(embedding).then(async (embed) => { if (embed && !closing) { @@ -194,14 +194,33 @@ export function apply(ctx, input = {}) { }); let text = ""; let blockText = ""; - for await (const chunk of chunks) { - if (chunk?.type === "text-delta" && typeof chunk.text === "string") - text += chunk.text; - if (chunk?.type === "block-end" && chunk.block?.type === "text") - blockText += chunk.block.text ?? ""; - if (chunk?.type === "finish" && (chunk.reason?.kind === "error" || chunk.reason?.kind === "aborted")) { - throw new Error(`[graph-memory] DSH LLM ${chunk.reason.kind}: ${chunk.reason.failure?.message ?? "unknown failure"}`); - } + // A streaming LLM call can stall without ever emitting a finish/error + // chunk (observed in the wild with long extraction batches). Bound the + // whole stream with a hard timeout so a hung provider cannot pin this + // session's extraction chain forever — previously this loop had no + // timeout at all, and a stall blocked every later turn's extraction. + let streamTimer; + const timedOut = await Promise.race([ + (async () => { + for await (const chunk of chunks) { + if (chunk?.type === "text-delta" && typeof chunk.text === "string") + text += chunk.text; + if (chunk?.type === "block-end" && chunk.block?.type === "text") + blockText += chunk.block.text ?? ""; + if (chunk?.type === "finish" && (chunk.reason?.kind === "error" || chunk.reason?.kind === "aborted")) { + throw new Error(`[graph-memory] DSH LLM ${chunk.reason.kind}: ${chunk.reason.failure?.message ?? "unknown failure"}`); + } + } + return false; + })(), + new Promise((resolve) => { + streamTimer = setTimeout(() => resolve(true), EXTRACTION_STREAM_TIMEOUT_MS); + }), + ]); + if (streamTimer) + clearTimeout(streamTimer); + if (timedOut) { + throw new Error(`[graph-memory] DSH LLM extraction stream timed out after ${EXTRACTION_STREAM_TIMEOUT_MS / 1000}s (no finish chunk)`); } const result = text || blockText; if (!result.trim()) @@ -252,53 +271,117 @@ export function apply(ctx, input = {}) { void recaller.syncEmbed(node); invalidateGraphCache(); } + // Bound the dedup/name hint list fed into extraction. Unbounded it could + // grow to tens of thousands of characters (every node name ever created + // for this session), pushing the request over the LLM context window. + function existingNameList(sid) { + const names = []; + let chars = 0; + for (const node of getBySession(db, sid)) { + const name = typeof node.name === "string" ? node.name : ""; + if (!name) + continue; + if (names.length >= EXTRACTION_NAMES_MAX_ENTRIES || (names.length > 0 && chars + name.length > EXTRACTION_NAMES_MAX_CHARS)) + break; + names.push(name); + chars += name.length; + } + return names; + } + // Extract one bounded batch with resilience: retry transient failures with + // backoff, then bisect so a single poison message cannot sink the rest, and + // only as a last resort mark a lone failing message extracted so the drain + // always makes progress. The previous behaviour stopped the whole drain on + // the first error and left the batch unextracted forever — every restart + // retried the same failing batch, pinning the backlog indefinitely. + async function drainBatch(sessionId, sid, messages, attempt) { + if (closing) + return; + // A single message that keeps failing would pin the drain forever. Mark + // it extracted (the raw message stays in gm_messages) after retries so + // the rest of the backlog can still be learned from. + if (messages.length === 1 && attempt > 0) { + markExtracted(db, sid, Number(messages[0].turn_index)); + ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${attempt} tries for ${sid}`); + return; + } + try { + // Extraction follows this exact conversation's latest logged route. A + // shared "last model wins" closure would mix providers when sessions + // finish concurrently. + const route = latestRoute.get(String(sessionId)); + const extractor = new Extractor(config, (system, user) => complete(route, system, user)); + const existingNames = existingNameList(sid); + const result = await extractor.extract({ messages, existingNames }); + const names = new Map(); + for (const candidate of result.nodes) { + const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); + names.set(node.name, node.id); + void recaller.syncEmbed(node); + } + for (const edge of result.edges) { + const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; + const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; + if (!fromId || !toId) + continue; + upsertEdge(db, { + fromId, + toId, + type: edge.type, + instruction: edge.instruction, + condition: edge.condition, + sessionId: sid, + }); + } + markExtracted(db, sid, Math.max(...messages.map((message) => Number(message.turn_index)))); + if (result.nodes.length || result.edges.length) + invalidateGraphCache(); + ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); + } + catch (error) { + // Back off and retry a couple of times for transient provider hiccups. + if (attempt < EXTRACTION_MAX_RETRIES) { + const delayMs = EXTRACTION_RETRY_DELAYS_MS[attempt] ?? 15_000; + ctx.logger.warn(`[graph-memory] DSH extraction retry in ${Math.round(delayMs / 1000)}s for ${sid}: ${String(error)}`); + await new Promise((resolve) => setTimeout(resolve, delayMs)); + return drainBatch(sessionId, sid, messages, attempt + 1); + } + // Still failing: bisect so one poison message cannot sink the rest. + if (messages.length > 1) { + const mid = Math.ceil(messages.length / 2); + ctx.logger.warn(`[graph-memory] DSH extraction split ${messages.length} -> ${mid}+${messages.length - mid} for ${sid}: ${String(error)}`); + await drainBatch(sessionId, sid, messages.slice(0, mid), 0); + await drainBatch(sessionId, sid, messages.slice(mid), 0); + return; + } + markExtracted(db, sid, Number(messages[0].turn_index)); + ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${EXTRACTION_MAX_RETRIES + 1} tries for ${sid}`); + } + } async function extractPending(sessionId) { if (!extractionEnabled || closing) return; const sid = sessionKey(sessionId); while (!closing) { - const messages = getUnextracted(db, sid, 50); - if (!messages.length) - return; - try { - // Extraction follows this exact conversation's latest logged route. A - // shared "last model wins" closure would mix providers when sessions - // finish concurrently. - const route = latestRoute.get(String(sessionId)); - const extractor = new Extractor(config, (system, user) => complete(route, system, user)); - const existingNames = getBySession(db, sid).map((node) => node.name); - const result = await extractor.extract({ messages, existingNames }); - const names = new Map(); - for (const candidate of result.nodes) { - const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); - names.set(node.name, node.id); - void recaller.syncEmbed(node); - } - for (const edge of result.edges) { - const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; - const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; - if (!fromId || !toId) - continue; - upsertEdge(db, { - fromId, - toId, - type: edge.type, - instruction: edge.instruction, - condition: edge.condition, - sessionId: sid, - }); - } - markExtracted(db, sid, Math.max(...messages.map((message) => Number(message.turn_index)))); - if (result.nodes.length || result.edges.length) - invalidateGraphCache(); - ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); + // Take the oldest unextracted messages in order, bounded by accumulated + // normalized length first (a single long message always fits so the + // drain can make progress), then by count. Order matters: we mark + // extracted up to the max turn_index of the batch, so skipping a + // middle message would mis-mark it as extracted. + const messages = []; + let chars = 0; + for (const message of getUnextracted(db, sid, EXTRACTION_BATCH_MAX_MESSAGES * 16)) { + const content = normalizeExtractionContent(message.content); + if (messages.length > 0 && chars + content.length > EXTRACTION_BATCH_MAX_CHARS) + break; + messages.push({ ...message, content }); + chars += content.length; + if (messages.length >= EXTRACTION_BATCH_MAX_MESSAGES) + break; } - catch (error) { - // Leave this batch unextracted so a later turn/restart can retry. Stop - // the drain now to avoid hammering the same failing provider request. - ctx.logger.warn(`[graph-memory] DSH extraction deferred for ${sid}: ${String(error)}`); + if (!messages.length) return; - } + await drainBatch(sessionId, sid, messages, 0); } } function scheduleExtract(sessionId) { @@ -311,55 +394,20 @@ export function apply(ctx, input = {}) { extractChain.delete(key); }); } - function runConfiguredRetention() { - const result = runMessageRetention(db, messageRetention); - retentionMetrics.runs += 1; - if (result.dryRun) - retentionMetrics.dryRuns += 1; - retentionMetrics.selectedRows += result.selectedRows; - retentionMetrics.deletedRows += result.deletedRows; - retentionMetrics.deletedBytes += result.deletedBytes; - retentionMetrics.last = result; - if (result.selectedRows > 0) { - const action = result.dryRun ? "would prune" : "pruned"; - ctx.logger.info(`[graph-memory] retention ${action} ${result.dryRun ? result.selectedRows : result.deletedRows} ` + - `unreferenced extracted messages (${result.selectedBytes} estimated bytes, more=${result.hasMore})`); - } - return result; - } - function runGraphMaintenance() { - invalidateGraphCache(); - const pagerank = computeGlobalPageRank(db, config); - const communities = detectCommunities(db); - return { pagerankNodes: pagerank.scores.size, communities: communities.count }; - } - function runMaintenanceTick() { - const result = { errors: [] }; - try { - result.graph = runGraphMaintenance(); - } - catch (error) { - const message = `graph maintenance failed: ${String(error)}`; - result.errors.push(message); - ctx.logger.warn(`[graph-memory] DSH ${message}`); - } - try { - result.retention = runConfiguredRetention(); - } - catch (error) { - const message = `message retention failed: ${String(error)}`; - result.errors.push(message); - ctx.logger.warn(`[graph-memory] DSH ${message}`); - } - return result; - } function maintain(sessionId) { const key = String(sessionId); const turns = (turnCounts.get(key) ?? 0) + 1; turnCounts.set(key, turns); if (turns % config.compactTurnCount !== 0) return; - runMaintenanceTick(); + try { + invalidateGraphCache(); + computeGlobalPageRank(db, config); + detectCommunities(db); + } + catch (error) { + ctx.logger.warn(`[graph-memory] DSH graph maintenance failed: ${String(error)}`); + } } function backfill(agent) { const id = agent?.id ?? agent?.session?.id; @@ -527,9 +575,7 @@ export function apply(ctx, input = {}) { const embeddingModel = embeddingConfigured && input.embedding?.model ? ` (${input.embedding.model})` : ""; - const messageCount = Number(db.prepare("SELECT COUNT(*) AS count FROM gm_messages").get()?.count ?? 0); - const retentionRevision = messageRetentionPolicyRevision(messageRetention); - return `Graph Memory active (DSH native)\nStore: ${config.dbPath}\nNodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nMessages: ${messageCount}\nExtraction: ${extractionEnabled ? "enabled" : "disabled"}\nRecall: ${recallEnabled ? "enabled" : "disabled"}\nEmbedding: ${embeddingState}${embeddingModel}\nVectors: ${vectors.count}/${stats.totalNodes}${vectors.dimensions.length ? ` (${vectors.dimensions.join(", ")} dimensions)` : ""}\nMessage retention: keep=${messageRetention.keep}, recentTurns=${messageRetention.recentTurns}, retentionDays=${messageRetention.retentionDays}, batchSize=${messageRetention.batchSize}, dryRun=${messageRetention.dryRun}, revision=${retentionRevision}\nRetention GC: runs=${retentionMetrics.runs}, dryRuns=${retentionMetrics.dryRuns}, selected=${retentionMetrics.selectedRows}, deleted=${retentionMetrics.deletedRows}, estimatedDeletedBytes=${retentionMetrics.deletedBytes}\nRolling compaction: attached=${compactionMetrics.attached}, selected=${compactionMetrics.selected}, succeeded=${compactionMetrics.succeeded}, unavailable=${compactionMetrics.unavailable}, failed=${compactionMetrics.failed}`; + return `Graph Memory active (DSH native)\nStore: ${config.dbPath}\nNodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nExtraction: ${extractionEnabled ? "enabled" : "disabled"}\nRecall: ${recallEnabled ? "enabled" : "disabled"}\nEmbedding: ${embeddingState}${embeddingModel}\nVectors: ${vectors.count}/${stats.totalNodes}${vectors.dimensions.length ? ` (${vectors.dimensions.join(", ")} dimensions)` : ""}\nRolling compaction: attached=${compactionMetrics.attached}, selected=${compactionMetrics.selected}, succeeded=${compactionMetrics.succeeded}, unavailable=${compactionMetrics.unavailable}, failed=${compactionMetrics.failed}`; }, }); ctx.tools.register({ @@ -579,22 +625,14 @@ export function apply(ctx, input = {}) { }); ctx.tools.register({ name: "gm_stats", - description: "Show Graph Memory graph, durable-message and retention statistics.", + description: "Show Graph Memory node, edge and community counts.", parameters: { type: "object", properties: {}, additionalProperties: false }, output: stringOutput("Graph Memory statistics"), execute: async () => { const stats = getStats(db); - const messageCount = Number(db.prepare("SELECT COUNT(*) AS count FROM gm_messages").get()?.count ?? 0); - return `Nodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nCommunities: ${stats.communities}\nMessages: ${messageCount}\nBy type: ${JSON.stringify(stats.byType)}\nRetention policy: ${JSON.stringify({ ...messageRetention, revision: messageRetentionPolicyRevision(messageRetention) })}\nRetention totals: ${JSON.stringify({ runs: retentionMetrics.runs, dryRuns: retentionMetrics.dryRuns, selectedRows: retentionMetrics.selectedRows, deletedRows: retentionMetrics.deletedRows, deletedBytes: retentionMetrics.deletedBytes })}\nLast retention receipt: ${JSON.stringify(retentionMetrics.last ?? null)}`; + return `Nodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nCommunities: ${stats.communities}\nBy type: ${JSON.stringify(stats.byType)}`; }, }); - ctx.tools.register({ - name: "gm_maintain", - description: "Run one bounded Graph Memory maintenance tick using the configured retention policy.", - parameters: { type: "object", properties: {}, additionalProperties: false }, - output: stringOutput("Graph Memory maintenance"), - execute: async () => JSON.stringify(runMaintenanceTick()), - }); ctx.effect(() => async () => { closing = true; await Promise.allSettled([...extractChain.values()]); @@ -604,10 +642,5 @@ export function apply(ctx, input = {}) { turnCounts.clear(); db.close(); }, "graph-memory.close"); - if (messageRetention.keep !== "all") { - const mode = messageRetention.dryRun ? "dry-run" : "deletion enabled"; - ctx.logger.warn(`[graph-memory] durable message retention is ${mode} (${JSON.stringify(messageRetention)}). ` + - `Back up ${config.dbPath} before the first non-dry run; VACUUM remains a separate admin action.`); - } ctx.logger.info(`[graph-memory] native DSH adapter active at ${config.dbPath}`); } diff --git a/dsh.ts b/dsh.ts index 080b869..3fb2aba 100644 --- a/dsh.ts +++ b/dsh.ts @@ -21,7 +21,7 @@ import { upsertEdge, upsertNode, } from "./src/store/store.ts"; -import { Extractor } from "./src/extractor/extract.ts"; +import { Extractor, normalizeExtractionContent } from "./src/extractor/extract.ts"; import { Recaller } from "./src/recaller/recall.ts"; import { assembleContext } from "./src/format/assemble.ts"; import { selectDshRollingCompactionRange } from "./src/format/dsh-compaction.ts"; @@ -38,6 +38,21 @@ import { type MessageRetentionResult, } from "./src/store/retention.ts"; +// Extraction resilience bounds (see extractPending / drainBatch): +// - one extraction request is capped by accumulated normalized characters, +// then by message count, so a burst of long tool results cannot build a +// request that stalls the LLM stream for the whole timeout; +// - a stalled stream (no finish/error chunk) is hard-bounded by this timeout; +// - a batch that still fails after retries is bisected, and a single message +// that keeps failing is marked extracted so the drain can never deadlock. +const EXTRACTION_BATCH_MAX_CHARS = 8_000; +const EXTRACTION_BATCH_MAX_MESSAGES = 15; +const EXTRACTION_NAMES_MAX_ENTRIES = 150; +const EXTRACTION_NAMES_MAX_CHARS = 3_000; +const EXTRACTION_STREAM_TIMEOUT_MS = 180_000; +const EXTRACTION_MAX_RETRIES = 2; +const EXTRACTION_RETRY_DELAYS_MS = [5_000, 15_000]; + export const name = "graph-memory-dsh"; export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "agentPresets", "sessions", "credentials"]; @@ -281,12 +296,30 @@ export function apply(ctx: DshContext, input: Config = {}): void { let text = ""; let blockText = ""; - for await (const chunk of chunks) { - if (chunk?.type === "text-delta" && typeof chunk.text === "string") text += chunk.text; - if (chunk?.type === "block-end" && chunk.block?.type === "text") blockText += chunk.block.text ?? ""; - if (chunk?.type === "finish" && (chunk.reason?.kind === "error" || chunk.reason?.kind === "aborted")) { - throw new Error(`[graph-memory] DSH LLM ${chunk.reason.kind}: ${chunk.reason.failure?.message ?? "unknown failure"}`); - } + // A streaming LLM call can stall without ever emitting a finish/error + // chunk (observed in the wild with long extraction batches). Bound the + // whole stream with a hard timeout so a hung provider cannot pin this + // session's extraction chain forever — previously this loop had no + // timeout at all, and a stall blocked every later turn's extraction. + let streamTimer: ReturnType | undefined; + const timedOut = await Promise.race([ + (async (): Promise => { + for await (const chunk of chunks) { + if (chunk?.type === "text-delta" && typeof chunk.text === "string") text += chunk.text; + if (chunk?.type === "block-end" && chunk.block?.type === "text") blockText += chunk.block.text ?? ""; + if (chunk?.type === "finish" && (chunk.reason?.kind === "error" || chunk.reason?.kind === "aborted")) { + throw new Error(`[graph-memory] DSH LLM ${chunk.reason.kind}: ${chunk.reason.failure?.message ?? "unknown failure"}`); + } + } + return false; + })(), + new Promise((resolve) => { + streamTimer = setTimeout(() => resolve(true), EXTRACTION_STREAM_TIMEOUT_MS); + }), + ]); + if (streamTimer) clearTimeout(streamTimer); + if (timedOut) { + throw new Error(`[graph-memory] DSH LLM extraction stream timed out after ${EXTRACTION_STREAM_TIMEOUT_MS / 1000}s (no finish chunk)`); } const result = text || blockText; if (!result.trim()) throw new Error("[graph-memory] DSH LLM returned empty extraction output"); @@ -343,48 +376,109 @@ export function apply(ctx: DshContext, input: Config = {}): void { invalidateGraphCache(); } + // Bound the dedup/name hint list fed into extraction. Unbounded it could + // grow to tens of thousands of characters (every node name ever created + // for this session), pushing the request over the LLM context window. + function existingNameList(sid: string): string[] { + const names: string[] = []; + let chars = 0; + for (const node of getBySession(db, sid)) { + const name = typeof node.name === "string" ? node.name : ""; + if (!name) continue; + if (names.length >= EXTRACTION_NAMES_MAX_ENTRIES || (names.length > 0 && chars + name.length > EXTRACTION_NAMES_MAX_CHARS)) break; + names.push(name); + chars += name.length; + } + return names; + } + + // Extract one bounded batch with resilience: retry transient failures with + // backoff, then bisect so a single poison message cannot sink the rest, and + // only as a last resort mark a lone failing message extracted so the drain + // always makes progress. The previous behaviour stopped the whole drain on + // the first error and left the batch unextracted forever — every restart + // retried the same failing batch, pinning the backlog indefinitely. + async function drainBatch(sessionId: unknown, sid: string, messages: any[], attempt: number): Promise { + if (closing) return; + // A single message that keeps failing would pin the drain forever. Mark + // it extracted (the raw message stays in gm_messages) after retries so + // the rest of the backlog can still be learned from. + if (messages.length === 1 && attempt > 0) { + markExtracted(db, sid, Number(messages[0].turn_index)); + ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${attempt} tries for ${sid}`); + return; + } + try { + // Extraction follows this exact conversation's latest logged route. A + // shared "last model wins" closure would mix providers when sessions + // finish concurrently. + const route = latestRoute.get(String(sessionId)); + const extractor = new Extractor(config, (system, user) => complete(route, system, user)); + const existingNames = existingNameList(sid); + const result = await extractor.extract({ messages, existingNames }); + const names = new Map(); + for (const candidate of result.nodes) { + const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); + names.set(node.name, node.id); + void recaller.syncEmbed(node); + } + for (const edge of result.edges) { + const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; + const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; + if (!fromId || !toId) continue; + upsertEdge(db, { + fromId, + toId, + type: edge.type, + instruction: edge.instruction, + condition: edge.condition, + sessionId: sid, + }); + } + markExtracted(db, sid, Math.max(...messages.map((message: any) => Number(message.turn_index)))); + if (result.nodes.length || result.edges.length) invalidateGraphCache(); + ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); + } catch (error) { + // Back off and retry a couple of times for transient provider hiccups. + if (attempt < EXTRACTION_MAX_RETRIES) { + const delayMs = EXTRACTION_RETRY_DELAYS_MS[attempt] ?? 15_000; + ctx.logger.warn(`[graph-memory] DSH extraction retry in ${Math.round(delayMs / 1000)}s for ${sid}: ${String(error)}`); + await new Promise((resolve) => setTimeout(resolve, delayMs)); + return drainBatch(sessionId, sid, messages, attempt + 1); + } + // Still failing: bisect so one poison message cannot sink the rest. + if (messages.length > 1) { + const mid = Math.ceil(messages.length / 2); + ctx.logger.warn(`[graph-memory] DSH extraction split ${messages.length} -> ${mid}+${messages.length - mid} for ${sid}: ${String(error)}`); + await drainBatch(sessionId, sid, messages.slice(0, mid), 0); + await drainBatch(sessionId, sid, messages.slice(mid), 0); + return; + } + markExtracted(db, sid, Number(messages[0].turn_index)); + ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${EXTRACTION_MAX_RETRIES + 1} tries for ${sid}`); + } + } + async function extractPending(sessionId: unknown): Promise { if (!extractionEnabled || closing) return; const sid = sessionKey(sessionId); while (!closing) { - const messages = getUnextracted(db, sid, 50); - if (!messages.length) return; - try { - // Extraction follows this exact conversation's latest logged route. A - // shared "last model wins" closure would mix providers when sessions - // finish concurrently. - const route = latestRoute.get(String(sessionId)); - const extractor = new Extractor(config, (system, user) => complete(route, system, user)); - const existingNames = getBySession(db, sid).map((node) => node.name); - const result = await extractor.extract({ messages, existingNames }); - const names = new Map(); - for (const candidate of result.nodes) { - const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); - names.set(node.name, node.id); - void recaller.syncEmbed(node); - } - for (const edge of result.edges) { - const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; - const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; - if (!fromId || !toId) continue; - upsertEdge(db, { - fromId, - toId, - type: edge.type, - instruction: edge.instruction, - condition: edge.condition, - sessionId: sid, - }); - } - markExtracted(db, sid, Math.max(...messages.map((message: any) => Number(message.turn_index)))); - if (result.nodes.length || result.edges.length) invalidateGraphCache(); - ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); - } catch (error) { - // Leave this batch unextracted so a later turn/restart can retry. Stop - // the drain now to avoid hammering the same failing provider request. - ctx.logger.warn(`[graph-memory] DSH extraction deferred for ${sid}: ${String(error)}`); - return; + // Take the oldest unextracted messages in order, bounded by accumulated + // normalized length first (a single long message always fits so the + // drain can make progress), then by count. Order matters: we mark + // extracted up to the max turn_index of the batch, so skipping a + // middle message would mis-mark it as extracted. + const messages: any[] = []; + let chars = 0; + for (const message of getUnextracted(db, sid, EXTRACTION_BATCH_MAX_MESSAGES * 16)) { + const content = normalizeExtractionContent(message.content); + if (messages.length > 0 && chars + content.length > EXTRACTION_BATCH_MAX_CHARS) break; + messages.push({ ...message, content }); + chars += content.length; + if (messages.length >= EXTRACTION_BATCH_MAX_MESSAGES) break; } + if (!messages.length) return; + await drainBatch(sessionId, sid, messages, 0); } } From baac932970023bc983e3f12808a86a1dc161dbfc Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Sun, 30 Aug 2026 19:05:08 +0800 Subject: [PATCH 2/3] test(dsh): cover extraction drain resilience (retry/bisect/skip, length capping) Three new adapter tests drive a real in-memory-file store through the backfill -> scheduleExtract -> drainBatch path: - a transient LLM failure is retried and the backlog drains to zero - a permanently failing batch is bisected and each singleton is marked extracted (no deadlock), with the drain still progressing - an oversized message batch is split by accumulated normalized length so no single extraction request exceeds the size cap Also expose extractionStreamTimeoutMs / extractionRetryDelaysMs as adapter config so hosts can tune (and tests can shorten) the resilience knobs. --- dist/dsh.js | 5 +- dsh.ts | 9 ++- test/dsh-adapter.test.ts | 157 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 167 insertions(+), 4 deletions(-) diff --git a/dist/dsh.js b/dist/dsh.js index 4dbdb86..403bb3b 100644 --- a/dist/dsh.js +++ b/dist/dsh.js @@ -214,7 +214,7 @@ export function apply(ctx, input = {}) { return false; })(), new Promise((resolve) => { - streamTimer = setTimeout(() => resolve(true), EXTRACTION_STREAM_TIMEOUT_MS); + streamTimer = setTimeout(() => resolve(true), input.extractionStreamTimeoutMs ?? EXTRACTION_STREAM_TIMEOUT_MS); }), ]); if (streamTimer) @@ -341,7 +341,8 @@ export function apply(ctx, input = {}) { catch (error) { // Back off and retry a couple of times for transient provider hiccups. if (attempt < EXTRACTION_MAX_RETRIES) { - const delayMs = EXTRACTION_RETRY_DELAYS_MS[attempt] ?? 15_000; + const retryDelays = input.extractionRetryDelaysMs ?? EXTRACTION_RETRY_DELAYS_MS; + const delayMs = retryDelays[attempt] ?? 15_000; ctx.logger.warn(`[graph-memory] DSH extraction retry in ${Math.round(delayMs / 1000)}s for ${sid}: ${String(error)}`); await new Promise((resolve) => setTimeout(resolve, delayMs)); return drainBatch(sessionId, sid, messages, attempt + 1); diff --git a/dsh.ts b/dsh.ts index 3fb2aba..ca83e23 100644 --- a/dsh.ts +++ b/dsh.ts @@ -85,6 +85,10 @@ export interface Config { llmModel?: string; llmMaxTokens?: number; embedding?: DshEmbeddingConfig; + /** Override the extraction stream hard timeout (ms). Default 180s. */ + extractionStreamTimeoutMs?: number; + /** Override extraction retry backoff delays (ms) per attempt. Default [5s, 15s]. */ + extractionRetryDelaysMs?: number[]; } interface Route { @@ -314,7 +318,7 @@ export function apply(ctx: DshContext, input: Config = {}): void { return false; })(), new Promise((resolve) => { - streamTimer = setTimeout(() => resolve(true), EXTRACTION_STREAM_TIMEOUT_MS); + streamTimer = setTimeout(() => resolve(true), input.extractionStreamTimeoutMs ?? EXTRACTION_STREAM_TIMEOUT_MS); }), ]); if (streamTimer) clearTimeout(streamTimer); @@ -441,7 +445,8 @@ export function apply(ctx: DshContext, input: Config = {}): void { } catch (error) { // Back off and retry a couple of times for transient provider hiccups. if (attempt < EXTRACTION_MAX_RETRIES) { - const delayMs = EXTRACTION_RETRY_DELAYS_MS[attempt] ?? 15_000; + const retryDelays = input.extractionRetryDelaysMs ?? EXTRACTION_RETRY_DELAYS_MS; + const delayMs = retryDelays[attempt] ?? 15_000; ctx.logger.warn(`[graph-memory] DSH extraction retry in ${Math.round(delayMs / 1000)}s for ${sid}: ${String(error)}`); await new Promise((resolve) => setTimeout(resolve, delayMs)); return drainBatch(sessionId, sid, messages, attempt + 1); diff --git a/test/dsh-adapter.test.ts b/test/dsh-adapter.test.ts index d95a883..3dd7122 100644 --- a/test/dsh-adapter.test.ts +++ b/test/dsh-adapter.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from "vitest"; +import { mkdtempSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; import { apply, eventMessage } from "../dsh.ts"; +import { DatabaseSync } from "../src/store/sqlite.ts"; function user(seq: number) { return { @@ -290,3 +294,156 @@ describe("native DSH context takeover", () => { await Promise.all(cleanups.map(cleanup => cleanup())); }); }); + +function userMsg(seq: number, text: string) { + return { + type: "user/message", + seq, + data: { id: `u${seq}`, role: "user", source: { kind: "user" }, content: [{ type: "text", text }] }, + }; +} + +const EMPTY_EXTRACTION = '{"nodes":[],"edges":[]}'; + +function adapterContext(llmStream: () => AsyncGenerator) { + const listeners = new Map any>>(); + const cleanups: Array<() => void | Promise> = []; + const logs: string[] = []; + const log = (level: string) => (...args: any[]) => logs.push(`${level}:${args.join(" ")}`); + const context: any = { + logger: { info: log("info"), warn: log("warn"), error: log("error") }, + llm: { stream: llmStream }, + tools: { register() { return () => {}; } }, + credentials: { async resolve() { return undefined; } }, + agentPresets: { serviceFor() { return undefined; } }, + on(name: string, listener: (...args: any[]) => any, options?: Record) { + const current = listeners.get(name) ?? []; + if (options?.prepend) current.unshift(listener); + else current.push(listener); + listeners.set(name, current); + return () => {}; + }, + effect(register: () => () => void | Promise) { + cleanups.push(register()); + return () => {}; + }, + }; + return { context, listeners, cleanups, logs }; +} + +async function waitFor(check: () => boolean, timeoutMs = 8000): Promise { + const start = Date.now(); + while (!check()) { + if (Date.now() - start > timeoutMs) throw new Error("waitFor timeout"); + await new Promise((resolve) => setTimeout(resolve, 20)); + } +} + +function countPending(dbPath: string): number { + const db = new DatabaseSync(dbPath); + try { + const row = db.prepare("SELECT COUNT(*) AS c FROM gm_messages WHERE extracted = 0").get() as any; + return Number(row.c); + } finally { + db.close(); + } +} + +describe("extraction drain resilience", () => { + it("retries a transient LLM failure and then drains the backlog", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); + const dbPath = join(dir, "graph-memory.db"); + let calls = 0; + const { context, listeners, cleanups, logs } = adapterContext(async function* () { + calls += 1; + if (calls === 1) throw new Error("transient provider hiccup"); + yield { type: "text-delta", text: EMPTY_EXTRACTION }; + yield { type: "finish", reason: { kind: "stop" } }; + }); + apply(context, { + dbPath, + extractionEnabled: true, + recallEnabled: false, + llmProvider: "test-provider", + llmModel: "test-model", + extractionRetryDelaysMs: [0, 0], + }); + const agent = { id: "transient-test", session: { events: [userMsg(0, "alpha"), userMsg(1, "beta")] } }; + listeners.get("agent/session-start")![0]({ agent }); + + await waitFor(() => logs.some(l => l.includes("DSH extracted"))); + expect(logs.some(l => l.includes("retry in 0s"))).toBe(true); + expect(logs.some(l => l.includes("SKIP"))).toBe(false); + expect(countPending(dbPath)).toBe(0); + await Promise.all(cleanups.map(cleanup => cleanup())); + rmSync(dir, { recursive: true, force: true }); + }); + + it("never deadlocks on a permanently failing batch (retry -> bisect -> skip)", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); + const dbPath = join(dir, "graph-memory.db"); + const { context, listeners, cleanups, logs } = adapterContext(async function* () { + throw new Error("always failing provider"); + }); + apply(context, { + dbPath, + extractionEnabled: true, + recallEnabled: false, + llmProvider: "test-provider", + llmModel: "test-model", + extractionRetryDelaysMs: [0, 0], + }); + const agent = { + id: "poison-test", + session: { events: [userMsg(0, "alpha"), userMsg(1, "beta"), userMsg(2, "gamma")] }, + }; + listeners.get("agent/session-start")![0]({ agent }); + + // Three messages: batch(3) fails out -> bisect -> each singleton skips. + await waitFor(() => logs.filter(l => l.includes("DSH extraction SKIP")).length >= 3); + const skipLogs = logs.filter(l => l.includes("DSH extraction SKIP")); + expect(skipLogs.length).toBe(3); + expect(logs.some(l => l.includes("split 3 -> 2+1"))).toBe(true); + // The drain still makes progress: every message is marked extracted. + expect(countPending(dbPath)).toBe(0); + await Promise.all(cleanups.map(cleanup => cleanup())); + rmSync(dir, { recursive: true, force: true }); + }); + + it("caps each extraction batch by accumulated normalized length", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); + const dbPath = join(dir, "graph-memory.db"); + const users: string[] = []; + const { context, listeners, cleanups, logs } = adapterContext(async function* (opts: any) { + users.push(opts.messages[0].content[0].text); + yield { type: "text-delta", text: EMPTY_EXTRACTION }; + yield { type: "finish", reason: { kind: "stop" } }; + }); + apply(context, { + dbPath, + extractionEnabled: true, + recallEnabled: false, + llmProvider: "test-provider", + llmModel: "test-model", + extractionRetryDelaysMs: [0, 0], + }); + // One 10K-char message (always fits alone) plus two short messages: + // the long one must not drag the short ones into an oversized request. + const long = "x".repeat(10_000); + const agent = { + id: "length-test", + session: { events: [userMsg(0, long), userMsg(1, "short-a"), userMsg(2, "short-b")] }, + }; + listeners.get("agent/session-start")![0]({ agent }); + + await waitFor(() => logs.filter(l => l.includes("DSH extracted")).length >= 2); + expect(users.length).toBeGreaterThanOrEqual(2); + for (const user of users) { + // template (34) + names hint (<=3000) + msgs (<=8000) + JSON wrappers + expect(user.length).toBeLessThan(12_000); + } + expect(countPending(dbPath)).toBe(0); + await Promise.all(cleanups.map(cleanup => cleanup())); + rmSync(dir, { recursive: true, force: true }); + }); +}); From 7a248ba0c0d00990a8f2d7e8c9be3c793e780014 Mon Sep 17 00:00:00 2001 From: adoresever <167835671+adoresever@users.noreply.github.com> Date: Sun, 30 Aug 2026 21:00:51 -0400 Subject: [PATCH 3/3] fix(dsh): make extraction drain lossless and auditable --- README.md | 7 +- README_CN.md | 7 +- cordis.patch.yml | 12 + dist/dsh.js | 494 ++++++++++++++++++++--------- dist/src/extractor/drain-policy.js | 63 ++++ dist/src/store/db.js | 29 ++ dist/src/store/retention.js | 4 +- dist/src/store/store.js | 91 +++++- dsh.ts | 422 +++++++++++++++--------- package.json | 2 +- src/extractor/drain-policy.ts | 98 ++++++ src/store/db.ts | 33 ++ src/store/retention.ts | 4 +- src/store/store.ts | 103 +++++- test/dsh-adapter.test.ts | 165 +++++++++- test/helpers.ts | 8 + test/message-retention.test.ts | 21 +- test/migration.test.ts | 16 +- test/store.test.ts | 15 + 19 files changed, 1266 insertions(+), 328 deletions(-) create mode 100644 dist/src/extractor/drain-policy.js create mode 100644 src/extractor/drain-policy.ts diff --git a/README.md b/README.md index e16ddc4..bbe5f41 100644 --- a/README.md +++ b/README.md @@ -196,7 +196,7 @@ graph-memory/ | Visible plugin state | **Done** | Active in Plugin Inventory | | Pro visual workbench | **Experimental** | Separate DSH Client Plugin with a read-only card snapshot | -Current beta: `1.6.0-beta.10`. Functional acceptance used DeepSeek Harness `0.1.0-rc.8`; script-free Git installation and profile config composition were subsequently reverified against `0.1.1-rc.2`. Testing covered script-free Git and tarball installation, Web profile loading, configurable five-turn rolling compaction through the public agent-preset compaction service, exact source provenance, bounded raw-message retention, token-budget enforcement, high-precision automatic recall, FTS5 fallback, and the Pro Lite Host, Typed Remote, and Client bundle boundaries. All 139 automated tests passed. Real model-backed acceptance also verified rolling checkpoint replacement, 1024-dimensional `text-embedding-v4` vectors, and automatic cross-project recall without an explicit memory tool call. +Current beta: `1.6.0-beta.11`. Functional acceptance used DeepSeek Harness `0.1.0-rc.8`; script-free Git installation and profile config composition were subsequently reverified against `0.1.1-rc.2`. Testing covered script-free Git and tarball installation, Web and Headless profile loading, configurable five-turn rolling compaction through the public agent-preset compaction service, exact source provenance, a lossless bounded extraction queue, failure quarantine and recovery, bounded raw-message retention, token-budget enforcement, high-precision automatic recall, FTS5 fallback, and the Pro Lite Host, Typed Remote, and Client bundle boundaries. All 149 automated tests passed. Real model-backed acceptance also verified rolling checkpoint replacement, 1024-dimensional `text-embedding-v4` vectors, and automatic cross-project recall without an explicit memory tool call.

Plugin enabled: graph-memory/dsh is active in the DSH plugin list
@@ -289,6 +289,9 @@ Before enabling deletion, back up `$DSH_HOME/graph-memory/graph-memory.db`, keep | `gm_record` | Persist a TASK, SKILL, or EVENT | | `gm_stats` | Graph, durable-message, and retention receipts/statistics | | `gm_maintain` | Run one bounded graph + configured retention maintenance tick | +| `gm_retry_extraction` | Requeue quarantined extraction failures without deleting or truncating source messages | + +The extraction queue bounds each temporary projection to 8,000 characters and 15 messages by default. One oversized message is extracted in semantic chunks while its durable SQLite event remains complete. Repeated failures become `quarantined`: they are never mislabeled as learned and retention cannot delete them. `gm_status` and `gm_stats` expose pending, succeeded, and quarantined counts. All bounds live under `extractionDrain`; see `cordis.patch.yml` for defaults. Automatic recall does not require an explicit `gm_search` tool call. The plugin retrieves relevant memory during Prompt Assembly. @@ -408,7 +411,7 @@ Release checks: ## Current limitations - Automatic extraction depends on auxiliary-model output stability. Use `gm_record` for critical beta knowledge. -- DSH does not yet expose `gm_update` and `gm_maintain`; those remain OpenClaw-entry tools. +- DSH does not yet expose `gm_update`; `gm_maintain` and `gm_retry_extraction` are native tools. - Pro Lite currently has a read-only card client; 2D/3D, split view, and controlled drag-to-context are not implemented. - npm registry publication is pending; install the current beta from a GitHub-built tarball. diff --git a/README_CN.md b/README_CN.md index f09086e..86da2a2 100644 --- a/README_CN.md +++ b/README_CN.md @@ -178,7 +178,7 @@ graph-memory/ | 插件状态可见 | **已完成** | 设置页 Plugin Inventory 显示 active | | Pro 可视化工作台 | **实验版可用** | 独立 DSH Client Plugin，当前为只读卡片式快照 | -当前 beta：`1.6.0-beta.10`。完整功能验收宿主为 DeepSeek Harness `0.1.0-rc.8`；随后又在 `0.1.1-rc.2` 上复验了无脚本 Git 安装与 profile 配置组合。验收已覆盖无安装脚本的 Git 与 tarball 安装、Web profile 原生加载、通过 Agent 公共 compaction 服务执行的可配置最近 5 轮滚动压缩、精确原文溯源、有界原始消息保留策略、token 预算、高精度自动召回、FTS5 降级，以及 Pro Lite Host、Typed Remote 和 Client bundle 边界；139 项自动化测试通过。真实模型验收还完成了滚动 checkpoint 替换、`text-embedding-v4` 1024 维向量写入，以及不调用记忆工具的跨项目自动召回。 +当前 beta：`1.6.0-beta.11`。完整功能验收宿主为 DeepSeek Harness `0.1.0-rc.8`；随后又在 `0.1.1-rc.2` 上复验了无脚本 Git 安装与 profile 配置组合。验收已覆盖无安装脚本的 Git 与 tarball 安装、Web / Headless profile 原生加载、通过 Agent 公共 compaction 服务执行的可配置最近 5 轮滚动压缩、精确原文溯源、无损有界抽取队列、失败隔离与恢复、有界原始消息保留策略、token 预算、高精度自动召回、FTS5 降级，以及 Pro Lite Host、Typed Remote 和 Client bundle 边界；149 项自动化测试通过。真实模型验收还完成了滚动 checkpoint 替换、`text-embedding-v4` 1024 维向量写入，以及不调用记忆工具的跨项目自动召回。

插件已启用：graph-memory/dsh 在 DSH 插件列表中处于 active
@@ -269,6 +269,9 @@ messageRetention: | `gm_record` | 确定性记录 TASK、SKILL 或 EVENT | | `gm_stats` | 查看图谱、原始消息和保留策略回执/统计 | | `gm_maintain` | 执行一次有界图维护与已配置的消息保留批次 | +| `gm_retry_extraction` | 将隔离的抽取失败消息重新入队，不删除或截断原始对话 | + +抽取队列默认按 8,000 个字符和 15 条消息限制单次临时投影。超长的单条消息会按语义边界分段抽取，SQLite 中的原始事件保持完整；连续失败的消息进入 `quarantined`，不会伪装成已抽取，也不会被消息保留策略删除。`gm_status` / `gm_stats` 会分别显示 pending、succeeded 和 quarantined 数量。参数统一位于 `extractionDrain`，默认值见 `cordis.patch.yml`。 自动召回不要求模型主动调用 `gm_search`；适配器会在 Prompt Assembly 阶段检索并注入相关记忆。 @@ -370,7 +373,7 @@ npm pack ## 当前限制 - 自动抽取依赖辅助模型输出稳定性；关键知识在 beta 阶段建议使用 `gm_record`。 -- DSH 版暂未提供 `gm_update` 和 `gm_maintain`，这两个工具目前属于 OpenClaw 入口。 +- DSH 版暂未提供 `gm_update`；`gm_maintain` 与 `gm_retry_extraction` 已作为原生工具提供。 - Pro Lite 目前只有只读卡片式 Client；2D/3D、分屏和受控拖拽尚未实现。 - npm registry 发布尚未完成，当前使用 GitHub 源码 tarball 安装。 diff --git a/cordis.patch.yml b/cordis.patch.yml index 178db3f..3357e2e 100644 --- a/cordis.patch.yml +++ b/cordis.patch.yml @@ -10,6 +10,18 @@ autoRecallMinScore: 0.6 recallTokenBudget: 4096 maintenanceInterval: 6 + # Extraction is lossless: these limits bound temporary LLM projections + # only. Original DSH events remain complete in SQLite. Repeated poison + # inputs are quarantined for explicit gm_retry_extraction recovery. + extractionDrain: + maxBatchChars: 8000 + maxBatchMessages: 15 + existingNamesMaxEntries: 150 + existingNamesMaxChars: 3000 + streamTimeoutMs: 180000 + maxRetries: 2 + retryDelaysMs: [5000, 15000] + shutdownGraceMs: 30000 # Durable events are separate from the compacted model surface. The # safe default never deletes raw messages. To opt in, first use # keep: referenced with dryRun: true and inspect gm_stats/gm_maintain. diff --git a/dist/dsh.js b/dist/dsh.js index 403bb3b..ba63885 100644 --- a/dist/dsh.js +++ b/dist/dsh.js @@ -7,8 +7,9 @@ */ import { createHash, randomUUID } from "node:crypto"; import { openDb } from "./src/store/db.js"; -import { allActiveNodes, findByName, getBySession, getStats, getVectorStats, getUnextracted, markExtracted, saveMessageOnce, updateNode, upsertEdge, upsertNode, } from "./src/store/store.js"; +import { allActiveNodes, findByName, getBySession, getStats, getVectorStats, getUnextracted, getExtractionStats, getPendingSessionIds, markMessagesExtracted, quarantineMessages, recordExtractionFailure, requeueQuarantined, saveMessageOnce, updateNode, upsertEdge, upsertNode, } from "./src/store/store.js"; import { Extractor, normalizeExtractionContent } from "./src/extractor/extract.js"; +import { normalizeExtractionDrainPolicy, splitExtractionContent, } from "./src/extractor/drain-policy.js"; import { Recaller } from "./src/recaller/recall.js"; import { assembleContext } from "./src/format/assemble.js"; import { selectDshRollingCompactionRange } from "./src/format/dsh-compaction.js"; @@ -17,22 +18,9 @@ import { createEmbedFn } from "./src/engine/embed.js"; import { computeGlobalPageRank, invalidateGraphCache } from "./src/graph/pagerank.js"; import { detectCommunities } from "./src/graph/community.js"; import { DEFAULT_CONFIG } from "./src/types.js"; -// Extraction resilience bounds (see extractPending / drainBatch): -// - one extraction request is capped by accumulated normalized characters, -// then by message count, so a burst of long tool results cannot build a -// request that stalls the LLM stream for the whole timeout; -// - a stalled stream (no finish/error chunk) is hard-bounded by this timeout; -// - a batch that still fails after retries is bisected, and a single message -// that keeps failing is marked extracted so the drain can never deadlock. -const EXTRACTION_BATCH_MAX_CHARS = 8_000; -const EXTRACTION_BATCH_MAX_MESSAGES = 15; -const EXTRACTION_NAMES_MAX_ENTRIES = 150; -const EXTRACTION_NAMES_MAX_CHARS = 3_000; -const EXTRACTION_STREAM_TIMEOUT_MS = 180_000; -const EXTRACTION_MAX_RETRIES = 2; -const EXTRACTION_RETRY_DELAYS_MS = [5_000, 15_000]; +import { messageRetentionPolicyRevision, normalizeMessageRetentionPolicy, runMessageRetention, } from "./src/store/retention.js"; export const name = "graph-memory-dsh"; -export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "agentPresets", "sessions", "credentials"]; +export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "sessions", "credentials"]; const HOST = "dsh"; const PLUGIN = "graph-memory"; function sessionKey(id) { @@ -110,6 +98,16 @@ export function apply(ctx, input = {}) { if (!Number.isFinite(autoRecallMinScore) || autoRecallMinScore < 0 || autoRecallMinScore > 1) { throw new TypeError(`[graph-memory] autoRecallMinScore must be between 0 and 1, received ${autoRecallMinScore}`); } + const maintenanceInterval = input.maintenanceInterval ?? DEFAULT_CONFIG.compactTurnCount; + if (!Number.isInteger(maintenanceInterval) || maintenanceInterval < 1) { + throw new TypeError(`[graph-memory] maintenanceInterval must be a positive integer, received ${maintenanceInterval}`); + } + const messageRetention = normalizeMessageRetentionPolicy(input.messageRetention); + const extractionDrain = normalizeExtractionDrainPolicy({ + ...input.extractionDrain, + streamTimeoutMs: input.extractionDrain?.streamTimeoutMs ?? input.extractionStreamTimeoutMs, + retryDelaysMs: input.extractionDrain?.retryDelaysMs ?? input.extractionRetryDelaysMs, + }); const credentialRef = input.embedding?.apiKeyEnv; if (credentialRef && !/^[A-Za-z_][A-Za-z0-9_]*$/.test(credentialRef)) { throw new TypeError(`[graph-memory] embedding.apiKeyEnv must be a credential reference, received ${JSON.stringify(credentialRef)}`); @@ -123,7 +121,7 @@ export function apply(ctx, input = {}) { const config = { ...DEFAULT_CONFIG, dbPath: input.dbPath ?? "~/.dsh/graph-memory/graph-memory.db", - compactTurnCount: input.maintenanceInterval ?? DEFAULT_CONFIG.compactTurnCount, + compactTurnCount: maintenanceInterval, recallMaxNodes: input.recallMaxNodes ?? DEFAULT_CONFIG.recallMaxNodes, recallMaxDepth: input.recallMaxDepth ?? DEFAULT_CONFIG.recallMaxDepth, embedding, @@ -140,6 +138,8 @@ export function apply(ctx, input = {}) { const embeddingConfigured = Boolean(input.embedding?.apiKeyEnv || input.embedding?.baseURL || input.embedding?.baseUrl); let embeddingState = embeddingConfigured ? "initializing" : "fts-only"; let closing = false; + let abortingExtraction = false; + const activeExtractionControllers = new Set(); let warnedMissingCompaction = false; const compactionAttached = new WeakSet(); const compactionMetrics = { @@ -149,6 +149,14 @@ export function apply(ctx, input = {}) { unavailable: 0, failed: 0, }; + const retentionMetrics = { + runs: 0, + dryRuns: 0, + selectedRows: 0, + deletedRows: 0, + deletedBytes: 0, + last: undefined, + }; if (embeddingConfigured) { void createEmbedFn(embedding).then(async (embed) => { if (embed && !closing) { @@ -179,30 +187,35 @@ export function apply(ctx, input = {}) { if (!selectedRoute) { throw new Error("[graph-memory] DSH has not recorded a model route yet; send one normal message first or configure llmProvider/llmModel"); } - const chunks = ctx.llm.stream({ - provider: selectedRoute.provider, - model: selectedRoute.model, - system, - temperature: 0.1, - maxTokens: input.llmMaxTokens ?? 4096, - messages: [{ - id: randomUUID(), - role: "user", - content: [{ type: "text", text: user }], - source: { kind: "plugin", plugin: PLUGIN }, - }], - }); + const controller = new AbortController(); + activeExtractionControllers.add(controller); let text = ""; let blockText = ""; - // A streaming LLM call can stall without ever emitting a finish/error - // chunk (observed in the wild with long extraction batches). Bound the - // whole stream with a hard timeout so a hung provider cannot pin this - // session's extraction chain forever — previously this loop had no - // timeout at all, and a stall blocked every later turn's extraction. let streamTimer; - const timedOut = await Promise.race([ - (async () => { - for await (const chunk of chunks) { + let iterator; + const timeoutError = new Error(`[graph-memory] DSH LLM extraction stream timed out after ${extractionDrain.streamTimeoutMs / 1000}s`); + try { + const chunks = ctx.llm.stream({ + provider: selectedRoute.provider, + model: selectedRoute.model, + system, + temperature: 0.1, + maxTokens: input.llmMaxTokens ?? 4096, + signal: controller.signal, + messages: [{ + id: randomUUID(), + role: "user", + content: [{ type: "text", text: user }], + source: { kind: "plugin", plugin: PLUGIN }, + }], + }); + iterator = chunks[Symbol.asyncIterator](); + const consume = (async () => { + while (true) { + const current = await iterator.next(); + if (current.done) + break; + const chunk = current.value; if (chunk?.type === "text-delta" && typeof chunk.text === "string") text += chunk.text; if (chunk?.type === "block-end" && chunk.block?.type === "text") @@ -211,21 +224,34 @@ export function apply(ctx, input = {}) { throw new Error(`[graph-memory] DSH LLM ${chunk.reason.kind}: ${chunk.reason.failure?.message ?? "unknown failure"}`); } } - return false; - })(), - new Promise((resolve) => { - streamTimer = setTimeout(() => resolve(true), input.extractionStreamTimeoutMs ?? EXTRACTION_STREAM_TIMEOUT_MS); - }), - ]); - if (streamTimer) - clearTimeout(streamTimer); - if (timedOut) { - throw new Error(`[graph-memory] DSH LLM extraction stream timed out after ${EXTRACTION_STREAM_TIMEOUT_MS / 1000}s (no finish chunk)`); + })(); + await Promise.race([ + consume, + new Promise((_resolve, reject) => { + controller.signal.addEventListener("abort", () => { + reject(controller.signal.reason ?? new Error("[graph-memory] extraction aborted")); + }, { once: true }); + }), + new Promise((_resolve, reject) => { + streamTimer = setTimeout(() => { + controller.abort(timeoutError); + reject(timeoutError); + }, extractionDrain.streamTimeoutMs); + }), + ]); + const result = text || blockText; + if (!result.trim()) + throw new Error("[graph-memory] DSH LLM returned empty extraction output"); + return result; + } + finally { + if (streamTimer) + clearTimeout(streamTimer); + activeExtractionControllers.delete(controller); + if (controller.signal.aborted && iterator?.return) { + void Promise.resolve(iterator.return()).catch(() => undefined); + } } - const result = text || blockText; - if (!result.trim()) - throw new Error("[graph-memory] DSH LLM returned empty extraction output"); - return result; } function ingest(sessionId, event) { const route = routeFromEvent(event); @@ -271,129 +297,215 @@ export function apply(ctx, input = {}) { void recaller.syncEmbed(node); invalidateGraphCache(); } - // Bound the dedup/name hint list fed into extraction. Unbounded it could - // grow to tens of thousands of characters (every node name ever created - // for this session), pushing the request over the LLM context window. + // Existing names are only deduplication hints. Select them deterministically + // so the same graph produces the same bounded prompt across restarts. function existingNameList(sid) { const names = []; let chars = 0; - for (const node of getBySession(db, sid)) { + const nodes = getBySession(db, sid).sort((left, right) => (right.updatedAt - left.updatedAt || + right.validatedCount - left.validatedCount || + left.name.localeCompare(right.name))); + for (const node of nodes) { const name = typeof node.name === "string" ? node.name : ""; if (!name) continue; - if (names.length >= EXTRACTION_NAMES_MAX_ENTRIES || (names.length > 0 && chars + name.length > EXTRACTION_NAMES_MAX_CHARS)) + if (names.length >= extractionDrain.existingNamesMaxEntries) break; + if (chars + name.length > extractionDrain.existingNamesMaxChars) + continue; names.push(name); chars += name.length; } return names; } - // Extract one bounded batch with resilience: retry transient failures with - // backoff, then bisect so a single poison message cannot sink the rest, and - // only as a last resort mark a lone failing message extracted so the drain - // always makes progress. The previous behaviour stopped the whole drain on - // the first error and left the batch unextracted forever — every restart - // retried the same failing batch, pinning the backlog indefinitely. - async function drainBatch(sessionId, sid, messages, attempt) { - if (closing) - return; - // A single message that keeps failing would pin the drain forever. Mark - // it extracted (the raw message stays in gm_messages) after retries so - // the rest of the backlog can still be learned from. - if (messages.length === 1 && attempt > 0) { - markExtracted(db, sid, Number(messages[0].turn_index)); - ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${attempt} tries for ${sid}`); - return; + const retryCancels = new Set(); + async function waitForRetry(delayMs) { + if (abortingExtraction) + return false; + if (delayMs === 0) + return true; + return new Promise((resolve) => { + let settled = false; + const finish = (value) => { + if (settled) + return; + settled = true; + clearTimeout(timer); + retryCancels.delete(cancel); + resolve(value); + }; + const timer = setTimeout(() => finish(true), delayMs); + const cancel = () => finish(false); + retryCancels.add(cancel); + }); + } + async function extractOnce(sessionId, sid, messages) { + const route = latestRoute.get(String(sessionId)); + const extractor = new Extractor(config, (system, user) => complete(route, system, user)); + const result = await extractor.extract({ messages, existingNames: existingNameList(sid) }); + const names = new Map(); + for (const candidate of result.nodes) { + const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); + names.set(node.name, node.id); + void recaller.syncEmbed(node); } - try { - // Extraction follows this exact conversation's latest logged route. A - // shared "last model wins" closure would mix providers when sessions - // finish concurrently. - const route = latestRoute.get(String(sessionId)); - const extractor = new Extractor(config, (system, user) => complete(route, system, user)); - const existingNames = existingNameList(sid); - const result = await extractor.extract({ messages, existingNames }); - const names = new Map(); - for (const candidate of result.nodes) { - const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); - names.set(node.name, node.id); - void recaller.syncEmbed(node); + for (const edge of result.edges) { + const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; + const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; + if (!fromId || !toId) + continue; + upsertEdge(db, { + fromId, + toId, + type: edge.type, + instruction: edge.instruction, + condition: edge.condition, + sessionId: sid, + }); + } + if (result.nodes.length || result.edges.length) + invalidateGraphCache(); + ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); + } + async function extractWithRetries(sessionId, sid, messages) { + const ids = Array.from(new Set(messages.map(message => String(message.id)))); + for (let attempt = 0; attempt <= extractionDrain.maxRetries; attempt += 1) { + if (abortingExtraction) + return new Error("[graph-memory] extraction aborted during shutdown"); + try { + await extractOnce(sessionId, sid, messages); + return; } - for (const edge of result.edges) { - const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; - const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; - if (!fromId || !toId) - continue; - upsertEdge(db, { - fromId, - toId, - type: edge.type, - instruction: edge.instruction, - condition: edge.condition, - sessionId: sid, - }); + catch (cause) { + const error = cause instanceof Error ? cause : new Error(String(cause)); + const retrying = attempt < extractionDrain.maxRetries; + const delayMs = retrying ? extractionDrain.retryDelaysMs[attempt] : 0; + recordExtractionFailure(db, ids, error.message, retrying ? Date.now() + delayMs : null); + if (!retrying) + return error; + ctx.logger.warn(`[graph-memory] DSH extraction retry ${attempt + 1}/${extractionDrain.maxRetries} in ${Math.round(delayMs / 1000)}s for ${sid}: ${error.message}`); + if (!await waitForRetry(delayMs)) + return new Error("[graph-memory] extraction aborted during shutdown"); } - markExtracted(db, sid, Math.max(...messages.map((message) => Number(message.turn_index)))); - if (result.nodes.length || result.edges.length) - invalidateGraphCache(); - ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); } - catch (error) { - // Back off and retry a couple of times for transient provider hiccups. - if (attempt < EXTRACTION_MAX_RETRIES) { - const retryDelays = input.extractionRetryDelaysMs ?? EXTRACTION_RETRY_DELAYS_MS; - const delayMs = retryDelays[attempt] ?? 15_000; - ctx.logger.warn(`[graph-memory] DSH extraction retry in ${Math.round(delayMs / 1000)}s for ${sid}: ${String(error)}`); - await new Promise((resolve) => setTimeout(resolve, delayMs)); - return drainBatch(sessionId, sid, messages, attempt + 1); - } - // Still failing: bisect so one poison message cannot sink the rest. - if (messages.length > 1) { - const mid = Math.ceil(messages.length / 2); - ctx.logger.warn(`[graph-memory] DSH extraction split ${messages.length} -> ${mid}+${messages.length - mid} for ${sid}: ${String(error)}`); - await drainBatch(sessionId, sid, messages.slice(0, mid), 0); - await drainBatch(sessionId, sid, messages.slice(mid), 0); + return new Error("[graph-memory] extraction retry loop ended unexpectedly"); + } + async function drainBatch(sessionId, sid, messages) { + if (abortingExtraction || !messages.length) + return; + if (messages.length === 1) { + const original = messages[0]; + const chunks = splitExtractionContent(String(original.content ?? ""), extractionDrain.maxBatchChars); + if (chunks.length > 1) { + for (let index = 0; index < chunks.length; index += 1) { + const error = await extractWithRetries(sessionId, sid, [{ ...original, content: chunks[index] }]); + if (error) { + quarantineMessages(db, [String(original.id)], error.message); + ctx.logger.warn(`[graph-memory] DSH extraction quarantined turn=${original.turn_index}, segment=${index + 1}/${chunks.length} for ${sid}: ${error.message}`); + return; + } + } + markMessagesExtracted(db, [String(original.id)]); + ctx.logger.info(`[graph-memory] DSH losslessly extracted turn=${original.turn_index} in ${chunks.length} bounded segments for ${sid}`); return; } - markExtracted(db, sid, Number(messages[0].turn_index)); - ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${EXTRACTION_MAX_RETRIES + 1} tries for ${sid}`); } + const error = await extractWithRetries(sessionId, sid, messages); + if (!error) { + markMessagesExtracted(db, messages.map(message => String(message.id))); + return; + } + if (messages.length > 1) { + const mid = Math.ceil(messages.length / 2); + ctx.logger.warn(`[graph-memory] DSH extraction split ${messages.length} -> ${mid}+${messages.length - mid} for ${sid}: ${error.message}`); + await drainBatch(sessionId, sid, messages.slice(0, mid)); + await drainBatch(sessionId, sid, messages.slice(mid)); + return; + } + quarantineMessages(db, [String(messages[0].id)], error.message); + ctx.logger.warn(`[graph-memory] DSH extraction quarantined turn=${messages[0].turn_index} after ${extractionDrain.maxRetries + 1} attempts for ${sid}: ${error.message}`); } async function extractPending(sessionId) { - if (!extractionEnabled || closing) + if (!extractionEnabled || abortingExtraction) return; const sid = sessionKey(sessionId); - while (!closing) { - // Take the oldest unextracted messages in order, bounded by accumulated - // normalized length first (a single long message always fits so the - // drain can make progress), then by count. Order matters: we mark - // extracted up to the max turn_index of the batch, so skipping a - // middle message would mis-mark it as extracted. + while (!abortingExtraction) { const messages = []; let chars = 0; - for (const message of getUnextracted(db, sid, EXTRACTION_BATCH_MAX_MESSAGES * 16)) { + for (const message of getUnextracted(db, sid, extractionDrain.maxBatchMessages * 16)) { const content = normalizeExtractionContent(message.content); - if (messages.length > 0 && chars + content.length > EXTRACTION_BATCH_MAX_CHARS) + const contentChars = Array.from(content).length; + if (messages.length > 0 && chars + contentChars > extractionDrain.maxBatchChars) break; messages.push({ ...message, content }); - chars += content.length; - if (messages.length >= EXTRACTION_BATCH_MAX_MESSAGES) + chars += contentChars; + if (messages.length >= extractionDrain.maxBatchMessages || chars >= extractionDrain.maxBatchChars) break; } if (!messages.length) return; - await drainBatch(sessionId, sid, messages, 0); + await drainBatch(sessionId, sid, messages); } } function scheduleExtract(sessionId) { + if (!extractionEnabled || closing) + return Promise.resolve(); const key = String(sessionId); - const previous = extractChain.get(key) ?? Promise.resolve(); - const next = previous.then(() => extractPending(sessionId)); + const previous = extractChain.get(key); + const running = previous + ? previous.then(() => extractPending(sessionId)) + : extractPending(sessionId); + const next = running.catch(error => { + ctx.logger.error(`[graph-memory] DSH extraction queue failed for ${key}: ${String(error)}`); + }); extractChain.set(key, next); - void next.finally(() => { + void next.then(() => { if (extractChain.get(key) === next) extractChain.delete(key); }); + return next; + } + function runConfiguredRetention() { + const result = runMessageRetention(db, messageRetention); + retentionMetrics.runs += 1; + if (result.dryRun) + retentionMetrics.dryRuns += 1; + retentionMetrics.selectedRows += result.selectedRows; + retentionMetrics.deletedRows += result.deletedRows; + retentionMetrics.deletedBytes += result.deletedBytes; + retentionMetrics.last = result; + if (result.selectedRows > 0) { + const action = result.dryRun ? "would prune" : "pruned"; + ctx.logger.info(`[graph-memory] retention ${action} ${result.dryRun ? result.selectedRows : result.deletedRows} ` + + `unreferenced extracted messages (${result.selectedBytes} estimated bytes, more=${result.hasMore})`); + } + return result; + } + function runGraphMaintenance() { + invalidateGraphCache(); + const pagerank = computeGlobalPageRank(db, config); + const communities = detectCommunities(db); + return { pagerankNodes: pagerank.scores.size, communities: communities.count }; + } + function runMaintenanceTick() { + const result = { errors: [] }; + try { + result.graph = runGraphMaintenance(); + } + catch (error) { + const message = `graph maintenance failed: ${String(error)}`; + result.errors.push(message); + ctx.logger.warn(`[graph-memory] DSH ${message}`); + } + try { + result.retention = runConfiguredRetention(); + } + catch (error) { + const message = `message retention failed: ${String(error)}`; + result.errors.push(message); + ctx.logger.warn(`[graph-memory] DSH ${message}`); + } + return result; } function maintain(sessionId) { const key = String(sessionId); @@ -401,14 +513,7 @@ export function apply(ctx, input = {}) { turnCounts.set(key, turns); if (turns % config.compactTurnCount !== 0) return; - try { - invalidateGraphCache(); - computeGlobalPageRank(db, config); - detectCommunities(db); - } - catch (error) { - ctx.logger.warn(`[graph-memory] DSH graph maintenance failed: ${String(error)}`); - } + runMaintenanceTick(); } function backfill(agent) { const id = agent?.id ?? agent?.session?.id; @@ -416,7 +521,6 @@ export function apply(ctx, input = {}) { return; for (const event of agent.session.events) ingest(id, event); - scheduleExtract(id); } // Graph Memory owns the rolling retention policy while DSH's public // compaction service owns the durable summary/replacement transaction. DSH @@ -434,7 +538,10 @@ export function apply(ctx, input = {}) { // Agent preset services live in an isolated standing scope. Use // DSH's public roster seam instead of reaching through Cordis scope // internals or requiring a change in Harness itself. - const compaction = ctx.agentPresets.serviceFor(agent, "compaction"); + const agentPresets = typeof ctx.get === "function" + ? ctx.get("agentPresets") + : ctx.agentPresets; + const compaction = agentPresets?.serviceFor?.(agent, "compaction"); if (!compaction?.compactRegion) { compactionMetrics.unavailable += 1; if (!warnedMissingCompaction) { @@ -475,6 +582,19 @@ export function apply(ctx, input = {}) { attachRollingCompaction(agent); backfill(agent); }); + // DSH declares this as a serial, awaited lifecycle event before turn/end is + // committed. It is the reliable drain boundary for one-shot Headless: LLM + // adapters are still registered here, unlike ordinary session/event emit + // observers whose returned promises are intentionally ignored. + ctx.on("agent/turn-stopping", async ({ agent, signal }) => { + if (signal?.aborted) + return; + const id = agent?.id ?? agent?.session?.id; + if (id === undefined) + return; + backfill(agent); + await scheduleExtract(id); + }); ctx.on("session/event", (session, event) => { const id = session?.id; if (id === undefined) @@ -488,7 +608,9 @@ export function apply(ctx, input = {}) { ingest(id, event); recordCompactionCapsule(id, event); if (event?.type === "turn/end") { - scheduleExtract(id); + // This is a background fallback for hosts that do not expose the Agent + // turn-stopping boundary. DSH itself drains synchronously above. + void scheduleExtract(id); maintain(id); } }); @@ -576,7 +698,10 @@ export function apply(ctx, input = {}) { const embeddingModel = embeddingConfigured && input.embedding?.model ? ` (${input.embedding.model})` : ""; - return `Graph Memory active (DSH native)\nStore: ${config.dbPath}\nNodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nExtraction: ${extractionEnabled ? "enabled" : "disabled"}\nRecall: ${recallEnabled ? "enabled" : "disabled"}\nEmbedding: ${embeddingState}${embeddingModel}\nVectors: ${vectors.count}/${stats.totalNodes}${vectors.dimensions.length ? ` (${vectors.dimensions.join(", ")} dimensions)` : ""}\nRolling compaction: attached=${compactionMetrics.attached}, selected=${compactionMetrics.selected}, succeeded=${compactionMetrics.succeeded}, unavailable=${compactionMetrics.unavailable}, failed=${compactionMetrics.failed}`; + const messageCount = Number(db.prepare("SELECT COUNT(*) AS count FROM gm_messages").get()?.count ?? 0); + const extraction = getExtractionStats(db); + const retentionRevision = messageRetentionPolicyRevision(messageRetention); + return `Graph Memory active (DSH native)\nStore: ${config.dbPath}\nNodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nMessages: ${messageCount}\nExtraction: ${extractionEnabled ? "enabled" : "disabled"} (pending=${extraction.pending}, succeeded=${extraction.succeeded}, quarantined=${extraction.quarantined})\nExtraction drain: maxChars=${extractionDrain.maxBatchChars}, maxMessages=${extractionDrain.maxBatchMessages}, retries=${extractionDrain.maxRetries}, timeoutMs=${extractionDrain.streamTimeoutMs}\nRecall: ${recallEnabled ? "enabled" : "disabled"}\nEmbedding: ${embeddingState}${embeddingModel}\nVectors: ${vectors.count}/${stats.totalNodes}${vectors.dimensions.length ? ` (${vectors.dimensions.join(", ")} dimensions)` : ""}\nMessage retention: keep=${messageRetention.keep}, recentTurns=${messageRetention.recentTurns}, retentionDays=${messageRetention.retentionDays}, batchSize=${messageRetention.batchSize}, dryRun=${messageRetention.dryRun}, revision=${retentionRevision}\nRetention GC: runs=${retentionMetrics.runs}, dryRuns=${retentionMetrics.dryRuns}, selected=${retentionMetrics.selectedRows}, deleted=${retentionMetrics.deletedRows}, estimatedDeletedBytes=${retentionMetrics.deletedBytes}\nRolling compaction: attached=${compactionMetrics.attached}, selected=${compactionMetrics.selected}, succeeded=${compactionMetrics.succeeded}, unavailable=${compactionMetrics.unavailable}, failed=${compactionMetrics.failed}`; }, }); ctx.tools.register({ @@ -626,22 +751,91 @@ export function apply(ctx, input = {}) { }); ctx.tools.register({ name: "gm_stats", - description: "Show Graph Memory node, edge and community counts.", + description: "Show Graph Memory graph, durable-message and retention statistics.", parameters: { type: "object", properties: {}, additionalProperties: false }, output: stringOutput("Graph Memory statistics"), execute: async () => { const stats = getStats(db); - return `Nodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nCommunities: ${stats.communities}\nBy type: ${JSON.stringify(stats.byType)}`; + const messageCount = Number(db.prepare("SELECT COUNT(*) AS count FROM gm_messages").get()?.count ?? 0); + return `Nodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nCommunities: ${stats.communities}\nMessages: ${messageCount}\nExtraction queue: ${JSON.stringify(getExtractionStats(db))}\nBy type: ${JSON.stringify(stats.byType)}\nRetention policy: ${JSON.stringify({ ...messageRetention, revision: messageRetentionPolicyRevision(messageRetention) })}\nRetention totals: ${JSON.stringify({ runs: retentionMetrics.runs, dryRuns: retentionMetrics.dryRuns, selectedRows: retentionMetrics.selectedRows, deletedRows: retentionMetrics.deletedRows, deletedBytes: retentionMetrics.deletedBytes })}\nLast retention receipt: ${JSON.stringify(retentionMetrics.last ?? null)}`; + }, + }); + ctx.tools.register({ + name: "gm_maintain", + description: "Run one bounded Graph Memory maintenance tick using the configured retention policy.", + parameters: { type: "object", properties: {}, additionalProperties: false }, + output: stringOutput("Graph Memory maintenance"), + execute: async () => JSON.stringify(runMaintenanceTick()), + }); + ctx.tools.register({ + name: "gm_retry_extraction", + description: "Requeue quarantined durable messages and retry knowledge extraction without deleting source text.", + parameters: { + type: "object", + properties: { + sessionId: { type: "string", description: "Optional DSH session id; omit to requeue every quarantined session" }, + }, + additionalProperties: false, + }, + output: stringOutput("Graph Memory extraction retry"), + execute: async (args = {}) => { + const requested = typeof args.sessionId === "string" && args.sessionId.trim() + ? args.sessionId.trim() + : undefined; + const sid = requested + ? requested.startsWith(`${HOST}:`) ? requested : sessionKey(requested) + : undefined; + const requeued = requeueQuarantined(db, sid); + const pending = sid ? [sid] : getPendingSessionIds(db); + let scheduled = 0; + for (const pendingSid of pending) { + const rawId = pendingSid.startsWith(`${HOST}:`) ? pendingSid.slice(HOST.length + 1) : pendingSid; + if (input.llmProvider && input.llmModel || latestRoute.has(rawId)) { + scheduleExtract(rawId); + scheduled += 1; + } + } + return `Requeued ${requeued} quarantined messages; scheduled ${scheduled} sessions.`; }, }); ctx.effect(() => async () => { closing = true; - await Promise.allSettled([...extractChain.values()]); + const chains = [...extractChain.values()]; + let graceTimer; + const drained = await Promise.race([ + Promise.allSettled(chains).then(() => true), + new Promise(resolve => { + graceTimer = setTimeout(() => resolve(false), extractionDrain.shutdownGraceMs); + }), + ]); + if (graceTimer) + clearTimeout(graceTimer); + if (!drained) { + abortingExtraction = true; + for (const cancel of [...retryCancels]) + cancel(); + for (const controller of activeExtractionControllers) { + controller.abort(new Error("[graph-memory] extraction shutdown grace elapsed")); + } + await Promise.allSettled([...extractChain.values()]); + } latestRoute.clear(); latestPrompt.clear(); recallCache.clear(); turnCounts.clear(); db.close(); }, "graph-memory.close"); + // With an explicit fallback route, recover durable pending work from prior + // process exits even when those sessions are not reopened in the UI. + if (extractionEnabled && input.llmProvider && input.llmModel) { + for (const sid of getPendingSessionIds(db)) { + scheduleExtract(sid.startsWith(`${HOST}:`) ? sid.slice(HOST.length + 1) : sid); + } + } + if (messageRetention.keep !== "all") { + const mode = messageRetention.dryRun ? "dry-run" : "deletion enabled"; + ctx.logger.warn(`[graph-memory] durable message retention is ${mode} (${JSON.stringify(messageRetention)}). ` + + `Back up ${config.dbPath} before the first non-dry run; VACUUM remains a separate admin action.`); + } ctx.logger.info(`[graph-memory] native DSH adapter active at ${config.dbPath}`); } diff --git a/dist/src/extractor/drain-policy.js b/dist/src/extractor/drain-policy.js new file mode 100644 index 0000000..3f9acff --- /dev/null +++ b/dist/src/extractor/drain-policy.js @@ -0,0 +1,63 @@ +export const DEFAULT_EXTRACTION_DRAIN_POLICY = { + maxBatchChars: 8_000, + maxBatchMessages: 15, + existingNamesMaxEntries: 150, + existingNamesMaxChars: 3_000, + streamTimeoutMs: 180_000, + maxRetries: 2, + retryDelaysMs: [5_000, 15_000], + shutdownGraceMs: 30_000, +}; +function integer(value, name, fallback, min, max) { + const resolved = value ?? fallback; + if (!Number.isInteger(resolved) || Number(resolved) < min || Number(resolved) > max) { + throw new TypeError(`[graph-memory] extractionDrain.${name} must be an integer between ${min} and ${max}, received ${String(resolved)}`); + } + return Number(resolved); +} +export function normalizeExtractionDrainPolicy(input) { + const maxRetries = integer(input?.maxRetries, "maxRetries", DEFAULT_EXTRACTION_DRAIN_POLICY.maxRetries, 0, 10); + const retryDelays = input?.retryDelaysMs ?? DEFAULT_EXTRACTION_DRAIN_POLICY.retryDelaysMs; + if (!Array.isArray(retryDelays) || retryDelays.length < maxRetries || retryDelays.length > 10) { + throw new TypeError(`[graph-memory] extractionDrain.retryDelaysMs must contain between ${maxRetries} and 10 delays`); + } + const retryDelaysMs = retryDelays.map((delay, index) => (integer(delay, `retryDelaysMs[${index}]`, 0, 0, 300_000))); + return { + maxBatchChars: integer(input?.maxBatchChars, "maxBatchChars", DEFAULT_EXTRACTION_DRAIN_POLICY.maxBatchChars, 64, 1_000_000), + maxBatchMessages: integer(input?.maxBatchMessages, "maxBatchMessages", DEFAULT_EXTRACTION_DRAIN_POLICY.maxBatchMessages, 1, 1_000), + existingNamesMaxEntries: integer(input?.existingNamesMaxEntries, "existingNamesMaxEntries", DEFAULT_EXTRACTION_DRAIN_POLICY.existingNamesMaxEntries, 0, 10_000), + existingNamesMaxChars: integer(input?.existingNamesMaxChars, "existingNamesMaxChars", DEFAULT_EXTRACTION_DRAIN_POLICY.existingNamesMaxChars, 0, 1_000_000), + streamTimeoutMs: integer(input?.streamTimeoutMs, "streamTimeoutMs", DEFAULT_EXTRACTION_DRAIN_POLICY.streamTimeoutMs, 10, 900_000), + maxRetries, + retryDelaysMs, + shutdownGraceMs: integer(input?.shutdownGraceMs, "shutdownGraceMs", DEFAULT_EXTRACTION_DRAIN_POLICY.shutdownGraceMs, 0, 300_000), + }; +} +/** + * Split only the temporary extraction projection. The durable message stays + * byte-for-byte unchanged in gm_messages. Boundaries prefer paragraphs and + * whitespace, and Array.from keeps surrogate pairs intact. + */ +export function splitExtractionContent(content, maxChars) { + const points = Array.from(content); + if (points.length <= maxChars) + return [content]; + const chunks = []; + let offset = 0; + while (offset < points.length) { + const hardEnd = Math.min(offset + maxChars, points.length); + let end = hardEnd; + if (hardEnd < points.length) { + const softStart = offset + Math.floor(maxChars * 0.6); + for (let cursor = hardEnd - 1; cursor >= softStart; cursor -= 1) { + if (/\s/u.test(points[cursor])) { + end = cursor + 1; + break; + } + } + } + chunks.push(points.slice(offset, end).join("")); + offset = end; + } + return chunks; +} diff --git a/dist/src/store/db.js b/dist/src/store/db.js index cb297d3..ebd558b 100644 --- a/dist/src/store/db.js +++ b/dist/src/store/db.js @@ -73,12 +73,41 @@ function migrate(db) { m8_backfill_community_signatures, m9_node_sources, m10_message_retention_index, + m11_extraction_queue_state, ]; for (let i = cur; i < steps.length; i++) { steps[i](db); db.prepare("INSERT INTO _migrations (v,at) VALUES (?,?)").run(i + 1, Date.now()); } } +// ─── 可审计抽取队列：待处理 / 成功 / 隔离 ────────────────────── +function m11_extraction_queue_state(db) { + const columns = new Set(db.prepare("PRAGMA table_info(gm_messages)").all() + .map(column => column.name)); + if (!columns.has("extraction_state")) { + db.exec(`ALTER TABLE gm_messages ADD COLUMN extraction_state TEXT NOT NULL DEFAULT 'pending' + CHECK(extraction_state IN ('pending', 'succeeded', 'quarantined'))`); + } + if (!columns.has("extraction_attempts")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_attempts INTEGER NOT NULL DEFAULT 0"); + } + if (!columns.has("extraction_error")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_error TEXT"); + } + if (!columns.has("extraction_next_retry_at")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_next_retry_at INTEGER"); + } + if (!columns.has("extraction_updated_at")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_updated_at INTEGER"); + } + db.exec(` + UPDATE gm_messages + SET extraction_state=CASE WHEN extracted=1 THEN 'succeeded' ELSE 'pending' END, + extraction_updated_at=COALESCE(extraction_updated_at, created_at); + CREATE INDEX IF NOT EXISTS ix_gm_msg_extraction_queue + ON gm_messages(extraction_state, extraction_next_retry_at, session_id, turn_index); + `); +} // ─── 有界原始消息保留策略查询 ──────────────────────────────── function m10_message_retention_index(db) { db.exec(` diff --git a/dist/src/store/retention.js b/dist/src/store/retention.js index dc511c4..b2f983f 100644 --- a/dist/src/store/retention.js +++ b/dist/src/store/retention.js @@ -102,7 +102,7 @@ function selectCandidates(db, policy, cutoffAt) { length(CAST(m.content AS BLOB)) AS content_bytes FROM gm_messages m ${recentJoin} - WHERE m.extracted=1 + WHERE m.extracted=1 AND m.extraction_state='succeeded' AND NOT EXISTS ( SELECT 1 FROM gm_node_sources source WHERE source.message_id=m.id ) @@ -139,7 +139,7 @@ export function runMessageRetention(db, policy, now = Date.now()) { const deleted = db.prepare(` DELETE FROM gm_messages WHERE id IN (${placeholders}) - AND extracted=1 + AND extracted=1 AND extraction_state='succeeded' AND NOT EXISTS ( SELECT 1 FROM gm_node_sources source WHERE source.message_id=gm_messages.id ) diff --git a/dist/src/store/store.js b/dist/src/store/store.js index b1644c0..c87510d 100644 --- a/dist/src/store/store.js +++ b/dist/src/store/store.js @@ -292,12 +292,95 @@ export function getMessages(db, sid, limit) { .all(sid); } export function getUnextracted(db, sid, limit) { - return db.prepare("SELECT * FROM gm_messages WHERE session_id=? AND extracted=0 ORDER BY turn_index LIMIT ?") - .all(sid, limit); + return db.prepare(` + SELECT * FROM gm_messages + WHERE session_id=? AND extracted=0 AND extraction_state='pending' + AND (extraction_next_retry_at IS NULL OR extraction_next_retry_at<=?) + ORDER BY turn_index, id LIMIT ? + `).all(sid, Date.now(), limit); } export function markExtracted(db, sid, upToTurn) { - db.prepare("UPDATE gm_messages SET extracted=1 WHERE session_id=? AND turn_index<=?") - .run(sid, upToTurn); + db.prepare(` + UPDATE gm_messages + SET extracted=1, extraction_state='succeeded', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE session_id=? AND turn_index<=? AND extraction_state='pending' + `).run(Date.now(), sid, upToTurn); +} +function messageIdPlaceholders(ids) { + return ids.map(() => "?").join(","); +} +/** Mark only the source rows actually accepted by the extractor. */ +export function markMessagesExtracted(db, ids) { + if (!ids.length) + return 0; + const result = db.prepare(` + UPDATE gm_messages + SET extracted=1, extraction_state='succeeded', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE id IN (${messageIdPlaceholders(ids)}) AND extraction_state='pending' + `).run(Date.now(), ...ids); + return Number(result.changes); +} +export function recordExtractionFailure(db, ids, error, nextRetryAt) { + if (!ids.length) + return 0; + const result = db.prepare(` + UPDATE gm_messages + SET extraction_attempts=extraction_attempts+1, extraction_error=?, + extraction_next_retry_at=?, extraction_updated_at=? + WHERE id IN (${messageIdPlaceholders(ids)}) AND extraction_state='pending' + `).run(error.slice(0, 2_000), nextRetryAt, Date.now(), ...ids); + return Number(result.changes); +} +/** A poison message remains durable and explicitly unlearned until retried. */ +export function quarantineMessages(db, ids, error) { + if (!ids.length) + return 0; + const result = db.prepare(` + UPDATE gm_messages + SET extracted=0, extraction_state='quarantined', extraction_error=?, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE id IN (${messageIdPlaceholders(ids)}) AND extraction_state='pending' + `).run(error.slice(0, 2_000), Date.now(), ...ids); + return Number(result.changes); +} +export function requeueQuarantined(db, sid) { + const result = sid + ? db.prepare(` + UPDATE gm_messages + SET extraction_state='pending', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE extraction_state='quarantined' AND session_id=? + `).run(Date.now(), sid) + : db.prepare(` + UPDATE gm_messages + SET extraction_state='pending', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE extraction_state='quarantined' + `).run(Date.now()); + return Number(result.changes); +} +export function getExtractionStats(db) { + const rows = db.prepare(` + SELECT extraction_state AS state, COUNT(*) AS count + FROM gm_messages GROUP BY extraction_state + `).all(); + const result = { pending: 0, succeeded: 0, quarantined: 0 }; + for (const row of rows) { + if (row.state in result) + result[row.state] = Number(row.count); + } + return result; +} +export function getPendingSessionIds(db, limit = 100) { + return db.prepare(` + SELECT session_id, MIN(turn_index) AS first_turn + FROM gm_messages + WHERE extracted=0 AND extraction_state='pending' + AND (extraction_next_retry_at IS NULL OR extraction_next_retry_at<=?) + GROUP BY session_id ORDER BY first_turn, session_id LIMIT ? + `).all(Date.now(), limit).map(row => row.session_id); } /** * 溯源选拉：按 session 拉取 user/assistant 核心对话（跳过 tool/toolResult） diff --git a/dsh.ts b/dsh.ts index ca83e23..6dd1b0a 100644 --- a/dsh.ts +++ b/dsh.ts @@ -15,13 +15,23 @@ import { getStats, getVectorStats, getUnextracted, - markExtracted, + getExtractionStats, + getPendingSessionIds, + markMessagesExtracted, + quarantineMessages, + recordExtractionFailure, + requeueQuarantined, saveMessageOnce, updateNode, upsertEdge, upsertNode, } from "./src/store/store.ts"; import { Extractor, normalizeExtractionContent } from "./src/extractor/extract.ts"; +import { + normalizeExtractionDrainPolicy, + splitExtractionContent, + type ExtractionDrainConfig, +} from "./src/extractor/drain-policy.ts"; import { Recaller } from "./src/recaller/recall.ts"; import { assembleContext } from "./src/format/assemble.ts"; import { selectDshRollingCompactionRange } from "./src/format/dsh-compaction.ts"; @@ -38,23 +48,8 @@ import { type MessageRetentionResult, } from "./src/store/retention.ts"; -// Extraction resilience bounds (see extractPending / drainBatch): -// - one extraction request is capped by accumulated normalized characters, -// then by message count, so a burst of long tool results cannot build a -// request that stalls the LLM stream for the whole timeout; -// - a stalled stream (no finish/error chunk) is hard-bounded by this timeout; -// - a batch that still fails after retries is bisected, and a single message -// that keeps failing is marked extracted so the drain can never deadlock. -const EXTRACTION_BATCH_MAX_CHARS = 8_000; -const EXTRACTION_BATCH_MAX_MESSAGES = 15; -const EXTRACTION_NAMES_MAX_ENTRIES = 150; -const EXTRACTION_NAMES_MAX_CHARS = 3_000; -const EXTRACTION_STREAM_TIMEOUT_MS = 180_000; -const EXTRACTION_MAX_RETRIES = 2; -const EXTRACTION_RETRY_DELAYS_MS = [5_000, 15_000]; - export const name = "graph-memory-dsh"; -export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "agentPresets", "sessions", "credentials"]; +export const inject = ["tools", "llm", "systemPrompt", "agentLoop", "agents", "sessions", "credentials"]; interface DshEmbeddingConfig { apiKeyEnv?: string; @@ -85,9 +80,11 @@ export interface Config { llmModel?: string; llmMaxTokens?: number; embedding?: DshEmbeddingConfig; - /** Override the extraction stream hard timeout (ms). Default 180s. */ + /** Bounded, lossless and durable extraction queue policy. */ + extractionDrain?: ExtractionDrainConfig; + /** @deprecated Use extractionDrain.streamTimeoutMs. */ extractionStreamTimeoutMs?: number; - /** Override extraction retry backoff delays (ms) per attempt. Default [5s, 15s]. */ + /** @deprecated Use extractionDrain.retryDelaysMs. */ extractionRetryDelaysMs?: number[]; } @@ -114,9 +111,10 @@ interface DshContext { agents?: { get(id: unknown): any; }; - agentPresets: { + agentPresets?: { serviceFor(agent: any, key: string): any; }; + get?(name: string): any; on(event: string, listener: (...args: any[]) => any, options?: Record): () => void; effect(register: () => (() => void | Promise), label?: string): () => void; } @@ -203,6 +201,11 @@ export function apply(ctx: DshContext, input: Config = {}): void { throw new TypeError(`[graph-memory] maintenanceInterval must be a positive integer, received ${maintenanceInterval}`); } const messageRetention = normalizeMessageRetentionPolicy(input.messageRetention); + const extractionDrain = normalizeExtractionDrainPolicy({ + ...input.extractionDrain, + streamTimeoutMs: input.extractionDrain?.streamTimeoutMs ?? input.extractionStreamTimeoutMs, + retryDelaysMs: input.extractionDrain?.retryDelaysMs ?? input.extractionRetryDelaysMs, + }); const credentialRef = input.embedding?.apiKeyEnv; if (credentialRef && !/^[A-Za-z_][A-Za-z0-9_]*$/.test(credentialRef)) { throw new TypeError(`[graph-memory] embedding.apiKeyEnv must be a credential reference, received ${JSON.stringify(credentialRef)}`); @@ -236,6 +239,8 @@ export function apply(ctx: DshContext, input: Config = {}): void { let embeddingState: "fts-only" | "initializing" | "vector-ready" | "degraded" = embeddingConfigured ? "initializing" : "fts-only"; let closing = false; + let abortingExtraction = false; + const activeExtractionControllers = new Set(); let warnedMissingCompaction = false; const compactionAttached = new WeakSet(); const compactionMetrics = { @@ -284,50 +289,67 @@ export function apply(ctx: DshContext, input: Config = {}): void { throw new Error("[graph-memory] DSH has not recorded a model route yet; send one normal message first or configure llmProvider/llmModel"); } - const chunks = ctx.llm.stream({ - provider: selectedRoute.provider, - model: selectedRoute.model, - system, - temperature: 0.1, - maxTokens: input.llmMaxTokens ?? 4096, - messages: [{ - id: randomUUID(), - role: "user", - content: [{ type: "text", text: user }], - source: { kind: "plugin", plugin: PLUGIN }, - }], - }); - + const controller = new AbortController(); + activeExtractionControllers.add(controller); let text = ""; let blockText = ""; - // A streaming LLM call can stall without ever emitting a finish/error - // chunk (observed in the wild with long extraction batches). Bound the - // whole stream with a hard timeout so a hung provider cannot pin this - // session's extraction chain forever — previously this loop had no - // timeout at all, and a stall blocked every later turn's extraction. let streamTimer: ReturnType | undefined; - const timedOut = await Promise.race([ - (async (): Promise => { - for await (const chunk of chunks) { + let iterator: AsyncIterator | undefined; + const timeoutError = new Error( + `[graph-memory] DSH LLM extraction stream timed out after ${extractionDrain.streamTimeoutMs / 1000}s`, + ); + try { + const chunks = ctx.llm.stream({ + provider: selectedRoute.provider, + model: selectedRoute.model, + system, + temperature: 0.1, + maxTokens: input.llmMaxTokens ?? 4096, + signal: controller.signal, + messages: [{ + id: randomUUID(), + role: "user", + content: [{ type: "text", text: user }], + source: { kind: "plugin", plugin: PLUGIN }, + }], + }); + iterator = chunks[Symbol.asyncIterator](); + const consume = (async () => { + while (true) { + const current = await iterator!.next(); + if (current.done) break; + const chunk = current.value; if (chunk?.type === "text-delta" && typeof chunk.text === "string") text += chunk.text; if (chunk?.type === "block-end" && chunk.block?.type === "text") blockText += chunk.block.text ?? ""; if (chunk?.type === "finish" && (chunk.reason?.kind === "error" || chunk.reason?.kind === "aborted")) { throw new Error(`[graph-memory] DSH LLM ${chunk.reason.kind}: ${chunk.reason.failure?.message ?? "unknown failure"}`); } } - return false; - })(), - new Promise((resolve) => { - streamTimer = setTimeout(() => resolve(true), input.extractionStreamTimeoutMs ?? EXTRACTION_STREAM_TIMEOUT_MS); - }), - ]); - if (streamTimer) clearTimeout(streamTimer); - if (timedOut) { - throw new Error(`[graph-memory] DSH LLM extraction stream timed out after ${EXTRACTION_STREAM_TIMEOUT_MS / 1000}s (no finish chunk)`); + })(); + await Promise.race([ + consume, + new Promise((_resolve, reject) => { + controller.signal.addEventListener("abort", () => { + reject(controller.signal.reason ?? new Error("[graph-memory] extraction aborted")); + }, { once: true }); + }), + new Promise((_resolve, reject) => { + streamTimer = setTimeout(() => { + controller.abort(timeoutError); + reject(timeoutError); + }, extractionDrain.streamTimeoutMs); + }), + ]); + const result = text || blockText; + if (!result.trim()) throw new Error("[graph-memory] DSH LLM returned empty extraction output"); + return result; + } finally { + if (streamTimer) clearTimeout(streamTimer); + activeExtractionControllers.delete(controller); + if (controller.signal.aborted && iterator?.return) { + void Promise.resolve(iterator.return()).catch(() => undefined); + } } - const result = text || blockText; - if (!result.trim()) throw new Error("[graph-memory] DSH LLM returned empty extraction output"); - return result; } function ingest(sessionId: unknown, event: any): boolean { @@ -380,121 +402,164 @@ export function apply(ctx: DshContext, input: Config = {}): void { invalidateGraphCache(); } - // Bound the dedup/name hint list fed into extraction. Unbounded it could - // grow to tens of thousands of characters (every node name ever created - // for this session), pushing the request over the LLM context window. + // Existing names are only deduplication hints. Select them deterministically + // so the same graph produces the same bounded prompt across restarts. function existingNameList(sid: string): string[] { const names: string[] = []; let chars = 0; - for (const node of getBySession(db, sid)) { + const nodes = getBySession(db, sid).sort((left, right) => ( + right.updatedAt - left.updatedAt || + right.validatedCount - left.validatedCount || + left.name.localeCompare(right.name) + )); + for (const node of nodes) { const name = typeof node.name === "string" ? node.name : ""; if (!name) continue; - if (names.length >= EXTRACTION_NAMES_MAX_ENTRIES || (names.length > 0 && chars + name.length > EXTRACTION_NAMES_MAX_CHARS)) break; + if (names.length >= extractionDrain.existingNamesMaxEntries) break; + if (chars + name.length > extractionDrain.existingNamesMaxChars) continue; names.push(name); chars += name.length; } return names; } - // Extract one bounded batch with resilience: retry transient failures with - // backoff, then bisect so a single poison message cannot sink the rest, and - // only as a last resort mark a lone failing message extracted so the drain - // always makes progress. The previous behaviour stopped the whole drain on - // the first error and left the batch unextracted forever — every restart - // retried the same failing batch, pinning the backlog indefinitely. - async function drainBatch(sessionId: unknown, sid: string, messages: any[], attempt: number): Promise { - if (closing) return; - // A single message that keeps failing would pin the drain forever. Mark - // it extracted (the raw message stays in gm_messages) after retries so - // the rest of the backlog can still be learned from. - if (messages.length === 1 && attempt > 0) { - markExtracted(db, sid, Number(messages[0].turn_index)); - ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${attempt} tries for ${sid}`); - return; + const retryCancels = new Set<() => void>(); + + async function waitForRetry(delayMs: number): Promise { + if (abortingExtraction) return false; + if (delayMs === 0) return true; + return new Promise((resolve) => { + let settled = false; + const finish = (value: boolean) => { + if (settled) return; + settled = true; + clearTimeout(timer); + retryCancels.delete(cancel); + resolve(value); + }; + const timer = setTimeout(() => finish(true), delayMs); + const cancel = () => finish(false); + retryCancels.add(cancel); + }); + } + + async function extractOnce(sessionId: unknown, sid: string, messages: any[]): Promise { + const route = latestRoute.get(String(sessionId)); + const extractor = new Extractor(config, (system, user) => complete(route, system, user)); + const result = await extractor.extract({ messages, existingNames: existingNameList(sid) }); + const names = new Map(); + for (const candidate of result.nodes) { + const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); + names.set(node.name, node.id); + void recaller.syncEmbed(node); } - try { - // Extraction follows this exact conversation's latest logged route. A - // shared "last model wins" closure would mix providers when sessions - // finish concurrently. - const route = latestRoute.get(String(sessionId)); - const extractor = new Extractor(config, (system, user) => complete(route, system, user)); - const existingNames = existingNameList(sid); - const result = await extractor.extract({ messages, existingNames }); - const names = new Map(); - for (const candidate of result.nodes) { - const { node } = upsertNode(db, candidate, sid, extractionSources(candidate, messages)); - names.set(node.name, node.id); - void recaller.syncEmbed(node); - } - for (const edge of result.edges) { - const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; - const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; - if (!fromId || !toId) continue; - upsertEdge(db, { - fromId, - toId, - type: edge.type, - instruction: edge.instruction, - condition: edge.condition, - sessionId: sid, - }); - } - markExtracted(db, sid, Math.max(...messages.map((message: any) => Number(message.turn_index)))); - if (result.nodes.length || result.edges.length) invalidateGraphCache(); - ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); - } catch (error) { - // Back off and retry a couple of times for transient provider hiccups. - if (attempt < EXTRACTION_MAX_RETRIES) { - const retryDelays = input.extractionRetryDelaysMs ?? EXTRACTION_RETRY_DELAYS_MS; - const delayMs = retryDelays[attempt] ?? 15_000; - ctx.logger.warn(`[graph-memory] DSH extraction retry in ${Math.round(delayMs / 1000)}s for ${sid}: ${String(error)}`); - await new Promise((resolve) => setTimeout(resolve, delayMs)); - return drainBatch(sessionId, sid, messages, attempt + 1); + for (const edge of result.edges) { + const fromId = names.get(edge.from) ?? findByName(db, edge.from)?.id; + const toId = names.get(edge.to) ?? findByName(db, edge.to)?.id; + if (!fromId || !toId) continue; + upsertEdge(db, { + fromId, + toId, + type: edge.type, + instruction: edge.instruction, + condition: edge.condition, + sessionId: sid, + }); + } + if (result.nodes.length || result.edges.length) invalidateGraphCache(); + ctx.logger.info(`[graph-memory] DSH extracted ${result.nodes.length} nodes and ${result.edges.length} edges from ${sid}`); + } + + async function extractWithRetries(sessionId: unknown, sid: string, messages: any[]): Promise { + const ids = Array.from(new Set(messages.map(message => String(message.id)))); + for (let attempt = 0; attempt <= extractionDrain.maxRetries; attempt += 1) { + if (abortingExtraction) return new Error("[graph-memory] extraction aborted during shutdown"); + try { + await extractOnce(sessionId, sid, messages); + return; + } catch (cause) { + const error = cause instanceof Error ? cause : new Error(String(cause)); + const retrying = attempt < extractionDrain.maxRetries; + const delayMs = retrying ? extractionDrain.retryDelaysMs[attempt] : 0; + recordExtractionFailure(db, ids, error.message, retrying ? Date.now() + delayMs : null); + if (!retrying) return error; + ctx.logger.warn(`[graph-memory] DSH extraction retry ${attempt + 1}/${extractionDrain.maxRetries} in ${Math.round(delayMs / 1000)}s for ${sid}: ${error.message}`); + if (!await waitForRetry(delayMs)) return new Error("[graph-memory] extraction aborted during shutdown"); } - // Still failing: bisect so one poison message cannot sink the rest. - if (messages.length > 1) { - const mid = Math.ceil(messages.length / 2); - ctx.logger.warn(`[graph-memory] DSH extraction split ${messages.length} -> ${mid}+${messages.length - mid} for ${sid}: ${String(error)}`); - await drainBatch(sessionId, sid, messages.slice(0, mid), 0); - await drainBatch(sessionId, sid, messages.slice(mid), 0); + } + return new Error("[graph-memory] extraction retry loop ended unexpectedly"); + } + + async function drainBatch(sessionId: unknown, sid: string, messages: any[]): Promise { + if (abortingExtraction || !messages.length) return; + if (messages.length === 1) { + const original = messages[0]; + const chunks = splitExtractionContent(String(original.content ?? ""), extractionDrain.maxBatchChars); + if (chunks.length > 1) { + for (let index = 0; index < chunks.length; index += 1) { + const error = await extractWithRetries(sessionId, sid, [{ ...original, content: chunks[index] }]); + if (error) { + quarantineMessages(db, [String(original.id)], error.message); + ctx.logger.warn(`[graph-memory] DSH extraction quarantined turn=${original.turn_index}, segment=${index + 1}/${chunks.length} for ${sid}: ${error.message}`); + return; + } + } + markMessagesExtracted(db, [String(original.id)]); + ctx.logger.info(`[graph-memory] DSH losslessly extracted turn=${original.turn_index} in ${chunks.length} bounded segments for ${sid}`); return; } - markExtracted(db, sid, Number(messages[0].turn_index)); - ctx.logger.warn(`[graph-memory] DSH extraction SKIP turn=${messages[0].turn_index} after ${EXTRACTION_MAX_RETRIES + 1} tries for ${sid}`); } + + const error = await extractWithRetries(sessionId, sid, messages); + if (!error) { + markMessagesExtracted(db, messages.map(message => String(message.id))); + return; + } + if (messages.length > 1) { + const mid = Math.ceil(messages.length / 2); + ctx.logger.warn(`[graph-memory] DSH extraction split ${messages.length} -> ${mid}+${messages.length - mid} for ${sid}: ${error.message}`); + await drainBatch(sessionId, sid, messages.slice(0, mid)); + await drainBatch(sessionId, sid, messages.slice(mid)); + return; + } + quarantineMessages(db, [String(messages[0].id)], error.message); + ctx.logger.warn(`[graph-memory] DSH extraction quarantined turn=${messages[0].turn_index} after ${extractionDrain.maxRetries + 1} attempts for ${sid}: ${error.message}`); } async function extractPending(sessionId: unknown): Promise { - if (!extractionEnabled || closing) return; + if (!extractionEnabled || abortingExtraction) return; const sid = sessionKey(sessionId); - while (!closing) { - // Take the oldest unextracted messages in order, bounded by accumulated - // normalized length first (a single long message always fits so the - // drain can make progress), then by count. Order matters: we mark - // extracted up to the max turn_index of the batch, so skipping a - // middle message would mis-mark it as extracted. + while (!abortingExtraction) { const messages: any[] = []; let chars = 0; - for (const message of getUnextracted(db, sid, EXTRACTION_BATCH_MAX_MESSAGES * 16)) { + for (const message of getUnextracted(db, sid, extractionDrain.maxBatchMessages * 16)) { const content = normalizeExtractionContent(message.content); - if (messages.length > 0 && chars + content.length > EXTRACTION_BATCH_MAX_CHARS) break; + const contentChars = Array.from(content).length; + if (messages.length > 0 && chars + contentChars > extractionDrain.maxBatchChars) break; messages.push({ ...message, content }); - chars += content.length; - if (messages.length >= EXTRACTION_BATCH_MAX_MESSAGES) break; + chars += contentChars; + if (messages.length >= extractionDrain.maxBatchMessages || chars >= extractionDrain.maxBatchChars) break; } if (!messages.length) return; - await drainBatch(sessionId, sid, messages, 0); + await drainBatch(sessionId, sid, messages); } } - function scheduleExtract(sessionId: unknown): void { + function scheduleExtract(sessionId: unknown): Promise { + if (!extractionEnabled || closing) return Promise.resolve(); const key = String(sessionId); - const previous = extractChain.get(key) ?? Promise.resolve(); - const next = previous.then(() => extractPending(sessionId)); + const previous = extractChain.get(key); + const running = previous + ? previous.then(() => extractPending(sessionId)) + : extractPending(sessionId); + const next = running.catch(error => { + ctx.logger.error(`[graph-memory] DSH extraction queue failed for ${key}: ${String(error)}`); + }); extractChain.set(key, next); - void next.finally(() => { + void next.then(() => { if (extractChain.get(key) === next) extractChain.delete(key); }); + return next; } function runConfiguredRetention(): MessageRetentionResult { @@ -561,7 +626,6 @@ export function apply(ctx: DshContext, input: Config = {}): void { const id = agent?.id ?? agent?.session?.id; if (id === undefined || !Array.isArray(agent?.session?.events)) return; for (const event of agent.session.events) ingest(id, event); - scheduleExtract(id); } // Graph Memory owns the rolling retention policy while DSH's public @@ -587,7 +651,10 @@ export function apply(ctx: DshContext, input: Config = {}): void { // Agent preset services live in an isolated standing scope. Use // DSH's public roster seam instead of reaching through Cordis scope // internals or requiring a change in Harness itself. - const compaction = ctx.agentPresets.serviceFor(agent, "compaction"); + const agentPresets = typeof ctx.get === "function" + ? ctx.get("agentPresets") + : ctx.agentPresets; + const compaction = agentPresets?.serviceFor?.(agent, "compaction"); if (!compaction?.compactRegion) { compactionMetrics.unavailable += 1; if (!warnedMissingCompaction) { @@ -636,6 +703,18 @@ export function apply(ctx: DshContext, input: Config = {}): void { backfill(agent); }); + // DSH declares this as a serial, awaited lifecycle event before turn/end is + // committed. It is the reliable drain boundary for one-shot Headless: LLM + // adapters are still registered here, unlike ordinary session/event emit + // observers whose returned promises are intentionally ignored. + ctx.on("agent/turn-stopping", async ({ agent, signal }: any) => { + if (signal?.aborted) return; + const id = agent?.id ?? agent?.session?.id; + if (id === undefined) return; + backfill(agent); + await scheduleExtract(id); + }); + ctx.on("session/event", (session: any, event: any) => { const id = session?.id; if (id === undefined) return; @@ -648,7 +727,9 @@ export function apply(ctx: DshContext, input: Config = {}): void { ingest(id, event); recordCompactionCapsule(id, event); if (event?.type === "turn/end") { - scheduleExtract(id); + // This is a background fallback for hosts that do not expose the Agent + // turn-stopping boundary. DSH itself drains synchronously above. + void scheduleExtract(id); maintain(id); } }); @@ -738,8 +819,9 @@ export function apply(ctx: DshContext, input: Config = {}): void { ? ` (${input.embedding.model})` : ""; const messageCount = Number((db.prepare("SELECT COUNT(*) AS count FROM gm_messages").get() as any)?.count ?? 0); + const extraction = getExtractionStats(db); const retentionRevision = messageRetentionPolicyRevision(messageRetention); - return `Graph Memory active (DSH native)\nStore: ${config.dbPath}\nNodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nMessages: ${messageCount}\nExtraction: ${extractionEnabled ? "enabled" : "disabled"}\nRecall: ${recallEnabled ? "enabled" : "disabled"}\nEmbedding: ${embeddingState}${embeddingModel}\nVectors: ${vectors.count}/${stats.totalNodes}${vectors.dimensions.length ? ` (${vectors.dimensions.join(", ")} dimensions)` : ""}\nMessage retention: keep=${messageRetention.keep}, recentTurns=${messageRetention.recentTurns}, retentionDays=${messageRetention.retentionDays}, batchSize=${messageRetention.batchSize}, dryRun=${messageRetention.dryRun}, revision=${retentionRevision}\nRetention GC: runs=${retentionMetrics.runs}, dryRuns=${retentionMetrics.dryRuns}, selected=${retentionMetrics.selectedRows}, deleted=${retentionMetrics.deletedRows}, estimatedDeletedBytes=${retentionMetrics.deletedBytes}\nRolling compaction: attached=${compactionMetrics.attached}, selected=${compactionMetrics.selected}, succeeded=${compactionMetrics.succeeded}, unavailable=${compactionMetrics.unavailable}, failed=${compactionMetrics.failed}`; + return `Graph Memory active (DSH native)\nStore: ${config.dbPath}\nNodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nMessages: ${messageCount}\nExtraction: ${extractionEnabled ? "enabled" : "disabled"} (pending=${extraction.pending}, succeeded=${extraction.succeeded}, quarantined=${extraction.quarantined})\nExtraction drain: maxChars=${extractionDrain.maxBatchChars}, maxMessages=${extractionDrain.maxBatchMessages}, retries=${extractionDrain.maxRetries}, timeoutMs=${extractionDrain.streamTimeoutMs}\nRecall: ${recallEnabled ? "enabled" : "disabled"}\nEmbedding: ${embeddingState}${embeddingModel}\nVectors: ${vectors.count}/${stats.totalNodes}${vectors.dimensions.length ? ` (${vectors.dimensions.join(", ")} dimensions)` : ""}\nMessage retention: keep=${messageRetention.keep}, recentTurns=${messageRetention.recentTurns}, retentionDays=${messageRetention.retentionDays}, batchSize=${messageRetention.batchSize}, dryRun=${messageRetention.dryRun}, revision=${retentionRevision}\nRetention GC: runs=${retentionMetrics.runs}, dryRuns=${retentionMetrics.dryRuns}, selected=${retentionMetrics.selectedRows}, deleted=${retentionMetrics.deletedRows}, estimatedDeletedBytes=${retentionMetrics.deletedBytes}\nRolling compaction: attached=${compactionMetrics.attached}, selected=${compactionMetrics.selected}, succeeded=${compactionMetrics.succeeded}, unavailable=${compactionMetrics.unavailable}, failed=${compactionMetrics.failed}`; }, }); @@ -797,7 +879,7 @@ export function apply(ctx: DshContext, input: Config = {}): void { execute: async () => { const stats = getStats(db); const messageCount = Number((db.prepare("SELECT COUNT(*) AS count FROM gm_messages").get() as any)?.count ?? 0); - return `Nodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nCommunities: ${stats.communities}\nMessages: ${messageCount}\nBy type: ${JSON.stringify(stats.byType)}\nRetention policy: ${JSON.stringify({ ...messageRetention, revision: messageRetentionPolicyRevision(messageRetention) })}\nRetention totals: ${JSON.stringify({ runs: retentionMetrics.runs, dryRuns: retentionMetrics.dryRuns, selectedRows: retentionMetrics.selectedRows, deletedRows: retentionMetrics.deletedRows, deletedBytes: retentionMetrics.deletedBytes })}\nLast retention receipt: ${JSON.stringify(retentionMetrics.last ?? null)}`; + return `Nodes: ${stats.totalNodes}\nEdges: ${stats.totalEdges}\nCommunities: ${stats.communities}\nMessages: ${messageCount}\nExtraction queue: ${JSON.stringify(getExtractionStats(db))}\nBy type: ${JSON.stringify(stats.byType)}\nRetention policy: ${JSON.stringify({ ...messageRetention, revision: messageRetentionPolicyRevision(messageRetention) })}\nRetention totals: ${JSON.stringify({ runs: retentionMetrics.runs, dryRuns: retentionMetrics.dryRuns, selectedRows: retentionMetrics.selectedRows, deletedRows: retentionMetrics.deletedRows, deletedBytes: retentionMetrics.deletedBytes })}\nLast retention receipt: ${JSON.stringify(retentionMetrics.last ?? null)}`; }, }); @@ -809,9 +891,57 @@ export function apply(ctx: DshContext, input: Config = {}): void { execute: async () => JSON.stringify(runMaintenanceTick()), }); + ctx.tools.register({ + name: "gm_retry_extraction", + description: "Requeue quarantined durable messages and retry knowledge extraction without deleting source text.", + parameters: { + type: "object", + properties: { + sessionId: { type: "string", description: "Optional DSH session id; omit to requeue every quarantined session" }, + }, + additionalProperties: false, + }, + output: stringOutput("Graph Memory extraction retry"), + execute: async (args: any = {}) => { + const requested = typeof args.sessionId === "string" && args.sessionId.trim() + ? args.sessionId.trim() + : undefined; + const sid = requested + ? requested.startsWith(`${HOST}:`) ? requested : sessionKey(requested) + : undefined; + const requeued = requeueQuarantined(db, sid); + const pending = sid ? [sid] : getPendingSessionIds(db); + let scheduled = 0; + for (const pendingSid of pending) { + const rawId = pendingSid.startsWith(`${HOST}:`) ? pendingSid.slice(HOST.length + 1) : pendingSid; + if (input.llmProvider && input.llmModel || latestRoute.has(rawId)) { + scheduleExtract(rawId); + scheduled += 1; + } + } + return `Requeued ${requeued} quarantined messages; scheduled ${scheduled} sessions.`; + }, + }); + ctx.effect(() => async () => { closing = true; - await Promise.allSettled([...extractChain.values()]); + const chains = [...extractChain.values()]; + let graceTimer: ReturnType | undefined; + const drained = await Promise.race([ + Promise.allSettled(chains).then(() => true), + new Promise(resolve => { + graceTimer = setTimeout(() => resolve(false), extractionDrain.shutdownGraceMs); + }), + ]); + if (graceTimer) clearTimeout(graceTimer); + if (!drained) { + abortingExtraction = true; + for (const cancel of [...retryCancels]) cancel(); + for (const controller of activeExtractionControllers) { + controller.abort(new Error("[graph-memory] extraction shutdown grace elapsed")); + } + await Promise.allSettled([...extractChain.values()]); + } latestRoute.clear(); latestPrompt.clear(); recallCache.clear(); @@ -819,6 +949,14 @@ export function apply(ctx: DshContext, input: Config = {}): void { db.close(); }, "graph-memory.close"); + // With an explicit fallback route, recover durable pending work from prior + // process exits even when those sessions are not reopened in the UI. + if (extractionEnabled && input.llmProvider && input.llmModel) { + for (const sid of getPendingSessionIds(db)) { + scheduleExtract(sid.startsWith(`${HOST}:`) ? sid.slice(HOST.length + 1) : sid); + } + } + if (messageRetention.keep !== "all") { const mode = messageRetention.dryRun ? "dry-run" : "deletion enabled"; ctx.logger.warn( diff --git a/package.json b/package.json index 18d2330..4badb0d 100755 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "graph-memory", - "version": "1.6.0-beta.10", + "version": "1.6.0-beta.11", "description": "Knowledge graph memory for DeepSeek Harness and OpenClaw — cross-session recall, PageRank, communities, and vector search", "keywords": [ "deepseek-harness", diff --git a/src/extractor/drain-policy.ts b/src/extractor/drain-policy.ts new file mode 100644 index 0000000..fde2895 --- /dev/null +++ b/src/extractor/drain-policy.ts @@ -0,0 +1,98 @@ +export interface ExtractionDrainConfig { + maxBatchChars?: number; + maxBatchMessages?: number; + existingNamesMaxEntries?: number; + existingNamesMaxChars?: number; + streamTimeoutMs?: number; + maxRetries?: number; + retryDelaysMs?: number[]; + shutdownGraceMs?: number; +} + +export interface ExtractionDrainPolicy { + maxBatchChars: number; + maxBatchMessages: number; + existingNamesMaxEntries: number; + existingNamesMaxChars: number; + streamTimeoutMs: number; + maxRetries: number; + retryDelaysMs: number[]; + shutdownGraceMs: number; +} + +export const DEFAULT_EXTRACTION_DRAIN_POLICY: ExtractionDrainPolicy = { + maxBatchChars: 8_000, + maxBatchMessages: 15, + existingNamesMaxEntries: 150, + existingNamesMaxChars: 3_000, + streamTimeoutMs: 180_000, + maxRetries: 2, + retryDelaysMs: [5_000, 15_000], + shutdownGraceMs: 30_000, +}; + +function integer( + value: unknown, + name: string, + fallback: number, + min: number, + max: number, +): number { + const resolved = value ?? fallback; + if (!Number.isInteger(resolved) || Number(resolved) < min || Number(resolved) > max) { + throw new TypeError(`[graph-memory] extractionDrain.${name} must be an integer between ${min} and ${max}, received ${String(resolved)}`); + } + return Number(resolved); +} + +export function normalizeExtractionDrainPolicy( + input: ExtractionDrainConfig | undefined, +): ExtractionDrainPolicy { + const maxRetries = integer(input?.maxRetries, "maxRetries", DEFAULT_EXTRACTION_DRAIN_POLICY.maxRetries, 0, 10); + const retryDelays = input?.retryDelaysMs ?? DEFAULT_EXTRACTION_DRAIN_POLICY.retryDelaysMs; + if (!Array.isArray(retryDelays) || retryDelays.length < maxRetries || retryDelays.length > 10) { + throw new TypeError(`[graph-memory] extractionDrain.retryDelaysMs must contain between ${maxRetries} and 10 delays`); + } + const retryDelaysMs = retryDelays.map((delay, index) => ( + integer(delay, `retryDelaysMs[${index}]`, 0, 0, 300_000) + )); + return { + maxBatchChars: integer(input?.maxBatchChars, "maxBatchChars", DEFAULT_EXTRACTION_DRAIN_POLICY.maxBatchChars, 64, 1_000_000), + maxBatchMessages: integer(input?.maxBatchMessages, "maxBatchMessages", DEFAULT_EXTRACTION_DRAIN_POLICY.maxBatchMessages, 1, 1_000), + existingNamesMaxEntries: integer(input?.existingNamesMaxEntries, "existingNamesMaxEntries", DEFAULT_EXTRACTION_DRAIN_POLICY.existingNamesMaxEntries, 0, 10_000), + existingNamesMaxChars: integer(input?.existingNamesMaxChars, "existingNamesMaxChars", DEFAULT_EXTRACTION_DRAIN_POLICY.existingNamesMaxChars, 0, 1_000_000), + streamTimeoutMs: integer(input?.streamTimeoutMs, "streamTimeoutMs", DEFAULT_EXTRACTION_DRAIN_POLICY.streamTimeoutMs, 10, 900_000), + maxRetries, + retryDelaysMs, + shutdownGraceMs: integer(input?.shutdownGraceMs, "shutdownGraceMs", DEFAULT_EXTRACTION_DRAIN_POLICY.shutdownGraceMs, 0, 300_000), + }; +} + +/** + * Split only the temporary extraction projection. The durable message stays + * byte-for-byte unchanged in gm_messages. Boundaries prefer paragraphs and + * whitespace, and Array.from keeps surrogate pairs intact. + */ +export function splitExtractionContent(content: string, maxChars: number): string[] { + const points = Array.from(content); + if (points.length <= maxChars) return [content]; + + const chunks: string[] = []; + let offset = 0; + while (offset < points.length) { + const hardEnd = Math.min(offset + maxChars, points.length); + let end = hardEnd; + if (hardEnd < points.length) { + const softStart = offset + Math.floor(maxChars * 0.6); + for (let cursor = hardEnd - 1; cursor >= softStart; cursor -= 1) { + if (/\s/u.test(points[cursor])) { + end = cursor + 1; + break; + } + } + } + chunks.push(points.slice(offset, end).join("")); + offset = end; + } + return chunks; +} diff --git a/src/store/db.ts b/src/store/db.ts index 371ea10..00e49e5 100755 --- a/src/store/db.ts +++ b/src/store/db.ts @@ -80,6 +80,7 @@ function migrate(db: DatabaseSyncInstance): void { m8_backfill_community_signatures, m9_node_sources, m10_message_retention_index, + m11_extraction_queue_state, ]; for (let i = cur; i < steps.length; i++) { steps[i](db); @@ -87,6 +88,38 @@ function migrate(db: DatabaseSyncInstance): void { } } +// ─── 可审计抽取队列：待处理 / 成功 / 隔离 ────────────────────── + +function m11_extraction_queue_state(db: DatabaseSyncInstance): void { + const columns = new Set( + (db.prepare("PRAGMA table_info(gm_messages)").all() as Array<{ name: string }>) + .map(column => column.name), + ); + if (!columns.has("extraction_state")) { + db.exec(`ALTER TABLE gm_messages ADD COLUMN extraction_state TEXT NOT NULL DEFAULT 'pending' + CHECK(extraction_state IN ('pending', 'succeeded', 'quarantined'))`); + } + if (!columns.has("extraction_attempts")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_attempts INTEGER NOT NULL DEFAULT 0"); + } + if (!columns.has("extraction_error")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_error TEXT"); + } + if (!columns.has("extraction_next_retry_at")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_next_retry_at INTEGER"); + } + if (!columns.has("extraction_updated_at")) { + db.exec("ALTER TABLE gm_messages ADD COLUMN extraction_updated_at INTEGER"); + } + db.exec(` + UPDATE gm_messages + SET extraction_state=CASE WHEN extracted=1 THEN 'succeeded' ELSE 'pending' END, + extraction_updated_at=COALESCE(extraction_updated_at, created_at); + CREATE INDEX IF NOT EXISTS ix_gm_msg_extraction_queue + ON gm_messages(extraction_state, extraction_next_retry_at, session_id, turn_index); + `); +} + // ─── 有界原始消息保留策略查询 ──────────────────────────────── function m10_message_retention_index(db: DatabaseSyncInstance): void { diff --git a/src/store/retention.ts b/src/store/retention.ts index 363c9f4..8c45bee 100644 --- a/src/store/retention.ts +++ b/src/store/retention.ts @@ -185,7 +185,7 @@ function selectCandidates( length(CAST(m.content AS BLOB)) AS content_bytes FROM gm_messages m ${recentJoin} - WHERE m.extracted=1 + WHERE m.extracted=1 AND m.extraction_state='succeeded' AND NOT EXISTS ( SELECT 1 FROM gm_node_sources source WHERE source.message_id=m.id ) @@ -228,7 +228,7 @@ export function runMessageRetention( const deleted = db.prepare(` DELETE FROM gm_messages WHERE id IN (${placeholders}) - AND extracted=1 + AND extracted=1 AND extraction_state='succeeded' AND NOT EXISTS ( SELECT 1 FROM gm_node_sources source WHERE source.message_id=gm_messages.id ) diff --git a/src/store/store.ts b/src/store/store.ts index 336b3df..e2f0c2f 100755 --- a/src/store/store.ts +++ b/src/store/store.ts @@ -367,13 +367,108 @@ export function getMessages(db: DatabaseSyncInstance, sid: string, limit?: numbe } export function getUnextracted(db: DatabaseSyncInstance, sid: string, limit: number): any[] { - return db.prepare("SELECT * FROM gm_messages WHERE session_id=? AND extracted=0 ORDER BY turn_index LIMIT ?") - .all(sid, limit) as any[]; + return db.prepare(` + SELECT * FROM gm_messages + WHERE session_id=? AND extracted=0 AND extraction_state='pending' + AND (extraction_next_retry_at IS NULL OR extraction_next_retry_at<=?) + ORDER BY turn_index, id LIMIT ? + `).all(sid, Date.now(), limit) as any[]; } export function markExtracted(db: DatabaseSyncInstance, sid: string, upToTurn: number): void { - db.prepare("UPDATE gm_messages SET extracted=1 WHERE session_id=? AND turn_index<=?") - .run(sid, upToTurn); + db.prepare(` + UPDATE gm_messages + SET extracted=1, extraction_state='succeeded', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE session_id=? AND turn_index<=? AND extraction_state='pending' + `).run(Date.now(), sid, upToTurn); +} + +function messageIdPlaceholders(ids: string[]): string { + return ids.map(() => "?").join(","); +} + +/** Mark only the source rows actually accepted by the extractor. */ +export function markMessagesExtracted(db: DatabaseSyncInstance, ids: string[]): number { + if (!ids.length) return 0; + const result = db.prepare(` + UPDATE gm_messages + SET extracted=1, extraction_state='succeeded', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE id IN (${messageIdPlaceholders(ids)}) AND extraction_state='pending' + `).run(Date.now(), ...ids); + return Number(result.changes); +} + +export function recordExtractionFailure( + db: DatabaseSyncInstance, + ids: string[], + error: string, + nextRetryAt: number | null, +): number { + if (!ids.length) return 0; + const result = db.prepare(` + UPDATE gm_messages + SET extraction_attempts=extraction_attempts+1, extraction_error=?, + extraction_next_retry_at=?, extraction_updated_at=? + WHERE id IN (${messageIdPlaceholders(ids)}) AND extraction_state='pending' + `).run(error.slice(0, 2_000), nextRetryAt, Date.now(), ...ids); + return Number(result.changes); +} + +/** A poison message remains durable and explicitly unlearned until retried. */ +export function quarantineMessages(db: DatabaseSyncInstance, ids: string[], error: string): number { + if (!ids.length) return 0; + const result = db.prepare(` + UPDATE gm_messages + SET extracted=0, extraction_state='quarantined', extraction_error=?, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE id IN (${messageIdPlaceholders(ids)}) AND extraction_state='pending' + `).run(error.slice(0, 2_000), Date.now(), ...ids); + return Number(result.changes); +} + +export function requeueQuarantined(db: DatabaseSyncInstance, sid?: string): number { + const result = sid + ? db.prepare(` + UPDATE gm_messages + SET extraction_state='pending', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE extraction_state='quarantined' AND session_id=? + `).run(Date.now(), sid) + : db.prepare(` + UPDATE gm_messages + SET extraction_state='pending', extraction_error=NULL, + extraction_next_retry_at=NULL, extraction_updated_at=? + WHERE extraction_state='quarantined' + `).run(Date.now()); + return Number(result.changes); +} + +export function getExtractionStats(db: DatabaseSyncInstance): { + pending: number; + succeeded: number; + quarantined: number; +} { + const rows = db.prepare(` + SELECT extraction_state AS state, COUNT(*) AS count + FROM gm_messages GROUP BY extraction_state + `).all() as Array<{ state: string; count: number }>; + const result = { pending: 0, succeeded: 0, quarantined: 0 }; + for (const row of rows) { + if (row.state in result) (result as any)[row.state] = Number(row.count); + } + return result; +} + +export function getPendingSessionIds(db: DatabaseSyncInstance, limit: number = 100): string[] { + return (db.prepare(` + SELECT session_id, MIN(turn_index) AS first_turn + FROM gm_messages + WHERE extracted=0 AND extraction_state='pending' + AND (extraction_next_retry_at IS NULL OR extraction_next_retry_at<=?) + GROUP BY session_id ORDER BY first_turn, session_id LIMIT ? + `).all(Date.now(), limit) as Array<{ session_id: string }>).map(row => row.session_id); } /** diff --git a/test/dsh-adapter.test.ts b/test/dsh-adapter.test.ts index 3dd7122..c66a680 100644 --- a/test/dsh-adapter.test.ts +++ b/test/dsh-adapter.test.ts @@ -1,10 +1,11 @@ import { describe, expect, it } from "vitest"; -import { mkdtempSync, rmSync } from "node:fs"; +import { existsSync, mkdtempSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { apply, eventMessage } from "../dsh.ts"; +import { apply, eventMessage, inject } from "../dsh.ts"; import { DatabaseSync } from "../src/store/sqlite.ts"; +import { normalizeExtractionContent } from "../src/extractor/extract.ts"; function user(seq: number) { return { @@ -22,6 +23,17 @@ describe("native DSH context takeover", () => { })).toThrow(/requires recentTurns or retentionDays/); }); + it("validates extraction bounds before opening the configured database", () => { + const dir = mkdtempSync(join(tmpdir(), "gm-config-")); + const dbPath = join(dir, "must-not-open.db"); + expect(() => apply({} as any, { + dbPath, + extractionDrain: { maxBatchChars: 1 }, + })).toThrow(/extractionDrain.maxBatchChars/); + expect(existsSync(dbPath)).toBe(false); + rmSync(dir, { recursive: true, force: true }); + }); + it("exposes the effective retention policy and bounded maintenance receipt", async () => { const tools = new Map(); const cleanups: Array<() => void | Promise> = []; @@ -305,15 +317,16 @@ function userMsg(seq: number, text: string) { const EMPTY_EXTRACTION = '{"nodes":[],"edges":[]}'; -function adapterContext(llmStream: () => AsyncGenerator) { +function adapterContext(llmStream: (options?: any) => AsyncGenerator) { const listeners = new Map any>>(); const cleanups: Array<() => void | Promise> = []; + const tools = new Map(); const logs: string[] = []; const log = (level: string) => (...args: any[]) => logs.push(`${level}:${args.join(" ")}`); const context: any = { logger: { info: log("info"), warn: log("warn"), error: log("error") }, llm: { stream: llmStream }, - tools: { register() { return () => {}; } }, + tools: { register(definition: any) { tools.set(definition.name, definition); return () => {}; } }, credentials: { async resolve() { return undefined; } }, agentPresets: { serviceFor() { return undefined; } }, on(name: string, listener: (...args: any[]) => any, options?: Record) { @@ -328,7 +341,7 @@ function adapterContext(llmStream: () => AsyncGenerator) { return () => {}; }, }; - return { context, listeners, cleanups, logs }; + return { context, listeners, cleanups, logs, tools }; } async function waitFor(check: () => boolean, timeoutMs = 8000): Promise { @@ -342,14 +355,40 @@ async function waitFor(check: () => boolean, timeoutMs = 8000): Promise { function countPending(dbPath: string): number { const db = new DatabaseSync(dbPath); try { - const row = db.prepare("SELECT COUNT(*) AS c FROM gm_messages WHERE extracted = 0").get() as any; + const row = db.prepare("SELECT COUNT(*) AS c FROM gm_messages WHERE extraction_state = 'pending'").get() as any; + return Number(row.c); + } finally { + db.close(); + } +} + +function countState(dbPath: string, state: string): number { + const db = new DatabaseSync(dbPath); + try { + const row = db.prepare("SELECT COUNT(*) AS c FROM gm_messages WHERE extraction_state = ?").get(state) as any; return Number(row.c); } finally { db.close(); } } +async function startAndEndTurn( + listeners: Map any>>, + agent: any, +): Promise { + listeners.get("agent/session-start")![0]({ agent }); + await listeners.get("agent/turn-stopping")![0]({ + agent, + signal: new AbortController().signal, + }); + await listeners.get("session/event")![0]({ id: agent.id }, { type: "turn/end", seq: 99_999 }); +} + describe("extraction drain resilience", () => { + it("does not require the UI-only agentPresets service", () => { + expect(inject).not.toContain("agentPresets"); + }); + it("retries a transient LLM failure and then drains the backlog", async () => { const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); const dbPath = join(dir, "graph-memory.db"); @@ -369,17 +408,17 @@ describe("extraction drain resilience", () => { extractionRetryDelaysMs: [0, 0], }); const agent = { id: "transient-test", session: { events: [userMsg(0, "alpha"), userMsg(1, "beta")] } }; - listeners.get("agent/session-start")![0]({ agent }); + await startAndEndTurn(listeners, agent); await waitFor(() => logs.some(l => l.includes("DSH extracted"))); - expect(logs.some(l => l.includes("retry in 0s"))).toBe(true); + expect(logs.some(l => l.includes("retry 1/2 in 0s"))).toBe(true); expect(logs.some(l => l.includes("SKIP"))).toBe(false); expect(countPending(dbPath)).toBe(0); await Promise.all(cleanups.map(cleanup => cleanup())); rmSync(dir, { recursive: true, force: true }); }); - it("never deadlocks on a permanently failing batch (retry -> bisect -> skip)", async () => { + it("quarantines permanently failing messages without pretending they were extracted", async () => { const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); const dbPath = join(dir, "graph-memory.db"); const { context, listeners, cleanups, logs } = adapterContext(async function* () { @@ -397,15 +436,14 @@ describe("extraction drain resilience", () => { id: "poison-test", session: { events: [userMsg(0, "alpha"), userMsg(1, "beta"), userMsg(2, "gamma")] }, }; - listeners.get("agent/session-start")![0]({ agent }); + await startAndEndTurn(listeners, agent); - // Three messages: batch(3) fails out -> bisect -> each singleton skips. - await waitFor(() => logs.filter(l => l.includes("DSH extraction SKIP")).length >= 3); - const skipLogs = logs.filter(l => l.includes("DSH extraction SKIP")); - expect(skipLogs.length).toBe(3); + await waitFor(() => logs.filter(l => l.includes("DSH extraction quarantined")).length >= 3); + expect(logs.filter(l => l.includes("DSH extraction quarantined"))).toHaveLength(3); expect(logs.some(l => l.includes("split 3 -> 2+1"))).toBe(true); - // The drain still makes progress: every message is marked extracted. expect(countPending(dbPath)).toBe(0); + expect(countState(dbPath, "quarantined")).toBe(3); + expect(countState(dbPath, "succeeded")).toBe(0); await Promise.all(cleanups.map(cleanup => cleanup())); rmSync(dir, { recursive: true, force: true }); }); @@ -434,7 +472,7 @@ describe("extraction drain resilience", () => { id: "length-test", session: { events: [userMsg(0, long), userMsg(1, "short-a"), userMsg(2, "short-b")] }, }; - listeners.get("agent/session-start")![0]({ agent }); + await startAndEndTurn(listeners, agent); await waitFor(() => logs.filter(l => l.includes("DSH extracted")).length >= 2); expect(users.length).toBeGreaterThanOrEqual(2); @@ -443,7 +481,102 @@ describe("extraction drain resilience", () => { expect(user.length).toBeLessThan(12_000); } expect(countPending(dbPath)).toBe(0); + const db = new DatabaseSync(dbPath); + const stored = db.prepare("SELECT content FROM gm_messages WHERE turn_index=0").get() as any; + expect(normalizeExtractionContent(stored.content)).toBe(long); + db.close(); + await Promise.all(cleanups.map(cleanup => cleanup())); + rmSync(dir, { recursive: true, force: true }); + }); + + it("passes an AbortSignal to DSH and aborts a stalled provider on timeout", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); + const dbPath = join(dir, "graph-memory.db"); + let observedSignal: AbortSignal | undefined; + const { context, listeners, cleanups } = adapterContext(async function* (opts: any) { + observedSignal = opts.signal; + await new Promise((_resolve, reject) => { + opts.signal.addEventListener("abort", () => reject(opts.signal.reason), { once: true }); + }); + if (false) yield undefined; + }); + apply(context, { + dbPath, + extractionEnabled: true, + recallEnabled: false, + llmProvider: "test-provider", + llmModel: "test-model", + extractionDrain: { streamTimeoutMs: 20, maxRetries: 0, retryDelaysMs: [] }, + }); + await startAndEndTurn(listeners, { + id: "timeout-test", session: { events: [userMsg(0, "stall")] }, + }); + + await waitFor(() => countState(dbPath, "quarantined") === 1); + expect(observedSignal).toBeInstanceOf(AbortSignal); + expect(observedSignal?.aborted).toBe(true); await Promise.all(cleanups.map(cleanup => cleanup())); rmSync(dir, { recursive: true, force: true }); }); + + it("requeues quarantined messages explicitly and learns them on a later healthy call", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); + const dbPath = join(dir, "graph-memory.db"); + let healthy = false; + const { context, listeners, cleanups, tools } = adapterContext(async function* () { + if (!healthy) throw new Error("provider unavailable"); + yield { type: "text-delta", text: EMPTY_EXTRACTION }; + yield { type: "finish", reason: { kind: "stop" } }; + }); + apply(context, { + dbPath, + extractionEnabled: true, + recallEnabled: false, + llmProvider: "test-provider", + llmModel: "test-model", + extractionDrain: { maxRetries: 0, retryDelaysMs: [] }, + }); + await startAndEndTurn(listeners, { + id: "retry-tool-test", session: { events: [userMsg(0, "remember me")] }, + }); + await waitFor(() => countState(dbPath, "quarantined") === 1); + + healthy = true; + const result = await tools.get("gm_retry_extraction").execute({ sessionId: "retry-tool-test" }); + expect(result).toContain("Requeued 1"); + await waitFor(() => countState(dbPath, "succeeded") === 1); + expect(countState(dbPath, "quarantined")).toBe(0); + await Promise.all(cleanups.map(cleanup => cleanup())); + rmSync(dir, { recursive: true, force: true }); + }); + + it("gracefully drains work scheduled immediately before host shutdown", async () => { + const dir = mkdtempSync(join(tmpdir(), "gm-resilience-")); + const dbPath = join(dir, "graph-memory.db"); + const { context, listeners, cleanups } = adapterContext(async function* () { + await new Promise(resolve => setTimeout(resolve, 20)); + yield { type: "text-delta", text: EMPTY_EXTRACTION }; + yield { type: "finish", reason: { kind: "stop" } }; + }); + apply(context, { + dbPath, + extractionEnabled: true, + recallEnabled: false, + llmProvider: "test-provider", + llmModel: "test-model", + extractionDrain: { shutdownGraceMs: 1_000 }, + }); + const agent = { id: "shutdown-test", session: { events: [userMsg(0, "last message")] } }; + listeners.get("agent/session-start")![0]({ agent }); + const ending = listeners.get("agent/turn-stopping")![0]({ + agent, + signal: new AbortController().signal, + }); + + await Promise.all(cleanups.map(cleanup => cleanup())); + await ending; + expect(countState(dbPath, "succeeded")).toBe(1); + expect(countPending(dbPath)).toBe(0); + rmSync(dir, { recursive: true, force: true }); + }); }); diff --git a/test/helpers.ts b/test/helpers.ts index 67c8b81..c19259f 100755 --- a/test/helpers.ts +++ b/test/helpers.ts @@ -61,11 +61,19 @@ export function createTestDb(): DatabaseSyncInstance { role TEXT NOT NULL, content TEXT NOT NULL, extracted INTEGER NOT NULL DEFAULT 0, + extraction_state TEXT NOT NULL DEFAULT 'pending' + CHECK(extraction_state IN ('pending', 'succeeded', 'quarantined')), + extraction_attempts INTEGER NOT NULL DEFAULT 0, + extraction_error TEXT, + extraction_next_retry_at INTEGER, + extraction_updated_at INTEGER, created_at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS ix_gm_msg_session ON gm_messages(session_id, turn_index); CREATE INDEX IF NOT EXISTS ix_gm_msg_retention ON gm_messages(extracted, created_at, session_id, turn_index); + CREATE INDEX IF NOT EXISTS ix_gm_msg_extraction_queue + ON gm_messages(extraction_state, extraction_next_retry_at, session_id, turn_index); `); // m3: 信号 diff --git a/test/message-retention.test.ts b/test/message-retention.test.ts index 83b22e7..cff6285 100644 --- a/test/message-retention.test.ts +++ b/test/message-retention.test.ts @@ -16,8 +16,8 @@ function insertMessage( ) { db.prepare(` INSERT INTO gm_messages - (id, session_id, turn_index, role, content, extracted, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?) + (id, session_id, turn_index, role, content, extracted, extraction_state, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) `).run( id, session, @@ -25,6 +25,7 @@ function insertMessage( role, options.content ?? `${role}-${turn}`, options.extracted ?? 1, + (options.extracted ?? 1) === 1 ? "succeeded" : "pending", options.createdAt ?? 1_700_000_000_000 + turn, ); } @@ -72,6 +73,22 @@ describe("message retention policy", () => { db.close(); }); + it("never deletes quarantined messages even if a legacy flag is inconsistent", () => { + const db = createTestDb(); + insertMessage(db, "safe", "s1", 1, "user"); + db.prepare(` + UPDATE gm_messages SET extraction_state='quarantined' WHERE id='safe' + `).run(); + + const result = runMessageRetention( + db, + normalizeMessageRetentionPolicy({ keep: "referenced", batchSize: 10 }), + ); + expect(result.deletedRows).toBe(0); + expect(ids(db)).toEqual(["safe"]); + db.close(); + }); + it("preserves complete event spans for the newest real user turns", () => { const db = createTestDb(); for (const [turn, role] of [ diff --git a/test/migration.test.ts b/test/migration.test.ts index 5f18a24..6d9b774 100644 --- a/test/migration.test.ts +++ b/test/migration.test.ts @@ -70,6 +70,10 @@ describe("database migrations", () => { INSERT INTO gm_communities (id, summary, node_count, embedding, created_at, updated_at) VALUES ('c1', 'legacy summary', 2, NULL, 1, 1); + INSERT INTO gm_messages + (id, session_id, turn_index, role, content, extracted, created_at) + VALUES ('done', 's1', 1, 'user', 'done', 1, 1), + ('todo', 's1', 2, 'assistant', 'todo', 0, 2); `); const migration = legacy.prepare("INSERT INTO _migrations (v, at) VALUES (?, ?)"); for (let version = 1; version <= 6; version++) migration.run(version, 1); @@ -85,12 +89,22 @@ describe("database migrations", () => { expect(row.member_signature).toMatch(/^[a-f0-9]{40}$/); expect( (upgraded.prepare("SELECT MAX(v) AS version FROM _migrations").get() as any).version, - ).toBe(10); + ).toBe(11); const sourceColumns = upgraded.prepare("PRAGMA table_info(gm_node_sources)").all() as Array<{ name: string }>; expect(sourceColumns.map((column) => column.name)).toEqual([ "node_id", "session_id", "message_id", "turn_index", ]); const messageIndexes = upgraded.prepare("PRAGMA index_list(gm_messages)").all() as Array<{ name: string }>; expect(messageIndexes.some((index) => index.name === "ix_gm_msg_retention")).toBe(true); + expect(messageIndexes.some((index) => index.name === "ix_gm_msg_extraction_queue")).toBe(true); + const messageColumns = upgraded.prepare("PRAGMA table_info(gm_messages)").all() as Array<{ name: string }>; + expect(messageColumns.map(column => column.name)).toContain("extraction_state"); + const queueRows = upgraded.prepare( + "SELECT id, extraction_state FROM gm_messages ORDER BY id", + ).all() as Array<{ id: string; extraction_state: string }>; + expect(queueRows).toEqual([ + { id: "done", extraction_state: "succeeded" }, + { id: "todo", extraction_state: "pending" }, + ]); }); }); diff --git a/test/store.test.ts b/test/store.test.ts index 3fb4fdd..d40e0b6 100755 --- a/test/store.test.ts +++ b/test/store.test.ts @@ -13,6 +13,7 @@ import { mergeNodes, edgesFrom, edgesTo, allActiveNodes, allEdges, searchNodes, topNodes, graphWalk, getBySession, saveMessage, saveMessageOnce, getMessages, getUnextracted, markExtracted, getEpisodicMessages, + getExtractionStats, markMessagesExtracted, quarantineMessages, requeueQuarantined, getNodeSourceMessages, saveSignal, pendingSignals, markSignalsDone, getStats, saveVector, vectorSearch, getAllVectors, upsertCommunitySummary, @@ -372,6 +373,20 @@ describe("messages & signals", () => { expect(unext[0].turn_index).toBe(3); }); + it("tracks exact extraction success and quarantine without crossing turn boundaries", () => { + saveMessageOnce(db, "a", "s1", 1, "user", "first"); + saveMessageOnce(db, "b", "s1", 1, "tool", "same turn, different event"); + saveMessageOnce(db, "c", "s1", 2, "assistant", "third"); + + markMessagesExtracted(db, ["b"]); + quarantineMessages(db, ["a"], "poison input"); + expect(getExtractionStats(db)).toEqual({ pending: 1, succeeded: 1, quarantined: 1 }); + expect((db.prepare("SELECT extracted FROM gm_messages WHERE id='a'").get() as any).extracted).toBe(0); + + expect(requeueQuarantined(db, "s1")).toBe(1); + expect(getExtractionStats(db)).toEqual({ pending: 2, succeeded: 1, quarantined: 0 }); + }); + it("saveSignal + pendingSignals + markSignalsDone", () => { saveSignal(db, "s1", { type: "tool_error", turnIndex: 3, data: { snippet: "Error: xxx" } }); saveSignal(db, "s1", { type: "task_completed", turnIndex: 5, data: { snippet: "done" } });