engine.ts

// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.

import type { FlowParsed, FlowNodeParsed, FlowEdgeParsed } from '@objectstack/spec/automation';
import type { ExecutionLog, ActionDescriptor } from '@objectstack/spec/automation';
import type { AutomationContext, AutomationResult, ResumeSignal, IAutomationService, ScreenSpec } from '@objectstack/spec/contracts';
import type { Logger } from '@objectstack/spec/contracts';
import { FlowSchema, FLOW_STRUCTURAL_NODE_TYPES, validateControlFlow, findRegionEntry, defineActionDescriptor } from '@objectstack/spec/automation';
import type { FlowRegionParsed } from '@objectstack/spec/automation';
import type { Connector } from '@objectstack/spec/integration';
import { ConnectorSchema } from '@objectstack/spec/integration';
// Static import (not a lazy `require`): the engine ships as ESM ("type":"module"),
// where a CommonJS `require('@objectstack/formula')` resolves to tsup's throwing
// `__require` stub. That threw on every CEL evaluation and the catch below
// silently returned `false`, so EVERY start-node / edge condition (record-change
// `previous.*`, `budget > 100000`, …) skipped its flow. A static import binds the
// engine at module load in both ESM and CJS builds.
import { ExpressionEngine, validateExpression } from '@objectstack/formula';

// ─── Node Executor Interface (Plugin Extension Point) ───────────────

/**
 * Each node type corresponds to a NodeExecutor.
 * Third-party plugins only need to implement this interface and register
 * it with the engine to extend automation capabilities.
 */
export interface NodeExecutor {
    /** Registry node type (built-in id or plugin-defined) */
    readonly type: string;

    /**
     * Optional ADR-0018 action descriptor. When present, it is published into
     * the engine's action registry and surfaced via {@link AutomationEngine.getActionDescriptors}
     * — feeding flow validation and the designer palette. Plugins SHOULD publish
     * one so their node appears in the palette and validates as a legal flow node.
     */
    readonly descriptor?: ActionDescriptor;

    /**
     * Execute a node
     * @param node - Current node definition
     * @param variables - Flow variable context (read/write)
     * @param context - Trigger context
     * @returns Execution result (may include output data, branch conditions, etc.)
     */
    execute(
        node: FlowNodeParsed,
        variables: Map<string, unknown>,
        context: AutomationContext,
    ): Promise<NodeExecutionResult>;
}

export interface NodeExecutionResult {
    success: boolean;
    output?: Record<string, unknown>;
    error?: string;
    /** Used by decision nodes — returns the selected branch label */
    branchLabel?: string;
    /**
     * ADR-0019 durable pause. When `true`, the node has done its on-entry work
     * (e.g. opened an approval request) and the run should **suspend** here: the
     * engine persists a continuation, stops traversal, and `execute()` returns
     * `{ status: 'paused', runId }`. The run is continued later via
     * {@link AutomationEngine.resume}. Any `output` is written to variables
     * before suspending. The node reads its own run id from the `$runId`
     * flow variable so it can map the run to external state.
     */
    suspend?: boolean;
    /**
     * Optional correlation key surfaced on the suspended-run record (e.g. an
     * approval request id). For observability / lookup; not required to resume.
     */
    correlation?: string;
    /**
     * Screen to render — set by a `screen` node that suspends to collect input.
     * Surfaced on the paused {@link AutomationResult} so a UI runner can render
     * the form and `resume()` with the values.
     */
    screen?: ScreenSpec;
    /**
     * #1479: step logs produced inside the node's structured region(s). A
     * container node (`loop` / `parallel` / `try_catch`) collects the
     * {@link AutomationEngine.runRegion} return value(s) here; {@link AutomationEngine.executeNode}
     * appends them to the parent run log right after the container's own step,
     * so per-iteration / per-branch body steps surface in run observability.
     */
    childSteps?: StepLogEntry[];
}

// ─── Trigger Interface (Plugin Extension Point) ─────────────────────

/**
 * A normalized description of *what* fires a flow, derived by the engine from
 * the flow's `start` node and handed to the matching {@link FlowTrigger} when a
 * flow is activated. Concrete triggers (record-change, schedule, …) read the
 * fields they care about and ignore the rest.
 *
 * The engine — not the trigger — owns parsing the start node, so trigger
 * plugins stay decoupled from flow-definition internals (mirrors how
 * `connector_action` keeps connectors decoupled from node config).
 */
export interface FlowTriggerBinding {
    /** Flow this binding activates. */
    readonly flowName: string;
    /** record-change: the object whose mutations fire the flow. */
    readonly object?: string;
    /** record-change: the start node's `triggerType` (e.g. 'record-after-update'). */
    readonly event?: string;
    /**
     * Optional trigger predicate copied from the start node's `condition`. The
     * engine evaluates it before running the flow; triggers may ignore it.
     */
    readonly condition?: string | { dialect?: string; source?: string; ast?: unknown };
    /** schedule: cron/interval descriptor (parsed but not yet acted on here). */
    readonly schedule?: unknown;
    /** The raw start-node `config`, for trigger-specific fields not modeled above. */
    readonly config?: Record<string, unknown>;
}

/**
 * Trigger interface. Schedule/Event/API triggers are registered via plugins.
 *
 * The engine completes the wiring: when a flow whose start node maps to this
 * trigger's {@link type} is registered (or when this trigger is registered
 * after such flows already exist), the engine calls {@link start} with the
 * parsed {@link FlowTriggerBinding} and a `callback` that runs the flow. The
 * trigger subscribes to its event source (e.g. an ObjectQL lifecycle hook) and
 * invokes `callback(ctx)` when it fires. {@link stop} tears that subscription
 * down when the flow is unregistered/disabled or the trigger is removed.
 */
export interface FlowTrigger {
    readonly type: string;
    start(binding: FlowTriggerBinding, callback: (ctx: AutomationContext) => Promise<void>): void;
    stop(flowName: string): void;
}

// ─── Connector Registry (Plugin Extension Point) ────────────────────

/**
 * Context handed to a connector action handler. Carries the live flow variable
 * map and the trigger context so a handler can read prior-node output, plus a
 * logger. The platform ships the registry + the `connector_action` dispatch
 * node (baseline, ADR-0018 §Addendum); *concrete* connectors — `connector-rest`,
 * `connector-slack`, … — are plugins that register handlers here.
 */
export interface ConnectorActionContext {
    readonly variables: Map<string, unknown>;
    readonly automation: AutomationContext;
    readonly logger: Logger;
}

/**
 * A handler for one connector action. Receives the (already-resolved) input
 * mapped from the flow node and returns the action's output, which the
 * `connector_action` node writes back into flow variables.
 */
export type ConnectorActionHandler = (
    input: Record<string, unknown>,
    ctx: ConnectorActionContext,
) => Promise<Record<string, unknown>>;

/**
 * A connector registered on the engine: its validated {@link Connector}
 * definition plus the handler for each action it declares.
 */
export interface RegisteredConnector {
    readonly def: Connector;
    readonly handlers: Record<string, ConnectorActionHandler>;
}

/**
 * A designer-facing view of one connector action — identity + its JSON-Schema
 * input/output. The runtime handler is intentionally omitted; this is metadata.
 */
export interface ConnectorActionDescriptor {
    readonly key: string;
    readonly label: string;
    readonly description?: string;
    readonly inputSchema?: Record<string, unknown>;
    readonly outputSchema?: Record<string, unknown>;
}

/**
 * A designer-facing descriptor for a registered connector: its identity plus
 * the actions it exposes. Served by `GET /api/v1/automation/connectors` so the
 * flow designer can populate the `connector_action` node's connector → action
 * → input pickers (ADR-0018 §Addendum, ADR-0022). Mirrors `ActionDescriptor`'s
 * role for node types, but for the connector registry.
 */
export interface ConnectorDescriptor {
    readonly name: string;
    readonly label: string;
    readonly type: string;
    readonly description?: string;
    readonly icon?: string;
    readonly actions: ConnectorActionDescriptor[];
}

// ─── Core Automation Engine ─────────────────────────────────────────

/**
 * Execution step log entry. Part of a {@link SuspendedRun}'s persisted state, so
 * it survives serialization to a durable {@link SuspendedRunStore}.
 */
export interface StepLogEntry {
    nodeId: string;
    nodeType: string;
    nodeLabel?: string;
    status: 'success' | 'failure' | 'skipped';
    startedAt: string;
    completedAt?: string;
    durationMs?: number;
    error?: { code: string; message: string; stack?: string };
    /**
     * #1479: structured-region grouping. When a step ran inside a `loop` /
     * `parallel` / `try_catch` body region, these tag it with its **immediate**
     * container so run observability can distinguish per-iteration / per-branch
     * body steps from top-level ones. Set by {@link AutomationEngine.runRegion}
     * (innermost wins — never overwritten as steps bubble through nested regions).
     */
    parentNodeId?: string;
    /** Zero-based loop iteration or parallel branch index of the enclosing region. */
    iteration?: number;
    /** Which region kind the step ran in: `loop-body` | `parallel-branch` | `try` | `catch`. */
    regionKind?: string;
}

/**
 * Internal execution log entry — compatible with ExecutionLog from spec.
 */
interface ExecutionLogEntry {
    id: string;
    flowName: string;
    flowVersion?: number;
    status: ExecutionLog['status'];
    startedAt: string;
    completedAt?: string;
    durationMs?: number;
    trigger: { type: string; userId?: string; object?: string; recordId?: string };
    steps: StepLogEntry[];
    variables?: Record<string, unknown>;
    output?: unknown;
    error?: string;
}

/**
 * Internal sentinel thrown by {@link AutomationEngine.executeNode} when a node
 * signals `suspend`. It unwinds the synchronous DAG recursion up to
 * `execute()` / `resume()`, which converts it into a persisted continuation
 * rather than a failed run. (Not exported — callers see `status: 'paused'`.)
 *
 * NOTE: suspend is supported on the serial / main execution path. A node that
 * suspends inside a `Promise.all` parallel branch will unwind that branch, but
 * sibling parallel branches already in flight are not cancelled — durable
 * pause across parallel gateways is out of scope for ADR-0019 M1.
 */
