Cert Sync

A certification and cross-surface sync agent for DataHub.

Built for the DataHub Agent Hackathon — challenge category: Agents That Do Real Work.

🎥 Demo video: https://youtu.be/8lgGDGlq6lc

For judges

This submission is a repo + setup instructions, not a hosted demo — no live URL required to evaluate it. Two ways to see it work:

1. Watch the demo video above — it walks the full detect → route → certify → write-back loop end to end, including a live Slack API pull.
2. Run it yourself — follow Setup below. You'll need your own free OpenRouter key (no sponsor LLM credits on this hackathon); if you'd rather not spend it, examples/ has a full committed run's output (all 5 scenarios, detect/route/certify/before-after) so you can inspect real results without running anything.

---

The problem

Metadata disagreements happen everywhere and nobody notices:

• A Slack thread says billing_amount means "net of adjustments."
• The dbt model's own SQL comment says it means "gross charge, before adjustments."
• Every downstream dashboard and agent just... picks one. Silently.

Nobody decided. Nobody was asked. The next AI agent that answers a question about that term is confidently wrong, and nothing in DataHub reflects the disagreement.

What Cert Sync does

Cert Sync watches real sources (Slack threads, docs) for a term, compares what they claim against the definition already registered in DataHub, and when it finds a genuine conflict:

1. Detects the drift and extracts what each source is actually claiming (LLM-based extraction, not string matching).
2. Resolves the real owner of the term from DataHub's own ownership metadata — routes to a human, never auto-decides.
3. Once a human certifies the correct definition, writes it back into DataHub as the system of record, with a full audit trail of where it came from.
4. Proves the fix reaches everyone downstream — a completely separate consumer agent, with no shared code or session, gives the wrong answer before certification and the right answer after, just by reading DataHub again.

It also runs as a standing service: a worker proactively scans for drift on a schedule, every detection/certification is persisted to a queryable audit log, and a dashboard shows the whole thing live.

Architecture

flowchart LR
    subgraph DH["DataHub (docker quickstart)"]
        GMS["GMS :8081"]
        UI["Frontend UI :9002"]
    end

    subgraph CS["Cert Sync (docker compose)"]
        API["api — FastAPI\n1 long-lived cert MCP session (mutation-enabled)\n+ fresh consumer MCP session per /ask"]
        Worker["worker\nscheduler.py"]
        Dashboard["dashboard\nStreamlit + api_client.py"]
    end

    Slack["Slack Web API"]
    SQLite[("SQLite\n(named volume)")]

    Worker -->|"POST /scan\nevery N min"| API
    Dashboard -->|"HTTP (httpx)"| API
    API -->|"MCP (stdio)"| GMS
    API --> Slack
    API --> SQLite
    GMS -.-> UI

Loading

Only the api service ever opens an MCP session or touches SQLite. The worker and dashboard only ever speak HTTP to api. eval/run_eval.py and the CLI pipeline (cert_agent/pipeline.py) are separate offline tools that never go through the live API — they're used for measuring detection precision/recall against the static scenario fixtures.

Built on the DataHub MCP Server (mcp-server-datahub, self-hosted, TOOLS_IS_MUTATION_ENABLED=true for the certifying agent, read-only for the consumer agent) — the two agents never share a session, which is what makes the "does the fix actually propagate" proof meaningful.

Results

Across all 5 scenarios (including one where the term has no registered owner at all), detection runs at 100% precision and recall. See examples/sample_eval_run/summary.json for a committed run's raw output, or Page B of the dashboard for a live view after running eval/run_eval.py yourself.

Prerequisites

• Docker + Docker Compose
• Python 3.11+ and uv (provides uvx, used to launch mcp-server-datahub)
• A free OpenRouter API key (used for drift detection and Q&A — see Notes on LLM usage below):
  1. Sign up at openrouter.ai (free, no card required).
  2. Go to openrouter.ai/keys → Create Key.
  3. Paste it into .env as OPENROUTER_API_KEY (see Configure environment below).
• A Slack workspace where you can install a bot (free tier is fine) — only needed if you want to see the live Slack ingestion for billing_amount; every other scenario works from static fixtures with no Slack setup at all

Setup

1. Start DataHub and load the demo dataset

datahub docker quickstart
# generate setup/healthcare.db locally (gitignored - not committed) - see
# setup/README.md for the source CSVs and full generation steps
python setup/create_db.py /path/to/healthcare-dataset-csvs
datahub ingest -c setup/ingest.yaml
python setup/add_lineage.py
python setup/add_metadata.py

