Check dbt Pipeline Conventions

This skill audits the codebase against the dbt pipeline conventions. It is LLM-based correctness validation — complements cfs validate which only checks artifact structure and code-marker presence (not engine config content).

Source of truth

Before reporting anything, read this THIS turn:

• .cf-studio/config/rules/architecture.md — "dbt Pipeline Conventions (summary)" states the hard rules this skill enforces

Where that file and the checks below disagree, the rules file wins — update the checks to match it.

What to check

Check 1 — Silver models: engine + order_by + write strategy

For every .sql file under src/ingestion/silver/ (excluding crm.disabled):

• engine must be insight_engine('ReplacingMergeTree', '_version') OR insight_engine('ReplacingMergeTree') (versionless, only for materialized='table'). A literal engine string is a violation whatever family it names — it cannot follow the install's topology.
• order_by must be ['unique_key']
• materialized must NOT be 'view' (views are forbidden for silver)
• If materialized='incremental' → incremental_strategy='delete+insert' AND unique_key='unique_key' (silver is physically deduplicated so any consumer — gold views, the product — can read silver WITHOUT FINAL). incremental_strategy='append' in a silver model is now a FAIL.

Bash discovery:

find src/ingestion/silver -name "*.sql" -not -path "*disabled*"

For each file: read the {{ config(...) }} block. Report violations with file path + the offending line.

Check 2 — Connector staging models: engine + order_by + unique_key projection

For every .sql file under src/ingestion/connectors/*/dbt/:

• If materialized is incremental or table → engine=insight_engine('ReplacingMergeTree', '_version') + order_by=['unique_key']
• If materialized is view → confirm it's a thin pass-through (no GROUP BY / window)
• If materialized is ephemeral → confirm it's a pass-through over a staging table dbt does not own (none exist today; the Jira field history is derived in dbt)
• The SELECT body MUST project a unique_key column (either propagated from bronze: u.unique_key AS unique_key, or computed: CAST(concat(...) AS String) AS unique_key)

Bash discovery:

find src/ingestion/connectors -name "*.sql" -path "*/dbt/*" -not -path "*disabled*"

Check 3 — unique_key formula in connector record producers

For every Airbyte YAML connector at src/ingestion/connectors/*/connector.yaml:

