qiming/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts

import { getDb, readSetting, writeSetting } from "../../db";

const STATE_KEY = "digital_employee_state_v1";
const MAX_EVENTS = 200;

export interface DigitalEmployeeServiceStatus {
  running?: boolean;
  error?: string;
  engineType?: string | null;
  port?: number;
  serverCount?: number;
  serverNames?: string[];
  pid?: number;
}

export interface DigitalEmployeeSessionSummary {
  total: number;
  active: number;
  items: Array<{
    id: string;
    title?: string;
    status?: string;
    engineType?: string;
    projectId?: string;
  }>;
}

export interface DigitalEmployeeSnapshot {
  generatedAt: string;
  services: Record<string, DigitalEmployeeServiceStatus | null>;
  sessions: DigitalEmployeeSessionSummary;
}

export interface DigitalEmployeeStateEvent {
  event_id: string;
  kind: string;
  occurred_at: string;
  message: string;
  plan_id: string | null;
  task_id: string | null;
}

export interface DigitalEmployeeState {
  version: 1;
  updatedAt: string | null;
  lastSnapshot: DigitalEmployeeSnapshot | null;
  events: DigitalEmployeeStateEvent[];
  pendingSyncCount?: number;
}

export interface DigitalEmployeeChatRequestRecord {
  user_id?: string;
  project_id?: string;
  prompt?: string;
  session_id?: string;
  request_id?: string;
  model_provider?: { provider?: string; model?: string; default_model?: string };
}

