create-ingestion

name	create-ingestion
description	Generates the Bronze config-driven ingestion layer from the approved LLD, DMS, STM, DQS, and a given story. Modes: - Story mode: read a STORY-NN-NNN, extract backtick-quoted deliverables from its Acceptance Criteria, generate only those that fall under this skill's domain of ownership. ROUTE-OUT for others. - Full mode: process every un-Done story classified as `ingestion`, topo-sorted by Depends On. Project-agnostic: the table list, critical-table designation, per-table DQ rule filenames, and runner module names are ALL read from the current run's LLD / DMS / STM / DQS / stories at runtime. No project identifiers or story IDs are hardcoded. Use when the user asks to: - Generate or update Bronze ingestion code for a story - Build the ingestion runner / factory / SparkSubmit wrapper - Generate per-table ingestion configs
argument-hint	[STORY-NN-NNN | 'full']
allowed-tools	Read, Write, Edit, Grep, Glob, Bash, AskUserQuestion, Skill
context	fork

Create Bronze Ingestion Framework

You are a senior Data Engineer. Your job is to translate LLD §2.3 and §5.1 into production-ready ingestion code: a generic runner, a TaskGroup factory, a SparkSubmitOperator wrapper, and one YAML config per Bronze table.

The LLD is the single source of truth. Never invent paths or tables — every module path, config file, and table name must already be named in the LLD. the output must be config-driven, not hard-coded.

Workspace Discovery

Before any file operation, run the discovery helper and substitute the returned tokens into every path this skill reads, writes, or edits:

python3 ${CLAUDE_PLUGIN_ROOT}/skills/validate-stories/scripts/status_rollup.py --mode discover

The JSON output supplies {workspace_root}, {project_root}, {project_name}, {stories_dir}, and {learnings_queue}. The plugin is project-agnostic — never hardcode project or chapter names in edits.

Coding Patterns & Libraries Handbook

Before any ingestion code is emitted, load the latest coding-patterns handbook:

PATTERNS_DIR=$(ls -d "{workspace_root}/inputs/code/v"* 2>/dev/null | sort -V | tail -1)
if [ -z "$PATTERNS_DIR" ] || [ ! -d "$PATTERNS_DIR" ]; then
  echo "CRITICAL: inputs/code/v*/ not found. Run /developer-plugin:refresh-libraries to initialize the library cache."
  exit 1
fi
LIBRARIES_FILE="$PATTERNS_DIR/LIBRARIES.md"

Required pattern docs for this skill:

• $PATTERNS_DIR/bronze-ingestion-pattern.md — registry → factory → runner shape
• $PATTERNS_DIR/spark-expectations-pattern.md — SE YAML rule shape + runner wrapper
• $PATTERNS_DIR/naming-conventions.md — audit columns, table names
• $PATTERNS_DIR/logging-error-handling.md — module logger + fail-fast
• $PATTERNS_DIR/LIBRARIES.md — pinned PySpark / Delta / SE versions

Library freshness check

LAST_VERIFIED=$(grep '^last_verified:' "$LIBRARIES_FILE" | awk '{print $2}')
TODAY=$(date -u +%Y-%m-%d)
AGE_DAYS=$(python3 -c "from datetime import date; print((date.fromisoformat('$TODAY') - date.fromisoformat('$LAST_VERIFIED')).days")

If AGE_DAYS > 30, pause and call AskUserQuestion with options Refresh now / Proceed with cached versions / Cancel. On Refresh, invoke /developer-plugin:refresh-libraries then resume.

References trailer (in output)

Emit a ### References section citing consumed pattern docs + LIBRARIES.md vintage. Add a stale-cache warning if the user opted to proceed with cached versions.

Domain of Ownership

This skill owns the project's Bronze ingestion layer — config-driven code that reads raw sources and writes versioned Bronze Delta tables. Specifically:

