吸收数字员工步骤自动调度
This commit is contained in:
baiyanyun
@@ -2,8 +2,11 @@ import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
const mockState = vi.hoisted(() => ({
|
||||
schedules: [] as Array<Record<string, unknown>>,
|
||||
readyCandidates: [] as Array<Record<string, unknown>>,
|
||||
startedRuns: [] as Array<Record<string, unknown>>,
|
||||
finishedRuns: [] as Array<Record<string, unknown>>,
|
||||
dispatchEvents: [] as Array<Record<string, unknown>>,
|
||||
governanceCommands: [] as Array<Record<string, unknown>>,
|
||||
settings: {
|
||||
enabled: true,
|
||||
catch_up_on_startup: true,
|
||||
@@ -23,6 +26,7 @@ vi.mock("../constants", () => ({
|
||||
|
||||
vi.mock("./stateService", () => ({
|
||||
readDigitalEmployeeCronSettings: () => mockState.settings,
|
||||
listDigitalEmployeeReadyPlanStepDispatchCandidates: () => mockState.readyCandidates,
|
||||
listDigitalEmployeeSchedules: () => mockState.schedules,
|
||||
listDueDigitalEmployeeSchedules: () => mockState.schedules,
|
||||
readDigitalEmployeeScheduleRuns: () => [],
|
||||
@@ -36,13 +40,28 @@ vi.mock("./stateService", () => ({
|
||||
mockState.finishedRuns.push({ id, status, ...options });
|
||||
return { id, status, ...options };
|
||||
},
|
||||
recordDigitalEmployeePlanStepDispatch: (input: Record<string, unknown>) => {
|
||||
mockState.dispatchEvents.push(input);
|
||||
return { eventId: `dispatch-${input.status}`, kind: `plan_step_dispatch_${input.status}`, payload: input };
|
||||
},
|
||||
recordDigitalEmployeeGovernanceCommand: (input: Record<string, unknown>) => {
|
||||
mockState.governanceCommands.push(input);
|
||||
return {
|
||||
planId: input.planId,
|
||||
taskId: input.taskId,
|
||||
runId: input.runId || `${input.taskId}:run:start_run:${input.commandId}`,
|
||||
};
|
||||
},
|
||||
}));
|
||||
|
||||
describe("digital employee scheduler service", () => {
|
||||
beforeEach(() => {
|
||||
vi.resetModules();
|
||||
mockState.readyCandidates = [];
|
||||
mockState.startedRuns = [];
|
||||
mockState.finishedRuns = [];
|
||||
mockState.dispatchEvents = [];
|
||||
mockState.governanceCommands = [];
|
||||
mockState.schedules = [{
|
||||
id: "schedule-1",
|
||||
planId: "plan-1",
|
||||
@@ -107,4 +126,59 @@ describe("digital employee scheduler service", () => {
|
||||
expect(mockState.finishedRuns[0]).toMatchObject({ status: "failed", error: "ComputerServer 500: boom" });
|
||||
expect(mockState.finishedRuns[0].nextRunAt).toBeTruthy();
|
||||
});
|
||||
|
||||
it("dispatches ready plan steps through local ComputerServer", async () => {
|
||||
mockState.schedules = [];
|
||||
mockState.readyCandidates = [readyStepCandidate()];
|
||||
const { runDigitalEmployeeSchedulerCheck } = await import("./schedulerService");
|
||||
|
||||
const result = await runDigitalEmployeeSchedulerCheck(new Date("2026-06-07T01:00:10.000Z"));
|
||||
|
||||
expect(result.readyStepDispatch).toMatchObject({ candidates: 1, dispatched: 1, skipped: 0, failed: 0 });
|
||||
expect(fetch).toHaveBeenCalledWith(
|
||||
"http://127.0.0.1:60006/computer/chat",
|
||||
expect.objectContaining({ method: "POST" }),
|
||||
);
|
||||
expect(mockState.dispatchEvents.map((event) => event.status)).toEqual(["started", "completed"]);
|
||||
expect(mockState.governanceCommands[0]).toMatchObject({
|
||||
commandKind: "start_run",
|
||||
planId: "plan-1",
|
||||
taskId: "task-1",
|
||||
stepId: "step-ready",
|
||||
});
|
||||
});
|
||||
|
||||
it("marks ready step dispatch failures", async () => {
|
||||
mockState.schedules = [];
|
||||
mockState.readyCandidates = [readyStepCandidate()];
|
||||
vi.stubGlobal("fetch", vi.fn(async () => ({
|
||||
ok: false,
|
||||
status: 502,
|
||||
text: async () => "bad gateway",
|
||||
})));
|
||||
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, 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.governanceCommands).toHaveLength(0);
|
||||
});
|
||||
});
|
||||
|
||||
function readyStepCandidate(): Record<string, unknown> {
|
||||
return {
|
||||
stepId: "step-ready",
|
||||
planId: "plan-1",
|
||||
taskId: "task-1",
|
||||
title: "汇总客户跟进",
|
||||
taskTitle: "执行客户跟进汇总",
|
||||
prompt: "执行客户跟进汇总",
|
||||
preferredSkillIds: ["qimingclaw-acp-session"],
|
||||
dispatchable: true,
|
||||
step: { id: "step-ready", payload: {} },
|
||||
task: { id: "task-1", payload: {} },
|
||||
plan: { id: "plan-1" },
|
||||
};
|
||||
}
|
||||
|
||||
@@ -3,18 +3,23 @@ import { LOCALHOST_HOSTNAME } from "../constants";
|
||||
import { getConfiguredPorts } from "../startupPorts";
|
||||
import {
|
||||
finishDigitalEmployeeScheduleRun,
|
||||
listDigitalEmployeeReadyPlanStepDispatchCandidates,
|
||||
listDigitalEmployeeSchedules,
|
||||
listDueDigitalEmployeeSchedules,
|
||||
readDigitalEmployeeCronSettings,
|
||||
readDigitalEmployeeScheduleRuns,
|
||||
recordDigitalEmployeeGovernanceCommand,
|
||||
recordDigitalEmployeePlanStepDispatch,
|
||||
recordDigitalEmployeeScheduleRunStarted,
|
||||
upsertDigitalEmployeeSchedule,
|
||||
type DigitalEmployeeReadyPlanStepDispatchCandidate,
|
||||
type DigitalEmployeeScheduleRecord,
|
||||
} from "./stateService";
|
||||
|
||||
const SWEEP_INTERVAL_MS = 30_000;
|
||||
const MAX_SCAN_MINUTES = 366 * 24 * 60;
|
||||
const RETRY_DELAYS_MS = [60_000, 5 * 60_000, 15 * 60_000];
|
||||
const MAX_READY_STEP_DISPATCH_PER_SWEEP = 3;
|
||||
|
||||
let schedulerTimer: ReturnType<typeof setInterval> | null = null;
|
||||
let schedulerSweeping = false;
|
||||
@@ -26,6 +31,15 @@ export interface DigitalEmployeeSchedulerCheckResult {
|
||||
activated: number;
|
||||
skipped: number;
|
||||
errors: Array<{ scheduleId: string; error: string }>;
|
||||
readyStepDispatch: DigitalEmployeeReadyStepDispatchResult;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeReadyStepDispatchResult {
|
||||
candidates: number;
|
||||
dispatched: number;
|
||||
skipped: number;
|
||||
failed: number;
|
||||
errors: Array<{ stepId: string; error: string }>;
|
||||
}
|
||||
|
||||
export function startDigitalEmployeeScheduler(): void {
|
||||
@@ -52,11 +66,11 @@ export async function runDigitalEmployeeSchedulerCheck(
|
||||
now = new Date(),
|
||||
): Promise<DigitalEmployeeSchedulerCheckResult> {
|
||||
if (schedulerSweeping) {
|
||||
return { checkedAt: now.toISOString(), due: 0, activated: 0, skipped: 0, errors: [] };
|
||||
return { checkedAt: now.toISOString(), due: 0, activated: 0, skipped: 0, errors: [], readyStepDispatch: emptyReadyStepDispatchResult() };
|
||||
}
|
||||
const settings = readDigitalEmployeeCronSettings();
|
||||
if (!settings.enabled) {
|
||||
return { checkedAt: now.toISOString(), due: 0, activated: 0, skipped: 0, errors: [] };
|
||||
return { checkedAt: now.toISOString(), due: 0, activated: 0, skipped: 0, errors: [], readyStepDispatch: emptyReadyStepDispatchResult() };
|
||||
}
|
||||
|
||||
schedulerSweeping = true;
|
||||
@@ -67,6 +81,7 @@ export async function runDigitalEmployeeSchedulerCheck(
|
||||
activated: 0,
|
||||
skipped: 0,
|
||||
errors: [],
|
||||
readyStepDispatch: emptyReadyStepDispatchResult(),
|
||||
};
|
||||
|
||||
try {
|
||||
@@ -92,6 +107,7 @@ export async function runDigitalEmployeeSchedulerCheck(
|
||||
}
|
||||
}
|
||||
}
|
||||
result.readyStepDispatch = await dispatchReadyPlanSteps(now);
|
||||
} finally {
|
||||
schedulerSweeping = false;
|
||||
}
|
||||
@@ -99,6 +115,104 @@ export async function runDigitalEmployeeSchedulerCheck(
|
||||
return result;
|
||||
}
|
||||
|
||||
export async function dispatchReadyPlanSteps(
|
||||
now = new Date(),
|
||||
): Promise<DigitalEmployeeReadyStepDispatchResult> {
|
||||
const candidates = listDigitalEmployeeReadyPlanStepDispatchCandidates(20);
|
||||
const result: DigitalEmployeeReadyStepDispatchResult = {
|
||||
candidates: candidates.length,
|
||||
dispatched: 0,
|
||||
skipped: 0,
|
||||
failed: 0,
|
||||
errors: [],
|
||||
};
|
||||
let attempts = 0;
|
||||
for (const candidate of candidates) {
|
||||
if (!candidate.dispatchable || !candidate.taskId) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
if (attempts >= MAX_READY_STEP_DISPATCH_PER_SWEEP) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
attempts += 1;
|
||||
try {
|
||||
await dispatchReadyPlanStep(candidate, now);
|
||||
result.dispatched += 1;
|
||||
} catch (error) {
|
||||
const message = normalizeError(error);
|
||||
result.failed += 1;
|
||||
result.errors.push({ stepId: candidate.stepId, error: message });
|
||||
recordDigitalEmployeePlanStepDispatch({
|
||||
stepId: candidate.stepId,
|
||||
planId: candidate.planId,
|
||||
taskId: candidate.taskId,
|
||||
status: "failed",
|
||||
reason: message,
|
||||
occurredAt: now.toISOString(),
|
||||
});
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function emptyReadyStepDispatchResult(): DigitalEmployeeReadyStepDispatchResult {
|
||||
return { candidates: 0, dispatched: 0, skipped: 0, failed: 0, errors: [] };
|
||||
}
|
||||
|
||||
async function dispatchReadyPlanStep(
|
||||
candidate: DigitalEmployeeReadyPlanStepDispatchCandidate,
|
||||
now: Date,
|
||||
): Promise<void> {
|
||||
const occurredAt = now.toISOString();
|
||||
const commandId = `ready-step-${safeId(candidate.stepId)}-${safeId(occurredAt)}`;
|
||||
const runId = `${candidate.taskId}:run:start_run:${safeId(commandId)}`;
|
||||
recordDigitalEmployeePlanStepDispatch({
|
||||
stepId: candidate.stepId,
|
||||
planId: candidate.planId,
|
||||
taskId: candidate.taskId,
|
||||
runId,
|
||||
status: "started",
|
||||
payload: { command_id: commandId, preferred_skill_ids: candidate.preferredSkillIds },
|
||||
occurredAt,
|
||||
});
|
||||
const chatRequest = buildReadyStepChatRequest(candidate, runId, occurredAt);
|
||||
const response = await postComputerChat(chatRequest);
|
||||
const ids = recordDigitalEmployeeGovernanceCommand({
|
||||
commandKind: "start_run",
|
||||
commandId,
|
||||
correlationId: commandId,
|
||||
planId: candidate.planId,
|
||||
taskId: candidate.taskId,
|
||||
runId,
|
||||
stepId: candidate.stepId,
|
||||
accepted: true,
|
||||
message: `自动派发 Ready 步骤：${candidate.title}`,
|
||||
payload: {
|
||||
source: "qimingclaw-ready-step-dispatcher",
|
||||
plan_id: candidate.planId,
|
||||
task_id: candidate.taskId,
|
||||
step_id: candidate.stepId,
|
||||
title: candidate.title,
|
||||
task_title: candidate.taskTitle,
|
||||
prompt: candidate.prompt,
|
||||
preferred_skill_ids: candidate.preferredSkillIds,
|
||||
chat_request: chatRequest,
|
||||
},
|
||||
result: { status: "running", response },
|
||||
});
|
||||
recordDigitalEmployeePlanStepDispatch({
|
||||
stepId: candidate.stepId,
|
||||
planId: candidate.planId,
|
||||
taskId: candidate.taskId,
|
||||
runId: ids.runId,
|
||||
status: "completed",
|
||||
payload: { command_id: commandId, response },
|
||||
occurredAt,
|
||||
});
|
||||
}
|
||||
|
||||
export async function runDigitalEmployeeScheduleNow(
|
||||
scheduleId: string,
|
||||
): Promise<{ ok: boolean; scheduleRunId?: string; error?: string }> {
|
||||
@@ -212,6 +326,38 @@ function buildScheduleChatRequest(
|
||||
};
|
||||
}
|
||||
|
||||
function buildReadyStepChatRequest(
|
||||
candidate: DigitalEmployeeReadyPlanStepDispatchCandidate,
|
||||
runId: string,
|
||||
occurredAt: string,
|
||||
) {
|
||||
const payload = objectPayload(candidate.step.payload);
|
||||
const taskPayload = objectPayload(candidate.task?.payload);
|
||||
const prompt = candidate.prompt || stringFromUnknown(payload.prompt) || stringFromUnknown(taskPayload.prompt) || candidate.title;
|
||||
return {
|
||||
user_id: stringFromUnknown(taskPayload.user_id) || "qimingclaw-ready-step-dispatcher",
|
||||
project_id: candidate.planId,
|
||||
prompt,
|
||||
session_id: stringFromUnknown(taskPayload.session_id) || `planstep-${candidate.stepId}`,
|
||||
request_id: `ready-step-${safeId(candidate.stepId)}-${safeId(occurredAt)}`,
|
||||
system_prompt: stringFromUnknown(taskPayload.system_prompt) || "你是 qimingclaw 本地数字员工，请执行已就绪的计划步骤并记录必要的结果。",
|
||||
original_user_prompt: prompt,
|
||||
open_long_memory: taskPayload.open_long_memory === true,
|
||||
agent_config: objectPayload(taskPayload.agent_config),
|
||||
model_provider: objectPayload(taskPayload.model_provider),
|
||||
ready_step_context: {
|
||||
plan_id: candidate.planId,
|
||||
task_id: candidate.taskId,
|
||||
step_id: candidate.stepId,
|
||||
run_id: runId,
|
||||
title: candidate.title,
|
||||
task_title: candidate.taskTitle,
|
||||
preferred_skill_ids: candidate.preferredSkillIds,
|
||||
dispatched_at: occurredAt,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function postComputerChat(body: Record<string, unknown>): Promise<unknown> {
|
||||
const { agent: agentPort } = getConfiguredPorts();
|
||||
const response = await fetch(`http://${LOCALHOST_HOSTNAME}:${agentPort}/computer/chat`, {
|
||||
|
||||
@@ -1153,6 +1153,85 @@ describe("digital employee state service", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("lists dispatchable ready plan steps and skips approval-blocked steps", async () => {
|
||||
const { listDigitalEmployeeReadyPlanStepDispatchCandidates } = await import("./stateService");
|
||||
|
||||
createGovernableTask({ taskStatus: "pending", stepStatus: "ready", runStatus: "completed", finishedAt: "2026-06-07T07:20:00.000Z" });
|
||||
|
||||
expect(listDigitalEmployeeReadyPlanStepDispatchCandidates()).toEqual([
|
||||
expect.objectContaining({
|
||||
stepId: "step-governance",
|
||||
planId: "plan-governance",
|
||||
taskId: "task-governance",
|
||||
dispatchable: true,
|
||||
preferredSkillIds: ["qimingclaw-task-agent"],
|
||||
}),
|
||||
]);
|
||||
|
||||
mockState.db?.approvals.set("approval-ready-step", {
|
||||
id: "approval-ready-step",
|
||||
plan_id: "plan-governance",
|
||||
task_id: "task-governance",
|
||||
run_id: null,
|
||||
title: "确认派发步骤",
|
||||
status: "pending",
|
||||
payload: JSON.stringify({ stepId: "step-governance" }),
|
||||
sync_status: "pending",
|
||||
created_at: "2026-06-07T07:30:00.000Z",
|
||||
updated_at: "2026-06-07T07:30:00.000Z",
|
||||
});
|
||||
|
||||
expect(listDigitalEmployeeReadyPlanStepDispatchCandidates()[0]).toMatchObject({
|
||||
stepId: "step-governance",
|
||||
dispatchable: false,
|
||||
skipReason: "approval_pending",
|
||||
});
|
||||
});
|
||||
|
||||
it("records ready plan step dispatch events and updates step status", async () => {
|
||||
const { recordDigitalEmployeePlanStepDispatch } = await import("./stateService");
|
||||
|
||||
createGovernableTask({ taskStatus: "pending", stepStatus: "ready", runStatus: "completed", finishedAt: "2026-06-07T07:20:00.000Z" });
|
||||
const started = recordDigitalEmployeePlanStepDispatch({
|
||||
stepId: "step-governance",
|
||||
planId: "plan-governance",
|
||||
taskId: "task-governance",
|
||||
runId: "run-dispatch",
|
||||
status: "started",
|
||||
payload: { command_id: "command-dispatch" },
|
||||
occurredAt: "2026-06-07T08:10:00.000Z",
|
||||
});
|
||||
|
||||
expect(started?.eventId).toBe("plan-step-dispatch:step-governance:started:2026-06-07T08-10-00-000Z");
|
||||
expect(mockState.db?.planSteps.get("step-governance")).toMatchObject({ status: "running", sync_status: "pending" });
|
||||
expect(mockState.db?.events.get(started!.eventId)).toMatchObject({
|
||||
kind: "plan_step_dispatch_started",
|
||||
plan_id: "plan-governance",
|
||||
task_id: "task-governance",
|
||||
run_id: "run-dispatch",
|
||||
});
|
||||
expect(mockState.db?.outbox.get("plan_step:upsert:step-governance")).toMatchObject({ entity_type: "plan_step" });
|
||||
expect(mockState.db?.outbox.get(`event:insert:${started!.eventId}`)).toMatchObject({ entity_type: "event" });
|
||||
|
||||
const failed = recordDigitalEmployeePlanStepDispatch({
|
||||
stepId: "step-governance",
|
||||
planId: "plan-governance",
|
||||
taskId: "task-governance",
|
||||
runId: "run-dispatch",
|
||||
status: "failed",
|
||||
reason: "ComputerServer 500",
|
||||
occurredAt: "2026-06-07T08:11:00.000Z",
|
||||
});
|
||||
|
||||
expect(failed?.kind).toBe("plan_step_dispatch_failed");
|
||||
expect(mockState.db?.planSteps.get("step-governance")).toMatchObject({ status: "blocked", sync_status: "pending" });
|
||||
expect(JSON.parse(mockState.db?.events.get(failed!.eventId)?.payload ?? "{}")).toMatchObject({
|
||||
step_id: "step-governance",
|
||||
status: "failed",
|
||||
reason: "ComputerServer 500",
|
||||
});
|
||||
});
|
||||
|
||||
it("blocks dependent plan steps on cancellation and returns them to pending on retry", async () => {
|
||||
const { recordDigitalEmployeeGovernanceCommand } = await import("./stateService");
|
||||
|
||||
|
||||
@@ -289,6 +289,40 @@ export interface DigitalEmployeeApprovalActionRecord {
|
||||
payload: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeReadyPlanStepDispatchCandidate {
|
||||
stepId: string;
|
||||
planId: string;
|
||||
taskId: string | null;
|
||||
title: string;
|
||||
taskTitle: string | null;
|
||||
prompt: string;
|
||||
preferredSkillIds: string[];
|
||||
dispatchable: boolean;
|
||||
skipReason?: string | null;
|
||||
step: DigitalEmployeePlanStepRecord;
|
||||
task?: DigitalEmployeeTaskRecord | null;
|
||||
plan?: DigitalEmployeePlanRecord | null;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeePlanStepDispatchInput {
|
||||
stepId: string;
|
||||
planId?: string | null;
|
||||
taskId?: string | null;
|
||||
runId?: string | null;
|
||||
status: "started" | "completed" | "failed" | "skipped";
|
||||
reason?: string | null;
|
||||
message?: string | null;
|
||||
source?: string | null;
|
||||
payload?: Record<string, unknown> | null;
|
||||
occurredAt?: string | null;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeePlanStepDispatchRecord {
|
||||
eventId: string;
|
||||
kind: "plan_step_dispatch_started" | "plan_step_dispatch_completed" | "plan_step_dispatch_failed" | "plan_step_dispatch_skipped";
|
||||
payload: Record<string, unknown>;
|
||||
}
|
||||
|
||||
export interface DigitalEmployeeRouteDecisionRecord {
|
||||
id: string;
|
||||
route_kind: string;
|
||||
@@ -1224,6 +1258,204 @@ export function recordDigitalEmployeeApprovalAction(
|
||||
return { eventId, kind, payload };
|
||||
}
|
||||
|
||||
export function listDigitalEmployeeReadyPlanStepDispatchCandidates(
|
||||
limit = 20,
|
||||
): DigitalEmployeeReadyPlanStepDispatchCandidate[] {
|
||||
const records = readDigitalEmployeeRuntimeRecords();
|
||||
const tasksByStepId = new Map<string, DigitalEmployeeTaskRecord>();
|
||||
for (const task of records.tasks) {
|
||||
const payload = objectRecord(task.payload) ?? {};
|
||||
const stepId = stringValue(payload.stepId ?? payload.step_id);
|
||||
if (stepId && !tasksByStepId.has(stepId)) tasksByStepId.set(stepId, task);
|
||||
}
|
||||
const tasksByPlanId = new Map<string, DigitalEmployeeTaskRecord[]>();
|
||||
for (const task of records.tasks) {
|
||||
const planId = stringValue(task.planId);
|
||||
if (!planId) continue;
|
||||
tasksByPlanId.set(planId, [...(tasksByPlanId.get(planId) ?? []), task]);
|
||||
}
|
||||
const plansById = new Map(records.plans.map((plan) => [plan.id, plan]));
|
||||
const activeRunsByTaskId = new Set(records.runs
|
||||
.filter((run) => isActiveRunStatus(run.status))
|
||||
.map((run) => stringValue(run.taskId))
|
||||
.filter(Boolean));
|
||||
const openApprovals = records.approvals.filter((approval) => isOpenApprovalStatus(approval.status));
|
||||
|
||||
return records.planSteps
|
||||
.filter((step) => normalizePlanStepStatus(step.status) === "ready")
|
||||
.sort((left, right) => 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;
|
||||
const plan = plansById.get(step.planId) ?? null;
|
||||
const taskPayload = objectRecord(task?.payload) ?? {};
|
||||
const stepPayload = objectRecord(step.payload) ?? {};
|
||||
const preferredSkillIds = normalizeSkillIds([
|
||||
...step.preferredSkillIds,
|
||||
...unknownArray(taskPayload.skills),
|
||||
stringValue(task?.assignedSkillId),
|
||||
]);
|
||||
const approvalOpen = openApprovals.some((approval) => {
|
||||
const payload = objectRecord(approval.payload) ?? {};
|
||||
return approval.planId === step.planId
|
||||
|| (task && approval.taskId === task.id)
|
||||
|| stringValue(payload.stepId ?? payload.step_id) === step.id;
|
||||
});
|
||||
const skipReason = readyPlanStepSkipReason({ step, task, activeRunsByTaskId, approvalOpen });
|
||||
const title = step.title || task?.title || step.id;
|
||||
const prompt = stringValue(stepPayload.prompt ?? stepPayload.description)
|
||||
|| stringValue(taskPayload.prompt ?? taskPayload.description)
|
||||
|| `执行计划步骤：${title}`;
|
||||
return {
|
||||
stepId: step.id,
|
||||
planId: step.planId,
|
||||
taskId: task?.id ?? null,
|
||||
title,
|
||||
taskTitle: task?.title ?? null,
|
||||
prompt,
|
||||
preferredSkillIds,
|
||||
dispatchable: !skipReason,
|
||||
skipReason,
|
||||
step,
|
||||
task,
|
||||
plan,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
export function recordDigitalEmployeePlanStepDispatch(
|
||||
input: DigitalEmployeePlanStepDispatchInput,
|
||||
): DigitalEmployeePlanStepDispatchRecord | null {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return null;
|
||||
const now = input.occurredAt || new Date().toISOString();
|
||||
const stepId = stringValue(input.stepId) || "unknown-step";
|
||||
const planId = stringValue(input.planId) || null;
|
||||
const taskId = stringValue(input.taskId) || null;
|
||||
const runId = stringValue(input.runId) || null;
|
||||
const status = input.status;
|
||||
const kind = `plan_step_dispatch_${status}` as DigitalEmployeePlanStepDispatchRecord["kind"];
|
||||
const payload: Record<string, unknown> = {
|
||||
source: stringValue(input.source) || "qimingclaw-ready-step-dispatcher",
|
||||
step_id: stepId,
|
||||
plan_id: planId,
|
||||
task_id: taskId,
|
||||
run_id: runId,
|
||||
status,
|
||||
reason: stringValue(input.reason) || null,
|
||||
...(input.payload ?? {}),
|
||||
};
|
||||
const eventId = `plan-step-dispatch:${safeId(stepId)}:${status}:${safeId(now)}`;
|
||||
const tx = db.transaction(() => {
|
||||
if (status === "started") {
|
||||
updatePlanStepDispatchStatus(stepId, "running", payload, now);
|
||||
} else if (status === "failed") {
|
||||
updatePlanStepDispatchStatus(stepId, "blocked", payload, now);
|
||||
}
|
||||
insertEvent(
|
||||
{
|
||||
event_id: eventId,
|
||||
kind,
|
||||
occurred_at: now,
|
||||
message: input.message || planStepDispatchMessage(status, stepId, stringValue(input.reason)),
|
||||
plan_id: planId,
|
||||
task_id: taskId,
|
||||
},
|
||||
runId,
|
||||
payload,
|
||||
);
|
||||
});
|
||||
tx();
|
||||
return { eventId, kind, payload };
|
||||
}
|
||||
|
||||
function firstRunnableTaskForStep(
|
||||
step: DigitalEmployeePlanStepRecord,
|
||||
tasks: DigitalEmployeeTaskRecord[],
|
||||
): DigitalEmployeeTaskRecord | null {
|
||||
const open = tasks.filter((task) => !isTerminalTaskStatus(task.status));
|
||||
return open.find((task) => {
|
||||
const payload = objectRecord(task.payload) ?? {};
|
||||
const dependsOn = normalizeSkillIds(unknownArray(payload.dependsOn ?? payload.depends_on));
|
||||
return dependsOn.includes(step.id);
|
||||
}) ?? open[0] ?? null;
|
||||
}
|
||||
|
||||
function readyPlanStepSkipReason(input: {
|
||||
step: DigitalEmployeePlanStepRecord;
|
||||
task: DigitalEmployeeTaskRecord | null;
|
||||
activeRunsByTaskId: Set<string>;
|
||||
approvalOpen: boolean;
|
||||
}): string | null {
|
||||
if (!input.task) return "missing_task";
|
||||
if (isTerminalTaskStatus(input.task.status)) return "task_terminal";
|
||||
if (input.activeRunsByTaskId.has(input.task.id)) return "task_already_running";
|
||||
if (input.approvalOpen) return "approval_pending";
|
||||
const payload = objectRecord(input.step.payload) ?? {};
|
||||
if (payload.auto_dispatch === false || payload.autoDispatch === false) return "auto_dispatch_disabled";
|
||||
return null;
|
||||
}
|
||||
|
||||
function isActiveRunStatus(status: string): boolean {
|
||||
return ["running", "pending", "queued", "started"].includes(stringValue(status).toLowerCase());
|
||||
}
|
||||
|
||||
function isTerminalTaskStatus(status: string): boolean {
|
||||
return ["completed", "succeeded", "success", "done", "skipped", "cancelled", "canceled"].includes(stringValue(status).toLowerCase());
|
||||
}
|
||||
|
||||
function isOpenApprovalStatus(status: string): boolean {
|
||||
return ["pending", "reviewing", "open", "waiting"].includes(stringValue(status).toLowerCase() || "pending");
|
||||
}
|
||||
|
||||
function updatePlanStepDispatchStatus(
|
||||
stepId: string,
|
||||
status: string,
|
||||
dispatchPayload: Record<string, unknown>,
|
||||
now: string,
|
||||
): void {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return;
|
||||
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;
|
||||
const previousPayload = objectRecord(parseStoredPayload(row.payload)) ?? {};
|
||||
const payload = {
|
||||
...previousPayload,
|
||||
lastDispatch: {
|
||||
status,
|
||||
occurredAt: now,
|
||||
payload: dispatchPayload,
|
||||
},
|
||||
};
|
||||
db.prepare(`
|
||||
UPDATE digital_plan_steps
|
||||
SET status = ?, payload = ?, sync_status = 'pending', updated_at = ?
|
||||
WHERE id = ?
|
||||
`).run(status, stringify(payload), now, stepId);
|
||||
markOutbox("plan_step", stepId, "upsert", {
|
||||
id: stepId,
|
||||
planId: stringValue(row.plan_id),
|
||||
title: stringValue(row.title),
|
||||
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 planStepDispatchMessage(status: string, stepId: string, reason: string): string {
|
||||
if (status === "started") return `计划步骤已自动派发：${stepId}`;
|
||||
if (status === "completed") return `计划步骤派发已触发执行：${stepId}`;
|
||||
if (status === "failed") return `计划步骤自动派发失败：${stepId}${reason ? `（${reason}）` : ""}`;
|
||||
return `计划步骤自动派发已跳过：${stepId}${reason ? `（${reason}）` : ""}`;
|
||||
}
|
||||
|
||||
function isDigitalEmployeeApprovalAction(action: string): action is DigitalEmployeeApprovalActionKind {
|
||||
return ["comment", "delegate", "mark_timeout", "mark_risk", "clear_risk"].includes(action);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user