class FlowSuspendSignal {
    readonly __flowSuspend = true as const;
    constructor(readonly nodeId: string, readonly correlation?: string, readonly screen?: ScreenSpec) {}
}

function isSuspendSignal(err: unknown): err is FlowSuspendSignal {
    return typeof err === 'object' && err !== null && (err as FlowSuspendSignal).__flowSuspend === true;
}

/**
 * A run paused at a node, awaiting {@link AutomationEngine.resume} (ADR-0019).
 *
 * Held in an in-memory hot cache and — when a {@link SuspendedRunStore} is
 * configured — mirrored to durable storage so the pause survives a process
 * restart. Every field is JSON-serializable (the engine's variable `Map` is
 * snapshotted as a plain object) so the whole record round-trips through a
 * store.
 */
export interface SuspendedRun {
    runId: string;
    flowName: string;
    flowVersion?: number;
    /** The node the run paused at; resume continues from its out-edges. */
    nodeId: string;
    /** Snapshot of the flow variable map at suspend time. */
    variables: Record<string, unknown>;
    steps: StepLogEntry[];
    context: AutomationContext;
    startedAt: string;
    startTime: number;
    correlation?: string;
    /** Screen the run paused on (screen-flow runtime), for re-fetch + UI render. */
    screen?: ScreenSpec;
}

/**
 * Pluggable durable store for suspended runs (ADR-0019). The engine persists a
 * {@link SuspendedRun} on suspend and deletes it on terminal completion; on
 * {@link AutomationEngine.resume} of a run not in the in-memory cache (e.g.
 * after a process restart) it rehydrates from here.
 *
 * The default is purely in-memory (no store); a host wires a DB-backed store
 * (`ObjectStoreSuspendedRunStore`, on `sys_automation_run`) for production /
 * serverless deployments where the process hibernates between suspend and
 * resume.
 */
export interface SuspendedRunStore {
    /** Persist (insert or replace) a suspended run. */
    save(run: SuspendedRun): Promise<void>;
    /** Load a suspended run by id, or `null` if not stored. */
    load(runId: string): Promise<SuspendedRun | null>;
    /** Remove a suspended run's durable record (idempotent). */
    delete(runId: string): Promise<void>;
    /** List all currently-stored suspended runs. */
    list(): Promise<SuspendedRun[]>;
}

export class AutomationEngine implements IAutomationService {
    /**
     * ADR-0044: maximum times a single node may be (re-)entered at the top
     * level of one run before the engine aborts it as a runaway back-edge
     * loop. Generous on purpose — the product guard (`maxRevisions`) sits
     * orders of magnitude lower.
     */
    static readonly MAX_NODE_REENTRIES = 100;

    private flows = new Map<string, FlowParsed>();
    private flowEnabled = new Map<string, boolean>();
    private flowVersionHistory = new Map<string, Array<{ version: number; definition: FlowParsed; createdAt: string }>>();
    private nodeExecutors = new Map<string, NodeExecutor>();
    private actionDescriptors = new Map<string, ActionDescriptor>();
    private triggers = new Map<string, FlowTrigger>();
    /**
     * Flows currently wired to a trigger, keyed by flow name → the trigger
     * `type` that owns the binding. Used to avoid double-binding and to know
     * which trigger to `stop()` when a flow is unregistered/disabled.
     */
    private boundFlowTriggers = new Map<string, string>();
    /** Connectors registered by integration plugins, keyed by connector name (ADR-0018 §Addendum). */
    private connectors = new Map<string, RegisteredConnector>();
    private executionLogs: ExecutionLogEntry[] = [];
    private maxLogSize = 1000;
    private logger: Logger;
    /**
     * Runs paused at a node, keyed by runId (ADR-0019). In-memory hot cache —
     * mirrored to {@link store} when one is configured, so a pause survives a
     * process restart. See {@link SuspendedRun}.
     */
    private suspendedRuns = new Map<string, SuspendedRun>();
    /**
     * Optional durable backing for {@link suspendedRuns}. When set, suspended
     * runs are persisted on suspend and rehydrated on resume after a restart;
     * when absent, behaviour is purely in-memory (the historical default).
     */
    private store?: SuspendedRunStore;
    /**
     * Run ids currently mid-resume — an in-process idempotency guard so a
     * duplicate `resume(runId)` can't re-enter and double-run side effects.
     */
    private resuming = new Set<string>();

    constructor(logger: Logger, store?: SuspendedRunStore) {
        this.logger = logger;
        this.store = store;
    }

    /**
     * Attach (or replace) the durable {@link SuspendedRunStore}. Used by the
     * service plugin to upgrade the engine to DB-backed persistence once the
     * ObjectQL engine is available (the engine is constructed earlier, during
     * `init`, before services are wired).
     */
    setSuspendedRunStore(store: SuspendedRunStore): void {
        this.store = store;
    }

    /**
     * Generate a process-unique run id. Includes a random component so ids do
     * not collide with runs persisted by a previous process lifetime (a plain
     * incrementing counter would reissue `run_1` after a restart, clashing with
     * a still-suspended durable run).
     */
    private nextRunId(): string {
        const g = globalThis as { crypto?: { randomUUID?: () => string } };
        const rand = g.crypto?.randomUUID
            ? g.crypto.randomUUID()
            : `${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`;
        return `run_${rand}`;
    }

    /**
     * Persist a suspended run to the in-memory cache and (best-effort) the
     * durable store. A store failure is logged but does not fail the run — the
     * in-memory copy still allows in-process resume; only cross-restart
     * durability is lost.
     */
    private async persistSuspendedRun(run: SuspendedRun): Promise<void> {
        this.suspendedRuns.set(run.runId, run);
        if (this.store) {
            try {
                await this.store.save(run);
            } catch (err) {
                this.logger.warn(
                    `[automation] failed to persist suspended run '${run.runId}' to durable store (kept in memory only): ${(err as Error).message}`,
                );
            }
        }
    }

    /**
     * Drop a suspended run from the in-memory cache and (best-effort) the
     * durable store. Called once the run is claimed for resume or reaches a
     * terminal state.
     */
    private async forgetSuspendedRun(runId: string): Promise<void> {
        this.suspendedRuns.delete(runId);
        if (this.store) {
            try {
                await this.store.delete(runId);
            } catch (err) {
                this.logger.warn(
                    `[automation] failed to delete suspended run '${runId}' from durable store: ${(err as Error).message}`,
                );
            }
        }
    }

    // ── Plugin Extension API ──────────────────────────────

    /** Register a node executor (called by plugins) */
    registerNodeExecutor(executor: NodeExecutor): void {
        if (this.nodeExecutors.has(executor.type)) {
            this.logger.warn(`Node executor '${executor.type}' replaced`);
        }
        this.nodeExecutors.set(executor.type, executor);

        // Publish the ADR-0018 action descriptor into the registry, so the
        // type validates as a legal flow node and appears in the designer
        // palette. A descriptor's `type` should match the executor's; we key
        // on the descriptor's `type` and warn on mismatch rather than silently
        // diverging.
        if (executor.descriptor) {
            const descriptorType = executor.descriptor.type;
            if (descriptorType !== executor.type) {
                this.logger.warn(
                    `Node executor '${executor.type}' publishes a descriptor for type '${descriptorType}' — registering under both.`,
                );
            }
            this.actionDescriptors.set(descriptorType, executor.descriptor);
        }

        this.logger.info(`Node executor registered: ${executor.type}`);
    }

    /**
     * Register a **deprecated alias** of a canonical node type (ADR-0018 M3).
     *
     * The alias is a real registered executor, so old saved flows whose nodes
     * use the alias type keep validating and running with no migration. At
     * execute time it delegates to the canonical executor (resolved live, so the
     * canonical may be registered before or after the alias), logging a one-time
     * deprecation warning. Its published descriptor is flagged `deprecated` +
     * `aliasOf` so the designer palette can hide or mark it while the canonical
     * type is the one offered for new authoring.
     *
     * This is how ADR-0018 collapses the five outbound verbs onto `http` /
     * `notify`: `http_request` / `http_call` / `webhook` become aliases of
     * `http`.
     */
    registerNodeAlias(
        alias: string,
        canonicalType: string,
        meta?: { name?: string; category?: ActionDescriptor['category']; paradigms?: ActionDescriptor['paradigms']; needsOutbox?: boolean },
    ): void {
        const engine = this;
        let warned = false;
        this.registerNodeExecutor({
            type: alias,
            descriptor: defineActionDescriptor({
                type: alias,
                version: '1.0.0',
                name: meta?.name ?? alias,
                description: `Deprecated alias of '${canonicalType}' (ADR-0018 M3). Author new flows with '${canonicalType}'.`,
                category: meta?.category ?? 'io',
                source: 'builtin',
                paradigms: meta?.paradigms ?? ['flow', 'approval'],
                supportsRetry: true,
                needsOutbox: meta?.needsOutbox ?? false,
                deprecated: true,
                aliasOf: canonicalType,
            }),
            async execute(node, variables, context) {
                if (!warned) {
                    warned = true;
                    engine.logger.warn(
                        `Node type '${alias}' is deprecated; use '${canonicalType}' (ADR-0018 M3). Existing flows keep running via the alias.`,
                    );
                }
                const target = engine.nodeExecutors.get(canonicalType);
                if (!target) {
                    return {
                        success: false,
                        error: `alias '${alias}' → '${canonicalType}': canonical executor not registered`,
                    };
                }
                return target.execute(node, variables, context);
            },
        });
        this.logger.info(`Node alias registered: ${alias} → ${canonicalType} (deprecated)`);
    }

    /** Unregister a node executor (hot-unplug) */
    unregisterNodeExecutor(type: string): void {
        const executor = this.nodeExecutors.get(type);
        this.nodeExecutors.delete(type);
        // Drop the published descriptor (keyed by descriptor.type, which may
        // differ from the executor type).
        this.actionDescriptors.delete(type);
        if (executor?.descriptor) {
            this.actionDescriptors.delete(executor.descriptor.type);
        }
        this.logger.info(`Node executor unregistered: ${type}`);
    }

