Workflow Engine Generalisation

Two cross-service workflows exist today:

1. provision_parties — in ores.workflow; synchronous saga; blocks until all three steps (save party, save account, link) complete.
2. ore_import — in ores.ore.service; the saga logic is embedded in ore_import_handler, which calls downstream services directly.

Neither workflow is durable: if ores.workflow or ores.ore.service restarts mid-execution, the workflow is lost. There is no mechanism to resume from the last completed step.

• No durability: an in-flight workflow is lost on service restart.
• No unified view: it is impossible to list all in-flight workflows from one place; they are scattered across services.
• Inconsistent patterns: provision_parties and ore_import implement the saga pattern differently; future workflows will diverge further.
• Domain logic leakage: ore_import_handler calls ores.trading, ores.iam, etc., conflating ORE-specific logic with cross-service orchestration.
• No async support: synchronous sagas cannot model steps that take minutes or hours (e.g. compute grid execution).
• Duplication: every new workflow must re-implement step tracking, compensation, error recording, and retry.

A single ores.workflow service that:

• Orchestrates all cross-service sagas (15–20 workflows)
• Persists all workflow and step state to PostgreSQL
• Is fully event-driven: never blocks on a domain service call
• Can be restarted at any time and resume all in-flight workflows
• Provides a unified audit trail and operational view

ores.workflow is a persistent state machine driven entirely by events.

On new workflow request:
  persist workflow_instance → dispatch step[0] command → return

On step-completed event:
  load workflow_instance → update step → dispatch step[n+1] → return
  (or begin compensation if step failed)

On service restart:
  load all in-progress workflows → re-dispatch pending commands → return

ores.workflow never waits for a response. It dispatches a command (fire-and-forget NATS publish) and returns. When the domain service completes the step, it publishes a completion event. The workflow engine processes that event and advances.

This pattern is uniform for all steps regardless of duration: a millisecond DB write and an hours-long compute job are handled identically.

Every domain service handler that participates in a workflow must:

1. Extract X-Workflow-Step-Id from the command headers.
2. Check whether this step_id has already been processed (by looking for an existing record associated with that key).
3. If already processed: publish the cached completion event again and return (idempotent re-notify).
4. If not processed: execute the operation, store workflow_step_id on the created/modified entity, publish the completion event.

This guarantees that re-dispatched commands after a service restart produce the same outcome without executing the operation twice.

The simplest implementation: store workflow_step_id as a nullable column on the entity created by the step (e.g. batch.workflow_step_id, report_input_bundle.workflow_step_id). A NULL means the entity was not created by a workflow step; a non-null value is the idempotency key.

On service start, ores.workflow executes a recovery pass:

1. Queries all workflow_instance rows with state = in_progress or state = compensating.
2. For each instance, loads the current step (the workflow_step row with state = in_progress).
3. Re-dispatches the step's command using the same step_id as the X-Workflow-Step-Id header.
4. Domain services deduplicate via the idempotency key and republish their completion events.

current_step_index   integer     not null default 0
step_count           integer     not null default 0
last_event_at        timestamptz null

command_subject      text        not null default ''
command_json         text        not null default ''   -- persisted before publish
command_published_at timestamptz null                  -- set after successful publish
idempotency_key      text        not null default ''   -- = step_id string
compensation_subject text        not null default ''   -- subject for compensation command
compensation_json    text        not null default ''   -- built from step result

The request_json column already exists and is used for the command body. command_published_at is null until the publish succeeds; on restart, any step where this is non-null and state = in_progress has its command re-published.

Implement after phase 2 validates the engine. See report execution plan.