support process run formatting

Author: paed01

AGENTS.md

@@ -73,6 +73,7 @@ Documented in `docs/Extend.md` and `docs/Extension.md`:
 - **Activities are armed through the inbound queue, then run consumer-driven.** Both start activities (`isStart`, no inbound trigger) and link catch events are armed by `Activity.init()`: it mints an executionId, emits the `init` event (whose placeholder in the process's `postponed` set blocks premature completion), increments an `initialized` counter, and queues a non-persistent `activity.init` message carrying that id (plus the activity `id`) on the activity's own `inbound-q`. The run is then driven by the inbound consumer: the idempotent `consumeInbound()` asserts the consumer when there are inbound triggers, or — even without sequence-flow triggers — when the activity is `initialized` (it reads the `initialized` counter rather than probing `inbound-q` for a pending-message count, so no synchronous smqp-only property is touched), and `_onInbound`'s `activity.init` case decrements the counter and calls `run()` with the carried id. `ProcessExecution._start` arms start activities this way (`init()` then `consumeInbound()`, no direct `run()`). A throwing link publishes `activity.link`; the catch's construction-time inbound-trigger handler calls the catch's own `init()` to arm it identically — there is no `activity.relink`. The `initialized` getter reads the counter (exec-state, not persisted); since the `activity.init` trigger is `persistent: false` it is dropped on recover and never re-consumed, so counter and trigger reset together (a stop/resume while armed cannot desync them).
 - **`run()` executionId resolution.** `run(runContent)` uses `runContent.initExecutionId` when `runContent.id` equals the activity's own id, otherwise mints a fresh `getUniqueId(id)`. The init/link handoff (`_onInbound`) always passes a matching `id`, so the reserved id is honoured; every other caller (e.g. delegated event runs passing `run(content.message)`) carries no matching id and gets a fresh one. `run()` does **not** peek the inbound queue, and there is no `initExecutionId` exec-state key. `initExecutionId` is **destructured out** of `runContent` so it never reaches the `run.enter`/`run.start` content (it would otherwise leak through `_createMessage`, which only overwrites `id`/`type`/flags, into every downstream message and saved state).
 - **Extension `activate`/`deactivate` fire symmetrically on activities and processes.** Both `Activity` and `Process` call their extensions' `activate(message)` on run enter / redelivered execute / resume and `deactivate(message)` on run leave / stop, each guarded by `if (this.extensions)`. Keep the two in sync: a lifecycle hook added on one side generally belongs on the other. `Definition` does not load or invoke extensions.
+- **Run transitions are formatted on both activities and processes.** `Activity._onRunMessage` and `Process._onRunMessage` route every non-passthrough run message through `this.formatter.format(message, cb)` (`MessageFormatter`) before `_continueRunMessage` runs the actual switch; the callback restores status, applies enriched content, or `emitFatal`s on a format error. Formatting is **unconditional** (not gated on `this.extensions`) so any handler — extension or not — can enrich a run by publishing a `format` message, optionally with an `endRoutingKey` to pause the run asynchronously until the matching end key arrives. The run-q message stays unacked across the wait, which is what serializes the run; nothing is `await`ed in the consumer.
 - **Multiple start events are mutually exclusive entry points.** The first start event to fire discards the others still armed, so two start branches can never both run.
 - **`stateVersion`.** `Definition.getState()` stamps `stateVersion` (the package major, hardcoded in `src/constants.js`); recovering an older major triggers migrations (e.g. start event reconciliation on resume). Unstamped legacy states are treated as version `0`. Bump the constant on each major release.
 

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Additions
 
 - process extensions are now activated and deactivated accordingly
+- process run messages can be formatted, including asynchronously, the same way as activities
 
 ## v18.0.2 - 2026-06-24
 

dist/process/Process.js

@@ -8,10 +8,12 @@ var _ProcessExecution = require("./ProcessExecution.js");
 var _shared = require("../shared.js");
 var _Api = require("../Api.js");
 var _EventBroker = require("../EventBroker.js");
+var _MessageFormatter = require("../MessageFormatter.js");
 var _messageHelper = require("../messageHelper.js");
 var _Errors = require("../error/Errors.js");
 var _constants = require("../constants.js");
 const K_LANES = Symbol.for('lanes');
+const K_FORMATTER = Symbol.for('formatter');
 
 /**
  * Owns one `<bpmn:process>`. Wraps the structural definition and orchestrates flow traversal,
@@ -49,12 +51,14 @@ function Process(processDef, context) {
     broker,
     on,
     once,
-    waitFor
+    waitFor,
+    emitFatal
   } = (0, _EventBroker.ProcessBroker)(this);
   this.broker = broker;
   this.on = on;
   this.once = once;
   this.waitFor = waitFor;
+  this.emitFatal = emitFatal;
   this[_constants.K_MESSAGE_HANDLERS] = {
     onApiMessage: this._onApiMessage.bind(this),
     onRunMessage: this._onRunMessage.bind(this),
@@ -84,6 +88,14 @@ Object.defineProperties(Process.prototype, {
       return this[_constants.K_EXTENSIONS];
     }
   },
+  formatter: {
+    get() {
+      let formatter = this[K_FORMATTER];
+      if (formatter) return formatter;
+      formatter = this[K_FORMATTER] = new _MessageFormatter.Formatter(this);
+      return formatter;
+    }
+  },
   stopped: {
     get() {
       return this[_constants.K_STOPPED];
@@ -292,14 +304,28 @@ Process.prototype._deactivateRunConsumers = function deactivateRunConsumers() {
 };
 
 /** @internal */
-Process.prototype._onRunMessage = function onRunMessage(routingKey, message) {
+Process.prototype._onRunMessage = function onRunMessage(routingKey, message, messageProperties) {
+  if (routingKey === 'run.resume') {
+    return this._onResumeMessage(message);
+  }
+  const preStatus = this[_constants.K_STATUS];
+  this[_constants.K_STATUS] = 'formatting';
+  return this.formatter.format(message, (err, formattedContent, formatted) => {
+    this[_constants.K_STATUS] = preStatus;
+    if (err) {
+      return this.emitFatal(err, message.content);
+    }
+    if (formatted) message.content = formattedContent;
+    this._continueRunMessage(routingKey, message, messageProperties);
+  });
+};
+
+/** @internal */
+Process.prototype._continueRunMessage = function continueRunMessage(routingKey, message) {
   const {
     content,
     fields
   } = message;
-  if (routingKey === 'run.resume') {
-    return this._onResumeMessage(message);
-  }
   this[_constants.K_STATE_MESSAGE] = message;
   switch (routingKey) {
     case 'run.enter':

docs/Extension.md

@@ -35,6 +35,8 @@ In some cases it may be required to add some extra data when an activity execute
 
 The basic flow is to publish a formatting message on the activity format queue.
 
+The same mechanism works on the **process**: publish a `format` message (optionally with an `endRoutingKey`) in response to a `process.*` event, and the process run pauses at the next transition until the matching end key is published — see the asynchronous example below, which applies equally to process events.
+
 ```javascript
 import * as elements from 'bpmn-elements';
 import { BpmnModdle } from 'bpmn-moddle';

src/process/Process.js

@@ -2,6 +2,7 @@ import { ProcessExecution } from './ProcessExecution.js';
 import { getUniqueId } from '../shared.js';
 import { ProcessApi } from '../Api.js';
 import { ProcessBroker } from '../EventBroker.js';
+import { Formatter } from '../MessageFormatter.js';
 import { cloneMessage, cloneContent, cloneParent } from '../messageHelper.js';
 import { makeErrorFromMessage } from '../error/Errors.js';
 import {
@@ -17,6 +18,7 @@ import {
 } from '../constants.js';
 
 const K_LANES = Symbol.for('lanes');
+const K_FORMATTER = Symbol.for('formatter');
 
 /**
  * Owns one `<bpmn:process>`. Wraps the structural definition and orchestrates flow traversal,
@@ -47,11 +49,12 @@ export function Process(processDef, context) {
   this[K_STATUS] = undefined;
   this[K_STOPPED] = false;
 
-  const { broker, on, once, waitFor } = ProcessBroker(this);
+  const { broker, on, once, waitFor, emitFatal } = ProcessBroker(this);
   this.broker = broker;
   this.on = on;
   this.once = once;
   this.waitFor = waitFor;
+  this.emitFatal = emitFatal;
 
   this[K_MESSAGE_HANDLERS] = {
     onApiMessage: this._onApiMessage.bind(this),
@@ -83,6 +86,14 @@ Object.defineProperties(Process.prototype, {
       return this[K_EXTENSIONS];
     },
   },
+  formatter: {
+    get() {
+      let formatter = this[K_FORMATTER];
+      if (formatter) return formatter;
+      formatter = this[K_FORMATTER] = new Formatter(this);
+      return formatter;
+    },
+  },
   stopped: {
     get() {
       return this[K_STOPPED];
@@ -278,13 +289,28 @@ Process.prototype._deactivateRunConsumers = function deactivateRunConsumers() {
 };
 
 /** @internal */
-Process.prototype._onRunMessage = function onRunMessage(routingKey, message) {
-  const { content, fields } = message;
-
+Process.prototype._onRunMessage = function onRunMessage(routingKey, message, messageProperties) {
   if (routingKey === 'run.resume') {
     return this._onResumeMessage(message);
   }
 
+  const preStatus = this[K_STATUS];
+  this[K_STATUS] = 'formatting';
+
+  return this.formatter.format(message, (err, formattedContent, formatted) => {
+    this[K_STATUS] = preStatus;
+    if (err) {
+      return this.emitFatal(err, message.content);
+    }
+    if (formatted) message.content = formattedContent;
+    this._continueRunMessage(routingKey, message, messageProperties);
+  });
+};
+
+/** @internal */
+Process.prototype._continueRunMessage = function continueRunMessage(routingKey, message) {
+  const { content, fields } = message;
+
   this[K_STATE_MESSAGE] = message;
 
   switch (routingKey) {