@objectstack/service-automation

The shipped provider for the kernel's automation service slot — a DAG flow execution engine implementing IAutomationService.

Slot criticality: optional (ServiceRequirementDef in @objectstack/spec/system).

Installation

pnpm add @objectstack/service-automation

Usage

The entry point is the kernel plugin AutomationServicePlugin. It seeds every built-in node executor, so it is the only plugin an automation capability needs.

import { LiteKernel } from '@objectstack/core';
import type { IAutomationService } from '@objectstack/spec/contracts';
import { AutomationServicePlugin } from '@objectstack/service-automation';

const kernel = new LiteKernel();
kernel.use(new AutomationServicePlugin());
await kernel.bootstrap();

const automation = kernel.getService<IAutomationService>('automation');
await automation.execute('escalate_high_priority_case', {
  object: 'crm_case',
  record: { id: 'case_1', priority: 'high' },
});

LiteKernel.use() is synchronous; ObjectKernel.use() returns a promise — await it there.

Plugin options

Every field of AutomationServicePluginOptions is optional.

| Option | Type | Default | Purpose | |:---|:---|:---|:---| | debug | boolean | false | Debug logging for flow execution. | | armRuntime | boolean | true | Bring up the runtime, not just the engine. false installs built-in nodes and fires automation:ready, then stops before anything is armed — no flow pull, no connector materialization, no wait-timer re-arm. | | suspendedRunStore | 'auto' \| 'memory' | 'auto' | 'auto' persists suspended runs to sys_automation_run when an ObjectQL engine is available; 'memory' never persists. | | maxLogSize | number | DEFAULT_MAX_EXECUTION_LOG_SIZE (1000) | In-memory execution-log ring buffer size. | | runSummaryLog | RunSummaryLogLevel | 'info' | Level for the one-line-per-terminal-run summary. Turning it down changes narration only — the summary is still computed, returned and persisted. | | runHistoryMaxPerFlow | number | DEFAULT_MAX_TERMINAL_RUNS_PER_FLOW (100) | Per-flow cap on terminal run-history rows; 0 disables the cap. | | credentialResolver | CredentialResolver | env-var resolver | Resolves a declarative connector's auth.credentialRef at boot. | | packageRoot | string | process.cwd() | Root that relative file refs in connector entries resolve against; reads are confined to it. |

The AGE half of run retention is declarative, not an option here: sys_automation_run declares retention: { maxAge: '30d', … } and the platform LifecycleService enforces it.

Flow Types

type declares how a flow starts:

| type | Starts when | |:---|:---| | record_change | a record is created / updated / deleted — bound on the start node | | schedule | a cron schedule fires | | screen | a user runs it interactively and supplies input | | autolaunched | another flow, an action or an API call invokes it | | api | it is exposed as an API-callable flow |

Flow Structure

A flow is a directed graph: a flat list of nodes joined by a flat list of edges. Nodes never contain child steps — branching, looping and error paths are all expressed as edges between top-level nodes.

The record-change binding lives on the start node's config ({ objectName, triggerType, condition }), not at the flow top level.

const escalateCase = {
  name: 'escalate_high_priority_case',
  label: 'Escalate High Priority Case',
  type: 'record_change',
  version: 1,
  status: 'active',

  nodes: [
    {
      id: 'start',
      type: 'start',
      label: 'Start',
      config: {
        objectName: 'crm_case',
        triggerType: 'record-after-write', // created OR updated
      },
    },
    {
      id: 'check_priority',
      type: 'decision',
      label: 'Is High Priority?',
      // Conditions are bare CEL — no braces. Each `label` MUST match an
      // out-edge's `label` exactly, or the branch cannot route (#4414).
      config: {
        conditions: [
          { label: 'High', expression: "record.priority == 'high'" },
          { label: 'Otherwise', expression: 'true' },
        ],
      },
    },
    {
      id: 'load_owner',
      type: 'get_record',
      label: 'Load Owner',
      config: {
        objectName: 'sys_user',
        filter: { id: { dialect: 'cel', source: 'record.owner_id' } },
        fields: ['id', 'name', 'email'],
        outputVariable: 'owner',
      },
    },
    {
      id: 'flag_case',
      type: 'update_record',
      label: 'Flag Case',
      config: {
        objectName: 'crm_case',
        filter: { id: { dialect: 'cel', source: 'record.id' } },
        // A computed field value is a CEL value envelope; a literal is written as it is.
        fields: { escalated: true, escalation_note: { dialect: 'cel', source: "'Escalated to ' + owner.name" } },
      },
    },
    { id: 'end', type: 'end', label: 'End' },
  ],

  edges: [
    { id: 'e1', source: 'start', target: 'check_priority', type: 'default' },
    // Branching is an edge, not a nested step list. A decision routes by
    // matching its branch `label` to an out-edge `label`.
    { id: 'e2', source: 'check_priority', target: 'load_owner', label: 'High', type: 'conditional' },
    { id: 'e3', source: 'check_priority', target: 'end', label: 'Otherwise', type: 'conditional' },
    { id: 'e4', source: 'load_owner', target: 'flag_case', type: 'default' },
    { id: 'e5', source: 'flag_case', target: 'end', type: 'default' },
  ],
};

