executeWorkflowContractThe single function every workflow contract runs through. Reads a row from workflow_definitions, walks 7 stages, writes audit rows to workflow_step_log + workflow_run_log + (optionally) reflexion_log + (when risk ≥ 3) proposed_actions + (per verify) workflow_verify_results. Caller decides what triggered the run; the runner decides what happens next.
POST /api/workflow/execute ?preview=trueDirect invocation. preview=true sets dry_run=true in opts — runs through all stages but skips writes inside executeFanOut (each step traces with status:'dry_run'). Used by ops console + manual triggers. src/index.ts:14304.
drainEvents() → subscription match → executeWorkflowContractCron-fired drainer reads events table since last cursor, matches against event_subscriptions, applies input_mapper JSONPath translation, then invokes the runner with invoked_by: "event:<type>:<id>". KV concurrency lock prevents double-fire. See substrate-event-ledger.
execute_workflow chat tool (admin-gated)Tool registered in palette; calls into the same executeWorkflowContract entry. Most chat surfaces stage a workflow_* proposed_action first and run on approval (HITL invariant per ADR-031).
scheduled handler → verify scheduler + cron-triggered drainsScheduled workflows (e.g. monthly_margin_review) fire from the cron handler. Also: the verify-check scheduler at 45 5 * * * reads pending workflow_verify_results rows and runs the verify SQL. src/index.ts:33255.
executeFanOut| kind | status | what it does today | side effects |
|---|---|---|---|
| kv_invalidate | REAL | iterates t.keys[], calls env.CACHE.delete(k) | KV deletes |
| stage_proposed_action | REAL | INSERT proposed_actions with status='pending'. R560 fix: INSERT failure now throws (was silent .catch) | proposed_actions row |
| d1_write | STUB | echoes {kind, note:'stub_not_implemented'}; marks step status='stub' | none |
| ns_push | STUB | same as d1_write — the Cal-Maine-class bug R560 closed (was reporting ok with zero NS writes) | none |
| http_call | STUB | stub | none |
| chat_tool | STUB | stub | none |
| hitl_email_draft | STUB | stub | none |
| flag | STUB | stub | none |
| workflow_class_invoke | STUB | stub (would invoke CF Workflows class, e.g. annual_roll_workflow) | none |
| loop_over_recipients | STUB | stub | none |
| loop_over_aging | STUB | stub | none |
| dispatch_workflow | STUB | stub (would recursively invoke executeWorkflowContract) | none |
| <unknown> | STUB | stepStatus='failed', output.note='unrecognized_step_kind' | workflow_step_log row with status='failed' |
R560 invariant: stubs no longer count toward steps_executed. Pre-R560, an all-stub workflow returned ok with executed=N; post-R560 it returns ok with executed=0 and the step trace shows status='stub'. This is how Mike spots which contracts still need wiring.
| stage | D1 writes | KV touches | events fired | error class on failure |
|---|---|---|---|---|
| 1. loadContract | none (1 SELECT) | none | none | workflow_type_not_found → status='failed' |
| 2. loadContext | none (N parallel SELECTs) | none | none | context.<name>:<msg> pushed to result.errors; non-fatal |
| 3. checkPreconditions | none | none | none | block-severity fail or unevaluable → status='aborted' |
| 4. stageHitlProposal | INSERT proposed_actions | none | none directly (HITL endpoint fires hitl.approved later) | action_id null → status='pending_hitl' with hitl_proposed_action_id=undefined |
| 5. executeFanOut (per step) | INSERT workflow_step_log (entry + exit); + per-kind: INSERT proposed_actions (stage_proposed_action only) or D1 writes (none yet for stubs) | DELETE per kv_invalidate.keys[] | none | per-step error string in trace[]; on_failure='abort' breaks loop |
| 6. scheduleVerifyChecks | INSERT workflow_verify_results (status='pending', one per verify check) | none | none | silently caught (run still completes); ids stripped from verify_check_ids |
| 7. executePostActions | INSERT workflow_run_log; if reflexion_enabled: INSERT reflexion_log | none | none today (see inconsistencies note) | silently caught; run already complete |
| bug class (pre-R560) | fix (R560) | visible to operator how |
|---|---|---|
loadContext silently swallowed query errors (.catch(()=>null)) | errors push to contextErrors[] → result.errors | operator sees context.<name>: <msg> in result body |
| checkPreconditions evaluated unevaluable rules as 'pass' (advisory) | block-severity unevaluable now blocks; warn-severity downgrades with marker | aborted runs surface the offending rule with "(unevaluable, downgraded to warning)" suffix |
stage_proposed_action INSERT used .catch(()=>null) — HITL bypass if INSERT failed | try/catch → explicit throw; outer catch marks step 'failed' | Mike sees missing approval card; step_log row with status='failed' + error message |
stub kinds (ns_push, d1_write, ...) returned ok and incremented executed | marked status='stub', NOT counted as executed | completed runs with executed=0 mean the contract still needs wiring |
| (event-driven) concurrent drainer firings could double-invoke runner | KV concurrency lock per drainer (TTL 300s); 2nd firing returns skipped_due_to_lock | see substrate-event-ledger |
Aggregate of every persisted row + side-effect a successful run can produce. Always rows happen on every non-aborted run; conditional rows happen per contract config.
| storage | row(s) | when |
|---|---|---|
workflow_run_log | 1 row (UPSERT on run_id) — status, duration_ms, errors_count, output_json | always (stage 7) |
workflow_step_log | 2 rows per fan-out target (entry status='running' + exit status='ok'|'stub'|'failed'|'skipped'|'dry_run') | always (stage 5, per step) |
workflow_verify_results | N rows, one per verify_checks_json entry; status='pending' | per contract (stage 6) |
proposed_actions | 1 row from HITL gate (risk ≥ 3); + 1 row per stage_proposed_action fan-out target | conditional (stages 4 + 5) |
reflexion_log | 1 row, entity_type='workflow_run', tags='workflow:<type>,status:<status>' | if reflexion_enabled=1 (stage 7) |
events | 0 rows from runner today; workflow.completed is documented intent but NOT wired (see inconsistencies) | n/a today |
KV env.CACHE | per kv_invalidate step: N keys deleted | conditional (stage 5) |
94 total fan-out targets across 22 contracts. 20 REAL, 74 STUB. 18 are HITL-gated (risk ≥ 3). Per-workflow detail: see index → Workflow contracts (v2).
| category | workflows | doc |
|---|---|---|
| AR & finance | ar_aging_action_plan, service_hold_trigger | ar-aging-action-plan, service-hold-trigger |
| Anomaly & guardrail | anomaly_remediation | anomaly-remediation |
| Bid & spec | bid_amendment_arrives, bid_award_notification, bid_price_update, spec_deviation_flagged, usda_drawdown_commit | bid-amendment-arrives, bid-award-notification, bid-price-update, spec-deviation-flagged, usda-drawdown-commit |
| Comms & triage | batch_notify, inbound_email_triage | batch-notify, inbound-email-triage |
| Customer ops | customer_invoice_dispute, customer_quote, draft_quote, new_customer | customer-invoice-dispute, customer-quote, draft-quote, new-customer |
| Item & vendor | new_assembly_item, new_ingredient_vendor_onboard, vendor_cost_update | new-assembly-item, new-ingredient-vendor-onboard, vendor-cost-update |
| Other | assembly_build_review, item_record_review | assembly-build-review, item-record-review |
| Scheduled cycle | annual_price_roll, monthly_margin_review, quarterly_bid_eligibility_refresh | annual-price-roll, monthly-margin-review, quarterly-bid-eligibility-refresh |
src/index.ts:14304 (HTTP), src/index.ts:33255 (cron 45 5 * * *)drainEvents() calls executeWorkflowContract