Owned path pattern	What lives here
src/<project_name>/bronze/**	Ingestion runner, factory, SparkSubmit wrapper, reconciliation task, DLQ writer — whatever LLD §2.3 + §5.1 name
airflow/configs/**	One YAML per Bronze table from LLD §5.1
dq_rules/**	Synced from latest {workspace_root}/outputs/dqs/v*/se-rules/*.yaml
contracts/dq/**	Per-table DQ thresholds from DQS §5
ddl/migrations/**	Bronze DDL as plain dated .sql from DMS §2 (no Liquibase, L-015)
tests/bronze/**	Bronze unit + integration tests

It does not own: src/<project_name>/utils/** (create-scaffold), src/<project_name>/{silver,gold}/** (future layer skills), airflow/dags/** (create-dag), contracts/*.yml except contracts/dq/** (create-scaffold; StructType contracts), .github/workflows/** + _infra/{ci,cd}/** (create-pipeline).

Deliverable Resolution Protocol

No Story → Deliverable Map is hardcoded. The current run's story file is the single source of truth for which paths to emit.

1. Story mode (STORY-NN-NNN): read the story file. Parse Acceptance Criteria. For each backtick-quoted path or identifier:

   • Path matches an owned pattern above → generate it.
   • Path matches another skill's domain → print ROUTE-TO: create-<kind> <path> and exit cleanly.
   • Identifier (no path) → resolve via LLD §2.3 interface contracts; if the LLD places it under this skill's domain, generate; else ROUTE-OUT.

2. Full / auto mode: enumerate every un-Done story whose classifier kind is ingestion (status_rollup.py --mode classify), topo-sorted by Depends On, and run each through Story mode.

Source-of-Truth Reads at Runtime

Project-specific facts are read each run — never hardcoded.

Fact needed	Where to read
Bronze table list	LLD §5.1 (Source → Bronze task table)
Source schema/format per table	STM Source-to-Bronze tab
Per-table empty_input_behavior	LLD §5.1 column — the tables flagged fail ARE the critical tables
Per-table SE rule file	ls {workspace_root}/outputs/dqs/v*/se-rules/*.yaml — discover filenames; do NOT assume any prefix like se-rules-<domain>-<table>.yaml
Per-table DQ thresholds	DQS §5
Runner/factory/wrapper module names	LLD §2.3 Module Interface Contracts
Compute knobs (shuffle, cores, mem)	config-template.yaml + LLD §6.1/§6.3
DAG name (for integration tests)	LLD §4.1

Workflow

Phase 0: Detect Input Mode and Upstream Gate

FIRST — resolve the effective argument. Do this before any other Phase 0 step.

The conversational argument is NOT the only source of truth — a parent orchestrator (e.g. implement-stories) may dispatch this skill via the Skill tool and the forwarded argument may not reach this fork. Check these sources in order; stop at the first non-empty hit:

1. $SKILL_ARG environment variable.
2. {workspace_root}/.skill-arg file — read its contents, then delete the file so it is consumed at most once.
3. The conversational argument supplied to the skill.
4. Auto-mode default — if $CLAUDE_AUTO_MODE=1 OR {workspace_root}/.auto-mode exists → use the skill's full-mode default (full LLD generation, no story ID).
5. Only if ALL four above are empty, ask the user via AskUserQuestion.

You MUST NOT ask the user until sources 1–4 have been checked. Auto-mode detection is a hard override — if the marker exists, proceed with the full-mode default and do not ask.

Mechanical resolver (copy-paste; $USER_ARG is the conversational arg):

resolve_skill_arg() {
  if [ -n "$SKILL_ARG" ]; then echo "$SKILL_ARG"; return; fi
  if [ -f "{workspace_root}/.skill-arg" ]; then
    cat "{workspace_root}/.skill-arg"; rm -f "{workspace_root}/.skill-arg"; return
  fi
  # caller supplies conversational arg as $1
  if [ -n "$1" ]; then echo "$1"; return; fi
  if [ "$CLAUDE_AUTO_MODE" = "1" ] || [ -f "{workspace_root}/.auto-mode" ]; then
    echo "__AUTO__"; return
  fi
  echo ""
}
RESOLVED_ARG=$(resolve_skill_arg "$USER_ARG")

If $RESOLVED_ARG is __AUTO__ or matches the full-mode pattern (empty or an .md path), proceed without asking — treat it as full mode.

Step 1 — Detect mode from the resolved argument:

• If the argument matches the pattern STORY-NN-NNN (e.g. STORY-02-002) or the user says something like "implement story 2 of epic 2" → Story mode. Normalize to STORY-{EPIC:02d}-{NUM:03d} format (e.g. "story 2 of epic 2" → STORY-02-002).
• If the argument is a file path (ends in .md) → Full mode. Skip to the LLD gate.
• If $RESOLVED_ARG is __AUTO__ (auto-mode default) → Full mode. Skip to the LLD gate.

Step 2 — Story mode: resolve and read the story file:

STORY_ID="STORY-02-002"   # substitute actual ID
EPIC_NUM=$(echo "$STORY_ID" | cut -d- -f2)
STORY_FILE=$(ls {stories_dir}/v*/EPIC-${EPIC_NUM}-*/STORY-${STORY_ID}*.md \
             {stories_dir}/v*/${STORY_ID}*.md 2>/dev/null | head -1)
echo "$STORY_FILE"

Read the story file and extract:

• Title and Story Points
• Sprint number
• Acceptance Criteria (the numbered AC list)
• Dependencies (Depends On: lines)

If the story file is not found, stop and tell the user which path was searched.

Verify the story classifies to ingestion (run status_rollup.py --mode classify --story STORY-NN-NNN). If the returned skill_kind is anything other than ingestion, emit ROUTE-TO: create-<kind> and exit cleanly so the orchestrator can dispatch the correct skill. Do NOT raise an error.

If the story classifies to ingestion but is NOT listed in the Story → Deliverable Map above, read its Acceptance Criteria at runtime and generate every backtick-quoted deliverable path. The classifier is the gate, not the map.

Step 3 — Story mode: dependency check:

For each story listed under Depends On:, open its story file and verify each backtick-quoted deliverable path from its ACs exists on disk. If any dependency's deliverables are missing, stop and list what is missing:

Dependency check failed:
  STORY-NN-NNN depends on STORY-NN-MMM
  Missing (declared in STORY-NN-MMM AC): {project_root}/<path>
Complete STORY-NN-MMM first, then re-run.

Step 4 — Set GENERATION_SCOPE:

In story mode, set GENERATION_SCOPE to the subset of candidate deliverables from the current story's AC that match this skill's owned path patterns. Phase 3/4 generation blocks check this scope and skip anything not in it; anything owned by another skill emits ROUTE-TO.

In full mode, GENERATION_SCOPE = all.