Node Types

The built-in node type ids (FLOW_BUILTIN_NODE_TYPES, from FlowNodeAction in @objectstack/spec):

start · end · decision · assignment · loop · create_record · update_record · delete_record · get_record · http · notify · script · screen · wait · subflow · map · connector_action · parallel_gateway · join_gateway · boundary_event

type is validated against the live action registry at registerFlow(), not against a closed enum, so plugin-registered node types are equally legal.

The registry is also why FlowNodeAction is not the whole list: the ADR-0031 structured constructs parallel and try_catch ship built-in executors (builtin/parallel-node.ts, builtin/try-catch-node.ts) without appearing in that enum. See Advanced Features below.

The CRUD quartet's config — the shape most often written from memory, and the one this README used to get wrong:

| Node | config keys | |:---|:---| | get_record | objectName, filter, fields, limit, outputVariable | | create_record | objectName, fields, outputVariable | | update_record | objectName, filter, fields — no outputVariable; the executor does not read one | | delete_record | objectName, filter |

filter is an object of field/value pairs ({ id: { dialect: 'cel', source: 'record.id' } }), not an array of { field, operator, value } triples; operator objects such as { "$ne": null } are legal values. Each operand is a CEL value envelope, evaluated before the query (one that evaluates to null refuses the node), a query placeholder the query engine resolves ({current_user_id}, {today}), or a literal. There is no recordId key — select by id through filter. Unknown keys are rejected at registerFlow().

For every other node's config, and for loops, parallel blocks, subflows, waits and error handling, see the maintained reference — Flows. This README deliberately does not keep a second copy of that per-node reference.

Expressions

A flow's expressions follow its positions: a condition is bare CEL, a value is a CEL value envelope, and text is a {{ }} template.

| Where | Dialect | Write it like | |:---|:---|:---| | Start-node condition | CEL — bare, no braces | record.amount > 500 | | Edge condition | CEL — bare, no braces | record.status == 'open' | | Decision conditions[].expression | CEL — bare, no braces | order_amount > 10000 | | Value slots — create_record / update_record fields, assignment values, the maps handed to a callee: subflow / map input, script inputs, a screen's defaults, field defaultValue and recordId, each evaluated before the screen is served, a loop's (with a body) or map's collection, a notify node's recipients (each entry of a list), templateData values and sourceId (its sourceObject, actorId and payload are literals, read as written; a {…} token there is refused), and an http node's body, evaluated once before the call (its method, durable, timeoutMs and signingSecret are literals, read as written; a {…} token there is refused) | CEL value envelope; a plain string is the literal text it spells (recordId's literal is the id, a string; a collection's literal is an inline array, and any string there is refused; a recipient entry that evaluates to a list adds each of its recipients) | { dialect: 'cel', source: "'Follow up on ' + record.name" }, 'open', collection: { dialect: 'cel', source: 'tasks' } | | A get_record / update_record / delete_record filter operand — a field's value, an operator's comparand, a list comparand's element | CEL value envelope, evaluated before the query (null refuses the node); or a literal — a query placeholder ({current_user_id}, {today}) among them, handed to the query engine untouched; any other {…} token is refused | { id: { dialect: 'cel', source: 'record.id' }, due_date: { $lt: '{today}' } } | | Text slots — notify title / message / actionUrl, http url and each headers value, screen title / description, end message | Template — {{ }} holes (ADR-0032 §3; a single-brace token is refused) | 'Deal won: {{ record.name }}', '{{ record.amount \| currency }}' |

A value slot refuses every {…} template token — registerFlow() and objectstack validate name its CEL spelling. A date macro is its CEL string form, on the UTC calendar:

