How a run executes

1. Entry

Integration builds a RunContext and calls:

from drtoller.framework.processing.dispatch import run_from_manifest
outcome = run_from_manifest(manifest_path, ctx, user_params=…)

Or a thin step hook (run_partition_loop, run_mapping, …).

2. Load + merge params

1. load_manifest_path — YAML + include fragments.
2. merged_params_from_manifest — params_defaults ⊕ UI/Airflow overlay.
3. Assert manifest.step_id == ctx.step_id.

3. Metrics setup (when needed)

With runtime.metrics_enabled and a prometheus plan / mode:

• compile_step_manifest → StoragePlan;
• setup_metrics → RunMetrics + resource sampler.

4. Dispatch

v3:

1. If pattern is in _PATTERN_RUNNERS → that runner (mapping, reduce_merge).
2. Else execution.entry is required → import_symbol(entry)(ctx, manifest_path=…).

Legacy: _MODE_RUNNERS[execution.mode] or manifest.runner.

Extend without touching the core: register_pattern_runner / register_mode_runner.

5. Mode internals (partition_loop)

compile → resolve ParallelPlan
  → build feed shard metrics / progress totals
  → inprocess 3-stage (reader | processor | writer)
     or parallel workers (processes / ray)
  → registry checkpoint (cadence from DRTML)
  → dataset metrics poller → SaturationSlice → Prometheus
  → finalize manifests / close session

The processor receives already materialized rows and returns ProcessResult.
Write: session.write → write_backend_for(spec.backend).

6. Finalize

finally: finalize_metrics(…, outcome=finished|failed).

Reduce / mapping / embed_train

• mapping — compile merge plan, mapping.build, output manifests JSON.
• reduce_merge — sum/merge per DRTML; metrics ticks via callback (storage does not import processing).
• embed_train — stateful W fold; shared embed_vocab/train_loop.run_train_over_groups; backends are transport only.

Responsibility matrix

Layer	Does	Does not
Domain	transform, in-RAM registry, choose K	I/O, metrics emit
Processing	lifecycle, queues, hooks	parquet math, raw SQL
Storage	read/write/feed	Prometheus, Airflow
Telemetry	emit / codegen	runtime psycopg
DB	transport	domain column defaults