Files
openclaw/extensions/memory-lancedb/memory-cli.ts
yt2102 6bd8e0387c fix(memory): close previous embedding provider before replacement (#113471)
* fix(memory): close previous embedding provider before replacement

* fix(memory): increase embedding worker close grace period for slow hosts

* fix(memory): clear this.provider after close in resetProviderInitializationForRetry

Co-authored-by: Sanjay Santhanam <notifications@github.com>

* fix(memory): serialize embedding provider replacement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): unify provider transition lifecycle

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): join worker shutdown lifecycle

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): block worker restart during close

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retain failed provider retirements

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): drain retirements on manager close

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): make worker close joinable

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): preserve sync before provider retirement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): separate worker exit from disposal errors

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): avoid shutdown admission gap

Co-authored-by: yt2102 <yt2102@qq.com>

* style(memory): format provider lifecycle fix

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): bound embedding worker termination

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): fail closed on fallback initialization errors

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retry primary after fallback creation failure

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retain disconnected embedding workers until exit

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): isolate shared fallback transition failures

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): serialize scoped manager retirement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retain failed global manager closes

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): serialize manager admission with teardown

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): serialize outer manager lifecycle

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): close request-scoped embedding providers

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retire qmd managers before replacement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): support synchronous embedding cleanup

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): isolate manager lifecycle scopes

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(cli): close request-scoped embedding providers

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): canonicalize manager lifecycle ownership

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retry primary after null fallback result

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): retain failed embedding provider closes

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): lease closable embedding providers

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): drain retained embedding providers

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): join lazy fallback teardown

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): drain embedding providers after HTTP close

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retain failed qmd candidate cleanup

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): aggregate retained qmd teardown failures

Co-authored-by: yt2102 <yt2102@qq.com>

* refactor(memory): keep qmd lifecycle policy unchanged

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retain failed worker construction clients

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(gateway): scope local embedding retirement by provider

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): drain provider generations before retirement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): drain admitted operations before teardown

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): preserve provider identity through vector search

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): lease provider generation through embedding

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): lease provider through index publication

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): keep sync harness hooks optional

Co-authored-by: yt2102 <yt2102@qq.com>

* refactor(memory): own generations in sync lifecycle

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): preserve query runtime across retirement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): pin FTS-only sync generations

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retry failed worker construction cleanup

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): drain admitted searches before closing

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): close manager and gateway admission races

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): serialize qmd wrapper replacement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): serialize failed qmd retirement

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): satisfy qmd lifecycle type and deadcode gates

Co-authored-by: yt2102 <yt2102@qq.com>

* test(memory): type qmd lifecycle doubles

Co-authored-by: yt2102 <yt2102@qq.com>

* style(gateway): clarify created embedding provider

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): drain availability probes before close

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): preserve cleanup ownership without blocking fallback

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retire lancedb embedding providers

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): preserve generic provider cleanup receiver

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): close lancedb CLI embeddings

Co-authored-by: yt2102 <yt2102@qq.com>

* test(memory): normalize abort rejection reason

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): retain lancedb provider retirements

Co-authored-by: yt2102 <yt2102@qq.com>

* fix(memory): drain lancedb embedding uses

Co-authored-by: yt2102 <yt2102@qq.com>

---------

Co-authored-by: Sanjay Santhanam <notifications@github.com>
Co-authored-by: Peter Steinberger <steipete@gmail.com>
2026-07-26 03:00:22 -04:00

229 lines
8.1 KiB
TypeScript

