吸收数字员工PlanStep远端策略拉取

2026-06-10 23:15:51 +08:00
parent 6f2ccdf783
commit c400367d22
17 changed files with 776 additions and 205 deletions
--- a/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts
+++ b/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.ts
@@ -2,14 +2,22 @@ import log from "electron-log";
 import { getDeviceId } from "../system/deviceId";
 import { getDomainTokenKey } from "@shared/utils/domain";
 import { ensureDigitalEmployeeSchema, getDb, readSetting } from "../../db";
+import {
+  normalizePlanStepDispatchPolicy,
+  readDigitalEmployeeUiState,
+  saveDigitalEmployeeUiState,
+  type DigitalEmployeePlanStepDispatchPolicy,
+} from "./stateService";

 const DEFAULT_SYNC_PATH = "/api/digital-employee/sync/outbox";
+const DEFAULT_PLAN_STEP_DISPATCH_POLICY_PATH = "/api/digital-employee/policies/plan-step-dispatch";
 const MANAGEMENT_SYNC_RECORD_PATH =
  "/system/content/content-digital-employee-sync";
 const DIGITAL_EMPLOYEE_SYNC_CONTRACT_VERSION = "qimingclaw.digital_employee.sync.v1";
 const DEFAULT_INTERVAL_MS = 30_000;
 const DEFAULT_BATCH_SIZE = 50;
 const REQUEST_TIMEOUT_MS = 15_000;
+const POLICY_PULL_THROTTLE_MS = 60_000;
 const MAX_RETRY_DELAY_MS = 30 * 60_000;
 const FAILED_RETRY_CANDIDATE_MULTIPLIER = 4;
 const ATTEMPT_HISTORY_RETENTION_MS = 30 * 24 * 60 * 60_000;
@@ -18,6 +26,8 @@ let syncTimer: ReturnType<typeof setInterval> | null = null;
 let syncing = false;
 let lastSyncAt: string | null = null;
 let lastSyncError: string | null = null;