    /** Register a trigger (called by plugins) */
    registerTrigger(trigger: FlowTrigger): void {
        this.triggers.set(trigger.type, trigger);
        this.logger.info(`Trigger registered: ${trigger.type}`);
        // A trigger may be registered *after* its flows (e.g. AutomationServicePlugin
        // pulls flows at start(); a trigger plugin wires up on kernel:ready, which
        // fires later). Activate any already-registered flow that maps to this type.
        for (const name of this.flows.keys()) {
            if (this.boundFlowTriggers.has(name)) continue;
            const resolved = this.resolveTriggerBinding(name);
            if (resolved?.triggerType === trigger.type) {
                this.activateFlowTrigger(name);
            }
        }
    }

    /** Unregister a trigger (hot-unplug) */
    unregisterTrigger(type: string): void {
        // Tear down every flow bound to this trigger before dropping it.
        for (const [name, boundType] of [...this.boundFlowTriggers]) {
            if (boundType !== type) continue;
            try {
                this.triggers.get(type)?.stop(name);
            } catch (err) {
                this.logger.warn(`Trigger '${type}' stop('${name}') failed: ${(err as Error).message}`);
            }
            this.boundFlowTriggers.delete(name);
        }
        this.triggers.delete(type);
        this.logger.info(`Trigger unregistered: ${type}`);
    }

    /**
     * Derive a flow's trigger binding from its `start` node, or `undefined` if
     * the flow has no auto-trigger (manual / screen). The convention —
     * established by the showcase flows — is that the start node carries the
     * trigger details in its `config`: `{ objectName, triggerType, condition }`
     * for record-change, or a `schedule` descriptor for time-based flows.
     */
    private resolveTriggerBinding(
        flowName: string,
    ): { triggerType: string; binding: FlowTriggerBinding } | undefined {
        const flow = this.flows.get(flowName);
        if (!flow) return undefined;
        const startNode = flow.nodes.find(n => n.type === 'start');
        const config = (startNode?.config ?? {}) as Record<string, unknown>;
        const triggerType = typeof config.triggerType === 'string' ? config.triggerType : undefined;

        if (triggerType && triggerType.startsWith('record-')) {
            return {
                triggerType: 'record_change',
                binding: {
                    flowName,
                    object: typeof config.objectName === 'string' ? config.objectName : undefined,
                    event: triggerType,
                    condition: (config.condition as FlowTriggerBinding['condition']) ?? undefined,
                    config,
                },
            };
        }

        if (config.schedule != null || flow.type === 'schedule') {
            return {
                triggerType: 'schedule',
                binding: { flowName, schedule: config.schedule, condition: (config.condition as FlowTriggerBinding['condition']) ?? undefined, config },
            };
        }

        // Inbound HTTP (ADR-0041 Tier 1): an `api` flow waits for an external
        // POST. The concrete trigger (`@objectstack/trigger-api`) mounts the
        // endpoint and enqueues; the binding's `config` carries the hook
        // details (`hookId`, `secret`) from the start node.
        if (flow.type === 'api' || triggerType === 'api') {
            return {
                triggerType: 'api',
                binding: { flowName, condition: (config.condition as FlowTriggerBinding['condition']) ?? undefined, config },
            };
        }

        return undefined;
    }

    /**
     * Bind a flow to its matching registered trigger (idempotent). No-op when
     * the flow has no trigger binding or no trigger is registered for its type
     * yet — {@link registerTrigger} re-attempts activation when one arrives.
     */
    private activateFlowTrigger(flowName: string): void {
        if (this.boundFlowTriggers.has(flowName)) return;
        const resolved = this.resolveTriggerBinding(flowName);
        if (!resolved) return;
        const trigger = this.triggers.get(resolved.triggerType);
        if (!trigger) return;
        try {
            trigger.start(resolved.binding, (ctx: AutomationContext) => this.execute(flowName, ctx).then(() => undefined));
            this.boundFlowTriggers.set(flowName, resolved.triggerType);
            this.logger.info(`Flow '${flowName}' bound to trigger '${resolved.triggerType}'`);
        } catch (err) {
            this.logger.warn(`Failed to bind flow '${flowName}' to trigger '${resolved.triggerType}': ${(err as Error).message}`);
        }
    }

    /** Unbind a flow from its trigger, if bound. */
    private deactivateFlowTrigger(flowName: string): void {
        const boundType = this.boundFlowTriggers.get(flowName);
        if (!boundType) return;
        try {
            this.triggers.get(boundType)?.stop(flowName);
        } catch (err) {
            this.logger.warn(`Trigger '${boundType}' stop('${flowName}') failed: ${(err as Error).message}`);
        }
        this.boundFlowTriggers.delete(flowName);
    }

    /** Active flow→trigger bindings (observability / tests). */
    getActiveTriggerBindings(): Array<{ flowName: string; triggerType: string }> {
        return [...this.boundFlowTriggers].map(([flowName, triggerType]) => ({ flowName, triggerType }));
    }

    /**
     * Register a connector (called by integration plugins, ADR-0018 §Addendum).
     * Validates the definition against {@link ConnectorSchema} and asserts every
     * declared action has a handler, so a half-wired connector fails loudly at
     * registration rather than silently at dispatch. Re-registering the same
     * name replaces (mirrors {@link registerNodeExecutor}).
     */
    registerConnector(def: Connector, handlers: Record<string, ConnectorActionHandler>): void {
        const parsed = ConnectorSchema.parse(def);
        for (const action of parsed.actions ?? []) {
            if (typeof handlers[action.key] !== 'function') {
                throw new Error(
                    `Connector '${parsed.name}': action '${action.key}' is declared but no handler was provided`,
                );
            }
        }
        if (this.connectors.has(parsed.name)) {
            this.logger.warn(`Connector '${parsed.name}' replaced`);
        }
        this.connectors.set(parsed.name, { def: parsed, handlers });
        this.logger.info(
            `Connector registered: ${parsed.name} (${Object.keys(handlers).length} action handlers)`,
        );
    }

    /** Unregister a connector (hot-unplug). */
    unregisterConnector(name: string): void {
        this.connectors.delete(name);
        this.logger.info(`Connector unregistered: ${name}`);
    }

    /**
     * Resolve the handler for a connector action, used by the baseline
     * `connector_action` node. Returns `undefined` when the connector or action
     * is not registered, so the node can fail the step with a clear error.
     */
    resolveConnectorAction(connectorId: string, actionId: string): ConnectorActionHandler | undefined {
        return this.connectors.get(connectorId)?.handlers[actionId];
    }

    /** Get all registered connector names. */
    getRegisteredConnectors(): string[] {
        return [...this.connectors.keys()];
    }

    /**
     * Get a designer-facing descriptor for every registered connector — its
     * identity plus the actions it exposes (input/output JSON Schema). Backs
     * `GET /api/v1/automation/connectors` so the designer can fill the
     * `connector_action` node's connector / action / input pickers (ADR-0022).
     * Handlers are omitted — they are runtime code, not metadata.
     */
    getConnectorDescriptors(): ConnectorDescriptor[] {
        return [...this.connectors.values()].map(({ def }) => ({
            name: def.name,
            label: def.label,
            type: def.type,
            description: def.description,
            icon: def.icon,
            actions: (def.actions ?? []).map((a) => ({
                key: a.key,
                label: a.label,
                description: a.description,
                inputSchema: a.inputSchema,
                outputSchema: a.outputSchema,
            })),
        }));
    }

    /** Get all registered node types */
    getRegisteredNodeTypes(): string[] {
        return [...this.nodeExecutors.keys()];
    }

    /**
     * Get all published action descriptors (ADR-0018). Backs both flow
     * validation and the designer palette (`GET /api/v1/automation/actions`).
     * Only executors that published a descriptor appear here.
     */
    getActionDescriptors(): ActionDescriptor[] {
        return [...this.actionDescriptors.values()];
    }

    /** Get the action descriptor for a single node type, if published. */
    getActionDescriptor(type: string): ActionDescriptor | undefined {
        return this.actionDescriptors.get(type);
    }

    /** Get all registered trigger types */
    getRegisteredTriggerTypes(): string[] {
        return [...this.triggers.keys()];
    }

    // ── IAutomationService Contract Implementation ────────

    registerFlow(name: string, definition: unknown): void {
        const parsed = FlowSchema.parse(definition);

        // DAG cycle detection
        this.detectCycles(parsed);

        // ADR-0031 — validate structured control-flow constructs (loop bodies,
        // parallel branches, try/catch regions) are well-formed (single-entry/
        // single-exit, acyclic). Reject the malformed before it can run.
        validateControlFlow(parsed);

        // ADR-0018 §M1 — validate node types against the live action registry.
        // The protocol no longer gates `type` with a closed enum; membership is
        // checked here instead. Soft-fail (warn, don't throw): a flow authored
        // against a plugin that is currently disabled should still register, and
        // executeNode() already throws NO_EXECUTOR at run time for unknown types.
        this.validateNodeTypes(name, parsed);

        // ADR-0032 §Decision 1a — parse-validate every predicate at registration,
        // so a malformed condition (e.g. the #1491 `{record.x}` template-brace-in-
        // CEL mistake) is a LOUD registration error with the offending source,
        // not a silent runtime `false`. Hard-fail: a broken predicate is never
        // safe to run.
        this.validateFlowExpressions(name, parsed);

        // Version history management
        const history = this.flowVersionHistory.get(name) ?? [];
        history.push({
            version: parsed.version,
            definition: parsed,
            createdAt: new Date().toISOString(),
        });
        this.flowVersionHistory.set(name, history);

        this.flows.set(name, parsed);
        if (!this.flowEnabled.has(name)) {
            this.flowEnabled.set(name, true);
        }
        this.logger.info(`Flow registered: ${name} (version ${parsed.version})`);

        // Re-bind in case the definition changed its trigger, then (re)activate.
        this.deactivateFlowTrigger(name);
        if (this.flowEnabled.get(name) !== false) {
            this.activateFlowTrigger(name);
        }
    }

    unregisterFlow(name: string): void {
        this.deactivateFlowTrigger(name);
        this.flows.delete(name);
        this.flowEnabled.delete(name);
        this.flowVersionHistory.delete(name);
        this.logger.info(`Flow unregistered: ${name}`);
    }

