---
description: Coding standards and technology stack rules for the dag-tools common resource repository.
---

# `dag-tools` Tech Stack & Rules

This project serves as the central hub for common Dagster utilities, resources, I/O managers, sensors, and asset wrappers used across our external repositories.

## 1. Dependency Management (`uv`)
- **Rule**: NEVER use `pip`, `poetry`, or `conda` directly. 
- **Tool**: Always use `uv` for all environment and dependency management.
- **Commands**: 
  - Install dependencies: `uv add <package>`
  - Run scripts: `uv run <script>`
  - Sync lockfile: `uv sync`
  - Start Dagster: `uv run dagster dev`

## 2. Shared Dagster Resources
- **Rule**: All assets and resources here must be highly generic and reusable. This repo does not execute domain-specific business pipelines.
- **Purpose**: Any IO Manager (e.g., S3, DuckDB), custom Sensor, Resource (e.g., dlt wrappers, API clients), or partition definitions used across multiple projects should be housed here.

## 3. Python Style
- **Type Hinting**: All function signatures and classes must include strict Python type hints (e.g., `def create_s3_io_manager(bucket: str) -> IOManagerDefinition:`).
- **Docstrings**: Provide a brief Google-style or standard docstring explaining the configuration kwargs and usage examples for all shared components. This is critical because other repositories will depend on these docstrings.
- **Imports**: Ensure all public-facing common tools are neatly exported in `dag_tools/__init__.py` so they are easily importable by downstream consumers (e.g., `from dag_tools import CommonS3IOManager`).

## 4. Ecosystem & Integrations Stack
- **Dependencies**: The framework explicitly supports `pyarrow`, `pandas`, `s3fs`, `httpx`, `pyodbc`, `oracledb`, and the `restate-sdk`. Do not reinstall duplicative dependencies natively (e.g. `requests` instead of `httpx`).
- **Dagster Version**: This project targets **Dagster 1.12+ (core)** / **0.28+ (libraries)**. The old experimental `dagster-components` library is deprecated — use `dagster.components` (built-in to core).
- **Restate SDK Constrains**: 
  - Never hardcode API Keys or DB passwords into Dagster `kwargs` mapped to Restate. Always pull from `os.environ.get()` inside Restate `ctx.run` blocks for strict Kubernetes injection isolation.
  - When fan-out querying DB sources during Restate extraction syncs, chunk Database executions strictly (e.g. 10,000 max size for HTTP DAG dispatches, 1,000 max size for legacy Oracle `IN` clauses).
  - **HTTP/2 Standard**: All Restate ASGI servers must use `Hypercorn` (or equivalent) to support HTTP/2, as required by modern Restate SDKs (`>=0.14.0`).
  - **Shared Worker Image**: Do NOT write bespoke Restate worker entrypoints or per-project Dockerfiles. Restate workers run the shared `restate-worker` image (repo-root `Dockerfile.restate-worker`); its env-driven entrypoint `dag_tools.restate_handlers.serve` selects services via `RESTATE_SERVICES` and self-registers via `RESTATE_ADMIN_URL` / `RESTATE_ADVERTISED_URI`. Register every new durable handler in `serve.SERVICE_REGISTRY` so it is selectable. Deployments configure workers purely through environment variables.

## 6. Docker Build Best Practices
- **Layer Caching**: When using `uv` with a local package, use the "metadata trick" (creating dummy `__init__.py` and `README.md`) to cache dependencies.
- **Shadowing Prevention**: ALWAYS `rm -rf` dummy metadata directories/files (created for caching) immediately before the final `uv pip install -e .` step. Failing to do so will cause the empty dummy directory to shadow the real package in the container's working directory, leading to `ModuleNotFoundError`.

## 4i. Verdict (`dag_tools.qual.verdict`)
- **Metadata parity is KEY-set only, never value comparison.** The recipe explicitly says "metadata values may differ" — a baseline timestamp metadata value will never equal the candidate's, and gating on that would NO_GO every qualification. Test `test_diff_compares_metadata_KEY_set_not_values` enforces this; don't relax it.
- **Duration deltas are reported but never gating.** Same recipe rule. A candidate that's 10x slower is fine for behavior preservation; it's the operator's call whether to GO. Do not add a "gates on duration" knob.
- **Strict by default.** Unknown gaps block GO. `GapAcceptance` exists so operators can explicitly accept "yes, I know orchestration isn't implemented and I'm OK going without it" — the verdict layer does NOT decide that on the operator's behalf. New gaps follow the same pattern: add a `GapAcceptance` field defaulting to False, surface a blocking issue when False, document the override flag in the CLI.
- **Candidate-preflight failure is a hard gate, period.** No GapAcceptance bypass exists or should be added — the operator should fix the deployment, not paper over it.

