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 21:00:59 +08:00
parent a6e0246418
commit 909f11dae5
17 changed files with 646 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -33,7 +33,10 @@ vi.mock("../constants", () => ({
vi.mock("./stateService", () => ({
readDigitalEmployeeCronSettings: () => mockState.settings,
readDigitalEmployeePlanStepDispatchPolicy: () => mockState.dispatchPolicy,
listDigitalEmployeeReadyPlanStepDispatchCandidates: () => mockState.readyCandidates,
listDigitalEmployeeReadyPlanStepDispatchCandidates: (options?: number | { planId?: string | null }) => {
const planId = typeof options === "object" && options ? options.planId : null;
return planId ? mockState.readyCandidates.filter((candidate) => candidate.planId === planId) : mockState.readyCandidates;
},
listDigitalEmployeeSchedules: () => mockState.schedules,
listDueDigitalEmployeeSchedules: () => mockState.schedules,
readDigitalEmployeeScheduleRuns: () => [],
@@ -203,6 +206,58 @@ describe("digital employee scheduler service", () => {
expect(mockState.dispatchEvents).toHaveLength(0);
});
it("allows manual plan dispatch to bypass the auto dispatch switch", async () => {
mockState.readyCandidates = [readyStepCandidate()];
mockState.dispatchPolicy = { ...mockState.dispatchPolicy, auto_dispatch_ready_steps: false };
const { dispatchReadyPlanSteps } = await import("./schedulerService");
const result = await dispatchReadyPlanSteps({
planId: "plan-1",
triggerSource: "manual",
bypassAutoDispatchSwitch: true,
now: new Date("2026-06-07T01:00:10.000Z"),
});
expect(result).toMatchObject({ plan_id: "plan-1", trigger_source: "manual", candidates: 1, dispatched: 1, skipped: 0, failed: 0 });
expect(mockState.dispatchEvents[0]).toMatchObject({ status: "started", payload: expect.objectContaining({ trigger_source: "manual" }) });
});
it("keeps the enabled switch as a manual plan dispatch kill switch", async () => {
mockState.readyCandidates = [readyStepCandidate()];
mockState.dispatchPolicy = { ...mockState.dispatchPolicy, enabled: false, auto_dispatch_ready_steps: false };
const { dispatchReadyPlanSteps } = await import("./schedulerService");
const result = await dispatchReadyPlanSteps({
planId: "plan-1",
triggerSource: "manual",
bypassAutoDispatchSwitch: true,
now: new Date("2026-06-07T01:00:10.000Z"),
});
expect(result).toMatchObject({ candidates: 1, dispatched: 0, skipped: 1, failed: 0 });
expect(fetch).not.toHaveBeenCalled();
expect(mockState.dispatchEvents).toHaveLength(0);
});
it("dispatches only the requested plan in manual plan dispatch", async () => {
mockState.readyCandidates = [
readyStepCandidate(),
readyStepCandidate({ planId: "plan-2", stepId: "step-plan-2", taskId: "task-plan-2" }),
];
const { dispatchReadyPlanSteps } = await import("./schedulerService");
const result = await dispatchReadyPlanSteps({
planId: "plan-2",
triggerSource: "manual",
bypassAutoDispatchSwitch: true,
now: new Date("2026-06-07T01:00:10.000Z"),
});
expect(result).toMatchObject({ plan_id: "plan-2", candidates: 1, dispatched: 1, skipped: 0, failed: 0 });
expect(result.dispatched_step_ids).toEqual(["step-plan-2"]);
expect(mockState.dispatchEvents.map((event) => event.stepId)).toEqual(["step-plan-2", "step-plan-2"]);
});
it("limits ready step dispatches by policy max per sweep", async () => {
mockState.readyCandidates = [
readyStepCandidate(),
@@ -219,18 +274,22 @@ describe("digital employee scheduler service", () => {
});
});
function readyStepCandidate(): Record<string, unknown> {
function readyStepCandidate(overrides: Record<string, unknown> = {}): Record<string, unknown> {
const stepId = typeof overrides.stepId === "string" ? overrides.stepId : "step-ready";
const taskId = typeof overrides.taskId === "string" ? overrides.taskId : "task-1";
const planId = typeof overrides.planId === "string" ? overrides.planId : "plan-1";
return {
stepId: "step-ready",
planId: "plan-1",
taskId: "task-1",
stepId,
planId,
taskId,
title: "汇总客户跟进",
taskTitle: "执行客户跟进汇总",
prompt: "执行客户跟进汇总",
preferredSkillIds: ["qimingclaw-acp-session"],
dispatchable: true,
step: { id: "step-ready", payload: {} },
task: { id: "task-1", payload: {} },
plan: { id: "plan-1" },
step: { id: stepId, payload: {} },
task: { id: taskId, payload: {} },
plan: { id: planId },
...overrides,
};
}

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

@@ -35,13 +35,29 @@ export interface DigitalEmployeeSchedulerCheckResult {
}
export interface DigitalEmployeeReadyStepDispatchResult {
ok?: boolean;
plan_id?: string | null;
trigger_source?: DigitalEmployeeReadyStepDispatchTriggerSource;
orchestration_id?: string;
candidates: number;
dispatched: number;
skipped: number;
failed: number;
dispatched_step_ids?: string[];
skipped_step_ids?: string[];
failed_step_ids?: string[];
errors: Array<{ stepId: string; error: string }>;
}
export type DigitalEmployeeReadyStepDispatchTriggerSource = "auto" | "manual";
export interface DigitalEmployeeReadyStepDispatchOptions {
now?: Date;
planId?: string | null;
triggerSource?: DigitalEmployeeReadyStepDispatchTriggerSource;
bypassAutoDispatchSwitch?: boolean;
}
export function startDigitalEmployeeScheduler(): void {
if (schedulerTimer) return;
schedulerTimer = setInterval(() => {
@@ -107,7 +123,7 @@ export async function runDigitalEmployeeSchedulerCheck(
}
}
}
result.readyStepDispatch = await dispatchReadyPlanSteps(now);
result.readyStepDispatch = await dispatchReadyPlanSteps({ now, triggerSource: "auto" });
} finally {
schedulerSweeping = false;
}
@@ -116,19 +132,32 @@ export async function runDigitalEmployeeSchedulerCheck(
}
export async function dispatchReadyPlanSteps(
now = new Date(),
options: Date | DigitalEmployeeReadyStepDispatchOptions = {},
): Promise<DigitalEmployeeReadyStepDispatchResult> {
const candidates = listDigitalEmployeeReadyPlanStepDispatchCandidates(20);
const now = options instanceof Date ? options : options.now ?? new Date();
const planId = options instanceof Date ? null : stringFromUnknown(options.planId) || null;
const triggerSource = options instanceof Date ? "auto" : options.triggerSource ?? "auto";
const bypassAutoDispatchSwitch = !(options instanceof Date) && options.bypassAutoDispatchSwitch === true;
const orchestrationId = `${triggerSource}-ready-steps-${safeId(planId || "all")}-${safeId(now.toISOString())}`;
const candidates = listDigitalEmployeeReadyPlanStepDispatchCandidates({ limit: 20, planId });
const result: DigitalEmployeeReadyStepDispatchResult = {
ok: true,
plan_id: planId,
trigger_source: triggerSource,
orchestration_id: orchestrationId,
candidates: candidates.length,
dispatched: 0,
skipped: 0,
failed: 0,
dispatched_step_ids: [],
skipped_step_ids: [],
failed_step_ids: [],
errors: [],
};
const policy = readDigitalEmployeePlanStepDispatchPolicy();
if (!policy.enabled || !policy.auto_dispatch_ready_steps) {
if (!policy.enabled || (!policy.auto_dispatch_ready_steps && !bypassAutoDispatchSwitch)) {
result.skipped = candidates.length;
result.skipped_step_ids = candidates.map((candidate) => candidate.stepId);
return result;
}
const maxPerSweep = Math.max(1, Math.min(Math.floor(policy.max_per_sweep), 10));
@@ -136,19 +165,23 @@ export async function dispatchReadyPlanSteps(
for (const candidate of candidates) {
if (!candidate.dispatchable || !candidate.taskId) {
result.skipped += 1;
result.skipped_step_ids?.push(candidate.stepId);
continue;
}
if (attempts >= maxPerSweep) {
result.skipped += 1;
result.skipped_step_ids?.push(candidate.stepId);
continue;
}
attempts += 1;
try {
await dispatchReadyPlanStep(candidate, now);
await dispatchReadyPlanStep(candidate, now, { triggerSource, orchestrationId, candidateCount: candidates.length });
result.dispatched += 1;
result.dispatched_step_ids?.push(candidate.stepId);
} catch (error) {
const message = normalizeError(error);
result.failed += 1;
result.failed_step_ids?.push(candidate.stepId);
result.errors.push({ stepId: candidate.stepId, error: message });
recordDigitalEmployeePlanStepDispatch({
stepId: candidate.stepId,
@@ -156,6 +189,8 @@ export async function dispatchReadyPlanSteps(
taskId: candidate.taskId,
status: "failed",
reason: message,
source: "qimingclaw-ready-step-dispatcher",
payload: { trigger_source: triggerSource, orchestration_id: orchestrationId, candidate_count: candidates.length },
occurredAt: now.toISOString(),
});
}
@@ -164,12 +199,22 @@ export async function dispatchReadyPlanSteps(
}
function emptyReadyStepDispatchResult(): DigitalEmployeeReadyStepDispatchResult {
return { candidates: 0, dispatched: 0, skipped: 0, failed: 0, errors: [] };
return {
candidates: 0,
dispatched: 0,
skipped: 0,
failed: 0,
dispatched_step_ids: [],
skipped_step_ids: [],
failed_step_ids: [],
errors: [],
};
}
async function dispatchReadyPlanStep(
candidate: DigitalEmployeeReadyPlanStepDispatchCandidate,
now: Date,
orchestration: { triggerSource: DigitalEmployeeReadyStepDispatchTriggerSource; orchestrationId: string; candidateCount: number },
): Promise<void> {
const occurredAt = now.toISOString();
const commandId = `ready-step-${safeId(candidate.stepId)}-${safeId(occurredAt)}`;
@@ -180,7 +225,13 @@ async function dispatchReadyPlanStep(
taskId: candidate.taskId,
runId,
status: "started",
payload: { command_id: commandId, preferred_skill_ids: candidate.preferredSkillIds },
payload: {
command_id: commandId,
preferred_skill_ids: candidate.preferredSkillIds,
trigger_source: orchestration.triggerSource,
orchestration_id: orchestration.orchestrationId,
candidate_count: orchestration.candidateCount,
},
occurredAt,
});
const chatRequest = buildReadyStepChatRequest(candidate, runId, occurredAt);
@@ -200,6 +251,8 @@ async function dispatchReadyPlanStep(
plan_id: candidate.planId,
task_id: candidate.taskId,
step_id: candidate.stepId,
trigger_source: orchestration.triggerSource,
orchestration_id: orchestration.orchestrationId,
title: candidate.title,
task_title: candidate.taskTitle,
prompt: candidate.prompt,
@@ -214,7 +267,13 @@ async function dispatchReadyPlanStep(
taskId: candidate.taskId,
runId: ids.runId,
status: "completed",
payload: { command_id: commandId, response },
payload: {
command_id: commandId,
response,
trigger_source: orchestration.triggerSource,
orchestration_id: orchestration.orchestrationId,
candidate_count: orchestration.candidateCount,
},
occurredAt,
});
}
@@ -232,6 +291,20 @@ export async function runDigitalEmployeeScheduleNow(
}
}
export async function runDigitalEmployeePlanReadyStepsNow(
planId: string,
): Promise<DigitalEmployeeReadyStepDispatchResult> {
const targetPlanId = stringFromUnknown(planId);
if (!targetPlanId) {
return { ...emptyReadyStepDispatchResult(), ok: false, plan_id: null, trigger_source: "manual", errors: [{ stepId: "", error: "plan id required" }] };
}
return dispatchReadyPlanSteps({
planId: targetPlanId,
triggerSource: "manual",
bypassAutoDispatchSwitch: true,
});
}
async function initializeScheduleNextRuns(now = new Date()): Promise<void> {
for (const schedule of listDigitalEmployeeSchedules()) {
if (!schedule.enabled || schedule.status !== "enabled" || schedule.deletedAt) continue;

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

@@ -1391,6 +1391,20 @@ describe("digital employee state service", () => {
});
});
it("filters ready plan step dispatch candidates by plan", async () => {
const { listDigitalEmployeeReadyPlanStepDispatchCandidates } = await import("./stateService");
createGovernableTask({ taskStatus: "pending", stepStatus: "ready", runStatus: "completed", finishedAt: "2026-06-07T07:20:00.000Z" });
addReadyPlanStepWithTask({ planId: "plan-other", stepId: "step-other-ready", taskId: "task-other-ready" });
expect(listDigitalEmployeeReadyPlanStepDispatchCandidates({ planId: "plan-governance" })).toEqual([
expect.objectContaining({ planId: "plan-governance", stepId: "step-governance" }),
]);
expect(listDigitalEmployeeReadyPlanStepDispatchCandidates({ planId: "plan-other" })).toEqual([
expect.objectContaining({ planId: "plan-other", stepId: "step-other-ready" }),
]);
});
it("records ready plan step dispatch events and updates step status", async () => {
const { recordDigitalEmployeePlanStepDispatch } = await import("./stateService");
@@ -1512,6 +1526,11 @@ describe("digital employee state service", () => {
});
expect(mockState.db?.planSteps.get("step-dependent")).toMatchObject({ status: "ready" });
addReadyPlanStepWithTask({ planId: "plan-governance", stepId: "step-dependent", taskId: "task-dependent" });
const { listDigitalEmployeeReadyPlanStepDispatchCandidates } = await import("./stateService");
expect(listDigitalEmployeeReadyPlanStepDispatchCandidates({ planId: "plan-governance" })).toEqual(expect.arrayContaining([
expect.objectContaining({ stepId: "step-dependent", dispatchable: true }),
]));
expect(JSON.parse(mockState.db?.planSteps.get("step-dependent")?.payload ?? "{}")).toMatchObject({
dependencyGraph: {
status: "ready",
@@ -1856,6 +1875,50 @@ function addPlanStep(options: {
});
}
function addReadyPlanStepWithTask(options: {
planId: string;
stepId: string;
taskId: string;
}): void {
if (!mockState.db?.plans.has(options.planId)) {
mockState.db?.plans.set(options.planId, {
id: options.planId,
title: options.planId,
objective: "验证计划级派发候选",
status: "running",
payload: JSON.stringify({ source: "test" }),
sync_status: "synced",
created_at: "2026-06-07T07:00:00.000Z",
updated_at: "2026-06-07T07:00:00.000Z",
});
}
const existingStep = mockState.db?.planSteps.get(options.stepId);
mockState.db?.planSteps.set(options.stepId, {
id: options.stepId,
plan_id: options.planId,
title: existingStep?.title ?? options.stepId,
status: "ready",
seq: existingStep?.seq ?? 1,
depends_on: existingStep?.depends_on ?? JSON.stringify([]),
preferred_skill_ids: existingStep?.preferred_skill_ids ?? JSON.stringify(["qimingclaw-task-agent"]),
payload: existingStep?.payload ?? JSON.stringify({ source: "test" }),
sync_status: existingStep?.sync_status ?? "synced",
created_at: existingStep?.created_at ?? "2026-06-07T07:00:00.000Z",
updated_at: "2026-06-07T07:00:00.000Z",
});
mockState.db?.tasks.set(options.taskId, {
id: options.taskId,
plan_id: options.planId,
title: `${options.stepId} 任务`,
status: "pending",
assigned_skill_id: "qimingclaw-task-agent",
payload: JSON.stringify({ source: "test", stepId: options.stepId }),
sync_status: "synced",
created_at: "2026-06-07T07:00:00.000Z",
updated_at: "2026-06-07T07:00:00.000Z",
});
}
function addGovernableTaskForStep(options: {
taskId: string;
runId: string;

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

@@ -1385,8 +1385,10 @@ export function recordDigitalEmployeeApprovalAction(
}
export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
limit = 20,
options: number | { limit?: number; planId?: string | null } = 20,
): DigitalEmployeeReadyPlanStepDispatchCandidate[] {
const limit = typeof options === "number" ? options : options.limit ?? 20;
const planIdFilter = typeof options === "number" ? "" : stringValue(options.planId);
const records = readDigitalEmployeeRuntimeRecords();
const tasksByStepId = new Map<string, DigitalEmployeeTaskRecord>();
for (const task of records.tasks) {
@@ -1408,8 +1410,9 @@ export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
const openApprovals = records.approvals.filter((approval) => isOpenApprovalStatus(approval.status));
return records.planSteps
.filter((step) => !planIdFilter || step.planId === planIdFilter)
.filter((step) => normalizePlanStepStatus(step.status) === "ready")
.sort((left, right) => left.seq - right.seq || left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id))
.sort((left, right) => left.planId.localeCompare(right.planId) || left.seq - right.seq || left.createdAt.localeCompare(right.createdAt) || left.id.localeCompare(right.id))
.slice(0, Math.max(1, limit))
.map((step) => {
const task = tasksByStepId.get(step.id) ?? firstRunnableTaskForStep(step, tasksByPlanId.get(step.planId) ?? []) ?? null;

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

@@ -564,6 +564,8 @@ describe("digital employee sync service", () => {
run_id: "run-1",
status: "failed",
reason: "computer chat failed",
trigger_source: "manual",
orchestration_id: "manual-ready-steps-plan-1",
});
insertEventEntity("event-plan-step-ready", "plan_step_ready", {
stepId: "step-2",
@@ -795,6 +797,8 @@ describe("digital employee sync service", () => {
run_id: "run-1",
status: "failed",
reason: "computer chat failed",
trigger_source: "manual",
orchestration_id: "manual-ready-steps-plan-1",
});
expect(businessView("event-plan-step-ready")).toMatchObject({
category: "dispatch",

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

@@ -693,6 +693,8 @@ function planStepDispatchBusinessView(payload: Record<string, unknown>): Record<
run_id: stringField(payload.run_id ?? payload.runId) || null,
status: stringField(payload.status) || null,
reason: stringField(payload.reason) || null,
trigger_source: stringField(payload.trigger_source ?? payload.triggerSource) || null,
orchestration_id: stringField(payload.orchestration_id ?? payload.orchestrationId) || null,
};
}