test(client): cover digital employee sync retries

qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/syncService.test.ts

@@ -0,0 +1,462 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+const mockState = vi.hoisted(() => ({
+  db: null as TestDb | null,
+  settings: new Map<string, unknown>(),
+  fetch: vi.fn(),
+}));
+
+vi.mock("electron-log", () => ({
+  default: {
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    debug: vi.fn(),
+  },
+}));
+
+vi.mock("../system/deviceId", () => ({
+  getDeviceId: vi.fn(() => "device-001"),
+}));
+
+vi.mock("../../db", () => ({
+  getDb: () => mockState.db,
+  readSetting: (key: string) => mockState.settings.get(key) ?? null,
+}));
+
+describe("digital employee sync service", () => {
+  beforeEach(() => {
+    vi.resetModules();
+    vi.useFakeTimers();
+    vi.setSystemTime(new Date("2026-06-07T08:00:00.000Z"));
+    mockState.fetch = vi.fn();
+    vi.stubGlobal("fetch", mockState.fetch);
+    mockState.settings = new Map<string, unknown>([
+      ["step1_config", { serverHost: "https://manage.example.com" }],
+      ["auth.username", "operator"],
+      ["auth.saved_key", "client-key-001"],
+      ["auth.tokens.manage.example.com", "login-token-001"],
+    ]);
+    mockState.db = new TestDb();
+  });
+
+  afterEach(() => {
+    mockState.db = null;
+    vi.unstubAllGlobals();
+    vi.useRealTimers();
+  });
+
+  it("skips flushing when management credentials are incomplete", async () => {
+    mockState.settings.delete("auth.saved_key");
+    insertOutbox("plan:upsert:plan-1", "plan", "plan-1");
+
+    const { flushDigitalEmployeeSyncOutbox } = await import("./syncService");
+    const status = await flushDigitalEmployeeSyncOutbox({ force: true });
+
+    expect(mockState.fetch).not.toHaveBeenCalled();
+    expect(status.missingCredentials).toEqual(["客户端 Key"]);
+    expect(status.pending).toBe(1);
+    expect(status.failed).toBe(0);
+    expect(readOutbox("plan:upsert:plan-1")).toMatchObject({
+      status: "pending",
+      attempts: 0,
+    });
+  });
+
+  it("marks entity-only acknowledgements as synced", async () => {
+    insertPlan("plan-1");
+    insertOutbox("plan:upsert:plan-1", "plan", "plan-1");
+    mockState.fetch.mockResolvedValue(
+      jsonResponse({
+        success: true,
+        acked: [
+          {
+            entity_type: "plan",
+            entity_id: "plan-1",
+            remote_id: "remote-plan-1",
+          },
+        ],
+      }),
+    );
+
+    const { flushDigitalEmployeeSyncOutbox } = await import("./syncService");
+    const status = await flushDigitalEmployeeSyncOutbox({ force: true });
+
+    expect(status.pending).toBe(0);
+    expect(status.failed).toBe(0);
+    expect(readOutbox("plan:upsert:plan-1")).toMatchObject({
+      status: "synced",
+      error: null,
+      attempts: 1,
+    });
+    expect(readPlan("plan-1")).toMatchObject({
+      sync_status: "synced",
+      remote_id: "remote-plan-1",
+      sync_error: null,
+    });
+    expect(readAttempts()).toEqual([
+      expect.objectContaining({
+        outbox_id: "plan:upsert:plan-1",
+        attempt_no: 1,
+        status: "synced",
+        error: null,
+      }),
+    ]);
+  });
+
+  it("keeps unacknowledged rows retryable after a partial backend ack", async () => {
+    insertPlan("plan-1");
+    insertPlan("plan-2");
+    insertOutbox("plan:upsert:plan-1", "plan", "plan-1");
+    insertOutbox("plan:upsert:plan-2", "plan", "plan-2");
+    mockState.fetch.mockResolvedValue(
+      jsonResponse({
+        success: true,
+        acked: [
+          {
+            outbox_id: "plan:upsert:plan-1",
+            entity_type: "plan",
+            entity_id: "plan-1",
+          },
+        ],
+      }),
+    );
+
+    const { flushDigitalEmployeeSyncOutbox } = await import("./syncService");
+    const status = await flushDigitalEmployeeSyncOutbox({ force: true });
+
+    expect(status.lastSyncError).toBe(
+      "management sync response did not acknowledge item",
+    );
+    expect(status.failed).toBe(1);
+    expect(status.failureSummary).toMatchObject({
+      total: 1,
+      dueForRetry: 0,
+      backoff: 1,
+      byEntityType: [{ entityType: "plan", count: 1 }],
+    });
+    expect(readOutbox("plan:upsert:plan-1")).toMatchObject({
+      status: "synced",
+    });
+    expect(readOutbox("plan:upsert:plan-2")).toMatchObject({
+      status: "failed",
+      attempts: 1,
+      error: "management sync response did not acknowledge item",
+    });
+    expect(status.recentFailures[0]).toMatchObject({
+      id: "plan:upsert:plan-2",
+      dueForRetry: false,
+      managementRecordUrl:
+        "https://manage.example.com/system/content/content-digital-employee-sync?client_key=client-key-001&entity_type=plan&entity_id=plan-2&outbox_id=plan%3Aupsert%3Aplan-2",
+      attemptHistory: [
+        expect.objectContaining({
+          attemptNo: 1,
+          status: "failed",
+          error: "management sync response did not acknowledge item",
+          endpoint: "https://manage.example.com/api/digital-employee/sync/outbox",
+        }),
+      ],
+    });
+  });
+});
+
+function insertPlan(id: string): void {
+  mockState.db?.plans.set(id, {
+    id,
+    remote_id: null,
+    sync_status: "pending",
+    sync_error: null,
+    last_synced_at: null,
+  });
+}
+
+function insertOutbox(id: string, entityType: string, entityId: string): void {
+  mockState.db?.outbox.set(id, {
+    id,
+    entity_type: entityType,
+    entity_id: entityId,
+    operation: "upsert",
+    payload: '{"ok":true}',
+    status: "pending",
+    attempts: 0,
+    last_attempt_at: null,
+    error: null,
+    created_at: "2026-06-07T07:55:00.000Z",
+    updated_at: "2026-06-07T07:55:00.000Z",
+  });
+}
+
+function readOutbox(id: string): Record<string, unknown> | undefined {
+  return mockState.db?.outbox.get(id);
+}
+
+function readPlan(id: string): Record<string, unknown> | undefined {
+  return mockState.db?.plans.get(id);
+}
+
+function readAttempts(): Array<Record<string, unknown>> {
+  return mockState.db?.attempts.map((attempt) => ({ ...attempt })) ?? [];
+}
+
+function jsonResponse(body: unknown): Response {
+  return {
+    ok: true,
+    status: 200,
+    text: vi.fn().mockResolvedValue(JSON.stringify(body)),
+  } as unknown as Response;
+}
+
+interface TestOutboxRow {
+  id: string;
+  entity_type: string;
+  entity_id: string;
+  operation: string;
+  payload: string;
+  status: string;
+  attempts: number;
+  last_attempt_at: string | null;
+  error: string | null;
+  created_at: string;
+  updated_at: string;
+}
+
+interface TestPlanRow {
+  id: string;
+  remote_id: string | null;
+  sync_status: string;
+  sync_error: string | null;
+  last_synced_at: string | null;
+}
+
+interface TestAttemptRow {
+  id: number;
+  outbox_id: string;
+  entity_type: string;
+  entity_id: string;
+  operation: string;
+  attempt_no: number;
+  status: string;
+  error: string | null;
+  endpoint: string | null;
+  started_at: string;
+  finished_at: string;
+}
+
+class TestDb {
+  outbox = new Map<string, TestOutboxRow>();
+  plans = new Map<string, TestPlanRow>();
+  attempts: TestAttemptRow[] = [];
+  private nextAttemptId = 1;
+
+  transaction<T>(callback: () => T): () => T {
+    return callback;
+  }
+
+  prepare(sql: string): {
+    all: (...args: unknown[]) => unknown[];
+    get: (...args: unknown[]) => unknown;
+    run: (...args: unknown[]) => unknown;
+  } {
+    const normalized = sql.replace(/\s+/g, " ").trim();
+    return {
+      all: (...args: unknown[]) => this.all(normalized, args),
+      get: (...args: unknown[]) => this.get(normalized, args),
+      run: (...args: unknown[]) => this.run(normalized, args),
+    };
+  }
+
+  private all(sql: string, args: unknown[]): unknown[] {
+    if (
+      sql.includes("FROM digital_sync_outbox") &&
+      sql.includes("status IN ('pending', 'failed')")
+    ) {
+      const limit = Number(args[0]);
+      return Array.from(this.outbox.values())
+        .filter((row) => row.status === "pending" || row.status === "failed")
+        .sort((left, right) => left.created_at.localeCompare(right.created_at))
+        .slice(0, limit)
+        .map(copyRow);
+    }
+
+    if (
+      sql.includes("FROM digital_sync_outbox") &&
+      sql.includes("WHERE status = 'failed'") &&
+      sql.includes("SELECT entity_type")
+    ) {
+      return Array.from(this.outbox.values())
+        .filter((row) => row.status === "failed")
+        .map((row) => ({
+          entity_type: row.entity_type,
+          attempts: row.attempts,
+          last_attempt_at: row.last_attempt_at,
+        }));
+    }
+
+    if (
+      sql.includes("FROM digital_sync_outbox") &&
+      sql.includes("WHERE status = 'failed'") &&
+      sql.includes("ORDER BY COALESCE")
+    ) {
+      const limit = Number(args[0]);
+      return Array.from(this.outbox.values())
+        .filter((row) => row.status === "failed")
+        .sort((left, right) =>
+          recentFailureSortKey(right).localeCompare(recentFailureSortKey(left)),
+        )
+        .slice(0, limit)
+        .map(copyRow);
+    }
+
+    if (sql.includes("FROM digital_sync_attempts")) {
+      const [outboxId, limit] = args as [string, number];
+      return this.attempts
+        .filter((row) => row.outbox_id === outboxId)
+        .sort((left, right) => {
+          const byTime = right.finished_at.localeCompare(left.finished_at);
+          return byTime || right.id - left.id;
+        })
+        .slice(0, limit)
+        .map((row) => ({
+          attempt_no: row.attempt_no,
+          status: row.status,
+          error: row.error,
+          endpoint: row.endpoint,
+          started_at: row.started_at,
+          finished_at: row.finished_at,
+        }));
+    }
+
+    throw new Error(`Unhandled all query: ${sql}`);
+  }
+
+  private get(sql: string, args: unknown[]): unknown {
+    if (
+      sql.includes("COUNT(*) AS count FROM digital_sync_outbox WHERE status = ?")
+    ) {
+      const [status] = args;
+      return {
+        count: Array.from(this.outbox.values()).filter(
+          (row) => row.status === status,
+        ).length,
+      };
+    }
+
+    throw new Error(`Unhandled get query: ${sql}`);
+  }
+
+  private run(sql: string, args: unknown[]): unknown {
+    if (sql.startsWith("UPDATE digital_sync_outbox SET status = 'syncing'")) {
+      const [lastAttemptAt, updatedAt, id] = args as [string, string, string];
+      const row = this.outbox.get(id);
+      if (row) {
+        row.status = "syncing";
+        row.attempts += 1;
+        row.last_attempt_at = lastAttemptAt;
+        row.updated_at = updatedAt;
+      }
+      return { changes: row ? 1 : 0 };
+    }
+
+    if (sql.startsWith("UPDATE digital_sync_outbox SET status = 'synced'")) {
+      const [updatedAt, id] = args as [string, string];
+      const row = this.outbox.get(id);
+      if (row) {
+        row.status = "synced";
+        row.error = null;
+        row.updated_at = updatedAt;
+      }
+      return { changes: row ? 1 : 0 };
+    }
+
+    if (sql.startsWith("UPDATE digital_sync_outbox SET status = 'failed'")) {
+      const [error, updatedAt, id] = args as [string, string, string];
+      const row = this.outbox.get(id);
+      if (row) {
+        row.status = "failed";
+        row.error = error;
+        row.updated_at = updatedAt;
+      }
+      return { changes: row ? 1 : 0 };
+    }
+
+    if (sql.startsWith("INSERT INTO digital_sync_attempts")) {
+      const [
+        outboxId,
+        entityType,
+        entityId,
+        operation,
+        attemptNo,
+        status,
+        error,
+        endpoint,
+        startedAt,
+        finishedAt,
+      ] = args as [
+        string,
+        string,
+        string,
+        string,
+        number,
+        string,
+        string | null,
+        string | null,
+        string,
+        string,
+      ];
+      this.attempts.push({
+        id: this.nextAttemptId,
+        outbox_id: outboxId,
+        entity_type: entityType,
+        entity_id: entityId,
+        operation,
+        attempt_no: attemptNo,
+        status,
+        error,
+        endpoint,
+        started_at: startedAt,
+        finished_at: finishedAt,
+      });
+      this.nextAttemptId += 1;
+      return { changes: 1 };
+    }
+
+    if (sql.startsWith("DELETE FROM digital_sync_attempts WHERE finished_at < ?")) {
+      const [cutoff] = args as [string];
+      const before = this.attempts.length;
+      this.attempts = this.attempts.filter((row) => row.finished_at >= cutoff);
+      return { changes: before - this.attempts.length };
+    }
+
+    if (sql.startsWith("UPDATE digital_plans SET sync_status = 'synced'")) {
+      const [remoteId, lastSyncedAt, id] = args as [string | null, string, string];
+      const row = this.plans.get(id);
+      if (row) {
+        row.sync_status = "synced";
+        row.remote_id = remoteId ?? row.remote_id;
+        row.last_synced_at = lastSyncedAt;
+        row.sync_error = null;
+      }
+      return { changes: row ? 1 : 0 };
+    }
+
+    if (sql.startsWith("UPDATE digital_plans SET sync_status = 'failed'")) {
+      const [syncError, id] = args as [string, string];
+      const row = this.plans.get(id);
+      if (row) {
+        row.sync_status = "failed";
+        row.sync_error = syncError;
+      }
+      return { changes: row ? 1 : 0 };
+    }
+
+    throw new Error(`Unhandled run query: ${sql}`);
+  }
+}
+
+function copyRow<T extends object>(row: T): T {
+  return { ...row };
+}
+
+function recentFailureSortKey(row: TestOutboxRow): string {
+  return row.last_attempt_at || row.updated_at || row.created_at;
+}

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