Step 5 — LLD gate (both modes):

Read the latest LLD and verify Status: Approved (or Updated - Pending Review if the user explicitly opts to proceed with a draft).

# Prefer pinned versions from the dev-lock (written by implement-stories).
# Falls through to the legacy `ls … | tail -1` pattern when no lockfile.
eval "$(python3 ${CLAUDE_PLUGIN_ROOT}/scripts/resolve_versions.py --export)"
LATEST_LLD_DIR="${LATEST_LLD_DIR:-$(ls -d {workspace_root}/outputs/lld/v* | sort -V | tail -1)}"
ls -t "$LATEST_LLD_DIR"/LLD-*.md | grep -v '\.bak$' | head -1

If not approved, stop and inform the user.

Phase 0.5: Path Scoping (optional, set by the orchestrator)

If {workspace_root}/.skill-paths exists, read it — each non-empty line is a path in scope for THIS invocation. Process ONLY those paths; ignore other paths the story AC names (they belong to other skills and the orchestrator will dispatch them separately). Delete .skill-paths after consuming it. When the file is absent (direct human invocation), fall through to the standard phases that scan every candidate deliverable against the Domain of Ownership.

Phase 1: Read Inputs

• LLD markdown: read §2.3 (Module Interface Contracts), §5.1 (Bronze task table), §3 (storage layout), §7 (configuration schema). The LLD §5.1 task table lists every table to generate a config for.
• Config template: {workspace_root}/outputs/lld/v{N}/config/config-template.yaml — source of truth for default storage paths, ingestion knobs (ingestion_config_dir, ingestion_dq_rules_dir, ingestion_default_empty_input_behavior, ingestion_spark_submit_class), and compute defaults.
• STM workbook (reference only): {workspace_root}/outputs/stm/v{N}/STM-*.xlsx → Source-to-Bronze tab. Column lists inform the per-table YAML schema block.
• Existing project tree: {project_root}/ — confirm target directories (src/{project_name}/bronze/, airflow/configs/, contracts/, dq_rules/) exist before writing. Never create new top-level folders.
• Pattern handbook: the canonical bronze-ingestion coding style lives in $PATTERNS_DIR/bronze-ingestion-pattern.md. Follow that; do NOT copy from any other example that might hard-code a TABLE_REGISTRY — the generated runner must be config-driven from airflow/configs/*.yml.
• DQS SE rules (source of truth for dq_rules/): discover files under {workspace_root}/outputs/dqs/v{N}/se-rules/*.yaml. Match each file to a Bronze table by reading the file's product_id (or by filename stem, as a fallback). Do NOT assume any filename prefix like se-rules-<domain>-<table>.yaml — that naming is project-specific. These are Spark Expectations-formatted rule sets (product_id, dq_env, rules[]) produced by the upstream DQ Engineer plugin. Never hand-write stubs when a matching SE file exists.

Phase 2: Clarify

Story mode — show a scope summary before generating anything:

Story:       STORY-02-002 — Generic Ingestion Runner (5 pts, Sprint 3)
Deliverable: {project_root}/src/{project_name}/bronze/ingestion_runner.py
Depends on:  ✓ STORY-02-001 (13 configs found)
LLD status:  Approved

Acceptance Criteria:
  1. Runner reads per-table YAML config
  2. Enforces StructType schema (no inference)
  3. Adds metadata columns: ds, _ingested_at, _source_batch_id
  4. Writes Delta partitioned by ds via dynamic partition overwrite (idempotent re-runs)
  5. Respects empty_input_behavior (fail raises, write_empty proceeds)

Then use AskUserQuestion to ask: "Generate this story's deliverable now? If a file already exists, overwrite or skip?"

Full mode — use AskUserQuestion to confirm (only where the LLD is silent):

• Which Bronze tables to include on this run (all 13 from §5.1, or a subset).
• Whether existing files should be overwritten or skipped.
• Source read format when STM leaves it ambiguous (CSV vs JDBC vs Delta).

Skip any question if the LLD/config-template gives an unambiguous answer.

Phase 3: Generate Code

Scope gate: In story mode, generate only the module(s) for the active story per the Story → Deliverable Map. Skip all others — do not create empty stubs.

Write Python modules to {project_root}/src/{project_name}/bronze/ (and utils/ for se_runner). Only write a module if it is in GENERATION_SCOPE:

1. ingestion_runner.py — reads a per-table YAML, enforces StructType (no schema inference), adds metadata columns (ds, _ingested_at, _source_batch_id), calls se_runner inline for row_dq + agg_dq, writes Delta partitioned by ds via dynamic partition overwrite (df.write.mode("overwrite").insertInto("unity.bronze.<table>") with spark.sql.sources.partitionOverwriteMode=dynamic; NEVER .option("replaceWhere", ...) — see the CRITICAL write block below), respects the per-table empty_input_behavior. _source_batch_id is deterministic ({table}:{ds}) so reruns are idempotent. Accepts a Spark session + config path via CLI (--config-path, --ds, --env). Delta wiring: _build_spark() must call delta.configure_spark_with_delta_pip(builder) and set spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog so local python -m runs work without PYSPARK_SUBMIT_ARGS. SparkSubmit production runs will pick up the same JAR via --packages. Metadata column names match the DQS SE rules in {workspace_root}/outputs/dqs/v{N}/se-rules/ — underscore prefix is required. TableConfig dataclass must include a quarantine_path_template field (default warehouse/{env}/quarantine/bronze/{table}/ per LLD §7) with a resolved_quarantine_path(env) method. Load it from the per-table YAML quarantine_path key; fall back to the default if absent. run_inline_dq signature: (df, cfg, env, dq_rules_dir). Define _DQ_ENV_MAP = {"DEV": "DEV", "STAGING": "QA", "PROD": "PROD"} (LLD §2.3) and map env before calling run_dq. Derive dq_rules_dir in this order (matches the AIRFLOW_CONFIGS_DIR pattern):
   1. Explicit kwarg passed by the caller.
   2. os.environ.get("DQ_RULES_DIR") — the canonical container path injected by the cookiecutter docker-compose.yml.
   3. Walk up from the config path: config_path.parent.parent.parent / "dq_rules" — only as a fallback for non-container runs (local pytest, etc.). Never hardcode /opt/dq_rules (relative or absolute) — spokane shipped a duplicate dq_rules bind mount because some generated code referenced that literal. The env var is the single resolution mechanism. se_action_if_failed from the per-table YAML is the fail-closed default for rules that omit their own action_if_failed; per-rule declarations take precedence (LLD §5.4). CRITICAL — Bronze BUSINESS tables are EXTERNAL + insertInto (LLD §2.3 / Decision 12).

The Bronze runner MUST land BUSINESS tables into PRE-CREATED Unity Catalog external Delta tables using insertInto("unity.bronze.<table>"). The tables are created at DEPLOY time by Liquibase (CREATE TABLE unity.bronze.<table> (...) USING DELTA LOCATION '<warehouse path>') — the runner NEVER creates business tables and NEVER calls saveAsTable on them. This is a deliberate design choice (deploy-time DDL owns the schema + location), NOT a connector limitation: on UC 0.5.0 saveAsTable IS supported for managed tables (and the re-enabled spark-expectations error/stats tables use it — see the se_runner spec below). Business Bronze tables simply stay external so their location and schema are governed by Liquibase, not inferred at write time. Path-based Delta writes (df.write.format("delta").save("/tmp/...")) for business tables leave UC empty until a manual docker cp + external-table registration — that gap remains prohibited for business data.

The skill's Phase 3 generator MUST emit the Spark session wiring shown in inputs/code/v1/scripts/ingestion_runner.py.snippet. UC is ALWAYS a named side catalog alongside DeltaCatalog — NEVER bind spark_catalog to UCSingleCatalog (binding breaks):

• spark.sql.extensions = io.delta.sql.DeltaSparkSessionExtension
• spark.sql.catalog.spark_catalog = org.apache.spark.sql.delta.catalog.DeltaCatalog
• spark.sql.catalog.unity = io.unitycatalog.spark.UCSingleCatalog
• spark.sql.catalog.unity.uri = os.environ["UC_URI"] (e.g. http://unity-catalog:8080)
• spark.sql.catalog.unity.token = ""
• spark.sql.catalog.unity.warehouse = unity
• spark.sql.defaultCatalog = unity

Idempotent write — DYNAMIC PARTITION OVERWRITE is THE single method (LLD v1.14 §2.3 / §13 Decision 15). The Bronze write MUST be:

df.write.mode("overwrite").insertInto(f"unity.bronze.{table}")

with the Spark session configured for dynamic partition overwrite: spark.sql.sources.partitionOverwriteMode = dynamic (the scaffold's build_spark_session owns this config — reference it, do NOT set it ad-hoc in the runner). This replaces ONLY the ds partition(s) present in the DataFrame, so re-runs are idempotent and other ds partitions are preserved.

NEVER use .option("replaceWhere", ...) with insertInto. Empirically confirmed via Delta DESCRIBE HISTORY: insertInto SILENTLY IGNORES .option("replaceWhere", ...) (replaceWhere only takes effect with .save() / .saveAsTable()), so re-runs APPEND and DOUBLE the data. Additionally, replaceWhere cannot coexist with partitionOverwriteMode=dynamic — that combination throws DELTA_REPLACE_WHERE_WITH_DYNAMIC_PARTITION_OVERWRITE. Do NOT use replaceWhere, overwritePartitions(), or any other variant: dynamic partition overwrite with insertInto is the one mandated idempotent write.

The validator (validate-dag UC-WIRING-001) rejects any generated src/**/bronze/**.py file that calls .save("/tmp/, .save("file://, or .save(<warehouse_path_var>) instead of insertInto, and rejects any saveAsTable against unity.bronze.<table>. The tests/test_uc_wiring.py regression scan blocks the same patterns from being introduced into snippets.

2. ingestion_factory.py — build_bronze_taskgroup(dag, configs_dir=None) scans <configs_dir>/*.yml at DAG parse time, returns an Airflow TaskGroup named bronze_ingestion with one SparkSubmitOperator per file. CRITICAL — env-driven path resolution. The factory MUST resolve configs_dir in this order: explicit arg → os.environ.get("AIRFLOW_CONFIGS_DIR") → /opt/airflow/configs. Never hardcode airflow/configs (relative) or any other relative path — when Airflow runs from /opt/airflow/ in the cookiecutter container, relative paths don't resolve. The validate-dag DAG-PATHS-002 rule rejects DAG files containing the literal string configs_dir="airflow/configs". The reconciliation runner under airflow/jobs/run_<layer>_recon.py follows the same pattern via --configs-dir argparse with the same default chain (see inputs/code/v1/scripts/reconciliation.py.snippet).

3. spark_submit_wrapper.py — thin helper that builds the SparkSubmitOperator with memory/cores/executors pulled from the pipeline config (§7) and passes --config-path to the runner module named by ingestion_spark_submit_class in config-template.yaml. Catalog: spark.sql.catalog.spark_catalog must be org.apache.spark.sql.delta.catalog.DeltaCatalog — same as the runner. The runner ALSO wires unity as a named side catalog (spark.sql.catalog.unity = io.unitycatalog.spark.UCSingleCatalog, spark.sql.defaultCatalog = unity) so Bronze can insertInto the pre-created unity.bronze.<table> external tables. spark_catalog stays bound to DeltaCatalog — NEVER bind it to UCSingleCatalog.

4. src/{project_name}/utils/se_runner.py — implement run_dq(df, *, table, env, dq_rules_dir, action_if_failed, quarantine_path). This module is the implementation of the LLD §2.3 se_runner.py interface contract.

   Version requirement (CRITICAL): these imports require spark-expectations >= 2.10.0. The YAML rule loader and WrappedDataFrameWriter were added in v2.10.0 (https://github.com/Nike-Inc/spark-expectations/releases/tag/v2.10.0). If the LIBRARIES.md catalogue ever pins SE below 2.10 these imports raise ModuleNotFoundError: No module named 'spark_expectations.rules' at runtime — the developer plugin's validate-dag SE-IMPORTS-001 rule and tests/test_se_runner_imports.py regression test gate against that downgrade. Pin the wiring details exactly:

   from spark_expectations.core.expectations import SparkExpectations, WrappedDataFrameWriter
   from spark_expectations.rules.plugins.yaml_loader import SparkExpectationsYamlRuleLoaderImpl
   
   # SE error/stats tables are MANAGED Unity Catalog tables addressed by a
   # FULLY-QUALIFIED name `catalog.schema.table` (UC 0.5.0). SE derives the
   # error-table name as `f"{target_table}_error"`; with an FQN target (e.g.
   # `unity.bronze.synthea_allergies`) that is the well-formed 3-part name
   # `unity.bronze.synthea_allergies_error`. The stats table is passed as an
   # FQN too (`unity.<schema>.<table>_stats`). SE's saveAsTable CREATES both
   # managed tables on first run (they live in the schema's storage_root) — do
   # NOT pre-create them in Liquibase; they are SE-owned.
   #
   # WHY FULLY-QUALIFIED NAMES ARE MANDATORY (verified root cause — NOT RTAS):
   # On UC 0.4.0 the error table was written by a BARE name
   # (`synthea_allergies_error`). UCSingleCatalog qualifies a bare name by
   # reading the session current-database; under spark-submit that is empty,
   # so `fullTableNameForApi` (UCSingleCatalog.scala:346) split an empty
   # namespace into a length-0 array and crashed on `namespace[0]` with
   # `ArrayIndexOutOfBoundsException: Index 0 out of bounds for length 0`.
   # That is an empty-namespace defect, NOT an RTAS/CTAS refusal — a deliberate
   # refusal throws a NAMED `UnsupportedOperationException`. Setting the session
   # current-database removed the AIOOBE on 0.4.0 (it became `SCHEMA_NOT_FOUND`),
   # proving the empty namespace was the cause. UC 0.5.0 fixes the namespace
   # handling AND supports managed creates, so an FQN target makes SE's
   # `<target>_error`/`_stats` saveAsTable succeed as managed tables.
   #
   # Requires: the bronze/silver/gold schema to have a managed location
   # (storage_root) configured by scripts/uc_init.py, plus delta-spark 4.1+ /
   # UC server 0.5.0 coordinated commits (see LIBRARIES.md).
   
   # Disable Kafka streaming stats and all notifications — required to avoid
   # Databricks secrets errors in local (non-Databricks) environments.
   user_conf = {
       "se.enable.error.table": True,   # managed UC error table by FQN (UC 0.5.0)
       "se.enable.stats.table": True,
       "se.streaming.enable": False,
       "spark.expectations.notifications.alert.flag.disable": True,
       "spark.expectations.notifications.email.enabled": False,
       "spark.expectations.notifications.slack.enabled": False,
       "spark.expectations.notifications.teams.enabled": False,
       "spark.expectations.notifications.pagerduty.enabled": False,
       "spark.expectations.notifications.zoom.enabled": False,
   }
   

   Load rules via SparkExpectationsYamlRuleLoaderImpl.load_rules(path, format="yaml", options={"dq_env": dq_env}). The error and stats writers are MANAGED — a WrappedDataFrameWriter().mode("append").format("delta") with no .option("path", ...) — passed as target_and_error_table_writer (at construction AND in with_expectations) and stats_table_writer. Pass stats_table=f"{se_target_table}_stats" (a fully-qualified managed-table name, e.g. unity.bronze.synthea_allergies_stats) — NOT a delta.\`path identifier. The managed SE **error table is now the rejected-row audit trail**; the legacyquarantine_path` becomes an optional secondary sink, no longer the sole record (it was a workaround for the disabled error table).

   CRITICAL — SE rule matching, view resolution, and per-table isolation (see IL/L "SE actually runs"). Without ALL of these, SE silently runs ZERO rules and the inline DQ gate is a no-op:

   1. product_id + target_table MUST match the loaded rule rows, or SE's reader.py filter (product_id == ctx.product_id AND table_name == target_table) selects ZERO rules → _row_dq is False → no row_dq / agg_dq / query_dq runs, no drop, no fail — DQ is a silent no-op.
      • Read the rules YAML and pass product_id=<rules file product_id:> (e.g. patient-360) — NEVER the bare table name.
      • Pass target_table=<dq_env[<ENV>].table_name> (the generator now emits this as the FULLY-QUALIFIED catalog.schema.table, e.g. unity.bronze.synthea_allergies; resolve it from the file rather than hardcoding). The SAME string is the SE rule filter key (so the rule rows' table_name must match it exactly — both FQN), the base for the managed error/stats table FQNs (<target>_error, <target>_stats), and the post-row_dq view name (<target_table.split('.')[-1]>_view). FQN is what makes SE's derived <target>_error a well-formed 3-part name — see the root-cause note above.
   2. Register bare-name TEMP VIEWS before with_expectations so referential query_dq SQL (which references sibling tables by 1-part name) resolves under spark.sql.defaultCatalog=unity (where a 1-part name would NOT resolve to unity.<schema>.<t>). For each unity.{bronze,silver,gold} schema that exists, CREATE OR REPLACE TEMP VIEW <t> AS SELECT * FROM unity.<schema>.<t> for every table; then register the in-flight df LAST under its own bare name so the self-table reference sees the rows being processed.
   3. SE stats AND error tables are PER-TABLE MANAGED UC tables — both carry the source schema (the error table holds failing source rows; the stats pre-create writes the source df), so each must be its own catalog table. Derive distinct FQNs per source table — unity.<schema>.<table>_error and unity.<schema>.<table>_stats — from the FQN target_table; a single shared name would collide across tables with DELTA_* schema errors. SE creates them as managed tables on first run (schema storage_root); they are NOT in Liquibase.
   4. Rule ids in the YAML use underscores (DQ_FLD_077), emitted by the generator — SE builds row_dq_<rule> column names and references them unquoted, so a hyphen (DQ-FLD-077) parses as subtraction and fails with UNRESOLVED_COLUMN. The runner does not need to transform ids; just do not re-introduce hyphens.
   5. run_dq MUST return ONLY the input DataFrame's columns — capture input_cols = df.columns BEFORE with_expectations and end with return validated.select(*input_cols). with_expectations APPENDS SE run-tracking columns (meta_dq_run_id, meta_dq_run_datetime) to the returned DataFrame. If the caller writes that DataFrame straight to the pre-created target table (e.g. Bronze write_bronze → insertInto(unity.bronze.<t>)), the two extra columns cause a Delta _LEGACY_ERROR_TEMP_DELTA_0007 schema mismatch (the target has neither column). Silver transforms dodge this only because they re-select( OUTPUT_COLUMNS) after run_dq; making run_dq itself column-stable fixes it for every caller (the run-tracking values live in the SE stats table, not the data table). When SE was a no-op this never surfaced because it appended nothing.

Phase 4: Generate Per-Table Configs

Scope gate: Skip this phase unless a deliverable under airflow/configs/**, dq_rules/**, contracts/dq/**, or ddl/migrations/** was extracted from the story's AC.

For every Bronze table enumerated in LLD §5.1 (count and names read at runtime — never hardcoded), write {project_root}/airflow/configs/<table>.yml with this shape (substitute <table> and replace other bracketed placeholders with values read from LLD / STM / config-template.yaml):

table: <table>                       # LLD §5.1 table name
source:
  schema: <pipeline_source_schema>   # from config-template §general
  table:  <source_table>             # STM Source-to-Bronze tab
  format: <csv|jdbc|delta>           # STM Source-to-Bronze tab
  path: <path-if-file-format>        # STM Source-to-Bronze tab
schema_ref: contracts/<table>.yml
output_path: warehouse/<env>/bronze/<bronze_table_name>/   # table ROOT — see note
metadata_columns:
  - ds
  - _ingested_at
  - _source_batch_id
empty_input_behavior: <fail|write_empty>  # from LLD §5.1 Empty Input column
dq_rules_table: <table>
se_action_if_failed: <fail|drop|ignore>   # from LLD §5.1 DQ Check column
quarantine_path: warehouse/<env>/quarantine/bronze/<table>/
timeout_minutes: <from LLD §4.2>
retries: <from LLD §4.2>
retry_delay_seconds: <from LLD §4.2>

Notes:

• <bronze_table_name> follows the DMS Bronze naming convention (DMS §3 or wherever naming is declared). Do NOT hardcode a source-system prefix (e.g. synthea_) — read the DMS.
• empty_input_behavior per table comes from LLD §5.1. Tables flagged fail are the "critical" tables for this project — there is no hardcoded list.
• Do NOT include ds={ds}/ in output_path — the runner writes Delta partitioned by ds, so the output directory is the table root.
• Do NOT rewrite contracts/<table>.yml here (owned by create-scaffold via DMS). If a contract is missing, flag it as a prerequisite.

For dq_rules/<table>.yml, sync from the DQS SE rules by discovery:

LATEST_DQS_DIR=$(ls -d {workspace_root}/outputs/dqs/v* | sort -V | tail -1)
# For each Bronze table from LLD §5.1, locate its SE rule file by reading
# product_id from each candidate YAML; fall back to filename-stem match.
for SE_FILE in "$LATEST_DQS_DIR"/se-rules/*.yaml; do
  PID=$(python3 -c "import yaml,sys; print(yaml.safe_load(open('$SE_FILE')).get('product_id',''))")
  # match PID (or filename stem) to a Bronze table name from LLD §5.1
  cp "$SE_FILE" "{project_root}/dq_rules/${TABLE}.yml"
  # FQN convention: dq_env.<ENV>.table_name MUST be the fully-qualified
  # `unity.bronze.<table>` (catalog.schema.table), never the bare name. It is
  # the SE rule-filter key (must equal the target_table se_runner passes to
  # with_expectations) AND the base for the managed <target>_error/_stats
  # table FQNs. A bare name reproduces the UC empty-namespace
  # `fullTableNameForApi` AIOOBE on the error-table write. If upstream emits a
  # bare table_name, qualify it in place:
  python3 - "$TABLE" "{project_root}/dq_rules/${TABLE}.yml" <<'PY'
import sys, yaml
table, path = sys.argv[1], sys.argv[2]
doc = yaml.safe_load(open(path))
for env, blk in (doc.get("dq_env") or {}).items():
    tn = (blk or {}).get("table_name", "")
    if tn and "." not in tn:
        blk["table_name"] = f"unity.bronze.{tn}"
yaml.safe_dump(doc, open(path, "w"), sort_keys=False)
PY
done

If a Bronze table has no matching SE rule file, stop and ask the DQ Engineer to produce one. Do NOT hand-write stubs — runtime loads expect the full SE schema (product_id, dq_env, rules[]).

Phase 4.5: Runtime Dependencies

Scope gate: Always run this check — pyproject.toml deps are required regardless of which story is being implemented.

The ingestion runner imports pyspark, writes Delta, and loads Spark Expectations rules; the factory uses SparkSubmitOperator. Ensure {project_root}/pyproject.toml declares each of these — add any that are missing:

Section	Required entry
[project].dependencies	pyspark[sql]>=4.0
[project].dependencies	delta-spark>=4.0
[project].dependencies	spark-expectations>=2.6.0
[project.optional-dependencies].dev	apache-airflow-providers-apache-spark>=4.0
[project.optional-dependencies].dev	pytest-mock>=3.12

If the LLD §2.1 does not list these (current LLD v1 omits them), note the drift in the change-set report so the Technical Lead can reconcile on the next LLD revision.

Phase 4.6: Generate Tests

Scope gate: In story mode, generate a test module only when a test path under tests/bronze/** was extracted from the story's AC (or when the story's AC explicitly calls for "unit tests covering "). Tests for modules emitted by this skill MUST accompany the module; do not ship untested ingestion code.

Write test modules under {project_root}/tests/bronze/ so the generated code is exercised in CI. Use pytest.importorskip("pyspark") so Spark- dependent tests skip cleanly when pyspark is not installed.

The exact set of test modules is driven by what this skill just generated, not by a hardcoded table. General guidance:

If you generated …	Write tests covering …
ingestion_runner.py	load_table_config (happy + each config-error branch), _parse_spark_type (every primitive + decimal + unsupported), load_struct_type, add_metadata_columns (assert _source_batch_id is deterministic + _ingested_at present), empty-input branches of ingest(), resolved_quarantine_path, _DQ_ENV_MAP (parametrised over runtime envs). Requires pyspark — import-skip otherwise.
airflow/configs/*.yml (per-table configs)	Parametrised sweep over {project_root}/airflow/configs/*.yml: required keys present, table: matches filename, referenced contracts/<t>.yml + dq_rules/<t>.yml exist, tables flagged fail in LLD §5.1 use fail, SE rules declare every dq_env value, the table count matches LLD §5.1 exactly. Pure YAML — no Spark.
ingestion_factory.py / DAG integration	Shells out to validate-ingestion/scripts/validate_ingestion.py against the project (expect PASS) and against a minimal scaffold missing runtime deps (expect CRITICAL on the missing libs).

Read the AC text of the active story verbatim — the story tells you which behaviour needs test coverage. Do not skip this phase.

Phase 5: Validate

Invoke /developer-plugin:validate-ingestion on the generated files. Fix any CRITICAL findings before reporting completion. Report WARNING/INFO findings to the user without blocking.

Story mode only — Acceptance Criteria verification:

After the validator passes, check each AC from the story file against the generated code and report a pass/fail table. Use Grep to verify structural presence. Examples for common ACs:

Acceptance Criterion	Check
"Runner reads per-table YAML config"	load_table_config function exists in ingestion_runner.py
"Enforces StructType schema (no inference)"	load_struct_type + no inferSchema in runner
"Metadata columns ds, _ingested_at, _source_batch_id"	add_metadata_columns present; _source_batch_id in its body
"dynamic partition overwrite (idempotent ds write)"	mode("overwrite").insertInto( in write_delta() AND no replaceWhere string anywhere in the runner
"empty_input_behavior respected"	EmptyInputError class + fail branch in ingest()
"SE runner loads YAML rules"	SparkExpectationsYamlRuleLoaderImpl in se_runner.py
"action_if_failed: fail|drop|ignore"	validation in load_table_config checking those three values
"Critical tables use fail"	test_per_table_configs.py::test_critical_tables_use_fail_behavior or direct YAML check

Print:

Story STORY-02-002 — Acceptance Criteria:
  AC 1: Runner reads per-table YAML config ................... PASS
  AC 2: Enforces StructType (no schema inference) ............ PASS
  AC 3: Metadata columns ds, _ingested_at, _source_batch_id .. PASS
  AC 4: dynamic partition overwrite (idempotent ds write) .... PASS
  AC 5: empty_input_behavior respected ....................... PASS
  Result: 5/5 AC PASS

If any AC fails (code structure is absent), report it as a CRITICAL and fix before declaring the story done.

Phase 5.5: Verification Compliance Self-Check (MANDATORY before reporting OK)

The story's ## Verification block is the contract. After every prior phase has emitted its files, run the AC verifier against the target story and refuse to declare OK if any mechanical verifier still fails.

python3 ${CLAUDE_PLUGIN_ROOT}/../scripts/verify_acs.py STORY-NN-NNN --json

(For batch / multi-story dispatch, run once per story.)

Parse the JSON output. For each AC in acs[]:

• status == "FAIL" and at least one check has a non-manual kind that failed → emit one CRITICAL line per failing check: CRITICAL STORY-NN-NNN AC<N>: <check.spec> — <check.detail> Then stop. Do NOT mark the plan task done. Do NOT print the OK trailer. The orchestrator's Phase 2 Step 3.5 reads this and halts the story.
• status == "FAIL" but every failing check is manual: → INFO only (manual checks can't fail mechanically; treat as author note).
• status == "PASS" / INDETERMINATE → continue.
• has_verification == false → emit one WARNING line STORY-NN-NNN: no Verification block — generation completed without AC compliance check. Then continue.

This phase is the only place this skill flips its overall result from OK to FAILED. Skills that ignore it leave gaps the orchestrator cannot see.

Output Summary

At the end, print a table: Module/Config | Path | Action (created|updated|skipped) so the user can see exactly what changed.

In story mode, prefix the table with the story ID and AC result summary:

Story: STORY-02-002 — Generic Ingestion Runner | Result: 5/5 AC PASS

Module/Config             | Path                                              | Action
ingestion_runner.py       | src/{project_name}/bronze/ingestion_runner.py     | created

Learnings & Corrections

Active Learnings

• L-001 (2026-05-12): Always: Default per-table YAML source.type=csv (let Spark read natively). Reserve duckdb source type for tables <100MB. Document the OOM risk.
• L-002 (2026-05-12): Always: Airflow 3.x SparkSubmitOperator dropped the driver_cores kwarg. Forward the value via spark.driver.cores in the conf dict.
• L-003 (2026-05-12): Always wrap the input df in a no-arg lambda: decorated = se.with_expectations(...)(lambda: df); validated = decorated().
• L-004 (2026-06-20, supersedes the 2026-05-12 path-based note): Write SE error/stats tables as MANAGED Unity Catalog tables addressed by a FULLY-QUALIFIED catalog.schema.table name (UC 0.5.0). Pass the FQN target to with_expectations so SE's derived f"{target}_error" is a well-formed 3-part name; use writers WITHOUT .option("path", ...). The 0.4.0 path-based workaround existed only because UCSingleCatalog crashed on the BARE error-table name with an empty-namespace ArrayIndexOutOfBoundsException in fullTableNameForApi (NOT an RTAS refusal); UC 0.5.0 fixes the namespace handling and supports managed creates, so the error table is re-enabled (se.enable.error.table=True) and becomes the rejected-row audit trail. Requires a schema managed location (storage_root) from uc_init.py.
• L-005 (2026-05-12): Always: SE evidence check filters on meta_dq_run_date (= Airflow ds) only — NOT on meta_dq_run_id. The contract proves 'DQ ran for today's data', not 'DQ ran for this exact Airflow run'.
• L-006 (2026-06-18): Always: the idempotent Bronze write is dynamic partition overwrite — df.write.mode("overwrite").insertInto("unity.bronze.<table>") with the session's spark.sql.sources.partitionOverwriteMode=dynamic (owned by the scaffold's build_spark_session). NEVER pass .option("replaceWhere", ...) to insertInto: empirically confirmed via Delta DESCRIBE HISTORY that insertInto SILENTLY IGNORES replaceWhere (it only applies to .save()/.saveAsTable()), so re-runs APPEND and double the data. replaceWhere also cannot coexist with partitionOverwriteMode=dynamic (throws DELTA_REPLACE_WHERE_WITH_DYNAMIC_PARTITION_OVERWRITE). Per LLD v1.14 §2.3 / §13 Decision 15.
• L-007 (2026-06-21): Emitted ingestion Python (factory/runner/wrapper + bronze tests) MUST pass make lint (ruff select=[E,F,I,UP], line-length 100). This skill had no lint gate and shipped UP035 (from typing import Mapping — import Mapping from collections.abc) plus E402/F401/I001 import debt in emitted tests, silently red-lining make lint. As the FINAL step ALWAYS run uv run ruff check --fix src/ tests/ && uv run ruff format src/ tests/ and confirm make lint exits 0 before reporting done (keep every import at module top — never mid-file).