DAGtor

DAGtor builds your SQL data pipelines from a sentence, heals them when they break, diagnosing root cause against DataHub's metadata, and learns from every fix so recurrences repair instantly.

Comment

Inspiration

Data pipelines break for boring, mechanical reasons — an upstream team renames a column, drops a table, or ships a migration on a Friday afternoon — and a human gets paged at 3 a.m. to read a stack trace, hunt down the current schema, rewrite the SQL, and re-run it. The maddening part is that the fix is almost always knowable from metadata: the real column name, the lineage, whether a field is PII. DataHub already holds exactly that. So we asked: what if an agent closed the loop itself — read the metadata, repair the pipeline, and remember the fix so it never has to think about it again? DAG + doctor → DAGtor.

What it does

DAGtor is an autonomous SRE for SQL data pipelines. It does three things:

  • Builds. Describe an analysis in plain English ("top 5 revenue segments by country"). DAGtor grounds the request in real DataHub tables, columns, and PII tags, then generates a runnable DAG — two loads → join → group → filter — pre-masking PII columns with SHA256() without being asked.
  • Heals. It runs the pipeline as real SQL over the DataHub fiction-retail sample (~727k rows). When a stage fails at runtime it intercepts the error (🔴), diagnoses the root cause against DataHub over MCP (list_schema_fields, get_lineage) (🟡), rewrites the SQL, and re-runs (🟢). Three repair strategies: soft (renamed column), governance (PII column → SHA256 masking), and strong (dropped table → reroute via lineage). Cases with no deterministic fix — a dropped table, a join that lost its ON predicate — escalate to Gemini with evidence gathered from DataHub, instead of guessing.
  • Learns. Every validated fix is written to an Experience Store, keyed by a failure signature. The next time the same failure appears, the fix is recalled instantly — no metadata round-trip, no model call.

A human-in-the-loop switch (Autopilot) lets fixes auto-apply, or pause on a before/after SQL diff for Approve & Apply.

How we built it

  • Frontend: Next.js (App Router, TypeScript) with @xyflow/react for the live DAG canvas, plus a marketing landing page and a docs page with dependency-free architecture diagrams.
  • Realtime: a custom Node server with native ws WebSockets streams node state from a server-side orchestrator — a globalThis-pinned state machine that drives the run loop and the fail → diagnose → heal → learn cycle.
  • Agent: @google/genai (Gemini) for intent routing, pipeline generation, and the function-calling diagnosis / escalation loop over the DataHub tools.
  • Metadata: DataHub over MCP (stdio) — the official mcp-server-datahub, with a protocol-compatible local server for zero-setup offline dev; the agent calls the DataHub Skills tools list_schema_fields and get_lineage.
  • Execution + memory: one database with grouped tables (the source dataset and the experience table) behind a driver — SQLite (better-sqlite3) locally, PostgreSQL (pg) for a remote deploy — with a SHA256() shim (pgcrypto) so the agent's generated SQL runs unchanged on either backend.
  • Deploy: Docker Compose (app + Postgres) and a Compute Engine runbook; bring-your-own Gemini key at runtime (Settings → Model), so no secret ships in the repo.

Challenges we ran into

  • Making healed SQL actually valid. The deterministic healer parses the missing column from the DB error, resolves the intended column from the real schema, and rewrites safely — including wrapping PII in SHA256() — so the re-run genuinely succeeds instead of failing differently.
  • Cross-backend SQL. SQLite has no SHA256; we register a UDF locally and a pgcrypto function on Postgres so the same generated SQL is portable.
  • A stateful, WebSocket-driven design. The in-memory orchestrator is a single process by nature, which shaped the deployment choice (a small VM over stateless autoscaling) and the BYOK key handling.
  • Guardrails over crashes. A join that loses its predicate would ask the database for billions of rows; a row-explosion guardrail trips and escalates rather than hanging the server.
  • Making healing visible. We surfaced the loop as inline chat heal-cards, a dedicated Healing view, and a "Memory" HUD, so the diagnosis and learning are legible without opening drawers.

Accomplishments that we're proud of

  • A genuine closed loop — build → run → fail → diagnose against DataHub → repair → learn → instant recall — all on real data, not mocks.
  • Governance-aware healing: a PII column stays masked through a repair, so a fix never quietly leaks emails downstream.
  • One agent that both generates metadata-grounded pipeline code and operates and repairs it — the build and the SRE halves in a single tool.
  • A zero-setup local demo (bundled MCP + SQLite) that swaps to real DataHub Core + Postgres with config only.

What we learned

  • Grounding the model in metadata (schemas, lineage, PII) turns "plausible SQL" into "correct SQL" and eliminates a whole class of hallucination.
  • The most valuable agent behavior isn't a clever one-shot fix — it's accumulating operational knowledge so the system gets faster and cheaper every time it sees a failure again.
  • Knowing when to escalate — solve deterministically vs. hand the model structured evidence — matters as much as the fix itself.

What's next for DAGtor

  • Persist rejected-proposal feedback into the Experience Store (learn from "no", not just "yes").
  • Broader heal strategies: type coercions, unit mismatches, partial-lineage reroutes.
  • Write-back to DataHub — annotate the datasets and lineage the agent touched, so the next person or agent inherits what it learned.
  • Scheduled, multi-pipeline monitoring and a Postgres-native execution path for larger datasets.

Built With

Share this project:

Updates