    async listFlows(): Promise<string[]> {
        return [...this.flows.keys()];
    }

    async getFlow(name: string): Promise<FlowParsed | null> {
        return this.flows.get(name) ?? null;
    }

    async toggleFlow(name: string, enabled: boolean): Promise<void> {
        if (!this.flows.has(name)) {
            throw new Error(`Flow '${name}' not found`);
        }
        this.flowEnabled.set(name, enabled);
        this.logger.info(`Flow '${name}' ${enabled ? 'enabled' : 'disabled'}`);
        // A disabled flow should stop receiving trigger events; a re-enabled one
        // should resume. execute() also guards disabled flows, but unbinding
        // avoids firing the trigger (and its event-source subscription) at all.
        if (enabled) {
            this.activateFlowTrigger(name);
        } else {
            this.deactivateFlowTrigger(name);
        }
    }

    /** Get flow version history */
    getFlowVersionHistory(name: string): Array<{ version: number; definition: FlowParsed; createdAt: string }> {
        return this.flowVersionHistory.get(name) ?? [];
    }

    /** Rollback flow to a specific version */
    rollbackFlow(name: string, version: number): void {
        const history = this.flowVersionHistory.get(name);
        if (!history) {
            throw new Error(`Flow '${name}' has no version history`);
        }
        const entry = history.find(h => h.version === version);
        if (!entry) {
            throw new Error(`Version ${version} not found for flow '${name}'`);
        }
        this.flows.set(name, entry.definition);
        this.logger.info(`Flow '${name}' rolled back to version ${version}`);
    }

    async listRuns(flowName: string, options?: { limit?: number; cursor?: string }): Promise<ExecutionLogEntry[]> {
        const limit = options?.limit ?? 20;
        const logs = this.executionLogs.filter(l => l.flowName === flowName);
        return logs.slice(-limit).reverse();
    }

    async getRun(runId: string): Promise<ExecutionLogEntry | null> {
        return this.executionLogs.find(l => l.id === runId) ?? null;
    }

    async execute(flowName: string, context?: AutomationContext): Promise<AutomationResult> {
        const startTime = Date.now();
        const flow = this.flows.get(flowName);

        if (!flow) {
            return { success: false, error: `Flow '${flowName}' not found` };
        }

        // Check if flow is disabled
        if (this.flowEnabled.get(flowName) === false) {
            return { success: false, error: `Flow '${flowName}' is disabled` };
        }

        // Initialize variable context
        const variables = new Map<string, unknown>();
        if (flow.variables) {
            for (const v of flow.variables) {
                if (v.isInput && context?.params?.[v.name] !== undefined) {
                    variables.set(v.name, context.params[v.name]);
                }
            }
        }
        // Inject trigger record. `$record` is the canonical handle; `record` is a
        // friendlier alias so templates/conditions can write `{record.title}` and
        // `record.status`. We also flatten the record's own fields to top-level
        // variables (so bare references like `status`/`budget` resolve in start
        // conditions and edge predicates) WITHOUT clobbering flow inputs already
        // seeded above. `previous` exposes the pre-update row for transition gates.
        if (context?.record) {
            variables.set('$record', context.record);
            variables.set('record', context.record);
            for (const [k, v] of Object.entries(context.record)) {
                if (!variables.has(k)) variables.set(k, v);
            }
        }
        if (context?.previous) {
            variables.set('previous', context.previous);
        }

        const runId = this.nextRunId();
        // Expose the run id to executors (ADR-0019): a pausing node (e.g. Approval)
        // reads `$runId` to map its external state back to this run for resume.
        variables.set('$runId', runId);
        // Expose flow identity to executors so externalized state (e.g. an
        // approval request row) can carry a human-readable origin. Captured in
        // the variable snapshot, so still present after a suspend/resume.
        variables.set('$flowName', flowName);
        variables.set('$flowLabel', flow.label ?? flowName);
        const startedAt = new Date().toISOString();
        const steps: StepLogEntry[] = [];

        try {
            // Find the start node
            const startNode = flow.nodes.find(n => n.type === 'start');
            if (!startNode) {
                return { success: false, error: 'Flow has no start node' };
            }

            // Trigger-condition gate. The start node's `condition` is the predicate
            // that decides whether the trigger event should launch this flow (e.g.
            // `status == "done" && previous.status != "done"`). The engine — not the
            // trigger — owns evaluating it, so every trigger type (record-change,
            // schedule, …) and a manual `execute()` share one gate. Plain-string
            // conditions are routed through CEL so bare field references resolve.
            const startCondition = (startNode.config as Record<string, unknown> | undefined)?.condition as
                | string
                | { dialect?: string; source?: string; ast?: unknown }
                | undefined;
            if (startCondition !== undefined && startCondition !== null && startCondition !== '') {
                const condExpr =
                    typeof startCondition === 'string' ? { dialect: 'cel', source: startCondition } : startCondition;
                if (!this.evaluateCondition(condExpr, variables)) {
                    this.logger.debug(`Flow '${flowName}' skipped: start condition not met`);
                    return { success: true, output: { skipped: true, reason: 'condition_not_met' } };
                }
            }

            // Validate node input schemas before execution
            this.validateNodeInputSchemas(flow, variables);

            // DAG traversal execution
            await this.executeNode(startNode, flow, variables, context ?? {}, steps);

            // Collect output variables
            const output: Record<string, unknown> = {};
            if (flow.variables) {
                for (const v of flow.variables) {
                    if (v.isOutput) {
                        output[v.name] = variables.get(v.name);
                    }
                }
            }

            const durationMs = Date.now() - startTime;

            // Record execution log
            this.recordLog({
                id: runId,
                flowName,
                flowVersion: flow.version,
                status: 'completed',
                startedAt,
                completedAt: new Date().toISOString(),
                durationMs,
                trigger: {
                    type: context?.event ?? 'manual',
                    userId: context?.userId,
                    object: context?.object,
                },
                steps,
                output,
            });

            return {
                success: true,
                output,
                durationMs,
            };
        } catch (err: unknown) {
            // A node asked to suspend the run (ADR-0019 durable pause). Snapshot
            // the live state, record a `paused` log, and return the run id so the
            // caller can later `resume()` it. This is NOT a failure.
            if (isSuspendSignal(err)) {
                const durationMs = Date.now() - startTime;
                await this.persistSuspendedRun({
                    runId,
                    flowName,
                    flowVersion: flow.version,
                    nodeId: err.nodeId,
                    variables: Object.fromEntries(variables),
                    steps,
                    context: context ?? {},
                    startedAt,
                    startTime,
                    correlation: err.correlation,
                    screen: err.screen,
                });
                this.recordLog({
                    id: runId,
                    flowName,
                    flowVersion: flow.version,
                    status: 'paused',
                    startedAt,
                    durationMs,
                    trigger: {
                        type: context?.event ?? 'manual',
                        userId: context?.userId,
                        object: context?.object,
                    },
                    steps,
                });
                return {
                    success: true,
                    status: 'paused',
                    runId,
                    durationMs,
                    screen: err.screen,
                };
            }

            const errorMessage = err instanceof Error ? err.message : String(err);

            // Record failed execution log
            const durationMs = Date.now() - startTime;
            this.recordLog({
                id: runId,
                flowName,
                flowVersion: flow.version,
                status: 'failed',
                startedAt,
                completedAt: new Date().toISOString(),
                durationMs,
                trigger: {
                    type: context?.event ?? 'manual',
                    userId: context?.userId,
                    object: context?.object,
                },
                steps,
                error: errorMessage,
            });

            // Error handling strategy
            if (flow.errorHandling?.strategy === 'retry') {
                return this.retryExecution(flowName, context, startTime, flow.errorHandling);
            }
            return {
                success: false,
                error: errorMessage,
                durationMs,
            };
        }
    }

    /**
     * Resume a run suspended at a node (ADR-0019 durable pause). Restores the
     * snapshotted variables, merges `signal.output` under the suspended node's
     * id, and continues traversal from that node's out-edges — optionally
     * restricted to the edge labelled `signal.branchLabel` (e.g. the approval
     * decision). The continuation may itself suspend again, in which case this
     * returns `{ status: 'paused', runId }` afresh.
     *
     * **Subflow chains (nested pause, linked-runs model).** A run paused at a
     * `subflow` node (correlation `subflow:<childRunId>`) DELEGATES the signal
     * down to the suspended child; a run that completes and carries
     * `$parentRunId` in its context BUBBLES its output up by auto-resuming the
     * parent. Both directions compose recursively, so arbitrarily nested
     * subflow pauses resolve from either end (UI holds the parent run id;
     * approval/wait infrastructure holds the child's).
     */
    async resume(runId: string, signal?: ResumeSignal): Promise<AutomationResult> {
        return this.resumeInternal(runId, signal, false);
    }