## 4h-1. Probe deploy-state status (`dag_tools.qual.probes.status`)
- **Location name is HARD-CODED to `dag-tools-probes`** — same constant the runner uses. If a future deployment needs configurability, surface it through the manifest, not a per-call override.
- **`ABSENT` is distinct from `LOADED` empty.** The workspace returning no `dag-tools-probes` entry is its own diagnostic state (operator forgot to register the location); don't collapse it into "no probes."
- **Partial loads are visible, not silenced.** Upstream-loaded-downstream-missing (or vice versa) usually means an `_IO_MANAGER_ERROR` fallback path is masking a class-import problem. Surface it; the operator needs the asymmetry to triage.
- **Unexpected probe-shaped assets are INFORMATIONAL.** Stale files in `DAGTOOLS_PROBES_DIR` get flagged but don't fail `all_loaded` — forcing operators to wipe the probe dir every iteration would be hostile.

## 4h0. Probe runner (`dag_tools.qual.probes`) + Q6 synthetic coverage
- **Probe state is in its OWN slot — never mixed with runnable-rep state.** `qualifications/<qual_id>/<side>/probes/state.json` for probes; `<side>/state.json` for runnable reps. Run records mirror: `<side>/probes/runs/.../` vs `<side>/runs/.../`. The two paths must not collide because a class can legitimately be both RUNNABLE (covered by a rep) and SYNTHETIC_REQUIRED (covered by a probe), and Q6 reads both.
- **The probe state schemas are SEPARATE TYPES** (`ProbeRepState` / `ProbeRepStatus`) from the rep state types (`RepState` / `RepStatus`). They have the same status values but distinct types — accidental cross-wiring fails type checks at the call site instead of silently working.
- **Launch BOTH the upstream AND downstream asset in ONE run.** Selecting only the downstream does NOT auto-materialize the upstream — Dagster treats it as a loaded input whose output must already exist, so downstream-only dies with `DagsterExecutionLoadInputError`. Both must run together: upstream writes via the real IO manager, downstream reads it back and asserts identity (the write→read round-trip is the probe's whole purpose). Live-tested against `dagster dev`; regression `test_run_probes_side_launches_against_dag_tools_probes_location` asserts both keys. Do NOT "optimize" back to downstream-only.
- **Location name is hard-coded to `dag-tools-probes`.** That's the contract with `dag_tools.probes_location`. If a future deployment needs configurability, surface it through the manifest's deployment config — not through a per-call override.
- **Reuse the Q2 `RunRecord` schema.** Probe runs use the same persisted shape so Q6 can diff them the same way runnable-rep runs are diffed.
- **`synthetic_classes_with_probe_coverage`** counts classes where probes PASSED on BOTH sides AND (when records exist) diff cleanly. One-sided wins don't count.
- **`synthetic_classes_red`** lists classes whose probes ran-and-FAILED OR ran-and-DIVERGED (both PASSED but records differ — exactly the regression Q6 exists to catch). Blocks GO **regardless** of `--accept-synthetic-coverage-missing` (which only excuses *missing* coverage). No opt-out exists for synthetic_classes_red. Regressions: `test_verdict_no_go_when_probe_failed_even_with_synthetic_accept`, `test_verdict_diverged_probes_blocks_go_even_with_passing_status`.
- **Graceful degradation.** PASSED state with no readable run record still counts as covered. The diff augments; it doesn't supplant.
- **`ClassVerdict.probe_diff`** is attached for SYNTHETIC_REQUIRED classes only. `UPGRADE_VERDICT.md` shows it in a dedicated probe-failure section for operators.

## 4h1. Synthetic probes (`dag_tools.qual.synthetic` + `dag_tools.probes_location`)
- **Generated source must always parse and the code location must always load.** The `dag-tools-probes` user-code location holds modules for every `SYNTHETIC_REQUIRED` class; one broken import blocks the whole location. The generator imports the real IO manager FQN inside a `try/except` with an `InMemoryIOManager` fallback and stamps the import error on `_IO_MANAGER_ERROR`; the probe asset raises that error at materialization, where operators can triage one probe at a time. Regressions: `test_generated_source_parses_as_python`, `test_generated_source_has_inmemory_fallback_for_missing_io_manager`.
- **Each probe binds its IO manager under a CLASS-UNIQUE resource key (`io_manager_<short>`).** `Definitions.merge` raises on conflicting resource keys; using bare `"io_manager"` for every probe would collide on merge and prevent the dag-tools-probes location from assembling more than one class. Regression: `test_each_probe_uses_class_unique_io_manager_resource_key` + `test_loaded_probes_merge_into_one_definitions_without_resource_collision`. Do not "simplify" back to a shared key.
- **Probes must NOT import `dag_tools.qual.*`.** They live in a separate code location with its own deploy cadence. If a probe imports the qual package, every `dag-tools` release becomes a probe redeploy. Regression: `test_generate_probe_source_is_self_contained`.
- **`probe_manifest.json` is written LAST.** `publish_bundle` writes every `<class_hash>.py` first and only then the manifest — same "pointer-written-last" invariant as `latest.json`. Don't reorder.
- **The dag-tools-probes location soft-fails per probe.** `load_probes_from_dir` captures failures on `ProbeLoadReport.failures` and continues — one syntax-broken `<class_hash>.py` MUST NOT block the whole location. The operator triages from the report. Regression: `test_loader_soft_fails_a_malformed_probe`.
- **The dag-tools-probes location loads cleanly with NO probes deployed.** `DAGTOOLS_PROBES_DIR` unset or pointing at a missing dir yields an empty `Definitions` — operators deploy the location once and add bundles over time. Regression: `test_loader_returns_empty_report_when_env_unset`.
- **Run + Q6-coverage integration is pending.** `qual synthetic` produces + the location deploys, but the Q2 runner doesn't yet launch probe assets per class and Q6 still requires `--accept-synthetic-coverage-missing` for synthetic classes. Until that integration ships, don't fake the coverage signal in `build_verdict`.
- **v1 deferrals surface as `ProbeModule.notes`**, not silent gaps. Partitions defs, partition mappings, and custom dbt translators that the generator can't synthesize are captured in notes and printed in the bundle output. When extending the generator, expand the synthesis; do not drop the notes path.

## 4h. Preflight (`dag_tools.qual.preflight`)
- **All checks become entries in `PreflightReport.checks`, never raise.** A `DagsterGraphQLError` from version lookup or workspace query becomes a failed `CheckResult` with the message in `detail`. Operators read the report; the runner should not crash because the deployment is misbehaving — that's WHY they're running preflight.
- **The candidate-side run-rendering check uses the baseline state, NOT the registry's `latest.json` or the class matrix.** Sampling reads `state.json` for the baseline side, deterministically sorts PASSED run_ids, and probes the top N. This is the event-log back-compat spot check — its whole point is reusing the runs that actually executed during Q2.
- **The baseline side skips the run-rendering check vacuously.** No priors exist yet. Don't add inference like "check baseline runs against the baseline deployment" — that's tautological and breaks the test `test_baseline_preflight_does_not_sample_baseline_runs`.
- **Version wildcards use `.x` suffix only.** `1.12.x` matches `1.12.0` through `1.12.999`; do NOT try to add SemVer-style range parsing — operators want to write what the manifest already shows them.

## 4g. Run execution (`dag_tools.qual.runs` + `dag_tools.qual.graphql`)
- **Resumability is the load-bearing invariant.** After every state transition (`pending` → `launched` → `passed`/`failed`), the runner saves `QualRunState` to both `~/.dagtools/quals/<id>/<side>-state.json` AND the registry's `qualifications/<id>/<side>/state.json`. If you add a new state transition, save state after it — tests in `test_runs_runner.py` enforce this.
- **Re-invocation MUST skip PASSED.** And it MUST reconcile LAUNCHED via GraphQL poll, not re-launch. A naive "just relaunch everything pending-or-not-passed" would burn duplicate compute and confuse downstream Q6 by writing two run records. The regression tests `test_run_side_skips_passed_reps_on_re_invocation` and `test_run_side_reconciles_launched_via_poll_not_relaunch` enforce both.
- **GraphQL queries use public fields only.** No `dagster._core.*` access. Add a fallback path for any field that's seen drift across the version range; soft-fail per-field rather than raise. New queries go in `graphql/client.py`; the launcher + runner consume them.
- **Only select GraphQL fields that EXIST on the target type — a nonexistent field 400s the WHOLE request.** `... on InvalidStepError { message }` broke every launch because `InvalidStepError` has only `invalidStepKey`. When editing a mutation/query, introspect the live schema first (`__type(name){ possibleTypes { name fields { name } } }`) and select only present fields. Mocked-transport unit tests can't catch this — it needs live testing or introspection. Regression: `test_launch_mutation_does_not_select_message_on_fieldless_error_types`.
- **The launch location/job come from the manifest, not a hardcoded default.** `run_side` passes `manifest.deployment.location_name` / `job_name` to `launch_representative`. The `"default"` fallback is degenerate — real deployments name their location after the user-deployment; `dagtools qual init --location-name` sets it. Regression: `test_run_side_threads_manifest_location_and_job_into_launch`.
- **No GraphQL library dependency.** httpx + raw query strings is the discipline. Tests mock httpx, not a GraphQL schema — keeps CI fast and portable.

## 4f. Equivalence-class matrix (`dag_tools.qual.classes`)
- **Read the manifest, not the registry.** Q1 (and every later Q-phase) walks `manifest.inventory_pins[]` and fetches each pinned `(repo, git_sha)` from the registry. Do **not** call `read_latest_pointer` to discover repos — that would let mid-qualification CI publishes shift the qualification target silently. The regression `test_classes_builder.py::test_q1_reads_pinned_sha_not_latest` enforces this.
- **The class key is THE contract**, not the asset key. Two assets with identical key components land in the same class; they're then treated as interchangeable for qualification purposes. Adding new key components is fine (they go in `ClassKeyComponents`) but they're effectively additive — old class hashes won't match new ones, so prefer doing this only at qual_id boundaries.
- **Custom dbt translators always force own classes.** The "is_custom_translator" signal comes from `dbt_projects.json`. When extending the dbt detection (e.g. multiple dbt projects in one repo), keep it conservative — over-segregation costs operator time, but missing a custom translator costs upgrade-day surprises.
- **Runnability is tag-driven**, not inference-driven. `synthetic_required: "true"` / `observe_only: "true"` are the contract; do not add FQN-based heuristics ("snowflake means synthetic_required") without a manifest knob to disable them — different orgs have different staging realities.

## 4e. Qualification manifest (`dag_tools.qual.qualify`)
- **One qual_id is immutable.** The manifest at `qualifications/<qual_id>/manifest.yaml` is written via `InventoryRegistry.put_qualification_manifest` which defaults to immutable. `--allow-overwrite` exists for the explicit "I really do want to redo this qualification" path; do NOT silently bypass.
- **Inventory pinning is point-in-time.** `create_qualification` reads `latest.json` for every repo once at init time and freezes those `(repo, git_sha)` pairs in `inventory_pins[]`. Later phases MUST read the manifest's pins, not call `read_latest_pointer` again — that would let mid-qualification CI publishes shift the qualification target.
- **`co_upgrade_risks` filters the Dagster family.** `compute_co_upgrade_risks` excludes anything matching `dagster*` from the diff because those are the explicit upgrade target. When extending the diff (e.g. to allowlist additional libraries), keep this filter in `dag_tools/qual/qualify/risks.py`, not scattered through callers.
- **YAML aliases match the recipe sample.** `CoUpgradeRisk` uses `from`/`to` aliases (not `from_version`/`to_version`) so the on-disk YAML matches what operators read in `docs/RECIPE.md`. Always serialize with `by_alias=True`.

## 4d. Read the Recipe before changing the qualification system
- The full spec for `dag_tools/qual/` (and the shared `dag_tools/inventory/` contract it depends on) lives in [`docs/RECIPE.md`](docs/RECIPE.md). It contains the ADRs (why the registry is separate from the gateway; why schema discipline replaces a package split; why FQN+MRO classification beats substring matching; etc.) and a table of regression tests that guard recipe invariants. Read it before changing publisher control flow, layout helpers, or schema fields.

## 4c. Survey (`dag_tools.qual.survey`)
- **Load failure means publish nothing.** `run_survey` refuses to write to the registry if any code location fails to load. This is THE recipe invariant for Phase 1 — the registry must never contain an inventory for code that doesn't load. The test `test_run_survey_refuses_to_publish_when_load_fails` enforces it; if you change publisher control flow, that test will tell you.
- **Capture every warning during load.** Use `_capture_warnings` (equivalent to `-W all`) around every `importlib` call. Warnings are part of the load-validation payload — they're how operators see Dagster deprecations and experimental APIs drifting across the fleet.
- **Soft-fail per item in introspection**, same discipline as `dag_tools.inventory.extractors`. One malformed sensor or dbt resource must never abort a whole introspection pass.
- **Schema versions live in `survey/schemas.py`** (one per artifact). Same additive-only evolution rules as `AssetRecord`.
- When adding a new artifact: define a pydantic model in `schemas.py` with its own `SCHEMA_VERSION_*` constant, add a filename constant to `registry/layout.py`, plumb it through `publisher.run_survey`'s artifact dict, and bump the schema version in any new commit that adds fields.

## 4b. Qualification Registry (`dag_tools.qual.registry`)
- **Layout is canonical**: every S3 key is constructed via helpers in `dag_tools/qual/registry/layout.py`. **Never** write a literal key string anywhere else — writers and readers must agree on the layout via the module, not by convention.
- **Immutability is enforced**: every per-build artifact and every per-qualification artifact is written via `S3Storage.put_immutable`, which refuses to overwrite (HEAD-then-PUT). The only mutable keys are `inventory/<repo>/latest.json` and the equivalent qualification pointers; those use `put_mutable`.
- **`latest.json` is written LAST**: `InventoryRegistry.publish_build` writes all per-SHA artifacts first, then updates the pointer. **Do not reorder.** This is the invariant that prevents readers from observing partial builds — keep all pointer writes at the end of their respective publish paths.
- **Machine-readable by default**: every `dagtools` CLI command emits JSON unless `--format table` is set. New sub-commands MUST follow this — operators pipe `dagtools` output into other tools.

## 4a. Inventory Contract (`dag_tools.inventory`)
- The shared `AssetRecord` schema in `dag_tools/inventory/schema.py` is consumed by **both** the runtime Domain Broker and the CI qualification survey. Cross-process compatibility is the whole point.
- **Evolution is strictly additive**: never rename or remove a field; only add `Optional[...]` fields with defaults. Bump `SCHEMA_VERSION` in the same commit. Readers use `extra="ignore"` to tolerate unknown fields from newer writers.
- For new IO manager classes, add an explicit entry to `dag_tools/inventory/classifier.py::FAMILY_REGISTRY` (FQN -> family). Do **not** rely on substring matching — it's a logged-at-WARNING last-resort fallback, never the intended path.
- For introspection that may not exist in older Dagster versions, version-gate it and **fail soft per-field** (record `None` + log WARNING). The extractor must never abort an entire `Definitions` walk because one asset is malformed — per-asset try/except is mandatory.

## 5. Dagster Component API Pattern (1.12 GA)
- **Rule**: All custom components MUST use the triple-inheritance pattern: `class MyComponent(Component, Resolvable, Model)`.
- **Imports**: `from dagster.components import Component, ComponentLoadContext` and `from dagster.components.resolved.base import Resolvable` and `from dagster.components.resolved.model import Model`.
- **No `ComponentSchema`**: Schemas are plain `pydantic.BaseModel` or declared directly on the component class as typed fields.
- **No manual `load()` overrides**: The `Resolvable` base handles YAML attribute parsing automatically.
- **`build_defs(self, context: ComponentLoadContext) -> Definitions`**: This is the only method you must implement.
- **Factory pattern for `@asset` closures**: Dagster 1.12 treats all function parameters (except `context`) as asset inputs. When creating `@asset` functions inside `build_defs()`, use a factory function to capture closure variables:
  ```python
  # WRONG: Dagster treats 'pk' and 'table' as asset inputs
  @asset(name=name)
  def my_asset(context, pk=pk_value, table=table_name): ...
  
  # CORRECT: Factory captures values via closure
  def _make_asset(_name, _pk, _table):
      @asset(name=_name)
      def my_asset(context): ...  # use _pk, _table from closure
      return my_asset
  ```
