Workflow Steps

Steps in Causet workflows are state transitions — either within a saga (advancing the entity’s step counter) or across entities via submit chains. Both mechanisms are built on the same event and intent infrastructure.

Saga Step Transitions

A saga step advances when the entity receives an event that matches a step’s on: field. The saga machinery writes the step’s set: values to the entity snapshot under the configured state_path.

Each transition is driven by a ledger event — not by the runtime calling a function. The saga only advances when the event is committed to the ledger and processed by the projection worker.

Submit Chains

A submit chain fires follow-up intents from one entity to another via op: submit in side_effects.

side_effects:
  rules:
    - name: trigger_artist_match
      when:
        - path: /status
          op: eq
          value: "parsed"
      then:
        - op: submit
          intent_type: MATCH_ARTIST
          payload:
            artist_raw: intent.artist_raw
            import_id: intent.import_id

When TICKET_PARSE_STARTED commits and this side effect fires, a new MATCH_ARTIST intent is submitted to the artist entity. That intent runs through its own preflight, core, and side_effects rules independently.

Execution Order

Side effects execute after core rules succeed and after the event is committed to the ledger. The sequence is:

1. Preflight rules evaluate — intent rejected if any fail
2. Core rules execute — state mutations applied
3. Event committed to ledger_events
4. Side effects fire (async, after ledger commit)

This means by the time a submit chain fires, the source entity’s state is already persisted.

Parallel Steps

Multiple op: submit operations in the same side_effects rule fire in parallel:

side_effects:
  rules:
    - name: fan_out_enrichment
      when:
        - path: /status
          op: eq
          value: "matched"
      then:
        - op: submit
          intent_type: FETCH_VENUE_DATA
          payload:
            venue_id: intent.venue_id
            import_id: intent.import_id
        - op: submit
          intent_type: FETCH_SETLIST_DATA
          payload:
            artist_id: intent.artist_id
            import_id: intent.import_id
        - op: submit
          intent_type: FETCH_TOUR_CONTEXT
          payload:
            tour_id: intent.tour_id
            import_id: intent.import_id

All three intents are submitted concurrently. Each runs through its own rules independently. The originating entity is not aware of their completion status — if you need to track completion, use a saga on the originating entity and advance it when each sub-intent emits a completion event.

op: schedule — Delayed Steps

The op: schedule operation enqueues an event or intent to fire after a delay:

side_effects:
  rules:
    - name: schedule_reminder
      when:
        - path: /status
          op: eq
          value: "registered"
      then:
        - op: schedule
          event_type: SHOW_REMINDER_SENT
          delay_seconds: 86400
          payload:
            user_id: intent.user_id
            show_id: intent.show_id

delay_seconds is relative to the time the intent is processed. Scheduled events fire as regular ledger events and can advance sagas or trigger further side effects.

Step Idempotency

Submitted intents must handle duplicate execution. If the same op: submit fires twice due to infrastructure retry:

• Preflight should reject duplicate processing using entity state checks
• Core should use idempotent operations (set rather than add where possible)

A common guard in preflight:

preflight:
  rules:
    - name: not_already_matched
      when:
        - path: /match_status
          op: neq
          value: "matched"
      error:
        code: ALREADY_MATCHED
        message: "artist already matched for this import"

This prevents the MATCH_ARTIST intent from reprocessing if it fires twice.

Error Propagation

When a submitted intent’s preflight rejects, the saga does not automatically advance. The rejection returns an error to the intent processor, but the originating entity is unaffected.

This is by design: submit chains are fire-and-forget. The originating entity does not observe the outcome of submitted intents unless a completion event is explicitly emitted back and a saga step listens for it.

Manual Saga Advancement

If a saga stalls because a downstream step never completes, you can submit a corrective intent that emits the expected event:

actions:
  FORCE_TICKET_MATCH:
    intent_type: FORCE_TICKET_MATCH
    state: ticket_import
    entity_key: import_id
    preflight:
      rules:
        - name: must_be_parsing
          when:
            - path: /_tmp/import_saga/step
              op: eq
              value: 1
          error:
            code: WRONG_SAGA_STATE
            message: "ticket import must be in parsing state"
    core:
      rules:
        - name: force_match
          when: {}
          then:
            - op: set
              path: /match_status
              value: "forced"
    events:
      - type: TICKET_ARTIST_MATCHED
        payload:
          import_id: intent.import_id
          forced: true

This emits TICKET_ARTIST_MATCHED, which advances the saga from parsing to matched.

Tracking Step Progress via Projections

To observe step transitions, project saga state fields that update on each relevant event:

projections:
  import_pipeline_status:
    source_events:
      - TICKET_PARSE_STARTED
      - TICKET_ARTIST_MATCHED
      - TICKET_METADATA_ENRICHED
      - TICKET_IMPORT_COMPLETE
      - TICKET_IMPORT_FAILED
    target:
      table: import_pipeline_status
      primary_key: [import_id]
    fields:
      import_id: TEXT
      current_step: INT
      last_event: TEXT
      last_updated_at: BIGINT
    derive:
      import_id: event.entity_id
      current_step: entity._tmp.import_saga.step
      last_event: event.event_type
      last_updated_at: event.ts

Query this table to find imports stuck in a given step for longer than expected:

SELECT import_id, current_step, last_updated_at
FROM import_pipeline_status
WHERE current_step < 4
  AND last_updated_at < extract(epoch from now() - interval '1 hour') * 1000;