feat(client): add digital employee sync schema
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import { readSetting, writeSetting } from "../../db";
|
||||
import { getDb, readSetting, writeSetting } from "../../db";
|
||||
|
||||
const STATE_KEY = "digital_employee_state_v1";
|
||||
const MAX_EVENTS = 200;
|
||||
@@ -45,6 +45,7 @@ export interface DigitalEmployeeState {
|
||||
updatedAt: string | null;
|
||||
lastSnapshot: DigitalEmployeeSnapshot | null;
|
||||
events: DigitalEmployeeStateEvent[];
|
||||
pendingSyncCount?: number;
|
||||
}
|
||||
|
||||
function defaultState(): DigitalEmployeeState {
|
||||
@@ -58,13 +59,15 @@ function defaultState(): DigitalEmployeeState {
|
||||
|
||||
export function readDigitalEmployeeState(): DigitalEmployeeState {
|
||||
const state = readSetting(STATE_KEY);
|
||||
const pendingSyncCount = countPendingSyncItems();
|
||||
if (!state || typeof state !== "object") {
|
||||
return defaultState();
|
||||
return { ...defaultState(), pendingSyncCount };
|
||||
}
|
||||
return {
|
||||
...defaultState(),
|
||||
...(state as Partial<DigitalEmployeeState>),
|
||||
version: 1,
|
||||
pendingSyncCount,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -89,9 +92,254 @@ export function recordDigitalEmployeeSnapshot(
|
||||
events: events.slice(0, MAX_EVENTS),
|
||||
};
|
||||
writeDigitalEmployeeState(next);
|
||||
persistSnapshotTables(snapshot, event);
|
||||
return next;
|
||||
}
|
||||
|
||||
function persistSnapshotTables(
|
||||
snapshot: DigitalEmployeeSnapshot,
|
||||
event: DigitalEmployeeStateEvent | null,
|
||||
): void {
|
||||
const db = getDb();
|
||||
if (!db) return;
|
||||
|
||||
const now = snapshot.generatedAt;
|
||||
const planStatus = snapshot.services.agent?.running || snapshot.services.mcp?.running
|
||||
? "running"
|
||||
: "pending";
|
||||
const taskStatus = snapshot.services.agent?.running ? "running" : "pending";
|
||||
const runStatus = planStatus === "running" ? "running" : "pending";
|
||||
|
||||
const planPayload = {
|
||||
services: snapshot.services,
|
||||
sessions: snapshot.sessions,
|
||||
};
|
||||
const taskPayload = {
|
||||
service: "agent",
|
||||
agent: snapshot.services.agent,
|
||||
mcp: snapshot.services.mcp,
|
||||
};
|
||||
const runPayload = {
|
||||
snapshot,
|
||||
};
|
||||
|
||||
const tx = db.transaction(() => {
|
||||
upsertPlan({
|
||||
id: "qimingclaw-client-readiness",
|
||||
title: "客户端运行状态巡检",
|
||||
objective: "确认 qimingclaw 客户端、Agent、MCP 和远程接入能力是否处于可执行状态。",
|
||||
status: planStatus,
|
||||
payload: planPayload,
|
||||
now,
|
||||
});
|
||||
upsertTask({
|
||||
id: "task-agent-status",
|
||||
planId: "qimingclaw-client-readiness",
|
||||
title: "读取 Agent 服务状态",
|
||||
status: taskStatus,
|
||||
assignedSkillId: "qimingclaw-acp-session",
|
||||
payload: taskPayload,
|
||||
now,
|
||||
});
|
||||
upsertRun({
|
||||
id: "run-service-snapshot",
|
||||
planId: "qimingclaw-client-readiness",
|
||||
taskId: "task-agent-status",
|
||||
status: runStatus,
|
||||
payload: runPayload,
|
||||
now,
|
||||
});
|
||||
if (event) {
|
||||
insertEvent(event, "run-service-snapshot", { snapshot });
|
||||
}
|
||||
});
|
||||
|
||||
tx();
|
||||
}
|
||||
|
||||
function stringify(value: unknown): string {
|
||||
return JSON.stringify(value ?? {});
|
||||
}
|
||||
|
||||
function countPendingSyncItems(): number {
|
||||
const db = getDb();
|
||||
if (!db) return 0;
|
||||
const row = db.prepare(`
|
||||
SELECT COUNT(*) AS count FROM digital_sync_outbox WHERE status = 'pending'
|
||||
`).get() as { count: number } | undefined;
|
||||
return row?.count ?? 0;
|
||||
}
|
||||
|
||||
function syncOutboxId(entityType: string, entityId: string, operation = "upsert"): string {
|
||||
return `${entityType}:${operation}:${entityId}`;
|
||||
}
|
||||
|
||||
function markOutbox(
|
||||
entityType: string,
|
||||
entityId: string,
|
||||
operation: string,
|
||||
payload: unknown,
|
||||
now: string,
|
||||
): void {
|
||||
const db = getDb();
|
||||
if (!db) return;
|
||||
db.prepare(`
|
||||
INSERT INTO digital_sync_outbox (
|
||||
id, entity_type, entity_id, operation, payload, status, attempts, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, 'pending', 0, ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
payload = excluded.payload,
|
||||
status = 'pending',
|
||||
updated_at = excluded.updated_at,
|
||||
error = NULL
|
||||
`).run(
|
||||
syncOutboxId(entityType, entityId, operation),
|
||||
entityType,
|
||||
entityId,
|
||||
operation,
|
||||
stringify(payload),
|
||||
now,
|
||||
now,
|
||||
);
|
||||
}
|
||||
|
||||
function upsertPlan(input: {
|
||||
id: string;
|
||||
title: string;
|
||||
objective: string;
|
||||
status: string;
|
||||
payload: unknown;
|
||||
now: string;
|
||||
}): void {
|
||||
const db = getDb();
|
||||
if (!db) return;
|
||||
db.prepare(`
|
||||
INSERT INTO digital_plans (
|
||||
id, title, objective, status, source, payload, sync_status, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, 'qimingclaw', ?, 'pending', ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
title = excluded.title,
|
||||
objective = excluded.objective,
|
||||
status = excluded.status,
|
||||
payload = excluded.payload,
|
||||
sync_status = CASE
|
||||
WHEN digital_plans.sync_status = 'synced' THEN 'pending'
|
||||
ELSE digital_plans.sync_status
|
||||
END,
|
||||
updated_at = excluded.updated_at
|
||||
`).run(
|
||||
input.id,
|
||||
input.title,
|
||||
input.objective,
|
||||
input.status,
|
||||
stringify(input.payload),
|
||||
input.now,
|
||||
input.now,
|
||||
);
|
||||
markOutbox("plan", input.id, "upsert", input, input.now);
|
||||
}
|
||||
|
||||
function upsertTask(input: {
|
||||
id: string;
|
||||
planId: string;
|
||||
title: string;
|
||||
status: string;
|
||||
assignedSkillId: string;
|
||||
payload: unknown;
|
||||
now: string;
|
||||
}): void {
|
||||
const db = getDb();
|
||||
if (!db) return;
|
||||
db.prepare(`
|
||||
INSERT INTO digital_tasks (
|
||||
id, plan_id, title, status, assigned_skill_id, payload, sync_status, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
plan_id = excluded.plan_id,
|
||||
title = excluded.title,
|
||||
status = excluded.status,
|
||||
assigned_skill_id = excluded.assigned_skill_id,
|
||||
payload = excluded.payload,
|
||||
sync_status = CASE
|
||||
WHEN digital_tasks.sync_status = 'synced' THEN 'pending'
|
||||
ELSE digital_tasks.sync_status
|
||||
END,
|
||||
updated_at = excluded.updated_at
|
||||
`).run(
|
||||
input.id,
|
||||
input.planId,
|
||||
input.title,
|
||||
input.status,
|
||||
input.assignedSkillId,
|
||||
stringify(input.payload),
|
||||
input.now,
|
||||
input.now,
|
||||
);
|
||||
markOutbox("task", input.id, "upsert", input, input.now);
|
||||
}
|
||||
|
||||
function upsertRun(input: {
|
||||
id: string;
|
||||
planId: string;
|
||||
taskId: string;
|
||||
status: string;
|
||||
payload: unknown;
|
||||
now: string;
|
||||
}): void {
|
||||
const db = getDb();
|
||||
if (!db) return;
|
||||
db.prepare(`
|
||||
INSERT INTO digital_runs (
|
||||
id, plan_id, task_id, status, started_at, payload, sync_status, created_at, updated_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?)
|
||||
ON CONFLICT(id) DO UPDATE SET
|
||||
plan_id = excluded.plan_id,
|
||||
task_id = excluded.task_id,
|
||||
status = excluded.status,
|
||||
payload = excluded.payload,
|
||||
sync_status = CASE
|
||||
WHEN digital_runs.sync_status = 'synced' THEN 'pending'
|
||||
ELSE digital_runs.sync_status
|
||||
END,
|
||||
updated_at = excluded.updated_at
|
||||
`).run(
|
||||
input.id,
|
||||
input.planId,
|
||||
input.taskId,
|
||||
input.status,
|
||||
input.now,
|
||||
stringify(input.payload),
|
||||
input.now,
|
||||
input.now,
|
||||
);
|
||||
markOutbox("run", input.id, "upsert", input, input.now);
|
||||
}
|
||||
|
||||
function insertEvent(
|
||||
event: DigitalEmployeeStateEvent,
|
||||
runId: string,
|
||||
payload: unknown,
|
||||
): void {
|
||||
const db = getDb();
|
||||
if (!db) return;
|
||||
db.prepare(`
|
||||
INSERT OR IGNORE INTO digital_events (
|
||||
id, plan_id, task_id, run_id, kind, message, payload, sync_status, occurred_at, created_at
|
||||
) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
|
||||
`).run(
|
||||
event.event_id,
|
||||
event.plan_id,
|
||||
event.task_id,
|
||||
runId,
|
||||
event.kind,
|
||||
event.message,
|
||||
stringify(payload),
|
||||
event.occurred_at,
|
||||
event.occurred_at,
|
||||
);
|
||||
markOutbox("event", event.event_id, "insert", event, event.occurred_at);
|
||||
}
|
||||
|
||||
function eventFromSnapshot(
|
||||
snapshot: DigitalEmployeeSnapshot,
|
||||
previous: DigitalEmployeeSnapshot | null,
|
||||
|
||||
Reference in New Issue
Block a user