    /**
     * @param skipBubble - Set when the caller is the subflow DELEGATION path,
     *   which continues the parent itself after the child completes — the
     *   child's own up-bubble must stay off so the parent isn't resumed twice.
     */
    private async resumeInternal(runId: string, signal: ResumeSignal | undefined, skipBubble: boolean): Promise<AutomationResult> {
        // Idempotency guard (set synchronously, before any await): reject a
        // concurrent duplicate resume of the same run so side effects can't run
        // twice. A duplicate that arrives *after* this one finishes finds no
        // suspended run and returns the "no suspended run" error below.
        if (this.resuming.has(runId)) {
            return { success: false, error: `Run '${runId}' is already being resumed` };
        }
        this.resuming.add(runId);
        try {
            // Hot path: suspended in this process. Cold path: rehydrate from the
            // durable store (e.g. the process restarted since the pause, ADR-0019).
            let run = this.suspendedRuns.get(runId) ?? null;
            if (!run && this.store) {
                try {
                    run = await this.store.load(runId);
                } catch (err) {
                    this.logger.warn(
                        `[automation] failed to load suspended run '${runId}' from durable store: ${(err as Error).message}`,
                    );
                }
            }
            if (!run) {
                return { success: false, error: `No suspended run '${runId}'` };
            }
            const flow = this.flows.get(run.flowName);
            if (!flow) {
                return { success: false, error: `Flow '${run.flowName}' not found for run '${runId}'` };
            }
            const node = flow.nodes.find(n => n.id === run.nodeId);
            if (!node) {
                return { success: false, error: `Suspended node '${run.nodeId}' no longer exists in flow '${run.flowName}'` };
            }

            // ── Subflow delegation (nested pause): this run is paused at a
            // `subflow` node whose child run itself suspended. The caller's
            // signal is meant for the node the CHILD paused on (its screen /
            // approval / wait), so forward it down. The child resumes with
            // bubbling off — when it completes, *this* invocation continues the
            // parent from the subflow node with the child's output, using the
            // same mapping as the synchronous path.
            if (typeof run.correlation === 'string' && run.correlation.startsWith('subflow:')) {
                const childRunId = run.correlation.slice('subflow:'.length);
                // Capture the child's row BEFORE resuming consumes it — the
                // output-variable mapping rides on the child's context.
                const childRun =
                    this.suspendedRuns.get(childRunId) ??
                    (this.store ? await this.store.load(childRunId).catch(() => null) : null);
                if (childRun) {
                    const childRes = await this.resumeInternal(childRunId, signal, true);
                    if (childRes.status === 'paused') {
                        // Child paused again (e.g. the next screen of a wizard).
                        // This run stays suspended; refresh its surfaced screen
                        // so a re-fetch (getSuspendedScreen) shows the new one.
                        if (childRes.screen && childRes.screen !== run.screen) {
                            await this.persistSuspendedRun({ ...run, screen: childRes.screen });
                        }
                        return {
                            success: true,
                            status: 'paused',
                            runId,
                            durationMs: Date.now() - run.startTime,
                            screen: childRes.screen,
                        };
                    }
                    if (!childRes.success) {
                        const error = `subflow run '${childRunId}' (${childRun.flowName}) failed: ${childRes.error ?? 'unknown error'}`;
                        await this.failSuspendedRun(run, error);
                        return { success: false, error, durationMs: Date.now() - run.startTime };
                    }
                    // Child completed — continue below with its output as the
                    // resume signal (replaces the caller's signal, which the
                    // child already consumed).
                    signal = this.buildSubflowResumeSignal(childRun.context, childRes.output);
                } else {
                    this.logger.warn(
                        `[automation] run '${runId}' is paused at subflow node '${run.nodeId}' but child run '${childRunId}' ` +
                            `is gone — continuing without child output`,
                    );
                }
            }

            // Consume the suspension *before* running downstream work — a run
            // resumes exactly once per pause, and a duplicate resume after a
            // partial restart must not double-run side effects.
            await this.forgetSuspendedRun(runId);

            // Restore variable context and apply the resume signal's output as if it
            // were the node's output, so downstream edges branch on it.
            const variables = new Map<string, unknown>(Object.entries(run.variables));
            if (signal?.output) {
                for (const [key, value] of Object.entries(signal.output)) {
                    variables.set(`${run.nodeId}.${key}`, value);
                }
            }
            // Bare flow variables — a `screen` node's collected inputs land under
            // their plain names so downstream `{var}` interpolation / conditions
            // read them directly (e.g. `new_assignee` → update_record fields).
            if (signal?.variables) {
                for (const [key, value] of Object.entries(signal.variables)) {
                    variables.set(key, value);
                }
            }

            const steps = run.steps;
            const context = run.context;

            try {
                // ── Map re-entry (sequential multi-instance, ADR-0037 A2).
                // A run paused at a `map` node (correlation `map:<childRunId>`)
                // does NOT continue past the node on resume — it RE-RUNS the
                // node so the executor can record the just-completed unit and
                // start the next item. The default path continues past the node.
                if (typeof run.correlation === 'string' && run.correlation.startsWith('map:')) {
                    await this.executeNode(node, flow, variables, context, steps);
                } else {
                    await this.traverseNext(node, flow, variables, context, steps, signal?.branchLabel);
                }

                // Collect output variables
                const output: Record<string, unknown> = {};
                if (flow.variables) {
                    for (const v of flow.variables) {
                        if (v.isOutput) output[v.name] = variables.get(v.name);
                    }
                }
                const durationMs = Date.now() - run.startTime;
                this.recordLog({
                    id: runId,
                    flowName: run.flowName,
                    flowVersion: run.flowVersion,
                    status: 'completed',
                    startedAt: run.startedAt,
                    completedAt: new Date().toISOString(),
                    durationMs,
                    trigger: {
                        type: context.event ?? 'manual',
                        userId: context.userId,
                        object: context.object,
                    },
                    steps,
                    output,
                });

                // ── Subflow up-bubble (nested pause): this run was a subflow
                // child whose parent suspended awaiting it. Auto-resume the
                // parent with our output, mapped like the synchronous path.
                // Skipped when the DELEGATION path drives the chain (it
                // continues the parent itself). Best-effort: the child's own
                // completion stands even if the parent continuation fails.
                if (!skipBubble) {
                    await this.bubbleToParent(run, output);
                }

                return { success: true, output, durationMs };
            } catch (err: unknown) {
                // Re-suspended at a downstream node: persist a fresh continuation.
                if (isSuspendSignal(err)) {
                    const durationMs = Date.now() - run.startTime;
                    await this.persistSuspendedRun({
                        ...run,
                        nodeId: err.nodeId,
                        variables: Object.fromEntries(variables),
                        steps,
                        correlation: err.correlation,
                        screen: err.screen,
                    });
                    this.recordLog({
                        id: runId,
                        flowName: run.flowName,
                        flowVersion: run.flowVersion,
                        status: 'paused',
                        startedAt: run.startedAt,
                        durationMs,
                        trigger: {
                            type: context.event ?? 'manual',
                            userId: context.userId,
                            object: context.object,
                        },
                        steps,
                    });
                    return { success: true, status: 'paused', runId, durationMs, screen: err.screen };
                }

                const errorMessage = err instanceof Error ? err.message : String(err);
                const durationMs = Date.now() - run.startTime;
                this.recordLog({
                    id: runId,
                    flowName: run.flowName,
                    flowVersion: run.flowVersion,
                    status: 'failed',
                    startedAt: run.startedAt,
                    completedAt: new Date().toISOString(),
                    durationMs,
                    trigger: {
                        type: context.event ?? 'manual',
                        userId: context.userId,
                        object: context.object,
                    },
                    steps,
                    error: errorMessage,
                });
                // Subflow chain: a child failing terminally fails every
                // ancestor awaiting it — they can never be resumed otherwise.
                // The delegation path handles its own level (skipBubble).
                if (!skipBubble) {
                    await this.failAncestors(run.context, errorMessage);
                }
                return { success: false, error: errorMessage, durationMs };
            }
        } finally {
            this.resuming.delete(runId);
        }
    }

    /**
     * Build the resume signal that maps a completed subflow child's output
     * into its parent — mirroring the synchronous path exactly: the engine's
     * standard `signal.output` merge lands it under `${subflowNodeId}.output`,
     * and `signal.variables` writes the bare `config.outputVariable` when the
     * child's context carries one (`$parentOutputVariable`).
     */
    private buildSubflowResumeSignal(childContext: AutomationContext | undefined, childOutput: unknown): ResumeSignal {
        const outVar = (childContext as Record<string, unknown> | undefined)?.$parentOutputVariable;
        return {
            output: { output: childOutput ?? null },
            ...(typeof outVar === 'string' && outVar
                ? { variables: { [outVar]: childOutput ?? null } }
                : {}),
        };
    }

    /**
     * Up-bubble for the subflow chain: when a completed run carries
     * `$parentRunId`, resume that parent with this run's output. Recursion via
     * the parent's own completion bubbles multi-level chains. Best-effort —
     * a failed parent continuation is logged, never thrown back at the
     * caller who resumed the child.
     */
    private async bubbleToParent(run: SuspendedRun, output: Record<string, unknown>): Promise<void> {
        const ctx = run.context as Record<string, unknown> | undefined;
        const parentRunId = ctx?.$parentRunId;
        if (typeof parentRunId !== 'string' || !parentRunId) return;
        try {
            // A `map` child (ADR-0037 A2): hand the unit's output to the map
            // node + flag the completion, so on re-entry it records this item
            // and starts the next. A plain subflow child uses the 1:1 mapping.
            const mapNode = ctx?.$parentMapNode;
            const sig = typeof mapNode === 'string' && mapNode
                ? { variables: { [`${mapNode}.$mapItemOutput`]: output ?? null, [`${mapNode}.$mapItemDone`]: true } }
                : this.buildSubflowResumeSignal(run.context, output);
            const parentRes = await this.resumeInternal(parentRunId, sig, false);
            if (!parentRes.success) {
                this.logger.warn(
                    `[automation] subflow run '${run.runId}' completed but resuming parent '${parentRunId}' failed: ${parentRes.error}`,
                );
            }
        } catch (err) {
            this.logger.warn(
                `[automation] subflow run '${run.runId}' completed but resuming parent '${parentRunId}' threw: ${(err as Error).message}`,
            );
        }
    }

    /**
     * Terminally fail a suspended run: consume its continuation and record a
     * `failed` log so it stops surfacing as resumable. Used when a subflow
     * descendant fails — the ancestor awaiting it can never be resumed.
     */
    private async failSuspendedRun(run: SuspendedRun, error: string): Promise<void> {
        await this.forgetSuspendedRun(run.runId);
        this.recordLog({
            id: run.runId,
            flowName: run.flowName,
            flowVersion: run.flowVersion,
            status: 'failed',
            startedAt: run.startedAt,
            completedAt: new Date().toISOString(),
            durationMs: Date.now() - run.startTime,
            trigger: {
                type: run.context?.event ?? 'manual',
                userId: run.context?.userId,
                object: run.context?.object,
            },
            steps: run.steps,
            error,
        });
    }