+let policyPulling = false;
+let lastPolicyPullAttemptMs = 0;

 function getDigitalEmployeeDb(): ReturnType<typeof getDb> {
  if (!ensureDigitalEmployeeSchema()) return null;
@@ -27,11 +37,21 @@ function getDigitalEmployeeDb(): ReturnType<typeof getDb> {
 interface DigitalEmployeeSyncConfig {
  enabled: boolean;
  endpoint: string | null;
+  policyEndpoint: string | null;
  clientKey: string | null;
  authToken: string | null;
  batchSize: number;
 }

+export interface DigitalEmployeePlanStepDispatchPolicyPullResult {
+  ok: boolean;
+  skipped?: boolean;
+  endpoint: string | null;
+  pulled_at: string | null;
+  policy: DigitalEmployeePlanStepDispatchPolicy;
+  error?: string | null;
+}
+
 interface OutboxRow {
  id: string;
  entity_type: string;
@@ -168,6 +188,63 @@ export function getDigitalEmployeeSyncStatus(): DigitalEmployeeSyncStatus {
  };
 }

+export async function pullDigitalEmployeePlanStepDispatchPolicy(
+  options: { force?: boolean } = {},
+): Promise<DigitalEmployeePlanStepDispatchPolicyPullResult> {
+  const currentState = readDigitalEmployeeUiState();
+  const currentPolicy = currentState.planStepDispatchPolicy ?? normalizePlanStepDispatchPolicy(null);
+  if (policyPulling) {
+    return { ok: true, skipped: true, endpoint: resolveSyncConfig().policyEndpoint, pulled_at: currentPolicy.last_pulled_at ?? null, policy: currentPolicy };
+  }
+  const nowMs = Date.now();
+  if (!options.force && nowMs - lastPolicyPullAttemptMs < POLICY_PULL_THROTTLE_MS) {
+    return { ok: true, skipped: true, endpoint: resolveSyncConfig().policyEndpoint, pulled_at: currentPolicy.last_pulled_at ?? null, policy: currentPolicy };
+  }
+  lastPolicyPullAttemptMs = nowMs;
+
+  const config = resolveSyncConfig();
+  const missingCredentials = missingSyncCredentialLabels(config);
+  if (!config.policyEndpoint || missingCredentials.length > 0) {
+    const error = !config.policyEndpoint ? "management policy endpoint unavailable" : `missing credentials: ${missingCredentials.join(", ")}`;
+    return recordPlanStepDispatchPolicyPullFailure(currentState, currentPolicy, config.policyEndpoint, error);
+  }
+
+  policyPulling = true;
+  try {
+    const response = await fetchJsonWithAuth(config, config.policyEndpoint);
+    const responseError = readSyncResponseError(response as SyncResponse);
+    if (responseError) throw new Error(responseError);
+    const remotePolicy = readRemotePlanStepDispatchPolicy(response);
+    if (!remotePolicy) throw new Error("management policy response missing plan step dispatch policy");
+    const pulledAt = new Date().toISOString();
+    const policy = normalizePlanStepDispatchPolicy({
+      ...remotePolicy,
+      source: "remote",
+      remote_updated_at: stringField(remotePolicy.updated_at) || stringField(remotePolicy.remote_updated_at) || null,
+      last_pulled_at: pulledAt,
+      last_pull_error: null,
+    });
+    saveDigitalEmployeeUiState({
+      action: "pull_plan_step_dispatch_policy",
+      metadata: {
+        source: "management-api",
+        endpoint: config.policyEndpoint,
+        ok: true,
+        plan_step_dispatch_policy: policy,
+      },
+      state: {
+        ...currentState,
+        planStepDispatchPolicy: policy,
+      },
+    });
+    return { ok: true, endpoint: config.policyEndpoint, pulled_at: pulledAt, policy, error: null };
+  } catch (error) {
+    return recordPlanStepDispatchPolicyPullFailure(currentState, currentPolicy, config.policyEndpoint, normalizeError(error));
+  } finally {
+    policyPulling = false;
+  }
+}
+
 export async function flushDigitalEmployeeSyncOutbox(
  options: { force?: boolean } = {},
 ): Promise<DigitalEmployeeSyncStatus> {
@@ -291,6 +368,13 @@ function resolveSyncConfig(): DigitalEmployeeSyncConfig {
      ? step1.serverHost.trim()
      : null;
  const endpoint = endpointOverride || buildEndpoint(serverHost);
+  const policyEndpointOverride =
+    typeof config.planStepDispatchPolicyEndpoint === "string" && config.planStepDispatchPolicyEndpoint.trim()
+      ? config.planStepDispatchPolicyEndpoint.trim()
+      : typeof config.policyEndpoint === "string" && config.policyEndpoint.trim()
+        ? config.policyEndpoint.trim()
+        : null;
+  const policyEndpoint = policyEndpointOverride || buildEndpoint(serverHost, DEFAULT_PLAN_STEP_DISPATCH_POLICY_PATH);
  const disabled = config.enabled === false;
  const batchSize =
    typeof config.batchSize === "number" && config.batchSize > 0
@@ -300,18 +384,19 @@ function resolveSyncConfig(): DigitalEmployeeSyncConfig {
  return {
    enabled: !disabled,
    endpoint,
+    policyEndpoint,
    clientKey: readClientKey(serverHost),
    authToken: readAuthToken(serverHost),
    batchSize,
  };
 }

-function buildEndpoint(serverHost: string | null): string | null {
+function buildEndpoint(serverHost: string | null, path = DEFAULT_SYNC_PATH): string | null {
  if (!serverHost) return null;
  const normalized = serverHost.startsWith("http")
    ? serverHost
    : `https://${serverHost}`;
-  return `${normalized.replace(/\/+$/, "")}${DEFAULT_SYNC_PATH}`;
+  return `${normalized.replace(/\/+$/, "")}${path}`;
 }

 function missingSyncCredentialLabels(
@@ -385,35 +470,85 @@ async function postOutbox(
  config: DigitalEmployeeSyncConfig,
  rows: OutboxRow[],
 ): Promise<SyncResponse> {
+  return fetchJsonWithAuth(config, config.endpoint!, {
+    client_key: config.clientKey,
+    device_id: getDeviceId(),
+    contract_version: DIGITAL_EMPLOYEE_SYNC_CONTRACT_VERSION,
+    items: rows.map(buildSyncItem),
+  }) as Promise<SyncResponse>;
+}
+
+async function fetchJsonWithAuth(
+  config: DigitalEmployeeSyncConfig,
+  endpoint: string,
+  requestBody?: Record<string, unknown>,
+): Promise<Record<string, unknown>> {
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), REQUEST_TIMEOUT_MS);
  try {
-    const response = await fetch(config.endpoint!, {
-      method: "POST",
+    const response = await fetch(endpoint, {
+      method: requestBody ? "POST" : "GET",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${config.authToken}`,
        "X-Qiming-Client-Key": config.clientKey!,
+        "X-Qiming-Device-Id": getDeviceId(),
      },
-      body: JSON.stringify({
-        client_key: config.clientKey,
-        device_id: getDeviceId(),
-        contract_version: DIGITAL_EMPLOYEE_SYNC_CONTRACT_VERSION,
-        items: rows.map(buildSyncItem),
-      }),
+      body: requestBody ? JSON.stringify(requestBody) : undefined,
      signal: controller.signal,
    });
    const text = await response.text();
-    const body = text ? (JSON.parse(text) as SyncResponse) : {};
+    const responseBody = text ? (JSON.parse(text) as Record<string, unknown>) : {};
    if (!response.ok) {
-      throw new Error(body.message || `HTTP ${response.status}`);
+      throw new Error(stringField(responseBody.message) || `HTTP ${response.status}`);
    }
-    return body;
+    return responseBody;
  } finally {
    clearTimeout(timer);
  }
 }

+function readRemotePlanStepDispatchPolicy(response: Record<string, unknown>): Record<string, unknown> | null {
+  const data = objectRecord(response.data) ?? response;
+  return objectRecord(data.plan_step_dispatch_policy)
+    ?? objectRecord(data.planStepDispatchPolicy)
+    ?? objectRecord(data.policy)
+    ?? (hasPlanStepDispatchPolicyFields(data) ? data : null);
+}
+
+function hasPlanStepDispatchPolicyFields(record: Record<string, unknown>): boolean {
+  return "enabled" in record || "max_per_sweep" in record || "auto_dispatch_ready_steps" in record;
+}
+
+function recordPlanStepDispatchPolicyPullFailure(
+  state: ReturnType<typeof readDigitalEmployeeUiState>,
+  currentPolicy: DigitalEmployeePlanStepDispatchPolicy,
+  endpoint: string | null,
+  error: string,
+): DigitalEmployeePlanStepDispatchPolicyPullResult {
+  const pulledAt = new Date().toISOString();
+  const policy = normalizePlanStepDispatchPolicy({
+    ...currentPolicy,
+    last_pulled_at: pulledAt,
+    last_pull_error: error,
+  });
+  saveDigitalEmployeeUiState({
+    action: "pull_plan_step_dispatch_policy",
+    metadata: {
+      source: "management-api",
+      endpoint,
+      ok: false,
+      error,
+      plan_step_dispatch_policy: policy,
+    },
+    state: {
+      ...state,
+      planStepDispatchPolicy: policy,
+    },
+  });
+  return { ok: false, endpoint, pulled_at: pulledAt, policy, error };
+}
+
 function buildSyncItem(row: OutboxRow): Record<string, unknown> {
  const payload = parsePayload(row.payload);
  const summary = readEntitySummary(row.entity_type, row.entity_id);
@@ -753,10 +888,16 @@ function planStepDispatchPolicyBusinessView(payload: Record<string, unknown>): R
    ?? {};
  return {
    action: stringField(metadata.action ?? payload.action) || "update_plan_step_dispatch_policy",
+    source: stringField(metadata.source ?? policy.source) || null,
+    pull_ok: typeof metadata.ok === "boolean" ? metadata.ok : null,
+    pull_error: stringField(metadata.error ?? policy.last_pull_error) || null,
+    pull_endpoint: stringField(metadata.endpoint) || null,
    policy_enabled: typeof policy.enabled === "boolean" ? policy.enabled : null,
    max_per_sweep: numberField(policy.max_per_sweep),
    auto_dispatch_ready_steps: typeof policy.auto_dispatch_ready_steps === "boolean" ? policy.auto_dispatch_ready_steps : null,
    updated_at: stringField(policy.updated_at) || null,
+    remote_updated_at: stringField(policy.remote_updated_at) || null,
+    last_pulled_at: stringField(policy.last_pulled_at) || null,
  };
 }

@@ -820,7 +961,8 @@ function isPlanStepDependencyEvent(kind: string): boolean {
 }

 function isPlanStepDispatchPolicyEvent(kind: string): boolean {
-  return kind === "digital_workday_update_plan_step_dispatch_policy";
+  return kind === "digital_workday_update_plan_step_dispatch_policy"
+    || kind === "digital_workday_pull_plan_step_dispatch_policy";
 }

 function isPlanStepLeaseEvent(kind: string): boolean {