feat(client): mirror agent events to digital employee
This commit is contained in:
@@ -2,6 +2,7 @@ import { agentService } from "../services/engines/unifiedAgent";
|
|||||||
import type { UnifiedSessionMessage } from "../services/engines/unifiedAgent";
|
import type { UnifiedSessionMessage } from "../services/engines/unifiedAgent";
|
||||||
import { pushSseEvent } from "../services/computerServer";
|
import { pushSseEvent } from "../services/computerServer";
|
||||||
import { firstTokenTrace } from "../services/engines/perf/firstTokenTrace";
|
import { firstTokenTrace } from "../services/engines/perf/firstTokenTrace";
|
||||||
|
import { recordDigitalEmployeeRuntimeEvent } from "../services/digitalEmployee/stateService";
|
||||||
import type { HandlerContext } from "@shared/types/ipc";
|
import type { HandlerContext } from "@shared/types/ipc";
|
||||||
import log from "electron-log";
|
import log from "electron-log";
|
||||||
|
|
||||||
@@ -11,6 +12,34 @@ const registeredHandlers: Array<{
|
|||||||
handler: (...args: any[]) => void;
|
handler: (...args: any[]) => void;
|
||||||
}> = [];
|
}> = [];
|
||||||
|
|
||||||
|
function readEventField(data: unknown, keys: string[]): string | undefined {
|
||||||
|
if (!data || typeof data !== "object") return undefined;
|
||||||
|
const record = data as Record<string, unknown>;
|
||||||
|
for (const key of keys) {
|
||||||
|
const value = record[key];
|
||||||
|
if (typeof value === "string" && value.trim()) return value;
|
||||||
|
}
|
||||||
|
return undefined;
|
||||||
|
}
|
||||||
|
|
||||||
|
function recordForwardedEvent(eventType: string, data: unknown): void {
|
||||||
|
recordDigitalEmployeeRuntimeEvent({
|
||||||
|
kind: eventType,
|
||||||
|
message:
|
||||||
|
readEventField(data, ["message", "description", "title", "subType"]) ||
|
||||||
|
eventType,
|
||||||
|
request_id: readEventField(data, ["requestId", "request_id"]),
|
||||||
|
session_id: readEventField(data, ["sessionId", "session_id", "id"]),
|
||||||
|
project_id: readEventField(data, ["projectId", "project_id"]),
|
||||||
|
status: eventType.includes("error")
|
||||||
|
? "failed"
|
||||||
|
: eventType.includes("idle") || eventType.includes("promptEnd")
|
||||||
|
? "completed"
|
||||||
|
: "running",
|
||||||
|
payload: data,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Unregister all event forwarders from agentService.
|
* Unregister all event forwarders from agentService.
|
||||||
* Call before re-registering or during cleanup.
|
* Call before re-registering or during cleanup.
|
||||||
@@ -65,6 +94,7 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
type: eventType,
|
type: eventType,
|
||||||
data,
|
data,
|
||||||
});
|
});
|
||||||
|
recordForwardedEvent(eventType, data);
|
||||||
};
|
};
|
||||||
agentService.on(eventType, handler);
|
agentService.on(eventType, handler);
|
||||||
registeredHandlers.push({ event: eventType, handler });
|
registeredHandlers.push({ event: eventType, handler });
|
||||||
@@ -75,6 +105,12 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
type: "error",
|
type: "error",
|
||||||
data: { message: error.message },
|
data: { message: error.message },
|
||||||
});
|
});
|
||||||
|
recordDigitalEmployeeRuntimeEvent({
|
||||||
|
kind: "agent_error",
|
||||||
|
message: error.message,
|
||||||
|
status: "failed",
|
||||||
|
payload: { message: error.message },
|
||||||
|
});
|
||||||
};
|
};
|
||||||
agentService.on("error", errorHandler);
|
agentService.on("error", errorHandler);
|
||||||
registeredHandlers.push({ event: "error", handler: errorHandler });
|
registeredHandlers.push({ event: "error", handler: errorHandler });
|
||||||
@@ -84,6 +120,11 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
type: "ready",
|
type: "ready",
|
||||||
data: {},
|
data: {},
|
||||||
});
|
});
|
||||||
|
recordDigitalEmployeeRuntimeEvent({
|
||||||
|
kind: "agent_ready",
|
||||||
|
message: "Agent service ready",
|
||||||
|
status: "running",
|
||||||
|
});
|
||||||
};
|
};
|
||||||
agentService.on("ready", readyHandler);
|
agentService.on("ready", readyHandler);
|
||||||
registeredHandlers.push({ event: "ready", handler: readyHandler });
|
registeredHandlers.push({ event: "ready", handler: readyHandler });
|
||||||
@@ -93,6 +134,11 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
type: "destroyed",
|
type: "destroyed",
|
||||||
data: {},
|
data: {},
|
||||||
});
|
});
|
||||||
|
recordDigitalEmployeeRuntimeEvent({
|
||||||
|
kind: "agent_destroyed",
|
||||||
|
message: "Agent service destroyed",
|
||||||
|
status: "completed",
|
||||||
|
});
|
||||||
};
|
};
|
||||||
agentService.on("destroyed", destroyedHandler);
|
agentService.on("destroyed", destroyedHandler);
|
||||||
registeredHandlers.push({ event: "destroyed", handler: destroyedHandler });
|
registeredHandlers.push({ event: "destroyed", handler: destroyedHandler });
|
||||||
@@ -105,6 +151,7 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
JSON.stringify(data).substring(0, 200),
|
JSON.stringify(data).substring(0, 200),
|
||||||
);
|
);
|
||||||
ctx.getMainWindow()?.webContents.send("computer:progress", data);
|
ctx.getMainWindow()?.webContents.send("computer:progress", data);
|
||||||
|
recordForwardedEvent("computer:progress", data);
|
||||||
const d = data as UnifiedSessionMessage;
|
const d = data as UnifiedSessionMessage;
|
||||||
if (d?.sessionId) {
|
if (d?.sessionId) {
|
||||||
// sessionId is now the ACP protocol UUID — same as acpSessionId
|
// sessionId is now the ACP protocol UUID — same as acpSessionId
|
||||||
@@ -139,6 +186,14 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
timestamp: new Date().toISOString(),
|
timestamp: new Date().toISOString(),
|
||||||
};
|
};
|
||||||
ctx.getMainWindow()?.webContents.send("computer:progress", event);
|
ctx.getMainWindow()?.webContents.send("computer:progress", event);
|
||||||
|
recordDigitalEmployeeRuntimeEvent({
|
||||||
|
kind: "computer_prompt_start",
|
||||||
|
message: "Computer prompt started",
|
||||||
|
request_id: data.requestId,
|
||||||
|
session_id: data.sessionId,
|
||||||
|
status: "running",
|
||||||
|
payload: event,
|
||||||
|
});
|
||||||
pushSseEvent(data.sessionId, "prompt_start", event);
|
pushSseEvent(data.sessionId, "prompt_start", event);
|
||||||
};
|
};
|
||||||
agentService.on("computer:promptStart", promptStartHandler);
|
agentService.on("computer:promptStart", promptStartHandler);
|
||||||
@@ -166,6 +221,13 @@ export function registerEventForwarders(ctx: HandlerContext): void {
|
|||||||
timestamp: new Date().toISOString(),
|
timestamp: new Date().toISOString(),
|
||||||
};
|
};
|
||||||
ctx.getMainWindow()?.webContents.send("computer:progress", event);
|
ctx.getMainWindow()?.webContents.send("computer:progress", event);
|
||||||
|
recordDigitalEmployeeRuntimeEvent({
|
||||||
|
kind: "computer_prompt_end",
|
||||||
|
message: data.description || data.reason || "Computer prompt ended",
|
||||||
|
session_id: data.sessionId,
|
||||||
|
status: data.reason === "error" ? "failed" : "completed",
|
||||||
|
payload: event,
|
||||||
|
});
|
||||||
pushSseEvent(data.sessionId, data.reason || "end_turn", event);
|
pushSseEvent(data.sessionId, data.reason || "end_turn", event);
|
||||||
};
|
};
|
||||||
agentService.on("computer:promptEnd", promptEndHandler);
|
agentService.on("computer:promptEnd", promptEndHandler);
|
||||||
|
|||||||
@@ -67,6 +67,16 @@ export interface DigitalEmployeeChatResultRecord {
|
|||||||
engine?: string;
|
engine?: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export interface DigitalEmployeeRuntimeEventRecord {
|
||||||
|
kind: string;
|
||||||
|
message?: string;
|
||||||
|
request_id?: string;
|
||||||
|
session_id?: string;
|
||||||
|
project_id?: string;
|
||||||
|
status?: "running" | "completed" | "failed" | "pending";
|
||||||
|
payload?: unknown;
|
||||||
|
}
|
||||||
|
|
||||||
function defaultState(): DigitalEmployeeState {
|
function defaultState(): DigitalEmployeeState {
|
||||||
return {
|
return {
|
||||||
version: 1,
|
version: 1,
|
||||||
@@ -230,6 +240,66 @@ export function recordDigitalEmployeeChatResult(
|
|||||||
tx();
|
tx();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function recordDigitalEmployeeRuntimeEvent(
|
||||||
|
event: DigitalEmployeeRuntimeEventRecord,
|
||||||
|
): void {
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
const ids = chatIds({
|
||||||
|
request_id: event.request_id,
|
||||||
|
session_id: event.session_id,
|
||||||
|
project_id: event.project_id,
|
||||||
|
prompt: event.message,
|
||||||
|
});
|
||||||
|
const status = event.status ?? "running";
|
||||||
|
const title = event.message || event.kind;
|
||||||
|
const db = getDb();
|
||||||
|
if (!db) return;
|
||||||
|
|
||||||
|
const tx = db.transaction(() => {
|
||||||
|
upsertPlan({
|
||||||
|
id: ids.planId,
|
||||||
|
title: `运行事件:${title}`,
|
||||||
|
objective: title,
|
||||||
|
status,
|
||||||
|
payload: event.payload ?? event,
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
upsertTask({
|
||||||
|
id: ids.taskId,
|
||||||
|
planId: ids.planId,
|
||||||
|
title,
|
||||||
|
status,
|
||||||
|
assignedSkillId: "qimingclaw-acp-session",
|
||||||
|
payload: event.payload ?? event,
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
upsertRun({
|
||||||
|
id: ids.runId,
|
||||||
|
planId: ids.planId,
|
||||||
|
taskId: ids.taskId,
|
||||||
|
status,
|
||||||
|
payload: event.payload ?? event,
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
if (status === "completed" || status === "failed") {
|
||||||
|
finishRun(ids.runId, status, now);
|
||||||
|
}
|
||||||
|
insertEvent(
|
||||||
|
{
|
||||||
|
event_id: `${ids.runId}:${event.kind}:${Date.now()}`,
|
||||||
|
kind: event.kind,
|
||||||
|
occurred_at: now,
|
||||||
|
message: title,
|
||||||
|
plan_id: ids.planId,
|
||||||
|
task_id: ids.taskId,
|
||||||
|
},
|
||||||
|
ids.runId,
|
||||||
|
event.payload ?? event,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
tx();
|
||||||
|
}
|
||||||
|
|
||||||
function persistSnapshotTables(
|
function persistSnapshotTables(
|
||||||
snapshot: DigitalEmployeeSnapshot,
|
snapshot: DigitalEmployeeSnapshot,
|
||||||
event: DigitalEmployeeStateEvent | null,
|
event: DigitalEmployeeStateEvent | null,
|
||||||
|
|||||||
@@ -379,6 +379,43 @@ crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts
|
|||||||
- 接 ACP event forwarder,把 token、tool、step、artifact 写入 Run/Event/Artifact。
|
- 接 ACP event forwarder,把 token、tool、step、artifact 写入 Run/Event/Artifact。
|
||||||
- 增加 sync worker,根据 `step1_config.serverHost` 和登录 token 推送 outbox。
|
- 增加 sync worker,根据 `step1_config.serverHost` 和登录 token 推送 outbox。
|
||||||
|
|
||||||
|
## Phase 6:ACP / computer 运行事件落库(已开始)
|
||||||
|
|
||||||
|
目标:让数字员工不只知道“管理端下发了任务”,也能看到 qimingclaw 执行过程中的真实运行事件。
|
||||||
|
|
||||||
|
当前 qimingclaw 已接入:
|
||||||
|
|
||||||
|
```text
|
||||||
|
crates/agent-electron-client/src/main/ipc/eventForwarders.ts
|
||||||
|
crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts
|
||||||
|
```
|
||||||
|
|
||||||
|
记录来源:
|
||||||
|
|
||||||
|
- ACP session SSE 事件:`message.updated`、`session.updated`、`session.status`、`session.idle`、`session.error` 等。
|
||||||
|
- computer 执行进度:`computer:progress`。
|
||||||
|
- prompt 生命周期:`computer:promptStart`、`computer:promptEnd`。
|
||||||
|
- Agent 生命周期:`ready`、`destroyed`、`error`。
|
||||||
|
|
||||||
|
当前落库方式:
|
||||||
|
|
||||||
|
- 运行中的事件会创建/更新对应 Plan、Task、Run。
|
||||||
|
- 完成和失败事件会结束 Run。
|
||||||
|
- 每条 Event 都写入 `digital_events`。
|
||||||
|
- Plan / Task / Run / Event 均进入 `digital_sync_outbox`,等待后续同步到管理端。
|
||||||
|
|
||||||
|
与管理端联动的边界:
|
||||||
|
|
||||||
|
- SQLite 作为本地事实缓存和离线队列,不作为最终中心库。
|
||||||
|
- 管理端 API 未确定前,qimingclaw 先沉淀统一的本地实体和 outbox。
|
||||||
|
- 后续 sync worker 只消费 `digital_sync_outbox`,避免业务写入链路直接依赖网络。
|
||||||
|
|
||||||
|
下一步:
|
||||||
|
|
||||||
|
- 增加 sync worker:读取 `digital_sync_outbox`,推送管理端,回填 `remote_id` / `sync_status`。
|
||||||
|
- 确认 qiming-backend 的数字员工同步 API 路径和鉴权方式。
|
||||||
|
- 将 Artifact / Approval 从普通 Event 中拆成正式实体。
|
||||||
|
|
||||||
## 管理端联动原则
|
## 管理端联动原则
|
||||||
|
|
||||||
数字员工状态是本地优先,但不能成为本地孤岛。
|
数字员工状态是本地优先,但不能成为本地孤岛。
|
||||||
|
|||||||
Reference in New Issue
Block a user