DAGtor

Overview

DAGtor turns a plain-English request into a runnable SQL pipeline, then supervises it. When a stage fails at runtime the agent diagnoses the root cause against DataHub's metadata graph, repairs the SQL itself, and remembers the fix — so the same failure heals instantly next time. Everything runs against the fiction-retail e-commerce sample (SQLite, ~727k rows), so runs and results are real.

The demo splits into two acts, mirroring the two subsystems:

1. 1Build. Send one message — e.g. Top 5 revenue by segment and country — and DAGtor builds the classic shape: two loads (orders, customers) → join → group → filter. Review it, then Deploy.
2. 2Run, heal & learn. Run the pipeline and watch a stage fail 🔴, get diagnosed 🟡, and heal 🟢 — with the fix saved to the Experience Store and recalled instantly on a re-run.

Architecture

DAGtor is a single Next.js process. The browser holds the chat + React Flow canvas; a custom Node server runs the API routes and a WebSocket bus that streams live state. The orchestrator is the server-side state machine at the centre — it drives the run loop, the node-colour transitions and the heal / approve cycle, calling out to Gemini, the DataHub MCP server, the SQL executor and the Experience Store.

The realtime bus and the runtime API-key store are pinned to globalThis so the Next route bundle and the custom WebSocket server (separate module graphs in the same process) share one instance.

Built on DataHub

The challenge asks entrants to build on the open-source DataHub platform together with at least one of: the MCP Server, Agent Context Kit, DataHub Skills, or the Analytics Agent. DAGtor points at DataHub Core (OSS) through the official MCP server — a config-only swap (see docs/development.md), with a protocol-compatible local server for offline dev — and uses three of the four:

• MCP Server. DAGtor speaks the Model Context Protocol to the official mcp-server-datahub against a live DataHub Core (OSS) graph — or a protocol-compatible bundled server for offline dev. The same list_schema_fields / get_lineage calls either way.
• Agent Context Kit. The agent is grounded in real DataHub context — table schemas, column types, lineage and PII / governance tags — so the builder pre-masks PII and the healer resolves renamed columns and dropped tables from the metadata graph rather than guessing.
• DataHub Skills. The agent is equipped with DataHub's official tool registry (npx skills add datahub-project/datahub-skills), giving Gemini explicit tools to call — get_lineage() and list_schema_fields() — which effectively eliminates hallucination during the healing process.

Data alignment. The execution database is populated with the same DataHub sample datasets that back the metadata graph, so the data the pipeline runs against and the metadata the agent diagnoses against stay perfectly in sync. The Analytics Agent is not used in this build.

Pipeline builder

Every message is routed by intent — build, explain, operate, or off-topic. A build request is grounded in real DataHub metadata (tables, columns, PII tags) before Gemini generates the DAG, and any PII-tagged column is pre-masked with SHA256() in the generated SQL.

The result is a draft pipeline on the canvas with a Deploy card. Deploy hands it to the orchestrator idle and ready to Run — a deliberately separate step, so the fail → heal → learn sequence is watched live rather than flashing past on deploy.

Self-healing & learning

A failed stage walks a fixed ladder: recall a validated fix from the Experience Store if the failure signature matches; otherwise diagnose by introspecting the referenced table's real schema and lineage in DataHub, escalating to Gemini for the harder cases; then propose a repair.

Human-in-the-loop. With Autopilot on, heals auto-apply. Off, the node pauses on 🔵 with a visual - old / + healed SQL diff and an Approve & Apply card; the run resumes on approval. A validated fix is written to the Experience Store keyed by a failure signature, so a recurrence is recalled instantly — no DataHub round-trip.

Node states

Every stage on the canvas is always in one of these states:

Fault injection (demo)

So the self-heal loop always has something to fix, DAGtor can arm deliberate faults on a deployed pipeline. On the canvas, the Inject faults panel lists four scenarios as toggles; the armed set is re-derived from each stage's clean baseline and applied on the next Run, so toggling one off cleanly removes its injection.

Getting started

The agent is MCP-native and works fully offline against a bundled DataHub-compatible server. Setting GOOGLE_API_KEY enables Gemini reasoning; you can also paste a key at runtime in Settings → Model.

cp .env.example .env      # optional; defaults work with no edits
pnpm install
pnpm fetch:dataset        # download the fiction-retail sample (~95 MB)
pnpm db:init              # create the SQLite experience store
pnpm dev                  # Next + WebSocket on http://localhost:3000

Then open the app and use Settings → Demo → Start fresh demo for a clean blank-canvas starting point.

Tech stack

• Frontend: Next.js (App Router, TypeScript) + @xyflow/react for the DAG canvas.
• Backend: Next API routes + a custom Node server with native ws WebSockets for real-time state.
• Agent engine: @google/genai targeting gemini-3.6-flash.
• Metadata: DataHub MCP server (schema fields, lineage, PII tags) over the Model Context Protocol.
• Data + memory: SQLite (better-sqlite3) — real SQL over the sample, plus the Experience Store.