# dag_tools Project Context
This document provides high-level context for AI tools and agents exploring this repository.

## Domain
`dag_tools` is the foundational, shared library for all of our Dagster data engineering projects. It does not run specific ETL pipelines itself. Instead, it provides the generalized infrastructure that other projects (like `pub-tools`) rely upon.

## Architecture
This repository contains:
1.  **IO Managers**: Custom Dagster IO Managers for handling data serialization/deserialization across cloud storage (S3) and databases (DuckDB, Postgres). Includes advanced native `ArrowIOManager` coercing S3 PyArrow formats into Pandas DataFrames.
2.  **Resources**: Reusable clients and connections (e.g., API clients, database connection pools, `dlt` pipeline wrappers, `ArrowResource`, `S3SensorResource`).
3.  **Sensors & Schedules**: Generalized sensors (e.g., S3 bucket sensors parsing `.metadata` flags for Chronological PyArrow partitioning) and common schedule definitions.
4.  **Asset Wrappers**: Traditional Python factory functions or decorators that standardize repetitive `@asset` patterns across the organization.
5.  **Components**: Formal Dagster 1.12 GA Declarative Components (`dagster.components`). Components use the `Component, Resolvable, Model` pattern and allow downstream users to deploy complex pipelines (e.g. `s3_sensor`, `dlt_pipeline`, `datahub_lineage`, `restate_dlt_sync`, `restate_api_sync`) using pure YAML architecture. Schemas are plain `pydantic.BaseModel` classes.
6.  **Utilities**: Helper functions for logging, metadata formatting, AssetKey normalization (`AssetNormalizationRegistry`), testing, dynamic Kubernetes resource tag resolution (`dag_tools.utils.k8s`), and multi-environment dbt compilation/validation (`scripts/compile_and_validate_dbt.py`).
7.  **Restate Handlers (Data Plane)**: Generic durable execution handlers (`GenericOracleAckService`, `GenericApiSyncService`, `SapInductionService`) that execute isolated API pushes and Database commits durably. The `SapInductionService` implements a **Durable State Machine** (NEW -> PENDING -> SUCCESS) with persistent retry tracking.
7a. **Inventory Contract (`dag_tools.inventory`)**: The shared structural-inventory schema for Dagster assets. A versioned `AssetRecord` pydantic model (`schema_version`, IO manager FQN + `family`, partitions/mapping FQNs, resource classes, integration libs, automation condition type, freshness/backfill, tags, asset checks, jobs, and a DataHub URN sidecar) plus a soft-failing extractor that walks a `Definitions`. **Two consumers**: the runtime Domain Broker (uses `io_manager_family` for mesh routing) and the `dagtools survey` CLI (publishes per-build inventory to MinIO for the Dagster upgrade qualification system). Evolution is **additive-only**; the long-lived Broker and the CI survey will be reading each other's records across versions.
7b. **Qualification System (`dag_tools.qual`)**: The desktop `dagtools` CLI (Typer) and the MinIO/S3 registry that backs it. `dag_tools.qual.registry` provides the `S3Storage` (immutable vs. mutable put modes), `InventoryRegistry` (composes layout + storage; `publish_build` writes per-SHA artifacts immutably then a mutable `latest.json` pointer LAST so readers never observe partial builds), and `compute_staleness` (the staleness reporter behind `dagtools registry status`). The layout module (`registry/layout.py`) owns every object-store key string. `dag_tools.qual.survey` provides `run_survey` — the publisher that loads every code location with `-W all` warning capture, refuses to publish if **any** load fails, and otherwise introspects assets, automation, io_managers summary, and dbt projects (custom-translator flagged). `dag_tools.qual.qualify` provides Q0 of Phase 2: `create_qualification` pins the registry's inventory snapshot into a `QualificationManifest` (immutable per qual_id), records the baseline/candidate version pair, and diffs non-Dagster library pins into `co_upgrade_risks[]` so a hidden dbt-core / warehouse bump can't propagate as a false Dagster regression. `dag_tools.qual.classes` provides Q1: `build_class_matrix` reads the manifest's pinned inventories (NOT the registry's current latest), groups assets by the recipe key (compute kind, IO manager FQN, partitioning, resources, integration libs, asset checks, automation condition + custom dbt translator FQNs), hashes each class deterministically (SHA-256[:12]), picks representatives spanning repos with `regression`-tag preference, classifies each as `RUNNABLE` / `SYNTHETIC_REQUIRED` / `OBSERVE_ONLY` by opt-in tags, and publishes `equivalence_classes.json` + `equivalence_classes.md` immutably. `dag_tools.qual.graphql` provides a version-tolerant `DagsterGraphQLClient` (launch + poll + event-log over httpx, no GraphQL library dependency, auth via `env:VAR_NAME` resolution). `dag_tools.qual.runs` provides Q2: `run_side` launches every RUNNABLE representative through the test deployment's Dagster GraphQL, polls each to terminal, builds a `RunRecord` (materializations + asset-check results + metadata-key union + failure step keys) and persists it under `qualifications/<qual_id>/<side>/runs/<class_hash>/<run_id>.json`. The per-`(qual_id, side)` `QualRunState` (pending/launched/passed/failed/skipped) is mirrored to the registry after every transition for desktop-crash recovery — re-invocation skips PASSED, reconciles LAUNCHED via run-id lookup rather than relaunching. `dag_tools.qual.preflight` provides Q3: `run_preflight` runs three checks against the test deployment via GraphQL — version matches the manifest's baseline/candidate (with `1.12.x`-style wildcard support), every code location reports `LOADED` (per-location error surfaced for triage), and (candidate side only) a deterministic sample of PASSED baseline runs still renders. Publishes `preflight.json` immutably. `dag_tools.qual.verdict` provides Q6 — the operator payoff: `build_verdict` walks every artifact under a qual_id (manifest, class matrix, both sides' state + run records, candidate preflight), diffs each representative's baseline vs candidate run for success / materialization count + asset-key set / metadata KEY-set (values may differ) / asset-check parity, rolls up to per-class verdicts, then applies the recipe's GO logic. Strict by default — `GapAcceptance` opt-ins are required to skip deferred orchestration, missing synthetic probe coverage, and co_upgrade_risks. Publishes `verdict.json` + `UPGRADE_VERDICT.md` immutably. `dag_tools.qual.synthetic` provides Q5 (generation): `generate_bundle` walks the class matrix, emits a self-contained Dagster module per `SYNTHETIC_REQUIRED` class (real IO manager FQN imported by name with an `InMemoryIOManager` fallback so the code location always loads; deterministic dict payload; upstream/downstream pair asserting payload identity after the IO manager round-trip; class-unique `io_manager_<short>` resource key so N probes merge cleanly), and produces a `ProbeManifest`. `publish_bundle` writes every `<class_hash>.py` to `qualifications/<qual_id>/probes/` first and only then `probe_manifest.json` (same "pointer-last" invariant as `latest.json`). `write_local_bundle` mirrors the bundle to `~/.dagtools/quals/<id>/probes/` for operator pickup. Generated probes deliberately do NOT import `dag_tools.qual.*` so the `dag-tools-probes` location's deploy cycle stays decoupled from `dag-tools` releases. v1 deferrals (partitions defs, partition mappings, custom dbt translators) surface as `ProbeModule.notes` so operators see the gap rather than silently miss it. `dag_tools.probes_location` is the deployable Dagster code location operators point their test deployment at — `load_probes_from_dir` dynamically loads every `<class_hash>.py` from `DAGTOOLS_PROBES_DIR`, merges via `Definitions.merge`, and soft-fails per probe so one broken file never blocks the whole location. The location loads cleanly with no probes deployed (so it can be deployed once and added to over time). `dag_tools.qual.probes` provides Q5c: `run_probes_side` launches each probe's downstream asset (deps pulls upstream automatically) against the `dag-tools-probes` user-code location via the same Dagster GraphQL launcher Q2 uses, polls to terminal, builds a `RunRecord` (reusing the Q2 schema for symmetry), and persists under `<side>/probes/runs/<class_hash>/<run_id>.json`. The per-(qual_id, side) `ProbeRunState` lives in a SEPARATE slot from the runnable-rep state (`<side>/probes/state.json` vs `<side>/state.json`) — a class can be both RUNNABLE (covered by rep) and SYNTHETIC_REQUIRED (covered by probe) with neither path clobbering the other. PASSED is sacred; LAUNCHED probes reconcile via run-id lookup rather than relaunching. Q6 reads probe state AND probe RunRecords on both sides: synthetic classes with PASSED probes on both sides AND (when records exist) a clean per-record diff count toward `synthetic_classes_with_probe_coverage` (removing the missing-coverage blocker), and synthetic classes whose probes either ran-and-FAILED or DIVERGED (both PASSED but materialization / metadata / asset-check parity broke) populate `synthetic_classes_red` and block GO **regardless** of `--accept-synthetic-coverage-missing` (which only excuses *missing* coverage). `ClassVerdict.probe_diff` carries the per-class diff for `UPGRADE_VERDICT.md` so operators see exactly what diverged. Graceful degradation: PASSED states with no readable run record still count as covered — the diff augments, doesn't supplant the status check. `dag_tools.qual.probes.status` (Q5d) provides `check_probes_status`: a GraphQL cross-reference of the probe manifest against the test deployment's `dag-tools-probes` location. Per probe it determines fully_loaded (both upstream + downstream loaded), partially_loaded (one missing — usually a probe-side import error), or missing (operator hasn't redeployed yet, `DAGTOOLS_PROBES_DIR` points elsewhere, or location is ERROR); also flags `unexpected_probe_asset_keys` from stale bundles still in the probes dir. `ABSENT` is reported as a distinct location load status when the workspace doesn't even register the location. CLI exposes `dagtools survey`, `dagtools registry status`, `dagtools qual init`, `dagtools qual classes`, `dagtools qual run`, `dagtools qual preflight`, `dagtools qual synthetic`, `dagtools qual probes run`, `dagtools qual probes status`, and `dagtools qual report`; `templates/Jenkinsfile.survey` is the Jenkins shared-library snippet. Phase 1 done; Phase 2 Q0+Q1+Q2+Q3+Q4+Q5 (generation + deploy target + runner)+Q6 (with probe coverage integration) done. The Q2 IO round-trip probe and local orchestration snapshots are deferred (orchestration is a separate command).
8.  **Control Plane vs. Data Plane Architecture**: The project is split into a Control Plane (Dagster orchestrator, Central Gateway, Domain Broker) and a Data Plane (Restate workers, Cortex Data Client). Restate workers run a single shared image (`Dockerfile.restate-worker`, published by CI). Its env-driven entrypoint `dag_tools.restate_handlers.serve` selects which handlers to host via `RESTATE_SERVICES` and self-registers with Restate on startup — consumers deploy the image and configure it purely through environment variables rather than writing bespoke entrypoints.
9.  **Federated Zero-Trust Data Mesh**: We implement a unified data mesh using DataHub URNs. The **Central Gateway** verifies Keycloak JWTs via Topaz AuthZ. The **Domain Broker** maps URNs to physical storage and mints temporary STS tickets. The **Cortex Data Client** uses Polars to fetch data natively. The **Cortex Polars IO Manager** forces Dagster to use this same client via M2M OAuth2, achieving 100% code portability between Jupyter and production.
10. **The "Holy Trinity" Pattern**: Our standard for complex integrations involves **dlt** (Extraction), **dbt** (Transformation into a stateful Outbox), and **Restate** (Durable Triggering).

**Full spec for the qualification system**: `docs/RECIPE.md` contains the recipe, the ADRs (why share code not service; why schema discipline replaces package split; FQN+MRO over substring), the regression-test guard rail table, and the implementation-status matrix.
## Component API (Dagster 1.12 GA)
All custom components inherit from `dagster.components.Component`, `dagster.components.resolved.base.Resolvable`, and `dagster.components.resolved.model.Model`. Key rules:
- **No `ComponentSchema`** — schemas are plain `pydantic.BaseModel` or use `Model` directly on the component class.
- **No manual `load()` overrides** — the base class handles YAML attribute resolution automatically via `Resolvable`.
- **`build_defs(self, context)` remains** the primary definition factory method.
- **Normalized Configuration** — components MUST use the Dagster configuration framework to wrap native tool configurations (dlt, dbt, restate), exposing a standardized Dagster-centric schema while handling internal translation to the tool's requirements.
- **Factory pattern for @asset closures** — Dagster 1.12 introspects function parameters as asset inputs; use factory functions to capture closure variables.

## Local Development
Other projects will typically install this repository as a dependency. When developing `dag-tools` locally:
```bash
uv run dagster dev
```
All new common tools should be accompanied by a small "example" usage asset to ensure they load properly in the local UI.