| You wrote | Write instead | |:---|:---| | '{TODAY()}' | { dialect: 'cel', source: 'isoDate(today())' } | | '{TODAY() + 90}' | { dialect: 'cel', source: 'isoDate(daysFromNow(90))' } | | '{TODAY() - 3}' | { dialect: 'cel', source: 'isoDate(daysAgo(3))' } | | '{NOW()}' | { dialect: 'cel', source: 'isoDatetime(now())' } | | '{NOW() + 2}' | { dialect: 'cel', source: 'isoDatetime(addDays(now(), 2))' } — daysFromNow would land on midnight | | '{TODAY() + days}' | { dialect: 'cel', source: 'isoDate(addDays(today(), days))' } — an absent days, or one that is not a number, now fails the run where the template silently added 0 |

Inside an object or list value, and in the legacy assignment shapes, an envelope is data: build the whole value as one envelope (a CEL map or list literal, its mixed values wrapped in dyn(…)), or move the assignment into the assignments map — an http body included: { dialect: 'cel', source: "{'id': record.id, 'text': dyn('Done: ' + record.title)}" }.

A loop's or a map's collection must evaluate to a list: anything else, null included, fails the node, as a token that resolved to nothing did, so a guard answering [] (has(vars.tasks) ? vars.tasks : []) iterates nothing where the list may be absent. Its bare variable name (collection: 'tasks'), which the structured loop and map never read, is refused too and named in an envelope. A loop with no body keeps its legacy flat-graph form, which reads collection as a bare variable name.

A subflow or map input, and a script's inputs, hand their values to a callee: each envelope is evaluated in the calling flow's scope (on a map, once per item, with the item variable bound) and handed over as the value it computes — a list stays a list. A value handed to a child flow is supplied, null included, so it wins over the child variable's defaultValue. Where the template resolved a token to nothing it handed nothing, and the default applied; the guarded form has(vars.x) ? vars.x : null hands null. To keep the default for an absent value, write it in the guard (has(vars.x) ? vars.x : 'standard'), or leave the key out where the value is never meant to be supplied. A function is handed null there, where the template handed undefined.

The run's user is current_user in every flow CEL expression: id, positions, organizationId, isPlatformAdmin, nothing more. In a run with no user (a schedule, a record change made by a system write) it is null, never a stand-in user. So {$User.Id} is refused in a value slot: write { dialect: 'cel', source: 'current_user.id' }, or, in a flow that can run without a user, current_user != null ? current_user.id : null (with no user it writes null, which on update_record clears the stored value). Every other {$User.<path>} ({$User.Email}) never resolved and is refused too: read the user record by current_user.id with a get_record on sys_user.

The two failure modes to memorize:

1. A plain string in a value slot — due_date: 'TODAY() + 7' or owner: 'current_user.id' writes that text into the field. Write it as a CEL envelope: { dialect: 'cel', source: 'isoDate(daysFromNow(7))' }.
2. Braces put into a condition — '{record.amount} > 500'. Since #4336 conditions reject this loudly: registerFlow() / objectstack validate refuse the flow with a CEL error naming the reference. Before that they were compared as text and were silently always-true or always-false.

There is no {!…} dialect. That is Salesforce syntax; the platform has never parsed it.

Service API

IAutomationService (from @objectstack/spec/contracts) declares two required members and a set of optional ones; this package implements them all.

import type { IAutomationService } from '@objectstack/spec/contracts';

// required
//   execute(flowName, context?)          -> Promise<AutomationResult>
//   listFlows()                          -> Promise<string[]>
// optional
//   registerFlow?(name, definition)      -> void
//   unregisterFlow?(name)                -> void
//   getFlow?(name)                       -> Promise<FlowParsed | null>
//   toggleFlow?(name, enabled)           -> Promise<void>
//   listRuns?(...)                       -> run history
//   getRun?(runId)                       -> Promise<ExecutionLog | null>
//   resume?(runId, signal?)              -> Promise<AutomationResult>
//   listSuspendedRuns?()                 -> suspended-run summaries
//   getSuspendedScreen?(runId)           -> Promise<ScreenSpec | null>
//   getActionDescriptors?()              -> ActionDescriptor[]
//   getConnectorDescriptors?()           -> ConnectorDescriptor[]
//   getFlowRuntimeStates?()              -> FlowRuntimeState[]
//   canonicalizeStoredFlow?(name, definition)

Execute a flow

execute takes the flow's machine name and an optional AutomationContext — it does not take an options object, and there is no inputs key.

const result = await automation.execute('escalate_high_priority_case', {
  record: { id: 'case_1', priority: 'high' },
  object: 'crm_case',
  event: 'on_update',
  userId: 'usr_123',
});