    /**
     * Cancel a suspended run (ADR-0044): consume its continuation and record a
     * terminal `cancelled` log so it stops surfacing as resumable. The
     * engine-level primitive behind "the submitter abandoned the revision
     * window" — recalling there leaves the run paused at a wait node with no
     * reject edge to resume down, so the run must end, not continue. Returns
     * `false` when no suspended run exists under the id (already terminal /
     * unknown), which callers treat as idempotent success.
     */
    async cancelRun(runId: string, reason?: string): Promise<boolean> {
        let run = this.suspendedRuns.get(runId) ?? null;
        if (!run && this.store) {
            try {
                run = await this.store.load(runId);
            } catch (err) {
                this.logger.warn(
                    `[automation] cancelRun: failed to load suspended run '${runId}' from durable store: ${(err as Error).message}`,
                );
            }
        }
        if (!run) return false;
        await this.forgetSuspendedRun(runId);
        this.recordLog({
            id: run.runId,
            flowName: run.flowName,
            flowVersion: run.flowVersion,
            status: 'cancelled',
            startedAt: run.startedAt,
            completedAt: new Date().toISOString(),
            durationMs: Date.now() - run.startTime,
            trigger: {
                type: run.context?.event ?? 'manual',
                userId: run.context?.userId,
                object: run.context?.object,
            },
            steps: run.steps,
            error: reason,
        });
        return true;
    }

    /**
     * Walk a failed run's `$parentRunId` chain and fail each suspended
     * ancestor (see {@link failSuspendedRun}). Bounded so a corrupt context
     * can't loop forever.
     */
    private async failAncestors(context: AutomationContext | undefined, error: string): Promise<void> {
        let parentId = (context as Record<string, unknown> | undefined)?.$parentRunId;
        let hops = 0;
        while (typeof parentId === 'string' && parentId && hops++ < 32) {
            const parent =
                this.suspendedRuns.get(parentId) ??
                (this.store ? await this.store.load(parentId).catch(() => null) : null);
            if (!parent) return;
            await this.failSuspendedRun(parent, `subflow descendant failed: ${error}`);
            parentId = (parent.context as Record<string, unknown> | undefined)?.$parentRunId;
        }
    }

