吸收数字员工步骤治理闭环

This commit is contained in:
baiyanyun
2026-06-09 11:31:29 +08:00
parent 5ebc0a640b
commit ef8252e2dd
9 changed files with 1061 additions and 154 deletions

View File

@@ -305,6 +305,9 @@ export interface DigitalEmployeeGovernanceCommandRecord {
commandId: string;
correlationId?: string;
planId?: string | null;
taskId?: string | null;
runId?: string | null;
stepId?: string | null;
scheduleId?: string | null;
accepted?: boolean;
message?: string;
@@ -1155,6 +1158,14 @@ export function recordDigitalEmployeeGovernanceCommand(
const commandId = command.commandId || `governance-${Date.now()}`;
const result = objectRecord(command.result) ?? {};
const payload = objectRecord(command.payload) ?? {};
if (isTaskGovernanceCommand(command.commandKind)) {
return applyDigitalEmployeeTaskGovernanceCommand({
...command,
commandId,
payload,
result,
});
}
const planId = stringValue(command.planId)
|| stringValue(result.plan_id)
|| stringValue(payload.plan_id)
@@ -1267,6 +1278,152 @@ export function recordDigitalEmployeeGovernanceCommand(
return ids;
}
export function applyDigitalEmployeeTaskGovernanceCommand(
command: DigitalEmployeeGovernanceCommandRecord,
): { planId: string; taskId: string; runId: string; stepId?: string | null } {
const now = new Date().toISOString();
const commandId = command.commandId || `governance-${Date.now()}`;
const result = objectRecord(command.result) ?? {};
const payload = objectRecord(command.payload) ?? {};
const db = getDigitalEmployeeDb();
const requestedRunId = stringValue(command.runId)
|| stringValue(result.run_id)
|| stringValue(result.latest_run_id)
|| stringValue(payload.run_id)
|| stringValue(payload.latest_run_id);
const requestedTaskId = stringValue(command.taskId)
|| stringValue(result.task_id)
|| stringValue(payload.task_id)
|| stringValue(payload.original_task_id);
const requestedPlanId = stringValue(command.planId)
|| stringValue(result.plan_id)
|| stringValue(payload.plan_id)
|| stringValue(payload.original_plan_id);
const requestedStepId = stringValue(command.stepId)
|| stringValue(result.step_id)
|| stringValue(payload.step_id)
|| stringValue(payload.stepId);
if (!db) {
const fallbackTaskId = requestedTaskId || `governance-task-${safeId(commandId)}`;
return {
planId: requestedPlanId || "qimingclaw-task-agent-plan",
taskId: fallbackTaskId,
runId: requestedRunId || taskGovernanceRunId(command.commandKind, fallbackTaskId, commandId),
stepId: requestedStepId || null,
};
}
const requestedRun = requestedRunId ? readDigitalEmployeeRunRow(requestedRunId) : null;
const existingTaskId = requestedTaskId || stringValue(requestedRun?.task_id);
const existingTask = existingTaskId ? readDigitalEmployeeTaskRow(existingTaskId) : null;
const latestRun = existingTaskId ? readLatestDigitalEmployeeRunForTask(existingTaskId) : null;
const targetRun = requestedRun || latestRun;
const taskId = existingTaskId || `governance-task-${safeId(commandId)}`;
const planId = requestedPlanId
|| stringValue(existingTask?.plan_id)
|| stringValue(targetRun?.plan_id)
|| "qimingclaw-task-agent-plan";
const existingTaskPayload = objectRecord(parseStoredPayload(existingTask?.payload)) ?? {};
const stepId = requestedStepId
|| stringValue(existingTaskPayload.stepId ?? existingTaskPayload.step_id)
|| findPlanStepIdForTask(planId, taskId);
const runId = taskGovernanceTargetRunId(command.commandKind, taskId, commandId, requestedRunId, targetRun);
const status = taskGovernanceStatusForCommand(command.commandKind, existingTask);
const nextAction = taskGovernanceNextAction(command.commandKind);
const payloadEnvelope = {
source: "qimingclaw-task-governance",
commandKind: command.commandKind,
commandId,
correlationId: command.correlationId,
planId,
taskId,
stepId: stepId || null,
runId,
sourceRunId: stringValue(targetRun?.id) || null,
accepted: command.accepted !== false,
message: command.message,
error: command.error,
status,
nextAction,
payload,
result,
};
const tx = db.transaction(() => {
if (!existingTask) {
upsertPlan({
id: planId,
title: stringValue(payload.plan_title ?? result.plan_title) || "Task Agent 控制平面",
objective: "记录 qimingclaw 数字员工任务级治理命令。",
status: "running",
payload: payloadEnvelope,
now,
});
upsertTask({
id: taskId,
planId,
title: stringValue(payload.title ?? payload.task_title ?? result.title) || governanceCommandTitle(command.commandKind),
status: taskStatusForCommand(command.commandKind, "pending"),
assignedSkillId: stringValue(payload.assigned_skill_id ?? result.assigned_skill_id) || "qimingclaw-task-agent",
payload: mergeGovernancePayload({}, payloadEnvelope, null),
now,
});
} else if (shouldUpdateTaskForGovernance(command.commandKind)) {
updateTaskForGovernance({
task: existingTask,
status: taskStatusForCommand(command.commandKind, stringValue(existingTask.status)),
payload: mergeGovernancePayload(existingTaskPayload, payloadEnvelope, stringValue(existingTask.status)),
now,
});
} else if (command.commandKind === "provide_task_input") {
updateTaskForGovernance({
task: existingTask,
status: stringValue(existingTask.status) || "pending",
payload: mergeGovernancePayload(existingTaskPayload, payloadEnvelope, stringValue(existingTask.status)),
now,
});
}
if (stepId && shouldUpdateStepForGovernance(command.commandKind)) {
updatePlanStepForGovernance({
stepId,
status: stepStatusForCommand(command.commandKind),
payloadEnvelope,
now,
});
}
persistTaskGovernanceRun({
commandKind: command.commandKind,
planId,
taskId,
runId,
targetRun,
payloadEnvelope,
now,
});
persistPayloadArtifactsAndApprovals({ planId, taskId, runId }, payloadEnvelope, now);
insertEvent(
{
event_id: `${runId}:${command.commandKind}:${safeId(commandId)}`,
kind: `governance_${command.commandKind}`,
occurred_at: now,
message: command.message || `${governanceCommandTitle(command.commandKind)} 已记录`,
plan_id: planId,
task_id: taskId,
},
runId,
payloadEnvelope,
);
});
tx();
return { planId, taskId, runId, stepId: stepId || null };
}
function governanceCommandTitle(commandKind: string): string {
switch (commandKind) {
case "create_plan":
@@ -1293,6 +1450,22 @@ function governanceCommandTitle(commandKind: string): string {
return "暂停数字员工计划";
case "cancel_plan":
return "取消数字员工计划";
case "retry_task":
return "重试数字员工任务";
case "skip_task":
return "跳过数字员工任务";
case "pause_task":
return "暂停数字员工任务";
case "resume_task":
return "恢复数字员工任务";
case "provide_task_input":
return "补充数字员工任务输入";
case "start_run":
return "启动数字员工任务运行";
case "cancel_task":
return "取消数字员工任务";
case "cancel_run":
return "取消数字员工运行";
default:
return `数字员工治理命令:${commandKind || "unknown"}`;
}
@@ -1452,6 +1625,375 @@ function persistScheduleGovernanceCommand(input: {
});
}
const TASK_GOVERNANCE_COMMANDS = new Set([
"retry_task",
"skip_task",
"pause_task",
"resume_task",
"provide_task_input",
"cancel_task",
"start_run",
"cancel_run",
]);
function isTaskGovernanceCommand(commandKind: string): boolean {
return TASK_GOVERNANCE_COMMANDS.has(commandKind);
}
function readDigitalEmployeeTaskRow(taskId: string): Record<string, unknown> | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, plan_id, title, status, assigned_skill_id, payload,
sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_tasks
WHERE id = ?
`).get(taskId) as Record<string, unknown> | undefined;
return row ?? null;
}
function readDigitalEmployeeRunRow(runId: string): Record<string, unknown> | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, plan_id, task_id, status, started_at, finished_at,
payload, sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_runs
WHERE id = ?
`).get(runId) as Record<string, unknown> | undefined;
return row ?? null;
}
function readLatestDigitalEmployeeRunForTask(taskId: string): Record<string, unknown> | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, plan_id, task_id, status, started_at, finished_at,
payload, sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_runs
WHERE task_id = ?
ORDER BY updated_at DESC, created_at DESC
LIMIT 1
`).get(taskId) as Record<string, unknown> | undefined;
return row ?? null;
}
function findPlanStepIdForTask(planId: string, taskId: string): string | null {
const task = readDigitalEmployeeTaskRow(taskId);
const taskPayload = objectRecord(parseStoredPayload(task?.payload)) ?? {};
const explicitStepId = stringValue(taskPayload.stepId ?? taskPayload.step_id);
if (explicitStepId) return explicitStepId;
const db = getDigitalEmployeeDb();
if (!db || !planId) return null;
const row = db.prepare(`
SELECT id
FROM digital_plan_steps
WHERE plan_id = ?
ORDER BY seq ASC, created_at ASC
LIMIT 1
`).get(planId) as Record<string, unknown> | undefined;
return stringValue(row?.id) || null;
}
function taskGovernanceRunId(commandKind: string, taskId: string, commandId: string): string {
return `${taskId}:run:${commandKind}:${safeId(commandId)}`;
}
function taskGovernanceTargetRunId(
commandKind: string,
taskId: string,
commandId: string,
requestedRunId: string,
targetRun: Record<string, unknown> | null,
): string {
if (commandKind === "retry_task" || commandKind === "start_run") {
return taskGovernanceRunId(commandKind, taskId, commandId);
}
return requestedRunId || stringValue(targetRun?.id) || taskGovernanceRunId(commandKind, taskId, commandId);
}
function taskGovernanceStatusForCommand(
commandKind: string,
existingTask: Record<string, unknown> | null,
): string {
if (commandKind === "cancel_run" || commandKind === "provide_task_input") {
return stringValue(existingTask?.status) || "accepted";
}
return taskStatusForCommand(commandKind, stringValue(existingTask?.status) || "pending");
}
function taskStatusForCommand(commandKind: string, previousStatus: string): string {
switch (commandKind) {
case "retry_task":
case "start_run":
case "resume_task":
return "running";
case "skip_task":
return "skipped";
case "cancel_task":
return "cancelled";
case "pause_task":
return "paused";
default:
return previousStatus || "pending";
}
}
function stepStatusForCommand(commandKind: string): string {
switch (commandKind) {
case "retry_task":
case "start_run":
case "resume_task":
return "running";
case "skip_task":
return "skipped";
case "cancel_task":
return "cancelled";
case "pause_task":
return "paused";
default:
return "pending";
}
}
function taskGovernanceNextAction(commandKind: string): string {
switch (commandKind) {
case "retry_task":
return "watch_retry_run";
case "skip_task":
return "continue_next_task";
case "pause_task":
return "await_resume";
case "resume_task":
case "start_run":
return "watch_progress";
case "provide_task_input":
return "resume_with_input";
case "cancel_task":
case "cancel_run":
return "close_execution";
default:
return "record_control_event";
}
}
function shouldUpdateTaskForGovernance(commandKind: string): boolean {
return ["retry_task", "skip_task", "pause_task", "resume_task", "cancel_task", "start_run"].includes(commandKind);
}
function shouldUpdateStepForGovernance(commandKind: string): boolean {
return ["retry_task", "skip_task", "pause_task", "resume_task", "cancel_task", "start_run"].includes(commandKind);
}
function mergeGovernancePayload(
existingPayload: Record<string, unknown>,
payloadEnvelope: Record<string, unknown>,
previousStatus: string | null,
): Record<string, unknown> {
const governanceHistory = unknownArray(existingPayload.governanceHistory).slice(-19);
return {
...existingPayload,
lastGovernanceCommand: payloadEnvelope,
governanceHistory: [
...governanceHistory,
{
commandKind: payloadEnvelope.commandKind,
commandId: payloadEnvelope.commandId,
previousStatus,
status: payloadEnvelope.status,
occurredAt: new Date().toISOString(),
},
],
};
}
function updateTaskForGovernance(input: {
task: Record<string, unknown>;
status: string;
payload: Record<string, unknown>;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
const taskId = stringValue(input.task.id);
db.prepare(`
UPDATE digital_tasks
SET status = ?, payload = ?, sync_status = 'pending', updated_at = ?
WHERE id = ?
`).run(input.status, stringify(input.payload), input.now, taskId);
markOutbox("task", taskId, "upsert", {
id: taskId,
planId: stringValue(input.task.plan_id),
title: stringValue(input.task.title),
status: input.status,
assignedSkillId: stringValue(input.task.assigned_skill_id),
payload: input.payload,
now: input.now,
}, input.now);
}
function updatePlanStepForGovernance(input: {
stepId: string;
status: string;
payloadEnvelope: 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(input.stepId) as Record<string, unknown> | undefined;
if (!row) return;
const previousPayload = objectRecord(parseStoredPayload(row.payload)) ?? {};
const payload = mergeGovernancePayload(previousPayload, input.payloadEnvelope, stringValue(row.status));
db.prepare(`
UPDATE digital_plan_steps
SET status = ?, payload = ?, sync_status = 'pending', updated_at = ?
WHERE id = ?
`).run(input.status, stringify(payload), input.now, input.stepId);
markOutbox("plan_step", input.stepId, "upsert", {
id: input.stepId,
planId: stringValue(row.plan_id),
title: stringValue(row.title),
status: input.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: input.now,
}, input.now);
}
function persistTaskGovernanceRun(input: {
commandKind: string;
planId: string;
taskId: string;
runId: string;
targetRun: Record<string, unknown> | null;
payloadEnvelope: Record<string, unknown>;
now: string;
}): void {
if (input.commandKind === "retry_task" || input.commandKind === "start_run") {
const attemptNo = countDigitalEmployeeRunsForTask(input.taskId) + 1;
upsertRun({
id: input.runId,
planId: input.planId,
taskId: input.taskId,
status: "running",
payload: {
...input.payloadEnvelope,
attemptNo,
},
now: input.now,
});
return;
}
const existingRun = input.targetRun || readDigitalEmployeeRunRow(input.runId);
const previousPayload = objectRecord(parseStoredPayload(existingRun?.payload)) ?? {};
const status = runStatusForTaskGovernance(input.commandKind, stringValue(existingRun?.status) || "pending");
const payload = mergeGovernancePayload(previousPayload, input.payloadEnvelope, stringValue(existingRun?.status) || null);
if (existingRun) {
updateRunForGovernance({
run: existingRun,
status,
payload,
finishedAt: terminalRunStatus(status) ? input.now : undefined,
now: input.now,
});
return;
}
upsertRun({
id: input.runId,
planId: input.planId,
taskId: input.taskId,
status,
payload,
now: input.now,
});
if (terminalRunStatus(status)) finishRunWithOutbox(input.runId, input.planId, input.taskId, status, payload, input.now);
}
function runStatusForTaskGovernance(commandKind: string, previousStatus: string): string {
switch (commandKind) {
case "skip_task":
return "skipped";
case "cancel_task":
case "cancel_run":
return "cancelled";
case "pause_task":
return "paused";
case "resume_task":
return "running";
default:
return previousStatus || "pending";
}
}
function terminalRunStatus(status: string): boolean {
return ["completed", "succeeded", "failed", "cancelled", "skipped"].includes(status);
}
function updateRunForGovernance(input: {
run: Record<string, unknown>;
status: string;
payload: Record<string, unknown>;
finishedAt?: string;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
const runId = stringValue(input.run.id);
if (input.finishedAt) {
db.prepare(`
UPDATE digital_runs
SET status = ?, payload = ?, finished_at = ?, sync_status = 'pending', updated_at = ?
WHERE id = ?
`).run(input.status, stringify(input.payload), input.finishedAt, input.now, runId);
} else {
db.prepare(`
UPDATE digital_runs
SET status = ?, payload = ?, sync_status = 'pending', updated_at = ?
WHERE id = ?
`).run(input.status, stringify(input.payload), input.now, runId);
}
markOutbox("run", runId, "upsert", {
id: runId,
planId: stringValue(input.run.plan_id),
taskId: stringValue(input.run.task_id),
status: input.status,
payload: input.payload,
finishedAt: input.finishedAt ?? optionalStringField(input.run, "finished_at"),
now: input.now,
}, input.now);
}
function finishRunWithOutbox(
runId: string,
planId: string,
taskId: string,
status: string,
payload: Record<string, unknown>,
now: string,
): void {
finishRun(runId, status, now);
markOutbox("run", runId, "upsert", { id: runId, planId, taskId, status, payload, finishedAt: now }, now);
}
function countDigitalEmployeeRunsForTask(taskId: string): number {
const db = getDigitalEmployeeDb();
if (!db) return 0;
const row = db.prepare(`
SELECT COUNT(*) AS count
FROM digital_runs
WHERE task_id = ?
`).get(taskId) as { count?: number } | undefined;
return Number(row?.count) || 0;
}
function extractPlanStepInputs(
payload: Record<string, unknown>,
planId: string,