Execution and orchestration reference

Execution selection

drtml_version: 3
step_id: enrich_documents
execution:
  entry: acme_pipeline.steps.enrich.hooks:run

For builtin structural patterns:

pattern: mapping
mapping:
  entry: drtoller.framework.processing.mapping_mode.run:run_mapping
  inputs: {source: local_ids}
  outputs: {crosswalk: canonical_mapping}

pattern: reduce_merge
reduce_merge:
  entry: drtoller.framework.processing.reduce_merge.run:run_reduce_merge
  inputs: {partials: partial_counts}
  outputs: {final: total_counts}
  group_by: [language, token_id]
  sum_columns: [count]
  upstream_runs_param: reduce.upstream_runs

Dispatch uses pattern registries first, then execution.entry; legacy manifests use mode registry then runner.

Processing block

processing:
  init: acme_pipeline.steps.enrich.impl:init_processor_ctx
  processor: acme_pipeline.steps.enrich.impl:process_batch
  batch_processor: acme_pipeline.steps.enrich.impl:process_batch
  outputs:
    rows: enriched_documents
  registry_flush:
    vocabulary: token_registry
  mapping_read_bindings:
    token_crosswalk: tokens
  feed_attach:
    metadata: document_metadata

• init: optional in-memory processor context factory.
• processor: required processor import.
• batch_processor: explicit batch callable when separate.
• outputs: logical processor output name → dataset id.
• registry_flush: registry key → output dataset id.
• mapping_read_bindings: processor context key → mapping_read alias.
• feed_attach: logical ProcessorBatch.attach key → attached dataset id.

Registry plugins

registry:
  - plugin: acme_pipeline.registry.tokens:TokenRegistry
    factory: create
    dataset: token_registry
    checkpoint: token_checkpoint

Registry implementations are in-memory domain state. Framework owns loading rows for resume, drain/checkpoint cadence, and persistence through declared datasets.

Orchestrator block

orchestrator:
  dag_id: enrich_documents_v1
  step_id: enrich_documents
  connections: [minio_s3, postgres_app]
  artifacts_root: env:DRTOLLER_ARTIFACTS_ROOT

  env_fallbacks:
    runtime.parallelism.ray_address: RAY_ADDRESS
    domain.model_name: DEFAULT_MODEL_NAME

  env_fallback_unless:
    runtime.parallelism.ray_address: domain.force_local

  params_hook: acme_pipeline.steps.enrich.params:complete_params

  xcom:
    - key: output_manifests
      from: outputs.output_manifests_json
    - key: row_count
      from: stats.rows_written

• connections: integration connection ids. Airflow currently applies MinIO credentials for minio_s3; new connection types need a connection hook.
• artifacts_root: absolute path or env:VAR resolved by integration.
• env_fallbacks: param → environment variable, applied when the user did not supply the param.
• env_fallback_unless: skip a fallback when another param is truthy.
• params_hook: optional module:fn(manifest, params, user_params) -> params; use sparingly for integration completion, not domain policy.
• xcom: extra Airflow XCom values resolved from run result dot paths. A standard <step_id>_outputs XCom is always pushed.

Manifest discovery outside a shared source checkout

Airflow and UI should discover step packages from deployment configuration, not hardcoded repository names:

• DRTOLLER_STEPS_ROOTS_FILE: newline-separated roots.
• DRTOLLER_STEPS_ROOTS: colon-separated roots.
• DRTOLLER_STEPS_ROOT: one root.

Repository-specific paths may exist as development fallbacks, but they are not part of the public step contract.