diff --git a/qimingclaw/crates/agent-electron-client/embedded/sgrobot-digital/src/lib/qimingclawAdapter.ts b/qimingclaw/crates/agent-electron-client/embedded/sgrobot-digital/src/lib/qimingclawAdapter.ts index e97c9af4..50058ab5 100644 --- a/qimingclaw/crates/agent-electron-client/embedded/sgrobot-digital/src/lib/qimingclawAdapter.ts +++ b/qimingclaw/crates/agent-electron-client/embedded/sgrobot-digital/src/lib/qimingclawAdapter.ts @@ -20,6 +20,8 @@ declare global { QimingClawBridge?: { digital?: { getSnapshot?: () => Promise; + getSyncStatus?: () => Promise; + flushSync?: () => Promise; }; }; } @@ -50,6 +52,17 @@ interface QimingclawServiceStatus { serverNames?: string[]; } +interface QimingclawSyncStatus { + running: boolean; + syncing: boolean; + enabled: boolean; + endpoint: string | null; + pending: number; + failed: number; + lastSyncAt: string | null; + lastSyncError: string | null; +} + interface QimingclawSnapshot { generatedAt: string; services: Record; diff --git a/qimingclaw/crates/agent-electron-client/src/main/bootstrap/startup.ts b/qimingclaw/crates/agent-electron-client/src/main/bootstrap/startup.ts index 3958af59..32df2af3 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/bootstrap/startup.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/bootstrap/startup.ts @@ -13,6 +13,7 @@ import { startSandboxService, stopSandboxService, } from "../services/sandbox/serviceBootstrap"; +import { startDigitalEmployeeSyncWorker } from "../services/digitalEmployee/syncService"; /** 依赖同步是否正在进行 */ let _depsSyncInProgress = false; @@ -86,6 +87,11 @@ export async function runStartupTasks(): Promise { } }); + // 数字员工本地状态与管理端同步:本地 outbox 优先,网络失败不影响执行链路。 + setImmediate(() => { + startDigitalEmployeeSyncWorker(); + }); + // 客户端升级后:若 appVersion 或 installVersion 变化,后台同步初始化依赖到写死版本 // 同步检查是否需要 dep sync,提前设置标志,避免 renderer 在 setImmediate 之前 // 读到 syncInProgress=false 而过早启动服务(竞态条件) diff --git a/qimingclaw/crates/agent-electron-client/src/main/ipc/digitalEmployeeHandlers.ts b/qimingclaw/crates/agent-electron-client/src/main/ipc/digitalEmployeeHandlers.ts index 50e2f9b7..5f2c4cf7 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/ipc/digitalEmployeeHandlers.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/ipc/digitalEmployeeHandlers.ts @@ -13,6 +13,10 @@ import { type DigitalEmployeeServiceStatus, type DigitalEmployeeSnapshot, } from "../services/digitalEmployee/stateService"; +import { + flushDigitalEmployeeSyncOutbox, + getDigitalEmployeeSyncStatus, +} from "../services/digitalEmployee/syncService"; export function registerDigitalEmployeeHandlers(ctx: HandlerContext): void { ipcMain.handle("digitalEmployee:getSnapshot", async () => { @@ -30,6 +34,14 @@ export function registerDigitalEmployeeHandlers(ctx: HandlerContext): void { ipcMain.handle("digitalEmployee:getState", async () => { return readDigitalEmployeeState(); }); + + ipcMain.handle("digitalEmployee:getSyncStatus", async () => { + return getDigitalEmployeeSyncStatus(); + }); + + ipcMain.handle("digitalEmployee:flushSync", async () => { + return flushDigitalEmployeeSyncOutbox(); + }); } function buildSnapshot(ctx: HandlerContext): DigitalEmployeeSnapshot { diff --git a/qimingclaw/crates/agent-electron-client/src/main/main.ts b/qimingclaw/crates/agent-electron-client/src/main/main.ts index abd8575b..a667f122 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/main.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/main.ts @@ -39,6 +39,7 @@ import { initAutoUpdater } from "./services/autoUpdater"; import { migrateDataDir, migrateSettingsPaths } from "./bootstrap/migrate"; import { getDeviceId, logSystemInfo } from "./services/system/deviceId"; import { initWebviewPolicy } from "./services/system/webviewPolicy"; +import { stopDigitalEmployeeSyncWorker } from "./services/digitalEmployee/syncService"; // macOS 26 Tahoe 兼容性:禁用 Fontations 字体后端 // 参考: https://github.com/electron/electron/issues/49522 @@ -574,6 +575,7 @@ app.on("before-quit", (e) => { log.info( "[App] Before quit - update install in progress, skipping preventDefault to allow installer", ); + stopDigitalEmployeeSyncWorker(); closeDb(); return; } @@ -586,6 +588,7 @@ app.on("before-quit", (e) => { void (async () => { const start = Date.now(); try { + stopDigitalEmployeeSyncWorker(); await cleanupAllProcesses(); } finally { const elapsed = Date.now() - start; diff --git a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts new file mode 100644 index 00000000..b3480842 --- /dev/null +++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts @@ -0,0 +1,363 @@ +import log from "electron-log"; +import { getDeviceId } from "../system/deviceId"; +import { getDb, readSetting } from "../../db"; + +const DEFAULT_SYNC_PATH = "/api/digital-employee/sync/outbox"; +const DEFAULT_INTERVAL_MS = 30_000; +const DEFAULT_BATCH_SIZE = 50; +const REQUEST_TIMEOUT_MS = 15_000; + +let syncTimer: ReturnType | null = null; +let syncing = false; +let lastSyncAt: string | null = null; +let lastSyncError: string | null = null; + +interface DigitalEmployeeSyncConfig { + enabled: boolean; + endpoint: string | null; + clientKey: string | null; + batchSize: number; +} + +interface OutboxRow { + id: string; + entity_type: string; + entity_id: string; + operation: string; + payload: string; + attempts: number; +} + +interface SyncAck { + outbox_id?: string; + id?: string; + entity_type?: string; + entity_id?: string; + remote_id?: string; +} + +interface SyncResponse { + success?: boolean; + acked?: SyncAck[]; + data?: { + acked?: SyncAck[]; + }; + message?: string; +} + +export interface DigitalEmployeeSyncStatus { + running: boolean; + syncing: boolean; + enabled: boolean; + endpoint: string | null; + pending: number; + failed: number; + lastSyncAt: string | null; + lastSyncError: string | null; +} + +export function startDigitalEmployeeSyncWorker(): void { + if (syncTimer) return; + syncTimer = setInterval(() => { + void flushDigitalEmployeeSyncOutbox().catch((error) => { + lastSyncError = normalizeError(error); + log.warn("[DigitalEmployeeSync] Scheduled sync failed:", error); + }); + }, DEFAULT_INTERVAL_MS); + void flushDigitalEmployeeSyncOutbox().catch((error) => { + lastSyncError = normalizeError(error); + log.debug("[DigitalEmployeeSync] Initial sync skipped/failed:", error); + }); + log.info("[DigitalEmployeeSync] Worker started"); +} + +export function stopDigitalEmployeeSyncWorker(): void { + if (!syncTimer) return; + clearInterval(syncTimer); + syncTimer = null; + log.info("[DigitalEmployeeSync] Worker stopped"); +} + +export function getDigitalEmployeeSyncStatus(): DigitalEmployeeSyncStatus { + const config = resolveSyncConfig(); + return { + running: Boolean(syncTimer), + syncing, + enabled: config.enabled, + endpoint: config.endpoint, + pending: countOutbox("pending"), + failed: countOutbox("failed"), + lastSyncAt, + lastSyncError, + }; +} + +export async function flushDigitalEmployeeSyncOutbox(): Promise { + if (syncing) return getDigitalEmployeeSyncStatus(); + + const config = resolveSyncConfig(); + if (!config.enabled || !config.endpoint || !config.clientKey) { + lastSyncError = null; + return getDigitalEmployeeSyncStatus(); + } + + const rows = readPendingOutbox(config.batchSize); + if (rows.length === 0) { + lastSyncError = null; + return getDigitalEmployeeSyncStatus(); + } + + syncing = true; + const now = new Date().toISOString(); + markRowsSyncing(rows, now); + + try { + const response = await postOutbox(config, rows); + const acked = response.data?.acked || response.acked || []; + if (response.success === false) { + throw new Error(response.message || "digital employee sync rejected"); + } + if (acked.length > 0) { + markRowsSynced(acked, now); + } else { + markRowsSynced( + rows.map((row) => ({ + outbox_id: row.id, + entity_type: row.entity_type, + entity_id: row.entity_id, + })), + now, + ); + } + lastSyncAt = now; + lastSyncError = null; + } catch (error) { + lastSyncError = normalizeError(error); + markRowsFailed(rows, now, lastSyncError); + log.warn("[DigitalEmployeeSync] Flush failed:", lastSyncError); + } finally { + syncing = false; + } + + return getDigitalEmployeeSyncStatus(); +} + +function resolveSyncConfig(): DigitalEmployeeSyncConfig { + const config = (readSetting("digital_employee_sync_config") || {}) as Record< + string, + unknown + >; + const step1 = (readSetting("step1_config") || {}) as Record; + const endpointOverride = + typeof config.endpoint === "string" && config.endpoint.trim() + ? config.endpoint.trim() + : null; + const serverHost = + typeof step1.serverHost === "string" && step1.serverHost.trim() + ? step1.serverHost.trim() + : null; + const endpoint = endpointOverride || buildEndpoint(serverHost); + const disabled = config.enabled === false; + const batchSize = + typeof config.batchSize === "number" && config.batchSize > 0 + ? Math.min(Math.floor(config.batchSize), 200) + : DEFAULT_BATCH_SIZE; + + return { + enabled: !disabled, + endpoint, + clientKey: readClientKey(serverHost), + batchSize, + }; +} + +function buildEndpoint(serverHost: string | null): string | null { + if (!serverHost) return null; + const normalized = serverHost.startsWith("http") + ? serverHost + : `https://${serverHost}`; + return `${normalized.replace(/\/+$/, "")}${DEFAULT_SYNC_PATH}`; +} + +function readClientKey(serverHost: string | null): string | null { + const username = readSetting("auth.username"); + const globalKey = readSetting("auth.saved_key"); + if (serverHost && typeof username === "string" && username.trim()) { + const domain = serverHost.replace(/^https?:\/\//, "").replace(/\/+$/, ""); + const domainKey = readSetting(`auth.saved_keys.${domain}_${username}`); + if (typeof domainKey === "string" && domainKey.trim()) return domainKey; + } + return typeof globalKey === "string" && globalKey.trim() ? globalKey : null; +} + +function readPendingOutbox(limit: number): OutboxRow[] { + const db = getDb(); + if (!db) return []; + return db + .prepare( + ` + SELECT id, entity_type, entity_id, operation, payload, attempts + FROM digital_sync_outbox + WHERE status IN ('pending', 'failed') + ORDER BY created_at ASC + LIMIT ? + `, + ) + .all(limit) as OutboxRow[]; +} + +async function postOutbox( + config: DigitalEmployeeSyncConfig, + rows: OutboxRow[], +): Promise { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS); + try { + const response = await fetch(config.endpoint!, { + method: "POST", + headers: { + "Content-Type": "application/json", + Authorization: `Bearer ${config.clientKey}`, + "X-Qiming-Client-Key": config.clientKey!, + }, + body: JSON.stringify({ + client_key: config.clientKey, + device_id: getDeviceId(), + items: rows.map((row) => ({ + outbox_id: row.id, + entity_type: row.entity_type, + entity_id: row.entity_id, + operation: row.operation, + payload: parsePayload(row.payload), + })), + }), + signal: controller.signal, + }); + const text = await response.text(); + const body = text ? (JSON.parse(text) as SyncResponse) : {}; + if (!response.ok) { + throw new Error(body.message || `HTTP ${response.status}`); + } + return body; + } finally { + clearTimeout(timer); + } +} + +function markRowsSyncing(rows: OutboxRow[], now: string): void { + const db = getDb(); + if (!db) return; + const tx = db.transaction(() => { + const stmt = db.prepare(` + UPDATE digital_sync_outbox + SET status = 'syncing', attempts = attempts + 1, last_attempt_at = ?, updated_at = ? + WHERE id = ? + `); + for (const row of rows) stmt.run(now, now, row.id); + }); + tx(); +} + +function markRowsSynced(acks: SyncAck[], now: string): void { + const db = getDb(); + if (!db) return; + const tx = db.transaction(() => { + const outboxStmt = db.prepare(` + UPDATE digital_sync_outbox + SET status = 'synced', error = NULL, updated_at = ? + WHERE id = ? + `); + for (const ack of acks) { + const outboxId = ack.outbox_id || ack.id; + if (!outboxId) continue; + outboxStmt.run(now, outboxId); + updateEntitySyncStatus(ack, now); + } + }); + tx(); +} + +function updateEntitySyncStatus(ack: SyncAck, now: string): void { + if (!ack.entity_type || !ack.entity_id) return; + const table = entityTable(ack.entity_type); + if (!table) return; + const db = getDb(); + if (!db) return; + db.prepare(` + UPDATE ${table} + SET sync_status = 'synced', + remote_id = COALESCE(?, remote_id), + last_synced_at = ?, + sync_error = NULL + WHERE id = ? + `).run(ack.remote_id || null, now, ack.entity_id); +} + +function markRowsFailed(rows: OutboxRow[], now: string, error: string): void { + const db = getDb(); + if (!db) return; + const tx = db.transaction(() => { + const stmt = db.prepare(` + UPDATE digital_sync_outbox + SET status = 'failed', error = ?, updated_at = ? + WHERE id = ? + `); + for (const row of rows) { + stmt.run(error, now, row.id); + markEntityFailed(row.entity_type, row.entity_id, error); + } + }); + tx(); +} + +function markEntityFailed(entityType: string, entityId: string, error: string): void { + const table = entityTable(entityType); + const db = getDb(); + if (!db || !table) return; + db.prepare(` + UPDATE ${table} + SET sync_status = 'failed', sync_error = ? + WHERE id = ? + `).run(error, entityId); +} + +function countOutbox(status: string): number { + const db = getDb(); + if (!db) return 0; + const row = db + .prepare("SELECT COUNT(*) AS count FROM digital_sync_outbox WHERE status = ?") + .get(status) as { count: number } | undefined; + return row?.count ?? 0; +} + +function entityTable(entityType: string): string | null { + switch (entityType) { + case "plan": + return "digital_plans"; + case "task": + return "digital_tasks"; + case "run": + return "digital_runs"; + case "event": + return "digital_events"; + case "artifact": + return "digital_artifacts"; + case "approval": + return "digital_approvals"; + default: + return null; + } +} + +function parsePayload(payload: string): unknown { + try { + return JSON.parse(payload); + } catch { + return payload; + } +} + +function normalizeError(error: unknown): string { + if (error instanceof Error) return error.message; + return String(error); +} diff --git a/qimingclaw/crates/agent-electron-client/src/preload/webviewPerfBridge.ts b/qimingclaw/crates/agent-electron-client/src/preload/webviewPerfBridge.ts index 730abc0d..7402b2ee 100644 --- a/qimingclaw/crates/agent-electron-client/src/preload/webviewPerfBridge.ts +++ b/qimingclaw/crates/agent-electron-client/src/preload/webviewPerfBridge.ts @@ -95,5 +95,11 @@ contextBridge.exposeInMainWorld("QimingClawBridge", { async getSnapshot() { return ipcRenderer.invoke("digitalEmployee:getSnapshot"); }, + async getSyncStatus() { + return ipcRenderer.invoke("digitalEmployee:getSyncStatus"); + }, + async flushSync() { + return ipcRenderer.invoke("digitalEmployee:flushSync"); + }, }, }); diff --git a/qimingclaw/docs/digital-employee-frontend-integration-plan.md b/qimingclaw/docs/digital-employee-frontend-integration-plan.md index 146250d5..66cf1976 100644 --- a/qimingclaw/docs/digital-employee-frontend-integration-plan.md +++ b/qimingclaw/docs/digital-employee-frontend-integration-plan.md @@ -412,9 +412,58 @@ crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts 下一步: -- 增加 sync worker:读取 `digital_sync_outbox`,推送管理端,回填 `remote_id` / `sync_status`。 - 确认 qiming-backend 的数字员工同步 API 路径和鉴权方式。 - 将 Artifact / Approval 从普通 Event 中拆成正式实体。 +- 数字员工页面展示同步状态:待同步、同步失败、最近同步时间。 + +## Phase 7:管理端 outbox 同步 worker(已开始) + +目标:SQLite 继续作为本地事实缓存和离线队列,同时把 Plan / Task / Run / Event 推送到管理端,避免数字员工数据停留在单机。 + +当前 qimingclaw 已新增: + +```text +crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts +crates/agent-electron-client/src/main/bootstrap/startup.ts +crates/agent-electron-client/src/main/ipc/digitalEmployeeHandlers.ts +crates/agent-electron-client/src/preload/webviewPerfBridge.ts +``` + +当前机制: + +- 启动后后台启动 sync worker,默认每 30 秒扫描 `digital_sync_outbox`。 +- 默认同步地址由 `step1_config.serverHost` 拼接: + +```text +/api/digital-employee/sync/outbox +``` + +- 可通过 settings `digital_employee_sync_config.endpoint` 覆盖同步地址。 +- 使用 `auth.saved_key` 或域名级 saved key 作为客户端凭证。 +- 请求同时携带 `Authorization: Bearer ` 和 `X-Qiming-Client-Key`,兼容后端后续选择。 +- 同步成功后回填 outbox 状态,并根据后端 ack 回填实体 `remote_id` / `last_synced_at`。 +- 同步失败时标记 failed 和错误信息,下轮继续补偿,不影响 `/computer/chat` 和 ACP 执行链路。 + +新增 IPC / Bridge: + +```text +digitalEmployee:getSyncStatus +digitalEmployee:flushSync +window.QimingClawBridge.digital.getSyncStatus() +window.QimingClawBridge.digital.flushSync() +``` + +当前边界: + +- 后端接口尚未真正确认,因此客户端先实现可配置 endpoint 和兼容 ack 协议。 +- worker 只消费 outbox,不直接侵入 Plan / Task / Run / Event 写入链路。 +- 管理端返回格式建议包含 `acked[]`,每项带 `outbox_id`、`entity_type`、`entity_id`、`remote_id`。 + +下一步: + +- 在 qiming-backend 落地 `/api/digital-employee/sync/outbox`。 +- 数字员工页面读取 `getSyncStatus()`,展示联动状态。 +- 增加 outbox 重试退避策略,避免接口长期不可用时频繁请求。 ## 管理端联动原则