399 lines
14 KiB
TypeScript
399 lines
14 KiB
TypeScript
import { agentService } from "../services/engines/unifiedAgent";
|
|
import type { UnifiedSessionMessage } from "../services/engines/unifiedAgent";
|
|
import { pushSseEvent } from "../services/computerServer";
|
|
import { firstTokenTrace } from "../services/engines/perf/firstTokenTrace";
|
|
import {
|
|
recordDigitalEmployeeRuntimeEvent,
|
|
recordDigitalEmployeeSkillCallAudit,
|
|
} from "../services/digitalEmployee/stateService";
|
|
import { evaluateDigitalEmployeeSkillCall } from "../services/digitalEmployee/skillExecutionPolicy";
|
|
import { mcpProxyManager } from "../services/packages/mcp";
|
|
import type { HandlerContext } from "@shared/types/ipc";
|
|
import log from "electron-log";
|
|
|
|
/** Stored handlers for cleanup */
|
|
const registeredHandlers: Array<{
|
|
event: string;
|
|
handler: (...args: any[]) => void;
|
|
}> = [];
|
|
|
|
function readEventField(data: unknown, keys: string[]): string | undefined {
|
|
for (const record of eventRecords(data)) {
|
|
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 {
|
|
const normalized = normalizeForwardedEvent(eventType, data);
|
|
recordDigitalEmployeeRuntimeEvent(normalized);
|
|
recordSkillCallAuditForForwardedEvent(eventType, data, normalized);
|
|
}
|
|
|
|
function recordSkillCallAuditForForwardedEvent(
|
|
eventType: string,
|
|
data: unknown,
|
|
normalized: Parameters<typeof recordDigitalEmployeeRuntimeEvent>[0],
|
|
): void {
|
|
const subtype = readEventField(data, ["subType", "sessionUpdate", "type", "messageType"]);
|
|
if (eventType !== "computer:progress" || (subtype !== "tool_call" && subtype !== "tool_call_update")) {
|
|
return;
|
|
}
|
|
try {
|
|
const toolName = readEventField(data, ["toolTitle", "title", "name", "kind"]);
|
|
const policy = evaluateDigitalEmployeeSkillCall({
|
|
source: "qimingclaw-acp",
|
|
toolName,
|
|
callId: readEventField(data, ["toolCallId", "id"]) || `${subtype}:${Date.now()}`,
|
|
sessionId: readEventField(data, ["sessionId", "session_id", "id"]),
|
|
mcpConfig: mcpProxyManager.getConfig(),
|
|
});
|
|
recordDigitalEmployeeSkillCallAudit({
|
|
callId: readEventField(data, ["toolCallId", "id"]) || `${subtype}:${Date.now()}`,
|
|
source: policy.source,
|
|
serverId: policy.server_id,
|
|
toolName: policy.tool_name || toolName,
|
|
skillId: policy.skill_id,
|
|
decision: policy.decision,
|
|
reasonCodes: policy.reason_codes,
|
|
policySnapshot: policy.policy_snapshot,
|
|
sessionId: normalized.session_id,
|
|
status: readEventField(data, ["status"]) || normalized.status,
|
|
});
|
|
} catch (error) {
|
|
log.warn("[EventForwarders] Failed to record skill call audit:", error);
|
|
}
|
|
}
|
|
|
|
function normalizeForwardedEvent(
|
|
eventType: string,
|
|
data: unknown,
|
|
): Parameters<typeof recordDigitalEmployeeRuntimeEvent>[0] {
|
|
const subtype = readEventField(data, ["subType", "sessionUpdate", "type", "messageType"]);
|
|
const toolName = readEventField(data, ["toolTitle", "title", "name", "kind"]);
|
|
const message = eventMessage(eventType, data, subtype, toolName);
|
|
return {
|
|
kind: eventKind(eventType, subtype),
|
|
message,
|
|
request_id: readEventField(data, ["requestId", "request_id"]),
|
|
session_id: readEventField(data, ["sessionId", "session_id", "id"]),
|
|
project_id: readEventField(data, ["projectId", "project_id"]),
|
|
status: eventStatus(eventType, data),
|
|
payload: runtimePayload(eventType, data, subtype, toolName),
|
|
};
|
|
}
|
|
|
|
function eventKind(eventType: string, subtype?: string): string {
|
|
if (eventType === "computer:progress" && subtype) return `computer_${subtype}`;
|
|
return eventType.replace(/[^a-zA-Z0-9]+/g, "_").replace(/^_+|_+$/g, "") || eventType;
|
|
}
|
|
|
|
function eventMessage(
|
|
eventType: string,
|
|
data: unknown,
|
|
subtype?: string,
|
|
toolName?: string,
|
|
): string {
|
|
if (subtype === "tool_call") return `工具调用开始：${toolName || "unknown"}`;
|
|
if (subtype === "tool_call_update") return `工具调用更新：${toolName || "unknown"}`;
|
|
if (subtype === "agent_message_chunk") return "Agent 输出内容";
|
|
if (subtype === "agent_thought_chunk") return "Agent 思考内容";
|
|
const explicit = readEventField(data, ["message", "description", "title"]);
|
|
if (explicit) return explicit;
|
|
if (eventType === "permission.updated") return `权限请求：${toolName || "待确认"}`;
|
|
if (eventType === "permission.replied") return `权限请求已回复：${toolName || "已处理"}`;
|
|
if (eventType === "file.edited") return `文件已编辑：${readEventField(data, ["filePath", "file_path", "path"]) || "unknown"}`;
|
|
if (eventType === "session.diff") return "会话产生文件差异";
|
|
return subtype || eventType;
|
|
}
|
|
|
|
function eventStatus(
|
|
eventType: string,
|
|
data: unknown,
|
|
): "running" | "completed" | "failed" | "pending" {
|
|
if (eventType.includes("error")) return "failed";
|
|
const status = (readEventField(data, ["status", "outcome", "reason"]) || "").toLowerCase();
|
|
if (["failed", "failure", "error", "rejected", "denied", "cancelled", "canceled"].includes(status)) return "failed";
|
|
if (["completed", "complete", "success", "succeeded", "done", "accepted", "approved", "end_turn"].includes(status)) return "completed";
|
|
if (eventType.includes("idle") || eventType.includes("promptEnd")) return "completed";
|
|
if (eventType.includes("permission")) return eventType.endsWith("replied") ? "completed" : "pending";
|
|
return "running";
|
|
}
|
|
|
|
function runtimePayload(
|
|
eventType: string,
|
|
data: unknown,
|
|
subtype?: string,
|
|
toolName?: string,
|
|
): Record<string, unknown> {
|
|
const primary = objectRecord(data) ?? {};
|
|
const nested = objectRecord(primary.data) ?? {};
|
|
const payload: Record<string, unknown> = {
|
|
...primary,
|
|
eventType,
|
|
subtype,
|
|
toolName,
|
|
};
|
|
const output = nested.rawOutput ?? nested.output ?? primary.output;
|
|
if (output !== undefined) payload.output = output;
|
|
const rawInput = nested.rawInput ?? nested.input ?? primary.input;
|
|
if (rawInput !== undefined) payload.input = rawInput;
|
|
const locations = Array.isArray(nested.locations) ? nested.locations : Array.isArray(primary.locations) ? primary.locations : [];
|
|
const artifacts = artifactHints(eventType, primary, nested, locations);
|
|
if (artifacts.length > 0) payload.artifacts = artifacts;
|
|
const approval = approvalHint(eventType, primary, nested, toolName);
|
|
if (approval) payload.approval_requests = [approval];
|
|
return payload;
|
|
}
|
|
|
|
function artifactHints(
|
|
eventType: string,
|
|
primary: Record<string, unknown>,
|
|
nested: Record<string, unknown>,
|
|
locations: unknown[],
|
|
): Array<Record<string, unknown>> {
|
|
const directPath = readEventField(primary, ["filePath", "file_path", "path", "uri"])
|
|
|| readEventField(nested, ["filePath", "file_path", "path", "uri"]);
|
|
const hints: Array<Record<string, unknown>> = locations
|
|
.map((location) => objectRecord(location))
|
|
.filter((location): location is Record<string, unknown> => Boolean(location?.path))
|
|
.map((location) => ({ kind: "file", path: location.path, payload: location }));
|
|
if (directPath && (eventType === "file.edited" || eventType === "session.diff")) {
|
|
hints.unshift({ kind: eventType === "file.edited" ? "file_edit" : "diff", path: directPath });
|
|
}
|
|
return hints;
|
|
}
|
|
|
|
function approvalHint(
|
|
eventType: string,
|
|
primary: Record<string, unknown>,
|
|
nested: Record<string, unknown>,
|
|
toolName?: string,
|
|
): Record<string, unknown> | null {
|
|
if (!eventType.includes("permission")) return null;
|
|
const id = readEventField(primary, ["permissionId", "permission_id", "id"])
|
|
|| readEventField(nested, ["permissionId", "permission_id", "id"])
|
|
|| toolName
|
|
|| eventType;
|
|
return {
|
|
id,
|
|
title: `权限确认：${toolName || readEventField(primary, ["title", "message"]) || "工具调用"}`,
|
|
status: eventType.endsWith("replied") ? "completed" : "pending",
|
|
source: "acp-permission",
|
|
payload: primary,
|
|
};
|
|
}
|
|
|
|
function eventRecords(data: unknown): Array<Record<string, unknown>> {
|
|
const primary = objectRecord(data);
|
|
if (!primary) return [];
|
|
return [primary, objectRecord(primary.data), objectRecord(primary.toolCall)]
|
|
.filter((record): record is Record<string, unknown> => Boolean(record));
|
|
}
|
|
|
|
function objectRecord(value: unknown): Record<string, unknown> | null {
|
|
return value && typeof value === "object" && !Array.isArray(value)
|
|
? value as Record<string, unknown>
|
|
: null;
|
|
}
|
|
|
|
/**
|
|
* Unregister all event forwarders from agentService.
|
|
* Call before re-registering or during cleanup.
|
|
*/
|
|
export function unregisterEventForwarders(): void {
|
|
if (registeredHandlers.length === 0) return;
|
|
for (const { event, handler } of registeredHandlers) {
|
|
agentService.off(event, handler);
|
|
}
|
|
registeredHandlers.length = 0;
|
|
log.info("[EventForwarders] Unregistered all event forwarders");
|
|
}
|
|
|
|
export function registerEventForwarders(ctx: HandlerContext): void {
|
|
// Ensure idempotency: unregister any previous handlers first
|
|
unregisterEventForwarders();
|
|
|
|
// Forward all agent SSE events to the renderer
|
|
const sseEventTypes = [
|
|
"message.updated",
|
|
"message.removed",
|
|
"message.part.updated",
|
|
"message.part.removed",
|
|
"permission.updated",
|
|
"permission.replied",
|
|
"session.created",
|
|
"session.updated",
|
|
"session.deleted",
|
|
"session.status",
|
|
"session.idle",
|
|
"session.error",
|
|
"session.diff",
|
|
"file.edited",
|
|
"server.connected",
|
|
];
|
|
|
|
log.info(
|
|
"[EventForwarders] Registering event forwarders for:",
|
|
sseEventTypes.join(", "),
|
|
);
|
|
|
|
for (const eventType of sseEventTypes) {
|
|
const handler = (data: unknown) => {
|
|
// Debug: log message events
|
|
if (eventType.startsWith("message")) {
|
|
log.debug(
|
|
`[EventForwarders] 📨 Received ${eventType}:`,
|
|
JSON.stringify(data).substring(0, 200),
|
|
);
|
|
}
|
|
ctx.getMainWindow()?.webContents.send("agent:event", {
|
|
type: eventType,
|
|
data,
|
|
});
|
|
recordForwardedEvent(eventType, data);
|
|
};
|
|
agentService.on(eventType, handler);
|
|
registeredHandlers.push({ event: eventType, handler });
|
|
}
|
|
|
|
const errorHandler = (error: Error) => {
|
|
ctx.getMainWindow()?.webContents.send("agent:event", {
|
|
type: "error",
|
|
data: { message: error.message },
|
|
});
|
|
recordDigitalEmployeeRuntimeEvent({
|
|
kind: "agent_error",
|
|
message: error.message,
|
|
status: "failed",
|
|
payload: { message: error.message },
|
|
});
|
|
};
|
|
agentService.on("error", errorHandler);
|
|
registeredHandlers.push({ event: "error", handler: errorHandler });
|
|
|
|
const readyHandler = () => {
|
|
ctx.getMainWindow()?.webContents.send("agent:event", {
|
|
type: "ready",
|
|
data: {},
|
|
});
|
|
recordDigitalEmployeeRuntimeEvent({
|
|
kind: "agent_ready",
|
|
message: "Agent service ready",
|
|
status: "running",
|
|
});
|
|
};
|
|
agentService.on("ready", readyHandler);
|
|
registeredHandlers.push({ event: "ready", handler: readyHandler });
|
|
|
|
const destroyedHandler = () => {
|
|
ctx.getMainWindow()?.webContents.send("agent:event", {
|
|
type: "destroyed",
|
|
data: {},
|
|
});
|
|
recordDigitalEmployeeRuntimeEvent({
|
|
kind: "agent_destroyed",
|
|
message: "Agent service destroyed",
|
|
status: "completed",
|
|
});
|
|
};
|
|
agentService.on("destroyed", destroyedHandler);
|
|
registeredHandlers.push({ event: "destroyed", handler: destroyedHandler });
|
|
|
|
// ==================== computer:* Event Forwarding (rcoder camelCase format) ====================
|
|
|
|
const progressHandler = (data: unknown) => {
|
|
log.debug(
|
|
"[EventForwarders] 📨 Received computer:progress:",
|
|
JSON.stringify(data).substring(0, 200),
|
|
);
|
|
ctx.getMainWindow()?.webContents.send("computer:progress", data);
|
|
recordForwardedEvent("computer:progress", data);
|
|
const d = data as UnifiedSessionMessage;
|
|
if (d?.sessionId) {
|
|
// sessionId is now the ACP protocol UUID — same as acpSessionId
|
|
log.debug(
|
|
`[EventForwarders] 📤 Pushing SSE event: sessionId=${d.sessionId}, subType=${d.subType}`,
|
|
);
|
|
pushSseEvent(d.sessionId, d.subType || "message", d);
|
|
}
|
|
};
|
|
agentService.on("computer:progress", progressHandler);
|
|
registeredHandlers.push({
|
|
event: "computer:progress",
|
|
handler: progressHandler,
|
|
});
|
|
|
|
const promptStartHandler = (data: {
|
|
sessionId: string;
|
|
acpSessionId?: string;
|
|
requestId?: string;
|
|
}) => {
|
|
firstTokenTrace.trace("event.prompt_start.forward", {
|
|
requestId: data.requestId,
|
|
sessionId: data.sessionId,
|
|
});
|
|
// sessionId is now the ACP protocol UUID
|
|
const event: UnifiedSessionMessage = {
|
|
sessionId: data.sessionId,
|
|
acpSessionId: data.acpSessionId,
|
|
messageType: "sessionPromptStart",
|
|
subType: "prompt_start",
|
|
data: { request_id: data.requestId },
|
|
timestamp: new Date().toISOString(),
|
|
};
|
|
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);
|
|
};
|
|
agentService.on("computer:promptStart", promptStartHandler);
|
|
registeredHandlers.push({
|
|
event: "computer:promptStart",
|
|
handler: promptStartHandler,
|
|
});
|
|
|
|
const promptEndHandler = (data: {
|
|
sessionId: string;
|
|
acpSessionId?: string;
|
|
reason?: string;
|
|
description?: string;
|
|
}) => {
|
|
firstTokenTrace.trace("event.prompt_end.forward", {
|
|
sessionId: data.sessionId,
|
|
});
|
|
// sessionId is now the ACP protocol UUID
|
|
const event: UnifiedSessionMessage = {
|
|
sessionId: data.sessionId,
|
|
acpSessionId: data.acpSessionId,
|
|
messageType: "sessionPromptEnd",
|
|
subType: data.reason || "end_turn",
|
|
data: { reason: data.reason, description: data.description },
|
|
timestamp: new Date().toISOString(),
|
|
};
|
|
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);
|
|
};
|
|
agentService.on("computer:promptEnd", promptEndHandler);
|
|
registeredHandlers.push({
|
|
event: "computer:promptEnd",
|
|
handler: promptEndHandler,
|
|
});
|
|
}
|