mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
refactor(memory): split QMD manager by domain (#113659)
* refactor(memory): split QMD manager by domain * refactor(memory): keep QMD base types private
This commit is contained in:
Peter Steinberger
committed by
GitHub
GitHub
parent
71e8f6a925
commit
9a7a5791c7
@@ -169,7 +169,6 @@ extensions/memory-core/src/memory/manager-search.test.ts
|
||||
extensions/memory-core/src/memory/manager-search.ts
|
||||
extensions/memory-core/src/memory/manager.ts
|
||||
extensions/memory-core/src/memory/qmd-manager.test.ts
|
||||
extensions/memory-core/src/memory/qmd-manager.ts
|
||||
extensions/memory-core/src/memory/search-manager.test.ts
|
||||
extensions/memory-core/src/memory/search-manager.ts
|
||||
extensions/memory-core/src/rem-evidence.ts
|
||||
|
||||
@@ -0,0 +1,493 @@
|
||||
import crypto from "node:crypto";
|
||||
import fs from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import type { FSWatcher } from "chokidar";
|
||||
import {
|
||||
createSubsystemLogger,
|
||||
resolveAgentContextLimits,
|
||||
resolveMemorySearchSyncConfig,
|
||||
resolveAgentWorkspaceDir,
|
||||
resolveStateDir,
|
||||
type OpenClawConfig,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
|
||||
import type {
|
||||
MemorySource,
|
||||
ResolvedMemoryBackendConfig,
|
||||
ResolvedQmdConfig,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
|
||||
import type {
|
||||
PluginStateLeaseContext,
|
||||
PluginStateLeaseRunner,
|
||||
} from "openclaw/plugin-sdk/plugin-state-runtime";
|
||||
import {
|
||||
QmdCollectionController,
|
||||
type ManagedQmdCollection as ManagedCollection,
|
||||
type QmdSearchRuntimeDebugContext,
|
||||
} from "./qmd-collection-controller.js";
|
||||
import { QmdCommandClient } from "./qmd-command-client.js";
|
||||
import { asQmdAbortError } from "./qmd-command-errors.js";
|
||||
import {
|
||||
QmdDocumentResolver,
|
||||
type QmdCollectionRoot as CollectionRoot,
|
||||
} from "./qmd-document-resolver.js";
|
||||
import { buildQmdProcessPath, MAX_QMD_OUTPUT_CHARS } from "./qmd-manager-helpers.js";
|
||||
import type {
|
||||
QmdRuntimeCollectionValidationCacheContext,
|
||||
QmdRuntimeManagedCollection,
|
||||
QmdRuntimeMultiCollectionProbeCacheContext,
|
||||
} from "./qmd-runtime-cache.js";
|
||||
import { QmdSessionExporter, resolveQmdSessionExporterConfig } from "./qmd-session-exporter.js";
|
||||
import type { MemoryWatchPressureWarningState } from "./watch-pressure.js";
|
||||
import type { MemoryWatchSettleQueue } from "./watch-settle.js";
|
||||
|
||||
export const qmdManagerLog = createSubsystemLogger("memory");
|
||||
|
||||
type QmdManagerMode = "full" | "status" | "cli";
|
||||
type QmdManagerRuntimeConfig = {
|
||||
workspaceDir: string;
|
||||
syncSettings: ReturnType<typeof resolveMemorySearchSyncConfig>;
|
||||
contextLimits: ReturnType<typeof resolveAgentContextLimits>;
|
||||
};
|
||||
|
||||
type SqliteDatabase = import("node:sqlite").DatabaseSync;
|
||||
|
||||
export abstract class QmdManagerBase {
|
||||
protected readonly agentId: string;
|
||||
protected readonly qmd: ResolvedQmdConfig;
|
||||
protected readonly workspaceDir: string;
|
||||
protected readonly contextLimits: ReturnType<typeof resolveAgentContextLimits>;
|
||||
protected readonly stateDir: string;
|
||||
protected readonly agentStateDir: string;
|
||||
protected readonly qmdDir: string;
|
||||
protected readonly xdgConfigHome: string;
|
||||
protected readonly xdgCacheHome: string;
|
||||
protected readonly indexPath: string;
|
||||
protected readonly env: NodeJS.ProcessEnv;
|
||||
protected readonly commands: QmdCommandClient;
|
||||
protected readonly withLease: PluginStateLeaseRunner;
|
||||
protected readonly collectionController: QmdCollectionController;
|
||||
protected readonly documentResolver: QmdDocumentResolver;
|
||||
protected readonly syncSettings: ReturnType<typeof resolveMemorySearchSyncConfig>;
|
||||
protected readonly managedCollectionNames: string[];
|
||||
protected readonly collectionRoots = new Map<string, CollectionRoot>();
|
||||
protected readonly sources = new Set<MemorySource>();
|
||||
protected readonly maxQmdOutputChars = MAX_QMD_OUTPUT_CHARS;
|
||||
protected readonly sessionExporter: QmdSessionExporter | null;
|
||||
protected updateTimer: NodeJS.Timeout | null = null;
|
||||
protected embedTimer: NodeJS.Timeout | null = null;
|
||||
protected watcher: FSWatcher | null = null;
|
||||
protected watchTimer: NodeJS.Timeout | null = null;
|
||||
protected readonly pendingWatchPaths: MemoryWatchSettleQueue = new Map();
|
||||
protected readonly watchPressureWarning: MemoryWatchPressureWarningState = { shown: false };
|
||||
protected pendingUpdate: Promise<void> | null = null;
|
||||
protected queuedForcedUpdate: Promise<void> | null = null;
|
||||
protected queuedForcedRuns = 0;
|
||||
protected dirty = false;
|
||||
protected closed = false;
|
||||
protected mode: QmdManagerMode = "full";
|
||||
protected readonly closeSignal: Promise<void>;
|
||||
protected resolveCloseSignal!: () => void;
|
||||
protected readonly closeAbortController = new AbortController();
|
||||
protected qmdRuntimeIdentityPromise: Promise<string> | null = null;
|
||||
protected db: SqliteDatabase | null = null;
|
||||
protected lastUpdateAt: number | null = null;
|
||||
protected lastEmbedAt: number | null = null;
|
||||
protected embedLeaseRetryPending = false;
|
||||
protected embedBackoffUntil: number | null = null;
|
||||
protected embedFailureCount = 0;
|
||||
protected vectorAvailable: boolean | null = null;
|
||||
protected vectorStatusDetail: string | null = null;
|
||||
protected readonly sessionWarm = new Set<string>();
|
||||
protected multiCollectionFilterSupported: boolean | null = null;
|
||||
|
||||
protected constructor(params: {
|
||||
agentId: string;
|
||||
resolved: ResolvedQmdConfig;
|
||||
runtimeConfig: QmdManagerRuntimeConfig;
|
||||
withLease: PluginStateLeaseRunner;
|
||||
}) {
|
||||
this.agentId = params.agentId;
|
||||
this.qmd = params.resolved;
|
||||
this.workspaceDir = params.runtimeConfig.workspaceDir;
|
||||
this.contextLimits = params.runtimeConfig.contextLimits;
|
||||
this.withLease = params.withLease;
|
||||
this.stateDir = resolveStateDir(process.env, os.homedir);
|
||||
this.agentStateDir = path.join(this.stateDir, "agents", this.agentId);
|
||||
this.qmdDir = path.join(this.agentStateDir, "qmd");
|
||||
this.syncSettings = params.runtimeConfig.syncSettings;
|
||||
// QMD manages collections in its index DB while XDG config and cache dirs
|
||||
// isolate contexts and index.sqlite per agent.
|
||||
this.xdgConfigHome = path.join(this.qmdDir, "xdg-config");
|
||||
this.xdgCacheHome = path.join(this.qmdDir, "xdg-cache");
|
||||
this.indexPath = path.join(this.xdgCacheHome, "qmd", "index.sqlite");
|
||||
|
||||
this.env = {
|
||||
...process.env,
|
||||
PATH: buildQmdProcessPath(process.env.PATH),
|
||||
XDG_CONFIG_HOME: this.xdgConfigHome,
|
||||
// QMD resolves index.yml relative to QMD_CONFIG_DIR rather than XDG_CONFIG_HOME.
|
||||
// Point it at the nested qmd config directory so per-agent collections are visible.
|
||||
QMD_CONFIG_DIR: path.join(this.xdgConfigHome, "qmd"),
|
||||
XDG_CACHE_HOME: this.xdgCacheHome,
|
||||
NO_COLOR: "1",
|
||||
};
|
||||
this.commands = new QmdCommandClient(
|
||||
this.qmd,
|
||||
this.env,
|
||||
this.workspaceDir,
|
||||
this.maxQmdOutputChars,
|
||||
);
|
||||
this.collectionController = new QmdCollectionController(
|
||||
this.qmd,
|
||||
this.agentId,
|
||||
this.workspaceDir,
|
||||
this.xdgConfigHome,
|
||||
async (args, opts) => await this.commands.run(args, opts),
|
||||
async (signal) => await this.buildQmdCollectionValidationCacheContext(signal),
|
||||
);
|
||||
this.documentResolver = new QmdDocumentResolver(
|
||||
this.workspaceDir,
|
||||
this.collectionRoots,
|
||||
() => this.ensureDb(),
|
||||
this.qmd.sessions.readable,
|
||||
);
|
||||
this.closeSignal = new Promise<void>((resolve) => {
|
||||
this.resolveCloseSignal = resolve;
|
||||
});
|
||||
const sessionExporterConfig = resolveQmdSessionExporterConfig({
|
||||
qmd: this.qmd,
|
||||
agentId: this.agentId,
|
||||
qmdDir: this.qmdDir,
|
||||
});
|
||||
this.sessionExporter = sessionExporterConfig
|
||||
? new QmdSessionExporter(
|
||||
sessionExporterConfig,
|
||||
this.agentId,
|
||||
this.workspaceDir,
|
||||
this.indexPath,
|
||||
(collection, collectionRelativePath, workspaceRelativePath, absolutePath) =>
|
||||
this.documentResolver.buildSearchPath(
|
||||
collection,
|
||||
collectionRelativePath,
|
||||
workspaceRelativePath,
|
||||
absolutePath,
|
||||
),
|
||||
)
|
||||
: null;
|
||||
if (sessionExporterConfig) {
|
||||
this.qmd.collections = [
|
||||
...this.qmd.collections,
|
||||
{
|
||||
name: sessionExporterConfig.collectionName,
|
||||
path: sessionExporterConfig.dir,
|
||||
pattern: "**/*.md",
|
||||
kind: "sessions",
|
||||
},
|
||||
];
|
||||
}
|
||||
this.managedCollectionNames = this.computeManagedCollectionNames();
|
||||
}
|
||||
|
||||
protected async initialize(mode: QmdManagerMode): Promise<void> {
|
||||
this.mode = mode;
|
||||
const startTime = Date.now();
|
||||
this.bootstrapCollections();
|
||||
if (mode === "status") {
|
||||
return;
|
||||
}
|
||||
|
||||
await fs.mkdir(this.xdgConfigHome, { recursive: true });
|
||||
await fs.mkdir(this.xdgCacheHome, { recursive: true });
|
||||
await fs.mkdir(path.dirname(this.indexPath), { recursive: true });
|
||||
if (this.sessionExporter) {
|
||||
await fs.mkdir(this.sessionExporter.config.dir, { recursive: true });
|
||||
}
|
||||
|
||||
// Reuse default-cache ML models despite the per-agent XDG cache override
|
||||
// so qmd does not download them separately for every agent.
|
||||
await this.symlinkSharedModels();
|
||||
|
||||
await this.ensureCollections();
|
||||
if (mode === "cli") {
|
||||
if (this.qmd.update.onBoot && this.qmd.update.waitForBootSync) {
|
||||
await this.runUpdate("boot:cli", true).catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd cli boot update failed: ${String(err)}`);
|
||||
});
|
||||
}
|
||||
qmdManagerLog.info(
|
||||
`qmd manager initialized for agent "${this.agentId}" mode=cli collections=${this.qmd.collections.length} durationMs=${Date.now() - startTime}`,
|
||||
);
|
||||
return;
|
||||
}
|
||||
|
||||
this.ensureWatcher();
|
||||
qmdManagerLog.info(
|
||||
`qmd manager initialized for agent "${this.agentId}" mode=full collections=${this.qmd.collections.length} durationMs=${Date.now() - startTime}`,
|
||||
);
|
||||
|
||||
if (this.qmd.update.onBoot) {
|
||||
const bootRun = this.runUpdate("boot", true);
|
||||
if (this.qmd.update.waitForBootSync) {
|
||||
await bootRun.catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd boot update failed: ${String(err)}`);
|
||||
});
|
||||
} else {
|
||||
void bootRun.catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd boot update failed: ${String(err)}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
if (this.qmd.update.intervalMs > 0) {
|
||||
this.updateTimer = setInterval(() => {
|
||||
void this.runUpdate("interval").catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd update failed (${String(err)})`);
|
||||
});
|
||||
}, this.qmd.update.intervalMs);
|
||||
}
|
||||
if (this.shouldScheduleEmbedTimer()) {
|
||||
const startPeriodicEmbedTimer = () => {
|
||||
this.embedTimer = setInterval(() => {
|
||||
void this.runUpdate("embed-interval").catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd embed interval update failed (${String(err)})`);
|
||||
});
|
||||
}, this.qmd.update.embedIntervalMs);
|
||||
};
|
||||
const initialDelayMs = this.resolveEmbedStartupJitterMs();
|
||||
if (initialDelayMs > 0) {
|
||||
this.embedTimer = setTimeout(() => {
|
||||
this.embedTimer = null;
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
void this.runUpdate("embed-interval")
|
||||
.catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd embed interval update failed (${String(err)})`);
|
||||
})
|
||||
.finally(() => {
|
||||
if (!this.closed) {
|
||||
startPeriodicEmbedTimer();
|
||||
}
|
||||
});
|
||||
}, initialDelayMs);
|
||||
} else {
|
||||
startPeriodicEmbedTimer();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected bootstrapCollections(): void {
|
||||
this.collectionRoots.clear();
|
||||
this.sources.clear();
|
||||
for (const collection of this.qmd.collections) {
|
||||
const kind: MemorySource = collection.kind === "sessions" ? "sessions" : "memory";
|
||||
this.collectionRoots.set(collection.name, { path: collection.path, kind });
|
||||
this.sources.add(kind);
|
||||
}
|
||||
}
|
||||
|
||||
protected qmdRuntimeCacheSources(): string[] {
|
||||
return [...this.sources].toSorted();
|
||||
}
|
||||
|
||||
protected qmdRuntimeCacheCollections(): QmdRuntimeManagedCollection[] {
|
||||
return this.qmd.collections.map((collection) => ({
|
||||
name: collection.name,
|
||||
kind: collection.kind,
|
||||
path: collection.path,
|
||||
pattern: collection.pattern,
|
||||
}));
|
||||
}
|
||||
|
||||
protected buildQmdRuntimeEnvironmentHash(): string {
|
||||
const relevantEnv = Object.fromEntries(
|
||||
Object.keys(this.env)
|
||||
.filter(
|
||||
(key) =>
|
||||
key === "PATH" ||
|
||||
key === "HOME" ||
|
||||
key === "LOCALAPPDATA" ||
|
||||
key === "XDG_CONFIG_HOME" ||
|
||||
key === "XDG_CACHE_HOME" ||
|
||||
key === "QMD_CONFIG_DIR" ||
|
||||
key.startsWith("QMD_"),
|
||||
)
|
||||
.toSorted()
|
||||
.map((key) => [key, this.env[key] ?? ""]),
|
||||
);
|
||||
return crypto.createHash("sha256").update(JSON.stringify(relevantEnv)).digest("hex");
|
||||
}
|
||||
|
||||
protected async buildQmdCollectionValidationCacheContext(
|
||||
signal?: AbortSignal,
|
||||
): Promise<QmdRuntimeCollectionValidationCacheContext> {
|
||||
return {
|
||||
workspaceDir: this.workspaceDir,
|
||||
agentId: this.agentId,
|
||||
qmdCommand: this.qmd.command,
|
||||
qmdVersion: await this.resolveQmdRuntimeIdentity(signal),
|
||||
qmdEnvironmentHash: this.buildQmdRuntimeEnvironmentHash(),
|
||||
qmdIndexPath: this.indexPath,
|
||||
searchMode: this.qmd.searchMode,
|
||||
collections: this.qmdRuntimeCacheCollections(),
|
||||
sources: this.qmdRuntimeCacheSources(),
|
||||
};
|
||||
}
|
||||
|
||||
protected async buildQmdMultiCollectionProbeCacheContext(): Promise<QmdRuntimeMultiCollectionProbeCacheContext> {
|
||||
return {
|
||||
workspaceDir: this.workspaceDir,
|
||||
agentId: this.agentId,
|
||||
qmdCommand: this.qmd.command,
|
||||
qmdVersion: await this.resolveQmdRuntimeIdentity(),
|
||||
qmdEnvironmentHash: this.buildQmdRuntimeEnvironmentHash(),
|
||||
qmdIndexPath: this.indexPath,
|
||||
searchMode: this.qmd.searchMode,
|
||||
sources: this.qmdRuntimeCacheSources(),
|
||||
};
|
||||
}
|
||||
|
||||
protected resolveQmdRuntimeIdentity(signal?: AbortSignal): Promise<string> {
|
||||
if (signal) {
|
||||
return this.readQmdRuntimeIdentity(signal);
|
||||
}
|
||||
this.qmdRuntimeIdentityPromise ??= this.readQmdRuntimeIdentity();
|
||||
return this.qmdRuntimeIdentityPromise;
|
||||
}
|
||||
|
||||
protected async readQmdRuntimeIdentity(signal?: AbortSignal): Promise<string> {
|
||||
const commandIdentity = `command:${this.qmd.command}`;
|
||||
try {
|
||||
const result = await this.runQmd(["--version"], {
|
||||
timeoutMs: Math.min(this.qmd.limits.timeoutMs, 2_000),
|
||||
signal,
|
||||
});
|
||||
const versionText = `${result.stdout}\n${result.stderr}`.trim();
|
||||
return versionText ? `${commandIdentity};version:${versionText}` : commandIdentity;
|
||||
} catch {
|
||||
if (signal?.aborted) {
|
||||
throw asQmdAbortError(signal);
|
||||
}
|
||||
return commandIdentity;
|
||||
}
|
||||
}
|
||||
|
||||
protected async ensureCollections(options?: {
|
||||
force?: boolean;
|
||||
debugContext?: QmdSearchRuntimeDebugContext;
|
||||
parentSignal?: AbortSignal;
|
||||
}): Promise<void> {
|
||||
await this.withQmdStoreWriteLease(async (lease) => {
|
||||
await this.collectionController.ensureCollections({ ...options, lease });
|
||||
}, options?.parentSignal);
|
||||
}
|
||||
|
||||
protected async tryRepairNullByteCollections(
|
||||
err: unknown,
|
||||
reason: string,
|
||||
lease: PluginStateLeaseContext,
|
||||
): Promise<boolean> {
|
||||
return await this.collectionController.tryRepairNullByteCollections(err, reason, lease);
|
||||
}
|
||||
|
||||
protected async tryRepairDuplicateDocumentConstraint(
|
||||
err: unknown,
|
||||
reason: string,
|
||||
lease: PluginStateLeaseContext,
|
||||
): Promise<boolean> {
|
||||
return await this.collectionController.tryRepairDuplicateDocumentConstraint(err, reason, lease);
|
||||
}
|
||||
|
||||
protected computeManagedCollectionNames(): string[] {
|
||||
const seen = new Set<string>();
|
||||
const names: string[] = [];
|
||||
for (const collection of this.qmd.collections) {
|
||||
const name = collection.name?.trim();
|
||||
if (!name || seen.has(name)) {
|
||||
continue;
|
||||
}
|
||||
seen.add(name);
|
||||
names.push(name);
|
||||
}
|
||||
return names;
|
||||
}
|
||||
|
||||
protected async symlinkSharedModels(): Promise<void> {
|
||||
const defaultCacheHome =
|
||||
process.env.XDG_CACHE_HOME ||
|
||||
(process.platform === "win32" ? process.env.LOCALAPPDATA : undefined) ||
|
||||
path.join(os.homedir(), ".cache");
|
||||
const defaultModelsDir = path.join(defaultCacheHome, "qmd", "models");
|
||||
const targetModelsDir = path.join(this.xdgCacheHome, "qmd", "models");
|
||||
try {
|
||||
const stat = await fs.stat(defaultModelsDir).catch((err: unknown) => {
|
||||
if ((err as NodeJS.ErrnoException).code === "ENOENT") {
|
||||
return null;
|
||||
}
|
||||
throw err;
|
||||
});
|
||||
if (!stat?.isDirectory()) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await fs.lstat(targetModelsDir);
|
||||
return;
|
||||
} catch {
|
||||
// Does not exist – proceed to create symlink.
|
||||
}
|
||||
try {
|
||||
await fs.symlink(defaultModelsDir, targetModelsDir, "dir");
|
||||
} catch (symlinkErr: unknown) {
|
||||
const code = (symlinkErr as NodeJS.ErrnoException).code;
|
||||
if (process.platform === "win32" && (code === "EPERM" || code === "ENOTSUP")) {
|
||||
await fs.symlink(defaultModelsDir, targetModelsDir, "junction");
|
||||
} else {
|
||||
throw symlinkErr;
|
||||
}
|
||||
}
|
||||
qmdManagerLog.debug(`symlinked qmd models: ${defaultModelsDir} → ${targetModelsDir}`);
|
||||
} catch (err) {
|
||||
// Non-fatal: if we can't symlink, qmd will fall back to downloading.
|
||||
qmdManagerLog.warn(`failed to symlink qmd models directory: ${String(err)}`);
|
||||
}
|
||||
}
|
||||
|
||||
protected async runQmd(
|
||||
args: string[],
|
||||
opts?: { timeoutMs?: number; discardOutput?: boolean; signal?: AbortSignal },
|
||||
): Promise<{ stdout: string; stderr: string }> {
|
||||
return await this.commands.run(args, opts);
|
||||
}
|
||||
|
||||
protected abstract runUpdate(reason: string, force?: boolean): Promise<void>;
|
||||
protected abstract ensureWatcher(): void;
|
||||
protected abstract shouldScheduleEmbedTimer(): boolean;
|
||||
protected abstract resolveEmbedStartupJitterMs(): number;
|
||||
protected abstract withQmdStoreWriteLease<T>(
|
||||
task: (lease: PluginStateLeaseContext) => Promise<T>,
|
||||
parentSignal?: AbortSignal,
|
||||
): Promise<T>;
|
||||
protected abstract ensureDb(): SqliteDatabase;
|
||||
}
|
||||
|
||||
export function resolveQmdManagerRuntimeConfig(
|
||||
cfg: OpenClawConfig,
|
||||
agentId: string,
|
||||
): QmdManagerRuntimeConfig {
|
||||
return {
|
||||
workspaceDir: resolveAgentWorkspaceDir(cfg, agentId),
|
||||
syncSettings: resolveMemorySearchSyncConfig(cfg, agentId),
|
||||
contextLimits: resolveAgentContextLimits(cfg, agentId),
|
||||
};
|
||||
}
|
||||
|
||||
export type QmdManagerCreateParams = {
|
||||
cfg: OpenClawConfig;
|
||||
agentId: string;
|
||||
resolved: ResolvedMemoryBackendConfig;
|
||||
withLease: PluginStateLeaseRunner;
|
||||
mode?: QmdManagerMode;
|
||||
runtimeConfig?: QmdManagerRuntimeConfig;
|
||||
};
|
||||
|
||||
export type { ManagedCollection };
|
||||
@@ -0,0 +1,156 @@
|
||||
import crypto from "node:crypto";
|
||||
import path from "node:path";
|
||||
import { resolveGlobalSingleton } from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
|
||||
import type { ResolvedQmdConfig } from "openclaw/plugin-sdk/memory-core-host-engine-storage";
|
||||
import { isFutureDateTimestampMs, MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime";
|
||||
import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
|
||||
export const SEARCH_PENDING_UPDATE_WAIT_MS = 500;
|
||||
export const MAX_QMD_OUTPUT_CHARS = 200_000;
|
||||
export const QMD_EMBED_BACKOFF_BASE_MS = 60_000;
|
||||
export const QMD_EMBED_BACKOFF_MAX_MS = 60 * 60 * 1000;
|
||||
|
||||
const QMD_EMBED_LEASE_MIN_WAIT_MS = 15 * 60 * 1000;
|
||||
const QMD_WRITE_LEASE_MIN_WAIT_MS = 5 * 60 * 1000;
|
||||
const QMD_EMBED_QUEUE_KEY = Symbol.for("openclaw.qmdEmbedQueueTail");
|
||||
const QMD_UPDATE_QUEUE_KEY = Symbol.for("openclaw.qmdUpdateQueueState");
|
||||
const IGNORED_MEMORY_WATCH_DIR_NAMES = new Set([
|
||||
".git",
|
||||
".cache",
|
||||
"node_modules",
|
||||
"vendor",
|
||||
"dist",
|
||||
"build",
|
||||
".pnpm-store",
|
||||
".venv",
|
||||
"venv",
|
||||
".tox",
|
||||
"__pycache__",
|
||||
]);
|
||||
|
||||
type QmdEmbedQueueState = {
|
||||
tail: Promise<void>;
|
||||
};
|
||||
|
||||
type QmdUpdateQueueState = {
|
||||
tails: Map<string, Promise<void>>;
|
||||
};
|
||||
|
||||
export function qmdUsesVectors(searchMode: ResolvedQmdConfig["searchMode"]): boolean {
|
||||
return searchMode !== "search";
|
||||
}
|
||||
|
||||
export function buildQmdProcessPath(rawPath: string | undefined): string {
|
||||
const nodeBinDir = path.dirname(process.execPath);
|
||||
const entries = rawPath?.split(path.delimiter).filter(Boolean) ?? [];
|
||||
if (entries.includes(nodeBinDir)) {
|
||||
return rawPath ?? nodeBinDir;
|
||||
}
|
||||
return [...entries, nodeBinDir].join(path.delimiter);
|
||||
}
|
||||
|
||||
export function normalizePositiveInteger(value: number | undefined, fallback: number): number {
|
||||
return typeof value === "number" && Number.isFinite(value)
|
||||
? Math.max(1, Math.floor(value))
|
||||
: fallback;
|
||||
}
|
||||
|
||||
export function getQmdEmbedQueueState(): QmdEmbedQueueState {
|
||||
return resolveGlobalSingleton<QmdEmbedQueueState>(QMD_EMBED_QUEUE_KEY, () => ({
|
||||
tail: Promise.resolve(),
|
||||
}));
|
||||
}
|
||||
|
||||
export function getQmdUpdateQueueState(): QmdUpdateQueueState {
|
||||
return resolveGlobalSingleton<QmdUpdateQueueState>(QMD_UPDATE_QUEUE_KEY, () => ({
|
||||
tails: new Map<string, Promise<void>>(),
|
||||
}));
|
||||
}
|
||||
|
||||
export function normalizeHanBm25Query(query: string): string {
|
||||
const trimmed = query.trim();
|
||||
// Keep Han/CJK BM25 queries intact so OpenClaw search semantics match direct qmd search.
|
||||
return trimmed;
|
||||
}
|
||||
|
||||
export function parseQmdStatusVectorCount(raw: string): number | null {
|
||||
for (const line of raw.split(/\r?\n/)) {
|
||||
const match = line.match(/^\s*Vectors(?:\s*[:=]\s*|\s+)(\d+)\b/i);
|
||||
if (match?.[1]) {
|
||||
const count = Number.parseInt(match[1], 10);
|
||||
if (Number.isFinite(count)) {
|
||||
return count;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
export function resolveStableJitterMs(params: { seed: string; windowMs: number }): number {
|
||||
if (params.windowMs <= 0) {
|
||||
return 0;
|
||||
}
|
||||
const hash = crypto.createHash("sha256").update(params.seed).digest();
|
||||
const bucket = hash.readUInt32BE(0);
|
||||
return bucket % (Math.floor(params.windowMs) + 1);
|
||||
}
|
||||
|
||||
function resolveQmdWriteLeaseOptions(expectedMs: number, minWaitMs: number) {
|
||||
const expected = Math.max(1, expectedMs);
|
||||
return {
|
||||
leaseMs: Math.min(MAX_TIMER_TIMEOUT_MS, Math.max(minWaitMs, expected * 2)),
|
||||
waitMs: Math.min(MAX_TIMER_TIMEOUT_MS, Math.max(minWaitMs, expected * 6)),
|
||||
};
|
||||
}
|
||||
|
||||
// Cross-process serialization for qmd embeds (heavy ML work, serialized globally).
|
||||
export function resolveQmdEmbedLeaseOptions(embedTimeoutMs: number) {
|
||||
return resolveQmdWriteLeaseOptions(embedTimeoutMs, QMD_EMBED_LEASE_MIN_WAIT_MS);
|
||||
}
|
||||
|
||||
// One per-agent write lease shared by the update and embed phases (both write the
|
||||
// same qmd index.sqlite), so a foreground `memory search` dirty-sync and a
|
||||
// background gateway update/embed never write the same store at once
|
||||
// (writer-vs-writer SQLITE_BUSY, #66339). Sized to the slower of the two writes
|
||||
// so a contending caller waits for the in-flight write instead of erroring.
|
||||
export function resolveQmdStoreWriteLeaseOptions(updateTimeoutMs: number, embedTimeoutMs: number) {
|
||||
return resolveQmdWriteLeaseOptions(
|
||||
Math.max(updateTimeoutMs, embedTimeoutMs),
|
||||
QMD_WRITE_LEASE_MIN_WAIT_MS,
|
||||
);
|
||||
}
|
||||
|
||||
function hasIgnoredMemoryWatchSegment(relativePath: string): boolean {
|
||||
const parts = relativePath
|
||||
.split(path.sep)
|
||||
.map((segment) => normalizeLowercaseStringOrEmpty(segment))
|
||||
.filter(Boolean);
|
||||
return parts.some((segment) => IGNORED_MEMORY_WATCH_DIR_NAMES.has(segment));
|
||||
}
|
||||
|
||||
export function shouldIgnoreMemoryWatchPath(watchPath: string, roots: readonly string[]): boolean {
|
||||
const normalized = path.normalize(watchPath);
|
||||
let matchedRelative: string | null = null;
|
||||
let matchedRootLength = -1;
|
||||
for (const watchRoot of roots) {
|
||||
const normalizedRoot = path.normalize(watchRoot);
|
||||
const relative = path.relative(normalizedRoot, normalized);
|
||||
if (relative !== ".." && !relative.startsWith(`..${path.sep}`) && !path.isAbsolute(relative)) {
|
||||
if (normalizedRoot.length > matchedRootLength) {
|
||||
matchedRelative = relative;
|
||||
matchedRootLength = normalizedRoot.length;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (matchedRelative !== null) {
|
||||
if (matchedRelative === "") {
|
||||
return false;
|
||||
}
|
||||
return hasIgnoredMemoryWatchSegment(matchedRelative);
|
||||
}
|
||||
return hasIgnoredMemoryWatchSegment(normalized);
|
||||
}
|
||||
|
||||
export function isEmbedBackoffActive(embedBackoffUntil: number | null): boolean {
|
||||
return embedBackoffUntil !== null && isFutureDateTimestampMs(embedBackoffUntil);
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
import fs from "node:fs/promises";
|
||||
import readline from "node:readline";
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import {
|
||||
buildMemoryReadResult,
|
||||
buildMemoryReadResultFromSlice,
|
||||
DEFAULT_MEMORY_READ_LINES,
|
||||
isFileMissingError,
|
||||
type MemoryEmbeddingProbeResult,
|
||||
type MemoryProviderStatus,
|
||||
type MemoryReadResult,
|
||||
type MemorySource,
|
||||
statRegularFile,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
|
||||
import { openNodeSqliteDatabase } from "openclaw/plugin-sdk/sqlite-runtime";
|
||||
import { isDefaultQmdMemoryPath as isDefaultMemoryPath } from "./qmd-document-resolver.js";
|
||||
import { qmdManagerLog } from "./qmd-manager-base.js";
|
||||
import {
|
||||
normalizePositiveInteger,
|
||||
parseQmdStatusVectorCount,
|
||||
qmdUsesVectors,
|
||||
} from "./qmd-manager-helpers.js";
|
||||
import { QmdManagerSearch } from "./qmd-manager-search.js";
|
||||
|
||||
type SqliteDatabase = import("node:sqlite").DatabaseSync;
|
||||
|
||||
export abstract class QmdManagerIo extends QmdManagerSearch {
|
||||
async readFile(params: {
|
||||
relPath: string;
|
||||
from?: number;
|
||||
lines?: number;
|
||||
}): Promise<MemoryReadResult> {
|
||||
const relPath = params.relPath?.trim();
|
||||
if (!relPath) {
|
||||
throw new Error("path required");
|
||||
}
|
||||
const absPath = this.resolveReadPath(relPath);
|
||||
if (!absPath.endsWith(".md")) {
|
||||
throw new Error("path required");
|
||||
}
|
||||
let statResult: Awaited<ReturnType<typeof statRegularFile>>;
|
||||
try {
|
||||
statResult = await statRegularFile(absPath);
|
||||
} catch (err) {
|
||||
if (err instanceof Error && err.message === "path must be a regular file") {
|
||||
throw new Error("path required", { cause: err });
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
if (statResult.missing) {
|
||||
return { text: "", path: relPath };
|
||||
}
|
||||
if (params.from !== undefined || params.lines !== undefined) {
|
||||
const startLine = normalizePositiveInteger(params.from, 1);
|
||||
const requestedCount = normalizePositiveInteger(
|
||||
params.lines ?? DEFAULT_MEMORY_READ_LINES,
|
||||
DEFAULT_MEMORY_READ_LINES,
|
||||
);
|
||||
const partial = await this.readPartialText(absPath, startLine, requestedCount);
|
||||
if (partial.missing) {
|
||||
return { text: "", path: relPath };
|
||||
}
|
||||
return buildMemoryReadResultFromSlice({
|
||||
selectedLines: partial.selectedLines,
|
||||
relPath,
|
||||
startLine,
|
||||
moreSourceLinesRemain: partial.moreSourceLinesRemain,
|
||||
maxChars: this.contextLimits?.memoryGetMaxChars,
|
||||
suggestReadFallback: isDefaultMemoryPath(relPath),
|
||||
});
|
||||
}
|
||||
const full = await this.readFullText(absPath);
|
||||
if (full.missing) {
|
||||
return { text: "", path: relPath };
|
||||
}
|
||||
return buildMemoryReadResult({
|
||||
content: full.text,
|
||||
relPath,
|
||||
from: params.from,
|
||||
lines: params.lines,
|
||||
defaultLines: DEFAULT_MEMORY_READ_LINES,
|
||||
maxChars: this.contextLimits?.memoryGetMaxChars,
|
||||
suggestReadFallback: isDefaultMemoryPath(relPath),
|
||||
});
|
||||
}
|
||||
|
||||
status(): MemoryProviderStatus {
|
||||
const counts = this.readCounts();
|
||||
return {
|
||||
backend: "qmd",
|
||||
provider: "qmd",
|
||||
model: "qmd",
|
||||
requestedProvider: "qmd",
|
||||
files: counts.totalDocuments,
|
||||
chunks: counts.totalDocuments,
|
||||
dirty: this.dirty,
|
||||
workspaceDir: this.workspaceDir,
|
||||
dbPath: this.indexPath,
|
||||
sources: Array.from(this.sources),
|
||||
sourceCounts: counts.sourceCounts,
|
||||
vector: {
|
||||
enabled: qmdUsesVectors(this.qmd.searchMode),
|
||||
available: this.vectorAvailable ?? undefined,
|
||||
semanticAvailable: this.vectorAvailable ?? undefined,
|
||||
loadError: this.vectorStatusDetail ?? undefined,
|
||||
},
|
||||
batch: {
|
||||
enabled: false,
|
||||
failures: 0,
|
||||
limit: 0,
|
||||
wait: false,
|
||||
concurrency: 0,
|
||||
pollIntervalMs: 0,
|
||||
timeoutMs: 0,
|
||||
},
|
||||
custom: {
|
||||
qmd: {
|
||||
collections: this.qmd.collections.length,
|
||||
lastUpdateAt: this.lastUpdateAt,
|
||||
embedFailures: this.embedFailureCount,
|
||||
embedBackoffUntil: this.embedBackoffUntil,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async probeEmbeddingAvailability(): Promise<MemoryEmbeddingProbeResult> {
|
||||
if (!qmdUsesVectors(this.qmd.searchMode)) {
|
||||
return { ok: true, checked: false };
|
||||
}
|
||||
const ok = await this.probeVectorAvailability();
|
||||
return {
|
||||
ok,
|
||||
error: ok ? undefined : (this.vectorStatusDetail ?? "QMD semantic vectors are unavailable"),
|
||||
};
|
||||
}
|
||||
|
||||
async probeVectorAvailability(): Promise<boolean> {
|
||||
if (!qmdUsesVectors(this.qmd.searchMode)) {
|
||||
this.vectorAvailable = false;
|
||||
this.vectorStatusDetail = null;
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
const result = await this.runQmd(["status"], {
|
||||
timeoutMs: this.qmd.limits.timeoutMs,
|
||||
});
|
||||
const vectorCount = parseQmdStatusVectorCount(`${result.stdout}\n${result.stderr}`);
|
||||
if (vectorCount === null) {
|
||||
this.vectorAvailable = false;
|
||||
this.vectorStatusDetail = "Could not determine QMD vector status from `qmd status`";
|
||||
return false;
|
||||
}
|
||||
this.vectorAvailable = vectorCount > 0;
|
||||
this.vectorStatusDetail =
|
||||
vectorCount > 0
|
||||
? null
|
||||
: "QMD index has 0 vectors; semantic search is unavailable until embeddings finish";
|
||||
return this.vectorAvailable;
|
||||
} catch (err) {
|
||||
const message = formatErrorMessage(err);
|
||||
this.vectorAvailable = false;
|
||||
this.vectorStatusDetail = `QMD status probe failed: ${message}`;
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
protected async readPartialText(
|
||||
absPath: string,
|
||||
from?: number,
|
||||
lines?: number,
|
||||
): Promise<
|
||||
{ missing: true } | { missing: false; selectedLines: string[]; moreSourceLinesRemain: boolean }
|
||||
> {
|
||||
const start = normalizePositiveInteger(from, 1);
|
||||
const count = normalizePositiveInteger(lines, Number.MAX_SAFE_INTEGER);
|
||||
let handle;
|
||||
try {
|
||||
handle = await fs.open(absPath);
|
||||
} catch (err) {
|
||||
if (isFileMissingError(err)) {
|
||||
return { missing: true };
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
const stream = handle.createReadStream({ encoding: "utf-8" });
|
||||
const rl = readline.createInterface({ input: stream, crlfDelay: Infinity });
|
||||
const selected: string[] = [];
|
||||
let index = 0;
|
||||
let moreSourceLinesRemain = false;
|
||||
try {
|
||||
for await (const line of rl) {
|
||||
index += 1;
|
||||
if (index < start) {
|
||||
continue;
|
||||
}
|
||||
if (selected.length >= count) {
|
||||
moreSourceLinesRemain = true;
|
||||
break;
|
||||
}
|
||||
selected.push(line);
|
||||
}
|
||||
} finally {
|
||||
rl.close();
|
||||
await handle.close();
|
||||
}
|
||||
return {
|
||||
missing: false,
|
||||
selectedLines: selected.slice(0, count),
|
||||
moreSourceLinesRemain,
|
||||
};
|
||||
}
|
||||
|
||||
protected async readFullText(
|
||||
absPath: string,
|
||||
): Promise<{ missing: true } | { missing: false; text: string }> {
|
||||
try {
|
||||
return { missing: false, text: await fs.readFile(absPath, "utf-8") };
|
||||
} catch (err) {
|
||||
if (isFileMissingError(err)) {
|
||||
return { missing: true };
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
protected ensureDb(): SqliteDatabase {
|
||||
if (this.db) {
|
||||
return this.db;
|
||||
}
|
||||
this.db = openNodeSqliteDatabase(this.indexPath, { readOnly: true });
|
||||
// busy_timeout is per-connection; keep it below the write path because this
|
||||
// synchronous read runs on the main thread and WAL readers rarely block.
|
||||
this.db.exec("PRAGMA busy_timeout = 1000");
|
||||
return this.db;
|
||||
}
|
||||
|
||||
protected resolveReadPath(relPath: string): string {
|
||||
return this.documentResolver.resolveReadPath(relPath);
|
||||
}
|
||||
|
||||
protected readCounts(): {
|
||||
totalDocuments: number;
|
||||
sourceCounts: Array<{ source: MemorySource; files: number; chunks: number }>;
|
||||
} {
|
||||
try {
|
||||
const rows = this.ensureDb()
|
||||
.prepare(
|
||||
"SELECT collection, COUNT(*) as c FROM documents WHERE active = 1 GROUP BY collection",
|
||||
)
|
||||
.all() as Array<{ collection: string; c: number }>;
|
||||
const bySource = new Map<MemorySource, { files: number; chunks: number }>();
|
||||
for (const source of this.sources) {
|
||||
bySource.set(source, { files: 0, chunks: 0 });
|
||||
}
|
||||
let total = 0;
|
||||
for (const row of rows) {
|
||||
const source = this.collectionRoots.get(row.collection)?.kind ?? "memory";
|
||||
const entry = bySource.get(source) ?? { files: 0, chunks: 0 };
|
||||
entry.files += row.c ?? 0;
|
||||
entry.chunks += row.c ?? 0;
|
||||
bySource.set(source, entry);
|
||||
total += row.c ?? 0;
|
||||
}
|
||||
return {
|
||||
totalDocuments: total,
|
||||
sourceCounts: Array.from(bySource.entries()).map(([source, value]) => ({
|
||||
source,
|
||||
files: value.files,
|
||||
chunks: value.chunks,
|
||||
})),
|
||||
};
|
||||
} catch (err) {
|
||||
qmdManagerLog.warn(`failed to read qmd index stats: ${String(err)}`);
|
||||
return {
|
||||
totalDocuments: 0,
|
||||
sourceCounts: Array.from(this.sources).map((source) => ({
|
||||
source,
|
||||
files: 0,
|
||||
chunks: 0,
|
||||
})),
|
||||
};
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
import path from "node:path";
|
||||
import chokidar from "chokidar";
|
||||
import { qmdManagerLog, type ManagedCollection } from "./qmd-manager-base.js";
|
||||
import { shouldIgnoreMemoryWatchPath } from "./qmd-manager-helpers.js";
|
||||
import { QmdManagerSync } from "./qmd-manager-sync.js";
|
||||
import { countChokidarWatchedEntries, warnIfMemoryWatchPressureHigh } from "./watch-pressure.js";
|
||||
import {
|
||||
recordMemoryWatchEventPath,
|
||||
settleMemoryWatchEventPaths,
|
||||
type MemoryWatchEventStats,
|
||||
} from "./watch-settle.js";
|
||||
|
||||
export abstract class QmdManagerLifecycle extends QmdManagerSync {
|
||||
protected ensureWatcher(): void {
|
||||
if (!this.syncSettings?.watch || this.watcher || this.closed) {
|
||||
return;
|
||||
}
|
||||
const watchPaths = new Set<string>();
|
||||
const watchRoots = new Set<string>();
|
||||
for (const collection of this.qmd.collections) {
|
||||
if (collection.kind === "sessions") {
|
||||
continue;
|
||||
}
|
||||
watchRoots.add(path.normalize(collection.path));
|
||||
watchPaths.add(this.resolveCollectionWatchPath(collection));
|
||||
}
|
||||
if (watchPaths.size === 0) {
|
||||
return;
|
||||
}
|
||||
const watchPathList = Array.from(watchPaths);
|
||||
const startTime = Date.now();
|
||||
qmdManagerLog.info(
|
||||
`qmd watcher starting for agent "${this.agentId}" paths=${watchPathList.length}`,
|
||||
);
|
||||
const watchRootList = Array.from(watchRoots);
|
||||
const watcher = chokidar.watch(watchPathList, {
|
||||
ignoreInitial: true,
|
||||
ignored: (watchPath) => shouldIgnoreMemoryWatchPath(watchPath, watchRootList),
|
||||
});
|
||||
this.watcher = watcher;
|
||||
const markDirty = (watchPath?: string, stats?: MemoryWatchEventStats) => {
|
||||
recordMemoryWatchEventPath(this.pendingWatchPaths, watchPath, stats);
|
||||
this.dirty = true;
|
||||
this.scheduleWatchSync();
|
||||
};
|
||||
watcher.on("add", markDirty);
|
||||
watcher.on("change", markDirty);
|
||||
watcher.on("unlink", markDirty);
|
||||
watcher.once("ready", () => {
|
||||
this.warnIfWatchPressure(countChokidarWatchedEntries(watcher));
|
||||
qmdManagerLog.info(
|
||||
`qmd watcher ready for agent "${this.agentId}" paths=${watchPathList.length} durationMs=${Date.now() - startTime}`,
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
protected warnIfWatchPressure(count: number): void {
|
||||
warnIfMemoryWatchPressureHigh(
|
||||
this.watchPressureWarning,
|
||||
count,
|
||||
"paths",
|
||||
"Large QMD collections can make OpenClaw run out of file watchers or open files.",
|
||||
"Remove large collections, or set memory.search.sync.watch to false and refresh memory manually.",
|
||||
(message) => qmdManagerLog.warn(message),
|
||||
);
|
||||
}
|
||||
|
||||
protected resolveCollectionWatchPath(collection: ManagedCollection): string {
|
||||
return path.join(path.normalize(collection.path), collection.pattern);
|
||||
}
|
||||
|
||||
protected scheduleWatchSync(): void {
|
||||
if (!this.syncSettings?.watch) {
|
||||
return;
|
||||
}
|
||||
if (this.watchTimer) {
|
||||
clearTimeout(this.watchTimer);
|
||||
}
|
||||
this.watchTimer = setTimeout(() => {
|
||||
this.watchTimer = null;
|
||||
void (async () => {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
if (!(await settleMemoryWatchEventPaths(this.pendingWatchPaths))) {
|
||||
if (!this.closed) {
|
||||
this.scheduleWatchSync();
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
await this.sync({ reason: "watch" });
|
||||
})().catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd watch sync failed: ${String(err)}`);
|
||||
});
|
||||
}, this.syncSettings.watchDebounceMs);
|
||||
}
|
||||
|
||||
protected async maybeWarmSession(sessionKey?: string): Promise<void> {
|
||||
if (this.mode === "cli" || !this.syncSettings?.onSessionStart) {
|
||||
return;
|
||||
}
|
||||
const key = sessionKey?.trim() || "";
|
||||
if (!key || this.sessionWarm.has(key)) {
|
||||
return;
|
||||
}
|
||||
this.sessionWarm.add(key);
|
||||
void this.sync({ reason: "session-start" }).catch((err: unknown) => {
|
||||
qmdManagerLog.warn(`qmd session-start sync failed: ${String(err)}`);
|
||||
});
|
||||
}
|
||||
|
||||
protected async maybeSyncDirtySearchState(): Promise<void> {
|
||||
if (this.mode === "cli" || !this.syncSettings?.onSearch || !this.dirty) {
|
||||
return;
|
||||
}
|
||||
await this.sync({ reason: "search" });
|
||||
}
|
||||
|
||||
async close(): Promise<void> {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
this.closed = true;
|
||||
this.resolveCloseSignal();
|
||||
this.closeAbortController.abort(new Error("qmd manager closed"));
|
||||
if (this.updateTimer) {
|
||||
clearInterval(this.updateTimer);
|
||||
this.updateTimer = null;
|
||||
}
|
||||
if (this.embedTimer) {
|
||||
clearTimeout(this.embedTimer);
|
||||
this.embedTimer = null;
|
||||
}
|
||||
if (this.watchTimer) {
|
||||
clearTimeout(this.watchTimer);
|
||||
this.watchTimer = null;
|
||||
}
|
||||
if (this.watcher) {
|
||||
await this.watcher.close().catch(() => undefined);
|
||||
this.watcher = null;
|
||||
}
|
||||
this.queuedForcedRuns = 0;
|
||||
await this.pendingUpdate?.catch(() => undefined);
|
||||
await this.queuedForcedUpdate?.catch(() => undefined);
|
||||
if (this.db) {
|
||||
this.db.close();
|
||||
this.db = null;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,477 @@
|
||||
import { truncateUtf16Safe } from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
|
||||
import {
|
||||
deriveQmdScopeChannel,
|
||||
deriveQmdScopeChatType,
|
||||
isQmdScopeAllowed,
|
||||
type QmdQueryResult,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-qmd";
|
||||
import type {
|
||||
MemorySearchResult,
|
||||
MemorySearchRuntimeDebug,
|
||||
MemorySource,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
|
||||
import { uniqueValues } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import { copyQmdSessionArtifactHit } from "../qmd-session-artifacts.js";
|
||||
import type { QmdSearchRuntimeDebugContext } from "./qmd-collection-controller.js";
|
||||
import type { QmdCommandPhaseReporter } from "./qmd-command-client.js";
|
||||
import {
|
||||
asQmdAbortError,
|
||||
isMissingCollectionSearchError,
|
||||
isUnsupportedQmdOptionError,
|
||||
} from "./qmd-command-errors.js";
|
||||
import type { QmdDocLocation as DocLocation } from "./qmd-document-resolver.js";
|
||||
import { qmdManagerLog } from "./qmd-manager-base.js";
|
||||
import { normalizeHanBm25Query, SEARCH_PENDING_UPDATE_WAIT_MS } from "./qmd-manager-helpers.js";
|
||||
import { QmdManagerLifecycle } from "./qmd-manager-lifecycle.js";
|
||||
import {
|
||||
clearQmdMultiCollectionProbeCache,
|
||||
readQmdMultiCollectionProbeCache,
|
||||
writeQmdMultiCollectionProbeCache,
|
||||
} from "./qmd-runtime-cache.js";
|
||||
|
||||
const SNIPPET_HEADER_RE = /@@\s*-([0-9]+),([0-9]+)/;
|
||||
|
||||
export abstract class QmdManagerSearchSupport extends QmdManagerLifecycle {
|
||||
protected recordSearchPlanDebug(params: {
|
||||
debugContext: QmdSearchRuntimeDebugContext;
|
||||
command: "query" | "search" | "vsearch";
|
||||
collectionNames: string[];
|
||||
collectionGroups: string[][];
|
||||
}): void {
|
||||
const sources = uniqueValues(
|
||||
params.collectionNames
|
||||
.map((collectionName) => this.collectionRoots.get(collectionName)?.kind)
|
||||
.filter((source): source is MemorySource => Boolean(source)),
|
||||
);
|
||||
params.debugContext.searchPlan = {
|
||||
command: params.command,
|
||||
collectionCount: params.collectionNames.length,
|
||||
groupCount: params.collectionGroups.length,
|
||||
sources,
|
||||
};
|
||||
}
|
||||
|
||||
protected beginQmdSearchRuntimeDebug(): QmdSearchRuntimeDebugContext {
|
||||
const debugContext: QmdSearchRuntimeDebugContext = {};
|
||||
const collectionValidation = this.collectionController.consumePendingValidationDebug();
|
||||
if (collectionValidation) {
|
||||
debugContext.collectionValidation = collectionValidation;
|
||||
}
|
||||
return debugContext;
|
||||
}
|
||||
|
||||
protected consumeQmdRuntimeDebug(
|
||||
debugContext: QmdSearchRuntimeDebugContext,
|
||||
): MemorySearchRuntimeDebug["qmd"] | undefined {
|
||||
const debug: NonNullable<MemorySearchRuntimeDebug["qmd"]> = {};
|
||||
if (debugContext.collectionValidation) {
|
||||
debug.collectionValidation = debugContext.collectionValidation;
|
||||
}
|
||||
if (debugContext.multiCollectionProbe) {
|
||||
debug.multiCollectionProbe = debugContext.multiCollectionProbe;
|
||||
}
|
||||
if (debugContext.searchPlan) {
|
||||
debug.searchPlan = debugContext.searchPlan;
|
||||
}
|
||||
return Object.keys(debug).length > 0 ? debug : undefined;
|
||||
}
|
||||
|
||||
protected async tryRepairMissingCollectionSearch(
|
||||
err: unknown,
|
||||
debugContext: QmdSearchRuntimeDebugContext,
|
||||
parentSignal?: AbortSignal,
|
||||
): Promise<boolean> {
|
||||
if (!isMissingCollectionSearchError(err)) {
|
||||
return false;
|
||||
}
|
||||
qmdManagerLog.warn(
|
||||
"qmd search failed because a managed collection is missing; repairing collections and retrying once",
|
||||
);
|
||||
await this.ensureCollections({ force: true, debugContext, parentSignal });
|
||||
return true;
|
||||
}
|
||||
|
||||
protected async runQmdSearch(
|
||||
args: string[],
|
||||
command: "query" | "search" | "vsearch",
|
||||
signal?: AbortSignal,
|
||||
reportCommandPhase?: QmdCommandPhaseReporter,
|
||||
): Promise<QmdQueryResult[]> {
|
||||
return await this.commands.search(args, command, signal, reportCommandPhase);
|
||||
}
|
||||
|
||||
protected async resolveDocLocation(
|
||||
docid?: string,
|
||||
hints?: { preferredCollection?: string; preferredFile?: string },
|
||||
): Promise<DocLocation | null> {
|
||||
return await this.documentResolver.resolveDocLocation(docid, hints);
|
||||
}
|
||||
|
||||
protected normalizeDocHints(hints?: { preferredCollection?: string; preferredFile?: string }): {
|
||||
preferredCollection?: string;
|
||||
preferredFile?: string;
|
||||
} {
|
||||
return this.documentResolver.normalizeDocHints(hints);
|
||||
}
|
||||
|
||||
protected toCollectionRelativePath(collection: string, filePath: string): string | null {
|
||||
return this.documentResolver.toCollectionRelativePath(collection, filePath);
|
||||
}
|
||||
|
||||
protected resolveSnippetLines(
|
||||
entry: QmdQueryResult,
|
||||
snippet: string,
|
||||
): { startLine: number; endLine: number } {
|
||||
const explicitStart = this.normalizeSnippetLine(entry.startLine);
|
||||
const explicitEnd = this.normalizeSnippetLine(entry.endLine);
|
||||
const headerLines = this.parseSnippetHeaderLines(snippet);
|
||||
if (explicitStart !== undefined && explicitEnd !== undefined) {
|
||||
return explicitStart <= explicitEnd
|
||||
? { startLine: explicitStart, endLine: explicitEnd }
|
||||
: { startLine: explicitEnd, endLine: explicitStart };
|
||||
}
|
||||
if (explicitStart !== undefined) {
|
||||
if (headerLines) {
|
||||
const width = headerLines.endLine - headerLines.startLine;
|
||||
return {
|
||||
startLine: explicitStart,
|
||||
endLine: explicitStart + Math.max(0, width),
|
||||
};
|
||||
}
|
||||
return { startLine: explicitStart, endLine: explicitStart };
|
||||
}
|
||||
if (explicitEnd !== undefined) {
|
||||
if (headerLines) {
|
||||
const width = headerLines.endLine - headerLines.startLine;
|
||||
return {
|
||||
startLine: Math.max(1, explicitEnd - Math.max(0, width)),
|
||||
endLine: explicitEnd,
|
||||
};
|
||||
}
|
||||
return { startLine: explicitEnd, endLine: explicitEnd };
|
||||
}
|
||||
if (headerLines) {
|
||||
return headerLines;
|
||||
}
|
||||
return { startLine: 1, endLine: snippet.split("\n").length };
|
||||
}
|
||||
|
||||
protected normalizeSnippetLine(value: unknown): number | undefined {
|
||||
return typeof value === "number" && Number.isFinite(value) && value > 0 ? value : undefined;
|
||||
}
|
||||
|
||||
protected parseSnippetHeaderLines(
|
||||
snippet: string,
|
||||
): { startLine: number; endLine: number } | null {
|
||||
const match = SNIPPET_HEADER_RE.exec(snippet);
|
||||
if (!match) {
|
||||
return null;
|
||||
}
|
||||
const start = Number(match[1]);
|
||||
const count = Number(match[2]);
|
||||
if (Number.isFinite(start) && Number.isFinite(count)) {
|
||||
return { startLine: start, endLine: start + count - 1 };
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
protected logScopeDenied(sessionKey?: string): void {
|
||||
const channel = deriveQmdScopeChannel(sessionKey) ?? "unknown";
|
||||
const chatType = deriveQmdScopeChatType(sessionKey) ?? "unknown";
|
||||
const key = sessionKey?.trim() || "<none>";
|
||||
qmdManagerLog.warn(
|
||||
`qmd search denied by scope (channel=${channel}, chatType=${chatType}, session=${key})`,
|
||||
);
|
||||
}
|
||||
|
||||
protected isScopeAllowed(sessionKey?: string): boolean {
|
||||
return isQmdScopeAllowed(this.qmd.scope, sessionKey);
|
||||
}
|
||||
|
||||
protected clampResultsByInjectedChars(results: MemorySearchResult[]): MemorySearchResult[] {
|
||||
const budget = this.qmd.limits.maxInjectedChars;
|
||||
if (!budget || budget <= 0) {
|
||||
return results;
|
||||
}
|
||||
let remaining = budget;
|
||||
const clamped: MemorySearchResult[] = [];
|
||||
for (const entry of results) {
|
||||
if (remaining <= 0) {
|
||||
break;
|
||||
}
|
||||
const snippet = entry.snippet ?? "";
|
||||
if (snippet.length <= remaining) {
|
||||
clamped.push(entry);
|
||||
remaining -= snippet.length;
|
||||
} else {
|
||||
const trimmed = truncateUtf16Safe(snippet, remaining);
|
||||
clamped.push(copyQmdSessionArtifactHit(entry, { ...entry, snippet: trimmed }));
|
||||
break;
|
||||
}
|
||||
}
|
||||
return clamped;
|
||||
}
|
||||
|
||||
protected diversifyResultsBySource(
|
||||
results: MemorySearchResult[],
|
||||
limit: number,
|
||||
): MemorySearchResult[] {
|
||||
const target = Math.max(0, limit);
|
||||
if (target <= 0) {
|
||||
return [];
|
||||
}
|
||||
if (results.length <= 1) {
|
||||
return results.slice(0, target);
|
||||
}
|
||||
const bySource = new Map<MemorySource, MemorySearchResult[]>();
|
||||
for (const entry of results) {
|
||||
const list = bySource.get(entry.source) ?? [];
|
||||
list.push(entry);
|
||||
bySource.set(entry.source, list);
|
||||
}
|
||||
if (!bySource.has("sessions") || !bySource.has("memory")) {
|
||||
return results.slice(0, target);
|
||||
}
|
||||
const sourceOrder = Array.from(bySource.entries())
|
||||
.toSorted((a, b) => (b[1][0]?.score ?? 0) - (a[1][0]?.score ?? 0))
|
||||
.map(([source]) => source);
|
||||
const diversified: MemorySearchResult[] = [];
|
||||
while (diversified.length < target) {
|
||||
let emitted = false;
|
||||
for (const source of sourceOrder) {
|
||||
const next = bySource.get(source)?.shift();
|
||||
if (!next) {
|
||||
continue;
|
||||
}
|
||||
diversified.push(next);
|
||||
emitted = true;
|
||||
if (diversified.length >= target) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!emitted) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return diversified;
|
||||
}
|
||||
|
||||
protected async waitForPendingUpdateBeforeSearch(): Promise<void> {
|
||||
const pending = this.pendingUpdate;
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
// Release the losing timer when the pending update settles first.
|
||||
let timeout: ReturnType<typeof setTimeout> | undefined;
|
||||
const wait = new Promise<void>((resolve) => {
|
||||
timeout = setTimeout(resolve, SEARCH_PENDING_UPDATE_WAIT_MS);
|
||||
});
|
||||
await Promise.race([pending.catch(() => undefined), wait]).finally(() => clearTimeout(timeout));
|
||||
}
|
||||
|
||||
protected async resolveCollectionSearchGroups(
|
||||
collectionNames: string[],
|
||||
signal?: AbortSignal,
|
||||
debugContext?: QmdSearchRuntimeDebugContext,
|
||||
): Promise<string[][]> {
|
||||
if (collectionNames.length <= 1) {
|
||||
return [collectionNames];
|
||||
}
|
||||
if (!(await this.supportsQmdMultiCollectionFilters(signal, debugContext))) {
|
||||
return collectionNames.map((collectionName) => [collectionName]);
|
||||
}
|
||||
return this.groupCollectionNamesBySource(collectionNames);
|
||||
}
|
||||
|
||||
protected async supportsQmdMultiCollectionFilters(
|
||||
signal?: AbortSignal,
|
||||
debugContext?: QmdSearchRuntimeDebugContext,
|
||||
): Promise<boolean> {
|
||||
if (signal?.aborted) {
|
||||
throw asQmdAbortError(signal);
|
||||
}
|
||||
if (this.multiCollectionFilterSupported !== null) {
|
||||
return this.multiCollectionFilterSupported;
|
||||
}
|
||||
const startedAt = Date.now();
|
||||
const cacheContext = await this.buildQmdMultiCollectionProbeCacheContext();
|
||||
const cached = await readQmdMultiCollectionProbeCache(cacheContext);
|
||||
if (cached.state === "hit") {
|
||||
this.multiCollectionFilterSupported = cached.value.multiCollectionProbe.supported;
|
||||
if (debugContext) {
|
||||
debugContext.multiCollectionProbe = {
|
||||
cacheState: "hit",
|
||||
elapsedMs: Math.max(0, Date.now() - startedAt),
|
||||
supported: this.multiCollectionFilterSupported,
|
||||
};
|
||||
}
|
||||
return this.multiCollectionFilterSupported;
|
||||
}
|
||||
try {
|
||||
const result = await this.runQmd(["--help"], {
|
||||
timeoutMs: Math.min(this.qmd.limits.timeoutMs, 5_000),
|
||||
signal,
|
||||
});
|
||||
const helpText = `${result.stdout}\n${result.stderr}`;
|
||||
this.multiCollectionFilterSupported =
|
||||
/\b(?:one or more collections|collection\(s\)|multiple -c flags)\b/i.test(helpText);
|
||||
const wroteCache = await writeQmdMultiCollectionProbeCache(
|
||||
cacheContext,
|
||||
this.multiCollectionFilterSupported,
|
||||
);
|
||||
if (debugContext) {
|
||||
debugContext.multiCollectionProbe = {
|
||||
cacheState: wroteCache ? "write" : "error",
|
||||
elapsedMs: Math.max(0, Date.now() - startedAt),
|
||||
supported: this.multiCollectionFilterSupported,
|
||||
};
|
||||
}
|
||||
} catch (err) {
|
||||
// Cancellation says nothing about QMD capabilities; leave the probe uncached.
|
||||
if (signal?.aborted) {
|
||||
throw asQmdAbortError(signal);
|
||||
}
|
||||
this.multiCollectionFilterSupported = false;
|
||||
if (debugContext) {
|
||||
debugContext.multiCollectionProbe = {
|
||||
cacheState: "error",
|
||||
elapsedMs: Math.max(0, Date.now() - startedAt),
|
||||
supported: false,
|
||||
};
|
||||
}
|
||||
qmdManagerLog.debug(`qmd multi-collection filter probe failed: ${String(err)}`);
|
||||
}
|
||||
return this.multiCollectionFilterSupported;
|
||||
}
|
||||
|
||||
protected async markQmdMultiCollectionFiltersUnsupported(
|
||||
debugContext: QmdSearchRuntimeDebugContext,
|
||||
): Promise<void> {
|
||||
const startedAt = Date.now();
|
||||
const cacheContext = await this.buildQmdMultiCollectionProbeCacheContext();
|
||||
this.multiCollectionFilterSupported = false;
|
||||
await clearQmdMultiCollectionProbeCache(cacheContext);
|
||||
const wroteCache = await writeQmdMultiCollectionProbeCache(cacheContext, false);
|
||||
debugContext.multiCollectionProbe = {
|
||||
cacheState: wroteCache ? "write" : "error",
|
||||
elapsedMs: Math.max(0, Date.now() - startedAt),
|
||||
supported: false,
|
||||
};
|
||||
}
|
||||
|
||||
protected async runQueryAcrossCollectionGroups(
|
||||
query: string,
|
||||
limit: number,
|
||||
collectionGroups: string[][],
|
||||
command: "query" | "search" | "vsearch",
|
||||
signal?: AbortSignal,
|
||||
reportCommandPhase?: QmdCommandPhaseReporter,
|
||||
): Promise<QmdQueryResult[]> {
|
||||
qmdManagerLog.debug(
|
||||
`qmd ${command} multi-source collection grouping active (${collectionGroups.length} groups)`,
|
||||
);
|
||||
const bestByResultKey = new Map<string, QmdQueryResult>();
|
||||
for (const collectionNames of collectionGroups) {
|
||||
const args = this.buildSearchArgs(command, query, limit);
|
||||
args.push(...this.buildCollectionFilterArgs(collectionNames));
|
||||
const parsed = await this.runQmdSearch(args, command, signal, reportCommandPhase);
|
||||
for (const entry of parsed) {
|
||||
const defaultCollection = collectionNames.length === 1 ? collectionNames[0] : undefined;
|
||||
const normalizedHints = this.normalizeDocHints({
|
||||
preferredCollection: entry.collection ?? defaultCollection,
|
||||
preferredFile: entry.file,
|
||||
});
|
||||
const normalizedDocId =
|
||||
typeof entry.docid === "string" && entry.docid.trim().length > 0
|
||||
? entry.docid
|
||||
: undefined;
|
||||
const withCollection = {
|
||||
...entry,
|
||||
docid: normalizedDocId,
|
||||
collection: normalizedHints.preferredCollection ?? entry.collection ?? defaultCollection,
|
||||
file: normalizedHints.preferredFile ?? entry.file,
|
||||
} satisfies QmdQueryResult;
|
||||
const resultKey = this.buildQmdResultKey(withCollection);
|
||||
if (!resultKey) {
|
||||
continue;
|
||||
}
|
||||
const prev = bestByResultKey.get(resultKey);
|
||||
const prevScore = typeof prev?.score === "number" ? prev.score : Number.NEGATIVE_INFINITY;
|
||||
const nextScore =
|
||||
typeof withCollection.score === "number"
|
||||
? withCollection.score
|
||||
: Number.NEGATIVE_INFINITY;
|
||||
if (!prev || nextScore > prevScore) {
|
||||
bestByResultKey.set(resultKey, withCollection);
|
||||
}
|
||||
}
|
||||
}
|
||||
return [...bestByResultKey.values()].toSorted((a, b) => (b.score ?? 0) - (a.score ?? 0));
|
||||
}
|
||||
|
||||
protected groupCollectionNamesBySource(collectionNames: string[]): string[][] {
|
||||
const groups = new Map<string, string[]>();
|
||||
for (const collectionName of collectionNames) {
|
||||
const source = this.collectionRoots.get(collectionName)?.kind ?? collectionName;
|
||||
const group = groups.get(source) ?? [];
|
||||
group.push(collectionName);
|
||||
groups.set(source, group);
|
||||
}
|
||||
return [...groups.values()];
|
||||
}
|
||||
|
||||
protected buildQmdResultKey(entry: QmdQueryResult): string | null {
|
||||
if (typeof entry.docid === "string" && entry.docid.trim().length > 0) {
|
||||
return `docid:${entry.docid}`;
|
||||
}
|
||||
const hints = this.normalizeDocHints({
|
||||
preferredCollection: entry.collection,
|
||||
preferredFile: entry.file,
|
||||
});
|
||||
if (!hints.preferredCollection || !hints.preferredFile) {
|
||||
return null;
|
||||
}
|
||||
const collectionRelativePath = this.toCollectionRelativePath(
|
||||
hints.preferredCollection,
|
||||
hints.preferredFile,
|
||||
);
|
||||
return collectionRelativePath
|
||||
? `file:${hints.preferredCollection}:${collectionRelativePath}`
|
||||
: null;
|
||||
}
|
||||
|
||||
protected listManagedCollectionNames(sources?: MemorySource[]): string[] {
|
||||
if (!sources?.length) {
|
||||
return this.managedCollectionNames;
|
||||
}
|
||||
const allowed = new Set(sources);
|
||||
return this.managedCollectionNames.filter((name) => {
|
||||
const source = this.collectionRoots.get(name)?.kind;
|
||||
return source ? allowed.has(source) : false;
|
||||
});
|
||||
}
|
||||
|
||||
protected buildCollectionFilterArgs(collectionNames: string[]): string[] {
|
||||
return collectionNames.filter(Boolean).flatMap((name) => ["-c", name]);
|
||||
}
|
||||
|
||||
protected buildSearchArgs(
|
||||
command: "query" | "search" | "vsearch",
|
||||
query: string,
|
||||
limit: number,
|
||||
): string[] {
|
||||
const normalizedQuery = command === "search" ? normalizeHanBm25Query(query) : query;
|
||||
if (command === "query") {
|
||||
const args = ["query", normalizedQuery, "--json", "-n", String(limit)];
|
||||
if (this.qmd.searchMode === "query" && this.qmd.rerank === false) {
|
||||
args.push("--no-rerank");
|
||||
}
|
||||
return args;
|
||||
}
|
||||
return [command, normalizedQuery, "--json", "-n", String(limit)];
|
||||
}
|
||||
|
||||
protected isUnsupportedQmdOptionError(err: unknown): boolean {
|
||||
return isUnsupportedQmdOptionError(err);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,283 @@
|
||||
import { truncateUtf16Safe } from "openclaw/plugin-sdk/memory-core-host-engine-foundation";
|
||||
import type { QmdQueryResult } from "openclaw/plugin-sdk/memory-core-host-engine-qmd";
|
||||
import type {
|
||||
MemorySearchResult,
|
||||
MemorySearchRuntimeDebug,
|
||||
MemorySource,
|
||||
} from "openclaw/plugin-sdk/memory-core-host-engine-storage";
|
||||
import { uniqueValues } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import {
|
||||
attachQmdSessionArtifactHit,
|
||||
resolveQmdSessionArtifactIdentity,
|
||||
} from "../qmd-session-artifacts.js";
|
||||
import { asQmdAbortError, isMissingCollectionSearchError } from "./qmd-command-errors.js";
|
||||
import { qmdManagerLog } from "./qmd-manager-base.js";
|
||||
import { QmdManagerSearchSupport } from "./qmd-manager-search-support.js";
|
||||
import {
|
||||
MEMORY_SEARCH_DEADLINE_CONTROL,
|
||||
type MemorySearchDeadlineControlOptions,
|
||||
} from "./search-deadline.js";
|
||||
|
||||
export abstract class QmdManagerSearch extends QmdManagerSearchSupport {
|
||||
async search(
|
||||
query: string,
|
||||
opts?: {
|
||||
maxResults?: number;
|
||||
minScore?: number;
|
||||
sessionKey?: string;
|
||||
qmdSearchModeOverride?: "query" | "search" | "vsearch";
|
||||
onDebug?: (debug: MemorySearchRuntimeDebug) => void;
|
||||
sources?: MemorySource[];
|
||||
/**
|
||||
* Caller-owned cancellation. When the caller stops waiting, abort kills
|
||||
* the in-flight qmd subprocess instead of leaving it orphaned.
|
||||
*/
|
||||
signal?: AbortSignal;
|
||||
} & MemorySearchDeadlineControlOptions,
|
||||
): Promise<MemorySearchResult[]> {
|
||||
if (!this.isScopeAllowed(opts?.sessionKey)) {
|
||||
this.logScopeDenied(opts?.sessionKey);
|
||||
return [];
|
||||
}
|
||||
const searchSignal = opts?.signal;
|
||||
const reportCommandPhase = opts?.[MEMORY_SEARCH_DEADLINE_CONTROL];
|
||||
if (searchSignal?.aborted) {
|
||||
throw asQmdAbortError(searchSignal);
|
||||
}
|
||||
const debugContext = this.beginQmdSearchRuntimeDebug();
|
||||
const trimmed = query.trim();
|
||||
if (!trimmed) {
|
||||
return [];
|
||||
}
|
||||
await this.maybeWarmSession(opts?.sessionKey);
|
||||
await this.maybeSyncDirtySearchState();
|
||||
await this.waitForPendingUpdateBeforeSearch();
|
||||
const resultLimit = Math.min(
|
||||
this.qmd.limits.maxResults,
|
||||
opts?.maxResults ?? this.qmd.limits.maxResults,
|
||||
);
|
||||
// Remember-only session exports are indexed for trusted recall but are not
|
||||
// part of ordinary manager searches. Explicit export keeps its existing
|
||||
// ordinary-access behavior; trusted recall always passes sources=sessions.
|
||||
const requestedSources = opts?.sources?.length
|
||||
? uniqueValues(opts.sources)
|
||||
: this.qmd.sessions.readable
|
||||
? undefined
|
||||
: (["memory"] satisfies MemorySource[]);
|
||||
const collectionNames = this.listManagedCollectionNames(requestedSources);
|
||||
const limit = resultLimit;
|
||||
if (collectionNames.length === 0) {
|
||||
qmdManagerLog.warn("qmd query skipped: no managed collections configured");
|
||||
return [];
|
||||
}
|
||||
const qmdSearchCommand = opts?.qmdSearchModeOverride ?? this.qmd.searchMode;
|
||||
let effectiveSearchMode: "query" | "search" | "vsearch" = qmdSearchCommand;
|
||||
let searchFallbackReason: string | undefined;
|
||||
const explicitSearchTool = this.qmd.searchTool;
|
||||
const mcporterEnabled = this.qmd.mcporter.enabled;
|
||||
const runSearchAttempt = async (
|
||||
allowMissingCollectionRepair: boolean,
|
||||
): Promise<QmdQueryResult[]> => {
|
||||
let attemptedCombinedCollectionFilter = false;
|
||||
try {
|
||||
if (mcporterEnabled) {
|
||||
const minScore = opts?.minScore ?? 0;
|
||||
if (explicitSearchTool) {
|
||||
if (collectionNames.length > 1) {
|
||||
return await this.commands.searchAcrossCollections({
|
||||
tool: explicitSearchTool,
|
||||
searchCommand: qmdSearchCommand,
|
||||
explicitToolOverride: true,
|
||||
query: trimmed,
|
||||
limit,
|
||||
minScore,
|
||||
collectionNames,
|
||||
signal: searchSignal,
|
||||
reportCommandPhase,
|
||||
});
|
||||
}
|
||||
return await this.commands.searchViaMcporter({
|
||||
mcporter: this.qmd.mcporter,
|
||||
tool: explicitSearchTool,
|
||||
searchCommand: qmdSearchCommand,
|
||||
explicitToolOverride: true,
|
||||
query: trimmed,
|
||||
limit,
|
||||
minScore,
|
||||
collection: collectionNames[0],
|
||||
timeoutMs: this.qmd.limits.timeoutMs,
|
||||
signal: searchSignal,
|
||||
reportCommandPhase,
|
||||
});
|
||||
}
|
||||
const tool = this.commands.resolveMcpTool(qmdSearchCommand);
|
||||
if (collectionNames.length > 1) {
|
||||
return await this.commands.searchAcrossCollections({
|
||||
tool,
|
||||
searchCommand: qmdSearchCommand,
|
||||
explicitToolOverride: false,
|
||||
query: trimmed,
|
||||
limit,
|
||||
minScore,
|
||||
collectionNames,
|
||||
signal: searchSignal,
|
||||
reportCommandPhase,
|
||||
});
|
||||
}
|
||||
return await this.commands.searchViaMcporter({
|
||||
mcporter: this.qmd.mcporter,
|
||||
tool,
|
||||
searchCommand: qmdSearchCommand,
|
||||
explicitToolOverride: false,
|
||||
query: trimmed,
|
||||
limit,
|
||||
minScore,
|
||||
collection: collectionNames[0],
|
||||
timeoutMs: this.qmd.limits.timeoutMs,
|
||||
signal: searchSignal,
|
||||
reportCommandPhase,
|
||||
});
|
||||
}
|
||||
const collectionGroups = await this.resolveCollectionSearchGroups(
|
||||
collectionNames,
|
||||
searchSignal,
|
||||
debugContext,
|
||||
);
|
||||
this.recordSearchPlanDebug({
|
||||
debugContext,
|
||||
command: qmdSearchCommand,
|
||||
collectionNames,
|
||||
collectionGroups,
|
||||
});
|
||||
attemptedCombinedCollectionFilter = collectionGroups.some((group) => group.length > 1);
|
||||
if (collectionGroups.length > 1) {
|
||||
return await this.runQueryAcrossCollectionGroups(
|
||||
trimmed,
|
||||
limit,
|
||||
collectionGroups,
|
||||
qmdSearchCommand,
|
||||
searchSignal,
|
||||
reportCommandPhase,
|
||||
);
|
||||
}
|
||||
const args = this.buildSearchArgs(qmdSearchCommand, trimmed, limit);
|
||||
args.push(...this.buildCollectionFilterArgs(collectionGroups[0] ?? collectionNames));
|
||||
return await this.runQmdSearch(args, qmdSearchCommand, searchSignal, reportCommandPhase);
|
||||
} catch (err) {
|
||||
if (allowMissingCollectionRepair && isMissingCollectionSearchError(err)) {
|
||||
throw err;
|
||||
}
|
||||
if (
|
||||
!mcporterEnabled &&
|
||||
qmdSearchCommand !== "query" &&
|
||||
this.isUnsupportedQmdOptionError(err)
|
||||
) {
|
||||
if (attemptedCombinedCollectionFilter) {
|
||||
await this.markQmdMultiCollectionFiltersUnsupported(debugContext);
|
||||
}
|
||||
effectiveSearchMode = "query";
|
||||
searchFallbackReason = "unsupported-search-flags";
|
||||
qmdManagerLog.warn(
|
||||
`qmd ${qmdSearchCommand} does not support configured flags; retrying search with qmd query`,
|
||||
);
|
||||
try {
|
||||
const collectionGroups = await this.resolveCollectionSearchGroups(
|
||||
collectionNames,
|
||||
searchSignal,
|
||||
debugContext,
|
||||
);
|
||||
this.recordSearchPlanDebug({
|
||||
debugContext,
|
||||
command: "query",
|
||||
collectionNames,
|
||||
collectionGroups,
|
||||
});
|
||||
if (collectionGroups.length > 1) {
|
||||
return await this.runQueryAcrossCollectionGroups(
|
||||
trimmed,
|
||||
limit,
|
||||
collectionGroups,
|
||||
"query",
|
||||
searchSignal,
|
||||
reportCommandPhase,
|
||||
);
|
||||
}
|
||||
const fallbackArgs = this.buildSearchArgs("query", trimmed, limit);
|
||||
fallbackArgs.push(
|
||||
...this.buildCollectionFilterArgs(collectionGroups[0] ?? collectionNames),
|
||||
);
|
||||
return await this.runQmdSearch(fallbackArgs, "query", searchSignal, reportCommandPhase);
|
||||
} catch (fallbackErr) {
|
||||
qmdManagerLog.warn(`qmd query fallback failed: ${String(fallbackErr)}`);
|
||||
throw fallbackErr instanceof Error ? fallbackErr : new Error(String(fallbackErr));
|
||||
}
|
||||
}
|
||||
const label = mcporterEnabled ? "mcporter/qmd" : `qmd ${qmdSearchCommand}`;
|
||||
qmdManagerLog.warn(`${label} failed: ${String(err)}`);
|
||||
throw err instanceof Error ? err : new Error(String(err));
|
||||
}
|
||||
};
|
||||
|
||||
let parsed: QmdQueryResult[];
|
||||
try {
|
||||
parsed = await runSearchAttempt(true);
|
||||
} catch (err) {
|
||||
if (!(await this.tryRepairMissingCollectionSearch(err, debugContext, searchSignal))) {
|
||||
throw err instanceof Error ? err : new Error(String(err));
|
||||
}
|
||||
parsed = await runSearchAttempt(false);
|
||||
}
|
||||
const results: MemorySearchResult[] = [];
|
||||
for (const entry of parsed) {
|
||||
const docHints = this.normalizeDocHints({
|
||||
preferredCollection: entry.collection,
|
||||
preferredFile: entry.file,
|
||||
});
|
||||
const doc = await this.resolveDocLocation(entry.docid, docHints);
|
||||
if (!doc) {
|
||||
continue;
|
||||
}
|
||||
const snippet = truncateUtf16Safe(entry.snippet ?? "", this.qmd.limits.maxSnippetChars);
|
||||
const lines = this.resolveSnippetLines(entry, snippet);
|
||||
const score = typeof entry.score === "number" ? entry.score : 0;
|
||||
const minScore = opts?.minScore ?? 0;
|
||||
if (score < minScore) {
|
||||
continue;
|
||||
}
|
||||
const result = {
|
||||
path: doc.rel,
|
||||
startLine: lines.startLine,
|
||||
endLine: lines.endLine,
|
||||
score,
|
||||
snippet,
|
||||
source: doc.source,
|
||||
} satisfies MemorySearchResult;
|
||||
const artifactIdentity =
|
||||
doc.source === "sessions"
|
||||
? resolveQmdSessionArtifactIdentity({
|
||||
artifactPath: doc.collectionRelativePath,
|
||||
collection: doc.collection,
|
||||
docid: entry.docid?.trim() || undefined,
|
||||
indexPath: this.indexPath,
|
||||
searchPath: doc.rel,
|
||||
})
|
||||
: null;
|
||||
results.push(
|
||||
artifactIdentity ? attachQmdSessionArtifactHit(result, artifactIdentity) : result,
|
||||
);
|
||||
}
|
||||
opts?.onDebug?.({
|
||||
backend: "qmd",
|
||||
configuredMode: qmdSearchCommand,
|
||||
effectiveMode: effectiveSearchMode,
|
||||
fallback: searchFallbackReason,
|
||||
qmd: this.consumeQmdRuntimeDebug(debugContext),
|
||||
});
|
||||
let ranked = results;
|
||||
if (opts?.sources?.length) {
|
||||
const allow = new Set(opts.sources);
|
||||
ranked = results.filter((result) => allow.has(result.source));
|
||||
}
|
||||
return this.clampResultsByInjectedChars(this.diversifyResultsBySource(ranked, resultLimit));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,441 @@
|
||||
import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime";
|
||||
import type { MemorySyncParams } from "openclaw/plugin-sdk/memory-core-host-engine-storage";
|
||||
import { resolveExpiresAtMsFromDurationMs } from "openclaw/plugin-sdk/number-runtime";
|
||||
import {
|
||||
PluginStateLeaseError,
|
||||
type PluginStateLeaseContext,
|
||||
} from "openclaw/plugin-sdk/plugin-state-runtime";
|
||||
import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime";
|
||||
import { asQmdAbortError, isSqliteBusyError } from "./qmd-command-errors.js";
|
||||
import { QmdManagerBase, qmdManagerLog } from "./qmd-manager-base.js";
|
||||
import {
|
||||
getQmdEmbedQueueState,
|
||||
getQmdUpdateQueueState,
|
||||
isEmbedBackoffActive,
|
||||
qmdUsesVectors,
|
||||
QMD_EMBED_BACKOFF_BASE_MS,
|
||||
QMD_EMBED_BACKOFF_MAX_MS,
|
||||
resolveQmdEmbedLeaseOptions,
|
||||
resolveQmdStoreWriteLeaseOptions,
|
||||
resolveStableJitterMs,
|
||||
} from "./qmd-manager-helpers.js";
|
||||
|
||||
export abstract class QmdManagerSync extends QmdManagerBase {
|
||||
async sync(params?: MemorySyncParams): Promise<void> {
|
||||
if (
|
||||
params?.sessions?.some((session) => session.sessionId.trim().length > 0) ||
|
||||
params?.archiveFiles?.some((sessionFile) => sessionFile.trim().length > 0)
|
||||
) {
|
||||
qmdManagerLog.debug("qmd sync ignoring targeted session hint; running regular update");
|
||||
}
|
||||
if (params?.progress) {
|
||||
params.progress({ completed: 0, total: 1, label: "Updating QMD index…" });
|
||||
}
|
||||
await this.runUpdate(params?.reason ?? "manual", params?.force);
|
||||
if (params?.progress) {
|
||||
params.progress({ completed: 1, total: 1, label: "QMD index updated" });
|
||||
}
|
||||
}
|
||||
|
||||
protected async runUpdate(
|
||||
reason: string,
|
||||
force?: boolean,
|
||||
opts?: { fromForcedQueue?: boolean },
|
||||
): Promise<void> {
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
if (this.pendingUpdate) {
|
||||
if (force) {
|
||||
return this.enqueueForcedUpdate(reason);
|
||||
}
|
||||
return this.pendingUpdate;
|
||||
}
|
||||
if (this.queuedForcedUpdate && !opts?.fromForcedQueue) {
|
||||
if (force) {
|
||||
return this.enqueueForcedUpdate(reason);
|
||||
}
|
||||
return this.queuedForcedUpdate;
|
||||
}
|
||||
if (this.shouldSkipUpdate(force)) {
|
||||
return;
|
||||
}
|
||||
const run = async () => {
|
||||
const startTime = Date.now();
|
||||
let updatePublished = false;
|
||||
qmdManagerLog.debug(
|
||||
`qmd sync started for agent "${this.agentId}" reason=${reason} force=${force === true}`,
|
||||
);
|
||||
try {
|
||||
await this.withQmdUpdateQueue(async (lease) => {
|
||||
const { signal } = lease;
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
if (this.sessionExporter) {
|
||||
await this.exportSessions(lease);
|
||||
this.throwIfAborted(signal);
|
||||
}
|
||||
await this.runQmdUpdateWithRetry(reason, lease);
|
||||
updatePublished = true;
|
||||
if (this.sessionExporter) {
|
||||
this.throwIfAborted(signal);
|
||||
this.refreshSessionArtifactDocIds(lease);
|
||||
}
|
||||
});
|
||||
} catch (err) {
|
||||
if (err instanceof PluginStateLeaseError && this.shouldPreserveLeaseRetry(err)) {
|
||||
this.dirty = true;
|
||||
if (updatePublished && qmdUsesVectors(this.qmd.searchMode)) {
|
||||
this.embedLeaseRetryPending = true;
|
||||
}
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
this.dirty = false;
|
||||
if (this.shouldRunEmbed(force)) {
|
||||
try {
|
||||
// Wait for embed capacity before taking the per-agent write lease. The
|
||||
// lease should protect active qmd writes only, not time spent queued
|
||||
// behind unrelated agents' embeds.
|
||||
const embedded = await this.withQmdEmbedQueue(async () => {
|
||||
await this.withQmdGlobalEmbedLease((globalLease) =>
|
||||
this.withQmdStoreWriteLease(async (lease) => {
|
||||
globalLease.assertOwned();
|
||||
lease.assertOwned();
|
||||
await this.runQmd(["embed"], {
|
||||
timeoutMs: this.qmd.update.embedTimeoutMs,
|
||||
discardOutput: true,
|
||||
signal: lease.signal,
|
||||
});
|
||||
}, globalLease.signal),
|
||||
);
|
||||
});
|
||||
if (!embedded) {
|
||||
return;
|
||||
}
|
||||
this.lastEmbedAt = Date.now();
|
||||
this.embedLeaseRetryPending = false;
|
||||
this.embedBackoffUntil = null;
|
||||
this.embedFailureCount = 0;
|
||||
} catch (err) {
|
||||
if (err instanceof PluginStateLeaseError) {
|
||||
if (this.shouldPreserveLeaseRetry(err)) {
|
||||
// The update already published documents. Keep both the dirty-sync
|
||||
// trigger and embed intent so contention cannot strand them unembedded.
|
||||
this.dirty = true;
|
||||
this.embedLeaseRetryPending = true;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
this.noteEmbedFailure(reason, err);
|
||||
}
|
||||
}
|
||||
if (this.closed) {
|
||||
return;
|
||||
}
|
||||
this.lastUpdateAt = Date.now();
|
||||
this.documentResolver.clearCache();
|
||||
qmdManagerLog.info(
|
||||
`qmd sync completed for agent "${this.agentId}" reason=${reason} durationMs=${Date.now() - startTime}`,
|
||||
);
|
||||
};
|
||||
this.pendingUpdate = run().finally(() => {
|
||||
this.pendingUpdate = null;
|
||||
});
|
||||
await this.pendingUpdate;
|
||||
}
|
||||
|
||||
protected async runQmdUpdateWithRetry(
|
||||
reason: string,
|
||||
lease: PluginStateLeaseContext,
|
||||
): Promise<void> {
|
||||
const { signal } = lease;
|
||||
const isBootRun = reason === "boot" || reason.startsWith("boot:");
|
||||
const maxAttempts = isBootRun ? 3 : 1;
|
||||
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
|
||||
try {
|
||||
await this.runQmdUpdateOnce(reason, lease);
|
||||
return;
|
||||
} catch (err) {
|
||||
if (attempt >= maxAttempts || !this.isRetryableUpdateError(err)) {
|
||||
throw err;
|
||||
}
|
||||
const delayMs = 500 * 2 ** (attempt - 1);
|
||||
qmdManagerLog.warn(
|
||||
`qmd update retry ${attempt}/${maxAttempts - 1} after failure (${reason}): ${String(err)}`,
|
||||
);
|
||||
await this.waitForRetryDelay(delayMs, signal);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
protected async runQmdUpdateOnce(reason: string, lease: PluginStateLeaseContext): Promise<void> {
|
||||
const { signal } = lease;
|
||||
try {
|
||||
lease.assertOwned();
|
||||
await this.runQmd(["update"], {
|
||||
timeoutMs: this.qmd.update.updateTimeoutMs,
|
||||
discardOutput: true,
|
||||
signal,
|
||||
});
|
||||
} catch (err) {
|
||||
if (
|
||||
!(await this.tryRepairNullByteCollections(err, reason, lease)) &&
|
||||
!(await this.tryRepairDuplicateDocumentConstraint(err, reason, lease))
|
||||
) {
|
||||
throw err;
|
||||
}
|
||||
lease.assertOwned();
|
||||
await this.runQmd(["update"], {
|
||||
timeoutMs: this.qmd.update.updateTimeoutMs,
|
||||
discardOutput: true,
|
||||
signal,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
protected isRetryableUpdateError(err: unknown): boolean {
|
||||
if (isSqliteBusyError(err)) {
|
||||
return true;
|
||||
}
|
||||
const message = formatErrorMessage(err);
|
||||
const normalized = normalizeLowercaseStringOrEmpty(message);
|
||||
return normalized.includes("timed out");
|
||||
}
|
||||
|
||||
protected throwIfAborted(signal: AbortSignal): void {
|
||||
if (signal.aborted) {
|
||||
throw asQmdAbortError(signal);
|
||||
}
|
||||
}
|
||||
|
||||
protected async waitForRetryDelay(delayMs: number, signal: AbortSignal): Promise<void> {
|
||||
this.throwIfAborted(signal);
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onAbort = () => {
|
||||
clearTimeout(timeout);
|
||||
reject(asQmdAbortError(signal));
|
||||
};
|
||||
const timeout = setTimeout(() => {
|
||||
signal.removeEventListener("abort", onAbort);
|
||||
resolve();
|
||||
}, delayMs);
|
||||
signal.addEventListener("abort", onAbort, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
protected shouldRunEmbed(force?: boolean): boolean {
|
||||
if (!qmdUsesVectors(this.qmd.searchMode)) {
|
||||
return false;
|
||||
}
|
||||
const now = Date.now();
|
||||
if (isEmbedBackoffActive(this.embedBackoffUntil)) {
|
||||
return false;
|
||||
}
|
||||
const embedIntervalMs = this.qmd.update.embedIntervalMs;
|
||||
return (
|
||||
this.embedLeaseRetryPending ||
|
||||
Boolean(force) ||
|
||||
this.lastEmbedAt === null ||
|
||||
(embedIntervalMs > 0 && now - this.lastEmbedAt > embedIntervalMs)
|
||||
);
|
||||
}
|
||||
|
||||
protected shouldPreserveLeaseRetry(err: PluginStateLeaseError): boolean {
|
||||
return (
|
||||
!this.closed &&
|
||||
err.code !== "PLUGIN_STATE_LEASE_ABORTED" &&
|
||||
err.code !== "PLUGIN_STATE_LEASE_INVALID_INPUT"
|
||||
);
|
||||
}
|
||||
|
||||
protected shouldScheduleEmbedTimer(): boolean {
|
||||
if (!qmdUsesVectors(this.qmd.searchMode)) {
|
||||
return false;
|
||||
}
|
||||
const embedIntervalMs = this.qmd.update.embedIntervalMs;
|
||||
if (embedIntervalMs <= 0) {
|
||||
return false;
|
||||
}
|
||||
const updateIntervalMs = this.qmd.update.intervalMs;
|
||||
return updateIntervalMs <= 0 || updateIntervalMs > embedIntervalMs;
|
||||
}
|
||||
|
||||
protected resolveEmbedStartupJitterMs(): number {
|
||||
const windowMs = this.qmd.update.embedIntervalMs;
|
||||
if (windowMs <= 0) {
|
||||
return 0;
|
||||
}
|
||||
const customCollections = this.qmd.collections
|
||||
.filter((collection) => collection.kind === "custom")
|
||||
.map((collection) => `${collection.path}\u0000${collection.pattern}`)
|
||||
.toSorted()
|
||||
.join("\u0001");
|
||||
if (!customCollections) {
|
||||
return 0;
|
||||
}
|
||||
return resolveStableJitterMs({
|
||||
seed: `${this.agentId}:${customCollections}`,
|
||||
windowMs,
|
||||
});
|
||||
}
|
||||
|
||||
protected async withQmdEmbedQueue(task: () => Promise<void>): Promise<boolean> {
|
||||
const queue = getQmdEmbedQueueState();
|
||||
const previous = queue.tail;
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
queue.tail = previous.then(
|
||||
() => current,
|
||||
() => current,
|
||||
);
|
||||
try {
|
||||
const waitResult = await Promise.race([
|
||||
previous.then(
|
||||
() => "ready" as const,
|
||||
() => "ready" as const,
|
||||
),
|
||||
this.closeSignal.then(() => "closed" as const),
|
||||
]);
|
||||
if (waitResult === "closed") {
|
||||
return false;
|
||||
}
|
||||
await task();
|
||||
return true;
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
}
|
||||
}
|
||||
|
||||
protected async withQmdGlobalEmbedLease<T>(
|
||||
task: (lease: PluginStateLeaseContext) => Promise<T>,
|
||||
): Promise<T> {
|
||||
return await this.withLease(
|
||||
{
|
||||
namespace: "qmd",
|
||||
key: "embed",
|
||||
database: { scope: "shared" },
|
||||
...resolveQmdEmbedLeaseOptions(this.qmd.update.embedTimeoutMs),
|
||||
signal: this.closeAbortController.signal,
|
||||
},
|
||||
async (lease) => await task(lease),
|
||||
);
|
||||
}
|
||||
|
||||
protected async withQmdStoreWriteLease<T>(
|
||||
task: (lease: PluginStateLeaseContext) => Promise<T>,
|
||||
parentSignal?: AbortSignal,
|
||||
): Promise<T> {
|
||||
// SQLite is the sole coordinator; never dual-lock sidecars. One per-agent
|
||||
// lease guards every write to index.sqlite while agents remain parallel.
|
||||
return await this.withLease(
|
||||
{
|
||||
namespace: "qmd",
|
||||
key: "write",
|
||||
database: { scope: "agent", agentId: this.agentId },
|
||||
...resolveQmdStoreWriteLeaseOptions(
|
||||
this.qmd.update.updateTimeoutMs,
|
||||
this.qmd.update.embedTimeoutMs,
|
||||
),
|
||||
signal: parentSignal
|
||||
? AbortSignal.any([this.closeAbortController.signal, parentSignal])
|
||||
: this.closeAbortController.signal,
|
||||
},
|
||||
async (lease) => await task(lease),
|
||||
);
|
||||
}
|
||||
|
||||
protected async withQmdUpdateQueue<T>(
|
||||
task: (lease: PluginStateLeaseContext) => Promise<T>,
|
||||
): Promise<T> {
|
||||
const queue = getQmdUpdateQueueState();
|
||||
const key = this.qmdDir;
|
||||
const previous = queue.tails.get(key) ?? Promise.resolve();
|
||||
let releaseCurrent!: () => void;
|
||||
const current = new Promise<void>((resolve) => {
|
||||
releaseCurrent = resolve;
|
||||
});
|
||||
const next = previous.then(
|
||||
() => current,
|
||||
() => current,
|
||||
);
|
||||
queue.tails.set(key, next);
|
||||
try {
|
||||
const waitResult = await Promise.race([
|
||||
previous.then(
|
||||
() => "ready" as const,
|
||||
() => "ready" as const,
|
||||
),
|
||||
this.closeSignal.then(() => "closed" as const),
|
||||
]);
|
||||
if (waitResult === "closed") {
|
||||
return undefined as T;
|
||||
}
|
||||
// The in-process queue is keyed per store; the per-agent write lease also
|
||||
// serializes separate processes and embed writes targeting this index.
|
||||
return await this.withQmdStoreWriteLease(task);
|
||||
} finally {
|
||||
releaseCurrent();
|
||||
void next.finally(() => {
|
||||
if (queue.tails.get(key) === next) {
|
||||
queue.tails.delete(key);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
protected noteEmbedFailure(reason: string, err: unknown): void {
|
||||
this.embedFailureCount += 1;
|
||||
const delayMs = Math.min(
|
||||
QMD_EMBED_BACKOFF_MAX_MS,
|
||||
QMD_EMBED_BACKOFF_BASE_MS * 2 ** Math.max(0, this.embedFailureCount - 1),
|
||||
);
|
||||
this.embedBackoffUntil = resolveExpiresAtMsFromDurationMs(delayMs) ?? null;
|
||||
qmdManagerLog.warn(
|
||||
`qmd embed failed (${reason}): ${String(err)}; backing off for ${Math.ceil(delayMs / 1000)}s`,
|
||||
);
|
||||
}
|
||||
|
||||
protected enqueueForcedUpdate(reason: string): Promise<void> {
|
||||
this.queuedForcedRuns += 1;
|
||||
if (!this.queuedForcedUpdate) {
|
||||
this.queuedForcedUpdate = this.drainForcedUpdates(reason).finally(() => {
|
||||
this.queuedForcedUpdate = null;
|
||||
});
|
||||
}
|
||||
return this.queuedForcedUpdate;
|
||||
}
|
||||
|
||||
protected async drainForcedUpdates(reason: string): Promise<void> {
|
||||
await this.pendingUpdate?.catch(() => undefined);
|
||||
while (!this.closed && this.queuedForcedRuns > 0) {
|
||||
this.queuedForcedRuns -= 1;
|
||||
await this.runUpdate(`${reason}:queued`, true, { fromForcedQueue: true });
|
||||
}
|
||||
}
|
||||
|
||||
protected shouldSkipUpdate(force?: boolean): boolean {
|
||||
if (force) {
|
||||
return false;
|
||||
}
|
||||
const debounceMs = this.qmd.update.debounceMs;
|
||||
if (debounceMs <= 0 || !this.lastUpdateAt) {
|
||||
return false;
|
||||
}
|
||||
return Date.now() - this.lastUpdateAt < debounceMs;
|
||||
}
|
||||
|
||||
protected async exportSessions(lease: PluginStateLeaseContext): Promise<void> {
|
||||
await this.sessionExporter?.exportSessions(lease);
|
||||
}
|
||||
|
||||
protected refreshSessionArtifactDocIds(lease: PluginStateLeaseContext): void {
|
||||
this.sessionExporter?.refreshArtifactDocIds(lease);
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user