byy/qiming
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

吸收数字员工PlanStep本地软租约

Browse Source
This commit is contained in:
baiyanyun
2026-06-10 22:29:52 +08:00
parent 909f11dae5
commit 6f2ccdf783
13 changed files with 604 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/schedulerService.test.ts
View File

@@ -7,6 +7,8 @@ const mockState = vi.hoisted(() => ({
finishedRuns: [] as Array<Record<string, unknown>>,
dispatchEvents: [] as Array<Record<string, unknown>>,
governanceCommands: [] as Array<Record<string, unknown>>,
acquiredLeases: [] as Array<Record<string, unknown>>,
releasedLeases: [] as Array<Record<string, unknown>>,
settings: {
enabled: true,
catch_up_on_startup: true,
@@ -54,6 +56,25 @@ vi.mock("./stateService", () => ({
mockState.dispatchEvents.push(input);
return { eventId: `dispatch-${input.status}`, kind: `plan_step_dispatch_${input.status}`, payload: input };
},
acquireDigitalEmployeePlanStepLease: (input: Record<string, unknown>) => {
mockState.acquiredLeases.push(input);
const candidate = mockState.readyCandidates.find((item) => item.stepId === input.stepId);
if (candidate?.leaseActive || candidate?.skipReason === "lease_active") {
return {
ok: false,
reason: "lease_active",
lease: { lease_id: `lease-${input.stepId}`, state: "active", owner_device_id: "device-remote", acquired_at: "2026-06-07T00:59:00.000Z", expires_at: "2026-06-07T01:04:00.000Z" },
};
}
return {
ok: true,
lease: { lease_id: `lease-${input.stepId}`, state: "active", owner_device_id: "device-001", acquired_at: input.occurredAt, expires_at: "2026-06-07T01:05:10.000Z" },
};
},
releaseDigitalEmployeePlanStepLease: (input: Record<string, unknown>) => {
mockState.releasedLeases.push(input);
return { ok: true, lease: { lease_id: input.leaseId, state: "released", owner_device_id: "device-001", acquired_at: "2026-06-07T01:00:10.000Z", expires_at: "2026-06-07T01:05:10.000Z", released_at: input.occurredAt, release_reason: input.reason } };
},
recordDigitalEmployeeGovernanceCommand: (input: Record<string, unknown>) => {
mockState.governanceCommands.push(input);
return {
@@ -72,6 +93,8 @@ describe("digital employee scheduler service", () => {
mockState.finishedRuns = [];
mockState.dispatchEvents = [];
mockState.governanceCommands = [];
mockState.acquiredLeases = [];
mockState.releasedLeases = [];
mockState.dispatchPolicy = {
enabled: true,
max_per_sweep: 3,
@@ -156,6 +179,8 @@ describe("digital employee scheduler service", () => {
expect.objectContaining({ method: "POST" }),
);
expect(mockState.dispatchEvents.map((event) => event.status)).toEqual(["started", "completed"]);
expect(mockState.acquiredLeases).toEqual([expect.objectContaining({ stepId: "step-ready", triggerSource: "auto" })]);
expect(mockState.releasedLeases).toEqual([expect.objectContaining({ stepId: "step-ready", leaseId: "lease-step-ready", reason: "completed" })]);
expect(mockState.governanceCommands[0]).toMatchObject({
commandKind: "start_run",
planId: "plan-1",
@@ -179,9 +204,24 @@ describe("digital employee scheduler service", () => {
expect(result.readyStepDispatch).toMatchObject({ candidates: 1, dispatched: 0, failed: 1 });
expect(result.readyStepDispatch.errors[0]).toMatchObject({ stepId: "step-ready", error: "ComputerServer 502: bad gateway" });
expect(mockState.dispatchEvents.map((event) => event.status)).toEqual(["started", "failed"]);
expect(mockState.releasedLeases).toEqual([expect.objectContaining({ stepId: "step-ready", leaseId: "lease-step-ready", reason: "failed" })]);
expect(mockState.governanceCommands).toHaveLength(0);
});
it("skips ready plan steps with active leases", async () => {
mockState.schedules = [];
mockState.readyCandidates = [readyStepCandidate({ dispatchable: false, skipReason: "lease_active", leaseActive: true })];
const { runDigitalEmployeeSchedulerCheck } = await import("./schedulerService");
const result = await runDigitalEmployeeSchedulerCheck(new Date("2026-06-07T01:00:10.000Z"));
expect(result.readyStepDispatch).toMatchObject({ candidates: 1, dispatched: 0, skipped: 1, failed: 0 });
expect(fetch).not.toHaveBeenCalled();
expect(mockState.acquiredLeases).toEqual([expect.objectContaining({ stepId: "step-ready", triggerSource: "auto" })]);
expect(mockState.releasedLeases).toHaveLength(0);
expect(mockState.dispatchEvents).toHaveLength(0);
});
it("honors disabled ready step dispatch policy", async () => {
mockState.readyCandidates = [readyStepCandidate()];
mockState.dispatchPolicy = { ...mockState.dispatchPolicy, enabled: false };

48
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/schedulerService.ts
View File

@@ -2,6 +2,7 @@ import log from "electron-log";
import { LOCALHOST_HOSTNAME } from "../constants";
import { getConfiguredPorts } from "../startupPorts";
import {
acquireDigitalEmployeePlanStepLease,
finishDigitalEmployeeScheduleRun,
listDigitalEmployeeReadyPlanStepDispatchCandidates,
listDigitalEmployeeSchedules,
@@ -12,7 +13,9 @@ import {
recordDigitalEmployeeGovernanceCommand,
recordDigitalEmployeePlanStepDispatch,
recordDigitalEmployeeScheduleRunStarted,
releaseDigitalEmployeePlanStepLease,
upsertDigitalEmployeeSchedule,
type DigitalEmployeePlanStepLease,
type DigitalEmployeeReadyPlanStepDispatchCandidate,
type DigitalEmployeeScheduleRecord,
} from "./stateService";
@@ -164,6 +167,17 @@ export async function dispatchReadyPlanSteps(
let attempts = 0;
for (const candidate of candidates) {
if (!candidate.dispatchable || !candidate.taskId) {
if (candidate.skipReason === "lease_active") {
acquireDigitalEmployeePlanStepLease({
stepId: candidate.stepId,
planId: candidate.planId,
taskId: candidate.taskId,
triggerSource,
orchestrationId,
candidateCount: candidates.length,
occurredAt: now.toISOString(),
});
}
result.skipped += 1;
result.skipped_step_ids?.push(candidate.stepId);
continue;
@@ -173,9 +187,23 @@ export async function dispatchReadyPlanSteps(
result.skipped_step_ids?.push(candidate.stepId);
continue;
}
attempts += 1;
const leaseResult = acquireDigitalEmployeePlanStepLease({
stepId: candidate.stepId,
planId: candidate.planId,
taskId: candidate.taskId,
triggerSource,
orchestrationId,
candidateCount: candidates.length,
occurredAt: now.toISOString(),
});
if (!leaseResult.ok || !leaseResult.lease) {
result.skipped += 1;
result.skipped_step_ids?.push(candidate.stepId);
continue;
}
try {
await dispatchReadyPlanStep(candidate, now, { triggerSource, orchestrationId, candidateCount: candidates.length });
attempts += 1;
await dispatchReadyPlanStep(candidate, now, { triggerSource, orchestrationId, candidateCount: candidates.length }, leaseResult.lease);
result.dispatched += 1;
result.dispatched_step_ids?.push(candidate.stepId);
} catch (error) {
@@ -183,6 +211,13 @@ export async function dispatchReadyPlanSteps(
result.failed += 1;
result.failed_step_ids?.push(candidate.stepId);
result.errors.push({ stepId: candidate.stepId, error: message });
releaseDigitalEmployeePlanStepLease({
stepId: candidate.stepId,
leaseId: leaseResult.lease.lease_id,
reason: "failed",
taskId: candidate.taskId,
occurredAt: now.toISOString(),
});
recordDigitalEmployeePlanStepDispatch({
stepId: candidate.stepId,
planId: candidate.planId,
@@ -215,6 +250,7 @@ async function dispatchReadyPlanStep(
candidate: DigitalEmployeeReadyPlanStepDispatchCandidate,
now: Date,
orchestration: { triggerSource: DigitalEmployeeReadyStepDispatchTriggerSource; orchestrationId: string; candidateCount: number },
lease: DigitalEmployeePlanStepLease,
): Promise<void> {
const occurredAt = now.toISOString();
const commandId = `ready-step-${safeId(candidate.stepId)}-${safeId(occurredAt)}`;
@@ -276,6 +312,14 @@ async function dispatchReadyPlanStep(
},
occurredAt,
});
releaseDigitalEmployeePlanStepLease({
stepId: candidate.stepId,
leaseId: lease.lease_id,
reason: "completed",
taskId: candidate.taskId,
runId: ids.runId,
occurredAt,
});
}
export async function runDigitalEmployeeScheduleNow(

83
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.test.ts
View File

@@ -16,6 +16,10 @@ vi.mock("../../db", () => ({
writeSetting: (key: string, value: unknown) => mockState.settings.set(key, value),
}));
vi.mock("../system/deviceId", () => ({
getDeviceId: vi.fn(() => "device-001"),
}));
describe("digital employee state service", () => {
beforeEach(() => {
vi.resetModules();
@@ -1405,6 +1409,71 @@ describe("digital employee state service", () => {
]);
});
it("acquires, respects, expires, and releases ready plan step leases", async () => {
const {
acquireDigitalEmployeePlanStepLease,
activeDigitalEmployeePlanStepLease,
listDigitalEmployeeReadyPlanStepDispatchCandidates,
releaseDigitalEmployeePlanStepLease,
} = await import("./stateService");
createGovernableTask({ taskStatus: "pending", stepStatus: "ready", runStatus: "completed", finishedAt: "2026-06-07T07:20:00.000Z" });
const acquired = acquireDigitalEmployeePlanStepLease({
stepId: "step-governance",
planId: "plan-governance",
taskId: "task-governance",
triggerSource: "manual",
orchestrationId: "manual-ready-steps-plan-governance",
candidateCount: 1,
occurredAt: "2026-06-07T08:10:00.000Z",
});
expect(acquired).toMatchObject({ ok: true, lease: expect.objectContaining({ state: "active", owner_device_id: "device-001" }) });
const leasedPayload = JSON.parse(mockState.db?.planSteps.get("step-governance")?.payload ?? "{}");
expect(activeDigitalEmployeePlanStepLease(leasedPayload, "2026-06-07T08:11:00.000Z")).toMatchObject({ lease_id: acquired.lease?.lease_id });
expect(listDigitalEmployeeReadyPlanStepDispatchCandidates()[0]).toMatchObject({ dispatchable: false, skipReason: "lease_active" });
expect(mockState.db?.outbox.get("plan_step:upsert:step-governance")).toMatchObject({ entity_type: "plan_step", status: "pending" });
expect(Array.from(mockState.db?.events.values() ?? [])).toEqual(expect.arrayContaining([
expect.objectContaining({ kind: "plan_step_lease_acquired", plan_id: "plan-governance", task_id: "task-governance" }),
]));
const blocked = acquireDigitalEmployeePlanStepLease({
stepId: "step-governance",
planId: "plan-governance",
taskId: "task-governance",
occurredAt: "2026-06-07T08:11:00.000Z",
});
expect(blocked).toMatchObject({ ok: false, reason: "lease_active", lease: expect.objectContaining({ lease_id: acquired.lease?.lease_id }) });
expect(Array.from(mockState.db?.events.values() ?? [])).toEqual(expect.arrayContaining([
expect.objectContaining({ kind: "plan_step_lease_skipped" }),
]));
const overwritten = acquireDigitalEmployeePlanStepLease({
stepId: "step-governance",
planId: "plan-governance",
taskId: "task-governance",
occurredAt: "2026-06-07T08:16:01.000Z",
});
expect(overwritten).toMatchObject({ ok: true, lease: expect.objectContaining({ state: "active" }) });
expect(overwritten.lease?.lease_id).not.toBe(acquired.lease?.lease_id);
const released = releaseDigitalEmployeePlanStepLease({
stepId: "step-governance",
leaseId: overwritten.lease?.lease_id,
reason: "completed",
taskId: "task-governance",
runId: "run-dispatch",
occurredAt: "2026-06-07T08:17:00.000Z",
});
expect(released).toMatchObject({ ok: true, lease: expect.objectContaining({ state: "released", release_reason: "completed" }) });
expect(activeDigitalEmployeePlanStepLease(JSON.parse(mockState.db?.planSteps.get("step-governance")?.payload ?? "{}"), "2026-06-07T08:17:01.000Z")).toBeNull();
expect(Array.from(mockState.db?.events.values() ?? [])).toEqual(expect.arrayContaining([
expect.objectContaining({ kind: "plan_step_lease_released", run_id: "run-dispatch" }),
]));
});
it("records ready plan step dispatch events and updates step status", async () => {
const { recordDigitalEmployeePlanStepDispatch } = await import("./stateService");
@@ -2431,6 +2500,20 @@ class TestDb {
return { changes: previous ? 1 : 0 };
}
if (sql.startsWith("UPDATE digital_plan_steps SET payload")) {
const [payload, updatedAt, id] = args as [string, string, string];
const previous = this.planSteps.get(id);
if (previous) {
this.planSteps.set(id, {
...previous,
payload,
updated_at: updatedAt,
sync_status: "pending",
});
}
return { changes: previous ? 1 : 0 };
}
if (sql.startsWith("UPDATE digital_plan_steps")) {
const [status, payload, updatedAt, id] = args as [string, string, string, string];
const previous = this.planSteps.get(id);

218
qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts
View File

@@ -1,8 +1,10 @@
import { ensureDigitalEmployeeSchema, getDb, readSetting, writeSetting } from "../../db";
import { getDeviceId } from "../system/deviceId";
const STATE_KEY = "digital_employee_state_v1";
const UI_STATE_KEY = "digital_employee_ui_state_v1";
const MAX_EVENTS = 200;
const PLAN_STEP_LEASE_TTL_MS = 5 * 60 * 1000;
export interface DigitalEmployeeServiceStatus {
running?: boolean;
@@ -304,6 +306,37 @@ export interface DigitalEmployeeReadyPlanStepDispatchCandidate {
plan?: DigitalEmployeePlanRecord | null;
}
export interface DigitalEmployeePlanStepLease {
lease_id: string;
state: "active" | "released" | "expired";
owner_device_id: string;
acquired_at: string;
expires_at: string;
released_at?: string | null;
release_reason?: string | null;
trigger_source?: string | null;
orchestration_id?: string | null;
candidate_count?: number | null;
}
export interface DigitalEmployeePlanStepLeaseResult {
ok: boolean;
lease?: DigitalEmployeePlanStepLease;
reason?: string | null;
}
export interface DigitalEmployeePlanStepLeaseInput {
stepId: string;
planId?: string | null;
taskId?: string | null;
runId?: string | null;
triggerSource?: string | null;
orchestrationId?: string | null;
candidateCount?: number | null;
occurredAt?: string | null;
ttlMs?: number | null;
}
export interface DigitalEmployeePlanStepDispatchInput {
stepId: string;
planId?: string | null;
@@ -1384,6 +1417,124 @@ export function recordDigitalEmployeeApprovalAction(
return { eventId, kind, payload };
}
export function activeDigitalEmployeePlanStepLease(
payload: unknown,
now = new Date().toISOString(),
): DigitalEmployeePlanStepLease | null {
const record = objectRecord(payload) ?? {};
const lease = objectRecord(record.dispatchLease ?? record.dispatch_lease);
if (!lease) return null;
const state = stringValue(lease.state) || "active";
const expiresAt = stringValue(lease.expires_at ?? lease.expiresAt);
if (state !== "active" || !expiresAt || expiresAt <= now) return null;
const leaseId = stringValue(lease.lease_id ?? lease.leaseId);
const ownerDeviceId = stringValue(lease.owner_device_id ?? lease.ownerDeviceId);
if (!leaseId || !ownerDeviceId) return null;
return {
lease_id: leaseId,
state: "active",
owner_device_id: ownerDeviceId,
acquired_at: stringValue(lease.acquired_at ?? lease.acquiredAt) || now,
expires_at: expiresAt,
released_at: stringValue(lease.released_at ?? lease.releasedAt) || null,
release_reason: stringValue(lease.release_reason ?? lease.releaseReason) || null,
trigger_source: stringValue(lease.trigger_source ?? lease.triggerSource) || null,
orchestration_id: stringValue(lease.orchestration_id ?? lease.orchestrationId) || null,
candidate_count: numberValue(lease.candidate_count ?? lease.candidateCount),
};
}
export function acquireDigitalEmployeePlanStepLease(
input: DigitalEmployeePlanStepLeaseInput,
): DigitalEmployeePlanStepLeaseResult {
const db = getDigitalEmployeeDb();
if (!db) return { ok: false, reason: "database_unavailable" };
const now = input.occurredAt || new Date().toISOString();
const stepId = stringValue(input.stepId);
if (!stepId) return { ok: false, reason: "missing_step" };
const row = db.prepare(`
SELECT id, plan_id, title, status, seq, depends_on, preferred_skill_ids, payload, created_at
FROM digital_plan_steps
WHERE id = ?
`).get(stepId) as Record<string, unknown> | undefined;
if (!row) return { ok: false, reason: "missing_step" };
const previousPayload = objectRecord(parseStoredPayload(row.payload)) ?? {};
const activeLease = activeDigitalEmployeePlanStepLease(previousPayload, now);
if (activeLease) {
recordDigitalEmployeePlanStepLeaseEvent({ row, lease: activeLease, kind: "plan_step_lease_skipped", reason: "lease_active", now, taskId: input.taskId, runId: input.runId });
return { ok: false, lease: activeLease, reason: "lease_active" };
}
const ttlMs = Math.max(60_000, Math.min(Number(input.ttlMs) || PLAN_STEP_LEASE_TTL_MS, 30 * 60 * 1000));
const lease: DigitalEmployeePlanStepLease = {
lease_id: `plan-step-lease:${safeId(stepId)}:${safeId(now)}`,
state: "active",
owner_device_id: getDeviceId(),
acquired_at: now,
expires_at: new Date(new Date(now).getTime() + ttlMs).toISOString(),
trigger_source: stringValue(input.triggerSource) || null,
orchestration_id: stringValue(input.orchestrationId) || null,
candidate_count: numberValue(input.candidateCount),
};
const payload = { ...previousPayload, dispatchLease: lease };
db.prepare(`
UPDATE digital_plan_steps
SET payload = ?, sync_status = 'pending', updated_at = ?
WHERE id = ?
`).run(stringify(payload), now, stepId);
markPlanStepLeaseOutbox(row, payload, now);
recordDigitalEmployeePlanStepLeaseEvent({ row, lease, kind: "plan_step_lease_acquired", now, taskId: input.taskId, runId: input.runId });
return { ok: true, lease };
}
export function releaseDigitalEmployeePlanStepLease(input: {
stepId: string;
leaseId?: string | null;
reason?: string | null;
taskId?: string | null;
runId?: string | null;
occurredAt?: string | null;
}): DigitalEmployeePlanStepLeaseResult {
const db = getDigitalEmployeeDb();
if (!db) return { ok: false, reason: "database_unavailable" };
const now = input.occurredAt || new Date().toISOString();
const stepId = stringValue(input.stepId);
if (!stepId) return { ok: false, reason: "missing_step" };
const row = db.prepare(`
SELECT id, plan_id, title, status, seq, depends_on, preferred_skill_ids, payload, created_at
FROM digital_plan_steps
WHERE id = ?
`).get(stepId) as Record<string, unknown> | undefined;
if (!row) return { ok: false, reason: "missing_step" };
const previousPayload = objectRecord(parseStoredPayload(row.payload)) ?? {};
const currentLease = objectRecord(previousPayload.dispatchLease ?? previousPayload.dispatch_lease);
const currentLeaseId = stringValue(currentLease?.lease_id ?? currentLease?.leaseId);
const expectedLeaseId = stringValue(input.leaseId);
if (!currentLease || (expectedLeaseId && currentLeaseId !== expectedLeaseId)) {
return { ok: false, reason: "lease_mismatch" };
}
const lease: DigitalEmployeePlanStepLease = {
lease_id: currentLeaseId || `plan-step-lease:${safeId(stepId)}:${safeId(now)}`,
state: "released",
owner_device_id: stringValue(currentLease.owner_device_id ?? currentLease.ownerDeviceId) || getDeviceId(),
acquired_at: stringValue(currentLease.acquired_at ?? currentLease.acquiredAt) || now,
expires_at: stringValue(currentLease.expires_at ?? currentLease.expiresAt) || now,
released_at: now,
release_reason: stringValue(input.reason) || "completed",
trigger_source: stringValue(currentLease.trigger_source ?? currentLease.triggerSource) || null,
orchestration_id: stringValue(currentLease.orchestration_id ?? currentLease.orchestrationId) || null,
candidate_count: numberValue(currentLease.candidate_count ?? currentLease.candidateCount),
};
const payload = { ...previousPayload, dispatchLease: lease };
db.prepare(`
UPDATE digital_plan_steps
SET payload = ?, sync_status = 'pending', updated_at = ?
WHERE id = ?
`).run(stringify(payload), now, stepId);
markPlanStepLeaseOutbox(row, payload, now);
recordDigitalEmployeePlanStepLeaseEvent({ row, lease, kind: "plan_step_lease_released", reason: lease.release_reason, now, taskId: input.taskId, runId: input.runId });
return { ok: true, lease };
}
export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
options: number | { limit?: number; planId?: string | null } = 20,
): DigitalEmployeeReadyPlanStepDispatchCandidate[] {
@@ -1408,6 +1559,7 @@ export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
.map((run) => stringValue(run.taskId))
.filter(Boolean));
const openApprovals = records.approvals.filter((approval) => isOpenApprovalStatus(approval.status));
const now = new Date().toISOString();
return records.planSteps
.filter((step) => !planIdFilter || step.planId === planIdFilter)
@@ -1431,6 +1583,8 @@ export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
|| stringValue(payload.stepId ?? payload.step_id) === step.id;
});
const skipReason = readyPlanStepSkipReason({ step, task, activeRunsByTaskId, approvalOpen });
const lease = activeDigitalEmployeePlanStepLease(stepPayload, now);
const effectiveSkipReason = lease ? "lease_active" : skipReason;
const title = step.title || task?.title || step.id;
const prompt = stringValue(stepPayload.prompt ?? stepPayload.description)
|| stringValue(taskPayload.prompt ?? taskPayload.description)
@@ -1443,8 +1597,8 @@ export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
taskTitle: task?.title ?? null,
prompt,
preferredSkillIds,
dispatchable: !skipReason,
skipReason,
dispatchable: !effectiveSkipReason,
skipReason: effectiveSkipReason,
step,
task,
plan,
@@ -1578,6 +1732,66 @@ function updatePlanStepDispatchStatus(
}, now);
}
function markPlanStepLeaseOutbox(row: Record<string, unknown>, payload: Record<string, unknown>, now: string): void {
const stepId = stringValue(row.id);
markOutbox("plan_step", stepId, "upsert", {
id: stepId,
planId: stringValue(row.plan_id),
title: stringValue(row.title),
status: stringValue(row.status),
seq: typeof row.seq === "number" ? row.seq : Number(row.seq) || 0,
dependsOn: parseStringArray(row.depends_on),
preferredSkillIds: parseStringArray(row.preferred_skill_ids),
payload,
now,
}, now);
}
function recordDigitalEmployeePlanStepLeaseEvent(input: {
row: Record<string, unknown>;
lease: DigitalEmployeePlanStepLease;
kind: "plan_step_lease_acquired" | "plan_step_lease_released" | "plan_step_lease_skipped";
reason?: string | null;
now: string;
taskId?: string | null;
runId?: string | null;
}): void {
const stepId = stringValue(input.row.id);
const planId = stringValue(input.row.plan_id) || null;
const taskId = stringValue(input.taskId) || null;
const payload = {
source: "qimingclaw-plan-step-lease",
step_id: stepId,
plan_id: planId,
task_id: taskId,
run_id: stringValue(input.runId) || null,
lease: input.lease,
lease_id: input.lease.lease_id,
lease_state: input.lease.state,
lease_owner_device_id: input.lease.owner_device_id,
lease_expires_at: input.lease.expires_at,
reason: stringValue(input.reason) || input.lease.release_reason || null,
};
insertEvent(
{
event_id: `${stepId}:lease:${input.kind}:${safeId(input.lease.lease_id)}:${safeId(input.now)}`,
kind: input.kind,
occurred_at: input.now,
message: planStepLeaseMessage(input.kind, stepId, stringValue(input.reason) || input.lease.release_reason),
plan_id: planId,
task_id: taskId,
},
stringValue(input.runId) || null,
payload,
);
}
function planStepLeaseMessage(kind: string, stepId: string, reason?: string | null): string {
if (kind === "plan_step_lease_acquired") return `计划步骤租约已获取：${stepId}`;
if (kind === "plan_step_lease_released") return `计划步骤租约已释放：${stepId}${reason ? `（${reason}）` : ""}`;
return `计划步骤租约跳过派发：${stepId}${reason ? `（${reason}）` : ""}`;
}
function planStepDispatchMessage(status: string, stepId: string, reason: string): string {
if (status === "started") return `计划步骤已自动派发：${stepId}`;
if (status === "completed") return `计划步骤派发已触发执行：${stepId}`;

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

@@ -342,6 +342,16 @@ describe("digital employee sync service", () => {
},
lastDispatch: { status: "failed" },
skipReason: "approval_pending",
dispatchLease: {
lease_id: "lease-step-1",
state: "active",
owner_device_id: "device-001",
acquired_at: "2026-06-07T07:59:00.000Z",
expires_at: "2026-06-07T08:04:00.000Z",
trigger_source: "manual",
orchestration_id: "manual-ready-steps-plan-1",
candidate_count: 1,
},
},
}),
);
@@ -382,8 +392,14 @@ describe("digital employee sync service", () => {
waiting_dependency_count: 1,
preferred_skill_ids: ["qimingclaw-mcp-tool-crm-crm-search"],
dispatchable: false,
skip_reason: "approval_pending",
skip_reason: "lease_active",
last_dispatch_status: "failed",
lease_id: "lease-step-1",
lease_state: "active",
lease_owner_device_id: "device-001",
lease_acquired_at: "2026-06-07T07:59:00.000Z",
lease_expires_at: "2026-06-07T08:04:00.000Z",
lease_active: true,
}),
}),
],
@@ -567,6 +583,26 @@ describe("digital employee sync service", () => {
trigger_source: "manual",
orchestration_id: "manual-ready-steps-plan-1",
});
insertEventEntity("event-lease", "plan_step_lease_released", {
step_id: "step-1",
plan_id: "plan-1",
task_id: "task-1",
run_id: "run-1",
lease_id: "lease-step-1",
lease_state: "released",
lease_owner_device_id: "device-001",
lease_expires_at: "2026-06-07T08:04:00.000Z",
reason: "completed",
lease: {
lease_id: "lease-step-1",
state: "released",
owner_device_id: "device-001",
acquired_at: "2026-06-07T07:59:00.000Z",
expires_at: "2026-06-07T08:04:00.000Z",
released_at: "2026-06-07T08:01:00.000Z",
release_reason: "completed",
},
});
insertEventEntity("event-plan-step-ready", "plan_step_ready", {
stepId: "step-2",
planId: "plan-1",
@@ -712,6 +748,7 @@ describe("digital employee sync service", () => {
{ entity_type: "event", entity_id: "event-artifact" },
{ entity_type: "event", entity_id: "event-skill" },
{ entity_type: "event", entity_id: "event-dispatch" },
{ entity_type: "event", entity_id: "event-lease" },
{ entity_type: "event", entity_id: "event-plan-step-ready" },
{ entity_type: "event", entity_id: "event-governance-input" },
{ entity_type: "event", entity_id: "event-notification-read" },
@@ -800,6 +837,21 @@ describe("digital employee sync service", () => {
trigger_source: "manual",
orchestration_id: "manual-ready-steps-plan-1",
});
expect(businessView("event-lease")).toMatchObject({
category: "dispatch",
step_id: "step-1",
plan_id: "plan-1",
task_id: "task-1",
run_id: "run-1",
lease_id: "lease-step-1",
lease_state: "released",
lease_owner_device_id: "device-001",
lease_acquired_at: "2026-06-07T07:59:00.000Z",
lease_expires_at: "2026-06-07T08:04:00.000Z",
lease_released_at: "2026-06-07T08:01:00.000Z",
lease_release_reason: "completed",
reason: "completed",
});
expect(businessView("event-plan-step-ready")).toMatchObject({
category: "dispatch",
step_id: "step-2",

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

@@ -508,9 +508,14 @@ function planStepBusinessView(
const payload = objectRecord(record.payload) ?? {};
const dependencyGraph = objectRecord(payload.dependencyGraph ?? payload.dependency_graph) ?? {};
const lastDispatch = objectRecord(payload.lastDispatch ?? payload.last_dispatch) ?? {};
const dispatchLease = objectRecord(payload.dispatchLease ?? payload.dispatch_lease) ?? {};
const dependsOn = stringArrayField(record.dependsOn ?? record.depends_on);
const blockedDependencies = stringArrayField(dependencyGraph.blockedDependencies ?? dependencyGraph.blocked_dependencies);
const waitingDependencies = stringArrayField(dependencyGraph.waitingDependencies ?? dependencyGraph.waiting_dependencies);
const leaseState = stringField(dispatchLease.state) || null;
const leaseExpiresAt = stringField(dispatchLease.expires_at ?? dispatchLease.expiresAt) || null;
const leaseActive = leaseState === "active" && leaseExpiresAt !== null && leaseExpiresAt > new Date().toISOString();
const skipReason = stringField(payload.skipReason ?? payload.skip_reason) || null;
return {
...base,
plan_id: stringField(record.planId ?? record.plan_id) || null,
@@ -524,9 +529,18 @@ function planStepBusinessView(
waiting_dependencies: waitingDependencies,
waiting_dependency_count: waitingDependencies.length,
preferred_skill_ids: stringArrayField(record.preferredSkillIds ?? record.preferred_skill_ids),
dispatchable: stringField(record.status) === "ready" && blockedDependencies.length === 0 && waitingDependencies.length === 0,
skip_reason: stringField(payload.skipReason ?? payload.skip_reason) || null,
dispatchable: !leaseActive && stringField(record.status) === "ready" && blockedDependencies.length === 0 && waitingDependencies.length === 0,
skip_reason: leaseActive ? "lease_active" : skipReason,
last_dispatch_status: stringField(lastDispatch.status) || null,
lease: Object.keys(dispatchLease).length > 0 ? dispatchLease : null,
lease_id: stringField(dispatchLease.lease_id ?? dispatchLease.leaseId) || null,
lease_state: leaseState,
lease_owner_device_id: stringField(dispatchLease.owner_device_id ?? dispatchLease.ownerDeviceId) || null,
lease_acquired_at: stringField(dispatchLease.acquired_at ?? dispatchLease.acquiredAt) || null,
lease_expires_at: leaseExpiresAt,
lease_released_at: stringField(dispatchLease.released_at ?? dispatchLease.releasedAt) || null,
lease_release_reason: stringField(dispatchLease.release_reason ?? dispatchLease.releaseReason) || null,
lease_active: leaseActive,
};
}
@@ -590,6 +604,7 @@ function eventSpecificBusinessView(
if (kind.startsWith("skill_call_")) return skillCallBusinessView(payload);
if (isPlanStepDependencyEvent(kind)) return planStepDependencyBusinessView(payload);
if (kind.startsWith("plan_step_dispatch_")) return planStepDispatchBusinessView(payload);
if (isPlanStepLeaseEvent(kind)) return planStepLeaseBusinessView(payload);
if (isPlanStepDispatchPolicyEvent(kind)) return planStepDispatchPolicyBusinessView(payload);
if (kind === "governance_provide_task_input") return governanceProvideTaskInputBusinessView(payload);
if (isNotificationEvent(kind)) return notificationBusinessView(payload);
@@ -698,6 +713,24 @@ function planStepDispatchBusinessView(payload: Record<string, unknown>): Record<
};
}
function planStepLeaseBusinessView(payload: Record<string, unknown>): Record<string, unknown> {
const lease = objectRecord(payload.lease) ?? payload;
return {
step_id: stringField(payload.step_id ?? payload.stepId) || null,
plan_id: stringField(payload.plan_id ?? payload.planId) || null,
task_id: stringField(payload.task_id ?? payload.taskId) || null,
run_id: stringField(payload.run_id ?? payload.runId) || null,
lease_id: stringField(payload.lease_id ?? payload.leaseId ?? lease.lease_id ?? lease.leaseId) || null,
lease_state: stringField(payload.lease_state ?? payload.leaseState ?? lease.state) || null,
lease_owner_device_id: stringField(payload.lease_owner_device_id ?? payload.leaseOwnerDeviceId ?? lease.owner_device_id ?? lease.ownerDeviceId) || null,
lease_acquired_at: stringField(lease.acquired_at ?? lease.acquiredAt) || null,
lease_expires_at: stringField(payload.lease_expires_at ?? payload.leaseExpiresAt ?? lease.expires_at ?? lease.expiresAt) || null,
lease_released_at: stringField(lease.released_at ?? lease.releasedAt) || null,
lease_release_reason: stringField(lease.release_reason ?? lease.releaseReason) || null,
reason: stringField(payload.reason) || null,
};
}
function planStepDependencyBusinessView(payload: Record<string, unknown>): Record<string, unknown> {
const dependencyGraph = objectRecord(payload.dependencyGraph ?? payload.dependency_graph) ?? {};
const dependencies = stringArrayField(payload.dependsOn ?? payload.depends_on);
@@ -771,6 +804,7 @@ function eventBusinessCategory(kind: string): string {
if (kind.startsWith("skill_call_")) return "skill";
if (isPlanStepDependencyEvent(kind)) return "dispatch";
if (kind.startsWith("plan_step_dispatch_")) return "dispatch";
if (isPlanStepLeaseEvent(kind)) return "dispatch";
if (isPlanStepDispatchPolicyEvent(kind)) return "dispatch";
if (isNotificationEvent(kind)) return "notification";
if (kind.startsWith("governance_")) return "governance";
@@ -789,6 +823,10 @@ function isPlanStepDispatchPolicyEvent(kind: string): boolean {
return kind === "digital_workday_update_plan_step_dispatch_policy";
}
function isPlanStepLeaseEvent(kind: string): boolean {
return kind === "plan_step_lease_acquired" || kind === "plan_step_lease_released" || kind === "plan_step_lease_skipped";
}
function isNotificationEvent(kind: string): boolean {
return kind === "digital_workday_update_notification_preferences"
|| kind === "digital_workday_notification_read"