import { parseStrictPositiveInteger } from "openclaw/plugin-sdk/number-runtime";
import type { OpenClawPluginApi } from "./api.js";
import type { Embeddings } from "./embeddings.js";
import {
MEMORY_QUERY_COLUMNS,
type MemoryQueryColumn,
type MemoryQueryFilter,
type MemoryDB,
} from "./lancedb-store.js";
import { normalizeRecallQuery } from "./memory-policy.js";
function parsePositiveIntegerOption(value: string | undefined, flag: string): number | undefined {
if (value === undefined) {
return undefined;
}
const parsed = parseStrictPositiveInteger(value);
if (parsed === undefined) {
throw new Error(`${flag} must be a positive integer`);
}
return parsed;
}
function parseMemoryCliColumns(value: unknown): MemoryQueryColumn[] {
if (typeof value !== "string") {
return [...MEMORY_QUERY_COLUMNS];
}
const columns = value.split(",").map((column) => column.trim());
const invalid = columns.filter(
(column): column is string =>
!MEMORY_QUERY_COLUMNS.includes(column as (typeof MEMORY_QUERY_COLUMNS)[number]),
);
if (invalid.length > 0) {
throw new Error(`Unsupported memory columns: ${invalid.join(", ")}`);
}
return columns as MemoryQueryColumn[];
}
function parseMemoryCliOrder(value: unknown): {
column: MemoryQueryColumn;
direction: 1 | -1;
} | null {
if (typeof value !== "string" || !value.trim()) {
return null;
}
const [column, direction = "asc", extra] = value.split(":");
if (
extra !== undefined ||
!MEMORY_QUERY_COLUMNS.includes(column as MemoryQueryColumn) ||
!["asc", "desc"].includes(direction.toLowerCase())
) {
throw new Error("--order-by must be <id|text|importance|category|createdAt>:<asc|desc>");
}
return {
column: column as MemoryQueryColumn,
direction: direction.toLowerCase() === "desc" ? -1 : 1,
};
}
export function parseMemoryCliFilter(rawValue: unknown): MemoryQueryFilter | undefined {
if (rawValue === undefined) {
return undefined;
}
if (typeof rawValue !== "string") {
throw new Error("--filter must be a string");
}
const filter = rawValue.trim();
if (filter.length > 200) {
throw new Error("Filter condition exceeds maximum length of 200 characters");
}
const match =
/^(id|text|importance|category|createdAt)\s*(=|!=|<>|<=|>=|<|>|LIKE)\s*(?:'((?:''|[^'])*)'|(-?(?:\d+(?:\.\d+)?|\.\d+)))$/i.exec(
filter,
);
if (!match) {
throw new Error(
"--filter must be one comparison using id, text, importance, category, or createdAt",
);
}
const rawColumn = match[1]!;
const rawOperator = match[2]!;
const rawString = match[3];
const rawNumber = match[4];
const column = MEMORY_QUERY_COLUMNS.find(
(candidate) => candidate.toLowerCase() === rawColumn.toLowerCase(),
);
if (!column) {
throw new Error(`Unsupported memory filter column: ${rawColumn}`);
}
const operator = rawOperator.toUpperCase() as MemoryQueryFilter["operator"];
const value = rawString !== undefined ? rawString.replaceAll("''", "'") : Number(rawNumber);
if (typeof value === "number" && !Number.isFinite(value)) {
throw new Error("--filter numeric value must be finite");
}
const expectsNumber = column === "importance" || column === "createdAt";
if (expectsNumber !== (typeof value === "number")) {
throw new Error(`--filter ${column} requires a ${expectsNumber ? "number" : "quoted string"}`);
}
if (operator === "LIKE" && typeof value !== "string") {
throw new Error("--filter LIKE requires a quoted string");
}
return { column, operator, value };
}
export function registerMemoryCli(
api: OpenClawPluginApi,
db: MemoryDB,
embeddings: Embeddings,
resolveCliAgentId: (rawAgentId: unknown) => string,
recallMaxChars: number | undefined,
): void {
api.registerCli(
({ program }) => {
const memory = program.command("ltm").description("LanceDB memory plugin commands");
memory
.command("list")
.description("List memories")
.option("--agent <id>", "Agent id (default: configured default agent)")
.option("--limit <n>", "Max results")
.option("--order-by-created-at", "Order memories by createdAt descending", false)
.action(async (opts) => {
const agentId = resolveCliAgentId(opts.agent);
const limit = parsePositiveIntegerOption(opts.limit, "--limit");
const entries = await db.list(agentId, limit, {
orderByCreatedAt: Boolean(opts.orderByCreatedAt),
});
console.log(JSON.stringify(entries, null, 2));
});
memory
.command("search")
.description("Search memories")
.argument("<query>", "Search query")
.option("--agent <id>", "Agent id (default: configured default agent)")
.option("--limit <n>", "Max results", "5")
.action(async (query, opts) => {
let operationError: unknown;
let operationFailed = false;
try {
const agentId = resolveCliAgentId(opts.agent);
const vector = await embeddings.embed(normalizeRecallQuery(query, recallMaxChars));
const limit = parsePositiveIntegerOption(opts.limit, "--limit");
const results = await db.search(agentId, vector, limit, 0.3);
const output = results.map((r) => ({
id: r.entry.id,
text: r.entry.text,
category: r.entry.category,
importance: r.entry.importance,
score: r.score,
}));
console.log(JSON.stringify(output, null, 2));
} catch (err) {
operationError = err;
operationFailed = true;
}
let closeError: unknown;
let closeFailed = false;
try {
await embeddings.close?.();
} catch (err) {
closeError = err;
closeFailed = true;
}
if (operationFailed) {
throw operationError;
}
if (closeFailed) {
throw closeError;
}
});
memory
.command("query")
.description("Query memories (non-vector search)")
.option("--agent <id>", "Agent id (default: configured default agent)")
.option("--cols <columns>", "Columns to select, comma-separated")
.option("--filter <condition>", "Filter condition")
.option("--limit <n>", "Limit number of results", "10")
.option("--order-by <order>", "Order by column and direction (e.g., createdAt:desc)")
.action(async (opts) => {
const agentId = resolveCliAgentId(opts.agent);
const outputColumns = parseMemoryCliColumns(opts.cols);
const order = parseMemoryCliOrder(opts.orderBy);
const selectedColumns = [...outputColumns];
if (order && !selectedColumns.includes(order.column)) {
selectedColumns.push(order.column);
}
const limit = parsePositiveIntegerOption(opts.limit, "--limit") ?? 10;
let rows = await db.query(agentId, {
columns: selectedColumns,
filter: parseMemoryCliFilter(opts.filter),
...(order ? {} : { limit }),
});
if (order) {
rows.sort((a, b) => {
const aValue = a[order.column] as number | string;
const bValue = b[order.column] as number | string;
if (aValue < bValue) {
return -1 * order.direction;
}
if (aValue > bValue) {
return order.direction;
}
return 0;
});
rows = rows.slice(0, limit);
if (!outputColumns.includes(order.column)) {
for (const row of rows) {
delete row[order.column];
}
}
}
console.log(JSON.stringify(rows, null, 2));
});
memory
.command("stats")
.description("Show memory statistics")
.option("--agent <id>", "Agent id (default: configured default agent)")
.action(async (opts) => {
const agentId = resolveCliAgentId(opts.agent);
const count = await db.count(agentId);
console.log(`Total memories: ${count}`);
});
},
{ commands: ["ltm"] },
);
}