export interface DigitalEmployeeChatResultRecord {
  success: boolean;
  code?: string;
  message?: string;
  project_id?: string;
  session_id?: string;
  request_id?: 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 {
  return {
    version: 1,
    updatedAt: null,
    lastSnapshot: null,
    events: [],
  };
}

export function readDigitalEmployeeState(): DigitalEmployeeState {
  const state = readSetting(STATE_KEY);
  const pendingSyncCount = countPendingSyncItems();
  if (!state || typeof state !== "object") {
    return { ...defaultState(), pendingSyncCount };
  }
  return {
    ...defaultState(),
    ...(state as Partial<DigitalEmployeeState>),
    version: 1,
    pendingSyncCount,
  };
}

export function writeDigitalEmployeeState(state: DigitalEmployeeState): void {
  writeSetting(STATE_KEY, state);
}

export function recordDigitalEmployeeSnapshot(
  snapshot: DigitalEmployeeSnapshot,
): DigitalEmployeeState {
  const previous = readDigitalEmployeeState();
  const events = [...previous.events];
  const event = eventFromSnapshot(snapshot, previous.lastSnapshot);
  if (event) {
    events.unshift(event);
  }

  const next: DigitalEmployeeState = {
    version: 1,
    updatedAt: snapshot.generatedAt,
    lastSnapshot: snapshot,
    events: events.slice(0, MAX_EVENTS),
  };
  writeDigitalEmployeeState(next);
  persistSnapshotTables(snapshot, event);
  return next;
}

export function recordDigitalEmployeeChatReceived(
  request: DigitalEmployeeChatRequestRecord,
): { planId: string; taskId: string; runId: string } {
  const now = new Date().toISOString();
  const ids = chatIds(request);
  const title = chatTitle(request);

  const db = getDb();
  if (!db) return ids;

  const tx = db.transaction(() => {
    upsertPlan({
      id: ids.planId,
      title: `远程任务：${title}`,
      objective: request.prompt || request.project_id || "管理端下发任务",
      status: "running",
      payload: request,
      now,
    });
    upsertTask({
      id: ids.taskId,
      planId: ids.planId,
      title,
      status: "running",
      assignedSkillId: "qimingclaw-computer-control",
      payload: request,
      now,
    });
    upsertRun({
      id: ids.runId,
      planId: ids.planId,
      taskId: ids.taskId,
      status: "running",
      payload: { request },
      now,
    });
    insertEvent(
      {
        event_id: `${ids.runId}:received`,
        kind: "computer_chat_received",
        occurred_at: now,
        message: `收到管理端任务：${title}`,
        plan_id: ids.planId,
        task_id: ids.taskId,
      },
      ids.runId,
      { request },
    );
  });
  tx();

  return ids;
}

export function recordDigitalEmployeeChatResult(
  request: DigitalEmployeeChatRequestRecord,
  result: DigitalEmployeeChatResultRecord,
): void {
  const now = new Date().toISOString();
  const ids = chatIds({
    ...request,
    request_id: request.request_id || result.request_id,
    session_id: request.session_id || result.session_id,
    project_id: request.project_id || result.project_id,
  });
  const status = result.success ? "completed" : "failed";
  const title = chatTitle(request);
  const db = getDb();
  if (!db) return;

  const tx = db.transaction(() => {
    upsertPlan({
      id: ids.planId,
      title: `远程任务：${title}`,
      objective: request.prompt || request.project_id || "管理端下发任务",
      status,
      payload: { request, result },
      now,
    });
    upsertTask({
      id: ids.taskId,
      planId: ids.planId,
      title,
      status,
      assignedSkillId: "qimingclaw-computer-control",
      payload: { request, result },
      now,
    });
    upsertRun({
      id: ids.runId,
      planId: ids.planId,
      taskId: ids.taskId,
      status,
      payload: { request, result },
      now,
    });
    finishRun(ids.runId, status, now);
    insertEvent(
      {
        event_id: `${ids.runId}:${status}`,
        kind: result.success ? "computer_chat_completed" : "computer_chat_failed",
        occurred_at: now,
        message: result.success
          ? `管理端任务完成：session=${result.session_id || request.session_id || "unknown"}`
          : `管理端任务失败：${result.message || result.code || "unknown error"}`,
        plan_id: ids.planId,
        task_id: ids.taskId,
      },
      ids.runId,
      { request, result },
    );
  });
  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(
  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 finishRun(runId: string, status: string, now: string): void {
  const db = getDb();
  if (!db) return;
  db.prepare(`
    UPDATE digital_runs
    SET status = ?, finished_at = ?, updated_at = ?, sync_status = 'pending'
    WHERE id = ?
  `).run(status, now, now, runId);
}

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 chatIds(request: DigitalEmployeeChatRequestRecord): {
  planId: string;
  taskId: string;
  runId: string;
} {
  const base = safeId(request.request_id || request.session_id || request.project_id || `${Date.now()}`);
  return {
    planId: `computer-chat-plan-${base}`,
    taskId: `computer-chat-task-${base}`,
    runId: `computer-chat-run-${base}`,
  };
}

function chatTitle(request: DigitalEmployeeChatRequestRecord): string {
  const prompt = (request.prompt || "").replace(/\s+/g, " ").trim();
  if (prompt) return prompt.length > 48 ? `${prompt.slice(0, 48)}...` : prompt;
  return request.project_id ? `项目 ${request.project_id}` : "管理端下发任务";
}

function safeId(value: string): string {
  return value.replace(/[^a-zA-Z0-9_-]+/g, "-").replace(/^-+|-+$/g, "") || "unknown";
}

function eventFromSnapshot(
  snapshot: DigitalEmployeeSnapshot,
  previous: DigitalEmployeeSnapshot | null,
): DigitalEmployeeStateEvent | null {
  const currentSummary = summarizeSnapshot(snapshot);
  const previousSummary = previous ? summarizeSnapshot(previous) : "";
  if (currentSummary === previousSummary) {
    return null;
  }
  return {
    event_id: `qimingclaw-snapshot-${Date.now()}`,
    kind: "service_snapshot",
    occurred_at: snapshot.generatedAt,
    message: currentSummary,
    plan_id: "qimingclaw-client-readiness",
    task_id: "task-agent-status",
  };
}

function summarizeSnapshot(snapshot: DigitalEmployeeSnapshot): string {
  const serviceNames = [
    ["agent", "Agent"],
    ["mcp", "MCP"],
    ["computerServer", "Computer Server"],
    ["lanproxy", "Lanproxy"],
    ["fileServer", "File Server"],
  ] as const;
  const services = serviceNames.map(([key, label]) => {
    const status = snapshot.services[key];
    return `${label}:${status?.running ? "running" : "stopped"}`;
  });
  services.push(`sessions:${snapshot.sessions.active}/${snapshot.sessions.total}`);
  return services.join(";");
}