if (result.success) {
  console.log('output:', result.output, 'in', result.durationMs, 'ms');
} else {
  console.error('failed:', result.error, result.code);
}

AutomationResult fields: success, output?, error?, durationMs?, code?, status?, runId?, screen?, successMessage?, errorMessage?, summary?. The machine-readable classification is code (not errorCode) — resume refusals such as RUN_NOT_FOUND, STORE_UNAVAILABLE, RESUME_IN_PROGRESS, plus the trigger-time FLOW_DISABLED / FLOW_NO_START_NODE.

⚠️ runAs on AutomationContext is derived by the engine from the flow definition — callers do not set it.

Register and inspect flows

automation.registerFlow?.('escalate_high_priority_case', escalateCase);

const names = await automation.listFlows();     // string[] of machine names
const parsed = await automation.getFlow?.('escalate_high_priority_case');

// A flow registered here is authored in the deployment, so its switch is its
// own `status`: re-register it with 'obsolete' to switch it off ('active' arms it).
automation.registerFlow?.('escalate_high_priority_case', { ...escalateCase, status: 'obsolete' });

// `toggleFlow` switches PACKAGED flows only — one a code package ships — by
// writing the ADR-0126 activation ledger. It refuses a flow no package ships
// with RESOURCE_CONFLICT / 409 and changes nothing.
await automation.toggleFlow?.('a_packaged_flow', false);

registerFlow validates against the live action registry and rejects unknown config keys. There is no registerTrigger method — a flow's trigger is declared on its start node (record_change) or by its type, and arming happens at registration.

REST API

Served by the runtime dispatcher's /automation domain when this service occupies the slot (paths shown with the /api/v1 wire prefix):

There is no flow-list route here: flows are metadata (ADR-0106), so list them with GET /api/v1/meta/flow. The former GET /api/v1/automation list was retired.

POST   /api/v1/automation                              # create a flow
GET    /api/v1/automation/actions                      # action descriptors
GET    /api/v1/automation/connectors                   # connector descriptors
GET    /api/v1/automation/_status                      # runtime status
GET    /api/v1/automation/:name                        # get one flow
PUT    /api/v1/automation/:name                        # update a flow
DELETE /api/v1/automation/:name                        # delete a flow
POST   /api/v1/automation/:name/trigger                # execute a flow
POST   /api/v1/automation/:name/toggle                 # enable / disable a packaged flow (a customer flow: its status, via PUT)
GET    /api/v1/automation/:name/runs                   # list runs
GET    /api/v1/automation/:name/runs/:runId            # run detail
GET    /api/v1/automation/:name/runs/:runId/screen     # screen spec of a parked run
POST   /api/v1/automation/:name/runs/:runId/resume     # resume a parked run
POST   /api/v1/automation/:name/runs/:runId/cancel              # cancel a suspended run — PLATFORM_ADMIN only (isSystem bypass), body { reason? }; 501 if unimplemented
POST   /api/v1/automation/:name/runs/:runId/restore-suspension  # re-arm a run stranded by a failed resume — same gate; body { reason? }; 501 if unimplemented
POST   /api/v1/automation/trigger/:name                # legacy execute shape

Advanced Features

loop, parallel and try_catch are structured control-flow constructs (ADR-0031). Each owns its body as a single-entry/single-exit region carried in config — a nested { nodes, edges } sub-graph, not a steps array — so the outer graph stays acyclic. A region runs in the enclosing variable scope; the container's ordinary out-edges are the continuation.

Loop

{
  id: 'notify_each',
  type: 'loop',
  label: 'For each task',
  config: {
    collection: { dialect: 'cel', source: 'tasks' }, // the list: a CEL value envelope, or an inline array
    iteratorVariable: 'task',   // current item, visible inside the body
    indexVariable: 'i',         // optional zero-based index
    maxIterations: 500,         // hard cap (clamped to the engine ceiling)
    body: {
      nodes: [{ id: 'send', type: 'notify', label: 'Notify', config: { /* … */ } }],
      edges: [],
    },
  },
}

Parallel Execution

Branches run concurrently and join implicitly when all complete — there is no author-visible split/join gateway.

{
  id: 'fan_out',
  type: 'parallel',
  label: 'Notify in parallel',
  config: {
    branches: [ // ≥ 2 regions
      { name: 'Email', nodes: [{ id: 'email', type: 'notify', label: 'Email', config: { /* … */ } }], edges: [] },
      { name: 'Slack', nodes: [{ id: 'slack', type: 'notify', label: 'Slack', config: { /* … */ } }], edges: [] },
    ],
  },
}

