diff --git a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts index c2afdd34..735fbd84 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts @@ -488,6 +488,126 @@ describe("digital employee sync service", () => { }); }); + it("syncs specialized event business views for management consumption", async () => { + insertEventEntity("event-route", "route_decision", { + source: "qimingclaw-route-decision", + routeDecision: { + id: "route-1", + route_kind: "PlanStepDispatch", + intent_kind: "schedule", + policy_result: "accepted", + reason_codes: ["ready_plan_step"], + candidate_skills: ["crm-search"], + created_command_id: "governance-command-1", + }, + }); + insertEventEntity("event-approval", "approval_action_recorded", { + approval_id: "approval-1", + action: "mark_risk", + status: "recorded", + actor: "operator", + comment_summary: { preview: "存在风险", length: 4 }, + delegate_summary: { target: "manager-1", label: "经理" }, + sla_summary: { due_at: "2026-06-08T09:00:00.000Z", sla_status: "tracked" }, + risk_summary: { risk_level: "high" }, + }); + insertEventEntity("event-artifact", "artifact_retention_marked", { + artifact_id: "artifact-1", + action: "mark_keep", + status: "recorded", + actor: "operator", + retention_status: "kept", + retention_until: "2026-07-01T00:00:00.000Z", + }); + insertEventEntity("event-skill", "skill_call_rejected", { + call_id: "call-1", + source: "qimingclaw-acp", + server_id: "server-1", + tool_name: "crm.search", + skill_id: "crm-search", + decision: "rejected", + reason_codes: ["skill_disabled"], + }); + insertEventEntity("event-dispatch", "plan_step_dispatch_failed", { + step_id: "step-1", + status: "failed", + reason: "computer chat failed", + }); + mockState.fetch.mockResolvedValue( + jsonResponse({ + success: true, + acked: [ + { entity_type: "event", entity_id: "event-route" }, + { entity_type: "event", entity_id: "event-approval" }, + { entity_type: "event", entity_id: "event-artifact" }, + { entity_type: "event", entity_id: "event-skill" }, + { entity_type: "event", entity_id: "event-dispatch" }, + ], + }), + ); + + const { flushDigitalEmployeeSyncOutbox } = await import("./syncService"); + const status = await flushDigitalEmployeeSyncOutbox({ force: true }); + const items = lastSyncRequestBody().items as Array<{ entity_id: string; business_view: Record }>; + const businessView = (eventId: string) => items.find((item) => item.entity_id === eventId)?.business_view; + + expect(status.failed).toBe(0); + expect(businessView("event-route")).toMatchObject({ + entity_type: "event", + local_id: "event-route", + category: "route", + kind: "route_decision", + route_decision_id: "route-1", + route_kind: "PlanStepDispatch", + intent_kind: "schedule", + policy_result: "accepted", + reason_codes: ["ready_plan_step"], + candidate_skills: ["crm-search"], + created_command_id: "governance-command-1", + attention_level: "action", + }); + expect(businessView("event-approval")).toMatchObject({ + category: "approval", + approval_id: "approval-1", + action: "mark_risk", + status: "recorded", + actor: "operator", + risk_level: "high", + due_at: "2026-06-08T09:00:00.000Z", + delegate_to: "manager-1", + comment_length: 4, + }); + expect(businessView("event-artifact")).toMatchObject({ + category: "artifact", + artifact_id: "artifact-1", + action: "mark_keep", + status: "recorded", + retention_status: "kept", + retention_until: "2026-07-01T00:00:00.000Z", + actor: "operator", + }); + expect(businessView("event-skill")).toMatchObject({ + category: "skill", + call_id: "call-1", + source: "qimingclaw-acp", + server_id: "server-1", + tool_name: "crm.search", + skill_id: "crm-search", + decision: "rejected", + reason_codes: ["skill_disabled"], + }); + expect(businessView("event-dispatch")).toMatchObject({ + category: "dispatch", + step_id: "step-1", + status: "failed", + reason: "computer chat failed", + }); + expect(readEntity("event", "event-route")).toMatchObject({ + sync_status: "synced", + sync_error: null, + }); + }); + it("marks memory sync failures as failed", async () => { insertEntity("memory", "memory-rejected"); insertOutbox("memory:upsert:memory-rejected", "memory", "memory-rejected"); @@ -524,17 +644,43 @@ function insertEntity(entityType: string, id: string): void { }); } +function insertEventEntity(id: string, kind: string, payload: Record): void { + mockState.db?.events.set(id, { + id, + plan_id: "plan-1", + task_id: "task-1", + run_id: "run-1", + kind, + message: `事件 ${kind}`, + payload: JSON.stringify(payload), + remote_id: null, + sync_status: "pending", + sync_error: null, + last_synced_at: null, + occurred_at: "2026-06-07T07:59:00.000Z", + }); + insertOutbox(`event:insert:${id}`, "event", id, JSON.stringify({ + event_id: id, + kind, + message: `事件 ${kind}`, + plan_id: "plan-1", + task_id: "task-1", + occurred_at: "2026-06-07T07:59:00.000Z", + }), "insert"); +} + function insertOutbox( id: string, entityType: string, entityId: string, payload = '{"ok":true}', + operation = "upsert", ): void { mockState.db?.outbox.set(id, { id, entity_type: entityType, entity_id: entityId, - operation: "upsert", + operation, payload, status: "pending", attempts: 0, @@ -571,6 +717,7 @@ function entityMap( if (entityType === "artifact") return mockState.db.artifacts; if (entityType === "approval") return mockState.db.approvals; if (entityType === "memory") return mockState.db.memories; + if (entityType === "event") return mockState.db.events; return undefined; } @@ -611,6 +758,12 @@ interface TestSyncEntityRow { status?: string; objective?: string | null; payload?: string; + plan_id?: string | null; + task_id?: string | null; + run_id?: string | null; + kind?: string; + message?: string; + occurred_at?: string; remote_id: string | null; sync_status: string; sync_error: string | null; @@ -640,6 +793,7 @@ class TestDb { artifacts = new Map(); approvals = new Map(); memories = new Map(); + events = new Map(); attempts: TestAttemptRow[] = []; private nextAttemptId = 1; @@ -760,6 +914,23 @@ class TestDb { return this.scheduleRuns.get(id); } + if (sql.startsWith("SELECT kind AS title, kind AS status, message, payload FROM digital_events")) { + const [id] = args as [string]; + const row = this.events.get(id); + if (!row) return undefined; + return { + title: row.kind, + status: row.kind, + message: row.message, + payload: row.payload, + }; + } + + if (sql.startsWith("SELECT plan_id, task_id, run_id, kind, message, payload, occurred_at FROM digital_events")) { + const [id] = args as [string]; + return this.events.get(id); + } + if ( sql.includes("COUNT(*) AS count FROM digital_sync_outbox WHERE status = ?") ) { @@ -857,7 +1028,7 @@ class TestDb { return { changes: before - this.attempts.length }; } - if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|artifacts|approvals|memories) SET sync_status = 'synced'/.test(sql)) { + if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|events|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) { @@ -869,7 +1040,7 @@ class TestDb { return { changes: row ? 1 : 0 }; } - if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|artifacts|approvals|memories) SET sync_status = 'failed'/.test(sql)) { + if (/^UPDATE digital_(plans|plan_steps|schedules|schedule_runs|events|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) { @@ -886,6 +1057,7 @@ class TestDb { 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_events")) return this.events; 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; 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 index 089c9a34..62f614f1 100644 --- a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts +++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts @@ -105,6 +105,16 @@ interface SyncResponse { message?: string; } +interface EventBusinessDetail { + plan_id: string | null; + task_id: string | null; + run_id: string | null; + kind: string | null; + message: string | null; + payload: Record; + occurred_at: string | null; +} + export interface DigitalEmployeeSyncStatus { running: boolean; syncing: boolean; @@ -469,15 +479,7 @@ function buildBusinessView( finished_at: stringField(record.finishedAt ?? record.finished_at) || null, }; case "event": - return { - ...base, - plan_id: stringField(record.plan_id ?? record.planId) || null, - task_id: stringField(record.task_id ?? record.taskId) || null, - run_id: stringField(record.run_id ?? record.runId) || null, - kind: stringField(record.kind) || summary?.status || null, - message: stringField(record.message) || summary?.summary || null, - occurred_at: stringField(record.occurred_at ?? record.occurredAt) || null, - }; + return buildEventBusinessView(base, row, record, summary); case "artifact": return { ...base, @@ -509,6 +511,149 @@ function buildBusinessView( } } +function buildEventBusinessView( + base: Record, + row: OutboxRow, + record: Record, + summary: DigitalEmployeeSyncEntitySummary | null, +): Record { + const detail = readEventBusinessDetail(row.entity_id, record, summary); + const kind = detail.kind || "event"; + return { + ...base, + category: eventBusinessCategory(kind), + plan_id: detail.plan_id, + task_id: detail.task_id, + run_id: detail.run_id, + kind, + message: detail.message, + occurred_at: detail.occurred_at, + ...eventSpecificBusinessView(kind, detail.payload, row.entity_id), + }; +} + +function readEventBusinessDetail( + eventId: string, + fallback: Record, + summary: DigitalEmployeeSyncEntitySummary | null, +): EventBusinessDetail { + const db = getDigitalEmployeeDb(); + const row = db?.prepare(` + SELECT plan_id, task_id, run_id, kind, message, payload, occurred_at + FROM digital_events + WHERE id = ? + `).get(eventId) as Record | undefined; + const eventPayload = parsePayload(stringField(row?.payload) || "{}"); + const fallbackPayload = objectRecord(fallback.payload) ?? fallback; + return { + plan_id: stringField(row?.plan_id ?? fallback.plan_id ?? fallback.planId) || null, + task_id: stringField(row?.task_id ?? fallback.task_id ?? fallback.taskId) || null, + run_id: stringField(row?.run_id ?? fallback.run_id ?? fallback.runId) || null, + kind: stringField(row?.kind ?? fallback.kind) || summary?.status || null, + message: stringField(row?.message ?? fallback.message) || summary?.summary || null, + payload: objectRecord(eventPayload) ?? fallbackPayload, + occurred_at: stringField(row?.occurred_at ?? fallback.occurred_at ?? fallback.occurredAt) || null, + }; +} + +function eventSpecificBusinessView( + kind: string, + payload: Record, + eventId: string, +): Record { + if (kind === "route_decision") return routeDecisionBusinessView(payload, eventId); + if (kind.startsWith("approval_action_")) return approvalActionBusinessView(payload); + if (kind.startsWith("artifact_access_") || kind.startsWith("artifact_retention_") || kind === "artifact_version_observed") { + return artifactEventBusinessView(payload); + } + if (kind.startsWith("skill_call_")) return skillCallBusinessView(payload); + if (kind.startsWith("plan_step_dispatch_")) return planStepDispatchBusinessView(payload); + return {}; +} + +function routeDecisionBusinessView(payload: Record, eventId: string): Record { + const decision = objectRecord(payload.routeDecision ?? payload.route_decision) ?? payload; + const policyResult = stringField(decision.policy_result ?? decision.policyResult) || "accepted"; + const reasonCodes = stringArrayField(decision.reason_codes ?? decision.reasonCodes); + const createdCommandId = stringField(decision.created_command_id ?? decision.createdCommandId) || null; + return { + route_decision_id: stringField(decision.id) || eventId, + route_kind: stringField(decision.route_kind ?? decision.routeKind) || "ChatOnly", + intent_kind: stringField(decision.intent_kind ?? decision.intentKind) || "chat", + policy_result: policyResult, + reason_codes: reasonCodes, + candidate_skills: stringArrayField(decision.candidate_skills ?? decision.candidateSkills), + created_command_id: createdCommandId, + attention_level: routeDecisionAttentionLevel(policyResult, reasonCodes, createdCommandId), + }; +} + +function approvalActionBusinessView(payload: Record): Record { + const comment = objectRecord(payload.comment_summary); + const delegate = objectRecord(payload.delegate_summary); + const sla = objectRecord(payload.sla_summary); + const risk = objectRecord(payload.risk_summary); + return { + approval_id: stringField(payload.approval_id ?? payload.approvalId) || null, + action: stringField(payload.action) || null, + status: stringField(payload.status) || null, + actor: stringField(payload.actor) || null, + risk_level: stringField(risk?.risk_level ?? payload.risk_level) || null, + due_at: stringField(sla?.due_at ?? payload.due_at) || null, + delegate_to: stringField(delegate?.target ?? delegate?.label ?? payload.delegate_to) || null, + comment_length: numberField(comment?.length ?? payload.comment_length), + }; +} + +function artifactEventBusinessView(payload: Record): Record { + const retention = objectRecord(payload.retention_summary); + return { + artifact_id: stringField(payload.artifact_id ?? payload.artifactId) || null, + action: stringField(payload.action) || null, + status: stringField(payload.status) || null, + retention_status: stringField(payload.retention_status ?? retention?.retention_status) || null, + retention_until: stringField(payload.retention_until ?? retention?.retention_until) || null, + actor: stringField(payload.actor) || null, + }; +} + +function skillCallBusinessView(payload: Record): Record { + return { + call_id: stringField(payload.call_id ?? payload.callId) || null, + source: stringField(payload.source) || null, + server_id: stringField(payload.server_id ?? payload.serverId) || null, + tool_name: stringField(payload.tool_name ?? payload.toolName) || null, + skill_id: stringField(payload.skill_id ?? payload.skillId) || null, + decision: stringField(payload.decision) || null, + reason_codes: stringArrayField(payload.reason_codes ?? payload.reasonCodes), + }; +} + +function planStepDispatchBusinessView(payload: Record): Record { + return { + step_id: stringField(payload.step_id ?? payload.stepId) || null, + status: stringField(payload.status) || null, + reason: stringField(payload.reason) || null, + }; +} + +function eventBusinessCategory(kind: string): string { + if (kind === "route_decision") return "route"; + if (kind.startsWith("approval_")) return "approval"; + if (kind.startsWith("artifact_")) return "artifact"; + if (kind.startsWith("skill_call_")) return "skill"; + if (kind.startsWith("plan_step_dispatch_")) return "dispatch"; + if (kind.startsWith("governance_")) return "governance"; + return "runtime"; +} + +function routeDecisionAttentionLevel(policyResult: string, reasonCodes: string[], createdCommandId: string | null): "info" | "action" | "warn" | "error" { + if (policyResult && policyResult !== "accepted") return "error"; + if (reasonCodes.includes("candidate_skills_disabled") || reasonCodes.includes("route_action_missing_context")) return "warn"; + if (createdCommandId) return "action"; + return "info"; +} + function compactPreview(value: string, maxLength = 120): string | null { const compact = value.replace(/\s+/g, " ").trim(); if (!compact) return null; @@ -799,6 +944,20 @@ function stringField(value: unknown): string { return typeof value === "string" && value.trim() ? value.trim() : ""; } +function stringArrayField(value: unknown): string[] { + if (!Array.isArray(value)) return []; + return value.map(stringField).filter(Boolean); +} + +function numberField(value: unknown): number | null { + if (typeof value === "number" && Number.isFinite(value)) return value; + if (typeof value === "string" && value.trim()) { + const parsed = Number(value); + return Number.isFinite(parsed) ? parsed : null; + } + return null; +} + function objectRecord(value: unknown): Record | null { return value && typeof value === "object" && !Array.isArray(value) ? value as Record diff --git a/qimingclaw/docs/digital-employee-frontend-integration-plan.md b/qimingclaw/docs/digital-employee-frontend-integration-plan.md index dc8b6002..72b830cd 100644 --- a/qimingclaw/docs/digital-employee-frontend-integration-plan.md +++ b/qimingclaw/docs/digital-employee-frontend-integration-plan.md @@ -599,7 +599,7 @@ GET /api/digital-employee/approvals - 建议:下一轮围绕 sandbox audit 与 token/cost 采集,把只读投影升级为可追溯的真实审计与成本统计。 12. **管理端业务查询模型** - - 已吸收:管理端已有通用 sync record 接收、查询、payload 摘要和深链;embedded adapter 已拆出 `/api/digital-employee/plans`、`/tasks`、`/runs`、`/events`、`/artifacts`、`/approvals` 与 `/plans/:planId` 只读业务查询接口,从 qimingclaw runtime、artifact、approval 和 canonical event 投影生成统一分页视图,本地业务视图投影已开始闭环。 + - 已吸收:管理端已有通用 sync record 接收、查询、payload 摘要和深链;embedded adapter 已拆出 `/api/digital-employee/plans`、`/tasks`、`/runs`、`/events`、`/artifacts`、`/approvals` 与 `/plans/:planId` 只读业务查询接口,从 qimingclaw runtime、artifact、approval 和 canonical event 投影生成统一分页视图;客户端 outbox 的 event `business_view` 已开始按 route decision、approval action、artifact lifecycle/access、skill call audit 和 ready PlanStep dispatch 细分常用字段,本地业务视图投影已开始闭环。 - 缺口:尚未接入管理端真实业务表、复杂组合检索、跨设备维度过滤和业务视图 UI。 - 建议:先稳定本地投影 API 和管理端消费模型,再决定是否拆物理业务表。