@@ -183,7 +183,7 @@ export async function flushDigitalEmployeeSyncOutbox(
       : [];
     recordSyncAttempts(syncedRows, config, "synced", null, startedAt, finishedAt);
     if (acked.length > 0) {
-      markRowsSynced(acked, finishedAt);
+      markRowsSynced(acked, finishedAt, rows);
     } else {
       markRowsSynced(
         rows.map((row) => ({
@@ -380,7 +380,11 @@ function markRowsSyncing(rows: OutboxRow[], now: string): void {
   tx();
 }
 
-function markRowsSynced(acks: SyncAck[], now: string): void {
+function markRowsSynced(
+  acks: SyncAck[],
+  now: string,
+  sourceRows: OutboxRow[] = [],
+): void {
   const db = getDb();
   if (!db) return;
   const tx = db.transaction(() => {
@@ -390,7 +394,10 @@ function markRowsSynced(acks: SyncAck[], now: string): void {
       WHERE id = ?
     `);
     for (const ack of acks) {
-      const outboxId = ack.outbox_id || ack.id;
+      const outboxId =
+        ack.outbox_id ||
+        ack.id ||
+        sourceRows.find((row) => ackMatchesRow(ack, row))?.id;
       if (!outboxId) continue;
       outboxStmt.run(now, outboxId);
       updateEntitySyncStatus(ack, now);