Files
qiming/qimingclaw/crates/agent-electron-client/src/main/services/digitalEmployee/stateService.ts
2026-06-09 23:10:01 +08:00

4188 lines
132 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import { ensureDigitalEmployeeSchema, getDb, readSetting, writeSetting } from "../../db";
const STATE_KEY = "digital_employee_state_v1";
const UI_STATE_KEY = "digital_employee_ui_state_v1";
const MAX_EVENTS = 200;
export interface DigitalEmployeeServiceStatus {
running?: boolean;
error?: string;
engineType?: string | null;
port?: number;
serverCount?: number;
serverNames?: string[];
pid?: number;
}
export interface DigitalEmployeeSessionSummary {
total: number;
active: number;
items: Array<{
id: string;
title?: string;
status?: string;
engineType?: string;
projectId?: string;
}>;
}
export interface DigitalEmployeeManagedService {
serviceId: string;
name: string;
kind: "agent" | "mcp" | "desktop" | "network" | "file" | "gui" | "unknown";
status: "running" | "stopped" | "degraded";
statusLabel: string;
running: boolean;
error?: string | null;
requiresHumanAction: boolean;
dependentTasks: string[];
restartCount: number;
lastSeenAt: string;
instance: {
pid?: number;
port?: number;
engineType?: string | null;
serverCount?: number;
serverNames?: string[];
};
lease: {
state: "active" | "stopped" | "degraded";
updatedAt: string;
};
}
export interface DigitalEmployeeSnapshot {
generatedAt: string;
services: Record<string, DigitalEmployeeServiceStatus | null>;
managedServices?: DigitalEmployeeManagedService[];
sessions: DigitalEmployeeSessionSummary;
}
export interface DigitalEmployeePlanRecord {
id: string;
remoteId?: string | null;
title: string;
objective?: string | null;
status: string;
source: string;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
}
export interface DigitalEmployeeTaskRecord {
id: string;
remoteId?: string | null;
planId?: string | null;
title: string;
status: string;
assignedSkillId?: string | null;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
}
export interface DigitalEmployeePlanStepRecord {
id: string;
remoteId?: string | null;
planId: string;
title: string;
status: string;
seq: number;
dependsOn: string[];
preferredSkillIds: string[];
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
}
export interface DigitalEmployeeScheduleRecord {
id: string;
remoteId?: string | null;
planId?: string | null;
name: string;
description?: string | null;
cronExpr: string;
timezone: string;
status: string;
enabled: boolean;
prompt?: string | null;
command?: string | null;
payload: unknown;
nextRunAt?: string | null;
lastRunAt?: string | null;
catchUpOnStartup: boolean;
maxCatchUpRuns: number;
maxRunHistory: number;
failureCount: number;
lastError?: string | null;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
deletedAt?: string | null;
}
export interface DigitalEmployeeScheduleRunRecord {
id: string;
remoteId?: string | null;
scheduleId: string;
planId?: string | null;
runId?: string | null;
dueAt: string;
startedAt?: string | null;
finishedAt?: string | null;
status: string;
attemptNo: number;
triggerKind: string;
error?: string | null;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
}
export interface DigitalEmployeeRunRecord {
id: string;
remoteId?: string | null;
planId?: string | null;
taskId?: string | null;
status: string;
startedAt?: string | null;
finishedAt?: string | null;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
}
export interface DigitalEmployeeEventRecord {
id: string;
remoteId?: string | null;
planId?: string | null;
taskId?: string | null;
runId?: string | null;
kind: string;
message: string;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
occurredAt: string;
createdAt: string;
}
export interface DigitalEmployeeRouteDecisionRecord {
id: string;
route_kind: string;
intent_kind: string;
reason_codes: string[];
candidate_skills: string[];
context_refs: Record<string, unknown>;
created_command_id: string | null;
correlation_id: string;
creates_plan: boolean;
creates_task_run: boolean;
approval_mode: string | null;
policy_result: string | null;
recorded_at: string;
entrypoint?: string | null;
actor?: string | null;
}
export interface DigitalEmployeeRouteDecisionInput {
id?: string | null;
idempotencyKey?: string | null;
correlationId?: string | null;
routeKind: string;
intentKind: string;
reasonCodes?: string[];
candidateSkills?: string[];
contextRefs?: Record<string, unknown>;
createdCommandId?: string | null;
createsPlan?: boolean;
createsTaskRun?: boolean;
approvalMode?: string | null;
policyResult?: string | null;
entrypoint?: string | null;
actor?: string | null;
message?: string | null;
envelope?: Record<string, unknown> | null;
recordedAt?: string | null;
}
export interface DigitalEmployeeArtifactRecord {
id: string;
remoteId?: string | null;
planId?: string | null;
taskId?: string | null;
runId?: string | null;
kind: string;
name?: string | null;
uri?: string | null;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
}
export interface DigitalEmployeeApprovalRecord {
id: string;
remoteId?: string | null;
planId?: string | null;
taskId?: string | null;
runId?: string | null;
title: string;
status: string;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
}
export interface DigitalEmployeeMemoryRecord {
id: string;
remoteId?: string | null;
key: string;
content: string;
category: string;
source: string;
sourcePath?: string | null;
sessionId?: string | null;
score?: number | null;
status: string;
payload: unknown;
syncStatus: string;
lastSyncedAt?: string | null;
syncError?: string | null;
createdAt: string;
updatedAt: string;
deletedAt?: string | null;
}
export interface DigitalEmployeeRuntimeRecords {
generatedAt: string;
plans: DigitalEmployeePlanRecord[];
planSteps: DigitalEmployeePlanStepRecord[];
schedules: DigitalEmployeeScheduleRecord[];
scheduleRuns: DigitalEmployeeScheduleRunRecord[];
tasks: DigitalEmployeeTaskRecord[];
runs: DigitalEmployeeRunRecord[];
events: DigitalEmployeeEventRecord[];
artifacts: DigitalEmployeeArtifactRecord[];
approvals: DigitalEmployeeApprovalRecord[];
memories: DigitalEmployeeMemoryRecord[];
}
export interface DigitalEmployeeMemoryUpsertInput {
id?: string;
key?: string;
content: string;
category?: string;
source?: string;
sourcePath?: string | null;
sessionId?: string | null;
score?: number | null;
status?: string;
payload?: Record<string, unknown>;
createdAt?: string;
updatedAt?: string;
deletedAt?: string | null;
}
export interface DigitalEmployeeQimingMemoryEntry {
id?: string;
text?: string;
content?: string;
key?: string;
category?: string;
source?: string;
sourcePath?: string;
sessionId?: string | null;
confidence?: number;
importance?: number;
status?: string;
createdAt?: number | string;
updatedAt?: number | string;
}
export interface DigitalEmployeeUiState {
selectedDutyIdsByDate: Record<string, string[]>;
confirmedDates: Record<string, string>;
reportScheduleTime: string;
reportGeneratedAtByDate: Record<string, string>;
decisionResultsByDate: Record<string, Record<string, string>>;
consoleRole?: string;
rolePreferences?: Record<string, unknown>;
profile: {
name: string;
company: string;
position: string;
currentStatus: string;
};
}
export interface DigitalEmployeeUiStateUpdate {
state: DigitalEmployeeUiState;
action?: string;
date?: string;
metadata?: Record<string, unknown>;
}
export interface DigitalEmployeeStateEvent {
event_id: string;
kind: string;
occurred_at: string;
message: string;
plan_id: string | null;
task_id: string | null;
}
export interface DigitalEmployeeState {
version: 1;
updatedAt: string | null;
lastSnapshot: DigitalEmployeeSnapshot | null;
events: DigitalEmployeeStateEvent[];
pendingSyncCount?: number;
}
export interface DigitalEmployeeChatRequestRecord {
user_id?: string;
project_id?: string;
prompt?: string;
session_id?: string;
request_id?: string;
model_provider?: { provider?: string; model?: string; default_model?: string };
}
export interface DigitalEmployeeChatResultRecord {
success: boolean;
code?: string;
message?: string;
project_id?: string;
session_id?: string;
request_id?: string;
engine?: string;
}
export interface DigitalEmployeeRuntimeEventRecord {
kind: string;
message?: string;
request_id?: string;
session_id?: string;
project_id?: string;
status?: "running" | "completed" | "failed" | "pending";
payload?: unknown;
}
export interface DigitalEmployeeDailyReportRecordInput {
reportId?: string | null;
date: string;
title: string;
summary: string;
totals: Record<string, unknown>;
detailLines: string[];
artifactSources?: unknown[];
filename: string;
textContent: string;
generatedAt?: string | null;
reportScheduleTime?: string | null;
source?: string | null;
model?: string | null;
}
export interface DigitalEmployeeDailyReportRecordResult {
reportId: string;
planId: string;
taskId: string;
runId: string;
artifactId: string;
eventId: string;
}
export interface DigitalEmployeeManagedServiceActionInput {
serviceId: string;
action: "restart";
status: "success" | "failed" | "rejected";
reason?: string | null;
actor?: string | null;
result?: Record<string, unknown> | null;
allowlisted: boolean;
recordedAt?: string | null;
}
export interface DigitalEmployeeManagedServiceActionResult {
eventId: string;
}
export interface DigitalEmployeeGovernanceCommandRecord {
commandKind: string;
commandId: string;
correlationId?: string;
planId?: string | null;
taskId?: string | null;
runId?: string | null;
stepId?: string | null;
scheduleId?: string | null;
accepted?: boolean;
message?: string;
error?: string | null;
payload?: Record<string, unknown> | null;
result?: Record<string, unknown> | null;
}
export interface DigitalEmployeeCronSettings {
enabled: boolean;
catch_up_on_startup: boolean;
max_run_history: number;
max_catch_up_runs: number;
max_concurrent_schedule_runs: number;
}
export interface DigitalEmployeeScheduleUpsertInput {
id?: string;
planId?: string | null;
name?: string;
description?: string | null;
cronExpr?: string;
timezone?: string;
status?: string;
enabled?: boolean;
prompt?: string | null;
command?: string | null;
payload?: Record<string, unknown>;
nextRunAt?: string | null;
catchUpOnStartup?: boolean;
maxCatchUpRuns?: number;
maxRunHistory?: number;
}
export interface DigitalEmployeeScheduleRunInput {
scheduleId: string;
planId?: string | null;
dueAt: string;
triggerKind: string;
status?: string;
attemptNo?: number;
runId?: string | null;
payload?: Record<string, unknown>;
}
const CRON_SETTINGS_KEY = "digital_employee_cron_settings_v1";
function defaultState(): DigitalEmployeeState {
return {
version: 1,
updatedAt: null,
lastSnapshot: null,
events: [],
};
}
function defaultUiState(): DigitalEmployeeUiState {
return {
selectedDutyIdsByDate: {},
confirmedDates: {},
reportScheduleTime: "18:00",
reportGeneratedAtByDate: {},
decisionResultsByDate: {},
consoleRole: "sales",
rolePreferences: {},
profile: {
name: "飞天数字员工",
company: "qimingclaw 客户端",
position: "客户端任务执行员",
currentStatus: "working",
},
};
}
function getDigitalEmployeeDb(): ReturnType<typeof getDb> {
if (!ensureDigitalEmployeeSchema()) return null;
return getDb();
}
function normalizeStringMap(value: unknown): Record<string, string> {
if (!value || typeof value !== "object" || Array.isArray(value)) return {};
return Object.fromEntries(
Object.entries(value as Record<string, unknown>)
.filter((entry): entry is [string, string] =>
typeof entry[0] === "string" && typeof entry[1] === "string",
),
);
}
function normalizeStringArrayMap(value: unknown): Record<string, string[]> {
if (!value || typeof value !== "object" || Array.isArray(value)) return {};
return Object.fromEntries(
Object.entries(value as Record<string, unknown>).map(([key, list]) => [
key,
Array.isArray(list)
? list.filter((item): item is string => typeof item === "string")
: [],
]),
);
}
function normalizeNestedStringMap(value: unknown): Record<string, Record<string, string>> {
if (!value || typeof value !== "object" || Array.isArray(value)) return {};
return Object.fromEntries(
Object.entries(value as Record<string, unknown>).map(([key, map]) => [
key,
normalizeStringMap(map),
]),
);
}
function normalizeRecord(value: unknown): Record<string, unknown> {
if (!value || typeof value !== "object" || Array.isArray(value)) return {};
return { ...value as Record<string, unknown> };
}
function normalizeUiState(
value: Partial<DigitalEmployeeUiState>,
): DigitalEmployeeUiState {
const fallback = defaultUiState();
const profile: Partial<DigitalEmployeeUiState["profile"]> = value.profile || {};
return {
selectedDutyIdsByDate: normalizeStringArrayMap(value.selectedDutyIdsByDate),
confirmedDates: normalizeStringMap(value.confirmedDates),
reportScheduleTime: typeof value.reportScheduleTime === "string" && value.reportScheduleTime.trim()
? value.reportScheduleTime
: fallback.reportScheduleTime,
reportGeneratedAtByDate: normalizeStringMap(value.reportGeneratedAtByDate),
decisionResultsByDate: normalizeNestedStringMap(value.decisionResultsByDate),
consoleRole: typeof value.consoleRole === "string" && value.consoleRole.trim()
? value.consoleRole
: fallback.consoleRole,
rolePreferences: normalizeRecord(value.rolePreferences),
profile: {
name: typeof profile.name === "string" && profile.name.trim()
? profile.name
: fallback.profile.name,
company: typeof profile.company === "string" && profile.company.trim()
? profile.company
: fallback.profile.company,
position: typeof profile.position === "string" && profile.position.trim()
? profile.position
: fallback.profile.position,
currentStatus: typeof profile.currentStatus === "string" && profile.currentStatus.trim()
? profile.currentStatus
: fallback.profile.currentStatus,
},
};
}
function digitalEmployeeUiActionMessage(action: string, date?: string): string {
const suffix = date ? `${date}` : "";
switch (action) {
case "confirm_workday":
return `数字员工今日任务已确认${suffix}`;
case "update_profile":
return "数字员工资料已更新";
case "update_report_schedule":
return `数字员工日报时间已更新${suffix}`;
case "generate_daily_report":
return `数字员工日报已生成${suffix}`;
case "decide_approval":
return `数字员工待决策事项已处理${suffix}`;
case "update_console_role":
return "数字员工角色已更新";
case "update_role_preferences":
return "数字员工运营偏好已更新";
default:
return `数字员工页面状态已更新${suffix}`;
}
}
export function readDigitalEmployeeState(): DigitalEmployeeState {
const state = readSetting(STATE_KEY);
const pendingSyncCount = countPendingSyncItems();
if (!state || typeof state !== "object") {
return { ...defaultState(), pendingSyncCount };
}
return {
...defaultState(),
...(state as Partial<DigitalEmployeeState>),
version: 1,
pendingSyncCount,
};
}
export function writeDigitalEmployeeState(state: DigitalEmployeeState): void {
writeSetting(STATE_KEY, state);
}
export function readDigitalEmployeeUiState(): DigitalEmployeeUiState {
const state = readSetting(UI_STATE_KEY);
if (!state || typeof state !== "object") return defaultUiState();
return normalizeUiState(state as Partial<DigitalEmployeeUiState>);
}
export function saveDigitalEmployeeUiState(
update: DigitalEmployeeUiStateUpdate,
): DigitalEmployeeUiState {
const state = normalizeUiState(update.state);
writeSetting(UI_STATE_KEY, state);
recordUiApprovalDecision(update);
if (update.action) {
recordDigitalEmployeeRuntimeEvent({
kind: `digital_workday_${update.action}`,
message: digitalEmployeeUiActionMessage(update.action, update.date),
status: "completed",
payload: {
action: update.action,
date: update.date,
metadata: update.metadata,
state,
},
});
}
return state;
}
export function readDigitalEmployeeCronSettings(): DigitalEmployeeCronSettings {
const stored = objectRecord(readSetting(CRON_SETTINGS_KEY));
return {
enabled: stored?.enabled !== false,
catch_up_on_startup: stored?.catch_up_on_startup !== false,
max_run_history: positiveInteger(stored?.max_run_history, 100),
max_catch_up_runs: positiveInteger(stored?.max_catch_up_runs, 5),
max_concurrent_schedule_runs: positiveInteger(stored?.max_concurrent_schedule_runs, 1),
};
}
export function saveDigitalEmployeeCronSettings(
patch: Partial<DigitalEmployeeCronSettings>,
): DigitalEmployeeCronSettings {
const current = readDigitalEmployeeCronSettings();
const next: DigitalEmployeeCronSettings = {
enabled: typeof patch.enabled === "boolean" ? patch.enabled : current.enabled,
catch_up_on_startup: typeof patch.catch_up_on_startup === "boolean" ? patch.catch_up_on_startup : current.catch_up_on_startup,
max_run_history: positiveInteger(patch.max_run_history, current.max_run_history),
max_catch_up_runs: positiveInteger(patch.max_catch_up_runs, current.max_catch_up_runs),
max_concurrent_schedule_runs: positiveInteger(patch.max_concurrent_schedule_runs, current.max_concurrent_schedule_runs),
};
writeSetting(CRON_SETTINGS_KEY, next);
return next;
}
function recordUiApprovalDecision(update: DigitalEmployeeUiStateUpdate): void {
if (update.action !== "decide_approval") return;
const metadata = objectRecord(update.metadata);
const decisionId = stringValue(metadata?.decision_id);
const decision = stringValue(metadata?.decision);
if (!decisionId || !decision) return;
const db = getDigitalEmployeeDb();
if (!db) return;
const existing = db.prepare(`
SELECT id, plan_id, task_id, run_id, title, status, payload, created_at
FROM digital_approvals
WHERE id = ?
`).get(decisionId) as Record<string, unknown> | undefined;
if (!existing) return;
const now = new Date().toISOString();
const previousStatus = stringField(existing, "status") || "pending";
const nextStatus = approvalDecisionStatus(decision);
const planId = stringField(existing, "plan_id");
const taskId = stringField(existing, "task_id");
const runId = stringField(existing, "run_id");
const title = stringField(existing, "title") || "待确认事项";
const storedPayload = parseStoredPayload(existing["payload"]);
const storedPayloadRecord = objectRecord(storedPayload);
const acpPermission = objectRecord(metadata?.acp_permission);
const payload = storedPayloadRecord
? {
...storedPayloadRecord,
decision,
decidedAt: now,
previousStatus,
...(acpPermission ? { acpPermission } : {}),
}
: {
value: storedPayload,
decision,
decidedAt: now,
previousStatus,
...(acpPermission ? { acpPermission } : {}),
};
upsertApproval({
id: decisionId,
planId,
taskId,
runId,
title,
status: nextStatus,
payload,
now,
});
const eventPayload = {
source: "qimingclaw-approval-decision",
approvalId: decisionId,
decision,
previousStatus,
nextStatus,
title,
planId,
taskId,
runId,
decidedAt: now,
...(acpPermission ? { acpPermission } : {}),
};
insertEvent(
{
event_id: `${decisionId}:decision:${safeId(decision)}:${safeId(now)}`,
kind: "approval_decision",
occurred_at: now,
message: `审批处理:${title} -> ${decision}`,
plan_id: planId,
task_id: taskId,
},
runId || null,
eventPayload,
);
}
function approvalDecisionStatus(decision: string): string {
const normalized = decision.toLowerCase();
if (["approve", "approved", "accept", "accepted"].includes(normalized)) return "approved";
if (["reject", "rejected", "deny", "denied"].includes(normalized)) return "rejected";
if (["dismiss", "dismissed", "skip", "skipped"].includes(normalized)) return "dismissed";
return "completed";
}
export function readDigitalEmployeeRuntimeRecords(
limit = 120,
): DigitalEmployeeRuntimeRecords {
const db = getDigitalEmployeeDb();
if (!db) {
return {
generatedAt: new Date().toISOString(),
plans: [],
planSteps: [],
schedules: [],
scheduleRuns: [],
tasks: [],
runs: [],
events: [],
artifacts: [],
approvals: [],
memories: [],
};
}
const safeLimit = Math.max(1, Math.min(Math.floor(limit), 300));
const plans = db.prepare(`
SELECT id, remote_id, title, objective, status, source, payload, sync_status,
last_synced_at, sync_error, created_at, updated_at
FROM digital_plans
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const planSteps = db.prepare(`
SELECT id, remote_id, plan_id, title, status, seq, depends_on,
preferred_skill_ids, payload, sync_status, last_synced_at,
sync_error, created_at, updated_at
FROM digital_plan_steps
ORDER BY plan_id ASC, seq ASC, created_at ASC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const schedules = db.prepare(`
SELECT id, remote_id, plan_id, name, description, cron_expr, timezone,
status, enabled, prompt, command, payload, next_run_at, last_run_at,
catch_up_on_startup, max_catch_up_runs, max_run_history,
failure_count, last_error, sync_status, last_synced_at, sync_error,
created_at, updated_at, deleted_at
FROM digital_schedules
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const scheduleRuns = db.prepare(`
SELECT id, remote_id, schedule_id, plan_id, run_id, due_at, started_at,
finished_at, status, attempt_no, trigger_kind, error, payload,
sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_schedule_runs
ORDER BY due_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const tasks = 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
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const runs = 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
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const events = db.prepare(`
SELECT id, remote_id, plan_id, task_id, run_id, kind, message, payload,
sync_status, last_synced_at, sync_error, occurred_at, created_at
FROM digital_events
ORDER BY occurred_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const artifacts = db.prepare(`
SELECT id, remote_id, plan_id, task_id, run_id, kind, name, uri, payload,
sync_status, last_synced_at, sync_error, created_at
FROM digital_artifacts
ORDER BY created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const approvals = db.prepare(`
SELECT id, remote_id, plan_id, task_id, run_id, title, status, payload,
sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_approvals
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
const memories = db.prepare(`
SELECT id, remote_id, key, content, category, source, source_path,
session_id, score, status, payload, sync_status, last_synced_at,
sync_error, created_at, updated_at, deleted_at
FROM digital_memories
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
return {
generatedAt: new Date().toISOString(),
plans: plans.map(toPlanRecord),
planSteps: planSteps.map(toPlanStepRecord),
schedules: schedules.map(toScheduleRecord),
scheduleRuns: scheduleRuns.map(toScheduleRunRecord),
tasks: tasks.map(toTaskRecord),
runs: runs.map(toRunRecord),
events: events.map(toEventRecord),
artifacts: artifacts.map(toArtifactRecord),
approvals: approvals.map(toApprovalRecord),
memories: memories.map(toMemoryRecord),
};
}
export function recordDigitalEmployeeRouteDecision(
input: DigitalEmployeeRouteDecisionInput,
): DigitalEmployeeRouteDecisionRecord {
const now = input.recordedAt || new Date().toISOString();
const idSource = stringValue(input.idempotencyKey) || stringValue(input.correlationId) || stringValue(input.id) || `${Date.now()}`;
const id = stringValue(input.id) || `route-decision:${safeId(idSource)}`;
const correlationId = stringValue(input.correlationId) || stringValue(input.idempotencyKey) || id;
const routeDecision: DigitalEmployeeRouteDecisionRecord = {
id,
route_kind: stringValue(input.routeKind) || "ChatOnly",
intent_kind: stringValue(input.intentKind) || "chat",
reason_codes: normalizeSkillIds(input.reasonCodes ?? []),
candidate_skills: normalizeSkillIds(input.candidateSkills ?? []),
context_refs: objectRecord(input.contextRefs) ?? {},
created_command_id: stringValue(input.createdCommandId) || null,
correlation_id: correlationId,
creates_plan: Boolean(input.createsPlan),
creates_task_run: Boolean(input.createsTaskRun),
approval_mode: stringValue(input.approvalMode) || null,
policy_result: stringValue(input.policyResult) || null,
recorded_at: now,
entrypoint: stringValue(input.entrypoint) || null,
actor: stringValue(input.actor) || null,
};
const db = getDigitalEmployeeDb();
if (!db) return routeDecision;
insertEvent(
{
event_id: id,
kind: "route_decision",
occurred_at: now,
message: input.message || routeDecisionMessage(routeDecision),
plan_id: null,
task_id: null,
},
null,
{
source: "qimingclaw-route-decision",
routeDecision,
envelope: objectRecord(input.envelope) ?? null,
},
);
return routeDecision;
}
export function listDigitalEmployeeRouteDecisions(options: { limit?: number } = {}): DigitalEmployeeRouteDecisionRecord[] {
const db = getDigitalEmployeeDb();
if (!db) return [];
const safeLimit = Math.max(1, Math.min(Math.floor(options.limit ?? 80), 300));
const rows = db.prepare(`
SELECT payload
FROM digital_events
WHERE kind = 'route_decision'
ORDER BY occurred_at DESC, created_at DESC
LIMIT ?
`).all(safeLimit) as Array<Record<string, unknown>>;
return rows
.map((row) => routeDecisionFromEventPayload(parseStoredPayload(row.payload)))
.filter((decision): decision is DigitalEmployeeRouteDecisionRecord => Boolean(decision));
}
function routeDecisionFromEventPayload(payload: unknown): DigitalEmployeeRouteDecisionRecord | null {
const record = objectRecord(payload);
const decision = objectRecord(record?.routeDecision ?? record?.route_decision ?? payload);
if (!decision) return null;
const id = stringValue(decision.id);
if (!id) return null;
return {
id,
route_kind: stringValue(decision.route_kind ?? decision.routeKind) || "ChatOnly",
intent_kind: stringValue(decision.intent_kind ?? decision.intentKind) || "chat",
reason_codes: parseStringArray(decision.reason_codes ?? decision.reasonCodes),
candidate_skills: parseStringArray(decision.candidate_skills ?? decision.candidateSkills),
context_refs: objectRecord(decision.context_refs ?? decision.contextRefs) ?? {},
created_command_id: stringValue(decision.created_command_id ?? decision.createdCommandId) || null,
correlation_id: stringValue(decision.correlation_id ?? decision.correlationId) || id,
creates_plan: Boolean(decision.creates_plan ?? decision.createsPlan),
creates_task_run: Boolean(decision.creates_task_run ?? decision.createsTaskRun),
approval_mode: stringValue(decision.approval_mode ?? decision.approvalMode) || null,
policy_result: stringValue(decision.policy_result ?? decision.policyResult) || null,
recorded_at: stringValue(decision.recorded_at ?? decision.recordedAt) || new Date().toISOString(),
entrypoint: stringValue(decision.entrypoint) || null,
actor: stringValue(decision.actor) || null,
};
}
function routeDecisionMessage(decision: DigitalEmployeeRouteDecisionRecord): string {
return `入口路由决策:${decision.route_kind} / ${decision.intent_kind}`;
}
export function listDigitalEmployeeMemories(options: {
query?: string | null;
category?: string | null;
limit?: number;
includeDeleted?: boolean;
} = {}): DigitalEmployeeMemoryRecord[] {
const db = getDigitalEmployeeDb();
if (!db) return [];
const rows = db.prepare(`
SELECT id, remote_id, key, content, category, source, source_path,
session_id, score, status, payload, sync_status, last_synced_at,
sync_error, created_at, updated_at, deleted_at
FROM digital_memories
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(Math.max(1, Math.min(Math.floor(options.limit ?? 200), 1000))) as Array<Record<string, unknown>>;
const query = stringValue(options.query).toLowerCase();
const category = stringValue(options.category).toLowerCase();
return rows
.map(toMemoryRecord)
.filter((memory) => options.includeDeleted === true || memory.status !== "deleted")
.filter((memory) => !category || memory.category.toLowerCase() === category)
.filter((memory) => {
if (!query) return true;
return `${memory.key} ${memory.content} ${memory.category} ${memory.source}`.toLowerCase().includes(query);
});
}
export function upsertDigitalEmployeeMemory(
input: DigitalEmployeeMemoryUpsertInput,
): DigitalEmployeeMemoryRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = new Date().toISOString();
const normalized = normalizeMemoryInput(input, now);
const tx = db.transaction(() => upsertMemory(normalized));
tx();
return readDigitalEmployeeMemory(normalized.id);
}
export function deleteDigitalEmployeeMemory(
keyOrId: string,
): DigitalEmployeeMemoryRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = new Date().toISOString();
const target = readDigitalEmployeeMemoryByKeyOrId(keyOrId);
if (!target) return null;
const tx = db.transaction(() => {
db.prepare(`
UPDATE digital_memories
SET status = 'deleted', deleted_at = ?, updated_at = ?, sync_status = 'pending'
WHERE id = ?
`).run(now, now, target.id);
markOutbox("memory", target.id, "delete", { ...target, status: "deleted", deletedAt: now, updatedAt: now }, now);
});
tx();
return readDigitalEmployeeMemory(target.id);
}
export function syncQimingclawMemoryEntries(
entries: DigitalEmployeeQimingMemoryEntry[],
): DigitalEmployeeMemoryRecord[] {
const db = getDigitalEmployeeDb();
if (!db) return [];
const now = new Date().toISOString();
const records = entries.map((entry) => normalizeQimingMemoryEntry(entry, now));
const tx = db.transaction(() => {
for (const record of records) upsertMemory(record);
});
tx();
return records
.map((record) => readDigitalEmployeeMemory(record.id))
.filter((record): record is DigitalEmployeeMemoryRecord => Boolean(record));
}
export function listDueDigitalEmployeeSchedules(
nowIso: string,
limit = 20,
): DigitalEmployeeScheduleRecord[] {
const db = getDigitalEmployeeDb();
if (!db) return [];
const rows = db.prepare(`
SELECT id, remote_id, plan_id, name, description, cron_expr, timezone,
status, enabled, prompt, command, payload, next_run_at, last_run_at,
catch_up_on_startup, max_catch_up_runs, max_run_history,
failure_count, last_error, sync_status, last_synced_at, sync_error,
created_at, updated_at, deleted_at
FROM digital_schedules
WHERE enabled = 1
AND status = 'enabled'
AND deleted_at IS NULL
AND next_run_at IS NOT NULL
AND next_run_at <= ?
ORDER BY next_run_at ASC, updated_at ASC
LIMIT ?
`).all(nowIso, Math.max(1, Math.min(Math.floor(limit), 100))) as Array<Record<string, unknown>>;
return rows.map(toScheduleRecord);
}
export function listDigitalEmployeeSchedules(limit = 1000): DigitalEmployeeScheduleRecord[] {
const db = getDigitalEmployeeDb();
if (!db) return [];
const rows = db.prepare(`
SELECT id, remote_id, plan_id, name, description, cron_expr, timezone,
status, enabled, prompt, command, payload, next_run_at, last_run_at,
catch_up_on_startup, max_catch_up_runs, max_run_history,
failure_count, last_error, sync_status, last_synced_at, sync_error,
created_at, updated_at, deleted_at
FROM digital_schedules
ORDER BY updated_at DESC, created_at DESC
LIMIT ?
`).all(Math.max(1, Math.min(Math.floor(limit), 5000))) as Array<Record<string, unknown>>;
return rows.map(toScheduleRecord);
}
export function upsertDigitalEmployeeSchedule(
input: DigitalEmployeeScheduleUpsertInput,
): DigitalEmployeeScheduleRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = new Date().toISOString();
const schedule = normalizeScheduleInput(input, now);
const tx = db.transaction(() => {
upsertSchedule(schedule);
});
tx();
return readDigitalEmployeeSchedule(schedule.id);
}
export function updateDigitalEmployeeScheduleStatus(
id: string,
status: string,
options: { enabled?: boolean; deleted?: boolean; nextRunAt?: string | null; error?: string | null } = {},
): DigitalEmployeeScheduleRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = new Date().toISOString();
const existing = readDigitalEmployeeSchedule(id);
if (!existing) return null;
const next: DigitalEmployeeScheduleUpsertInput = {
id: existing.id,
planId: existing.planId,
name: existing.name,
description: existing.description,
cronExpr: existing.cronExpr,
timezone: existing.timezone,
status,
enabled: options.enabled ?? (status === "enabled"),
prompt: existing.prompt,
command: existing.command,
payload: objectRecord(existing.payload) ?? { value: existing.payload },
nextRunAt: options.nextRunAt === undefined ? existing.nextRunAt : options.nextRunAt,
catchUpOnStartup: existing.catchUpOnStartup,
maxCatchUpRuns: existing.maxCatchUpRuns,
maxRunHistory: existing.maxRunHistory,
};
const schedule = normalizeScheduleInput(next, now);
const tx = db.transaction(() => {
upsertSchedule({
...schedule,
lastRunAt: existing.lastRunAt,
failureCount: existing.failureCount,
lastError: options.error ?? existing.lastError,
deletedAt: options.deleted ? now : existing.deletedAt,
});
});
tx();
return readDigitalEmployeeSchedule(id);
}
export function readDigitalEmployeeSchedule(id: string): DigitalEmployeeScheduleRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, plan_id, name, description, cron_expr, timezone,
status, enabled, prompt, command, payload, next_run_at, last_run_at,
catch_up_on_startup, max_catch_up_runs, max_run_history,
failure_count, last_error, sync_status, last_synced_at, sync_error,
created_at, updated_at, deleted_at
FROM digital_schedules
WHERE id = ?
`).get(id) as Record<string, unknown> | undefined;
return row ? toScheduleRecord(row) : null;
}
export function readDigitalEmployeeScheduleRuns(
scheduleId: string,
limit = 20,
): DigitalEmployeeScheduleRunRecord[] {
const db = getDigitalEmployeeDb();
if (!db) return [];
const rows = db.prepare(`
SELECT id, remote_id, schedule_id, plan_id, run_id, due_at, started_at,
finished_at, status, attempt_no, trigger_kind, error, payload,
sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_schedule_runs
WHERE schedule_id = ?
ORDER BY due_at DESC, attempt_no DESC
LIMIT ?
`).all(scheduleId, Math.max(1, Math.min(Math.floor(limit), 200))) as Array<Record<string, unknown>>;
return rows.map(toScheduleRunRecord);
}
export function recordDigitalEmployeeScheduleRunStarted(
input: DigitalEmployeeScheduleRunInput,
): DigitalEmployeeScheduleRunRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = new Date().toISOString();
const schedule = readDigitalEmployeeSchedule(input.scheduleId);
const planId = input.planId || schedule?.planId || `schedule-plan-${safeId(input.scheduleId)}`;
const runId = input.runId || `schedule-run-${safeId(input.scheduleId)}-${safeId(input.dueAt)}-${input.attemptNo || 1}`;
const taskId = `schedule-task-${safeId(input.scheduleId)}`;
const payload = {
source: "qimingclaw-schedule-run",
scheduleId: input.scheduleId,
dueAt: input.dueAt,
triggerKind: input.triggerKind,
attemptNo: input.attemptNo || 1,
schedule,
...(input.payload ?? {}),
};
const rowInput = {
id: `${input.scheduleId}:${safeId(input.dueAt)}:${input.attemptNo || 1}`,
scheduleId: input.scheduleId,
planId,
runId,
dueAt: input.dueAt,
status: input.status || "running",
attemptNo: input.attemptNo || 1,
triggerKind: input.triggerKind,
payload,
now,
};
const tx = db.transaction(() => {
upsertPlan({
id: planId,
title: schedule?.name || "数字员工定时任务",
objective: schedule?.description || schedule?.prompt || schedule?.command || "按本地调度计划执行数字员工任务。",
status: "running",
payload,
now,
});
upsertTask({
id: taskId,
planId,
title: schedule?.name || "执行数字员工定时任务",
status: "running",
assignedSkillId: "qimingclaw-scheduler",
payload,
now,
});
upsertRun({ id: runId, planId, taskId, status: "running", payload, now });
upsertScheduleRun(rowInput);
insertEvent({
event_id: `${runId}:schedule_started`,
kind: "schedule_run_started",
occurred_at: now,
message: `定时任务开始执行:${schedule?.name || input.scheduleId}`,
plan_id: planId,
task_id: taskId,
}, runId, payload);
});
tx();
return readDigitalEmployeeScheduleRun(rowInput.id);
}
export function finishDigitalEmployeeScheduleRun(
scheduleRunId: string,
status: "completed" | "failed",
options: { error?: string | null; nextRunAt?: string | null; response?: unknown } = {},
): DigitalEmployeeScheduleRunRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const existing = readDigitalEmployeeScheduleRun(scheduleRunId);
if (!existing) return null;
const now = new Date().toISOString();
const payloadRecord = objectRecord(existing.payload) ?? {};
const schedule = readDigitalEmployeeSchedule(existing.scheduleId);
const tx = db.transaction(() => {
db.prepare(`
UPDATE digital_schedule_runs
SET status = ?, finished_at = ?, error = ?, payload = ?, sync_status = CASE
WHEN sync_status = 'synced' THEN 'pending'
ELSE sync_status
END, updated_at = ?
WHERE id = ?
`).run(status, now, options.error ?? null, stringify({
...payloadRecord,
response: options.response,
error: options.error ?? null,
finishedAt: now,
}), now, scheduleRunId);
if (existing.runId) finishRun(existing.runId, status, now);
if (schedule) {
upsertSchedule({
...scheduleToUpsertInput(schedule, now),
lastRunAt: now,
nextRunAt: options.nextRunAt === undefined ? schedule.nextRunAt : options.nextRunAt,
failureCount: status === "failed" ? schedule.failureCount + 1 : 0,
lastError: status === "failed" ? options.error ?? "schedule run failed" : null,
});
}
insertEvent({
event_id: `${existing.runId || scheduleRunId}:schedule_${status}`,
kind: `schedule_run_${status}`,
occurred_at: now,
message: status === "completed" ? "定时任务执行完成" : `定时任务执行失败:${options.error || "unknown"}`,
plan_id: existing.planId ?? null,
task_id: null,
}, existing.runId || "", { scheduleRunId, error: options.error, response: options.response });
markOutbox("schedule_run", scheduleRunId, "upsert", { id: scheduleRunId, status, error: options.error }, now);
});
tx();
return readDigitalEmployeeScheduleRun(scheduleRunId);
}
export function recordDigitalEmployeeSnapshot(
snapshot: DigitalEmployeeSnapshot,
): DigitalEmployeeState {
const previous = readDigitalEmployeeState();
const events = [...previous.events];
const event = eventFromSnapshot(snapshot, previous.lastSnapshot);
if (event) {
events.unshift(event);
}
const next: DigitalEmployeeState = {
version: 1,
updatedAt: snapshot.generatedAt,
lastSnapshot: snapshot,
events: events.slice(0, MAX_EVENTS),
};
writeDigitalEmployeeState(next);
persistSnapshotTables(snapshot, event);
return next;
}
export function recordDigitalEmployeeChatReceived(
request: DigitalEmployeeChatRequestRecord,
): { planId: string; taskId: string; runId: string } {
const now = new Date().toISOString();
const ids = chatIds(request);
const title = chatTitle(request);
const db = getDigitalEmployeeDb();
if (!db) return ids;
const tx = db.transaction(() => {
upsertPlan({
id: ids.planId,
title: `远程任务:${title}`,
objective: request.prompt || request.project_id || "管理端下发任务",
status: "running",
payload: request,
now,
});
upsertTask({
id: ids.taskId,
planId: ids.planId,
title,
status: "running",
assignedSkillId: "qimingclaw-computer-control",
payload: request,
now,
});
upsertRun({
id: ids.runId,
planId: ids.planId,
taskId: ids.taskId,
status: "running",
payload: { request },
now,
});
persistPayloadArtifactsAndApprovals(ids, request, now);
insertEvent(
{
event_id: `${ids.runId}:received`,
kind: "computer_chat_received",
occurred_at: now,
message: `收到管理端任务:${title}`,
plan_id: ids.planId,
task_id: ids.taskId,
},
ids.runId,
{ request },
);
});
tx();
return ids;
}
export function recordDigitalEmployeeChatResult(
request: DigitalEmployeeChatRequestRecord,
result: DigitalEmployeeChatResultRecord,
): void {
const now = new Date().toISOString();
const ids = chatIds({
...request,
request_id: request.request_id || result.request_id,
session_id: request.session_id || result.session_id,
project_id: request.project_id || result.project_id,
});
const status = result.success ? "completed" : "failed";
const title = chatTitle(request);
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
upsertPlan({
id: ids.planId,
title: `远程任务:${title}`,
objective: request.prompt || request.project_id || "管理端下发任务",
status,
payload: { request, result },
now,
});
upsertTask({
id: ids.taskId,
planId: ids.planId,
title,
status,
assignedSkillId: "qimingclaw-computer-control",
payload: { request, result },
now,
});
upsertRun({
id: ids.runId,
planId: ids.planId,
taskId: ids.taskId,
status,
payload: { request, result },
now,
});
finishRun(ids.runId, status, now);
persistPayloadArtifactsAndApprovals(ids, { request, result }, now);
insertEvent(
{
event_id: `${ids.runId}:${status}`,
kind: result.success ? "computer_chat_completed" : "computer_chat_failed",
occurred_at: now,
message: result.success
? `管理端任务完成session=${result.session_id || request.session_id || "unknown"}`
: `管理端任务失败:${result.message || result.code || "unknown error"}`,
plan_id: ids.planId,
task_id: ids.taskId,
},
ids.runId,
{ request, result },
);
});
tx();
}
export function recordDigitalEmployeeRuntimeEvent(
event: DigitalEmployeeRuntimeEventRecord,
): void {
const now = new Date().toISOString();
const ids = chatIds({
request_id: event.request_id,
session_id: event.session_id,
project_id: event.project_id,
prompt: event.message,
});
const status = event.status ?? "running";
const title = event.message || event.kind;
const db = getDigitalEmployeeDb();
if (!db) return;
const tx = db.transaction(() => {
upsertPlan({
id: ids.planId,
title: `运行事件:${title}`,
objective: title,
status,
payload: event.payload ?? event,
now,
});
upsertTask({
id: ids.taskId,
planId: ids.planId,
title,
status,
assignedSkillId: "qimingclaw-acp-session",
payload: event.payload ?? event,
now,
});
upsertRun({
id: ids.runId,
planId: ids.planId,
taskId: ids.taskId,
status,
payload: event.payload ?? event,
now,
});
if (status === "completed" || status === "failed") {
finishRun(ids.runId, status, now);
}
persistPayloadArtifactsAndApprovals(ids, event.payload ?? event, now, {
sourceEventKind: event.kind,
});
insertEvent(
{
event_id: `${ids.runId}:${event.kind}:${Date.now()}`,
kind: event.kind,
occurred_at: now,
message: title,
plan_id: ids.planId,
task_id: ids.taskId,
},
ids.runId,
event.payload ?? event,
);
});
tx();
}
export function recordDigitalEmployeeDailyReport(
input: DigitalEmployeeDailyReportRecordInput,
): DigitalEmployeeDailyReportRecordResult | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = input.generatedAt || new Date().toISOString();
const date = input.date || now.slice(0, 10);
const planId = `daily-report:${date}`;
const taskId = `${planId}:generate`;
const reportId = input.reportId || `${planId}:${safeId(now)}`;
const runId = reportId;
const artifactId = `${reportId}:artifact:text`;
const eventId = `${reportId}:event:generated`;
const payload = {
report_id: reportId,
date,
title: input.title,
summary: input.summary,
totals: input.totals ?? {},
detail_lines: input.detailLines ?? [],
artifact_sources: input.artifactSources ?? [],
filename: input.filename,
format: "text",
text_content: input.textContent,
generated_at: now,
report_schedule_time: input.reportScheduleTime ?? null,
source: input.source ?? "qimingclaw-adapter",
model: input.model ?? "qimingclaw-adapter",
};
const tx = db.transaction(() => {
upsertPlan({
id: planId,
title: `${date} 数字员工日报`,
objective: "沉淀数字员工日报产物与同步事实。",
status: "completed",
payload: { date, latest_report_id: reportId, source: "daily_report" },
now,
});
upsertTask({
id: taskId,
planId,
title: "生成数字员工日报",
status: "completed",
assignedSkillId: "qimingclaw-artifact-report",
payload: { date, latest_report_id: reportId, source: "daily_report" },
now,
});
upsertRun({
id: runId,
planId,
taskId,
status: "completed",
payload,
now,
});
finishRun(runId, "completed", now);
upsertArtifact({
id: artifactId,
planId,
taskId,
runId,
kind: "daily_report",
name: input.filename,
uri: null,
payload,
now,
});
insertEvent({
event_id: eventId,
kind: "daily_report_generated",
occurred_at: now,
message: `数字员工日报已生成:${date}`,
plan_id: planId,
task_id: taskId,
}, runId, payload);
});
tx();
return { reportId, planId, taskId, runId, artifactId, eventId };
}
export function recordDigitalEmployeeManagedServiceAction(
input: DigitalEmployeeManagedServiceActionInput,
): DigitalEmployeeManagedServiceActionResult | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const now = input.recordedAt || new Date().toISOString();
const normalizedServiceId = safeId(input.serviceId || "unknown-service");
const eventId = `managed-service:${normalizedServiceId}:${input.action}:${input.status}:${safeId(now)}`;
const kind = input.status === "rejected"
? "managed_service_restart_rejected"
: "managed_service_restart";
const payload = {
service_id: input.serviceId,
action: input.action,
status: input.status,
reason: input.reason ?? null,
actor: input.actor ?? null,
result: input.result ?? null,
allowlisted: input.allowlisted,
};
insertEvent({
event_id: eventId,
kind,
occurred_at: now,
message: input.status === "rejected"
? `服务重启已拒绝:${input.serviceId}`
: `服务重启${input.status === "success" ? "成功" : "失败"}${input.serviceId}`,
plan_id: null,
task_id: null,
}, null, payload);
return { eventId };
}
export function recordDigitalEmployeeGovernanceCommand(
command: DigitalEmployeeGovernanceCommandRecord,
): { planId: string; taskId: string; runId: string } {
const now = new Date().toISOString();
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)
|| `governance-plan-${safeId(commandId)}`;
const ids = {
planId,
taskId: `governance-task-${safeId(commandId)}`,
runId: `governance-run-${safeId(commandId)}`,
};
const status = governanceCommandStatus(command);
const payloadEnvelope = {
source: "qimingclaw-governance-command",
commandKind: command.commandKind,
commandId,
correlationId: command.correlationId,
planId,
scheduleId: command.scheduleId || stringValue(result.schedule_id) || null,
accepted: command.accepted !== false,
message: command.message,
error: command.error,
payload,
result,
};
const db = getDigitalEmployeeDb();
if (!db) return ids;
const existingPlan = readExistingPlanSummary(planId);
const title = stringValue(payload.title ?? result.title)
|| existingPlan?.title
|| governanceCommandTitle(command.commandKind);
const objective = stringValue(payload.objective ?? payload.description ?? payload.requested_action)
|| existingPlan?.objective
|| command.message
|| title;
const tx = db.transaction(() => {
upsertPlan({
id: ids.planId,
title,
objective,
status,
payload: payloadEnvelope,
now,
});
upsertTask({
id: ids.taskId,
planId: ids.planId,
title: governanceCommandTitle(command.commandKind),
status,
assignedSkillId: "qimingclaw-governance-command",
payload: payloadEnvelope,
now,
});
if (command.commandKind === "create_plan") {
upsertCreatedPlanSteps({
planId: ids.planId,
payload,
status: "pending",
commandId,
now,
});
upsertCreatedPlanTasks({
planId: ids.planId,
payload,
status: "pending",
commandId,
now,
});
}
persistScheduleGovernanceCommand({
commandKind: command.commandKind,
commandId,
planId: ids.planId,
runId: ids.runId,
scheduleId: command.scheduleId || stringValue(result.schedule_id) || stringValue(payload.schedule_id),
payload,
result,
payloadEnvelope,
status,
now,
});
upsertRun({
id: ids.runId,
planId: ids.planId,
taskId: ids.taskId,
status,
payload: payloadEnvelope,
now,
});
if (status !== "running" && status !== "pending") {
finishRun(ids.runId, status, now);
}
persistPayloadArtifactsAndApprovals(ids, payloadEnvelope, now);
insertEvent(
{
event_id: `${ids.runId}:${command.commandKind}:${safeId(commandId)}`,
kind: `governance_${command.commandKind}`,
occurred_at: now,
message: command.message || `${governanceCommandTitle(command.commandKind)} 已记录`,
plan_id: ids.planId,
task_id: ids.taskId,
},
ids.runId,
payloadEnvelope,
);
});
tx();
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":
return "创建数字员工计划";
case "submit_plan":
return "提交数字员工计划";
case "approve_plan":
return "批准数字员工计划";
case "activate_plan":
return "激活数字员工计划";
case "create_schedule":
return "创建数字员工定时任务";
case "update_schedule":
return "更新数字员工定时任务";
case "pause_schedule":
return "暂停数字员工定时任务";
case "resume_schedule":
return "恢复数字员工定时任务";
case "delete_schedule":
return "删除数字员工定时任务";
case "run_schedule_now":
return "立即运行数字员工计划";
case "pause_plan":
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"}`;
}
}
function readExistingPlanSummary(planId: string): { title: string; objective: string } | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
try {
const row = db.prepare("SELECT title, objective FROM digital_plans WHERE id = ?").get(planId) as Record<string, unknown> | undefined;
if (!row) return null;
return {
title: stringValue(row.title),
objective: stringValue(row.objective),
};
} catch {
return null;
}
}
function upsertCreatedPlanSteps(input: {
planId: string;
payload: Record<string, unknown>;
status: string;
commandId: string;
now: string;
}): void {
extractPlanStepInputs(input.payload, input.planId).forEach((step) => {
upsertPlanStep({
id: step.id,
planId: input.planId,
title: step.title,
status: step.status || input.status,
seq: step.seq,
dependsOn: step.dependsOn,
preferredSkillIds: step.preferredSkillIds,
payload: {
source: "qimingclaw-create-plan",
commandId: input.commandId,
dependsOn: step.dependsOn,
preferredSkillIds: step.preferredSkillIds,
original: step.original,
},
now: input.now,
});
});
}
function upsertCreatedPlanTasks(input: {
planId: string;
payload: Record<string, unknown>;
status: string;
commandId: string;
now: string;
}): void {
extractPlanTaskInputs(input.payload, input.planId).forEach((task, index) => {
const title = task.title || `计划任务 ${index + 1}`;
const taskId = task.id || `${input.planId}:task:${safeId(`${index + 1}-${title}`)}`;
upsertTask({
id: taskId,
planId: input.planId,
title,
status: input.status,
assignedSkillId: task.assignedSkillId,
payload: {
source: "qimingclaw-create-plan",
commandId: input.commandId,
stepId: task.stepId,
description: task.description,
dependsOn: task.dependsOn,
skills: task.skills,
original: task.original,
},
now: input.now,
});
});
}
function persistScheduleGovernanceCommand(input: {
commandKind: string;
commandId: string;
planId: string;
runId: string;
scheduleId: string;
payload: Record<string, unknown>;
result: Record<string, unknown>;
payloadEnvelope: Record<string, unknown>;
status: string;
now: string;
}): void {
if (![
"create_schedule",
"update_schedule",
"pause_schedule",
"resume_schedule",
"delete_schedule",
"run_schedule_now",
].includes(input.commandKind)) return;
const existing = input.scheduleId ? readDigitalEmployeeSchedule(input.scheduleId) : null;
const scheduleId = input.scheduleId || existing?.id || `schedule-${safeId(input.commandId)}`;
const cronExpr = stringValue(input.payload.cron_expr ?? input.payload.cron ?? input.payload.schedule)
|| existing?.cronExpr
|| "0 9 * * *";
const enabled = input.commandKind === "pause_schedule" || input.commandKind === "delete_schedule"
? false
: input.payload.enabled !== false;
const scheduleStatus = input.commandKind === "delete_schedule"
? "deleted"
: input.commandKind === "pause_schedule"
? "paused"
: "enabled";
if (input.commandKind === "run_schedule_now") {
upsertScheduleRun({
id: `${scheduleId}:${safeId(input.now)}:1`,
scheduleId,
planId: input.planId,
runId: input.runId,
dueAt: input.now,
status: "running",
attemptNo: 1,
triggerKind: "manual",
payload: input.payloadEnvelope,
now: input.now,
});
return;
}
upsertSchedule({
id: scheduleId,
planId: input.planId,
name: stringValue(input.payload.name ?? input.result.name ?? input.payload.title)
|| existing?.name
|| governanceCommandTitle(input.commandKind),
description: stringValue(input.payload.description ?? input.result.description) || existing?.description || null,
cronExpr,
timezone: stringValue(input.payload.timezone ?? input.result.timezone) || existing?.timezone || "system",
status: scheduleStatus,
enabled,
prompt: stringValue(input.payload.prompt) || existing?.prompt || null,
command: stringValue(input.payload.command) || existing?.command || null,
payload: {
...input.payloadEnvelope,
previous: existing,
},
nextRunAt: stringValue(input.result.next_run_at ?? input.payload.next_run_at) || existing?.nextRunAt || null,
catchUpOnStartup: existing?.catchUpOnStartup ?? readDigitalEmployeeCronSettings().catch_up_on_startup,
maxCatchUpRuns: existing?.maxCatchUpRuns ?? readDigitalEmployeeCronSettings().max_catch_up_runs,
maxRunHistory: existing?.maxRunHistory ?? readDigitalEmployeeCronSettings().max_run_history,
createdAt: existing?.createdAt || input.now,
updatedAt: input.now,
lastRunAt: existing?.lastRunAt,
failureCount: existing?.failureCount ?? 0,
lastError: existing?.lastError,
deletedAt: input.commandKind === "delete_schedule" ? input.now : existing?.deletedAt,
});
}
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", "cancel_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);
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: {
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,
): Array<{
id: string;
title: string;
status: string;
seq: number;
dependsOn: string[];
preferredSkillIds: string[];
original: unknown;
}> {
const steps = unknownArray(payload.steps)
.map((step, index) => planStepInputFromRecord(objectRecord(step), index, planId))
.filter((step): step is NonNullable<typeof step> => Boolean(step));
if (steps.length > 0) return steps;
if (unknownArray(payload.tasks).length === 0) return [];
return [{
id: `${planId}:step:default`,
title: "默认执行步骤",
status: "pending",
seq: 1,
dependsOn: [],
preferredSkillIds: [],
original: { source: "default_step", task_count: unknownArray(payload.tasks).length },
}];
}
function planStepInputFromRecord(
record: Record<string, unknown> | null,
index: number,
planId: string,
): ReturnType<typeof extractPlanStepInputs>[number] | null {
if (!record) return null;
const title = stringValue(record.title ?? record.name ?? record.summary)
|| `计划步骤 ${index + 1}`;
const stepId = stringValue(record.step_id ?? record.stepId ?? record.id)
|| `${planId}:step:${index + 1}-${safeId(title)}`;
return {
id: stepId,
title,
status: stringValue(record.status) || "pending",
seq: typeof record.seq === "number" ? record.seq : Number(record.seq) || index + 1,
dependsOn: normalizeSkillIds(unknownArray(record.depends_on ?? record.dependsOn)),
preferredSkillIds: normalizeSkillIds([
...unknownArray(record.skills),
...unknownArray(record.preferred_skills),
...unknownArray(record.preferredSkills),
]),
original: record,
};
}
function extractPlanTaskInputs(payload: Record<string, unknown>, planId: string): Array<{
id: string;
stepId: string;
title: string;
description: string;
assignedSkillId: string;
dependsOn: string[];
skills: string[];
original: unknown;
}> {
const rawSteps = unknownArray(payload.steps);
const defaultStepId = rawSteps.length > 0 ? "" : `${planId}:step:default`;
const explicitTasks = unknownArray(payload.tasks)
.map((task, index) => planTaskInputFromRecord(objectRecord(task), index, defaultStepId));
const stepTasks = rawSteps.flatMap((step, stepIndex) => {
const stepRecord = objectRecord(step);
if (!stepRecord) return [];
const nestedTasks = unknownArray(stepRecord.tasks);
if (nestedTasks.length > 0) {
return nestedTasks.map((task, taskIndex) => planTaskInputFromRecord(
objectRecord(task),
taskIndex,
stringValue(stepRecord.step_id ?? stepRecord.id) || `step-${stepIndex + 1}`,
stepRecord,
));
}
return [planTaskInputFromRecord(stepRecord, stepIndex, stringValue(stepRecord.step_id ?? stepRecord.id) || `step-${stepIndex + 1}`)];
});
return dedupeByKey(
[...explicitTasks, ...stepTasks].filter((task): task is NonNullable<typeof task> => Boolean(task)),
(task) => task.id || `${task.stepId}:${task.title}`,
);
}
function planTaskInputFromRecord(
record: Record<string, unknown> | null,
index: number,
fallbackStepId: string,
parentStep?: Record<string, unknown>,
): ReturnType<typeof extractPlanTaskInputs>[number] | null {
if (!record) return null;
const stepId = stringValue(record.step_id ?? record.stepId) || fallbackStepId;
const title = stringValue(record.title ?? record.name ?? record.task_title)
|| stringValue(parentStep?.title ?? parentStep?.name)
|| `计划任务 ${index + 1}`;
const description = stringValue(record.description ?? record.summary)
|| stringValue(parentStep?.description ?? parentStep?.summary)
|| title;
const skills = normalizeSkillIds([
...unknownArray(record.skills),
...unknownArray(record.preferred_skills),
...unknownArray(parentStep?.skills),
...unknownArray(parentStep?.preferred_skills),
stringValue(record.assigned_skill_id ?? record.assignedSkillId),
]);
return {
id: stringValue(record.task_id ?? record.id),
stepId,
title,
description,
assignedSkillId: skills[0] ?? "qimingclaw-plan-template",
dependsOn: normalizeSkillIds(unknownArray(record.depends_on ?? parentStep?.depends_on)),
skills,
original: record,
};
}
function normalizeSkillIds(values: unknown[]): string[] {
return [...new Set(values
.map((value) => stringValue(value))
.filter(Boolean))];
}
function governanceCommandStatus(
command: DigitalEmployeeGovernanceCommandRecord,
): string {
if (command.accepted === false || command.error) return "failed";
const resultStatus = stringValue(command.result?.status);
if (resultStatus) return resultStatus;
switch (command.commandKind) {
case "activate_plan":
case "run_schedule_now":
return "running";
case "create_plan":
return "draft";
case "submit_plan":
return "reviewing";
case "approve_plan":
return "approved";
case "pause_schedule":
return "paused";
case "delete_schedule":
return "deleted";
case "pause_plan":
return "paused";
case "cancel_plan":
return "cancelled";
default:
return "completed";
}
}
function persistSnapshotTables(
snapshot: DigitalEmployeeSnapshot,
event: DigitalEmployeeStateEvent | null,
): void {
const db = getDigitalEmployeeDb();
if (!db) return;
const now = snapshot.generatedAt;
const planStatus = snapshot.services.agent?.running || snapshot.services.mcp?.running
? "running"
: "pending";
const taskStatus = snapshot.services.agent?.running ? "running" : "pending";
const runStatus = planStatus === "running" ? "running" : "pending";
const planPayload = {
services: snapshot.services,
sessions: snapshot.sessions,
};
const taskPayload = {
service: "agent",
agent: snapshot.services.agent,
mcp: snapshot.services.mcp,
};
const runPayload = {
snapshot,
};
const tx = db.transaction(() => {
upsertPlan({
id: "qimingclaw-client-readiness",
title: "客户端运行状态巡检",
objective: "确认 qimingclaw 客户端、Agent、MCP 和远程接入能力是否处于可执行状态。",
status: planStatus,
payload: planPayload,
now,
});
upsertTask({
id: "task-agent-status",
planId: "qimingclaw-client-readiness",
title: "读取 Agent 服务状态",
status: taskStatus,
assignedSkillId: "qimingclaw-acp-session",
payload: taskPayload,
now,
});
upsertRun({
id: "run-service-snapshot",
planId: "qimingclaw-client-readiness",
taskId: "task-agent-status",
status: runStatus,
payload: runPayload,
now,
});
if (event) {
insertEvent(event, "run-service-snapshot", { snapshot });
}
});
tx();
}
function stringify(value: unknown): string {
return JSON.stringify(value ?? {});
}
function stringField(row: Record<string, unknown>, key: string): string {
const value = row[key];
return typeof value === "string" ? value : "";
}
function optionalStringField(
row: Record<string, unknown>,
key: string,
): string | null {
const value = row[key];
return typeof value === "string" && value.trim() ? value : null;
}
function numberField(
row: Record<string, unknown>,
key: string,
fallback: number,
): number {
const value = row[key];
const numeric = typeof value === "number" ? value : Number(value);
return Number.isFinite(numeric) ? numeric : fallback;
}
function booleanField(
row: Record<string, unknown>,
key: string,
fallback: boolean,
): boolean {
const value = row[key];
if (typeof value === "boolean") return value;
if (typeof value === "number") return value !== 0;
if (typeof value === "string") return !["0", "false", "no"].includes(value.toLowerCase());
return fallback;
}
function positiveInteger(value: unknown, fallback: number): number {
const numeric = typeof value === "number" ? value : Number(value);
if (!Number.isFinite(numeric) || numeric <= 0) return fallback;
return Math.floor(numeric);
}
function parseStoredPayload(value: unknown): unknown {
if (typeof value !== "string" || !value.trim()) return {};
try {
return JSON.parse(value);
} catch {
return value;
}
}
function parseStringArray(value: unknown): string[] {
const parsed = parseStoredPayload(value);
if (!Array.isArray(parsed)) return [];
return parsed.filter((item): item is string => typeof item === "string" && item.trim().length > 0);
}
function toPlanRecord(row: Record<string, unknown>): DigitalEmployeePlanRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
title: stringField(row, "title"),
objective: optionalStringField(row, "objective"),
status: stringField(row, "status"),
source: stringField(row, "source"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
};
}
function toTaskRecord(row: Record<string, unknown>): DigitalEmployeeTaskRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: optionalStringField(row, "plan_id"),
title: stringField(row, "title"),
status: stringField(row, "status"),
assignedSkillId: optionalStringField(row, "assigned_skill_id"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
};
}
function toPlanStepRecord(row: Record<string, unknown>): DigitalEmployeePlanStepRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: stringField(row, "plan_id"),
title: stringField(row, "title"),
status: stringField(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: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
};
}
function toScheduleRecord(row: Record<string, unknown>): DigitalEmployeeScheduleRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: optionalStringField(row, "plan_id"),
name: stringField(row, "name"),
description: optionalStringField(row, "description"),
cronExpr: stringField(row, "cron_expr"),
timezone: stringField(row, "timezone") || "system",
status: stringField(row, "status") || "enabled",
enabled: booleanField(row, "enabled", true),
prompt: optionalStringField(row, "prompt"),
command: optionalStringField(row, "command"),
payload: parseStoredPayload(row.payload),
nextRunAt: optionalStringField(row, "next_run_at"),
lastRunAt: optionalStringField(row, "last_run_at"),
catchUpOnStartup: booleanField(row, "catch_up_on_startup", true),
maxCatchUpRuns: numberField(row, "max_catch_up_runs", 5),
maxRunHistory: numberField(row, "max_run_history", 100),
failureCount: numberField(row, "failure_count", 0),
lastError: optionalStringField(row, "last_error"),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
deletedAt: optionalStringField(row, "deleted_at"),
};
}
function toScheduleRunRecord(row: Record<string, unknown>): DigitalEmployeeScheduleRunRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
scheduleId: stringField(row, "schedule_id"),
planId: optionalStringField(row, "plan_id"),
runId: optionalStringField(row, "run_id"),
dueAt: stringField(row, "due_at"),
startedAt: optionalStringField(row, "started_at"),
finishedAt: optionalStringField(row, "finished_at"),
status: stringField(row, "status"),
attemptNo: numberField(row, "attempt_no", 1),
triggerKind: stringField(row, "trigger_kind") || "scheduled",
error: optionalStringField(row, "error"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
};
}
function toRunRecord(row: Record<string, unknown>): DigitalEmployeeRunRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: optionalStringField(row, "plan_id"),
taskId: optionalStringField(row, "task_id"),
status: stringField(row, "status"),
startedAt: optionalStringField(row, "started_at"),
finishedAt: optionalStringField(row, "finished_at"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
};
}
function toEventRecord(row: Record<string, unknown>): DigitalEmployeeEventRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: optionalStringField(row, "plan_id"),
taskId: optionalStringField(row, "task_id"),
runId: optionalStringField(row, "run_id"),
kind: stringField(row, "kind"),
message: stringField(row, "message"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
occurredAt: stringField(row, "occurred_at"),
createdAt: stringField(row, "created_at"),
};
}
function toArtifactRecord(
row: Record<string, unknown>,
): DigitalEmployeeArtifactRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: optionalStringField(row, "plan_id"),
taskId: optionalStringField(row, "task_id"),
runId: optionalStringField(row, "run_id"),
kind: stringField(row, "kind"),
name: optionalStringField(row, "name"),
uri: optionalStringField(row, "uri"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
};
}
function toApprovalRecord(
row: Record<string, unknown>,
): DigitalEmployeeApprovalRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
planId: optionalStringField(row, "plan_id"),
taskId: optionalStringField(row, "task_id"),
runId: optionalStringField(row, "run_id"),
title: stringField(row, "title"),
status: stringField(row, "status"),
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
};
}
function toMemoryRecord(row: Record<string, unknown>): DigitalEmployeeMemoryRecord {
return {
id: stringField(row, "id"),
remoteId: optionalStringField(row, "remote_id"),
key: stringField(row, "key"),
content: stringField(row, "content"),
category: stringField(row, "category") || "fact",
source: stringField(row, "source") || "qimingclaw",
sourcePath: optionalStringField(row, "source_path"),
sessionId: optionalStringField(row, "session_id"),
score: typeof row.score === "number" ? row.score : row.score == null ? null : Number(row.score) || null,
status: stringField(row, "status") || "active",
payload: parseStoredPayload(row.payload),
syncStatus: stringField(row, "sync_status"),
lastSyncedAt: optionalStringField(row, "last_synced_at"),
syncError: optionalStringField(row, "sync_error"),
createdAt: stringField(row, "created_at"),
updatedAt: stringField(row, "updated_at"),
deletedAt: optionalStringField(row, "deleted_at"),
};
}
function countPendingSyncItems(): number {
const db = getDigitalEmployeeDb();
if (!db) return 0;
const row = db.prepare(`
SELECT COUNT(*) AS count FROM digital_sync_outbox WHERE status = 'pending'
`).get() as { count: number } | undefined;
return row?.count ?? 0;
}
function syncOutboxId(entityType: string, entityId: string, operation = "upsert"): string {
return `${entityType}:${operation}:${entityId}`;
}
function markOutbox(
entityType: string,
entityId: string,
operation: string,
payload: unknown,
now: string,
): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_sync_outbox (
id, entity_type, entity_id, operation, payload, status, attempts, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, 'pending', 0, ?, ?)
ON CONFLICT(id) DO UPDATE SET
payload = excluded.payload,
status = 'pending',
updated_at = excluded.updated_at,
error = NULL
`).run(
syncOutboxId(entityType, entityId, operation),
entityType,
entityId,
operation,
stringify(payload),
now,
now,
);
}
function normalizeMemoryInput(
input: DigitalEmployeeMemoryUpsertInput,
now: string,
): Required<Pick<DigitalEmployeeMemoryUpsertInput, "id" | "key" | "content" | "category" | "source" | "payload">> & {
sourcePath: string | null;
sessionId: string | null;
score: number | null;
status: string;
createdAt: string;
updatedAt: string;
deletedAt: string | null;
} {
const content = stringValue(input.content);
const key = stringValue(input.key) || memoryKeyFromContent(content);
return {
id: stringValue(input.id) || `memory-${safeId(key)}`,
key,
content,
category: stringValue(input.category) || "fact",
source: stringValue(input.source) || "qimingclaw-digital-local",
sourcePath: stringValue(input.sourcePath) || null,
sessionId: stringValue(input.sessionId) || null,
score: typeof input.score === "number" && Number.isFinite(input.score) ? input.score : null,
status: stringValue(input.status) || "active",
payload: input.payload ?? {},
createdAt: input.createdAt || now,
updatedAt: input.updatedAt || now,
deletedAt: input.deletedAt ?? null,
};
}
function normalizeQimingMemoryEntry(
entry: DigitalEmployeeQimingMemoryEntry,
now: string,
): ReturnType<typeof normalizeMemoryInput> {
const content = stringValue(entry.text ?? entry.content);
const createdAt = memoryTimestampToIso(entry.createdAt) || now;
const updatedAt = memoryTimestampToIso(entry.updatedAt) || createdAt;
const score = typeof entry.importance === "number"
? entry.importance
: typeof entry.confidence === "number"
? entry.confidence
: null;
return normalizeMemoryInput({
id: stringValue(entry.id) || undefined,
key: stringValue(entry.key) || stringValue(entry.id) || memoryKeyFromContent(content),
content,
category: stringValue(entry.category) || "fact",
source: stringValue(entry.source) || "qimingclaw-memory-service",
sourcePath: stringValue(entry.sourcePath) || null,
sessionId: stringValue(entry.sessionId) || null,
score,
status: stringValue(entry.status) || "active",
payload: {
source: "qimingclaw-memory-service",
original: entry,
},
createdAt,
updatedAt,
deletedAt: stringValue(entry.status) === "deleted" ? updatedAt : null,
}, now);
}
function memoryTimestampToIso(value: number | string | undefined): string | null {
if (typeof value === "number" && Number.isFinite(value)) {
return new Date(value).toISOString();
}
if (typeof value === "string" && value.trim()) {
const parsed = Date.parse(value);
return Number.isFinite(parsed) ? new Date(parsed).toISOString() : null;
}
return null;
}
function memoryKeyFromContent(content: string): string {
return safeId(content.slice(0, 80) || `memory-${Date.now()}`);
}
function readDigitalEmployeeMemory(id: string): DigitalEmployeeMemoryRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, key, content, category, source, source_path,
session_id, score, status, payload, sync_status, last_synced_at,
sync_error, created_at, updated_at, deleted_at
FROM digital_memories
WHERE id = ?
`).get(id) as Record<string, unknown> | undefined;
return row ? toMemoryRecord(row) : null;
}
function readDigitalEmployeeMemoryByKeyOrId(keyOrId: string): DigitalEmployeeMemoryRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, key, content, category, source, source_path,
session_id, score, status, payload, sync_status, last_synced_at,
sync_error, created_at, updated_at, deleted_at
FROM digital_memories
WHERE id = ? OR key = ?
ORDER BY CASE WHEN status = 'deleted' THEN 1 ELSE 0 END ASC, updated_at DESC, created_at DESC
LIMIT 1
`).get(keyOrId, keyOrId) as Record<string, unknown> | undefined;
return row ? toMemoryRecord(row) : null;
}
function upsertMemory(input: ReturnType<typeof normalizeMemoryInput>): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_memories (
id, key, content, category, source, source_path, session_id, score,
status, payload, sync_status, created_at, updated_at, deleted_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
key = excluded.key,
content = excluded.content,
category = excluded.category,
source = excluded.source,
source_path = excluded.source_path,
session_id = excluded.session_id,
score = excluded.score,
status = excluded.status,
payload = excluded.payload,
deleted_at = excluded.deleted_at,
sync_status = CASE
WHEN digital_memories.sync_status = 'synced' THEN 'pending'
ELSE digital_memories.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.key,
input.content,
input.category,
input.source,
input.sourcePath,
input.sessionId,
input.score,
input.status,
stringify(input.payload),
input.createdAt,
input.updatedAt,
input.deletedAt,
);
markOutbox("memory", input.id, input.deletedAt ? "delete" : "upsert", input, input.updatedAt);
}
function normalizeScheduleInput(
input: DigitalEmployeeScheduleUpsertInput,
now: string,
): DigitalEmployeeScheduleUpsertInput & {
id: string;
name: string;
cronExpr: string;
timezone: string;
status: string;
enabled: boolean;
payload: Record<string, unknown>;
catchUpOnStartup: boolean;
maxCatchUpRuns: number;
maxRunHistory: number;
createdAt: string;
updatedAt: string;
lastRunAt?: string | null;
failureCount?: number;
lastError?: string | null;
deletedAt?: string | null;
} {
const settings = readDigitalEmployeeCronSettings();
const name = stringValue(input.name) || "数字员工定时任务";
const cronExpr = stringValue(input.cronExpr) || stringValue(input.payload?.cron_expr) || stringValue(input.payload?.schedule) || "0 9 * * *";
return {
...input,
id: stringValue(input.id) || `schedule-${safeId(input.planId || name)}-${safeId(cronExpr)}`,
name,
cronExpr,
timezone: stringValue(input.timezone) || "system",
status: stringValue(input.status) || (input.enabled === false ? "paused" : "enabled"),
enabled: input.enabled !== false,
payload: input.payload ?? {},
catchUpOnStartup: input.catchUpOnStartup ?? settings.catch_up_on_startup,
maxCatchUpRuns: positiveInteger(input.maxCatchUpRuns, settings.max_catch_up_runs),
maxRunHistory: positiveInteger(input.maxRunHistory, settings.max_run_history),
createdAt: now,
updatedAt: now,
};
}
function scheduleToUpsertInput(
schedule: DigitalEmployeeScheduleRecord,
now: string,
): ReturnType<typeof normalizeScheduleInput> {
return {
id: schedule.id,
planId: schedule.planId,
name: schedule.name,
description: schedule.description,
cronExpr: schedule.cronExpr,
timezone: schedule.timezone,
status: schedule.status,
enabled: schedule.enabled,
prompt: schedule.prompt,
command: schedule.command,
payload: objectRecord(schedule.payload) ?? { value: schedule.payload },
nextRunAt: schedule.nextRunAt,
catchUpOnStartup: schedule.catchUpOnStartup,
maxCatchUpRuns: schedule.maxCatchUpRuns,
maxRunHistory: schedule.maxRunHistory,
createdAt: schedule.createdAt || now,
updatedAt: now,
lastRunAt: schedule.lastRunAt,
failureCount: schedule.failureCount,
lastError: schedule.lastError,
deletedAt: schedule.deletedAt,
};
}
function upsertSchedule(input: ReturnType<typeof normalizeScheduleInput>): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_schedules (
id, plan_id, name, description, cron_expr, timezone, status, enabled,
prompt, command, payload, next_run_at, last_run_at, catch_up_on_startup,
max_catch_up_runs, max_run_history, failure_count, last_error,
sync_status, created_at, updated_at, deleted_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
name = excluded.name,
description = excluded.description,
cron_expr = excluded.cron_expr,
timezone = excluded.timezone,
status = excluded.status,
enabled = excluded.enabled,
prompt = excluded.prompt,
command = excluded.command,
payload = excluded.payload,
next_run_at = excluded.next_run_at,
last_run_at = excluded.last_run_at,
catch_up_on_startup = excluded.catch_up_on_startup,
max_catch_up_runs = excluded.max_catch_up_runs,
max_run_history = excluded.max_run_history,
failure_count = excluded.failure_count,
last_error = excluded.last_error,
deleted_at = excluded.deleted_at,
sync_status = CASE
WHEN digital_schedules.sync_status = 'synced' THEN 'pending'
ELSE digital_schedules.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.planId ?? null,
input.name,
input.description ?? null,
input.cronExpr,
input.timezone,
input.status,
input.enabled ? 1 : 0,
input.prompt ?? null,
input.command ?? null,
stringify(input.payload),
input.nextRunAt ?? null,
input.lastRunAt ?? null,
input.catchUpOnStartup ? 1 : 0,
input.maxCatchUpRuns,
input.maxRunHistory,
input.failureCount ?? 0,
input.lastError ?? null,
input.createdAt,
input.updatedAt,
input.deletedAt ?? null,
);
markOutbox("schedule", input.id, input.deletedAt ? "delete" : "upsert", input, input.updatedAt);
}
function upsertScheduleRun(input: {
id: string;
scheduleId: string;
planId: string;
runId: string;
dueAt: string;
status: string;
attemptNo: number;
triggerKind: string;
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_schedule_runs (
id, schedule_id, plan_id, run_id, due_at, started_at, status, attempt_no,
trigger_kind, payload, sync_status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
run_id = excluded.run_id,
status = excluded.status,
payload = excluded.payload,
sync_status = CASE
WHEN digital_schedule_runs.sync_status = 'synced' THEN 'pending'
ELSE digital_schedule_runs.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.scheduleId,
input.planId,
input.runId,
input.dueAt,
input.now,
input.status,
input.attemptNo,
input.triggerKind,
stringify(input.payload),
input.now,
input.now,
);
markOutbox("schedule_run", input.id, "upsert", input, input.now);
}
function readDigitalEmployeeScheduleRun(id: string): DigitalEmployeeScheduleRunRecord | null {
const db = getDigitalEmployeeDb();
if (!db) return null;
const row = db.prepare(`
SELECT id, remote_id, schedule_id, plan_id, run_id, due_at, started_at,
finished_at, status, attempt_no, trigger_kind, error, payload,
sync_status, last_synced_at, sync_error, created_at, updated_at
FROM digital_schedule_runs
WHERE id = ?
`).get(id) as Record<string, unknown> | undefined;
return row ? toScheduleRunRecord(row) : null;
}
function upsertPlan(input: {
id: string;
title: string;
objective: string;
status: string;
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_plans (
id, title, objective, status, source, payload, sync_status, created_at, updated_at
) VALUES (?, ?, ?, ?, 'qimingclaw', ?, 'pending', ?, ?)
ON CONFLICT(id) DO UPDATE SET
title = excluded.title,
objective = excluded.objective,
status = excluded.status,
payload = excluded.payload,
sync_status = CASE
WHEN digital_plans.sync_status = 'synced' THEN 'pending'
ELSE digital_plans.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.title,
input.objective,
input.status,
stringify(input.payload),
input.now,
input.now,
);
markOutbox("plan", input.id, "upsert", input, input.now);
}
function upsertPlanStep(input: {
id: string;
planId: string;
title: string;
status: string;
seq: number;
dependsOn: string[];
preferredSkillIds: string[];
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_plan_steps (
id, plan_id, title, status, seq, depends_on, preferred_skill_ids,
payload, sync_status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
title = excluded.title,
status = excluded.status,
seq = excluded.seq,
depends_on = excluded.depends_on,
preferred_skill_ids = excluded.preferred_skill_ids,
payload = excluded.payload,
sync_status = CASE
WHEN digital_plan_steps.sync_status = 'synced' THEN 'pending'
ELSE digital_plan_steps.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.planId,
input.title,
input.status,
input.seq,
stringify(input.dependsOn),
stringify(input.preferredSkillIds),
stringify(input.payload),
input.now,
input.now,
);
markOutbox("plan_step", input.id, "upsert", input, input.now);
}
function upsertTask(input: {
id: string;
planId: string;
title: string;
status: string;
assignedSkillId: string;
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_tasks (
id, plan_id, title, status, assigned_skill_id, payload, sync_status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
title = excluded.title,
status = excluded.status,
assigned_skill_id = excluded.assigned_skill_id,
payload = excluded.payload,
sync_status = CASE
WHEN digital_tasks.sync_status = 'synced' THEN 'pending'
ELSE digital_tasks.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.planId,
input.title,
input.status,
input.assignedSkillId,
stringify(input.payload),
input.now,
input.now,
);
markOutbox("task", input.id, "upsert", input, input.now);
}
function upsertRun(input: {
id: string;
planId: string;
taskId: string;
status: string;
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_runs (
id, plan_id, task_id, status, started_at, payload, sync_status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, 'pending', ?, ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
task_id = excluded.task_id,
status = excluded.status,
payload = excluded.payload,
sync_status = CASE
WHEN digital_runs.sync_status = 'synced' THEN 'pending'
ELSE digital_runs.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.planId,
input.taskId,
input.status,
input.now,
stringify(input.payload),
input.now,
input.now,
);
markOutbox("run", input.id, "upsert", input, input.now);
}
function finishRun(runId: string, status: string, now: string): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
UPDATE digital_runs
SET status = ?, finished_at = ?, updated_at = ?, sync_status = 'pending'
WHERE id = ?
`).run(status, now, now, runId);
}
function upsertArtifact(input: {
id: string;
planId: string;
taskId: string;
runId: string;
kind: string;
name: string | null;
uri: string | null;
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_artifacts (
id, plan_id, task_id, run_id, kind, name, uri, payload, sync_status, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
task_id = excluded.task_id,
run_id = excluded.run_id,
kind = excluded.kind,
name = excluded.name,
uri = excluded.uri,
payload = excluded.payload,
sync_status = CASE
WHEN digital_artifacts.sync_status = 'synced' THEN 'pending'
ELSE digital_artifacts.sync_status
END
`).run(
input.id,
input.planId,
input.taskId,
input.runId,
input.kind,
input.name,
input.uri,
stringify(input.payload),
input.now,
);
markOutbox("artifact", input.id, "upsert", input, input.now);
}
function upsertApproval(input: {
id: string;
planId: string;
taskId: string;
runId: string;
title: string;
status: string;
payload: unknown;
now: string;
}): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT INTO digital_approvals (
id, plan_id, task_id, run_id, title, status, payload, sync_status, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
ON CONFLICT(id) DO UPDATE SET
plan_id = excluded.plan_id,
task_id = excluded.task_id,
run_id = excluded.run_id,
title = excluded.title,
status = excluded.status,
payload = excluded.payload,
sync_status = CASE
WHEN digital_approvals.sync_status = 'synced' THEN 'pending'
ELSE digital_approvals.sync_status
END,
updated_at = excluded.updated_at
`).run(
input.id,
input.planId,
input.taskId,
input.runId,
input.title,
input.status,
stringify(input.payload),
input.now,
input.now,
);
markOutbox("approval", input.id, "upsert", input, input.now);
}
function insertEvent(
event: DigitalEmployeeStateEvent,
runId: string | null,
payload: unknown,
): void {
const db = getDigitalEmployeeDb();
if (!db) return;
db.prepare(`
INSERT OR IGNORE INTO digital_events (
id, plan_id, task_id, run_id, kind, message, payload, sync_status, occurred_at, created_at
) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?, ?)
`).run(
event.event_id,
event.plan_id,
event.task_id,
runId,
event.kind,
event.message,
stringify(payload),
event.occurred_at,
event.occurred_at,
);
markOutbox("event", event.event_id, "insert", event, event.occurred_at);
}
function persistPayloadArtifactsAndApprovals(
ids: { planId: string; taskId: string; runId: string },
payload: unknown,
now: string,
context: ArtifactExtractionContext = {},
): void {
const sources = payloadSources(payload);
const artifacts = sources.flatMap((source) => extractArtifacts(source, context));
const approvals = sources.flatMap((source) => extractApprovals(source));
dedupeByKey(artifacts, (artifact) => artifact.key).slice(0, 20).forEach((artifact, index) => {
upsertArtifact({
id: `${ids.runId}:artifact:${safeId(artifact.key || String(index + 1))}`,
planId: ids.planId,
taskId: ids.taskId,
runId: ids.runId,
kind: artifact.kind,
name: artifact.name,
uri: artifact.uri,
payload: artifact.payload,
now,
});
});
dedupeByKey(approvals, (approval) => approval.key).slice(0, 20).forEach((approval, index) => {
upsertApproval({
id: `${ids.runId}:approval:${safeId(approval.key || String(index + 1))}`,
planId: ids.planId,
taskId: ids.taskId,
runId: ids.runId,
title: approval.title,
status: approval.status,
payload: approval.payload,
now,
});
});
}
function payloadSources(payload: unknown): unknown[] {
const record = objectRecord(payload);
if (!record) return [payload];
return [
payload,
record.result,
record.response,
record.output,
record.payload,
].filter((item) => item !== undefined && item !== null);
}
interface ExtractedArtifact {
key: string;
kind: string;
name: string | null;
uri: string | null;
payload: unknown;
}
interface ArtifactExtractionContext {
sourceEventKind?: string;
}
interface DeliveryMetadata {
stage: string;
status: string | null;
workspace_id: string | null;
project_id: string | null;
file_id: string | null;
version: string | null;
filename: string | null;
size: number | null;
uri: string | null;
error: string | null;
source_tool: string | null;
source_event_kind: string | null;
}
function extractArtifacts(payload: unknown, context: ArtifactExtractionContext = {}): ExtractedArtifact[] {
const record = objectRecord(payload);
if (!record) return typeof payload === "string" && looksLikeArtifactUri(payload)
? [artifactFromValue(payload, 0)]
: [];
const fileServerArtifacts = extractFileServerArtifacts(record, context);
if (fileServerArtifacts.length > 0) return fileServerArtifacts;
const values = [
...unknownArray(record.artifacts),
...unknownArray(record.outputs),
...unknownArray(record.files),
...unknownArray(record.attachments),
];
for (const key of ["artifact", "outputPath", "output_path", "filePath", "file_path", "path", "uri", "url"]) {
if (record[key]) values.push(record[key]);
}
return values
.map((value, index) => artifactFromValue(value, index))
.filter((artifact) => artifact.name || artifact.uri);
}
function artifactFromValue(value: unknown, index: number): ExtractedArtifact {
const record = objectRecord(value);
if (record) {
const uri = stringValue(record.uri ?? record.url ?? record.path ?? record.filePath ?? record.file_path ?? record.outputPath ?? record.output_path);
const name = stringValue(record.name ?? record.title ?? record.filename) || artifactName(uri) || `产物 ${index + 1}`;
const kind = stringValue(record.kind ?? record.type) || "artifact";
return {
key: stringValue(record.id ?? record.artifact_id) || uri || name,
kind,
name,
uri: uri || null,
payload: value,
};
}
const text = String(value ?? "").trim();
return {
key: text,
kind: "artifact",
name: artifactName(text) || `产物 ${index + 1}`,
uri: looksLikeArtifactUri(text) ? text : null,
payload: value,
};
}
const FILE_SERVER_TOOLS = new Set([
"create_workspace",
"create_workspace_v2",
"push_skills_to_workspace",
"push_skills_to_workspace_v2",
"get_file_list",
"files_update",
"upload_file",
"upload_files",
"download_all_files",
"create_project",
"upload_project",
"get_project_content",
"get_project_content_by_version",
"backup_current_version",
"export_project",
"delete_project",
"upload_attachment_file",
"copy_project",
]);
function extractFileServerArtifacts(
record: Record<string, unknown>,
context: ArtifactExtractionContext,
): ExtractedArtifact[] {
const sourceTool = normalizeFileServerTool(record);
if (!isFileServerPayload(record, sourceTool, context)) return [];
const values = [
...unknownArray(record.artifacts),
...unknownArray(record.outputs),
...unknownArray(record.files),
...unknownArray(record.attachments),
];
if (values.length === 0) values.push(record);
return values.map((value, index) => {
const item = objectRecord(value);
return fileServerArtifactFromRecord(item ? { ...record, ...item } : record, sourceTool, context, index);
}).filter((artifact) => artifact.name || artifact.uri);
}
function fileServerArtifactFromRecord(
record: Record<string, unknown>,
sourceTool: string,
context: ArtifactExtractionContext,
index: number,
): ExtractedArtifact {
const delivery = buildFileServerDelivery(record, sourceTool, context);
const name = delivery.filename
|| stringValue(record.name ?? record.title ?? record.projectName ?? record.project_name)
|| delivery.workspace_id
|| artifactName(delivery.uri || "")
|| `File Server 产物 ${index + 1}`;
return {
key: delivery.file_id
|| stringValue(record.id ?? record.artifact_id)
|| delivery.uri
|| `${delivery.stage}:${delivery.workspace_id || delivery.project_id || name}`,
kind: delivery.stage,
name,
uri: delivery.uri,
payload: {
...record,
delivery,
},
};
}
function buildFileServerDelivery(
record: Record<string, unknown>,
sourceTool: string,
context: ArtifactExtractionContext,
): DeliveryMetadata {
const uri = stringValue(
record.downloadUrl
?? record.download_url
?? record.exportPath
?? record.export_path
?? record.uri
?? record.url
?? record.path
?? record.filePath
?? record.file_path
?? record.outputPath
?? record.output_path,
);
return {
stage: fileServerDeliveryStage(sourceTool),
status: stringValue(record.status ?? record.deliveryStatus ?? record.delivery_status) || null,
workspace_id: stringValue(record.workspaceId ?? record.workspace_id ?? record.workspace) || null,
project_id: stringValue(record.projectId ?? record.project_id ?? record.project) || null,
file_id: stringValue(record.fileId ?? record.file_id ?? record.id ?? record.artifact_id) || null,
version: stringValue(record.version ?? record.fileVersion ?? record.file_version) || null,
filename: stringValue(record.filename ?? record.fileName ?? record.file_name ?? record.name) || artifactName(uri) || null,
size: numberValue(record.size ?? record.fileSize ?? record.file_size ?? record.bytes),
uri: uri || null,
error: stringValue(record.error ?? record.message_error ?? record.lastError ?? record.last_error) || null,
source_tool: sourceTool || null,
source_event_kind: context.sourceEventKind || null,
};
}
function isFileServerPayload(
record: Record<string, unknown>,
sourceTool: string,
context: ArtifactExtractionContext,
): boolean {
if (sourceTool && FILE_SERVER_TOOLS.has(sourceTool)) return true;
const text = [
record.source,
record.service,
record.provider,
record.server,
context.sourceEventKind,
].map((value) => stringValue(value).toLowerCase()).join(" ");
return text.includes("qiming-file-server")
|| text.includes("qiming_file_server")
|| text.includes("file_server")
|| text.includes("file-server");
}
function normalizeFileServerTool(record: Record<string, unknown>): string {
return stringValue(record.toolName ?? record.tool_name ?? record.tool ?? record.action ?? record.command)
.toLowerCase()
.replace(/-/g, "_");
}
function fileServerDeliveryStage(tool: string): string {
if (tool.includes("workspace")) return "workspace";
if (tool.includes("download")) return "download";
if (tool.includes("export")) return "export";
if (tool.includes("upload") || tool.includes("push_skills")) return "upload";
if (tool.includes("files_update") || tool.includes("file_list") || tool.includes("project_content")) return "file_update";
return "artifact";
}
function numberValue(value: unknown): number | null {
if (typeof value === "number" && Number.isFinite(value)) return value;
if (typeof value === "string" && value.trim()) {
const parsed = Number(value);
return Number.isFinite(parsed) ? parsed : null;
}
return null;
}
interface ExtractedApproval {
key: string;
title: string;
status: string;
payload: unknown;
}
function extractApprovals(payload: unknown): ExtractedApproval[] {
const record = objectRecord(payload);
if (!record) return [];
const values = [
...unknownArray(record.approvals),
...unknownArray(record.approval_requests),
...unknownArray(record.approvalRequests),
...unknownArray(record.decisions),
];
if (record.needApproval === true || record.need_approval === true || record.requiresApproval === true) {
values.push(record);
}
return values.map((value, index) => approvalFromValue(value, index));
}
function approvalFromValue(value: unknown, index: number): ExtractedApproval {
const record = objectRecord(value);
if (!record) {
return {
key: String(index + 1),
title: `待确认事项 ${index + 1}`,
status: "pending",
payload: value,
};
}
const title = stringValue(record.title ?? record.message ?? record.reason ?? record.summary)
|| `待确认事项 ${index + 1}`;
return {
key: stringValue(record.id ?? record.approval_id ?? record.key) || title,
title,
status: stringValue(record.status) || "pending",
payload: value,
};
}
function objectRecord(value: unknown): Record<string, unknown> | null {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: null;
}
function unknownArray(value: unknown): unknown[] {
return Array.isArray(value) ? value : [];
}
function stringValue(value: unknown): string {
return typeof value === "string" ? value.trim() : "";
}
function looksLikeArtifactUri(value: string): boolean {
return /^(file|https?):\/\//i.test(value) || value.includes("/") || value.includes("\\");
}
function artifactName(value: string): string {
if (!value) return "";
const normalized = value.split(/[?#]/)[0] || value;
return normalized.split(/[\\/]/).filter(Boolean).pop() ?? "";
}
function dedupeByKey<T>(items: T[], keyOf: (item: T) => string): T[] {
const seen = new Set<string>();
return items.filter((item) => {
const key = keyOf(item);
if (!key || seen.has(key)) return false;
seen.add(key);
return true;
});
}
function chatIds(request: DigitalEmployeeChatRequestRecord): {
planId: string;
taskId: string;
runId: string;
} {
const base = safeId(request.request_id || request.session_id || request.project_id || `${Date.now()}`);
return {
planId: `computer-chat-plan-${base}`,
taskId: `computer-chat-task-${base}`,
runId: `computer-chat-run-${base}`,
};
}
function chatTitle(request: DigitalEmployeeChatRequestRecord): string {
const prompt = (request.prompt || "").replace(/\s+/g, " ").trim();
if (prompt) return prompt.length > 48 ? `${prompt.slice(0, 48)}...` : prompt;
return request.project_id ? `项目 ${request.project_id}` : "管理端下发任务";
}
function safeId(value: string): string {
return value.replace(/[^a-zA-Z0-9_-]+/g, "-").replace(/^-+|-+$/g, "") || "unknown";
}
function eventFromSnapshot(
snapshot: DigitalEmployeeSnapshot,
previous: DigitalEmployeeSnapshot | null,
): DigitalEmployeeStateEvent | null {
const currentSummary = summarizeSnapshot(snapshot);
const previousSummary = previous ? summarizeSnapshot(previous) : "";
if (currentSummary === previousSummary) {
return null;
}
return {
event_id: `qimingclaw-snapshot-${Date.now()}`,
kind: "service_snapshot",
occurred_at: snapshot.generatedAt,
message: currentSummary,
plan_id: "qimingclaw-client-readiness",
task_id: "task-agent-status",
};
}
function summarizeSnapshot(snapshot: DigitalEmployeeSnapshot): string {
const serviceNames = [
["agent", "Agent"],
["mcp", "MCP"],
["computerServer", "Computer Server"],
["lanproxy", "Lanproxy"],
["fileServer", "File Server"],
] as const;
const services = serviceNames.map(([key, label]) => {
const status = snapshot.services[key];
return `${label}:${status?.running ? "running" : "stopped"}`;
});
services.push(`sessions:${snapshot.sessions.active}/${snapshot.sessions.total}`);
return services.join(";");
}