refactor(automation): register workflow actions

This commit is contained in:
ZacharyZhang-NY committed 2026-04-22 18:54:31 -04:00
1 parent 8bf2299092
commit 0d9f4f15eb
7 files changed
+524 -387

No files matched your search

@@ -0,0 +1,380 @@
import type { AutomationAction } from "@shared/types";
import { dispatchAgentMessage } from "../messaging/dispatch";
import { validateTaskCreation } from "../task-quality-gates";
import {
createAutomationActionHandler,
type AutomationActionHandler,
type TypedAutomationActionExecution,
} from "./automation-action-runtime";
const HEARTBEAT_WAKE_TRIGGER_MAP: Record<string, string> = {
run_completed: "subtask_completed",
run_failed: "report_failed",
hire_approved: "assignment",
agent_idle: "timer",
};
const ESCALATION_STATUS_TRIGGER_MAP: Record<string, string> = {
in_review: "subtask_completed",
blocked: "report_blocked",
};
const ESCALATION_RULE_TRIGGER_MAP: Record<string, string> = {
run_completed: "subtask_completed",
run_failed: "report_failed",
};
const TASK_PRIORITY_SCORE: Record<string, number> = { critical: 4, high: 3, medium: 2, low: 1 };
export const AUTOMATION_ACTION_HANDLERS = {
assign_task: createAutomationActionHandler("assign_task", handleAssignTask),
create_task: createAutomationActionHandler("create_task", handleCreateTask),
notify_agent: createAutomationActionHandler("notify_agent", handleAgentNotification),
send_message: createAutomationActionHandler("send_message", handleAgentNotification),
trigger_heartbeat: createAutomationActionHandler("trigger_heartbeat", handleTriggerHeartbeat),
create_approval: createAutomationActionHandler("create_approval", handleCreateApproval),
update_task_status: createAutomationActionHandler("update_task_status", handleUpdateTaskStatus),
escalate_to_manager: createAutomationActionHandler("escalate_to_manager", handleEscalateToManager),
cross_department_notify: createAutomationActionHandler("cross_department_notify", handleCrossDepartmentNotify),
schedule_meeting: createAutomationActionHandler("schedule_meeting", handleScheduleMeeting),
create_document: createAutomationActionHandler("create_document", handleCreateDocument),
auto_approve: createAutomationActionHandler("auto_approve", handleAutoApprove),
reassign_task: createAutomationActionHandler("reassign_task", handleReassignTask),
run_task: createAutomationActionHandler("run_task", handleRunTask),
} satisfies Record<AutomationAction, AutomationActionHandler>;
function deriveHeartbeatWakeTrigger(triggerType: string, previousStatus: string | null): string {
if (triggerType === "task_status_changed" && previousStatus === "in_review") return "comment";
return HEARTBEAT_WAKE_TRIGGER_MAP[triggerType] ?? "manual";
}
function deriveEscalationWakeTrigger(newStatus: string | null, ruleTrigger: string | null): string {
if (newStatus && ESCALATION_STATUS_TRIGGER_MAP[newStatus]) return ESCALATION_STATUS_TRIGGER_MAP[newStatus];
if (ruleTrigger && ESCALATION_RULE_TRIGGER_MAP[ruleTrigger]) return ESCALATION_RULE_TRIGGER_MAP[ruleTrigger];
return "report_blocked";
}
function handleAssignTask({ deps, ctx }: TypedAutomationActionExecution<"assign_task">): void {
const { db, logger, wakeAgentIfPossible } = deps;
if (ctx.taskId) {
const assignedId = db.tasks.autoAssignTask(ctx.taskId, ctx.companyId);
if (assignedId) {
wakeAgentIfPossible(assignedId, ctx.companyId, "assignment");
}
return;
}
if (!ctx.agentId) return;
const snapshot = db.snap.listSnapshot();
const unassigned = snapshot.tasks
.filter((task) => task.companyId === ctx.companyId && !task.assigneeAgentId && ["backlog", "todo"].includes(task.status))
.sort((a, b) => (TASK_PRIORITY_SCORE[b.priority] ?? 0) - (TASK_PRIORITY_SCORE[a.priority] ?? 0));
if (unassigned.length === 0) return;
const task = unassigned[0]!;
db.tasks.saveTask({
id: task.id,
companyId: task.companyId,
projectId: task.projectId,
goalId: task.goalId,
parentId: task.parentId,
title: task.title,
description: task.description,
assigneeAgentId: ctx.agentId,
workspaceId: task.workspaceId,
requiresFilesystem: task.requiresFilesystem,
priority: task.priority,
status: "todo",
});
wakeAgentIfPossible(ctx.agentId, ctx.companyId, "assignment");
logger.info(`[automation] Auto-assigned task "${task.title}" to idle agent ${ctx.agentId}`);
}
function handleCreateTask({ deps, callbacks, ctx, config }: TypedAutomationActionExecution<"create_task">): void {
const { db, logger } = deps;
const { handleTaskCreated } = callbacks;
const title = config.title ?? "[Auto] Task from automation rule";
const description = config.description ?? "";
const priority = config.priority ?? "medium";
const assigneeAgentId = config.assigneeAgentId ?? null;
const taskValidation = validateTaskCreation({ title, description });
if (!taskValidation.valid) {
logger.warn(`[automation] Skipping create_task - validation failed: ${taskValidation.errors.join("; ")}`);
return;
}
const taskId = db.tasks.saveTask({
companyId: ctx.companyId,
projectId: ctx.projectId ?? config.projectId ?? null,
title,
description,
assigneeAgentId,
priority,
status: assigneeAgentId ? "todo" : "backlog",
requiresFilesystem: config.requiresFilesystem,
});
handleTaskCreated(taskId, ctx.companyId);
logger.info(`[automation] Created task: ${title} (${taskId})`);
}
function handleAgentNotification<K extends "notify_agent" | "send_message">({
deps,
rule,
ctx,
config,
}: TypedAutomationActionExecution<K>): void {
const { db, logger, emitEvent, publishDomainChanged, wakeAgentIfPossible } = deps;
const toAgentId = config.toAgentId ?? ctx.agentId ?? null;
const fromAgentId = config.fromAgentId ?? ctx.agentId ?? toAgentId;
if (!fromAgentId) return;
const defaultChannel = toAgentId ? "direct" : "company";
const requestedChannel = config.channel ?? defaultChannel;
const channel = (["direct", "department", "company", "project", "incident"] as const).includes(requestedChannel)
? requestedChannel
: defaultChannel;
let channelTargetId: string | null = config.channelTargetId ?? config.department ?? ctx.department ?? null;
if (channel === "department" && !channelTargetId) {
try {
const sender = db.entityCrud.getAgent(fromAgentId);
channelTargetId = sender.department ?? null;
} catch (err) {
logger.warn(`[automation] getAgent failed for sender ${fromAgentId}; dropping department target:`, err instanceof Error ? err.message : String(err));
channelTargetId = null;
}
}
const channelNeedsTarget = channel === "department" || channel === "project";
if (channelNeedsTarget && !channelTargetId) {
logger.info(
`[automation] Skipping rule "${rule.name}" - no ${channel} channel target available for sender ${fromAgentId}`,
);
return;
}
const dispatchResult = dispatchAgentMessage(
{ db, emitEvent, publishDomainChanged, wakeAgentIfPossible },
{
companyId: ctx.companyId,
fromAgentId,
toAgentId: channel === "direct" ? toAgentId : null,
channel,
channelTargetId,
subject: config.subject ?? "Automation notification",
body: config.body ?? `Automation rule triggered for task ${ctx.taskId ?? "n/a"}`,
priority: config.priority ?? "normal",
},
"message",
);
if (!dispatchResult.ok) {
logger.warn(
`[automation] Failed to dispatch message for rule ${rule.name} (${rule.id}): ${dispatchResult.error.code} ${dispatchResult.error.message}`,
);
}
}
function handleTriggerHeartbeat({ deps, rule, ctx, config }: TypedAutomationActionExecution<"trigger_heartbeat">): void {
const { db, wakeAgentIfPossible } = deps;
let targetAgentId = config.agentId ?? null;
if (!targetAgentId && ctx.agentId) {
const triggerType = rule.trigger ?? "";
if (triggerType === "run_completed" || triggerType === "run_failed") {
const manager = db.agents.getManagerForAgent(ctx.agentId);
targetAgentId = manager?.id ?? ctx.agentId;
} else {
targetAgentId = ctx.agentId;
}
}
if (targetAgentId) {
wakeAgentIfPossible(targetAgentId, ctx.companyId, deriveHeartbeatWakeTrigger(rule.trigger ?? "", ctx.previousStatus ?? null));
}
}
function handleCreateApproval({ deps, ctx, config }: TypedAutomationActionExecution<"create_approval">): void {
deps.db.approvalsMod.requestApproval({
companyId: ctx.companyId,
relatedTaskId: ctx.taskId ?? null,
requestedByAgentId: ctx.agentId ?? null,
type: config.approvalType ?? "document_review",
payloadSummary: config.summary ?? "Auto-generated approval request",
impactSummary: config.impact ?? "Triggered by automation rule",
});
}
function handleUpdateTaskStatus({
deps,
callbacks,
rule,
ctx,
config,
}: TypedAutomationActionExecution<"update_task_status">): void {
if (!ctx.taskId) return;
const { db, publishDomainChanged } = deps;
const { handleTaskStatusChange } = callbacks;
const newStatus = config.status ?? "todo";
const previousTask = db.entityCrud.getTask(ctx.taskId);
db.tasks.advanceTaskStatus(ctx.taskId, newStatus, "Status changed by automation rule");
publishDomainChanged();
if (previousTask.status !== newStatus && rule.trigger !== "task_status_changed") {
handleTaskStatusChange(ctx.taskId, ctx.companyId, previousTask.status, newStatus);
}
}
function handleEscalateToManager({
deps,
rule,
ctx,
config,
}: TypedAutomationActionExecution<"escalate_to_manager">): void {
const { db, logger, emitEvent, publishDomainChanged, wakeAgentIfPossible } = deps;
const targetAgentId = ctx.agentId;
if (!targetAgentId) return;
let agentName = "an agent";
try {
agentName = db.entityCrud.getAgent(targetAgentId).name;
} catch (err) {
logger.warn(`[automation] getAgent failed for escalation target ${targetAgentId}; using generic label:`, err instanceof Error ? err.message : String(err));
}
let taskTitle = ctx.taskId ?? "unknown";
try {
if (ctx.taskId) taskTitle = db.entityCrud.getTask(ctx.taskId).title;
} catch (err) {
logger.warn(`[automation] getTask failed for escalation task ${ctx.taskId}; using id as title:`, err instanceof Error ? err.message : String(err));
}
const manager = db.agents.getManagerForAgent(targetAgentId);
const escalationBody = config.body ?? `Task "${taskTitle}" needs your attention. ${agentName} has flagged this as requiring manager intervention.`;
const wakeTrigger = deriveEscalationWakeTrigger(ctx.newStatus ?? null, rule.trigger ?? null);
if (manager) {
const escalationResult = dispatchAgentMessage(
{ db, emitEvent, publishDomainChanged, wakeAgentIfPossible },
{
companyId: ctx.companyId,
fromAgentId: targetAgentId,
toAgentId: manager.id,
channel: "direct",
subject: `Escalation from ${agentName}: ${taskTitle}`,
body: escalationBody,
priority: "urgent",
},
wakeTrigger,
);
if (!escalationResult.ok) {
logger.warn(
`[automation] Failed to escalate "${taskTitle}" to manager: ${escalationResult.error.code} ${escalationResult.error.message}`,
);
}
return;
}
logger.info(`[automation] Skipping escalation for top-level agent "${agentName}" - no manager to escalate to.`);
}
function handleCrossDepartmentNotify({
deps,
rule,
ctx,
config,
}: TypedAutomationActionExecution<"cross_department_notify">): void {
const { db, logger, emitEvent, publishDomainChanged, wakeAgentIfPossible } = deps;
const targetDepartment = rule.targetDepartment ?? config.department;
if (!targetDepartment) return;
const departmentAgents = db.agents.getIdleAgentsForDepartment(ctx.companyId, targetDepartment);
const senderAgentId = ctx.agentId ?? departmentAgents[0]?.id;
if (!senderAgentId) return;
const dispatchResult = dispatchAgentMessage(
{ db, emitEvent, publishDomainChanged, wakeAgentIfPossible },
{
companyId: ctx.companyId,
fromAgentId: senderAgentId,
toAgentId: null,
channel: "department",
channelTargetId: targetDepartment,
subject: config.subject ?? "Cross-department notification",
body: config.body ?? "Automation triggered a cross-department alert.",
priority: "normal",
},
"message",
);
if (!dispatchResult.ok) {
logger.warn(
`[automation] Failed cross-department notification for ${targetDepartment}: ${dispatchResult.error.code} ${dispatchResult.error.message}`,
);
}
}
function handleScheduleMeeting({ deps, ctx, config }: TypedAutomationActionExecution<"schedule_meeting">): void {
const meetingType = config.meetingType ?? "department_sync";
const title = config.title ?? "Auto-scheduled meeting";
const offsetMinutes = config.scheduleOffsetMinutes ?? 60;
const scheduledAt = config.scheduledAt ?? new Date(Date.now() + offsetMinutes * 60_000).toISOString();
deps.db.knowledgeMod.saveMeeting({
companyId: ctx.companyId,
type: meetingType,
title,
organizerAgentId: ctx.agentId ?? null,
participantAgentIds: config.participantAgentIds ?? "[]",
scheduledAt,
durationMinutes: config.durationMinutes ?? 30,
});
}
function handleCreateDocument({ deps, ctx, config }: TypedAutomationActionExecution<"create_document">): void {
deps.db.documentsMod.saveDocument({
companyId: ctx.companyId,
type: config.documentType ?? "status_report",
title: config.title ?? "Auto-generated document",
content: config.content ?? "",
authorAgentId: ctx.agentId ?? null,
projectId: ctx.projectId ?? null,
});
}
function handleAutoApprove({ deps, callbacks, ctx }: TypedAutomationActionExecution<"auto_approve">): void {
const snapshot = deps.db.snap.listSnapshot();
const pendingApprovals = snapshot.approvals.filter(
(approval) => approval.companyId === ctx.companyId && approval.state === "pending",
);
for (const approval of pendingApprovals.slice(0, 5)) {
deps.db.approvalsMod.decideApproval({
companyId: ctx.companyId,
approvalId: approval.id,
state: "approved",
decisionNote: "Auto-approved by automation rule.",
});
callbacks.handleApprovalResolved(approval.id, ctx.companyId, "approved");
}
}
function handleReassignTask({ deps, ctx }: TypedAutomationActionExecution<"reassign_task">): void {
if (!ctx.taskId) return;
const newAssignee = deps.db.tasks.autoAssignTask(ctx.taskId, ctx.companyId);
if (newAssignee) {
deps.wakeAgentIfPossible(newAssignee, ctx.companyId, "assignment");
}
}
function handleRunTask({ deps, ctx, config }: TypedAutomationActionExecution<"run_task">): void {
const targetTaskId = config.taskId ?? ctx.taskId;
if (!targetTaskId) return;
const task = deps.db.entityCrud.getTask(targetTaskId);
if (!task.assigneeAgentId) return;
deps.wakeAgentIfPossible(task.assigneeAgentId, ctx.companyId, "assignment");
}
@@ -0,0 +1,62 @@
import type { AppDatabase } from "../database";
import type { AutomationAction, DesktopEvent } from "@shared/types";
import type { AutomationEventContext } from "./types";
import type { AutomationActionConfigMap } from "./automation-action-schemas";
export interface ActionDeps {
db: AppDatabase;
logger: {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
emitEvent: (event: DesktopEvent) => void;
publishDomainChanged: () => void;
wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null;
}
export interface ActionCallbacks {
handleTaskCreated: (taskId: string, companyId: string) => void;
handleApprovalResolved: (approvalId: string, companyId: string, decision: string) => void;
handleTaskStatusChange: (taskId: string, companyId: string, previousStatus: string, newStatus: string) => void;
}
export interface AutomationRuleInput {
id?: string;
name?: string;
action: string;
actionConfigJson: string;
targetDepartment: string | null;
trigger?: string;
}
export interface AutomationRuleExecution extends Omit<AutomationRuleInput, "action"> {
action: AutomationAction;
}
export type AutomationActionExecution = {
deps: ActionDeps;
callbacks: ActionCallbacks;
rule: AutomationRuleExecution;
ctx: AutomationEventContext;
config: AutomationActionConfigMap[AutomationAction];
};
export type TypedAutomationActionExecution<K extends AutomationAction> = Omit<AutomationActionExecution, "config"> & {
config: AutomationActionConfigMap[K];
};
export type TypedAutomationActionHandler<K extends AutomationAction> = (input: TypedAutomationActionExecution<K>) => void;
export type AutomationActionHandler = (input: AutomationActionExecution) => void;
export function createAutomationActionHandler<K extends AutomationAction>(
action: K,
handler: TypedAutomationActionHandler<K>,
): AutomationActionHandler {
return (input) => {
if (input.rule.action !== action) {
throw new Error(`Automation handler ${action} received ${input.rule.action}.`);
}
handler(input as TypedAutomationActionExecution<K>);
};
}
@@ -10,7 +10,7 @@
*/
import { z } from "zod";
import { approvalTypeEnum } from "@shared/types";
import { approvalTypeEnum, type AutomationAction } from "@shared/types";
const messageChannelSchema = z.enum(["direct", "department", "company", "project", "incident"]);
const messagePrioritySchema = z.enum(["urgent", "normal", "low"]);
@@ -87,7 +87,7 @@ export const runTaskConfigSchema = z.object({
taskId: z.string().optional(),
});
const ACTION_SCHEMAS = {
export const ACTION_SCHEMAS = {
assign_task: assignTaskConfigSchema,
create_task: createTaskConfigSchema,
notify_agent: notifyAgentConfigSchema,
@@ -102,7 +102,14 @@ const ACTION_SCHEMAS = {
auto_approve: autoApproveConfigSchema,
reassign_task: reassignTaskConfigSchema,
run_task: runTaskConfigSchema,
} as const;
} as const satisfies Record<AutomationAction, z.ZodTypeAny>;
export const AUTOMATION_ACTIONS = Object.freeze(Object.keys(ACTION_SCHEMAS)) as readonly AutomationAction[];
export const WORKFLOW_ACTIONS: ReadonlySet<AutomationAction> = new Set(AUTOMATION_ACTIONS);
export function isAutomationAction(action: string): action is AutomationAction {
return WORKFLOW_ACTIONS.has(action as AutomationAction);
}
export type AutomationActionConfigMap = {
[K in keyof typeof ACTION_SCHEMAS]: z.infer<typeof ACTION_SCHEMAS[K]>;
+20 -364
View File
@@ -1,374 +1,30 @@
import type { AppDatabase } from "../database";
import type { AutomationAction, DesktopEvent } from "@shared/types";
import { dispatchAgentMessage } from "../messaging/dispatch";
import { validateTaskCreation } from "../task-quality-gates";
import type { AutomationAction } from "@shared/types";
import { AUTOMATION_ACTION_HANDLERS } from "./automation-action-handlers";
import { isAutomationAction, parseActionConfig } from "./automation-action-schemas";
import type {
ActionCallbacks,
ActionDeps,
AutomationRuleExecution,
AutomationRuleInput,
} from "./automation-action-runtime";
import type { AutomationEventContext } from "./types";
import { parseActionConfig, type AutomationActionConfigMap } from "./automation-action-schemas";
// Map trigger type → default wake trigger when automation fires a heartbeat.
const HEARTBEAT_WAKE_TRIGGER_MAP: Record<string, string> = {
run_completed: "subtask_completed",
run_failed: "report_failed",
hire_approved: "assignment",
agent_idle: "timer",
};
function deriveHeartbeatWakeTrigger(triggerType: string, previousStatus: string | null): string {
if (triggerType === "task_status_changed" && previousStatus === "in_review") return "comment";
return HEARTBEAT_WAKE_TRIGGER_MAP[triggerType] ?? "manual";
}
// Map the run/state combination → wake trigger for escalation to manager.
// Status-based mapping takes precedence; rule-trigger mapping is the fallback.
const ESCALATION_STATUS_TRIGGER_MAP: Record<string, string> = {
in_review: "subtask_completed",
blocked: "report_blocked",
};
const ESCALATION_RULE_TRIGGER_MAP: Record<string, string> = {
run_completed: "subtask_completed",
run_failed: "report_failed",
};
function deriveEscalationWakeTrigger(newStatus: string | null, ruleTrigger: string | null): string {
if (newStatus && ESCALATION_STATUS_TRIGGER_MAP[newStatus]) return ESCALATION_STATUS_TRIGGER_MAP[newStatus];
if (ruleTrigger && ESCALATION_RULE_TRIGGER_MAP[ruleTrigger]) return ESCALATION_RULE_TRIGGER_MAP[ruleTrigger];
return "report_blocked";
}
export interface ActionDeps {
db: AppDatabase;
logger: {
info: (...args: unknown[]) => void;
warn: (...args: unknown[]) => void;
error: (...args: unknown[]) => void;
};
emitEvent: (event: DesktopEvent) => void;
publishDomainChanged: () => void;
wakeAgentIfPossible: (agentId: string | null | undefined, companyId: string, trigger: string) => string | null;
}
export interface ActionCallbacks {
handleTaskCreated: (taskId: string, companyId: string) => void;
handleApprovalResolved: (approvalId: string, companyId: string, decision: string) => void;
handleTaskStatusChange: (taskId: string, companyId: string, previousStatus: string, newStatus: string) => void;
}
export { AUTOMATION_ACTION_HANDLERS } from "./automation-action-handlers";
export type { ActionCallbacks, ActionDeps } from "./automation-action-runtime";
export function executeAutomationAction(
deps: ActionDeps,
callbacks: ActionCallbacks,
rule: {
id?: string;
name?: string;
action: string;
actionConfigJson: string;
targetDepartment: string | null;
trigger?: string;
},
rule: AutomationRuleInput,
ctx: AutomationEventContext,
): void {
const { db, logger, emitEvent, publishDomainChanged, wakeAgentIfPossible } = deps;
const { handleTaskCreated, handleApprovalResolved, handleTaskStatusChange } = callbacks;
const action = parseAutomationAction(rule.action);
const executionRule: AutomationRuleExecution = { ...rule, action };
const config = parseActionConfig(action, rule.actionConfigJson);
AUTOMATION_ACTION_HANDLERS[action]({ deps, callbacks, rule: executionRule, ctx, config });
}
const action = rule.action as AutomationAction;
// Validate config against the per-action schema; schema failures surface
// as thrown errors so the caller (evaluateAutomationRules) records + logs.
const config = parseActionConfig(action, rule.actionConfigJson) as AutomationActionConfigMap[typeof action] & Record<string, unknown>;
switch (action) {
case "assign_task": {
if (ctx.taskId) {
const assignedId = db.tasks.autoAssignTask(ctx.taskId, ctx.companyId);
if (assignedId) {
wakeAgentIfPossible(assignedId, ctx.companyId, "assignment");
}
} else if (ctx.agentId) {
const snapshot = db.snap.listSnapshot();
const unassigned = snapshot.tasks
.filter((task) => task.companyId === ctx.companyId && !task.assigneeAgentId && ["backlog", "todo"].includes(task.status))
.sort((a, b) => {
const priorityScore = { critical: 4, high: 3, medium: 2, low: 1 } as Record<string, number>;
return (priorityScore[b.priority] ?? 0) - (priorityScore[a.priority] ?? 0);
});
if (unassigned.length > 0) {
const task = unassigned[0]!;
db.tasks.saveTask({
id: task.id,
companyId: task.companyId,
projectId: task.projectId,
goalId: task.goalId,
parentId: task.parentId,
title: task.title,
description: task.description,
assigneeAgentId: ctx.agentId,
workspaceId: task.workspaceId,
requiresFilesystem: task.requiresFilesystem,
priority: task.priority,
status: "todo",
});
wakeAgentIfPossible(ctx.agentId, ctx.companyId, "assignment");
logger.info(`[automation] Auto-assigned task "${task.title}" to idle agent ${ctx.agentId}`);
}
}
break;
}
case "create_task": {
const cfg = config as AutomationActionConfigMap["create_task"];
const title = cfg.title ?? "[Auto] Task from automation rule";
const description = cfg.description ?? "";
const priority = cfg.priority ?? "medium";
const assigneeAgentId = cfg.assigneeAgentId ?? null;
const taskValidation = validateTaskCreation({ title, description });
if (!taskValidation.valid) {
logger.warn(`[automation] Skipping create_task — validation failed: ${taskValidation.errors.join("; ")}`);
break;
}
const taskId = db.tasks.saveTask({
companyId: ctx.companyId,
projectId: ctx.projectId ?? cfg.projectId ?? null,
title,
description,
assigneeAgentId,
priority,
status: assigneeAgentId ? "todo" : "backlog",
requiresFilesystem: cfg.requiresFilesystem,
});
handleTaskCreated(taskId, ctx.companyId);
logger.info(`[automation] Created task: ${title} (${taskId})`);
break;
}
case "notify_agent":
case "send_message": {
const cfg = config as AutomationActionConfigMap["notify_agent"];
const toAgentId = cfg.toAgentId ?? ctx.agentId ?? null;
const fromAgentId = cfg.fromAgentId ?? ctx.agentId ?? toAgentId;
if (!fromAgentId) break;
const defaultChannel = toAgentId ? "direct" : "company";
const requestedChannel = cfg.channel ?? defaultChannel;
const channel = (["direct", "department", "company", "project", "incident"] as const).includes(requestedChannel)
? requestedChannel
: defaultChannel;
let channelTargetId: string | null = cfg.channelTargetId ?? cfg.department ?? ctx.department ?? null;
if (channel === "department" && !channelTargetId && fromAgentId) {
try {
const sender = db.entityCrud.getAgent(fromAgentId);
channelTargetId = sender.department ?? null;
} catch (err) {
logger.warn(`[automation] getAgent failed for sender ${fromAgentId}; dropping department target:`, err instanceof Error ? err.message : String(err));
channelTargetId = null;
}
}
// `dispatchAgentMessage` rejects department / project messages
// that arrive without a channel target. The default rule
// "Notify department on task completion" is agent-agnostic
// (the operator may not have a department per agent), so a
// single-person company or any agent without a department
// routinely lands here. Treat the missing target as "no
// recipients available" and skip — the previous behaviour
// logged a noisy WARN that looked like a real failure.
const channelNeedsTarget = channel === "department" || channel === "project";
if (channelNeedsTarget && !channelTargetId) {
logger.info(
`[automation] Skipping rule "${rule.name}" — no ${channel} channel target available for sender ${fromAgentId}`,
);
break;
}
const dispatchResult = dispatchAgentMessage(
{ db, emitEvent, publishDomainChanged, wakeAgentIfPossible },
{
companyId: ctx.companyId,
fromAgentId,
toAgentId: channel === "direct" ? toAgentId : null,
channel: channel as "direct" | "department" | "company" | "project" | "incident",
channelTargetId,
subject: (config.subject as string) ?? "Automation notification",
body: (config.body as string) ?? `Automation rule triggered for task ${ctx.taskId ?? "n/a"}`,
priority: (config.priority as "urgent" | "normal" | "low") ?? "normal",
},
"message",
);
if (!dispatchResult.ok) {
logger.warn(
`[automation] Failed to dispatch message for rule ${rule.name} (${rule.id}): ${dispatchResult.error.code} ${dispatchResult.error.message}`,
);
}
break;
}
case "trigger_heartbeat": {
let targetAgentId = (config.agentId as string) ?? null;
if (!targetAgentId && ctx.agentId) {
const triggerType = rule.trigger ?? "";
if (triggerType === "run_completed" || triggerType === "run_failed") {
const manager = db.agents.getManagerForAgent(ctx.agentId);
targetAgentId = manager?.id ?? ctx.agentId;
} else {
targetAgentId = ctx.agentId;
}
}
if (targetAgentId) {
wakeAgentIfPossible(targetAgentId, ctx.companyId, deriveHeartbeatWakeTrigger(rule.trigger ?? "", ctx.previousStatus ?? null));
}
break;
}
case "create_approval": {
const approvalType = (config.approvalType as string) ?? "document_review";
db.approvalsMod.requestApproval({
companyId: ctx.companyId,
relatedTaskId: ctx.taskId ?? null,
requestedByAgentId: ctx.agentId ?? null,
type: approvalType,
payloadSummary: (config.summary as string) ?? "Auto-generated approval request",
impactSummary: (config.impact as string) ?? "Triggered by automation rule",
});
break;
}
case "update_task_status": {
if (!ctx.taskId) break;
const newStatus = (config.status as string) ?? "todo";
const previousTask = db.entityCrud.getTask(ctx.taskId);
db.tasks.advanceTaskStatus(ctx.taskId, newStatus, "Status changed by automation rule");
publishDomainChanged();
if (previousTask.status !== newStatus && rule.trigger !== "task_status_changed") {
handleTaskStatusChange(ctx.taskId, ctx.companyId, previousTask.status, newStatus);
}
break;
}
case "escalate_to_manager": {
const targetAgentId = ctx.agentId;
if (!targetAgentId) break;
let agentName = "an agent";
try {
agentName = db.entityCrud.getAgent(targetAgentId).name;
} catch (err) {
logger.warn(`[automation] getAgent failed for escalation target ${targetAgentId}; using generic label:`, err instanceof Error ? err.message : String(err));
}
let taskTitle = ctx.taskId ?? "unknown";
try {
if (ctx.taskId) taskTitle = db.entityCrud.getTask(ctx.taskId).title;
} catch (err) {
logger.warn(`[automation] getTask failed for escalation task ${ctx.taskId}; using id as title:`, err instanceof Error ? err.message : String(err));
}
const manager = db.agents.getManagerForAgent(targetAgentId);
const escalationBody = (config.body as string) ?? `Task "${taskTitle}" needs your attention. ${agentName} has flagged this as requiring manager intervention.`;
const wakeTrigger = deriveEscalationWakeTrigger(ctx.newStatus ?? null, rule.trigger ?? null);
if (manager) {
const escalationResult = dispatchAgentMessage(
{ db, emitEvent, publishDomainChanged, wakeAgentIfPossible },
{
companyId: ctx.companyId,
fromAgentId: targetAgentId,
toAgentId: manager.id,
channel: "direct",
subject: `Escalation from ${agentName}: ${taskTitle}`,
body: escalationBody,
priority: "urgent",
},
wakeTrigger,
);
if (!escalationResult.ok) {
logger.warn(
`[automation] Failed to escalate "${taskTitle}" to manager: ${escalationResult.error.code} ${escalationResult.error.message}`,
);
}
} else {
// No manager found — agent is likely the CEO or top-level.
// Skip posting to company channel to avoid flooding it with system noise.
// The task status change is already visible in the task list and activity log.
logger.info(`[automation] Skipping escalation for top-level agent "${agentName}" — no manager to escalate to.`);
}
break;
}
case "cross_department_notify": {
const targetDepartment = rule.targetDepartment ?? (config.department as string);
if (!targetDepartment) break;
const departmentAgents = db.agents.getIdleAgentsForDepartment(ctx.companyId, targetDepartment);
const senderAgentId = ctx.agentId ?? departmentAgents[0]?.id;
if (senderAgentId) {
const dispatchResult = dispatchAgentMessage(
{ db, emitEvent, publishDomainChanged, wakeAgentIfPossible },
{
companyId: ctx.companyId,
fromAgentId: senderAgentId,
toAgentId: null,
channel: "department",
channelTargetId: targetDepartment,
subject: (config.subject as string) ?? "Cross-department notification",
body: (config.body as string) ?? "Automation triggered a cross-department alert.",
priority: "normal",
},
"message",
);
if (!dispatchResult.ok) {
logger.warn(
`[automation] Failed cross-department notification for ${targetDepartment}: ${dispatchResult.error.code} ${dispatchResult.error.message}`,
);
}
}
break;
}
case "schedule_meeting": {
const cfg = config as AutomationActionConfigMap["schedule_meeting"];
const meetingType = cfg.meetingType ?? "department_sync";
const title = cfg.title ?? "Auto-scheduled meeting";
// Prefer an absolute scheduledAt; otherwise honor scheduleOffsetMinutes.
// Default to +60min only when the rule specifies no time at all.
const offsetMinutes = cfg.scheduleOffsetMinutes ?? 60;
const scheduledAt = cfg.scheduledAt ?? new Date(Date.now() + offsetMinutes * 60_000).toISOString();
db.knowledgeMod.saveMeeting({
companyId: ctx.companyId,
type: meetingType,
title,
organizerAgentId: ctx.agentId ?? null,
participantAgentIds: cfg.participantAgentIds ?? "[]",
scheduledAt,
durationMinutes: cfg.durationMinutes ?? 30,
});
break;
}
case "create_document": {
const documentType = (config.documentType as string) ?? "status_report";
const title = (config.title as string) ?? "Auto-generated document";
db.documentsMod.saveDocument({
companyId: ctx.companyId,
type: documentType,
title,
content: (config.content as string) ?? "",
authorAgentId: ctx.agentId ?? null,
projectId: ctx.projectId ?? null,
});
break;
}
case "auto_approve": {
const snapshot = db.snap.listSnapshot();
const pendingApprovals = snapshot.approvals.filter(
(approval) => approval.companyId === ctx.companyId && approval.state === "pending",
);
for (const approval of pendingApprovals.slice(0, 5)) {
db.approvalsMod.decideApproval({
companyId: ctx.companyId,
approvalId: approval.id,
state: "approved",
decisionNote: "Auto-approved by automation rule.",
});
handleApprovalResolved(approval.id, ctx.companyId, "approved");
}
break;
}
case "reassign_task": {
if (!ctx.taskId) break;
const newAssignee = db.tasks.autoAssignTask(ctx.taskId, ctx.companyId);
if (newAssignee) {
wakeAgentIfPossible(newAssignee, ctx.companyId, "assignment");
}
break;
}
case "run_task": {
const targetTaskId = (config.taskId as string) ?? ctx.taskId;
if (!targetTaskId) break;
const task = db.entityCrud.getTask(targetTaskId);
if (!task.assigneeAgentId) break;
wakeAgentIfPossible(task.assigneeAgentId, ctx.companyId, "assignment");
break;
}
default:
throw new Error(`Unknown automation action: ${rule.action}`);
}
function parseAutomationAction(action: string): AutomationAction {
if (isAutomationAction(action)) return action;
throw new Error(`Unknown automation action: ${action}`);
}
+4 -3
View File
@@ -1,9 +1,10 @@
import { z } from "zod";
import type { AppDatabase } from "../database";
import type { AgentRecord } from "@shared/types";
import { WORKFLOW_ACTIONS, type WorkflowStepState, type AutomationEventContext } from "./types";
import type { WorkflowStepState, AutomationEventContext } from "./types";
import type { ActionDeps, ActionCallbacks } from "./automation-actions";
import { executeAutomationAction } from "./automation-actions";
import { isAutomationAction } from "./automation-action-schemas";
// Raw workflow step shape accepted from the DB `stepsJson` column. All fields
// are optional/loose because legacy rows used free-form typeof checks; zod
@@ -36,13 +37,13 @@ export function normalizeWorkflowSteps(raw: unknown): WorkflowStepState[] {
const parsed = workflowStepsInputSchema.parse(raw);
return parsed.map((step, index) => {
const rawAction = step.action ?? "send_message";
if (!WORKFLOW_ACTIONS.has(rawAction as WorkflowStepState["action"])) {
if (!isAutomationAction(rawAction)) {
throw new Error(`Workflow step ${index + 1} has unsupported action "${rawAction}".`);
}
return {
id: step.id && step.id.length > 0 ? step.id : `step-${index + 1}`,
name: step.name && step.name.length > 0 ? step.name : `Step ${index + 1}`,
action: rawAction as WorkflowStepState["action"],
action: rawAction,
configJson: step.configJson ?? "{}",
assigneeDepartment: step.assigneeDepartment ?? null,
assigneeAgentId: step.assigneeAgentId ?? null,
-17
View File
@@ -12,23 +12,6 @@ export const IMPORTANT_DOCUMENT_TYPES = new Set([
"budget_proposal", "architecture_decision", "contract", "sop",
]);
export const WORKFLOW_ACTIONS: ReadonlySet<AutomationAction> = new Set([
"assign_task",
"create_task",
"notify_agent",
"send_message",
"trigger_heartbeat",
"create_approval",
"update_task_status",
"escalate_to_manager",
"cross_department_notify",
"schedule_meeting",
"create_document",
"auto_approve",
"reassign_task",
"run_task",
]);
export interface WorkflowStepState {
id: string;
name: string;
@@ -0,0 +1,48 @@
import { readFileSync } from "node:fs";
import { join } from "node:path";
import { describe, expect, it } from "vitest";
import { AUTOMATION_ACTION_HANDLERS } from "../../src/main/automation/automation-actions";
import {
ACTION_SCHEMAS,
AUTOMATION_ACTIONS,
WORKFLOW_ACTIONS,
isAutomationAction,
parseActionConfig,
} from "../../src/main/automation/automation-action-schemas";
import { normalizeWorkflowSteps } from "../../src/main/automation/automation-workflow";
describe("automation action registry", () => {
it("drives schemas, workflow validation, and execution handlers from one action set", () => {
const actionNames = [...AUTOMATION_ACTIONS].sort();
expect([...WORKFLOW_ACTIONS].sort()).toEqual(actionNames);
expect(Object.keys(ACTION_SCHEMAS).sort()).toEqual(actionNames);
expect(Object.keys(AUTOMATION_ACTION_HANDLERS).sort()).toEqual(actionNames);
for (const action of AUTOMATION_ACTIONS) {
expect(isAutomationAction(action)).toBe(true);
expect(AUTOMATION_ACTION_HANDLERS[action]).toEqual(expect.any(Function));
expect(() => parseActionConfig(action, "{}")).not.toThrow();
expect(normalizeWorkflowSteps([{ action, configJson: "{}" }])[0]).toMatchObject({
action,
status: "pending",
});
}
});
it("rejects unsupported workflow actions before execution", () => {
expect(isAutomationAction("unknown_action")).toBe(false);
expect(() => normalizeWorkflowSteps([{ action: "unknown_action" }])).toThrow(
'Workflow step 1 has unsupported action "unknown_action".',
);
});
it("keeps automation execution table-driven", () => {
const dispatchSource = readFileSync(join(process.cwd(), "src/main/automation/automation-actions.ts"), "utf8");
const handlerSource = readFileSync(join(process.cwd(), "src/main/automation/automation-action-handlers.ts"), "utf8");
expect(dispatchSource).not.toContain("switch (action)");
expect(handlerSource).not.toContain("switch (action)");
expect(dispatchSource).toContain("AUTOMATION_ACTION_HANDLERS");
expect(handlerSource).toContain("satisfies Record<AutomationAction, AutomationActionHandler>");
});
});