mirror of
https://github.com/TencentCloud/TencentDB-Agent-Memory
synced 2026-07-10 12:34:27 +00:00
feat: release v0.2.2 — TCVDB backend, BM25 hybrid retrieval, pipeline refactor
This commit is contained in:
chrishuan
@@ -10,7 +10,6 @@
|
||||
* All processing is local, zero external API dependencies.
|
||||
*/
|
||||
|
||||
import fs from "node:fs";
|
||||
import path from "node:path";
|
||||
import { createRequire } from "node:module";
|
||||
import type { OpenClawPluginApi } from "openclaw/plugin-sdk/core";
|
||||
@@ -19,54 +18,33 @@ import type { MemoryTdaiConfig } from "./src/config.js";
|
||||
import { performAutoRecall } from "./src/hooks/auto-recall.js";
|
||||
import { performAutoCapture } from "./src/hooks/auto-capture.js";
|
||||
import { MemoryPipelineManager } from "./src/utils/pipeline-manager.js";
|
||||
import { SceneExtractor } from "./src/scene/scene-extractor.js";
|
||||
import { CheckpointManager } from "./src/utils/checkpoint.js";
|
||||
import { PersonaTrigger } from "./src/persona/persona-trigger.js";
|
||||
import { PersonaGenerator } from "./src/persona/persona-generator.js";
|
||||
import { prewarmEmbeddedAgent } from "./src/utils/clean-context-runner.js";
|
||||
import {
|
||||
prewarmEmbeddedAgent,
|
||||
setPreferredEmbeddedAgentRuntime,
|
||||
} from "./src/utils/clean-context-runner.js";
|
||||
import { SessionFilter } from "./src/utils/session-filter.js";
|
||||
import { extractL1Memories } from "./src/record/l1-extractor.js";
|
||||
import { readConversationMessagesGroupedBySessionId } from "./src/conversation/l0-recorder.js";
|
||||
import type { ConversationMessage } from "./src/conversation/l0-recorder.js";
|
||||
import { VectorStore } from "./src/store/vector-store.js";
|
||||
import { createEmbeddingService } from "./src/store/embedding.js";
|
||||
import type { IMemoryStore } from "./src/store/types.js";
|
||||
import type { EmbeddingService } from "./src/store/embedding.js";
|
||||
import { executeMemorySearch, formatSearchResponse } from "./src/tools/memory-search.js";
|
||||
import { executeConversationSearch, formatConversationSearchResponse } from "./src/tools/conversation-search.js";
|
||||
import { LocalMemoryCleaner } from "./src/utils/memory-cleaner.js";
|
||||
import { getOrCreateInstanceId, initReporter, report } from "./src/report/reporter.js";
|
||||
import { registerMemoryTdaiCli } from "./src/cli/index.js";
|
||||
import {
|
||||
initDataDirectories,
|
||||
initStores,
|
||||
resetStores,
|
||||
createPipelineManager,
|
||||
createL1Runner,
|
||||
createPersister,
|
||||
createL2Runner,
|
||||
createL3Runner,
|
||||
} from "./src/utils/pipeline-factory.js";
|
||||
import { getOrCreateInstanceId, initReporter, report, resetReporter } from "./src/report/reporter.js";
|
||||
import { ensureL2L3Local } from "./src/profile/profile-sync.js";
|
||||
|
||||
const TAG = "[memory-tdai]";
|
||||
|
||||
/**
|
||||
* Initialize all required data directories under the plugin data root.
|
||||
*
|
||||
* Called once at plugin registration time so downstream modules
|
||||
* (L0 recorder, L1 writer, scene extractor, persona generator, etc.)
|
||||
* don't need to lazily mkdir on every write — the directories are
|
||||
* guaranteed to exist from startup.
|
||||
*
|
||||
* Directory layout:
|
||||
* <pluginDataDir>/
|
||||
* ├── conversations/ — L0 daily JSONL shards (one message per line)
|
||||
* ├── records/ — L1 daily JSONL shards (extracted memories)
|
||||
* ├── scene_blocks/ — L2 scene block .md files (LLM-managed)
|
||||
* ├── .metadata/ — checkpoint, scene_index.json
|
||||
* └── .backup/ — rotating backups (persona, scene_blocks)
|
||||
*/
|
||||
function initDataDirectories(dataDir: string): void {
|
||||
const dirs = [
|
||||
"conversations",
|
||||
"records",
|
||||
"scene_blocks",
|
||||
".metadata",
|
||||
".backup",
|
||||
];
|
||||
for (const sub of dirs) {
|
||||
fs.mkdirSync(path.join(dataDir, sub), { recursive: true });
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Epoch ms when the plugin was registered (cold-start timestamp).
|
||||
* Used as a fallback cursor in performAutoCapture when no checkpoint
|
||||
@@ -151,20 +129,20 @@ function sweepStaleCaches(): void {
|
||||
|
||||
export default function register(api: OpenClawPluginApi) {
|
||||
pluginStartTimestamp = Date.now();
|
||||
setPreferredEmbeddedAgentRuntime(api.runtime.agent);
|
||||
// Reset reporter singleton so config changes take effect on hot-reload.
|
||||
resetReporter();
|
||||
const _require = createRequire(import.meta.url);
|
||||
const pluginVersion = (() => { try { return (_require("./package.json") as { version?: string }).version ?? "unknown"; } catch { return "unknown"; } })();
|
||||
api.logger.info(
|
||||
api.logger.debug?.(
|
||||
`${TAG} Registering plugin ... ` +
|
||||
`startTimestamp=${pluginStartTimestamp} (${new Date(pluginStartTimestamp).toISOString()})`,
|
||||
);
|
||||
|
||||
// Persistent instance ID for metric reporting (populated async below)
|
||||
let instanceId: string | undefined;
|
||||
|
||||
let cfg: MemoryTdaiConfig;
|
||||
try {
|
||||
cfg = parseConfig(api.pluginConfig as Record<string, unknown> | undefined);
|
||||
api.logger.info(
|
||||
api.logger.debug?.(
|
||||
`${TAG} Config parsed: ` +
|
||||
`capture=${cfg.capture.enabled}, ` +
|
||||
`recall=${cfg.recall.enabled}(maxResults=${cfg.recall.maxResults}), ` +
|
||||
@@ -186,12 +164,26 @@ export default function register(api: OpenClawPluginApi) {
|
||||
// Resolve plugin data directory via runtime API (avoid importing internal paths directly)
|
||||
const pluginDataDir = path.join(api.runtime.state.resolveStateDir(), "memory-tdai");
|
||||
initDataDirectories(pluginDataDir);
|
||||
api.logger.info(`${TAG} Data dir: ${pluginDataDir} (all subdirectories initialized)`);
|
||||
api.logger.debug?.(`${TAG} Data dir: ${pluginDataDir} (all subdirectories initialized)`);
|
||||
|
||||
// Kick off instanceId resolution immediately after data dir is ready.
|
||||
// getOrCreateInstanceId only reads/writes a small UUID file and caches the
|
||||
// result — starting it here means it will almost certainly be settled before
|
||||
// the first L1 runner fires, avoiding the need to defer metric reporting.
|
||||
let instanceId: string | undefined;
|
||||
getOrCreateInstanceId(pluginDataDir).then((id) => {
|
||||
instanceId = id;
|
||||
// initReporter is guarded by a "already initialised" check, so calling it
|
||||
// here is safe even if the registration-complete call below fires first.
|
||||
initReporter({ enabled: cfg.report.enabled, type: cfg.report.type, logger: api.logger, instanceId: id, pluginVersion });
|
||||
}).catch((err) => {
|
||||
api.logger.warn(`${TAG} Failed to initialize instanceId for metrics: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
|
||||
// Unified session/agent filter: combines internal-session detection + user-configured excludeAgents
|
||||
const sessionFilter = new SessionFilter(cfg.capture.excludeAgents);
|
||||
if (cfg.capture.excludeAgents.length > 0) {
|
||||
api.logger.info(`${TAG} Agent exclude patterns: ${cfg.capture.excludeAgents.join(", ")}`);
|
||||
api.logger.debug?.(`${TAG} Agent exclude patterns: ${cfg.capture.excludeAgents.join(", ")}`);
|
||||
}
|
||||
|
||||
// Daily local JSONL cleaner (L0/L1), enabled only when retentionDays is configured.
|
||||
@@ -205,13 +197,13 @@ export default function register(api: OpenClawPluginApi) {
|
||||
logger: api.logger,
|
||||
});
|
||||
sharedMemoryCleaner.start();
|
||||
api.logger.info(`${TAG} Memory cleaner started (singleton)`);
|
||||
api.logger.debug?.(`${TAG} Memory cleaner started (singleton)`);
|
||||
} else {
|
||||
api.logger.info(`${TAG} Memory cleaner already started in this process, reusing existing instance`);
|
||||
api.logger.debug?.(`${TAG} Memory cleaner already started in this process, reusing existing instance`);
|
||||
}
|
||||
memoryCleaner = sharedMemoryCleaner;
|
||||
} else {
|
||||
api.logger.info(`${TAG} Memory cleaner disabled (retentionDays not configured)`);
|
||||
api.logger.debug?.(`${TAG} Memory cleaner disabled (retentionDays not configured)`);
|
||||
}
|
||||
|
||||
// Hardcoded actor ID (legacy, to be removed)
|
||||
@@ -228,7 +220,7 @@ export default function register(api: OpenClawPluginApi) {
|
||||
// ============================
|
||||
|
||||
// Shared references for tools (populated when extraction scheduler creates them)
|
||||
let sharedVectorStore: VectorStore | undefined;
|
||||
let sharedVectorStore: IMemoryStore | undefined;
|
||||
let sharedEmbeddingService: EmbeddingService | undefined;
|
||||
|
||||
/**
|
||||
@@ -272,12 +264,14 @@ export default function register(api: OpenClawPluginApi) {
|
||||
};
|
||||
|
||||
// tdai_memory_search — Agent-callable L1 memory search tool
|
||||
// TODO: implement hard per-turn call limit via before_tool_call hook + execute early-return (方案 D)
|
||||
api.registerTool(
|
||||
{
|
||||
name: "tdai_memory_search",
|
||||
label: "Memory Search",
|
||||
description:
|
||||
"Search through the user's long-term memories. Use this when you need to recall specific information about the user's preferences, past events, instructions, or context from previous conversations. Returns relevant memory records ranked by relevance.",
|
||||
"Search through the user's long-term memories. Use this when you need to recall specific information about the user's preferences, past events, instructions, or context from previous conversations. Returns relevant memory records ranked by relevance. " +
|
||||
"Limit: tdai_memory_search and tdai_conversation_search share a combined limit of 3 calls per turn. Stop searching after 3 total attempts.",
|
||||
parameters: {
|
||||
type: "object",
|
||||
properties: {
|
||||
@@ -367,6 +361,7 @@ export default function register(api: OpenClawPluginApi) {
|
||||
);
|
||||
|
||||
// tdai_conversation_search — Agent-callable L0 conversation search tool
|
||||
// TODO: implement hard per-turn call limit via before_tool_call hook + execute early-return (方案 D)
|
||||
api.registerTool(
|
||||
{
|
||||
name: "tdai_conversation_search",
|
||||
@@ -375,7 +370,8 @@ export default function register(api: OpenClawPluginApi) {
|
||||
"Search through past conversation history (raw dialogue records). " +
|
||||
"Use this when tdai_memory_search (structured memories) doesn't have the information you need, " +
|
||||
"or when you want to find specific past conversations, dialogue context, or exact words " +
|
||||
"the user said before. Returns relevant individual messages ranked by relevance.",
|
||||
"the user said before. Returns relevant individual messages ranked by relevance. " +
|
||||
"Limit: tdai_memory_search and tdai_conversation_search share a combined limit of 3 calls per turn. Stop searching after 3 total attempts.",
|
||||
parameters: {
|
||||
type: "object",
|
||||
properties: {
|
||||
@@ -464,7 +460,7 @@ export default function register(api: OpenClawPluginApi) {
|
||||
// (migrated from legacy before_agent_start to before_prompt_build so that
|
||||
// event.messages is guaranteed to be available — session is already loaded)
|
||||
if (cfg.recall.enabled) {
|
||||
api.logger.info(`${TAG} Registering before_prompt_build hook (auto-recall)`);
|
||||
api.logger.debug?.(`${TAG} Registering before_prompt_build hook (auto-recall)`);
|
||||
api.on("before_prompt_build", async (event, ctx) => {
|
||||
const startMs = Date.now();
|
||||
api.logger.debug?.(`${TAG} [before_prompt_build] Hook triggered`);
|
||||
@@ -619,343 +615,128 @@ export default function register(api: OpenClawPluginApi) {
|
||||
scheduler.start({});
|
||||
}
|
||||
|
||||
// Pre-warm the embedded agent import so the first extraction run doesn't
|
||||
// pay the cold-start cost (~35s jiti compile → <50ms with dist/ path).
|
||||
prewarmEmbeddedAgent(api.logger);
|
||||
// Pre-warm the embedded agent entrypoint. When runtime already exposes
|
||||
// runEmbeddedPiAgent this becomes a no-op; otherwise it still preloads
|
||||
// the legacy dist bridge to reduce first-run cold start.
|
||||
prewarmEmbeddedAgent(api.logger, api.runtime.agent);
|
||||
};
|
||||
|
||||
if (cfg.extraction.enabled) {
|
||||
// === Initialize VectorStore (always) + EmbeddingService (only when embedding enabled) ===
|
||||
let vectorStore: VectorStore | undefined;
|
||||
// === Store + scheduler initialization (async, runs eagerly) ===
|
||||
// Wrapped in an async IIFE because register() is synchronous.
|
||||
// initStores() is once-async: the first call creates the store,
|
||||
// subsequent calls (e.g. from seed CLI) reuse the cached result.
|
||||
let vectorStore: IMemoryStore | undefined;
|
||||
let embeddingService: EmbeddingService | undefined;
|
||||
|
||||
// VectorStore is always created as the metadata store for L0/L1 records.
|
||||
// It works as a pure SQLite store even without embedding — keyword search,
|
||||
// L0/L1 reads, and pipeline queries all use structured SQL, not vectors.
|
||||
try {
|
||||
const dims = cfg.embedding.dimensions; // 0 when provider="none" → vec0 tables deferred
|
||||
const dbPath = path.join(pluginDataDir, "vectors.db");
|
||||
vectorStore = new VectorStore(dbPath, dims, api.logger);
|
||||
const storeReady = (async () => {
|
||||
const stores = await initStores(cfg, pluginDataDir, api.logger);
|
||||
vectorStore = stores.vectorStore;
|
||||
embeddingService = stores.embeddingService;
|
||||
|
||||
// Create EmbeddingService only when embedding is enabled (remote provider configured)
|
||||
if (cfg.embedding.enabled) {
|
||||
// Share with tools immediately
|
||||
sharedVectorStore = vectorStore;
|
||||
sharedEmbeddingService = embeddingService;
|
||||
|
||||
// Keep cleaner's SQLite handle updated (singleton cleaner may start earlier).
|
||||
memoryCleaner?.setVectorStore(vectorStore);
|
||||
|
||||
if (vectorStore?.pullProfiles) {
|
||||
try {
|
||||
if (cfg.embedding.provider !== "local" && cfg.embedding.apiKey) {
|
||||
// Remote embedding provider (OpenAI-compatible API: OpenAI, Azure, self-hosted, etc.)
|
||||
embeddingService = createEmbeddingService({
|
||||
provider: cfg.embedding.provider,
|
||||
baseUrl: cfg.embedding.baseUrl,
|
||||
apiKey: cfg.embedding.apiKey,
|
||||
model: cfg.embedding.model,
|
||||
dimensions: cfg.embedding.dimensions,
|
||||
proxyUrl: cfg.embedding.proxyUrl,
|
||||
maxInputChars: cfg.embedding.maxInputChars,
|
||||
timeoutMs: cfg.embedding.timeoutMs,
|
||||
}, api.logger);
|
||||
} else {
|
||||
// Local provider (node-llama-cpp) — preserved internally but not reachable from user config
|
||||
embeddingService = createEmbeddingService({
|
||||
provider: "local",
|
||||
modelPath: cfg.embedding.model || undefined,
|
||||
modelCacheDir: cfg.embedding.modelCacheDir,
|
||||
}, api.logger);
|
||||
}
|
||||
await ensureL2L3Local(pluginDataDir, vectorStore, api.logger);
|
||||
} catch (err) {
|
||||
api.logger.warn(
|
||||
`${TAG} EmbeddingService init failed, continuing with keyword-only mode: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
embeddingService = undefined;
|
||||
api.logger.warn(`${TAG} Startup L2/L3 pull failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
} else {
|
||||
api.logger.info(`${TAG} Embedding disabled by config, VectorStore will serve as metadata-only store`);
|
||||
}
|
||||
|
||||
// Init VectorStore with provider info (undefined when no embedding → skips provider change detection)
|
||||
const providerInfo = embeddingService?.getProviderInfo();
|
||||
const initResult = vectorStore.init(providerInfo);
|
||||
|
||||
// If VectorStore entered degraded mode (e.g. sqlite-vec load failed),
|
||||
// treat it as unavailable and fall back to keyword-only mode.
|
||||
if (vectorStore.isDegraded()) {
|
||||
api.logger.warn(
|
||||
`${TAG} VectorStore is in degraded mode, falling back to keyword dedup`,
|
||||
);
|
||||
vectorStore = undefined;
|
||||
embeddingService = undefined;
|
||||
} else {
|
||||
// If embedding provider/model/dimensions changed, re-embed all existing texts
|
||||
if (stores.needsReindex && embeddingService && vectorStore) {
|
||||
const svc = embeddingService;
|
||||
const vs = vectorStore;
|
||||
api.logger.info(
|
||||
`${TAG} VectorStore initialized: ${dbPath} (${dims}D, provider=${cfg.embedding.provider})`,
|
||||
`${TAG} Embedding config changed (${stores.reindexReason}). ` +
|
||||
`Starting background re-embed of all stored texts...`,
|
||||
);
|
||||
|
||||
// If embedding provider/model/dimensions changed, re-embed all existing texts
|
||||
if (initResult.needsReindex && embeddingService) {
|
||||
const svc = embeddingService; // capture for async closure
|
||||
const vs = vectorStore; // capture for async closure
|
||||
vs.reindexAll(
|
||||
(text) => svc.embed(text),
|
||||
(done, total, layer) => {
|
||||
if (done === total || done % 50 === 0) {
|
||||
api.logger.debug?.(`${TAG} Re-embed progress: ${layer} ${done}/${total}`);
|
||||
}
|
||||
},
|
||||
).then(({ l1Count, l0Count }) => {
|
||||
api.logger.info(
|
||||
`${TAG} Embedding config changed (${initResult.reason}). ` +
|
||||
`Starting background re-embed of all stored texts...`,
|
||||
`${TAG} Re-embed complete: L1=${l1Count} records, L0=${l0Count} messages`,
|
||||
);
|
||||
// Run re-embed asynchronously so it doesn't block plugin startup
|
||||
vs.reindexAll(
|
||||
(text) => svc.embed(text),
|
||||
(done, total, layer) => {
|
||||
if (done === total || done % 50 === 0) {
|
||||
api.logger.debug?.(`${TAG} Re-embed progress: ${layer} ${done}/${total}`);
|
||||
}
|
||||
},
|
||||
).then(({ l1Count, l0Count }) => {
|
||||
api.logger.info(
|
||||
`${TAG} Re-embed complete: L1=${l1Count} records, L0=${l0Count} messages`,
|
||||
);
|
||||
}).catch((err) => {
|
||||
api.logger.error(
|
||||
`${TAG} Re-embed failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
}).catch((err) => {
|
||||
api.logger.error(
|
||||
`${TAG} Re-embed failed (non-fatal): ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
api.logger.warn(
|
||||
`${TAG} VectorStore init failed; vector/FTS recall and dedup conflict detection will be unavailable: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
vectorStore = undefined;
|
||||
embeddingService = undefined;
|
||||
}
|
||||
})();
|
||||
|
||||
// Share vectorStore/embeddingService with tdai_memory_search tool
|
||||
sharedVectorStore = vectorStore;
|
||||
sharedEmbeddingService = embeddingService;
|
||||
// === Create pipeline manager (sync — does not need store) ===
|
||||
scheduler = createPipelineManager(cfg, api.logger, sessionFilter);
|
||||
|
||||
// Keep cleaner's SQLite handle updated (singleton cleaner may start earlier).
|
||||
memoryCleaner?.setVectorStore(vectorStore);
|
||||
// Wire runners after store is ready
|
||||
storeReady.then(() => {
|
||||
// L1 runner via shared factory
|
||||
scheduler!.setL1Runner(createL1Runner({
|
||||
pluginDataDir,
|
||||
cfg,
|
||||
openclawConfig: api.config,
|
||||
vectorStore,
|
||||
embeddingService,
|
||||
logger: api.logger,
|
||||
getInstanceId: () => instanceId,
|
||||
}));
|
||||
|
||||
// === Create pipeline manager ===
|
||||
scheduler = new MemoryPipelineManager(
|
||||
{
|
||||
everyNConversations: cfg.pipeline.everyNConversations,
|
||||
enableWarmup: cfg.pipeline.enableWarmup,
|
||||
l1: { idleTimeoutSeconds: cfg.pipeline.l1IdleTimeoutSeconds },
|
||||
l2: {
|
||||
delayAfterL1Seconds: cfg.pipeline.l2DelayAfterL1Seconds,
|
||||
minIntervalSeconds: cfg.pipeline.l2MinIntervalSeconds,
|
||||
maxIntervalSeconds: cfg.pipeline.l2MaxIntervalSeconds,
|
||||
sessionActiveWindowHours: cfg.pipeline.sessionActiveWindowHours,
|
||||
},
|
||||
},
|
||||
api.logger,
|
||||
sessionFilter,
|
||||
);
|
||||
// Persister via shared factory
|
||||
scheduler!.setPersister(createPersister(pluginDataDir, api.logger));
|
||||
|
||||
// L1 runner: read L0 from DB (primary) or JSONL (fallback) → local LLM extraction → L1 JSONL + VectorStore
|
||||
scheduler.setL1Runner(async ({ sessionKey }) => {
|
||||
// L1 reads L0 data from VectorStore DB (primary, indexed query).
|
||||
// Fallback: read from L0 JSONL files when VectorStore is unavailable.
|
||||
if (!api.config) {
|
||||
api.logger.debug?.(`${TAG} [pipeline-l1] No OpenClaw config, skipping L1 extraction`);
|
||||
return { processedCount: 0 };
|
||||
}
|
||||
|
||||
const checkpoint = new CheckpointManager(pluginDataDir, api.logger);
|
||||
const cp = await checkpoint.read();
|
||||
const runnerState = checkpoint.getRunnerState(cp, sessionKey);
|
||||
|
||||
api.logger.info(
|
||||
`${TAG} [pipeline-l1] Session ${sessionKey}: ` +
|
||||
`l1_cursor=${runnerState.last_l1_cursor || "(start)"}`,
|
||||
);
|
||||
|
||||
try {
|
||||
// Read L0 messages since last L1 cursor, grouped by sessionId.
|
||||
// Within the same sessionKey, different sessionIds represent different
|
||||
// conversation instances (e.g. after /reset). Each group is extracted
|
||||
// independently so its sessionId is correctly associated with L1 records.
|
||||
//
|
||||
// Primary path: read from VectorStore DB (indexed query, fast).
|
||||
// Fallback: read from L0 JSONL files (scan + parse, slower).
|
||||
let groups: Array<{ sessionId: string; messages: ConversationMessage[] }>;
|
||||
|
||||
if (vectorStore && !vectorStore.isDegraded()) {
|
||||
// DB path: fast indexed query
|
||||
// NOTE: When last_l1_cursor is 0 (first L1 run), we pass undefined
|
||||
// to query all messages — but only those captured AFTER plugin start
|
||||
// (L0 capture uses pluginStartTimestamp as floor, so DB won't contain
|
||||
// pre-existing messages). This is safe because auto-capture already
|
||||
// filters out messages older than pluginStartTimestamp.
|
||||
const l1Cursor = runnerState.last_l1_cursor > 0
|
||||
? runnerState.last_l1_cursor
|
||||
: undefined;
|
||||
const dbGroups = vectorStore.queryL0GroupedBySessionId(
|
||||
sessionKey,
|
||||
l1Cursor,
|
||||
);
|
||||
// Cast role from string to "user" | "assistant" (DB stores as string)
|
||||
groups = dbGroups.map((g) => ({
|
||||
sessionId: g.sessionId,
|
||||
messages: g.messages.map((m) => ({
|
||||
id: m.id,
|
||||
role: m.role as "user" | "assistant",
|
||||
content: m.content,
|
||||
timestamp: m.timestamp,
|
||||
})),
|
||||
}));
|
||||
api.logger.debug?.(
|
||||
`${TAG} [pipeline-l1] L0 data source: VectorStore DB`,
|
||||
);
|
||||
} else {
|
||||
// Fallback: JSONL files
|
||||
api.logger.debug?.(
|
||||
`${TAG} [pipeline-l1] L0 data source: JSONL files (VectorStore unavailable)`,
|
||||
);
|
||||
const jsonlGroups = await readConversationMessagesGroupedBySessionId(
|
||||
sessionKey,
|
||||
// L2 runner: read L1 records (incremental) → SceneExtractor
|
||||
scheduler!.setL2Runner(async (sessionKey: string, cursor?: string) => {
|
||||
try {
|
||||
const l2Runner = createL2Runner({
|
||||
pluginDataDir,
|
||||
runnerState.last_l1_cursor || undefined,
|
||||
api.logger,
|
||||
50, // Match DB path limit (queryL0ForL1 default)
|
||||
);
|
||||
// Convert SessionIdMessageGroup[] to the same shape
|
||||
groups = jsonlGroups.map((g) => ({
|
||||
sessionId: g.sessionId,
|
||||
messages: g.messages,
|
||||
}));
|
||||
}
|
||||
|
||||
if (groups.length === 0) {
|
||||
api.logger.debug?.(`${TAG} [pipeline-l1] No new L0 messages for session ${sessionKey}`);
|
||||
return { processedCount: 0 };
|
||||
}
|
||||
|
||||
const totalMessages = groups.reduce((sum, g) => sum + g.messages.length, 0);
|
||||
api.logger.info(
|
||||
`${TAG} [pipeline-l1] Processing ${totalMessages} L0 messages across ${groups.length} sessionId group(s) for session ${sessionKey}`,
|
||||
);
|
||||
|
||||
let totalExtracted = 0;
|
||||
let totalStored = 0;
|
||||
let lastSceneName: string | undefined;
|
||||
let maxTimestamp = 0;
|
||||
|
||||
for (const group of groups) {
|
||||
api.logger.debug?.(
|
||||
`${TAG} [pipeline-l1] Group sessionId=${group.sessionId || "(empty)"}: ${group.messages.length} messages`,
|
||||
);
|
||||
|
||||
const l1Result = await extractL1Memories({
|
||||
messages: group.messages,
|
||||
sessionKey,
|
||||
sessionId: group.sessionId,
|
||||
baseDir: pluginDataDir,
|
||||
config: api.config,
|
||||
options: {
|
||||
enableDedup: cfg.extraction.enableDedup,
|
||||
maxMemoriesPerSession: cfg.extraction.maxMemoriesPerSession,
|
||||
model: cfg.extraction.model,
|
||||
previousSceneName: lastSceneName ?? (runnerState.last_scene_name || undefined),
|
||||
vectorStore,
|
||||
embeddingService,
|
||||
conflictRecallTopK: cfg.embedding.conflictRecallTopK,
|
||||
},
|
||||
cfg,
|
||||
openclawConfig: api.config,
|
||||
vectorStore,
|
||||
logger: api.logger,
|
||||
instanceId,
|
||||
});
|
||||
|
||||
totalExtracted += l1Result.extractedCount;
|
||||
totalStored += l1Result.storedCount;
|
||||
if (l1Result.lastSceneName) {
|
||||
lastSceneName = l1Result.lastSceneName;
|
||||
}
|
||||
|
||||
const groupMaxTs = Math.max(...group.messages.map((m) => m.timestamp));
|
||||
maxTimestamp = Math.max(maxTimestamp, groupMaxTs);
|
||||
return await l2Runner(sessionKey, cursor);
|
||||
} catch (err) {
|
||||
api.logger.error(`${TAG} [pipeline-l2] L2 failed: ${err instanceof Error ? err.stack ?? err.message : String(err)}`);
|
||||
throw err;
|
||||
}
|
||||
});
|
||||
|
||||
// Update checkpoint on disk — cursor is the global max timestamp across all groups
|
||||
await checkpoint.markL1ExtractionComplete(
|
||||
sessionKey,
|
||||
totalStored,
|
||||
maxTimestamp,
|
||||
lastSceneName,
|
||||
);
|
||||
|
||||
api.logger.info(
|
||||
`${TAG} [pipeline-l1] L1 complete: extracted=${totalExtracted}, stored=${totalStored} (${groups.length} group(s))`,
|
||||
);
|
||||
|
||||
return { processedCount: totalMessages };
|
||||
} catch (err) {
|
||||
api.logger.error(`${TAG} [pipeline-l1] L1 failed: ${err instanceof Error ? err.stack ?? err.message : String(err)}`);
|
||||
// ── error_degradation metric ──
|
||||
if (instanceId) {
|
||||
report("error_degradation", {
|
||||
module: "l1-extraction",
|
||||
action: "extractL1Memories",
|
||||
errorType: "exception",
|
||||
errorMessage: err instanceof Error ? err.message : String(err),
|
||||
degradedTo: null,
|
||||
impact: "blocking",
|
||||
// L3 runner: persona trigger + generation
|
||||
scheduler!.setL3Runner(async () => {
|
||||
try {
|
||||
const l3Runner = createL3Runner({
|
||||
pluginDataDir,
|
||||
cfg,
|
||||
openclawConfig: api.config,
|
||||
vectorStore,
|
||||
logger: api.logger,
|
||||
instanceId,
|
||||
});
|
||||
await l3Runner();
|
||||
} catch (err) {
|
||||
api.logger.error(`${TAG} [pipeline-l3] Failed: ${err instanceof Error ? err.stack ?? err.message : String(err)}`);
|
||||
}
|
||||
throw err; // rethrow so pipeline-manager can retry
|
||||
}
|
||||
});
|
||||
}).catch((err) => {
|
||||
api.logger.error(
|
||||
`${TAG} Store init failed; vector/FTS recall and dedup will be unavailable: ${err instanceof Error ? err.message : String(err)}`,
|
||||
);
|
||||
});
|
||||
|
||||
// Persister: saves pipeline session states to checkpoint
|
||||
scheduler.setPersister(async (states) => {
|
||||
const checkpoint = new CheckpointManager(pluginDataDir, api.logger);
|
||||
await checkpoint.mergePipelineStates(states);
|
||||
});
|
||||
|
||||
// L2 runner: read L1 records (incremental) → SceneExtractor
|
||||
scheduler.setL2Runner(async (sessionKey: string, cursor?: string) => {
|
||||
try {
|
||||
return await runLocalL2Extraction(api, cfg, pluginDataDir, sessionKey, vectorStore, cursor, instanceId);
|
||||
} catch (err) {
|
||||
api.logger.error(`${TAG} [pipeline-l2] L2 failed: ${err instanceof Error ? err.stack ?? err.message : String(err)}`);
|
||||
throw err; // rethrow so pipeline-manager can handle retry/fallback (consistent with L1 runner)
|
||||
}
|
||||
});
|
||||
|
||||
// L3 runner: persona trigger + generation
|
||||
scheduler.setL3Runner(async () => {
|
||||
try {
|
||||
const trigger = new PersonaTrigger({
|
||||
dataDir: pluginDataDir,
|
||||
interval: cfg.persona.triggerEveryN,
|
||||
logger: api.logger,
|
||||
});
|
||||
|
||||
const { should, reason } = await trigger.shouldGenerate();
|
||||
if (!should) {
|
||||
api.logger.debug?.(`${TAG} [pipeline-l3] Persona generation not needed`);
|
||||
return;
|
||||
}
|
||||
|
||||
if (!api.config) {
|
||||
api.logger.warn(`${TAG} [pipeline-l3] No OpenClaw config, skipping persona generation`);
|
||||
return;
|
||||
}
|
||||
|
||||
api.logger.info(`${TAG} [pipeline-l3] Starting persona generation: ${reason}`);
|
||||
const generator = new PersonaGenerator({
|
||||
dataDir: pluginDataDir,
|
||||
config: api.config,
|
||||
model: cfg.persona.model,
|
||||
backupCount: cfg.persona.backupCount,
|
||||
logger: api.logger,
|
||||
instanceId,
|
||||
});
|
||||
const genResult = await generator.generate(reason);
|
||||
api.logger.info(`${TAG} [pipeline-l3] Persona generation ${genResult ? "succeeded" : "skipped (no changes)"}`);
|
||||
} catch (err) {
|
||||
api.logger.error(`${TAG} [pipeline-l3] Failed: ${err instanceof Error ? err.stack ?? err.message : String(err)}`);
|
||||
}
|
||||
});
|
||||
|
||||
// Capture vectorStore reference for cleanup
|
||||
const vectorStoreRef = vectorStore;
|
||||
|
||||
// Register a SINGLE gateway_stop hook for ordered shutdown.
|
||||
// Order: memoryCleaner → scheduler → vectorStore → embeddingService
|
||||
// Order: memoryCleaner → scheduler → vectorStore → embeddingService → resetStores
|
||||
// (memoryCleaner may use VectorStore during cleanup, so it must stop first)
|
||||
//
|
||||
// The entire hook is wrapped with a 3 s timeout to guarantee we never
|
||||
@@ -965,8 +746,11 @@ export default function register(api: OpenClawPluginApi) {
|
||||
const GATEWAY_STOP_TIMEOUT_MS = 3_000;
|
||||
const hookStartMs = Date.now();
|
||||
|
||||
// Ensure store init has completed before tearing down
|
||||
await storeReady.catch(() => {});
|
||||
|
||||
const doCleanup = async (): Promise<void> => {
|
||||
// 1. Stop the memory cleaner
|
||||
// 1. Stop the memory cleaner first (it may be running deleteL1ExpiredByUpdatedTime)
|
||||
if (memoryCleaner) {
|
||||
try {
|
||||
memoryCleaner.destroy();
|
||||
@@ -983,16 +767,20 @@ export default function register(api: OpenClawPluginApi) {
|
||||
const t = Date.now();
|
||||
await scheduler.destroy();
|
||||
api.logger.info(`${TAG} [gateway_stop] Scheduler destroyed (${Date.now() - t}ms)`);
|
||||
} else {
|
||||
api.logger.info(`${TAG} [gateway_stop] Scheduler was never started, skipping destroy`);
|
||||
}
|
||||
|
||||
// 3. Close VectorStore
|
||||
if (vectorStoreRef) {
|
||||
vectorStoreRef.close();
|
||||
// 3. Close VectorStore last (after all consumers are done)
|
||||
if (vectorStore) {
|
||||
api.logger.info(`${TAG} [gateway_stop] Closing VectorStore`);
|
||||
vectorStore.close();
|
||||
}
|
||||
|
||||
// 4. Release embedding service resources
|
||||
// 4. Release embedding service resources (model memory, GPU, etc.)
|
||||
if (embeddingService?.close) {
|
||||
try {
|
||||
api.logger.info(`${TAG} [gateway_stop] Closing EmbeddingService`);
|
||||
await embeddingService.close();
|
||||
} catch (err) {
|
||||
api.logger.warn(`${TAG} [gateway_stop] EmbeddingService close error: ${err instanceof Error ? err.message : String(err)}`);
|
||||
@@ -1021,11 +809,14 @@ export default function register(api: OpenClawPluginApi) {
|
||||
if (timeoutId !== undefined) clearTimeout(timeoutId);
|
||||
}
|
||||
|
||||
// 5. Reset store singleton cache so hot-restart can re-initialize
|
||||
resetStores();
|
||||
|
||||
api.logger.info(`${TAG} [gateway_stop] Cleanup finished, all resources released (${Date.now() - hookStartMs}ms)`);
|
||||
});
|
||||
}
|
||||
|
||||
api.logger.info(`${TAG} Registering agent_end hook (auto-capture)`);
|
||||
api.logger.debug?.(`${TAG} Registering agent_end hook (auto-capture)`);
|
||||
api.on("agent_end", async (event, ctx) => {
|
||||
const startMs = Date.now();
|
||||
api.logger.debug?.(`${TAG} [agent_end] Hook triggered`);
|
||||
@@ -1139,7 +930,7 @@ export default function register(api: OpenClawPluginApi) {
|
||||
}
|
||||
});
|
||||
} else {
|
||||
api.logger.info(`${TAG} Auto-capture disabled`);
|
||||
api.logger.debug?.(`${TAG} Auto-capture disabled`);
|
||||
}
|
||||
|
||||
// memoryCleaner gateway_stop is handled in the unified handler above (inside extraction.enabled block).
|
||||
@@ -1159,177 +950,30 @@ export default function register(api: OpenClawPluginApi) {
|
||||
});
|
||||
}
|
||||
|
||||
api.logger.info(
|
||||
// ============================
|
||||
// CLI registration
|
||||
// ============================
|
||||
|
||||
api.registerCli(
|
||||
({ program, config, logger: cliLogger }) => {
|
||||
const memoryTdai = program
|
||||
.command("memory-tdai")
|
||||
.description("memory-tdai plugin commands (seed, query, stats)");
|
||||
|
||||
registerMemoryTdaiCli(memoryTdai, {
|
||||
config,
|
||||
pluginConfig: api.pluginConfig,
|
||||
stateDir: api.runtime.state.resolveStateDir(),
|
||||
logger: cliLogger,
|
||||
});
|
||||
},
|
||||
{ commands: ["memory-tdai"] },
|
||||
);
|
||||
|
||||
api.logger.debug?.(
|
||||
`${TAG} Plugin registration complete (v3). ` +
|
||||
`startTimestamp=${pluginStartTimestamp} (${new Date(pluginStartTimestamp).toISOString()})`,
|
||||
);
|
||||
|
||||
// Resolve persistent instance ID for metric reporting (used by agent_turn, error_degradation, etc.)
|
||||
getOrCreateInstanceId(pluginDataDir).then((id) => {
|
||||
instanceId = id;
|
||||
initReporter({ enabled: cfg.report.enabled, type: cfg.report.type, logger: api.logger, instanceId: id, pluginVersion });
|
||||
}).catch((err) => {
|
||||
api.logger.warn(`${TAG} Failed to initialize instanceId for metrics: ${err instanceof Error ? err.message : String(err)}`);
|
||||
});
|
||||
}
|
||||
|
||||
// ============================
|
||||
// L2 extraction implementations
|
||||
// ============================
|
||||
|
||||
/**
|
||||
* Local L2 extraction: read L1 records → SceneExtractor.
|
||||
*
|
||||
* Uses **incremental** reads when VectorStore is available:
|
||||
* 1. Receive the pipeline cursor (`last_extraction_updated_time`) from pipeline-manager
|
||||
* 2. Query only L1 records updated AFTER that cursor via `queryMemoryRecords`
|
||||
* 3. Return the latest `updatedAt` from the batch so pipeline-manager can advance the cursor
|
||||
*
|
||||
* Falls back to JSONL read (with client-side time filtering) when VectorStore is unavailable.
|
||||
*/
|
||||
async function runLocalL2Extraction(
|
||||
api: OpenClawPluginApi,
|
||||
cfg: MemoryTdaiConfig,
|
||||
pluginDataDir: string,
|
||||
sessionKey: string,
|
||||
vectorStore?: VectorStore,
|
||||
updatedAfter?: string,
|
||||
instanceId?: string,
|
||||
): Promise<{ latestCursor?: string } | void> {
|
||||
api.logger.debug?.(
|
||||
`${TAG} [L2-local] session=${sessionKey}, updatedAfter=${updatedAfter ?? "(full)"}`,
|
||||
);
|
||||
|
||||
let records: Array<{ content: string; created_at: string; id: string; updatedAt: string }>;
|
||||
|
||||
// Prefer incremental SQLite query when VectorStore is available
|
||||
if (vectorStore && !vectorStore.isDegraded()) {
|
||||
const { queryMemoryRecords } = await import("./src/record/l1-reader.js");
|
||||
const memRecords = queryMemoryRecords(vectorStore, {
|
||||
sessionKey,
|
||||
updatedAfter,
|
||||
}, api.logger);
|
||||
|
||||
if (memRecords.length === 0) {
|
||||
api.logger.debug?.(
|
||||
`${TAG} [L2-local] No new L1 records since cursor (session=${sessionKey}, updatedAfter=${updatedAfter ?? "(full)"}), skipping scene extraction`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
api.logger.debug?.(
|
||||
`${TAG} [L2-local] Incremental query returned ${memRecords.length} record(s) (session=${sessionKey})`,
|
||||
);
|
||||
|
||||
records = memRecords.map((r) => ({
|
||||
content: r.content,
|
||||
created_at: r.createdAt,
|
||||
id: r.id,
|
||||
updatedAt: r.updatedAt,
|
||||
}));
|
||||
} else {
|
||||
// Fallback: read JSONL files with client-side time filtering
|
||||
api.logger.debug?.(`${TAG} [L2-local] VectorStore unavailable, falling back to JSONL read`);
|
||||
const { readAllMemoryRecords } = await import("./src/record/l1-reader.js");
|
||||
let allRecords = await readAllMemoryRecords(pluginDataDir, api.logger);
|
||||
|
||||
// Apply updatedAfter filter on JSONL records (same semantics as SQLite path)
|
||||
if (updatedAfter) {
|
||||
const beforeCount = allRecords.length;
|
||||
allRecords = allRecords.filter((r) => {
|
||||
const t = r.updatedAt || r.createdAt || "";
|
||||
return t > updatedAfter;
|
||||
});
|
||||
api.logger.debug?.(
|
||||
`${TAG} [L2-local] JSONL time filter: ${beforeCount} → ${allRecords.length} record(s) (updatedAfter=${updatedAfter})`,
|
||||
);
|
||||
}
|
||||
|
||||
if (allRecords.length === 0) {
|
||||
api.logger.debug?.(`${TAG} [L2-local] No new L1 records found (JSONL fallback), skipping scene extraction`);
|
||||
return;
|
||||
}
|
||||
|
||||
records = allRecords.map((r) => ({
|
||||
content: r.content,
|
||||
created_at: r.createdAt,
|
||||
id: r.id,
|
||||
updatedAt: r.updatedAt,
|
||||
}));
|
||||
}
|
||||
|
||||
const extractor = new SceneExtractor({
|
||||
dataDir: pluginDataDir,
|
||||
config: api.config!,
|
||||
model: cfg.persona.model,
|
||||
maxScenes: cfg.persona.maxScenes,
|
||||
sceneBackupCount: cfg.persona.sceneBackupCount,
|
||||
logger: api.logger,
|
||||
instanceId,
|
||||
});
|
||||
|
||||
const memories = records.map((r) => ({
|
||||
content: r.content,
|
||||
created_at: r.created_at,
|
||||
id: r.id,
|
||||
}));
|
||||
|
||||
// ── Checkpoint guard ──────────────────────────────────────────────
|
||||
// Snapshot critical counters BEFORE the LLM agent runs.
|
||||
// The LLM operates on checkpoint via raw file tools (write_to_file /
|
||||
// replace_in_file) which bypass CheckpointManager's file lock.
|
||||
// If the LLM accidentally overwrites the entire checkpoint (e.g. via
|
||||
// write_to_file), system-managed counters like scenes_processed and
|
||||
// memories_since_last_persona can be reset to stale values.
|
||||
// After extraction we detect and repair such corruption.
|
||||
const preCheckpoint = new CheckpointManager(pluginDataDir, api.logger);
|
||||
const preState = await preCheckpoint.read();
|
||||
const preScenesProcessed = preState.scenes_processed;
|
||||
const preMemoriesSince = preState.memories_since_last_persona;
|
||||
const preTotalProcessed = preState.total_processed;
|
||||
|
||||
const extractResult = await extractor.extract(memories);
|
||||
if (extractResult.success && extractResult.memoriesProcessed > 0) {
|
||||
const checkpoint = new CheckpointManager(pluginDataDir, api.logger);
|
||||
|
||||
// Detect and repair LLM-caused checkpoint corruption.
|
||||
// If the LLM wrote the entire checkpoint file (instead of using
|
||||
// replace_in_file on specific fields), system-managed counters may
|
||||
// have been overwritten with stale/zero values.
|
||||
const postState = await checkpoint.read();
|
||||
if (
|
||||
postState.scenes_processed < preScenesProcessed ||
|
||||
postState.total_processed < preTotalProcessed
|
||||
) {
|
||||
api.logger.warn(
|
||||
`${TAG} [L2-local] ⚠️ Checkpoint corruption detected! ` +
|
||||
`scenes_processed: ${preScenesProcessed} → ${postState.scenes_processed}, ` +
|
||||
`total_processed: ${preTotalProcessed} → ${postState.total_processed}, ` +
|
||||
`memories_since: ${preMemoriesSince} → ${postState.memories_since_last_persona}. ` +
|
||||
`Repairing...`,
|
||||
);
|
||||
await checkpoint.write({
|
||||
...postState,
|
||||
scenes_processed: Math.max(postState.scenes_processed, preScenesProcessed),
|
||||
total_processed: Math.max(postState.total_processed, preTotalProcessed),
|
||||
memories_since_last_persona: Math.max(postState.memories_since_last_persona, preMemoriesSince),
|
||||
});
|
||||
api.logger.info(`${TAG} [L2-local] Checkpoint repaired`);
|
||||
}
|
||||
|
||||
await checkpoint.incrementScenesProcessed();
|
||||
|
||||
// Return the max updatedAt from this batch as the new cursor
|
||||
const latestCursor = records.reduce((latest, r) => {
|
||||
return r.updatedAt > latest ? r.updatedAt : latest;
|
||||
}, "");
|
||||
|
||||
api.logger.debug?.(
|
||||
`${TAG} [L2-local] Extraction complete: processed=${extractResult.memoriesProcessed}, latestCursor=${latestCursor}`,
|
||||
);
|
||||
|
||||
return { latestCursor: latestCursor || undefined };
|
||||
}
|
||||
}
|
||||
|
||||
// ============================
|
||||
|
||||
Reference in New Issue
Block a user