Workflow Engine

Audience: Engineers building or debugging multi-step workflows.
What you will accomplish: Understand how workflows run as durable state machines — with gates, idempotency, crash recovery, and compensation — so you can define, test, and operate them.
Prerequisites: Read MCP Surface first to understand tool risk tiers.
Estimated time: ~12 minutes.

The workflow engine (cam.core.orchestrator) turns multi-step operational processes into durable, resumable state machines. A workflow can pause at a human gate for hours or days, survive server restarts, and resume exactly where it left off.

What is a workflow engine?

A workflow engine turns a multi-step operational process (like client intake: extract → create contact → create matter → compute deadlines → draft email → await approval → send) into a durable state machine.

"Durable" means every step's state is persisted to the database. If the server crashes mid-workflow, it picks up exactly where it left off on restart. If a step needs human approval, the workflow pauses (sometimes for days) and resumes when the approver acts.

This is fundamentally different from a script: scripts run start-to-finish or fail; workflows survive failures, wait for humans, and never silently skip a step.

State machine

queued ──► running ──► succeeded
              │   ├──► awaiting_approval ──►(approve)──► running
              │   │                         └──►(reject)──► rejected
              │   ├──►(transient, attempts<cap)──► running   [retry]
              │   └──►(fatal | attempts==cap)──► parked
              └── parked ──►(re-drive)──► running
                           └──►(cancel)──► cancelled

Step states: pending → running → succeeded | failed

• failed loops to running on retry or escalates the run to parked
• compensated marks a reversed step
• skipped marks a bypassed step

Idempotency — exactly-once external effects

Every step's external effect is keyed on "{run_id}:{step_name}".

Before executing, the engine checks the IdempotencyStore:

• Miss → execute handler, reserve, record output
• Hit → reuse stored output, skip handler (no re-effect)

This gives exactly-once external effect under at-least-once execution. Even if two Celery workers pick up the same run, only one will execute each step.

Gate lifecycle

Human-in-the-loop gates are first-class citizens in the state machine, not afterthoughts.

1. 1

   Step 1: Engine hits a GATE:* step

   The run transitions to awaiting_approval. No further steps execute.

   Run status → awaiting_approval
2. 2

   Step 2: GateRequest row created

   Persistence layer writes a GateRequest with tokens for each approval channel (MCP, web, email).

   Tokens issued per channel
3. 3

   Step 3: Approver resolves the gate

   Approver calls one channel's endpoint with the raw token.

   POST /approvals/{token}
   { "decision": "approve", "comment": "LGTM" }

4. 4

   Step 4: resolve_gate() validates

   Signature → expiry → single-use (atomic) → authz. Any failure returns 403 and does not advance the run.

   Validation chain
5. 5

   Step 5: Run resumes or terminates

   Approve → run status → running, re-enqueued. Reject → run status → rejected.

   Decision recorded in audit log

Gate configuration per workflow:

• required_role — who can approve (e.g., attorney)
• channels — how the approver is notified (mcp, web, email)
• ttl_seconds — how long the gate stays open before auto-reject

Retry and compensation

Retry policy

• TransientError → retry with exponential full-jitter backoff, max 5 attempts
• FatalError or attempts exhausted → park immediately
• RateLimitError → respect Retry-After header, then retry

Compensation

On park, registered compensation hooks run in reverse order (best-effort). If a step created a contact and a matter, and the matter step fails fatally, the compensation hook for create_contact can delete the orphaned contact.

Crash recovery

On startup, the engine sweeps all runs:

Because every external effect is idempotent, re-enqueuing a running run is safe even if the previous worker died mid-step.

Workflow definition example

@workflow("intake", version=1)
class IntakeWorkflow:
    steps = [
        "parse_lead",        # extract fields from email/form
        "dedupe_contact",    # CRM lookup — ambiguous → blocking gap
        "create_contact",    # write (confirm)
        "create_matter",     # write (confirm)
        "compute_deadlines", # deadline engine
        "open_tasks",        # checklist
        "draft_welcome",     # email.draft — no send yet
        "GATE:human_review", # pause for approval
        "send_welcome",      # email.send — post-gate only
    ]

Each step is a handler method on the workflow class. The engine calls them in order, managing state transitions and idempotency automatically.

Workflow registration

Workflows are registered at startup with their service dependencies:

from cam.core.workflows.intake.workflow import register_intake_workflow
from cam.core.services import ServiceContainer
 
services = ServiceContainer(...)  # inject connectors, stores, etc.
register_intake_workflow(services)

The deprecated configure_services() shim exists only for backward compatibility and logs a deprecation warning.

Step handler contract

Every step handler receives:

• ctx: StepContext — run_id, step_name, inputs, services
• Returns: StepResult — status, output, next_step_hint, compensation_hook

async def create_contact(ctx: StepContext) -> StepResult:
    contact = await ctx.services.crm.upsert_contact(ctx.inputs["contact"])
    return StepResult(
        status=StepStatus.SUCCEEDED,
        output={"contact_id": contact.id},
        idempotency_key=f"{ctx.run_id}:create_contact",
    )

Distributed execution

For production multi-worker deployments, the engine uses Redis SETNX distributed locking:

• Before iterating steps, the worker attempts to acquire cam:run_lock:{run_id}
• Only one worker executes a given run at a time
• Lock is released on completion, park, or gate

This enables N Celery workers to process independent runs concurrently without collision.

Testing workflows

The engine includes an InMemoryRunStore for fast, deterministic testing:

from cam.core.orchestrator import WorkflowEngine, InMemoryRunStore
 
engine = WorkflowEngine(store=InMemoryRunStore())
run = await engine.start("intake", inputs={...})
 
# Simulate gate approval
await engine.resolve_gate(run.id, token="test-token", decision="approve")
 
# Assert final state
assert run.status == RunStatus.SUCCEEDED

All workflow tests run without PostgreSQL or Redis, completing in milliseconds.

Observability

Every workflow step starts an OpenTelemetry span:

• Span name = workflow.{workflow_name}.step.{step_name}
• run_id tag on every span
• Parent span = the workflow run span
• Error spans capture the full exception chain and ConnectorError taxonomy

Structured logs (structlog) include:

• event — step lifecycle (step_start, step_complete, step_retry, gate_open, gate_resolve)
• run_id, step_name, attempt
• duration_ms
• PII-scrubbed inputs/outputs

Current limitations

Common mistakes

Verification

After reading this page, verify your understanding:

1. Describe what happens when a transient error occurs on step 3 of a 9-step workflow
2. Explain why re-enqueuing a running run after a crash is safe
3. Name the three gate channels and how single-use atomicity is enforced
4. Show how to write a step handler that returns a compensation hook

Next steps

• Client Intake Workflow — End-to-end example with gates
• Status Update Emails — Webhook + sweep dual trigger
• Document Generation — Template → DOCX/PDF pipeline
• Document Routing — Privilege-aware classification and filing

Last updated: 2026-06-01