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 627af72..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 { Extractor } from "./src/extractor/extract.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";
@@ -19,7 +20,7 @@ import { detectCommunities } from "./src/graph/community.js";
import { DEFAULT_CONFIG } from "./src/types.js";
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) {
@@ -102,6 +103,11 @@ export function apply(ctx, input = {}) {
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)}`);
@@ -132,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 = {
@@ -179,34 +187,71 @@ 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 = "";
- 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"}`);
+ let streamTimer;
+ 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")
+ 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"}`);
+ }
+ }
+ })();
+ 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);
@@ -252,64 +297,173 @@ export function apply(ctx, input = {}) {
void recaller.syncEmbed(node);
invalidateGraphCache();
}
+ // 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;
+ 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 >= extractionDrain.existingNamesMaxEntries)
+ break;
+ if (chars + name.length > extractionDrain.existingNamesMaxChars)
+ continue;
+ names.push(name);
+ chars += name.length;
+ }
+ return names;
+ }
+ 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);
+ }
+ 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;
+ }
+ 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");
+ }
+ }
+ 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;
+ }
+ }
+ 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) {
- 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}`);
+ while (!abortingExtraction) {
+ const messages = [];
+ let chars = 0;
+ for (const message of getUnextracted(db, sid, extractionDrain.maxBatchMessages * 16)) {
+ const content = normalizeExtractionContent(message.content);
+ const contentChars = Array.from(content).length;
+ if (messages.length > 0 && chars + contentChars > extractionDrain.maxBatchChars)
+ break;
+ messages.push({ ...message, content });
+ chars += contentChars;
+ if (messages.length >= extractionDrain.maxBatchMessages || chars >= extractionDrain.maxBatchChars)
+ 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);
}
}
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);
@@ -367,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
@@ -385,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) {
@@ -426,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)
@@ -439,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);
}
});
@@ -528,8 +699,9 @@ export function apply(ctx, input = {}) {
? ` (${input.embedding.model})`
: "";
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"}\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}`;
},
});
ctx.tools.register({
@@ -585,7 +757,7 @@ export function apply(ctx, input = {}) {
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}\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({
@@ -595,15 +767,71 @@ export function apply(ctx, input = {}) {
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)}). ` +
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 080b869..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 } from "./src/extractor/extract.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";
@@ -39,7 +49,7 @@ import {
} from "./src/store/retention.ts";
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;
@@ -70,6 +80,12 @@ export interface Config {
llmModel?: string;
llmMaxTokens?: number;
embedding?: DshEmbeddingConfig;
+ /** Bounded, lossless and durable extraction queue policy. */
+ extractionDrain?: ExtractionDrainConfig;
+ /** @deprecated Use extractionDrain.streamTimeoutMs. */
+ extractionStreamTimeoutMs?: number;
+ /** @deprecated Use extractionDrain.retryDelaysMs. */
+ extractionRetryDelaysMs?: number[];
}
interface Route {
@@ -95,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;
}
@@ -184,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)}`);
@@ -217,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