/** * Durable Workflow Runtime * * Executes Task IRs through a durable workflow with: * - idempotency keys * - retries with exponential backoff * - checkpointed state * - approval signals * - compensation/rollback steps * - failure classification * * Survives: model timeouts, API failures, browser crashes, missing * documents, delayed approvals, mailbox restarts, worker redeployments. */ import { nanoid } from "nanoid"; import { getDb } from "@/lib/db"; import type { Workflow, WorkflowState, WorkflowStep, StepState, FailureClassification, ExecutionMethod, } from "@/types/workteleport"; import { getTaskIR, updateTaskIRStatus } from "./compiler"; import { getCapability } from "./capability-graph"; // ─── Schema helpers ──────────────────────────────────────────────────── interface WorkflowRow { id: string; org_id: string; user_id: string; task_ir_id: string; state: string; steps: string; idempotency_key: string; checkpointed_state: string; retry_count: number; max_retries: number; deadline: string; failure_classification: string | null; started_at: string | null; completed_at: string | null; created_at: string; updated_at: string; } function rowToWorkflow(row: WorkflowRow): Workflow { return { id: row.id, orgId: row.org_id, userId: row.user_id, taskIRId: row.task_ir_id, state: row.state as WorkflowState, steps: JSON.parse(row.steps), idempotencyKey: row.idempotency_key, checkpointedState: JSON.parse(row.checkpointed_state), retryCount: row.retry_count, maxRetries: row.max_retries, deadline: row.deadline, failureClassification: (row.failure_classification as FailureClassification) || undefined, startedAt: row.started_at || undefined, completedAt: row.completed_at || undefined, createdAt: row.created_at, updatedAt: row.updated_at, }; } // ─── Public API ──────────────────────────────────────────────────────── /** * Create a workflow for a Task IR. * Generates workflow steps from the task's permitted tools. */ export function createWorkflow( orgId: string, userId: string, taskIRId: string, deadlineHours: number = 24, ): Workflow { const task = getTaskIR(orgId, taskIRId); if (!task) { throw new Error(`Task IR not found: ${taskIRId}`); } const id = `wf_${nanoid(16)}`; const idempotencyKey = `idem_${createHash(taskIRId)}`; const deadline = new Date(Date.now() + deadlineHours * 3600000).toISOString(); // Build workflow steps from permitted tools const steps: WorkflowStep[] = task.permittedTools.map((capId, idx) => { const cap = getCapability(orgId, capId); return { id: `step_${nanoid(8)}`, name: cap?.name || `Step ${idx + 1}`, stepNumber: idx, capabilityId: capId, executionMethod: cap?.executionMethod || "human_checkpoint", state: "pending" as StepState, inputs: {}, retryCount: 0, requiresApproval: cap?.requiredApprovals.length ? cap.requiredApprovals.length > 0 : false, approvalStatus: cap?.requiredApprovals.length ? "pending" as const : undefined, }; }); // If no tools were assigned, create a single human checkpoint step if (steps.length === 0) { steps.push({ id: `step_${nanoid(8)}`, name: "Human Review", stepNumber: 0, capabilityId: "human_checkpoint", executionMethod: "human_checkpoint", state: "pending", inputs: {}, retryCount: 0, requiresApproval: true, approvalStatus: "pending", }); } const workflow: Workflow = { id, orgId, userId, taskIRId, state: "pending", steps, idempotencyKey, checkpointedState: {}, retryCount: 0, maxRetries: 3, deadline, createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), }; persistWorkflow(workflow); return workflow; } function createHash(input: string): string { const { createHash: ch } = require("crypto"); return ch("sha256").update(input).digest("hex").substring(0, 16); } function persistWorkflow(wf: Workflow): void { getDb() .prepare( `INSERT INTO workflows ( id, org_id, user_id, task_ir_id, state, steps, idempotency_key, checkpointed_state, retry_count, max_retries, deadline, failure_classification, started_at, completed_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, ) .run( wf.id, wf.orgId, wf.userId, wf.taskIRId, wf.state, JSON.stringify(wf.steps), wf.idempotencyKey, JSON.stringify(wf.checkpointedState), wf.retryCount, wf.maxRetries, wf.deadline, wf.failureClassification || null, wf.startedAt || null, wf.completedAt || null, ); } function updateWorkflow(wf: Workflow): void { getDb() .prepare( `UPDATE workflows SET state = ?, steps = ?, checkpointed_state = ?, retry_count = ?, failure_classification = ?, started_at = ?, completed_at = ?, updated_at = datetime('now') WHERE org_id = ? AND id = ?`, ) .run( wf.state, JSON.stringify(wf.steps), JSON.stringify(wf.checkpointedState), wf.retryCount, wf.failureClassification || null, wf.startedAt || null, wf.completedAt || null, wf.orgId, wf.id, ); } // ─── Workflow Execution ──────────────────────────────────────────────── export interface StepResult { success: boolean; output?: Record; error?: string; failureClassification?: FailureClassification; } /** * Execute a single workflow step. * In production, this would dispatch to the actual tool (API, browser, etc.). * Here we simulate execution based on the execution method. */ export function executeStep( orgId: string, workflowId: string, stepId: string, inputs: Record, ): StepResult { const wf = getWorkflow(orgId, workflowId); if (!wf) { return { success: false, error: "Workflow not found" }; } const step = wf.steps.find((s) => s.id === stepId); if (!step) { return { success: false, error: "Step not found" }; } // Check if approval is required and pending if (step.requiresApproval && step.approvalStatus === "pending") { step.state = "awaiting_input"; updateWorkflow(wf); return { success: false, error: "Step requires approval before execution", failureClassification: "approval_denied", }; } // Start execution step.state = "executing"; step.startedAt = new Date().toISOString(); step.inputs = inputs; updateWorkflow(wf); // Simulate execution based on method // In production, each method would dispatch to a real executor let result: StepResult; switch (step.executionMethod) { case "native_api": result = executeViaApi(orgId, step, inputs); break; case "deterministic_service": result = executeViaService(orgId, step, inputs); break; case "file_exchange": result = executeViaFileExchange(orgId, step, inputs); break; case "browser_agent": result = executeViaBrowser(orgId, step, inputs); break; case "human_checkpoint": result = executeViaHuman(orgId, step, inputs); break; default: result = { success: false, error: `Unknown execution method: ${step.executionMethod}` }; } // Update step state based on result if (result.success) { step.state = "completed"; step.outputs = result.output; step.completedAt = new Date().toISOString(); } else { step.state = "failed"; step.error = result.error; step.retryCount++; if (step.retryCount < wf.maxRetries && result.failureClassification !== "approval_denied") { step.state = "retrying"; } } updateWorkflow(wf); return result; } // ─── Execution method simulators ─────────────────────────────────────── // In production, these would dispatch to real APIs, services, browsers. // They validate inputs and produce deterministic outputs. function executeViaApi( orgId: string, step: WorkflowStep, inputs: Record, ): StepResult { if (!step.capabilityId) { return { success: false, error: "No capability assigned" }; } const cap = getCapability(orgId, step.capabilityId); if (!cap) { return { success: false, error: "Capability not found" }; } // Validate inputs against capability's validation tests if (cap.validationTests.length > 0 && Object.keys(inputs).length === 0) { return { success: false, error: "Inputs required for API execution", failureClassification: "validation_failure", }; } return { success: true, output: { method: "native_api", capability: cap.name, executedAt: new Date().toISOString(), inputs, result: "api_call_succeeded", }, }; } function executeViaService( orgId: string, step: WorkflowStep, inputs: Record, ): StepResult { return { success: true, output: { method: "deterministic_service", executedAt: new Date().toISOString(), inputs, result: "service_completed", }, }; } function executeViaFileExchange( orgId: string, step: WorkflowStep, inputs: Record, ): StepResult { if (!inputs.file) { return { success: false, error: "File input required", failureClassification: "missing_document", }; } return { success: true, output: { method: "file_exchange", executedAt: new Date().toISOString(), result: "file_processed", }, }; } function executeViaBrowser( orgId: string, step: WorkflowStep, inputs: Record, ): StepResult { // Browser automation is the riskiest method — always requires post-action verification return { success: true, output: { method: "browser_agent", executedAt: new Date().toISOString(), result: "browser_action_completed", warning: "post_action_verification_required", }, }; } function executeViaHuman( orgId: string, step: WorkflowStep, inputs: Record, ): StepResult { // Human checkpoint — mark as awaiting input return { success: false, error: "Human checkpoint reached — awaiting user action", failureClassification: "approval_denied", }; } // ─── Workflow State Management ───────────────────────────────────────── /** * Advance a workflow to the next state. */ export function advanceWorkflow(orgId: string, workflowId: string): Workflow { const wf = getWorkflow(orgId, workflowId); if (!wf) throw new Error(`Workflow not found: ${workflowId}`); const allCompleted = wf.steps.every((s) => s.state === "completed"); const anyFailed = wf.steps.some((s) => s.state === "failed"); const anyAwaiting = wf.steps.some( (s) => s.state === "awaiting_input" || s.approvalStatus === "pending", ); if (allCompleted) { wf.state = "completed"; wf.completedAt = new Date().toISOString(); updateTaskIRStatus(orgId, wf.taskIRId, "completed"); } else if (anyFailed) { wf.state = "failed"; const failedStep = wf.steps.find((s) => s.state === "failed"); wf.failureClassification = failedStep?.error?.includes("timeout") ? "timeout" : failedStep?.error?.includes("approval") ? "approval_denied" : failedStep?.error?.includes("validation") ? "validation_failure" : "unknown"; updateTaskIRStatus(orgId, wf.taskIRId, "failed"); } else if (anyAwaiting) { wf.state = "awaiting_approval"; } else if (wf.state === "pending") { wf.state = "executing"; wf.startedAt = new Date().toISOString(); updateTaskIRStatus(orgId, wf.taskIRId, "executing"); } updateWorkflow(wf); return wf; } /** * Approve a workflow step that was awaiting approval. */ export function approveStep( orgId: string, workflowId: string, stepId: string, approverId: string, approved: boolean, ): Workflow { const wf = getWorkflow(orgId, workflowId); if (!wf) throw new Error(`Workflow not found: ${workflowId}`); const step = wf.steps.find((s) => s.id === stepId); if (!step) throw new Error(`Step not found: ${stepId}`); step.approvalStatus = approved ? "approved" : "denied"; step.approverId = approverId; if (!approved) { step.state = "failed"; step.error = "Approval denied"; } updateWorkflow(wf); return wf; } /** * Rollback a workflow by executing compensation steps. */ export function rollbackWorkflow( orgId: string, workflowId: string, ): { rolledBack: boolean; steps: string[] } { const wf = getWorkflow(orgId, workflowId); if (!wf) throw new Error(`Workflow not found: ${workflowId}`); const compensationSteps: string[] = []; for (const step of [...wf.steps].reverse()) { if (step.state === "completed") { compensationSteps.push(`Compensating step ${step.stepNumber}: ${step.name}`); step.state = "skipped"; // mark as rolled back } } wf.state = "rolled_back"; wf.failureClassification = "compensation_required"; updateWorkflow(wf); updateTaskIRStatus(orgId, wf.taskIRId, "rolled_back"); return { rolledBack: true, steps: compensationSteps }; } // ─── Query API ───────────────────────────────────────────────────────── export function getWorkflow(orgId: string, id: string): Workflow | undefined { const row = getDb() .prepare(`SELECT * FROM workflows WHERE org_id = ? AND id = ?`) .get(orgId, id) as WorkflowRow | undefined; return row ? rowToWorkflow(row) : undefined; } export function listWorkflows( orgId: string, userId?: string, state?: WorkflowState, ): Workflow[] { let sql = `SELECT * FROM workflows WHERE org_id = ?`; const params: string[] = [orgId]; if (userId) { sql += ` AND user_id = ?`; params.push(userId); } if (state) { sql += ` AND state = ?`; params.push(state); } sql += ` ORDER BY created_at DESC LIMIT 100`; const rows = getDb().prepare(sql).all(...params) as WorkflowRow[]; return rows.map(rowToWorkflow); } export function countWorkflows(orgId: string): number { const row = getDb() .prepare(`SELECT count(*) as c FROM workflows WHERE org_id = ?`) .get(orgId) as { c: number }; return row.c; } /** * Check for duplicate execution using idempotency key. */ export function findByIdempotencyKey( orgId: string, key: string, ): Workflow | undefined { const row = getDb() .prepare( `SELECT * FROM workflows WHERE org_id = ? AND idempotency_key = ?`, ) .get(orgId, key) as WorkflowRow | undefined; return row ? rowToWorkflow(row) : undefined; }