See setup/README.md for full details on the dataset (a forked healthcare pipeline built from the Kaggle Healthcare Dataset, CC0 licensed, synthetic data only).

2. Configure environment

cp .env.example .env

Fill in DATAHUB_GMS_URL (defaults to http://localhost:8081) and OPENROUTER_API_KEY. SLACK_BOT_TOKEN is optional — leave it blank to skip Slack setup; detection then falls back to the static source_a_*.md fixture for every scenario, including billing_amount.

3. (Optional) Set up the live Slack integration

Only scenario_01_billing_amount is wired to Slack. To see it live instead of from a fixture:

1. Create a Slack workspace if you don't have one: slack.com/get-started.
2. Go to api.slack.com/apps → Create New App → From scratch → name it (e.g. certsync-demo-bot).
3. OAuth & Permissions → add Bot Token Scopes: channels:history, channels:read, chat:write, users:read.
4. Install to Workspace, then copy the Bot User OAuth Token (xoxb-...) into .env as SLACK_BOT_TOKEN.
5. Create a public channel (e.g. #finance-reporting-demo) and /invite @certsync-demo-bot.
6. Get the channel ID from the channel details panel (starts with C) and set it in slack_channels.yaml:

   billing_amount:
     channel_id: "C0XXXXXXX"
     channel_name: "finance-reporting-demo"

7. Seed the channel with the demo thread once:

   python setup/seed_slack_channel.py

4. Run the stack

docker compose up --build

This brings up:

• api — http://localhost:8010 (FastAPI, holds the MCP sessions, healthcheck at /health)
• worker — proactively hits /scan every SCAN_INTERVAL_SECONDS (default 1800s / 30 min — deliberately conservative, see Notes on LLM usage)
• dashboard — http://localhost:8501 (Streamlit UI)

Open http://localhost:8501 and walk through "Live Walkthrough" for scenario_01_billing_amount.

Verifying the Slack integration is actually live

The Source A panel shows a LIVE via Slack API badge and the exact fetch timestamp. To prove it's a real API pull and not a cached fixture:

1. Post any message into the seeded Slack channel from another client (Slack app, curl against chat.postMessage, or python demo/video_pipeline/post_live_message.py <channel_id> "<text>").
2. Click 🔄 Refresh from Slack in the dashboard.
3. The new message appears and the fetch timestamp updates — with no dashboard restart, no redeploy, no code change.

Examples

examples/sample_eval_run/summary.json is a real, committed run across all 5 scenarios — full drift detection, resolved owner, certified before/after definitions, and consumer-agent answers before and after certification. See examples/README.md for how to read it. Useful if you want to see actual output without running the stack or spending LLM quota.

Project structure

cert_agent/          Detection, owner resolution, certification (mutation-enabled MCP session)
consumer_agent/      Read-only Q&A agent, proves the fix propagates downstream
integrations/        Slack Web API client
persistence/         SQLite audit-log layer
services/            Shared business logic used by the API (detect/certify/ask/scan)
api/                 FastAPI service — the only process that opens MCP sessions
worker/              Proactive drift-scan scheduler (HTTP-only, no MCP/business logic)
demo/                Streamlit dashboard + video-production tooling (demo/video_pipeline/)
eval/                Offline precision/recall evaluation against scenario fixtures
scenarios/           5 scenario fixtures (sources + expected outcomes)
setup/               DataHub healthcare dataset setup (pipeline, lineage, metadata, Slack seeding)

Notes on LLM usage

Drift detection and Q&A call an LLM via OpenRouter — there's no sponsor LLM credit for this hackathon, so you'll need your own free OpenRouter key (see Prerequisites). Free-tier OpenRouter keys cap at 50 requests/day account-wide. Two things keep that from being a bottleneck:

• The worker's proactive scan defaults to every 30 minutes (SCAN_INTERVAL_SECONDS=1800), not 5 — each scan is one LLM call per scenario (5 total), so a shorter interval left running unattended would exhaust a fresh free-tier key before you get to click anything in the dashboard.
• examples/ has a full sample run committed, so you can see real output without spending any quota at all.

If you do run out mid-session, cert_agent/drift_detector.py also supports a GEMINI_API_KEY fallback, or point OPENROUTER_API_KEY at a paid tier.

Pre-existing code disclosure

The healthcare dataset (setup/) is derived from the Kaggle Healthcare Dataset (CC0 1.0, public domain, synthetic records) — column normalization, planted data-quality issues, pipeline staging, and DataHub metadata/lineage on top of it were built for this project. All agent, service, API, worker, and dashboard code was written for this hackathon submission.

License

Apache 2.0 — see LICENSE.