吸收数字员工长期记忆闭环
This commit is contained in:
baiyanyun
@@ -182,6 +182,26 @@ const DIGITAL_EMPLOYEE_SCHEMA_SQL = `
|
||||
updated_at TEXT NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS digital_memories (
|
||||
id TEXT PRIMARY KEY,
|
||||
remote_id TEXT,
|
||||
key TEXT NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
category TEXT NOT NULL DEFAULT 'fact',
|
||||
source TEXT NOT NULL DEFAULT 'qimingclaw',
|
||||
source_path TEXT,
|
||||
session_id TEXT,
|
||||
score REAL,
|
||||
status TEXT NOT NULL DEFAULT 'active',
|
||||
payload TEXT NOT NULL DEFAULT '{}',
|
||||
sync_status TEXT NOT NULL DEFAULT 'pending',
|
||||
last_synced_at TEXT,
|
||||
sync_error TEXT,
|
||||
created_at TEXT NOT NULL,
|
||||
updated_at TEXT NOT NULL,
|
||||
deleted_at TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS digital_sync_outbox (
|
||||
id TEXT PRIMARY KEY,
|
||||
entity_type TEXT NOT NULL,
|
||||
@@ -222,6 +242,9 @@ const DIGITAL_EMPLOYEE_SCHEMA_SQL = `
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_tasks_plan ON digital_tasks(plan_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_runs_task ON digital_runs(task_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_events_plan ON digital_events(plan_id, occurred_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_memories_key ON digital_memories(key, status, updated_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_memories_category ON digital_memories(category, status, updated_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_memories_sync ON digital_memories(sync_status);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_sync_outbox_status ON digital_sync_outbox(status, created_at);
|
||||
CREATE INDEX IF NOT EXISTS idx_digital_sync_attempts_outbox ON digital_sync_attempts(outbox_id, finished_at);
|
||||
`;
|
||||
|
||||
@@ -17,8 +17,13 @@ import {
|
||||
saveDigitalEmployeeCronSettings,
|
||||
readDigitalEmployeeScheduleRuns,
|
||||
saveDigitalEmployeeUiState,
|
||||
listDigitalEmployeeMemories,
|
||||
upsertDigitalEmployeeMemory,
|
||||
deleteDigitalEmployeeMemory,
|
||||
syncQimingclawMemoryEntries,
|
||||
type DigitalEmployeeManagedService,
|
||||
type DigitalEmployeeGovernanceCommandRecord,
|
||||
type DigitalEmployeeMemoryUpsertInput,
|
||||
type DigitalEmployeeServiceStatus,
|
||||
type DigitalEmployeeSnapshot,
|
||||
type DigitalEmployeeUiStateUpdate,
|
||||
@@ -33,6 +38,7 @@ import {
|
||||
} from "../services/digitalEmployee/syncService";
|
||||
import { listDigitalEmployeePlanTemplates } from "../services/digitalEmployee/planTemplateService";
|
||||
import { listDigitalEmployeeSkillCapabilities } from "../services/digitalEmployee/skillCatalogService";
|
||||
import { memoryService } from "../services/memory";
|
||||
|
||||
export function registerDigitalEmployeeHandlers(ctx: HandlerContext): void {
|
||||
ipcMain.handle("digitalEmployee:getSnapshot", async () => {
|
||||
@@ -70,6 +76,50 @@ export function registerDigitalEmployeeHandlers(ctx: HandlerContext): void {
|
||||
return listDigitalEmployeePlanTemplates();
|
||||
});
|
||||
|
||||
ipcMain.handle("digitalEmployee:listMemories", async (_, options?: { query?: string; category?: string; limit?: number }) => {
|
||||
if (memoryService.isInitialized()) {
|
||||
syncQimingclawMemoryEntries(memoryService.listMemories({ status: "active", limit: options?.limit ?? 300 }));
|
||||
}
|
||||
return listDigitalEmployeeMemories(options ?? {});
|
||||
});
|
||||
|
||||
ipcMain.handle("digitalEmployee:addMemory", async (_, input: DigitalEmployeeMemoryUpsertInput) => {
|
||||
const content = typeof input?.content === "string" ? input.content : "";
|
||||
const category = typeof input?.category === "string" ? input.category : "fact";
|
||||
const key = typeof input?.key === "string" && input.key.trim() ? input.key.trim() : undefined;
|
||||
let memoryServiceId: string | null = null;
|
||||
if (memoryService.isInitialized()) {
|
||||
memoryServiceId = memoryService.addMemory({
|
||||
id: input.id || key,
|
||||
text: content,
|
||||
category: category as never,
|
||||
source: "core",
|
||||
sourcePath: "MEMORY.md",
|
||||
status: "active",
|
||||
});
|
||||
}
|
||||
return upsertDigitalEmployeeMemory({
|
||||
...input,
|
||||
id: input.id || memoryServiceId || key,
|
||||
key: key || memoryServiceId || input.id,
|
||||
content,
|
||||
category,
|
||||
source: memoryServiceId ? "qimingclaw-memory-service" : "qimingclaw-digital-local",
|
||||
payload: {
|
||||
...(input.payload ?? {}),
|
||||
memoryServiceId,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
ipcMain.handle("digitalEmployee:deleteMemory", async (_, keyOrId: string) => {
|
||||
const deleted = deleteDigitalEmployeeMemory(keyOrId);
|
||||
if (deleted && memoryService.isInitialized()) {
|
||||
memoryService.deleteMemory(deleted.id);
|
||||
}
|
||||
return deleted;
|
||||
});
|
||||
|
||||
ipcMain.handle(
|
||||
"digitalEmployee:saveUiState",
|
||||
async (_, update: DigitalEmployeeUiStateUpdate) => {
|
||||
|
||||
@@ -604,6 +604,100 @@ describe("digital employee state service", () => {
|
||||
kind: "governance_resume_task",
|
||||
});
|
||||
});
|
||||
|
||||
it("upserts, lists, and syncs digital employee memories", async () => {
|
||||
const {
|
||||
listDigitalEmployeeMemories,
|
||||
readDigitalEmployeeRuntimeRecords,
|
||||
syncQimingclawMemoryEntries,
|
||||
upsertDigitalEmployeeMemory,
|
||||
} = await import("./stateService");
|
||||
|
||||
const localMemory = upsertDigitalEmployeeMemory({
|
||||
key: "preferred-report-style",
|
||||
content: "用户偏好日报先给结论再列证据。",
|
||||
category: "preference",
|
||||
source: "qimingclaw-digital-local",
|
||||
score: 0.88,
|
||||
payload: { source: "test" },
|
||||
});
|
||||
|
||||
expect(localMemory).toMatchObject({
|
||||
id: "memory-preferred-report-style",
|
||||
key: "preferred-report-style",
|
||||
category: "preference",
|
||||
status: "active",
|
||||
syncStatus: "pending",
|
||||
});
|
||||
expect(mockState.db?.outbox.get("memory:upsert:memory-preferred-report-style")).toMatchObject({
|
||||
entity_type: "memory",
|
||||
entity_id: "memory-preferred-report-style",
|
||||
operation: "upsert",
|
||||
status: "pending",
|
||||
});
|
||||
|
||||
syncQimingclawMemoryEntries([
|
||||
{
|
||||
id: "mem-from-service",
|
||||
text: "用户希望所有提交信息使用中文。",
|
||||
category: "preference",
|
||||
source: "core",
|
||||
sourcePath: "MEMORY.md",
|
||||
importance: 0.9,
|
||||
status: "active",
|
||||
createdAt: 1780800000000,
|
||||
updatedAt: 1780800300000,
|
||||
},
|
||||
]);
|
||||
|
||||
expect(listDigitalEmployeeMemories({ query: "中文" })).toEqual([
|
||||
expect.objectContaining({
|
||||
id: "mem-from-service",
|
||||
key: "mem-from-service",
|
||||
content: "用户希望所有提交信息使用中文。",
|
||||
}),
|
||||
]);
|
||||
expect(listDigitalEmployeeMemories({ category: "preference" })).toHaveLength(2);
|
||||
expect(readDigitalEmployeeRuntimeRecords().memories).toEqual(expect.arrayContaining([
|
||||
expect.objectContaining({ id: "mem-from-service" }),
|
||||
expect.objectContaining({ id: "memory-preferred-report-style" }),
|
||||
]));
|
||||
});
|
||||
|
||||
it("soft deletes digital employee memories and writes delete outbox", async () => {
|
||||
const {
|
||||
deleteDigitalEmployeeMemory,
|
||||
listDigitalEmployeeMemories,
|
||||
upsertDigitalEmployeeMemory,
|
||||
} = await import("./stateService");
|
||||
|
||||
upsertDigitalEmployeeMemory({
|
||||
id: "memory-delete-me",
|
||||
key: "delete-me",
|
||||
content: "这条记忆将被删除。",
|
||||
category: "fact",
|
||||
});
|
||||
|
||||
const deleted = deleteDigitalEmployeeMemory("delete-me");
|
||||
|
||||
expect(deleted).toMatchObject({
|
||||
id: "memory-delete-me",
|
||||
key: "delete-me",
|
||||
status: "deleted",
|
||||
deletedAt: "2026-06-07T08:00:00.000Z",
|
||||
syncStatus: "pending",
|
||||
});
|
||||
expect(listDigitalEmployeeMemories({ includeDeleted: false })).toEqual([]);
|
||||
expect(listDigitalEmployeeMemories({ includeDeleted: true })).toEqual([
|
||||
expect.objectContaining({ id: "memory-delete-me", status: "deleted" }),
|
||||
]);
|
||||
expect(mockState.db?.outbox.get("memory:delete:memory-delete-me")).toMatchObject({
|
||||
entity_type: "memory",
|
||||
entity_id: "memory-delete-me",
|
||||
operation: "delete",
|
||||
status: "pending",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
function createGovernableTask(options: {
|
||||
@@ -749,6 +843,23 @@ interface TestApprovalRow {
|
||||
updated_at: string;
|
||||
}
|
||||
|
||||
interface TestMemoryRow {
|
||||
id: string;
|
||||
key: string;
|
||||
content: string;
|
||||
category: string;
|
||||
source: string;
|
||||
source_path: string | null;
|
||||
session_id: string | null;
|
||||
score: number | null;
|
||||
status: string;
|
||||
payload: string;
|
||||
sync_status: string;
|
||||
created_at: string;
|
||||
updated_at: string;
|
||||
deleted_at: string | null;
|
||||
}
|
||||
|
||||
interface TestScheduleRow {
|
||||
id: string;
|
||||
plan_id: string | null;
|
||||
@@ -813,6 +924,7 @@ class TestDb {
|
||||
events = new Map<string, TestEventRow>();
|
||||
artifacts = new Map<string, TestArtifactRow>();
|
||||
approvals = new Map<string, TestApprovalRow>();
|
||||
memories = new Map<string, TestMemoryRow>();
|
||||
schedules = new Map<string, TestScheduleRow>();
|
||||
scheduleRuns = new Map<string, TestScheduleRunRow>();
|
||||
outbox = new Map<string, TestOutboxRow>();
|
||||
@@ -844,6 +956,7 @@ class TestDb {
|
||||
if (sql.includes("FROM digital_events")) return Array.from(this.events.values());
|
||||
if (sql.includes("FROM digital_artifacts")) return Array.from(this.artifacts.values());
|
||||
if (sql.includes("FROM digital_approvals")) return Array.from(this.approvals.values());
|
||||
if (sql.includes("FROM digital_memories")) return Array.from(this.memories.values());
|
||||
return [];
|
||||
}
|
||||
|
||||
@@ -882,6 +995,18 @@ class TestDb {
|
||||
if (sql.includes("FROM digital_schedule_runs") && sql.includes("WHERE id = ?")) {
|
||||
return this.scheduleRuns.get(args[0] as string);
|
||||
}
|
||||
if (sql.includes("FROM digital_memories") && sql.includes("WHERE id = ? OR key = ?")) {
|
||||
return Array.from(this.memories.values())
|
||||
.filter((memory) => memory.id === args[0] || memory.key === args[1])
|
||||
.sort((left, right) => {
|
||||
const leftDeleted = left.status === "deleted" ? 1 : 0;
|
||||
const rightDeleted = right.status === "deleted" ? 1 : 0;
|
||||
return leftDeleted - rightDeleted || right.updated_at.localeCompare(left.updated_at) || right.created_at.localeCompare(left.created_at);
|
||||
})[0];
|
||||
}
|
||||
if (sql.includes("FROM digital_memories") && sql.includes("WHERE id = ?")) {
|
||||
return this.memories.get(args[0] as string);
|
||||
}
|
||||
return { count: 0 };
|
||||
}
|
||||
|
||||
@@ -1258,6 +1383,57 @@ class TestDb {
|
||||
return { changes: 1 };
|
||||
}
|
||||
|
||||
if (sql.startsWith("INSERT INTO digital_memories")) {
|
||||
const [id, key, content, category, source, sourcePath, sessionId, score, status, payload, createdAt, updatedAt, deletedAt] = args as [
|
||||
string,
|
||||
string,
|
||||
string,
|
||||
string,
|
||||
string,
|
||||
string | null,
|
||||
string | null,
|
||||
number | null,
|
||||
string,
|
||||
string,
|
||||
string,
|
||||
string,
|
||||
string | null,
|
||||
];
|
||||
const previous = this.memories.get(id);
|
||||
this.memories.set(id, {
|
||||
id,
|
||||
key,
|
||||
content,
|
||||
category,
|
||||
source,
|
||||
source_path: sourcePath,
|
||||
session_id: sessionId,
|
||||
score,
|
||||
status,
|
||||
payload,
|
||||
sync_status: previous?.sync_status === "synced" ? "pending" : previous?.sync_status ?? "pending",
|
||||
created_at: previous?.created_at ?? createdAt,
|
||||
updated_at: updatedAt,
|
||||
deleted_at: deletedAt,
|
||||
});
|
||||
return { changes: 1 };
|
||||
}
|
||||
|
||||
if (sql.startsWith("UPDATE digital_memories")) {
|
||||
const [deletedAt, updatedAt, id] = args as [string, string, string];
|
||||
const previous = this.memories.get(id);
|
||||
if (previous) {
|
||||
this.memories.set(id, {
|
||||
...previous,
|
||||
status: "deleted",
|
||||
deleted_at: deletedAt,
|
||||
updated_at: updatedAt,
|
||||
sync_status: "pending",
|
||||
});
|
||||
}
|
||||
return { changes: previous ? 1 : 0 };
|
||||
}
|
||||
|
||||
if (sql.startsWith("INSERT INTO digital_sync_outbox")) {
|
||||
const [id, entityType, entityId, operation, payload, createdAt, updatedAt] = args as [
|
||||
string,
|
||||
|
||||
@@ -218,6 +218,26 @@ export interface DigitalEmployeeApprovalRecord {
|
||||
updatedAt: string;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeMemoryRecord {
|
||||
id: string;
|
||||
remoteId?: string | null;
|
||||
key: string;
|
||||
content: string;
|
||||
category: string;
|
||||
source: string;
|
||||
sourcePath?: string | null;
|
||||
sessionId?: string | null;
|
||||
score?: number | null;
|
||||
status: string;
|
||||
payload: unknown;
|
||||
syncStatus: string;
|
||||
lastSyncedAt?: string | null;
|
||||
syncError?: string | null;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
deletedAt?: string | null;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeRuntimeRecords {
|
||||
generatedAt: string;
|
||||
plans: DigitalEmployeePlanRecord[];
|
||||
@@ -229,6 +249,39 @@ export interface DigitalEmployeeRuntimeRecords {
|
||||
events: DigitalEmployeeEventRecord[];
|
||||
artifacts: DigitalEmployeeArtifactRecord[];
|
||||
approvals: DigitalEmployeeApprovalRecord[];
|
||||
memories: DigitalEmployeeMemoryRecord[];
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeMemoryUpsertInput {
|
||||
id?: string;
|
||||
key?: string;
|
||||
content: string;
|
||||
category?: string;
|
||||
source?: string;
|
||||
sourcePath?: string | null;
|
||||
sessionId?: string | null;
|
||||
score?: number | null;
|
||||
status?: string;
|
||||
payload?: Record<string, unknown>;
|
||||
createdAt?: string;
|
||||
updatedAt?: string;
|
||||
deletedAt?: string | null;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeQimingMemoryEntry {
|
||||
id?: string;
|
||||
text?: string;
|
||||
content?: string;
|
||||
key?: string;
|
||||
category?: string;
|
||||
source?: string;
|
||||
sourcePath?: string;
|
||||
sessionId?: string | null;
|
||||
confidence?: number;
|
||||
importance?: number;
|
||||
status?: string;
|
||||
createdAt?: number | string;
|
||||
updatedAt?: number | string;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeUiState {
|
||||
@@ -624,6 +677,7 @@ export function readDigitalEmployeeRuntimeRecords(
|
||||
events: [],
|
||||
artifacts: [],
|
||||
approvals: [],
|
||||
memories: [],
|
||||
};
|
||||
}
|
||||
const safeLimit = Math.max(1, Math.min(Math.floor(limit), 300));
|
||||
@@ -695,6 +749,14 @@ export function readDigitalEmployeeRuntimeRecords(
|
||||
ORDER BY updated_at DESC, created_at DESC
|
||||
LIMIT ?
|
||||
`).all(safeLimit) as Array<Record<string, unknown>>;
|
||||
const memories = db.prepare(`
|
||||
SELECT id, remote_id, key, content, category, source, source_path,
|
||||
session_id, score, status, payload, sync_status, last_synced_at,
|
||||
sync_error, created_at, updated_at, deleted_at
|
||||
FROM digital_memories
|
||||
ORDER BY updated_at DESC, created_at DESC
|
||||
LIMIT ?
|
||||
`).all(safeLimit) as Array<Record<string, unknown>>;
|
||||
|
||||
return {
|
||||
generatedAt: new Date().toISOString(),
|
||||
@@ -707,9 +769,86 @@ export function readDigitalEmployeeRuntimeRecords(
|
||||
events: events.map(toEventRecord),
|
||||
artifacts: artifacts.map(toArtifactRecord),
|
||||
approvals: approvals.map(toApprovalRecord),
|
||||
memories: memories.map(toMemoryRecord),
|
||||
};
|
||||
}
|
||||
|
||||
export function listDigitalEmployeeMemories(options: {
|
||||
query?: string | null;
|
||||
category?: string | null;
|
||||
limit?: number;
|
||||
includeDeleted?: boolean;
|
||||
} = {}): DigitalEmployeeMemoryRecord[] {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return [];
|
||||
const rows = db.prepare(`
|
||||
SELECT id, remote_id, key, content, category, source, source_path,
|
||||
session_id, score, status, payload, sync_status, last_synced_at,
|
||||
sync_error, created_at, updated_at, deleted_at
|
||||
FROM digital_memories
|
||||
ORDER BY updated_at DESC, created_at DESC
|
||||
LIMIT ?
|
||||
`).all(Math.max(1, Math.min(Math.floor(options.limit ?? 200), 1000))) as Array<Record<string, unknown>>;
|
||||
const query = stringValue(options.query).toLowerCase();
|
||||
const category = stringValue(options.category).toLowerCase();
|
||||
return rows
|
||||
.map(toMemoryRecord)
|
||||
.filter((memory) => options.includeDeleted === true || memory.status !== "deleted")
|
||||
.filter((memory) => !category || memory.category.toLowerCase() === category)
|
||||
.filter((memory) => {
|
||||
if (!query) return true;
|
||||
return `${memory.key} ${memory.content} ${memory.category} ${memory.source}`.toLowerCase().includes(query);
|
||||
});
|
||||
}
|
||||
|
||||
export function upsertDigitalEmployeeMemory(
|
||||
input: DigitalEmployeeMemoryUpsertInput,
|
||||
): DigitalEmployeeMemoryRecord | null {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return null;
|
||||
const now = new Date().toISOString();
|
||||
const normalized = normalizeMemoryInput(input, now);
|
||||
const tx = db.transaction(() => upsertMemory(normalized));
|
||||
tx();
|
||||
return readDigitalEmployeeMemory(normalized.id);
|
||||
}
|
||||
|
||||
export function deleteDigitalEmployeeMemory(
|
||||
keyOrId: string,
|
||||
): DigitalEmployeeMemoryRecord | null {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return null;
|
||||
const now = new Date().toISOString();
|
||||
const target = readDigitalEmployeeMemoryByKeyOrId(keyOrId);
|
||||
if (!target) return null;
|
||||
const tx = db.transaction(() => {
|
||||
db.prepare(`
|
||||
UPDATE digital_memories
|
||||
SET status = 'deleted', deleted_at = ?, updated_at = ?, sync_status = 'pending'
|
||||
WHERE id = ?
|
||||
`).run(now, now, target.id);
|
||||
markOutbox("memory", target.id, "delete", { ...target, status: "deleted", deletedAt: now, updatedAt: now }, now);
|
||||
});
|
||||
tx();
|
||||
return readDigitalEmployeeMemory(target.id);
|
||||
}
|
||||
|
||||
export function syncQimingclawMemoryEntries(
|
||||
entries: DigitalEmployeeQimingMemoryEntry[],
|
||||
): DigitalEmployeeMemoryRecord[] {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return [];
|
||||
const now = new Date().toISOString();
|
||||
const records = entries.map((entry) => normalizeQimingMemoryEntry(entry, now));
|
||||
const tx = db.transaction(() => {
|
||||
for (const record of records) upsertMemory(record);
|
||||
});
|
||||
tx();
|
||||
return records
|
||||
.map((record) => readDigitalEmployeeMemory(record.id))
|
||||
.filter((record): record is DigitalEmployeeMemoryRecord => Boolean(record));
|
||||
}
|
||||
|
||||
export function listDueDigitalEmployeeSchedules(
|
||||
nowIso: string,
|
||||
limit = 20,
|
||||
@@ -2452,6 +2591,28 @@ function toApprovalRecord(
|
||||
};
|
||||
}
|
||||
|
||||
function toMemoryRecord(row: Record<string, unknown>): DigitalEmployeeMemoryRecord {
|
||||
return {
|
||||
id: stringField(row, "id"),
|
||||
remoteId: optionalStringField(row, "remote_id"),
|
||||
key: stringField(row, "key"),
|
||||
content: stringField(row, "content"),
|
||||
category: stringField(row, "category") || "fact",
|
||||
source: stringField(row, "source") || "qimingclaw",
|
||||
sourcePath: optionalStringField(row, "source_path"),
|
||||
sessionId: optionalStringField(row, "session_id"),
|
||||
score: typeof row.score === "number" ? row.score : row.score == null ? null : Number(row.score) || null,
|
||||
status: stringField(row, "status") || "active",
|
||||
payload: parseStoredPayload(row.payload),
|
||||
syncStatus: stringField(row, "sync_status"),
|
||||
lastSyncedAt: optionalStringField(row, "last_synced_at"),
|
||||
syncError: optionalStringField(row, "sync_error"),
|
||||
createdAt: stringField(row, "created_at"),
|
||||
updatedAt: stringField(row, "updated_at"),
|
||||
deletedAt: optionalStringField(row, "deleted_at"),
|
||||
};
|
||||
}
|
||||
|
||||
function countPendingSyncItems(): number {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return 0;
|
||||
@@ -2494,6 +2655,154 @@ function markOutbox(
|
||||
);
|
||||
}
|
||||
|
||||
function normalizeMemoryInput(
|
||||
input: DigitalEmployeeMemoryUpsertInput,
|
||||
now: string,
|
||||
): Required<Pick<DigitalEmployeeMemoryUpsertInput, "id" | "key" | "content" | "category" | "source" | "payload">> & {
|
||||
sourcePath: string | null;
|
||||
sessionId: string | null;
|
||||
score: number | null;
|
||||
status: string;
|
||||
createdAt: string;
|
||||
updatedAt: string;
|
||||
deletedAt: string | null;
|
||||
} {
|
||||
const content = stringValue(input.content);
|
||||
const key = stringValue(input.key) || memoryKeyFromContent(content);
|
||||
return {
|
||||
id: stringValue(input.id) || `memory-${safeId(key)}`,
|
||||
key,
|
||||
content,
|
||||
category: stringValue(input.category) || "fact",
|
||||
source: stringValue(input.source) || "qimingclaw-digital-local",
|
||||
sourcePath: stringValue(input.sourcePath) || null,
|
||||
sessionId: stringValue(input.sessionId) || null,
|
||||
score: typeof input.score === "number" && Number.isFinite(input.score) ? input.score : null,
|
||||
status: stringValue(input.status) || "active",
|
||||
payload: input.payload ?? {},
|
||||
createdAt: input.createdAt || now,
|
||||
updatedAt: input.updatedAt || now,
|
||||
deletedAt: input.deletedAt ?? null,
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeQimingMemoryEntry(
|
||||
entry: DigitalEmployeeQimingMemoryEntry,
|
||||
now: string,
|
||||
): ReturnType<typeof normalizeMemoryInput> {
|
||||
const content = stringValue(entry.text ?? entry.content);
|
||||
const createdAt = memoryTimestampToIso(entry.createdAt) || now;
|
||||
const updatedAt = memoryTimestampToIso(entry.updatedAt) || createdAt;
|
||||
const score = typeof entry.importance === "number"
|
||||
? entry.importance
|
||||
: typeof entry.confidence === "number"
|
||||
? entry.confidence
|
||||
: null;
|
||||
return normalizeMemoryInput({
|
||||
id: stringValue(entry.id) || undefined,
|
||||
key: stringValue(entry.key) || stringValue(entry.id) || memoryKeyFromContent(content),
|
||||
content,
|
||||
category: stringValue(entry.category) || "fact",
|
||||
source: stringValue(entry.source) || "qimingclaw-memory-service",
|
||||
sourcePath: stringValue(entry.sourcePath) || null,
|
||||
sessionId: stringValue(entry.sessionId) || null,
|
||||
score,
|
||||
status: stringValue(entry.status) || "active",
|
||||
payload: {
|
||||
source: "qimingclaw-memory-service",
|
||||
original: entry,
|
||||
},
|
||||
createdAt,
|
||||
updatedAt,
|
||||
deletedAt: stringValue(entry.status) === "deleted" ? updatedAt : null,
|
||||
}, now);
|
||||
}
|
||||
|
||||
function memoryTimestampToIso(value: number | string | undefined): string | null {
|
||||
if (typeof value === "number" && Number.isFinite(value)) {
|
||||
return new Date(value).toISOString();
|
||||
}
|
||||
if (typeof value === "string" && value.trim()) {
|
||||
const parsed = Date.parse(value);
|
||||
return Number.isFinite(parsed) ? new Date(parsed).toISOString() : null;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function memoryKeyFromContent(content: string): string {
|
||||
return safeId(content.slice(0, 80) || `memory-${Date.now()}`);
|
||||
}
|
||||
|
||||
function readDigitalEmployeeMemory(id: string): DigitalEmployeeMemoryRecord | null {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return null;
|
||||
const row = db.prepare(`
|
||||
SELECT id, remote_id, key, content, category, source, source_path,
|
||||
session_id, score, status, payload, sync_status, last_synced_at,
|
||||
sync_error, created_at, updated_at, deleted_at
|
||||
FROM digital_memories
|
||||
WHERE id = ?
|
||||
`).get(id) as Record<string, unknown> | undefined;
|
||||
return row ? toMemoryRecord(row) : null;
|
||||
}
|
||||
|
||||
function readDigitalEmployeeMemoryByKeyOrId(keyOrId: string): DigitalEmployeeMemoryRecord | null {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return null;
|
||||
const row = db.prepare(`
|
||||
SELECT id, remote_id, key, content, category, source, source_path,
|
||||
session_id, score, status, payload, sync_status, last_synced_at,
|
||||
sync_error, created_at, updated_at, deleted_at
|
||||
FROM digital_memories
|
||||
WHERE id = ? OR key = ?
|
||||
ORDER BY CASE WHEN status = 'deleted' THEN 1 ELSE 0 END ASC, updated_at DESC, created_at DESC
|
||||
LIMIT 1
|
||||
`).get(keyOrId, keyOrId) as Record<string, unknown> | undefined;
|
||||
return row ? toMemoryRecord(row) : null;
|
||||
}
|
||||
|
||||
function upsertMemory(input: ReturnType<typeof normalizeMemoryInput>): void {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return;
|
||||
db.prepare(`
|
||||
INSERT INTO digital_memories (
|
||||
id, key, content, category, source, source_path, session_id, score,
|
||||
status, payload, sync_status, created_at, updated_at, deleted_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
key = excluded.key,
|
||||
content = excluded.content,
|
||||
category = excluded.category,
|
||||
source = excluded.source,
|
||||
source_path = excluded.source_path,
|
||||
session_id = excluded.session_id,
|
||||
score = excluded.score,
|
||||
status = excluded.status,
|
||||
payload = excluded.payload,
|
||||
deleted_at = excluded.deleted_at,
|
||||
sync_status = CASE
|
||||
WHEN digital_memories.sync_status = 'synced' THEN 'pending'
|
||||
ELSE digital_memories.sync_status
|
||||
END,
|
||||
updated_at = excluded.updated_at
|
||||
`).run(
|
||||
input.id,
|
||||
input.key,
|
||||
input.content,
|
||||
input.category,
|
||||
input.source,
|
||||
input.sourcePath,
|
||||
input.sessionId,
|
||||
input.score,
|
||||
input.status,
|
||||
stringify(input.payload),
|
||||
input.createdAt,
|
||||
input.updatedAt,
|
||||
input.deletedAt,
|
||||
);
|
||||
markOutbox("memory", input.id, input.deletedAt ? "delete" : "upsert", input, input.updatedAt);
|
||||
}
|
||||
|
||||
function normalizeScheduleInput(
|
||||
input: DigitalEmployeeScheduleUpsertInput,
|
||||
now: string,
|
||||
|
||||
@@ -439,6 +439,71 @@ describe("digital employee sync service", () => {
|
||||
remote_id: "remote-schedule-run-1",
|
||||
});
|
||||
});
|
||||
|
||||
it("syncs memory business views and acknowledgements", async () => {
|
||||
insertEntity("memory", "memory-1");
|
||||
insertOutbox(
|
||||
"memory:upsert:memory-1",
|
||||
"memory",
|
||||
"memory-1",
|
||||
JSON.stringify({
|
||||
key: "commit-language",
|
||||
content: "用户要求提交信息使用中文，并保留相关上下文。",
|
||||
category: "preference",
|
||||
source: "qimingclaw-memory-service",
|
||||
status: "active",
|
||||
}),
|
||||
);
|
||||
mockState.fetch.mockResolvedValue(
|
||||
jsonResponse({
|
||||
success: true,
|
||||
acked: [{ entity_type: "memory", entity_id: "memory-1", remote_id: "remote-memory-1" }],
|
||||
}),
|
||||
);
|
||||
|
||||
const { flushDigitalEmployeeSyncOutbox } = await import("./syncService");
|
||||
const status = await flushDigitalEmployeeSyncOutbox({ force: true });
|
||||
|
||||
expect(status.failed).toBe(0);
|
||||
expect(lastSyncRequestBody()).toMatchObject({
|
||||
items: [
|
||||
expect.objectContaining({
|
||||
entity_type: "memory",
|
||||
business_view: expect.objectContaining({
|
||||
entity_type: "memory",
|
||||
local_id: "memory-1",
|
||||
key: "commit-language",
|
||||
category: "preference",
|
||||
source: "qimingclaw-memory-service",
|
||||
status: "active",
|
||||
content_preview: "用户要求提交信息使用中文，并保留相关上下文。",
|
||||
}),
|
||||
}),
|
||||
],
|
||||
});
|
||||
expect(readEntity("memory", "memory-1")).toMatchObject({
|
||||
sync_status: "synced",
|
||||
remote_id: "remote-memory-1",
|
||||
sync_error: null,
|
||||
});
|
||||
});
|
||||
|
||||
it("marks memory sync failures as failed", async () => {
|
||||
insertEntity("memory", "memory-rejected");
|
||||
insertOutbox("memory:upsert:memory-rejected", "memory", "memory-rejected");
|
||||
mockState.fetch.mockResolvedValue(
|
||||
jsonResponse({ code: "0001", message: "记忆同步被拒绝", data: null }),
|
||||
);
|
||||
|
||||
const { flushDigitalEmployeeSyncOutbox } = await import("./syncService");
|
||||
const status = await flushDigitalEmployeeSyncOutbox({ force: true });
|
||||
|
||||
expect(status.failed).toBe(1);
|
||||
expect(readEntity("memory", "memory-rejected")).toMatchObject({
|
||||
sync_status: "failed",
|
||||
sync_error: "记忆同步被拒绝",
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
function insertPlan(id: string): void {
|
||||
@@ -505,6 +570,7 @@ function entityMap(
|
||||
if (entityType === "schedule_run") return mockState.db.scheduleRuns;
|
||||
if (entityType === "artifact") return mockState.db.artifacts;
|
||||
if (entityType === "approval") return mockState.db.approvals;
|
||||
if (entityType === "memory") return mockState.db.memories;
|
||||
return undefined;
|
||||
}
|
||||
|
||||
@@ -573,6 +639,7 @@ class TestDb {
|
||||
scheduleRuns = new Map<string, TestSyncEntityRow>();
|
||||
artifacts = new Map<string, TestSyncEntityRow>();
|
||||
approvals = new Map<string, TestSyncEntityRow>();
|
||||
memories = new Map<string, TestSyncEntityRow>();
|
||||
attempts: TestAttemptRow[] = [];
|
||||
private nextAttemptId = 1;
|
||||
|
||||
@@ -673,6 +740,11 @@ class TestDb {
|
||||
return this.approvals.get(id);
|
||||
}
|
||||
|
||||
if (sql.startsWith("SELECT key AS title, status, content AS summary, payload FROM digital_memories")) {
|
||||
const [id] = args as [string];
|
||||
return this.memories.get(id);
|
||||
}
|
||||
|
||||
if (sql.startsWith("SELECT title, status, NULL AS objective, payload FROM digital_plan_steps")) {
|
||||
const [id] = args as [string];
|
||||
return this.planSteps.get(id);
|
||||
@@ -785,7 +857,7 @@ class TestDb {
|
||||
return { changes: before - this.attempts.length };
|
||||
}
|
||||
|
||||
if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|artifacts|approvals) SET sync_status = 'synced'/.test(sql)) {
|
||||
if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|artifacts|approvals|memories) SET sync_status = 'synced'/.test(sql)) {
|
||||
const [remoteId, lastSyncedAt, id] = args as [string | null, string, string];
|
||||
const row = this.entityRowsForUpdate(sql).get(id);
|
||||
if (row) {
|
||||
@@ -797,7 +869,7 @@ class TestDb {
|
||||
return { changes: row ? 1 : 0 };
|
||||
}
|
||||
|
||||
if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|artifacts|approvals) SET sync_status = 'failed'/.test(sql)) {
|
||||
if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|artifacts|approvals|memories) SET sync_status = 'failed'/.test(sql)) {
|
||||
const [syncError, id] = args as [string, string];
|
||||
const row = this.entityRowsForUpdate(sql).get(id);
|
||||
if (row) {
|
||||
@@ -813,6 +885,7 @@ class TestDb {
|
||||
private entityRowsForUpdate(sql: string): Map<string, TestSyncEntityRow> {
|
||||
if (sql.startsWith("UPDATE digital_artifacts")) return this.artifacts;
|
||||
if (sql.startsWith("UPDATE digital_approvals")) return this.approvals;
|
||||
if (sql.startsWith("UPDATE digital_memories")) return this.memories;
|
||||
if (sql.startsWith("UPDATE digital_schedule_runs")) return this.scheduleRuns;
|
||||
if (sql.startsWith("UPDATE digital_schedules")) return this.schedules;
|
||||
if (sql.startsWith("UPDATE digital_plan_steps")) return this.planSteps;
|
||||
|
||||
@@ -496,11 +496,25 @@ function buildBusinessView(
|
||||
run_id: stringField(record.runId ?? record.run_id) || null,
|
||||
decision: stringField(record.decision) || null,
|
||||
};
|
||||
case "memory":
|
||||
return {
|
||||
...base,
|
||||
key: stringField(record.key) || row.entity_id,
|
||||
category: stringField(record.category) || null,
|
||||
source: stringField(record.source) || null,
|
||||
content_preview: compactPreview(stringField(record.content) || summary?.summary || ""),
|
||||
};
|
||||
default:
|
||||
return base;
|
||||
}
|
||||
}
|
||||
|
||||
function compactPreview(value: string, maxLength = 120): string | null {
|
||||
const compact = value.replace(/\s+/g, " ").trim();
|
||||
if (!compact) return null;
|
||||
return compact.length > maxLength ? `${compact.slice(0, maxLength)}...` : compact;
|
||||
}
|
||||
|
||||
function markRowsSyncing(rows: OutboxRow[], now: string): void {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return;
|
||||
@@ -774,6 +788,8 @@ function entitySummarySelect(entityType: string): string | null {
|
||||
return "SELECT name AS title, kind AS status, uri AS summary, payload FROM digital_artifacts WHERE id = ?";
|
||||
case "approval":
|
||||
return "SELECT title, status, NULL AS objective, payload FROM digital_approvals WHERE id = ?";
|
||||
case "memory":
|
||||
return "SELECT key AS title, status, content AS summary, payload FROM digital_memories WHERE id = ?";
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
@@ -876,6 +892,8 @@ function entityTable(entityType: string): string | null {
|
||||
return "digital_artifacts";
|
||||
case "approval":
|
||||
return "digital_approvals";
|
||||
case "memory":
|
||||
return "digital_memories";
|
||||
default:
|
||||
return null;
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user