Error Handling

{
  id: 'guarded',
  type: 'try_catch',
  label: 'Charge with fallback',
  config: {
    try: { nodes: [{ id: 'charge', type: 'http', label: 'Charge', config: { /* … */ } }], edges: [] },
    catch: { nodes: [{ id: 'flag', type: 'update_record', label: 'Flag failure', config: { /* … */ } }], edges: [] },
    errorVariable: '$error',
    retry: { maxRetries: 3, backoffMs: 1000, backoffMultiplier: 2 },
  },
}

Subflows

{
  id: 'validate',
  type: 'subflow',
  label: 'Validate Address',
  config: {
    flowName: 'validate_address',
    // Each value is a CEL value envelope, evaluated in this flow's scope.
    input: {
      street: { dialect: 'cel', source: 'input.street' },
      city: { dialect: 'cel', source: 'input.city' },
    },
    outputVariable: 'validated_address',
  },
}

Wait

wait suspends the run durably. Its contract is the node-level waitEventConfig block — not config:

{
  id: 'hold',
  type: 'wait',
  label: 'Wait 24h',
  waitEventConfig: {
    eventType: 'timer',          // 'timer' | 'signal' | 'webhook' | 'manual' | 'condition'
    timerDuration: 'PT24H',      // ISO 8601 duration
  },
}

The node resumes down its ordinary out-edges; there is no nextSteps key.

BPMN parallel_gateway / join_gateway / boundary_event remain in the protocol as the interop representation and map onto these constructs on import/export — they are not the native authoring model.

Best Practices

1. Keep Flows Simple: Break complex logic into multiple flows
2. Use Descriptive Names: Name flows and steps clearly
3. Handle Errors: Always include error handling for critical operations
4. Test Thoroughly: Test flows with various input scenarios
5. Monitor Performance: Track flow execution times and optimize slow flows
6. Version Control: Store flow definitions in version control
7. Document Intent: Add descriptions to flows and steps

Performance Considerations

• Parallel Execution: DAG engine automatically parallelizes independent steps
• Batch Processing: Use loop steps efficiently for large collections
• Query Optimization: Filter queries early to reduce data volume
• Async Execution: Long-running flows execute asynchronously

Exports

import {
  AutomationEngine, AutomationServicePlugin, createPackageFileLoader,
  InMemorySuspendedRunStore, ObjectStoreSuspendedRunStore, SysAutomationRun,
  installBuiltinNodes, registerLogicNodes, registerCrudNodes,
  registerScreenNodes, registerHttpNodes, registerConnectorNodes,
  resolveRunDataContext, stampSystemInsertOwner, UnscopedRunDataAccessError,
  summarizeRun, formatRunSummaryLine,
  DEFAULT_MAX_EXECUTION_LOG_SIZE, DEFAULT_MAX_TERMINAL_RUNS_PER_FLOW,
  MAX_PERSISTED_HISTORY_STEPS,
} from '@objectstack/service-automation';

Types: AutomationEngineOptions, AutomationServicePluginOptions, RunSummaryLogLevel, NodeExecutor, NodeExecutionResult, SuspensionRelease, SuspensionReleaseReason, FlowTrigger, FlowTriggerBinding, RegisteredConnector, SuspendedRun, SuspendedRunStore, SuspendedRunStoreEngine, ObjectStoreSuspendedRunStoreOptions, RunRecord, StepLogEntry, UnknownNodeTypeAuditEntry, RunDataContext, RunIdentityContext, ConnectorProviderFactory, ConnectorProviderContext, ConnectorMaterialization, ConnectorMaterializationHandler, ConnectorOrigin, ConnectorState, ConnectorDescriptor, ConnectorActionDescriptor, ConnectorActionHandler, ConnectorActionContext.

The connector types are re-exports from @objectstack/spec/integration — connector plugins should import them from there rather than coupling to this engine.

AutomationEngine is the engine underneath the plugin, exported for hosts that build their own kernel integration; AutomationServicePlugin is the entry point for everyone else. The built-in node installers are functions, not plugins — the platform's foundational nodes are built in, not installed.

License

Apache-2.0. See LICENSING.md.

See Also

• @objectstack/spec/automation
• Automation — the automation docs section: hooks, flows, workflows, approvals, webhooks, connectors
• Automation Protocol — the complete schema reference for the automation protocol