From 2373e0b5f3993be3cc284f0aba5f194081a1690b Mon Sep 17 00:00:00 2001 From: baiyanyun Date: Fri, 5 Jun 2026 16:43:23 +0800 Subject: [PATCH] feat(client): add digital employee sync schema --- .../agent-electron-client/src/main/db.ts | 120 +++++++++ .../services/digitalEmployee/stateService.ts | 252 +++++++++++++++++- ...ital-employee-frontend-integration-plan.md | 48 +++- 3 files changed, 416 insertions(+), 4 deletions(-) diff --git a/qimingclaw/crates/agent-electron-client/src/main/db.ts b/qimingclaw/crates/agent-electron-client/src/main/db.ts index ba14e10b..571fec4f 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/db.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/db.ts @@ -21,6 +21,126 @@ export function initDatabase(): void { key TEXT PRIMARY KEY, value TEXT NOT NULL ); + + CREATE TABLE IF NOT EXISTS digital_plans ( + id TEXT PRIMARY KEY, + remote_id TEXT, + title TEXT NOT NULL, + objective TEXT, + status TEXT NOT NULL, + source TEXT NOT NULL DEFAULT 'local', + 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 + ); + + CREATE TABLE IF NOT EXISTS digital_tasks ( + id TEXT PRIMARY KEY, + remote_id TEXT, + plan_id TEXT, + title TEXT NOT NULL, + status TEXT NOT NULL, + assigned_skill_id TEXT, + 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, + FOREIGN KEY(plan_id) REFERENCES digital_plans(id) + ); + + CREATE TABLE IF NOT EXISTS digital_runs ( + id TEXT PRIMARY KEY, + remote_id TEXT, + plan_id TEXT, + task_id TEXT, + status TEXT NOT NULL, + started_at TEXT, + finished_at TEXT, + 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, + FOREIGN KEY(plan_id) REFERENCES digital_plans(id), + FOREIGN KEY(task_id) REFERENCES digital_tasks(id) + ); + + CREATE TABLE IF NOT EXISTS digital_events ( + id TEXT PRIMARY KEY, + remote_id TEXT, + plan_id TEXT, + task_id TEXT, + run_id TEXT, + kind TEXT NOT NULL, + message TEXT NOT NULL, + payload TEXT NOT NULL DEFAULT '{}', + sync_status TEXT NOT NULL DEFAULT 'pending', + last_synced_at TEXT, + sync_error TEXT, + occurred_at TEXT NOT NULL, + created_at TEXT NOT NULL, + FOREIGN KEY(plan_id) REFERENCES digital_plans(id), + FOREIGN KEY(task_id) REFERENCES digital_tasks(id), + FOREIGN KEY(run_id) REFERENCES digital_runs(id) + ); + + CREATE TABLE IF NOT EXISTS digital_artifacts ( + id TEXT PRIMARY KEY, + remote_id TEXT, + plan_id TEXT, + task_id TEXT, + run_id TEXT, + kind TEXT NOT NULL, + name TEXT, + uri TEXT, + payload TEXT NOT NULL DEFAULT '{}', + sync_status TEXT NOT NULL DEFAULT 'pending', + last_synced_at TEXT, + sync_error TEXT, + created_at TEXT NOT NULL + ); + + CREATE TABLE IF NOT EXISTS digital_approvals ( + id TEXT PRIMARY KEY, + remote_id TEXT, + plan_id TEXT, + task_id TEXT, + run_id TEXT, + title TEXT NOT NULL, + status TEXT NOT NULL, + 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 + ); + + CREATE TABLE IF NOT EXISTS digital_sync_outbox ( + id TEXT PRIMARY KEY, + entity_type TEXT NOT NULL, + entity_id TEXT NOT NULL, + operation TEXT NOT NULL, + payload TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + attempts INTEGER NOT NULL DEFAULT 0, + last_attempt_at TEXT, + error TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + + CREATE INDEX IF NOT EXISTS idx_digital_plans_sync ON digital_plans(sync_status); + 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_sync_outbox_status ON digital_sync_outbox(status, created_at); `); log.info('Database tables created'); } catch (error) { diff --git a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts index e7e45291..72bee95a 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts @@ -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), 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, diff --git a/qimingclaw/docs/digital-employee-frontend-integration-plan.md b/qimingclaw/docs/digital-employee-frontend-integration-plan.md index 5aae3dd8..ebafc686 100644 --- a/qimingclaw/docs/digital-employee-frontend-integration-plan.md +++ b/qimingclaw/docs/digital-employee-frontend-integration-plan.md @@ -340,8 +340,8 @@ digitalEmployee:getState 当前边界： -- 状态服务仍是轻量版本，先使用 settings 表持久化。 -- Plan / Task / Run / Event 还没有独立 SQLite 表。 +- 状态服务仍保留 settings JSON 作为兼容快照。 +- Plan / Task / Run / Event 已开始写入独立 SQLite 表。 - 当前事件主要记录服务状态变化，还没有接入 `/computer/chat` 和 ACP token/event stream。 下一步： @@ -350,6 +350,50 @@ digitalEmployee:getState - 在 computer server 的 `/computer/chat` 入口创建 Task / Run。 - 在 ACP event forwarder 中写入 Run/Event。 +## 管理端联动原则 + +数字员工状态是本地优先，但不能成为本地孤岛。 + +本地 SQLite 表设计必须同时满足： + +- 离线可用：客户端无网络时仍能展示任务、运行明细和日报。 +- 可追溯：本地每个 Plan / Task / Run / Event 都有稳定本地 ID。 +- 可同步：预留管理端 ID 和同步状态。 +- 可补偿：同步失败后可通过 outbox 重试。 + +当前已新增/规划的本地表： + +```text +digital_plans +digital_tasks +digital_runs +digital_events +digital_artifacts +digital_approvals +digital_sync_outbox +``` + +同步字段： + +```text +remote_id +sync_status +last_synced_at +sync_error +``` + +同步流建议： + +```text +本地写入 Plan/Task/Run/Event + -> digital_sync_outbox pending + -> 管理端 sync API + -> 回填 remote_id + -> sync_status = synced +``` + +管理端 API 尚未在 qiming-backend 中确认最终路径，因此当前 qimingclaw 先落本地同步契约，不硬编码远端接口。 + ### 产出 - 数字员工页面可以从源码重新构建。