吸收数字员工步骤依赖调度
This commit is contained in:
baiyanyun
@@ -1921,7 +1921,7 @@ function shouldUpdateTaskForGovernance(commandKind: string): boolean {
|
||||
}
|
||||
|
||||
function shouldUpdateStepForGovernance(commandKind: string): boolean {
|
||||
return ["retry_task", "skip_task", "pause_task", "resume_task", "cancel_task", "start_run"].includes(commandKind);
|
||||
return ["retry_task", "skip_task", "pause_task", "resume_task", "cancel_task", "start_run", "cancel_run"].includes(commandKind);
|
||||
}
|
||||
|
||||
function mergeGovernancePayload(
|
||||
@@ -2003,6 +2003,195 @@ function updatePlanStepForGovernance(input: {
|
||||
payload,
|
||||
now: input.now,
|
||||
}, input.now);
|
||||
advancePlanStepDependencyGraph({
|
||||
planId: stringValue(row.plan_id),
|
||||
changedStepId: input.stepId,
|
||||
now: input.now,
|
||||
});
|
||||
}
|
||||
|
||||
function advancePlanStepDependencyGraph(input: {
|
||||
planId: string;
|
||||
changedStepId: string;
|
||||
now: string;
|
||||
}): void {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db || !input.planId || !input.changedStepId) return;
|
||||
const rows = db.prepare(`
|
||||
SELECT id, plan_id, title, status, seq, depends_on, preferred_skill_ids, payload, created_at
|
||||
FROM digital_plan_steps
|
||||
WHERE plan_id = ?
|
||||
ORDER BY seq ASC, created_at ASC
|
||||
`).all(input.planId) as Array<Record<string, unknown>>;
|
||||
if (rows.length === 0) return;
|
||||
|
||||
const stepsById = new Map(rows.map((row) => [stringValue(row.id), row]));
|
||||
const queue = [input.changedStepId];
|
||||
const visited = new Set<string>();
|
||||
|
||||
while (queue.length > 0) {
|
||||
const sourceStepId = queue.shift() || "";
|
||||
if (!sourceStepId || visited.has(sourceStepId)) continue;
|
||||
visited.add(sourceStepId);
|
||||
for (const row of rows) {
|
||||
const stepId = stringValue(row.id);
|
||||
const dependsOn = parseStringArray(row.depends_on);
|
||||
if (!stepId || !dependsOn.includes(sourceStepId)) continue;
|
||||
const currentStatus = normalizePlanStepStatus(row.status);
|
||||
if (!canAdvanceDependentPlanStep(currentStatus)) continue;
|
||||
const nextState = evaluatePlanStepDependencyState(row, stepsById, input.now, input.changedStepId);
|
||||
if (!nextState || nextState.status === currentStatus) continue;
|
||||
updatePlanStepForDependencyState({
|
||||
row,
|
||||
dependsOn,
|
||||
status: nextState.status,
|
||||
eventKind: nextState.eventKind,
|
||||
message: nextState.message,
|
||||
dependencyGraph: nextState.dependencyGraph,
|
||||
now: input.now,
|
||||
});
|
||||
row.status = nextState.status;
|
||||
row.payload = stringify({
|
||||
...(objectRecord(parseStoredPayload(row.payload)) ?? {}),
|
||||
dependencyGraph: nextState.dependencyGraph,
|
||||
});
|
||||
queue.push(stepId);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function evaluatePlanStepDependencyState(
|
||||
row: Record<string, unknown>,
|
||||
stepsById: Map<string, Record<string, unknown>>,
|
||||
now: string,
|
||||
triggeredByStepId: string,
|
||||
): {
|
||||
status: "ready" | "blocked" | "pending";
|
||||
eventKind: "plan_step_ready" | "plan_step_blocked" | "plan_step_waiting";
|
||||
message: string;
|
||||
dependencyGraph: Record<string, unknown>;
|
||||
} | null {
|
||||
const stepId = stringValue(row.id);
|
||||
const dependsOn = parseStringArray(row.depends_on);
|
||||
if (!stepId || dependsOn.length === 0) return null;
|
||||
const dependencyStates = dependsOn.map((dependencyId) => {
|
||||
const dependency = stepsById.get(dependencyId);
|
||||
return {
|
||||
id: dependencyId,
|
||||
status: normalizePlanStepStatus(dependency?.status),
|
||||
exists: Boolean(dependency),
|
||||
};
|
||||
});
|
||||
const blockedDependencies = dependencyStates
|
||||
.filter((dependency) => isBlockingPlanStepStatus(dependency.status))
|
||||
.map((dependency) => dependency.id);
|
||||
const waitingDependencies = dependencyStates
|
||||
.filter((dependency) => !dependency.exists || !isSatisfiedPlanStepStatus(dependency.status))
|
||||
.map((dependency) => dependency.id);
|
||||
const satisfiedDependencies = dependencyStates
|
||||
.filter((dependency) => isSatisfiedPlanStepStatus(dependency.status))
|
||||
.map((dependency) => dependency.id);
|
||||
|
||||
const dependencyGraph = {
|
||||
evaluatedAt: now,
|
||||
triggeredByStepId,
|
||||
dependencies: dependencyStates,
|
||||
satisfiedDependencies,
|
||||
waitingDependencies,
|
||||
blockedDependencies,
|
||||
};
|
||||
|
||||
if (blockedDependencies.length > 0) {
|
||||
return {
|
||||
status: "blocked",
|
||||
eventKind: "plan_step_blocked",
|
||||
message: `计划步骤“${stringValue(row.title) || stepId}”因依赖阻塞，等待人工处理。`,
|
||||
dependencyGraph: { ...dependencyGraph, status: "blocked" },
|
||||
};
|
||||
}
|
||||
if (waitingDependencies.length === 0) {
|
||||
return {
|
||||
status: "ready",
|
||||
eventKind: "plan_step_ready",
|
||||
message: `计划步骤“${stringValue(row.title) || stepId}”依赖已满足，等待启动。`,
|
||||
dependencyGraph: { ...dependencyGraph, status: "ready" },
|
||||
};
|
||||
}
|
||||
return {
|
||||
status: "pending",
|
||||
eventKind: "plan_step_waiting",
|
||||
message: `计划步骤“${stringValue(row.title) || stepId}”重新等待依赖完成。`,
|
||||
dependencyGraph: { ...dependencyGraph, status: "pending" },
|
||||
};
|
||||
}
|
||||
|
||||
function updatePlanStepForDependencyState(input: {
|
||||
row: Record<string, unknown>;
|
||||
dependsOn: string[];
|
||||
status: "ready" | "blocked" | "pending";
|
||||
eventKind: "plan_step_ready" | "plan_step_blocked" | "plan_step_waiting";
|
||||
message: string;
|
||||
dependencyGraph: Record<string, unknown>;
|
||||
now: string;
|
||||
}): void {
|
||||
const db = getDigitalEmployeeDb();
|
||||
if (!db) return;
|
||||
const stepId = stringValue(input.row.id);
|
||||
const planId = stringValue(input.row.plan_id);
|
||||
const payload = {
|
||||
...(objectRecord(parseStoredPayload(input.row.payload)) ?? {}),
|
||||
dependencyGraph: input.dependencyGraph,
|
||||
};
|
||||
db.prepare(`
|
||||
UPDATE digital_plan_steps
|
||||
SET status = ?, payload = ?, sync_status = 'pending', updated_at = ?
|
||||
WHERE id = ?
|
||||
`).run(input.status, stringify(payload), input.now, stepId);
|
||||
markOutbox("plan_step", stepId, "upsert", {
|
||||
id: stepId,
|
||||
planId,
|
||||
title: stringValue(input.row.title),
|
||||
status: input.status,
|
||||
seq: typeof input.row.seq === "number" ? input.row.seq : Number(input.row.seq) || 0,
|
||||
dependsOn: input.dependsOn,
|
||||
preferredSkillIds: parseStringArray(input.row.preferred_skill_ids),
|
||||
payload,
|
||||
now: input.now,
|
||||
}, input.now);
|
||||
insertEvent(
|
||||
{
|
||||
event_id: `${stepId}:dependency:${input.status}:${safeId(stringValue(input.dependencyGraph.triggeredByStepId))}:${safeId(input.now)}`,
|
||||
kind: input.eventKind,
|
||||
occurred_at: input.now,
|
||||
message: input.message,
|
||||
plan_id: planId,
|
||||
task_id: null,
|
||||
},
|
||||
null,
|
||||
{
|
||||
source: "qimingclaw-plan-step-dependency",
|
||||
stepId,
|
||||
planId,
|
||||
status: input.status,
|
||||
dependencyGraph: input.dependencyGraph,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
function canAdvanceDependentPlanStep(status: string): boolean {
|
||||
return ["pending", "ready", "blocked"].includes(status);
|
||||
}
|
||||
|
||||
function isSatisfiedPlanStepStatus(status: string): boolean {
|
||||
return ["completed", "succeeded", "success", "done", "skipped"].includes(status);
|
||||
}
|
||||
|
||||
function isBlockingPlanStepStatus(status: string): boolean {
|
||||
return ["failed", "error", "cancelled", "canceled", "rejected", "blocked"].includes(status);
|
||||
}
|
||||
|
||||
function normalizePlanStepStatus(status: unknown): string {
|
||||
return stringValue(status).toLowerCase() || "pending";
|
||||
}
|
||||
|
||||
function persistTaskGovernanceRun(input: {
|
||||
@@ -3247,7 +3436,7 @@ function upsertApproval(input: {
|
||||
|
||||
function insertEvent(
|
||||
event: DigitalEmployeeStateEvent,
|
||||
runId: string,
|
||||
runId: string | null,
|
||||
payload: unknown,
|
||||
): void {
|
||||
const db = getDigitalEmployeeDb();
|
||||
|
||||
Reference in New Issue
Block a user