    /**
     * List the runs currently suspended awaiting {@link resume} (ADR-0019).
     * Backs operability surfaces such as a "pending approvals" view.
     *
     * Synchronous — reads the in-memory cache only, so after a process restart
     * runs that suspended in a prior lifetime are not listed here even though
     * they remain durably stored and resumable by id. Use
     * {@link listSuspendedRunsDurable} to include those.
     */
    listSuspendedRuns(): Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }> {
        return [...this.suspendedRuns.values()].map(r => ({
            runId: r.runId,
            flowName: r.flowName,
            nodeId: r.nodeId,
            correlation: r.correlation,
        }));
    }

    /**
     * Like {@link listSuspendedRuns} but includes runs held only in the durable
     * {@link SuspendedRunStore} (e.g. suspended before a restart). The in-memory
     * cache takes precedence on id collisions. Falls back to the in-memory list
     * when no store is configured.
     */
    async listSuspendedRunsDurable(): Promise<Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }>> {
        const byId = new Map<string, { runId: string; flowName: string; nodeId: string; correlation?: string }>();
        if (this.store) {
            try {
                for (const r of await this.store.list()) {
                    byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation });
                }
            } catch (err) {
                this.logger.warn(`[automation] failed to list suspended runs from durable store: ${(err as Error).message}`);
            }
        }
        // In-memory entries win — they are the freshest copy.
        for (const r of this.suspendedRuns.values()) {
            byId.set(r.runId, { runId: r.runId, flowName: r.flowName, nodeId: r.nodeId, correlation: r.correlation });
        }
        return [...byId.values()];
    }

    /**
     * The screen a paused run is currently waiting on (screen-flow runtime), or
     * `null` if the run isn't suspended / didn't pause at a screen node. Lets a
     * UI flow-runner re-fetch the form after a refresh.
     */
    getSuspendedScreen(runId: string): ScreenSpec | null {
        return this.suspendedRuns.get(runId)?.screen ?? null;
    }

    // ── DAG Traversal Core ──────────────────────────────────

    private recordLog(entry: ExecutionLogEntry): void {
        this.executionLogs.push(entry);
        // Evict oldest logs when exceeding max size
        if (this.executionLogs.length > this.maxLogSize) {
            this.executionLogs.splice(0, this.executionLogs.length - this.maxLogSize);
        }
    }

    /**
     * Validate each node's `type` against the live action registry (ADR-0018).
     * A type is known if it is structural (start/end), has a registered
     * executor, or has a published action descriptor. Unknown types are
     * warned about (not rejected) so flows authored against a temporarily
     * absent plugin still register; the runtime surfaces a hard NO_EXECUTOR
     * error if such a node is actually executed.
     */
    private validateNodeTypes(flowName: string, flow: FlowParsed): void {
        const known = new Set<string>([
            ...FLOW_STRUCTURAL_NODE_TYPES,
            ...this.nodeExecutors.keys(),
            ...this.actionDescriptors.keys(),
        ]);
        const unknown = [...new Set(
            flow.nodes.map(n => n.type).filter(t => !known.has(t)),
        )];
        if (unknown.length > 0) {
            this.logger.warn(
                `Flow '${flowName}' references node type(s) with no registered executor or descriptor: ` +
                `${unknown.join(', ')}. They will fail at execution time unless a plugin registers them. ` +
                `Registered types: ${[...known].join(', ') || '(none)'}`,
            );
        }
    }

    /**
     * ADR-0032 §Decision 1a — parse-validate every predicate in the flow at
     * registration. Predicates are bare CEL; this catches the #1491 class
     * (`{record.x}` template braces in a condition → CEL parse error) and any
     * other malformed predicate LOUDLY, with the offending location + source +
     * a corrective hint, instead of letting it fail silently at run time.
     *
     * Only the *predicate* surfaces are checked here (start/node `config.condition`
     * and `edge.condition`) — node string fields are templates (a different
     * dialect) and are validated by the template engine, not as CEL.
     */
    private validateFlowExpressions(flowName: string, flow: FlowParsed): void {
        const failures: string[] = [];

        const check = (where: string, raw: unknown): void => {
            if (raw == null) return;
            // Conditions are predicates (bare CEL). Delegate to the one shared
            // validator (ADR-0032 §5) so the corrective message matches the CLI
            // build and the agent `validate_expression` tool exactly.
            const result = validateExpression('predicate', raw as string | { dialect?: string; source?: string });
            for (const e of result.errors) {
                failures.push(`  • ${where}: ${e.message}\n      source: \`${e.source}\``);
            }
        };

        for (const node of flow.nodes) {
            const cfg = (node.config ?? {}) as Record<string, unknown>;
            // start-node trigger gate + decision/branch predicates live in config.condition
            check(`node '${node.id}' (${node.type}) condition`, cfg.condition);
        }
        for (const edge of flow.edges) {
            check(`edge '${edge.id}' (${edge.source}→${edge.target}) condition`, edge.condition as unknown);
        }

        if (failures.length > 0) {
            throw new Error(
                `Flow '${flowName}' has ${failures.length} invalid condition${failures.length > 1 ? 's' : ''} (ADR-0032 §1a). ` +
                `Conditions are bare CEL — do not wrap field references in \`{…}\` template braces:\n${failures.join('\n')}`,
            );
        }
    }

    /**
     * Detect cycles in the flow graph (DAG validation).
     * Uses DFS with coloring (white/gray/black) to detect back edges.
     * Throws an error with cycle details if a cycle is found.
     *
     * ADR-0044: edges explicitly typed `back` (declared back-edges — e.g. a
     * revise/rework loop re-entering an approval node) are excluded from the
     * analysis: the graph **minus `back` edges** must be a DAG. An unmarked
     * cycle is still rejected — authors opt in edge by edge. At run time a
     * `back` edge traverses like any default edge; the re-entry runaway guard
     * lives in {@link executeNode}.
     */
    private detectCycles(flow: FlowParsed): void {
        const WHITE = 0, GRAY = 1, BLACK = 2;
        const color = new Map<string, number>();
        const parent = new Map<string, string>();

        // Build adjacency list from edges
        const adj = new Map<string, string[]>();
        for (const node of flow.nodes) {
            color.set(node.id, WHITE);
            adj.set(node.id, []);
        }
        for (const edge of flow.edges) {
            if (edge.type === 'back') continue; // ADR-0044 declared back-edge
            const targets = adj.get(edge.source);
            if (targets) targets.push(edge.target);
        }

        const dfs = (nodeId: string): string[] | null => {
            color.set(nodeId, GRAY);
            for (const neighbor of adj.get(nodeId) ?? []) {
                if (color.get(neighbor) === GRAY) {
                    // Back edge found — reconstruct cycle
                    const cycle = [neighbor, nodeId];
                    let cur = nodeId;
                    while (cur !== neighbor) {
                        cur = parent.get(cur)!;
                        if (cur) cycle.push(cur);
                        else break;
                    }
                    return cycle.reverse();
                }
                if (color.get(neighbor) === WHITE) {
                    parent.set(neighbor, nodeId);
                    const result = dfs(neighbor);
                    if (result) return result;
                }
            }
            color.set(nodeId, BLACK);
            return null;
        };

        for (const node of flow.nodes) {
            if (color.get(node.id) === WHITE) {
                const cycle = dfs(node.id);
                if (cycle) {
                    throw new Error(
                        `Flow contains a cycle: ${cycle.join(' → ')}. Only DAG flows are allowed — ` +
                        `to author an intentional rework loop, mark the cycle-closing edge with type: 'back' (ADR-0044).`,
                    );
                }
            }
        }
    }

    /**
     * Get the runtime type name of a value for schema validation.
     */
    private getValueType(value: unknown): string {
        if (Array.isArray(value)) return 'array';
        if (typeof value === 'object' && value !== null) return 'object';
        return typeof value;
    }

    /**
     * Validate node input schemas before execution.
     * Checks that node config matches declared inputSchema if present.
     */
    private validateNodeInputSchemas(flow: FlowParsed, _variables: Map<string, unknown>): void {
        for (const node of flow.nodes) {
            if (node.inputSchema && node.config) {
                for (const [paramName, paramDef] of Object.entries(node.inputSchema)) {
                    if (paramDef.required && !(paramName in (node.config as Record<string, unknown>))) {
                        throw new Error(
                            `Node '${node.id}' missing required input parameter '${paramName}'`,
                        );
                    }
                    const value = (node.config as Record<string, unknown>)[paramName];
                    if (value !== undefined) {
                        const actualType = this.getValueType(value);
                        if (actualType !== paramDef.type) {
                            throw new Error(
                                `Node '${node.id}' parameter '${paramName}' expected type '${paramDef.type}' but got '${actualType}'`,
                            );
                        }
                    }
                }
            }
        }
    }

    /**
     * Execute a node with timeout support, fault edge handling, and step logging.
     */
    private async executeNode(
        node: FlowNodeParsed,
        flow: FlowParsed,
        variables: Map<string, unknown>,
        context: AutomationContext,
        steps: StepLogEntry[],
    ): Promise<void> {
        if (node.type === 'end') return;

        // ADR-0044 runaway guard: declared back-edges make re-entering a node
        // legal, so a misauthored unconditional loop could otherwise spin
        // forever. Count this node's prior *top-level* visits in the run's step
        // log (region body steps carry `parentNodeId` and are excluded — a
        // 200-iteration `loop` region is legitimate) and fail the run loudly
        // past the cap. Product-level guards (e.g. an approval node's
        // `maxRevisions`) terminate far earlier; this is the engine backstop.
        const priorVisits = steps.reduce(
            (n, s) => (s.nodeId === node.id && s.parentNodeId === undefined ? n + 1 : n), 0,
        );
        if (priorVisits >= AutomationEngine.MAX_NODE_REENTRIES) {
            throw new Error(
                `Node '${node.id}' was entered ${priorVisits} times in one run — aborting as a runaway loop ` +
                `(back-edge cycles must terminate; see ADR-0044)`,
            );
        }

        const stepStart = Date.now();
        const stepStartedAt = new Date().toISOString();

        // Find executor
        const executor = this.nodeExecutors.get(node.type);
        if (!executor) {
            // start node without executor is fine — just skip
            if (node.type !== 'start') {
                steps.push({
                    nodeId: node.id,
                    nodeType: node.type,
                    status: 'failure',
                    startedAt: stepStartedAt,
                    completedAt: new Date().toISOString(),
                    durationMs: Date.now() - stepStart,
                    error: { code: 'NO_EXECUTOR', message: `No executor registered for node type '${node.type}'` },
                });
                throw new Error(`No executor registered for node type '${node.type}'`);
            }
            // Log start node step
            steps.push({
                nodeId: node.id,
                nodeType: node.type,
                status: 'success',
                startedAt: stepStartedAt,
                completedAt: new Date().toISOString(),
                durationMs: Date.now() - stepStart,
            });
        } else {
            // Execute node with optional timeout
            let result: NodeExecutionResult;
            try {
                if (node.timeoutMs && node.timeoutMs > 0) {
                    result = await this.executeWithTimeout(
                        executor.execute(node, variables, context),
                        node.timeoutMs,
                        node.id,
                    );
                } else {
                    result = await executor.execute(node, variables, context);
                }
            } catch (execErr: unknown) {
                const errMsg = execErr instanceof Error ? execErr.message : String(execErr);
                steps.push({
                    nodeId: node.id,
                    nodeType: node.type,
                    status: 'failure',
                    startedAt: stepStartedAt,
                    completedAt: new Date().toISOString(),
                    durationMs: Date.now() - stepStart,
                    error: { code: 'EXECUTION_ERROR', message: errMsg },
                });

                // Check for fault edges
                const faultEdge = flow.edges.find(e => e.source === node.id && e.type === 'fault');
                if (faultEdge) {
                    variables.set('$error', { nodeId: node.id, message: errMsg });
                    const faultTarget = flow.nodes.find(n => n.id === faultEdge.target);
                    if (faultTarget) {
                        await this.executeNode(faultTarget, flow, variables, context, steps);
                        return;
                    }
                }
                throw execErr;
            }

            if (!result.success) {
                const errMsg = result.error ?? 'Unknown error';
                steps.push({
                    nodeId: node.id,
                    nodeType: node.type,
                    status: 'failure',
                    startedAt: stepStartedAt,
                    completedAt: new Date().toISOString(),
                    durationMs: Date.now() - stepStart,
                    error: { code: 'NODE_FAILURE', message: errMsg },
                });

                // Write error output to variable context for downstream nodes
                variables.set('$error', { nodeId: node.id, message: errMsg, output: result.output });

                // Check for fault edges
                const faultEdge = flow.edges.find(e => e.source === node.id && e.type === 'fault');
                if (faultEdge) {
                    const faultTarget = flow.nodes.find(n => n.id === faultEdge.target);
                    if (faultTarget) {
                        await this.executeNode(faultTarget, flow, variables, context, steps);
                        return;
                    }
                }
                throw new Error(`Node '${node.id}' failed: ${errMsg}`);
            }

            // Log successful step
            steps.push({
                nodeId: node.id,
                nodeType: node.type,
                status: 'success',
                startedAt: stepStartedAt,
                completedAt: new Date().toISOString(),
                durationMs: Date.now() - stepStart,
            });

            // #1479: fold a structured-region container's body/branch/handler
            // steps into the run log, right after the container's own step.
            if (result.childSteps?.length) {
                steps.push(...result.childSteps);
            }

            // Write back output variables
            if (result.output) {
                for (const [key, value] of Object.entries(result.output)) {
                    variables.set(`${node.id}.${key}`, value);
                }
            }

            // ADR-0019 durable pause: the node did its on-entry work and asked to
            // suspend here. Output is already written above; unwind the recursion
            // up to execute()/resume(), which persists a continuation. Traversal
            // of this node's out-edges happens on resume, not now.
            if (result.suspend) {
                throw new FlowSuspendSignal(node.id, result.correlation, result.screen);
            }
        }

        // Continue to the node's successors.
        await this.traverseNext(node, flow, variables, context, steps);
    }

    /**
     * Traverse a node's out-edges and execute its successors. Split out of
     * {@link executeNode} so {@link resume} can re-enter traversal from a
     * suspended node without re-running the node body.
     *
     * @param branchLabel - When set (e.g. from a resume signal), restrict
     *   traversal to out-edges whose `label` matches — this is how an Approval
     *   node's `approve`/`reject` decision selects its downstream branch. When
     *   no edge carries the label, traversal falls back to the normal edge set.
     */
    private async traverseNext(
        node: FlowNodeParsed,
        flow: FlowParsed,
        variables: Map<string, unknown>,
        context: AutomationContext,
        steps: StepLogEntry[],
        branchLabel?: string,
    ): Promise<void> {
        // Find next nodes — separate conditional and unconditional edges
        let outEdges = flow.edges.filter(
            e => e.source === node.id && e.type !== 'fault',
        );

        // Branch selection (resume): prefer edges tagged with the decision label.
        if (branchLabel) {
            const labeled = outEdges.filter(e => e.label === branchLabel);
            if (labeled.length > 0) outEdges = labeled;
        }

        const conditionalEdges: FlowEdgeParsed[] = [];
        const unconditionalEdges: FlowEdgeParsed[] = [];
        for (const edge of outEdges) {
            if (edge.condition) {
                conditionalEdges.push(edge);
            } else {
                unconditionalEdges.push(edge);
            }
        }

        // Conditional edges: evaluate sequentially (mutually exclusive)
        for (const edge of conditionalEdges) {
            if (this.evaluateCondition(edge.condition!, variables)) {
                const nextNode = flow.nodes.find(n => n.id === edge.target);
                if (nextNode) {
                    await this.executeNode(nextNode, flow, variables, context, steps);
                }
            }
        }

        // Unconditional edges: execute in parallel (Promise.all)
        if (unconditionalEdges.length > 0) {
            const parallelTasks = unconditionalEdges
                .map(edge => flow.nodes.find(n => n.id === edge.target))
                .filter((n): n is FlowNodeParsed => n != null)
                .map(nextNode => this.executeNode(nextNode, flow, variables, context, steps));

            await Promise.all(parallelTasks);
        }
    }

    /**
     * Execute a structured control-flow **region** (ADR-0031) — the nested
     * body of a `loop` container (or, later, a `parallel` branch / `try_catch`
     * region). The region is a self-contained single-entry/single-exit
     * sub-graph carried in the container's `config`; it runs in the **enclosing
     * variable scope** (the caller's `variables` map), so the iterator variable
     * and any body mutations are visible to the surrounding flow — a region is
     * NOT a separate `subflow` invocation.
     *
     * The region executes against a synthetic flow view of its own
     * nodes/edges, so the main DAG traversal (`traverseNext`) is never aware of
     * scope markers — keeping the shared traversal untouched.
     *
     * #1479: the executed body steps are **returned** (tagged with `grouping`)
     * so the calling container node can fold them into the parent run log via
     * `NodeExecutionResult.childSteps`. Tagging only fills fields left undefined,
     * so when regions nest, each step keeps its **innermost** container's
     * `parentNodeId` / `iteration` / `regionKind`. On failure the region throws
     * as before (preserving `try_catch` retry semantics); a failed attempt's
     * partial steps are not surfaced.
     *
     * Durable pause (`suspend`) inside a region is not supported in this
     * iteration — it is converted into a clear error (mirrors the `subflow`
     * nested-pause guard).
     */
    async runRegion(
        region: FlowRegionParsed,
        variables: Map<string, unknown>,
        context: AutomationContext,
        grouping?: { parentNodeId: string; iteration?: number; regionKind?: string },
    ): Promise<StepLogEntry[]> {
        const entryId = findRegionEntry(region);
        const entry = region.nodes.find(n => n.id === entryId);
        if (!entry) {
            throw new Error(`region entry node '${entryId}' not found`);
        }
        // A synthetic flow view — executeNode/traverseNext only read `nodes`/`edges`.
        const subFlow = { nodes: region.nodes, edges: region.edges ?? [] } as unknown as FlowParsed;
        const regionSteps: StepLogEntry[] = [];
        try {
            await this.executeNode(entry, subFlow, variables, context, regionSteps);
        } catch (err) {
            if (isSuspendSignal(err)) {
                throw new Error(
                    `durable pause inside a structured region (node '${err.nodeId}') is not supported`,
                );
            }
            throw err;
        }
        // Tag this region's steps with their immediate container. Innermost wins:
        // a step that already carries a `parentNodeId` (set by a nested region)
        // is left untouched.
        if (grouping) {
            for (const step of regionSteps) {
                if (step.parentNodeId === undefined) {
                    step.parentNodeId = grouping.parentNodeId;
                    if (grouping.iteration !== undefined) step.iteration = grouping.iteration;
                    if (grouping.regionKind !== undefined) step.regionKind = grouping.regionKind;
                }
            }
        }
        return regionSteps;
    }

    /**
     * Execute a promise with timeout using Promise.race.
     */
    private executeWithTimeout(
        promise: Promise<NodeExecutionResult>,
        timeoutMs: number,
        nodeId: string,
    ): Promise<NodeExecutionResult> {
        return Promise.race([
            promise,
            new Promise<NodeExecutionResult>((_, reject) =>
                setTimeout(() => reject(new Error(`Node '${nodeId}' timed out after ${timeoutMs}ms`)), timeoutMs),
            ),
        ]);
    }

    /**
     * Safe expression evaluator.
     * Uses simple operator-based parsing without `new Function`.
     * Supports: comparisons (>, <, >=, <=, ==, !=, ===, !==),
     * boolean literals (true, false), and basic arithmetic.
     */
    evaluateCondition(expression: string | { dialect?: string; source?: string; ast?: unknown }, variables: Map<string, unknown>): boolean {
        // M9.5+ wiring: route Expression envelopes through @objectstack/formula
        // ExpressionEngine. CEL is the default; legacy `{var}` template syntax
        // is preserved as a fallback for back-compat.
        const isEnvelope = typeof expression === 'object' && expression != null && 'dialect' in expression;
        const dialect = isEnvelope ? (expression as { dialect?: string }).dialect : undefined;
        const exprStr = typeof expression === 'string' ? expression : ((expression as { source?: string })?.source ?? '');

        if (isEnvelope && dialect && dialect !== 'cel' && dialect !== 'flow' && dialect !== 'template') {
            // Other dialects (cron, js) are not boolean predicates here.
            return false;
        }

        // CEL path — bind `vars` scope for `{step.result}` style references via
        // the equivalent `vars.step.result` CEL identifier path.
        if (dialect === 'cel' || (isEnvelope && !dialect)) {
            try {
                const vars: Record<string, unknown> = {};
                for (const [key, value] of variables) {
                    // Convert "step.result" keys into nested object paths.
                    const segs = key.split('.');
                    let cursor = vars;
                    for (let i = 0; i < segs.length - 1; i++) {
                        if (typeof cursor[segs[i]] !== 'object' || cursor[segs[i]] === null) {
                            cursor[segs[i]] = {};
                        }
                        cursor = cursor[segs[i]] as Record<string, unknown>;
                    }
                    cursor[segs[segs.length - 1]] = value;
                }
                // Expose variables two ways under `extra`: as a `vars` namespace
                // (so `vars.step.result` keeps working) AND spread to top level (so
                // bare identifiers like `status` / `previous.status` resolve — the
                // natural authoring style for record-change start conditions).
                const result = ExpressionEngine.evaluate(
                    { dialect: 'cel', source: exprStr },
                    { extra: { ...vars, vars }, record: vars },
                );
                // ADR-0032 §Decision 1c — NO silent fallback. A non-`ok` result is a
                // real fault (malformed predicate, or — pre build-validation — a
                // `{…}` template mistakenly written into a CEL condition). Surfacing
                // it as a thrown, attributed error makes execute()'s catch record a
                // loud flow failure, instead of the old `return false` that made a
                // broken condition indistinguishable from "condition not met" (#1491).
                if (!result.ok) {
                    throw new Error(
                        `condition failed to evaluate as CEL: ${result.error?.message ?? 'unknown error'} — ` +
                        `source: \`${exprStr}\`. Conditions are bare CEL (e.g. \`record.rating >= 4\`); ` +
                        `do not wrap field references in \`{…}\` template braces.`,
                    );
                }
                return Boolean(result.value);
            } catch (err) {
                // Re-throw with the source attached (ADR-0032 §1d — errors written
                // for self-correction). Never swallow to `false`.
                const msg = (err as Error)?.message ?? String(err);
                throw new Error(
                    msg.includes('source:') ? msg : `condition evaluation error: ${msg} — source: \`${exprStr}\``,
                );
            }
        }

        // Legacy template path: {varName} → value, then primitive compare.
        let resolved = exprStr;
        for (const [key, value] of variables) {
            resolved = resolved.split(`{${key}}`).join(String(value));
        }
        resolved = resolved.trim();

        try {
            // Boolean literals
            if (resolved === 'true') return true;
            if (resolved === 'false') return false;

            // Comparison operators (ordered by length to match longer operators first)
            const operators = ['===', '!==', '>=', '<=', '!=', '==', '>', '<'] as const;
            for (const op of operators) {
                const idx = resolved.indexOf(op);
                if (idx !== -1) {
                    const left = resolved.slice(0, idx).trim();
                    const right = resolved.slice(idx + op.length).trim();
                    return this.compareValues(left, op, right);
                }
            }

            // Numeric truthy check
            const numVal = Number(resolved);
            if (!isNaN(numVal)) return numVal !== 0;

            return false;
        } catch {
            return false;
        }
    }

    /**
     * Compare two string-represented values with an operator.
     */
    private compareValues(left: string, op: string, right: string): boolean {
        const lNum = Number(left);
        const rNum = Number(right);
        const bothNumeric = !isNaN(lNum) && !isNaN(rNum) && left !== '' && right !== '';

        if (bothNumeric) {
            switch (op) {
                case '>': return lNum > rNum;
                case '<': return lNum < rNum;
                case '>=': return lNum >= rNum;
                case '<=': return lNum <= rNum;
                case '==': case '===': return lNum === rNum;
                case '!=': case '!==': return lNum !== rNum;
                default: return false;
            }
        }
        // String comparison
        switch (op) {
            case '==': case '===': return left === right;
            case '!=': case '!==': return left !== right;
            case '>': return left > right;
            case '<': return left < right;
            case '>=': return left >= right;
            case '<=': return left <= right;
            default: return false;
        }
    }

    /**
     * Retry execution with exponential backoff, jitter, and recursive protection.
     * Uses an iterative loop with an internal retry flag to prevent recursive call stacking.
     */
    private async retryExecution(
        flowName: string,
        context: AutomationContext | undefined,
        startTime: number,
        errorHandling: {
            maxRetries?: number;
            retryDelayMs?: number;
            backoffMultiplier?: number;
            maxRetryDelayMs?: number;
            jitter?: boolean;
        },
    ): Promise<AutomationResult> {
        const maxRetries = errorHandling.maxRetries ?? 3;
        const baseDelay = errorHandling.retryDelayMs ?? 1000;
        const multiplier = errorHandling.backoffMultiplier ?? 1;
        const maxDelay = errorHandling.maxRetryDelayMs ?? 30000;
        const useJitter = errorHandling.jitter ?? false;

        let lastError = 'Max retries exceeded';
        for (let i = 0; i < maxRetries; i++) {
            // Calculate delay with exponential backoff
            let delay = Math.min(baseDelay * Math.pow(multiplier, i), maxDelay);
            if (useJitter) {
                delay = delay * (0.5 + Math.random() * 0.5);
            }
            await new Promise(r => setTimeout(r, delay));

            // Execute directly without recursion into retryExecution again
            const result = await this.executeWithoutRetry(flowName, context);
            if (result.success) return result;
            lastError = result.error ?? 'Unknown error';
        }
        return { success: false, error: lastError, durationMs: Date.now() - startTime };
    }

    /**
     * Execute a flow without triggering retry logic (used by retryExecution to prevent recursion).
     */
    private async executeWithoutRetry(
        flowName: string,
        context?: AutomationContext,
    ): Promise<AutomationResult> {
        const startTime = Date.now();
        const flow = this.flows.get(flowName);

        if (!flow) {
            return { success: false, error: `Flow '${flowName}' not found` };
        }
        if (this.flowEnabled.get(flowName) === false) {
            return { success: false, error: `Flow '${flowName}' is disabled` };
        }

        const variables = new Map<string, unknown>();
        if (flow.variables) {
            for (const v of flow.variables) {
                if (v.isInput && context?.params?.[v.name] !== undefined) {
                    variables.set(v.name, context.params[v.name]);
                }
            }
        }
        if (context?.record) {
            variables.set('$record', context.record);
        }

        const runId = this.nextRunId();
        const startedAt = new Date().toISOString();
        const steps: StepLogEntry[] = [];

        try {
            const startNode = flow.nodes.find(n => n.type === 'start');
            if (!startNode) {
                return { success: false, error: 'Flow has no start node' };
            }

            await this.executeNode(startNode, flow, variables, context ?? {}, steps);

            const output: Record<string, unknown> = {};
            if (flow.variables) {
                for (const v of flow.variables) {
                    if (v.isOutput) {
                        output[v.name] = variables.get(v.name);
                    }
                }
            }

            const durationMs = Date.now() - startTime;
            this.recordLog({
                id: runId,
                flowName,
                flowVersion: flow.version,
                status: 'completed',
                startedAt,
                completedAt: new Date().toISOString(),
                durationMs,
                trigger: {
                    type: context?.event ?? 'manual',
                    userId: context?.userId,
                    object: context?.object,
                },
                steps,
                output,
            });

            return { success: true, output, durationMs };
        } catch (err: unknown) {
            const errorMessage = err instanceof Error ? err.message : String(err);
            const durationMs = Date.now() - startTime;
            this.recordLog({
                id: runId,
                flowName,
                flowVersion: flow.version,
                status: 'failed',
                startedAt,
                completedAt: new Date().toISOString(),
                durationMs,
                trigger: {
                    type: context?.event ?? 'manual',
                    userId: context?.userId,
                    object: context?.object,
                },
                steps,
                error: errorMessage,
            });
            return { success: false, error: errorMessage, durationMs };
        }
    }
}