• Find every AddFields block with path: [unique_key] (or path: [unique] — that's a deviation)
• Formula MUST start with {{ config['insight_tenant_id'] }}-{{ config['insight_source_id'] }}-
• Connector-specific tail follows (e.g., {{ record['id'] }} or composite)

For every Python CDK connector at src/ingestion/connectors/*/source_*/:

• Check streams/base.py (or equivalent) for a _make_unique_key helper
• Helper must include tenant_id and source_id as the first two arguments
• Helper output must contain those two before the natural key parts

Check 4 — no promotion machinery reappears

Bronze tables are ReplacingMergeTree by construction (the destination creates them via append_dedup with primary key unique_key); the old bronze→RMT promotion is retired (#2877). FAIL if any of these reappear:

• a promote_bronze_to_rmt macro or any call to it
• a <connector>__bronze_promoted.sql model
• a -- depends_on: {{ ref('<connector>__bronze_promoted') }} line

Check 5 — Airbyte sync mode in the catalog builders

• src/ingestion/reconcile-connectors/python/normalize_catalog.py must emit destinationSyncMode = "append_dedup" with primaryKey = [["unique_key"]] for every stream, and FAIL (non-zero exit) on any stream whose schema lacks a unique_key property
• src/ingestion/scripts/bootstrap-db/create-connector-tables.sh must enforce the same contract (snapshot shape = real sync shape; keyless stream = hard error)
• Must NOT emit append or overwrite anywhere

Check 6 — Ephemeral wrapping for staging tables dbt does not own

• Any model materialized as ephemeral must SELECT only from source(...) (not ref(...)) — i.e., it's a thin wrapper for a non-dbt-managed table
• The union_by_tag macro must contain the ephemeral handling branch (check src/ingestion/dbt/macros/union_by_tag.sql for materialized == 'ephemeral')

Check 7 — Incremental by default for silver/staging

Every silver model and every dbt-owned staging model with append/event semantics MUST be materialized='incremental'. materialized='table' is allowed ONLY in three justified cases:

1. Full-refresh source + current-state semantics: small lookup/dimension tables whose upstream is a full-refresh API.
   • Allowed: class_people, class_hr_working_hours
2. Aggregation that scans all data: heavy GROUP BY / multi-CTE join requiring full rebuild.
   • Allowed: mtr_git_person_totals, mtr_git_person_weekly
3. Explode/fan-out staging over full-refresh bronze: one bronze row → many output rows AND bronze is full-refresh + overwrite.
   • Allowed: jira__changelog_items, jira__issue_field_snapshot

For each .sql file under src/ingestion/silver/ and src/ingestion/connectors/*/dbt/:

• If materialized='table' AND model name is in the allow-list above → PASS
• If materialized='table' AND model name is NOT in the allow-list → FAIL with suggestion: "Convert to materialized='incremental'. For silver use incremental_strategy='delete+insert' + unique_key='unique_key'; for staging use incremental_strategy='append'. Scope the incremental boundary to one source instance with {{ silver_incremental_watermark([...]) }} — never a table-wide max(_version), which lets one producer's boundary permanently exclude a slower producer's rows. If upstream lacks _version, amend the SELECT to project toUnixTimestamp64Milli(_airbyte_extracted_at) AS _version."
• If materialized='view' for a silver class_* / fct_* / mtr_* → FAIL (views forbidden in silver per check 1)
• If materialized='ephemeral' → cross-checked by Check 6

Bash discovery:

grep -lE "materialized\s*=\s*'table'" src/ingestion/silver/ src/ingestion/connectors/*/dbt/ -r 2>/dev/null

Cross-reference each match against the allow-list and report PASS / FAIL.

Check 8 — Read-time deduplication of every RMT read

RMT only collapses duplicates on background merge (never guaranteed at query time), so every read of an RMT relation must be deduplicated at read time unless the producer is physically unique. A duplicate that slips through inflates metrics (e.g. the slack_active_days = 42 incident).

A read is OK if ANY of:

• it goes through the union_by_tag macro (which dedups the union by unique_key via QUALIFY ROW_NUMBER() … ORDER BY _version DESC, or LIMIT 1 BY for versionless), OR
• the read carries FINAL / argMax / QUALIFY ROW_NUMBER / LIMIT 1 BY in its own subquery scope, OR
• the producer is incremental_strategy='delete+insert' or materialized='table' (physically unique → reader needs no dedup).

FAIL conditions (scan connectors/*/dbt/ + silver/ + gold migrations scripts/migrations/*.sql):

1. A direct ref()/source() read of an RMT producer (append/incremental) with no dedup in scope and not via union_by_tag.
2. Aggregation over un-deduplicated bronze: count()/sum()/avg() over a source('bronze_*') read without FINAL or LIMIT 1 BY unique_key (bronze is ReplacingMergeTree). Inflation is baked into one output row and is unrecoverable downstream (e.g. bitbucket_cloud__commits, claude_admin__ai_dev_usage).
3. A snapshot() macro call whose source is read without FINAL (spurious SCD2 versions). The shared macros/snapshot.sql must read FROM {{ source_ref }} FINAL.
4. A gold view in scripts/migrations/*.sql reading a staging.*/silver.* append+RMT table without FINAL. (Reads of delete+insert silver are OK. Note migrations are immutable history — verify against the latest migration that (re)defines the view.)

Deterministic complement: src/ingestion/dbt/audit_rmt_read_dedup.py performs this scan mechanically and exits non-zero on a gap — runnable in CI. Use it to generate candidates, then confirm each by reading (it cannot resolve outer-scope dedup or column-level semantics).

How to run

When user invokes /check-dbt-conventions:

1. Read all 5 spec docs (DESIGN.md + 4 ADRs) THIS turn — confirm understanding
2. Run the 8 checks systematically using Glob / Grep / Read (optionally run python3 src/ingestion/dbt/audit_rmt_read_dedup.py for Check 8)
3. Report per-check status: PASS / FAIL with file paths + line numbers + the offending text
4. Summarize: total checks, failures by category
5. For each failure, suggest the minimal fix (e.g., "change order_by='(insight_source_id, comment_id)' to order_by=['unique_key']")

Output format

=== check-dbt-conventions ===
Rules read: .cf-studio/config/rules/architecture.md (this turn)

Check 1 — Silver engine + order_by — PASS (34 models, 0 violations)
Check 2 — Connector staging engine/order_by/unique_key — FAIL (2 violations)
  - src/ingestion/connectors/.../foo.sql:8 — order_by='(...)' should be ['unique_key']
  - src/ingestion/connectors/.../bar.sql:14 — missing unique_key projection in SELECT
Check 3 — unique_key formula — PASS
Check 4 — no promotion machinery — PASS
Check 5 — Airbyte append_dedup catalog builders — PASS
Check 6 — Ephemeral wrapping — PASS
Check 7 — Incremental-by-default — PASS
Check 8 — Read-time dedup of RMT reads — FAIL (1 violation)
  - src/ingestion/connectors/.../foo.sql:42 — count()/sum() over source('bronze_*') without FINAL/LIMIT 1 BY

Summary: 6/8 PASS, 2 FAIL (Checks 2, 8 — fixes above)

Refuse to report PASS without having read each spec doc this turn (anti-pattern: stale reasoning). Refuse to invent file paths — only report files actually scanned.

When to invoke

User-driven, on demand:

• Before merging a connector PR
• After refactoring silver models
• Periodically (CI) — run before dbt build to surface convention drift