UPSTREAM
A desired-state control plane for DataHub.
You declare the estate you want. Upstream measures the gap, explains it with cited evidence, prices every proposed change, waits for a human decision, and writes the ratified result back into DataHub.
Apache-2.0 · Python 3.11 · DataHub v1.7.0 OSS
Contents
- The problem
- The idea in one paragraph
- System architecture
- The specification
- The read path — eight Skills, never a silent fallback
- The conformance dial
- Controller one — coverage
- Controller two — reliability
- The arithmetic: confidence and risk
- Simulation — pricing a change before proposing it
- Write-back, and immunization
- Memory — why the second time is cheaper
- The operator console
- The API: read, mutate, stream
- Evidence and evaluation
- Safety model
- What is verified, and what is not
- Engineering notes
- Contributed back to DataHub
- Running it
1. The problem
Two failures show up in every data organisation, and neither is fixed by more tooling.
Nobody can say what "good" looks like. Which tables need an owner? Which need a freshness check, a glossary term, a documented grain? Every team has opinions; almost none have those opinions written down anywhere a machine can read. So catalogs fill with assets nobody can vouch for, governance becomes a quarterly spreadsheet exercise, and the honest answer to "is our metadata healthy?" is a shrug.
Sometimes a pipeline breaks and nothing fails. A vendor renames a column in a feed you do
not control. The staging model still runs. The cast still succeeds — it just emits NULL now.
Three transformations later a revenue mart is wrong, every job in the DAG is green, and you
find out from the person who trusted the number. This class of failure is invisible to job
monitoring by construction, because no job failed.
DataHub is an excellent answer to what is my estate. Neither problem is a storage problem. They are both decision problems sitting on top of storage.
2. The idea in one paragraph
DataHub is the system of record. Upstream is the system of intelligence.
Upstream is a reconciliation loop over DataHub, in the Kubernetes sense of the word: a declared desired state, a controller that continuously observes actual state, and a bounded actuator that closes the difference. It stores nothing of its own — no catalog table, no lineage table, no entity registry. Every fact it reasons over is read from DataHub through DataHub's own retrieval Skills, and every conclusion it reaches is written back into DataHub as a first-class entity: an incident, a runbook, a structured property, an assertion.
Delete the Upstream container and nothing is lost. The learning is in the graph.
3. System architecture
3.1 The layering
┌──────────────────────────────────────────────────────────────────────────┐
│ estate.yaml the desired state, declared │
│ tiers · requirements · policies 31 lines of configuration │
└───────────────────────────┬──────────────────────────────────────────────┘
│ selectors resolve through DataHub search
┌───────────────────────────▼──────────────────────────────────────────────┐
│ READ PATH — 8 context slices over 5 DataHub tools │
│ search · lineage · assertions · quality · governance │
│ glossary · documentation · ownership │
│ every slice reports its transport; a fallback is logged, never silent │
└───────────────────────────┬──────────────────────────────────────────────┘
│ MetadataContext — one object, no cache of record
┌───────────────────────────▼──────────────────────────────────────────────┐
│ CONFORMANCE — one pure function per requirement, scored per check │
│ no model participates · deterministic · the same number everywhere │
└───────────────────────────┬──────────────────────────────────────────────┘
│ violations, ranked by risk
┌───────────────────────────▼──────────────────────────────────────────────┐
│ CONTROLLERS (LangGraph) │
│ │
│ coverage observe → plan → ratify → actuate → verify │
│ reliability triage → lineage → correlate → forensics → critic │
│ → simulate → [gate] → remediate → verify → scribe │
└───────────────────────────┬──────────────────────────────────────────────┘
│ typed, cited, priced proposals
┌───────────────────────────▼──────────────────────────────────────────────┐
│ THE HUMAN GATE LangGraph interrupt() + checkpointer │
│ console or CLI nothing is written before a person says yes │
└───────────────────────────┬──────────────────────────────────────────────┘
│ ratified writes only
┌───────────────────────────▼──────────────────────────────────────────────┐
│ WRITE BACK INTO DATAHUB │
│ incidents · runbooks · assertions · structured properties · tags │
│ every write logged to a ledger that `reset` can undo │
└──────────────────────────────────────────────────────────────────────────┘
Read the arrows literally. Metadata flows up into intelligence; conclusions flow back down into the system of record. Upstream is a client of DataHub in both directions.
3.2 Division of responsibility
| DataHub owns | Upstream owns |
|---|---|
| metadata storage, the entity graph | reasoning, reconciliation |
| lineage, search, the Timeline API | memory, simulation, risk |
| assertions, incidents, properties | proposals, orchestration, the human gate |
| ingestion | nothing that is stored |
Three consequences, each enforced in code rather than asserted in prose:
- No metadata store. If a fact is not in DataHub, Upstream does not know it. The fix is to write it to DataHub, not to cache it locally.
- No ingestion. Where this repository runs
datahub ingest, it is running DataHub's. - Every conclusion is a DataHub write. RCAs become incidents. Runbooks become institutional memory links. Learned checks become assertions. Risk becomes structured properties.
3.3 Repository shape
estate.yaml the desired state
upstream/
├ spec/ loader; selectors resolve through DataHub search
├ datahub/ the read path (8 Skills) and one writer per aspect
├ agents/ triage · lineage · correlator · forensics · critic
│ simulator · remediator · verifier · scribe
├ controllers/ the two control loops, the shared write dispatch,
│ and the ratify controller behind the console
├ conformance/ the dial — one pure function per requirement
├ console/ the read model: one projector per screen + the receipt journal
└ api/ FastAPI: read, mutate, stream. Orchestration only
console/ the operator console (Vite + React 18 + TypeScript)
contrib/ a Skill contributed back to DataHub
spike/ the verified DataHub surface, and the document recording it
demo-stack/ sample-pack loaders, DuckDB mirror, chaos + governance decay
evals/ the ten-case suite, its runner, and committed run records
examples/ generated reports and LLM cassettes, committed as evidence
tests/ 430 tests; fixtures pinned from live responses
~11,900 lines of Python, ~6,900 of TypeScript, ~5,000 of tests.
4. The specification: estate.yaml
The entire configuration surface is one file. Tiers of assets, and the requirements each tier must meet.
tiers:
- name: tier-1
match:
- domains: ["Data Platform Team"]
- name_patterns: ["fct_*", "dim_*"]
platforms: ["dbt"]
requirements:
require_owner: true
require_glossary_term: true
min_assertions: 1
freshness_sla_hours: 24
max_open_incident_hours: 4
policies:
actuation: ratify # dry-run | ratify | auto — never silent
max_proposals_per_run: 25
reliability:
min_confidence_to_remediate: 0.75
immunize_on_resolve: true
Two design decisions matter here.
Selectors resolve through DataHub search, not through a list of URNs. The spec follows the
estate as it grows. Add a fct_* model tomorrow and it is in scope tonight, with no
configuration change.
actuation is a policy, not a flag in code. dry-run prices and writes nothing. ratify
pauses at a real gate. auto is available and logged. The controller reads the policy off the
record rather than off the request, so a caller cannot talk the system past a dry-run estate
by asserting it is not in one.
5. The read path — eight Skills, never a silent fallback
Every fact reaches the agent through DataHub's own retrieval surface. Eight context slices are projected from five distinct tools:
| Slice | Tool | Consumer |
|---|---|---|
| asset-search | search |
conformance, triage |
| lineage-retrieval | get_lineage |
lineage walker, blast radius |
| assertion-retrieval | get_dataset_assertions |
conformance, verifier |
| quality-retrieval | list_schema_fields |
forensics, proposer |
| governance-retrieval | get_entities → structuredProperties, health |
memory, risk |
| glossary-retrieval | get_entities → glossaryTerms |
conformance, impact |
| documentation-retrieval | get_entities → description |
conformance, proposer |
| ownership-retrieval | get_entities → ownership |
conformance, proposer |
Ordering is the argument. Eight skill invocations appear in the trace before the first line of reasoning. An agent that reasons before it perceives is the failure this design exists to prevent.
Where a tool cannot answer, the SDK floor answers and the trace says so. Two slices fall back on v1.7.0 — documentation and assertions — for reasons documented in §17. The fallback returns the same payload shape and is logged on every invocation. The claim "every read goes through DataHub" stays literally true; the claim "every read goes through an agent tool" is not made anywhere.
The floor is a tested path, not a documented intention: forcing the agent transport to fail on every call reproduces the identical conformance score.
6. The conformance dial
One number for the distance between the spec and the graph.
tier-1 16 assets 128 checks 75 passed 59%
tier-2 2 assets 10 checks 4 passed 40%
raw 8 assets 24 checks 2 passed 8%
overall 26 assets 162 checks 81 passed 50%
Scored per check, not per asset, so the number moves smoothly as gaps close rather than jumping when an asset crosses a threshold. One pure function per requirement — no model participates in scoring, which is what makes it the same number on every machine.
The raw tier scoring 8% is not a scoring artefact. It is the vendor feed boundary carrying no schema contract, no owner and no test — which is precisely why the drift scenario in §8 propagates silently. The dial and the incident are the same story told twice.
7. Controller one — coverage: closing the declared gap
observe ──> plan ──> ratify ──> actuate ──> verify
│ │ │ │ │
50% 45 priced human 45 writes 83%
proposals gate
observe — score the estate, and fetch lineage and schema only for assets that have a gap.
plan — violations become proposals. Every proposal is grounded: an owner comes from git history, the lineage graph, a sibling asset or the domain — never from a guess. A description is written from the columns and upstreams the read path already returned. A check is a deterministic template. Then every proposal is priced by the simulator, and proposals with no simulated benefit are dropped rather than shown.
ratify — a real pause. LangGraph interrupt() with a checkpointer and a stable thread id.
Get either wrong and the gate silently becomes a no-op that still prints as though it ran;
there is a test that fails if that ever happens.
actuate — the approved writes, each logged to a ledger. One failing write never takes the run down with it.
verify — re-observe. A controller that does not re-read cannot claim it changed anything.
Measured result: 50% → 83%. Projected +32.7%, measured +32.7%, across 53 checks. The what-if and the what-is run through the same scoring code over the same context object, so the number on the approval screen is the number you get. A disagreement there is a bug, not a rounding difference.
28 violations survive on purpose. Freshness requirements need an evaluation, not a declaration. Some assets have no owner anywhere in the graph to infer from. Others ask for a judgement — a business term, a retention period — that belongs to a human. Each is listed with its reason. An agent that filled those in would be guessing into a catalog people trust.
8. Controller two — reliability: diagnosing a silent failure
The scenario: a vendor renames discount_amt → discount_amount. A staging macro reads the
old name, emits NULL of the right type, and the mart's discount column goes 100% empty. No
job fails. The only signal anywhere is one quality assertion at the very end of the chain.
triage ──> lineage ──> correlate ──> forensics ──> critic ──┬──> report ──> simulate
│ │ │ │ │ └──> abstain
8 skills 6 hops Timeline + bounded confidence
+ memory upstream run history read-only from 5 signals
+ risk MODIFY, not SQL
REMOVE+ADD
correlate reads DataHub's Timeline API. The critical platform detail: a column rename
arrives as a single MODIFY event, not REMOVE + ADD, and the field is operation
rather than changeType. A correlator that waits for a removal misses exactly the failure this
project is built around.
forensics runs bounded, read-only SQL against the warehouse mirror to confirm or refute the
hypothesis — SELECT only, LIMIT injected, with a per-run query budget. A refused query keeps
its statement and its error in the evidence ledger rather than disappearing.
critic scores what survived. A hypothesis citing no evidence is dropped in code, not discouraged in a prompt.
Every claim in the final write-up carries indices into an evidence ledger. The root-cause analysis is the ledger, rendered — not prose the model was asked to justify afterwards.
9. The arithmetic: confidence and risk
Both numbers are computed and decomposable. Neither is ever reported by a model.
Confidence — five orthogonal signals
| Signal | Weight | What it measures |
|---|---|---|
| evidence | 0.60 | cited support × temporal fit × diversity |
| memory | 0.10 | a recorded conclusion on this subgraph |
| timeline | 0.10 | the change precedes the symptom |
| forensic | 0.10 | a query independently confirms it |
| simulation | 0.10 | the fix removes the symptom |
A hypothesis with no hard artifact — no timeline event, no query result, no run record — is
multiplied by 0.6, because a story without an artifact is a story. Below
min_confidence_to_remediate (0.75) the run abstains and names the one probe that would
settle it.
Risk — eight independent graph facts
risk = 0.20 · business_criticality tier · domain · exec-facing consumers
+ 0.15 · historical_fragility decay-weighted incident history
+ 0.15 · assertion_health_gap 1 − passing / required
+ 0.10 · lineage_depth normalised hops × fan-out
+ 0.10 · governance_violations open estate.yaml violations on the asset
+ 0.10 · incident_similarity resemblance to past incidents here
+ 0.10 · business_impact glossary terms, revenue domains, dashboards
+ 0.10 · (1 − confidence) the controller's own uncertainty is a risk
The eighth term is the one people miss: acting on a shaky conclusion is riskier than acting on a
solid one, so confidence feeds risk, and risk gates actuation. risk.explain() prints the
arithmetic beside the priority in every trace.
10. Simulation — pricing a change before proposing it
Nothing reaches a human unpriced. Before a proposal is shown, the simulator computes:
- conformance delta — how far the dial moves, through the same checks as the live dial
- blast radius — downstream assets, and how many are tier-1
- MTTR reduction — how much earlier a boundary check would have caught this
- checks fixed / broken
Conformance is not additive, and the system refuses to pretend otherwise. Proposals share requirements, so summing per-proposal deltas double-counts. A selection is priced only when it matches a set the simulator actually ran; anything else is reported as unpriced, with the reason. This is the single most-tempting place to fabricate a number, and it is where the design is most deliberate.
11. Write-back, and immunization
Diagnosis is where most agents stop. Four artifacts land in DataHub, and they are what make the next run cheaper:
| Artifact | Where it goes | Who reads it |
|---|---|---|
| Incident + RCA | the symptom asset — resolved only if verified | a human, in the UI |
| Runbook | the cause asset, where the next engineer looks | a human |
fragilityScore · lastRootCause · mttrMinutes · knownFailureModes |
both assets | the next run |
| Immunization assertions | the ingestion boundary | DataHub, continuously |
The last row is the point. On resolving a vendor-drift incident, the system registers a schema contract at the boundary and a null guard on the column that emptied — one hop from the vendor rather than three transformations downstream at the mart, where the symptom appeared. The next occurrence of this failure fails the build that introduces it.
The citations in the RCA are written by code, not by the model. And the fix is verified before it is believed: the transformation project and warehouse are copied to a temporary directory, the patch is applied there, the models are rebuilt and the failing check re-evaluated on the copy. A fix nobody confirmed leaves the incident open.
The patch itself is constructed, not generated. A model that writes SQL from a description
will occasionally write plausible SQL that drops a column, and no reviewer catches that at 2am.
The remediator reads the live schema, finds the transformation still naming the removed column,
and rewrites it to prefer the new name and fall back to the old — a rule applied to text the
code has already located and verified. The model writes the pull-request prose, which is the
part where being wrong is cheap. Nothing is applied: the diff is written to examples/ and
checked with git apply --check, because a branch is a human's decision.
12. Memory — why the second time is cheaper
Re-break the same scenario and run the same command:
| cold | warm | |
|---|---|---|
| plan | walk lineage → correlate → forensics → critic | confirm the remembered cause → forensics → critic |
| tool calls | 48 | 8 |
| forensic queries | 2 | 1 |
| evidence collected | 18 | 11 |
Recognition comes from the structured properties the system wrote for itself to read — not from prose similarity, which cannot work here and it is worth saying why. A symptom line is a dozen words; a resolved incident's RCA is three hundred. A set-similarity score with a union denominator therefore reports a perfect topical match at 0.007, indistinguishable from noise. That bug was invisible until there was something to recall.
A warm start reorders the investigation; it does not skip the evidence or the critic. If the memory is wrong, the confidence does not materialise and the run abstains.
13. The operator console
Five screens over a live HTTP API, served by the same projectors that generate the committed artifacts — so the live payload and the recorded one cannot drift.
| Screen | Answers |
|---|---|
| Mission Control | What is broken, and what is waiting on a person |
| Investigation Detail | Why the system believes it — the evidence ledger, cited |
| Simulation Workspace | Which remediation to approve, and what each is worth |
| Approval Center | Every controller's pending work, in one queue |
| Proposal Review | Exactly what one approval writes, aspect by aspect |
The frontend computes nothing. Filtering, ordering, paging, counting, pricing, ranking and even the empty-state copy are backend projections. React renders what it is handed. A payload the projector cannot fill arrives carrying the reason it is empty rather than a zero — "not priced", "the record kept no urn", "the run never reached this stage".
Two controllers put work into one queue, because a person has one attention, not one per controller.
14. The API: read, mutate, stream
read route → dependency → projector → response
write route → dependency → controller → writer → receipt → projector → response
live GET /api/events — the stored receipt, streamed verbatim
Every route is one statement. Run resolution, the GMS probe and error translation live in one dependency module, so six read endpoints share one answer to which run, is DataHub up, and this record cannot be projected.
The receipt is the event
Every mutation answers with a typed MutationReceipt: what was written, to which aspect,
and the URN DataHub returned — or the reason nothing was. That same object is the line appended
to the journal and the frame GET /api/events broadcasts, byte for byte. One schema for the
write path and the live path, with no translation between them; the stream replays stored bytes
rather than re-rendering them, so history cannot change shape after a deploy.
Resume is the stored sequence number, which a browser returns as Last-Event-ID on reconnect,
so a dropped connection replays exactly what it missed, once.
Approving is not the same as writing, and the receipt distinguishes them. A proposal writes
if it carries a drafted payload and policy allows; otherwise the decision is recorded with the
reason — a draft pull request Upstream never merges, a row from a report that kept no payload,
a dry-run estate, or a proposal already decided.
A half-written batch answers 200 partial, never a 500. DataHub has no transaction across
aspects, so a batch is not atomic. Discarding the writes that landed is the one outcome an
operator cannot recover from.
A receipt names the screens it invalidated; each store re-asks its projector rather than patching state by hand.
15. Evidence and evaluation
Ten seeded cases across five failure scenarios, scored against committed run records.
make eval # inject every scenario and run the loop, live
make eval-replay # re-score the records — no API key, no network, no DataHub
| Scenario | Cases | Root cause @1 | Correct abstention | Median wall-clock |
|---|---|---|---|---|
| vendor schema drift | 4 | 1/1 | 3/3 | 109s |
| silent skipped job | 2 | 0/1 | 1/1 | 107s |
| boundary assertion failure | 2 | 1/2 | n/a | 64s |
| unrelated noise (no cause) | 1 | n/a | 1/1 | 160s |
| recurrence (warm memory) | 1 | 1/1 | n/a | 71s |
| overall | 10 | 3/5 | 5/5 | 95s |
Confidence separates right answers from wrong ones by 0.256 — mean 0.850 on correct conclusions against 0.594 on everything else. That row decides whether the five-signal model is load-bearing or decorative, which is why it is published rather than described.
Both misses are explained, and neither is a wrong answer — the system abstained both times. One is a gap in the demo estate: the freshness injector ages rows in the warehouse but publishes no staleness signal to DataHub, so there is nothing for the read path to perceive. The other is the confidence model working as specified: a real failing assertion with no timeline evidence behind it reaches 0.60 on a single hard artifact and declines, because requiring several independent signals to move together is what a 0.85 is meant to mean.
Every artifact under examples/ came from a live run. make eval-replay needs no key, no
network and no DataHub — anyone can reproduce the grid from a clone.
16. Safety model
| Boundary | Mechanism |
|---|---|
| No silent writes | actuation: ratify policy; LangGraph interrupt() with a checkpointer |
| No unbounded SQL | SELECT only, LIMIT injected, read-only connection, per-run query budget |
| No applied patches | diffs written to examples/, checked with git apply --check; a branch is a human's decision |
| No unverified resolution | the fix is rebuilt in a sandbox copy; unverified means the incident stays open |
| No invented conclusions | uncited hypotheses dropped in code; below-threshold runs abstain and name the next probe |
| No irreversible action | every write logged to a ledger; reset --yes undoes all of them |
| No double-write | decisions are journalled; a repeat approval is refused, naming the receipt that already answered |
| Bounded autonomy | per-run budgets on tool calls, queries and model calls, printed in the trace |
17. What is verified, and what is not
Every DataHub surface this system depends on was executed against a live OSS quickstart and
read back. make spike reports PASS only for writes it could confirm afterwards; anything
unconfirmed returns PARTIAL or FAIL. That distinction is not pedantry — a malformed
assertion write passed silently here until read-back was added.
| Surface | Result |
|---|---|
Incidents (raiseIncident / updateIncidentStatus) |
OK |
| Structured properties (definition + read-modify-write) | OK |
External assertions (AssertionInfo + AssertionRunEvent) |
FIXED — see below |
Documentation (editableDatasetProperties + institutional memory) |
OK |
| Tags | exercised |
| Ownership | OK |
| Timeline API (read) | OK |
| Skills read path (5 tools → 8 slices) | OK |
| MCP server | unavailable on this version |
Not claimed anywhere in this project: that a skill executed over both ACK and MCP
transports — datahub-agent-context 1.7.0 ships the tool implementations but no server
entrypoint, so the transport layer is written for it and reports the resolution honestly rather
than claiming a round trip that did not happen. Scale beyond ~30 assets is also unproven; the
read path is per-asset and would need batching.
Full detail, including every deviation from the documentation, is in
spike/DATAHUB-USAGE.md, and everything this project leans on is in
LIMITATIONS.md.
18. Engineering notes
430 tests, ruff clean, TypeScript strict with no any. The tests worth mentioning are the
ones that assert absence: that a projector emits an empty state with its reason rather than a
zero, that the gate is a real pause, that a rejected write leaves the estate unchanged, that a
recorded payload never absorbs decisions made on the developer's machine.
The frontend contract is the schema. API tests parse the console's own TypeScript interfaces and assert that every response satisfies them — required fields present, no undeclared fields. Restating the payloads as Pydantic models would have produced two specifications that drift.
The API response is asserted equal to the projector's own output, so the live payload and the committed recording cannot diverge.
Determinism where it matters. Conformance, risk, simulation and ranking are pure functions.
Model calls are recorded to cassettes; LLM_MODE=replay reproduces a run with no key and no
network. Reports carry no timestamps, so a committed run diffs on what changed in the estate
rather than on when the command was typed.
Honest failure everywhere. A record that never reached a stage produces a 503 naming the stage. An unknown proposal is a 404 listing what the batch actually holds. A filter the queue does not offer is a 400 naming the ones it does — because a silently dropped filter looks exactly like a filter that matched everything.
19. Contributed back to DataHub
reconcile-estate, packaged as a DataHub Skill any agent can call: give it a spec, get back the
ranked, remediable gap. It reads and changes nothing. The conformance scorer that powers the
dial is useful outside this project, and shipping it as a Skill rather than an endpoint is the
form DataHub's own agent surface expects.
20. Running it
cp .env.example .env # a DataHub token and one model key
pip install -e ".[dbt]" # Python 3.11
make up # DataHub OSS quickstart, version-pinned
make seed # DataHub's own sample estate, mirrored, then decayed
make score # the gap between estate.yaml and the graph
make reconcile # close it, behind a human gate
make break scenario=vendor-drift # a silent failure
make investigate urn="urn:li:dataset:(urn:li:dataPlatform:dbt,demo.main.fct_revenue_daily,PROD)"
Prefer a screen? make api and make console-live put all of it behind five screens at
localhost:5273. Want neither? make eval-replay reproduces the evaluation grid with no key,
no network and no DataHub.
The demo estate is DataHub's own sample data — showcase-ecommerce and bootstrap — loaded,
mirrored into DuckDB for forensic SQL, then broken on purpose by a deterministic scenario
library. A judge gets the same estate every number here was measured against. Reproducibility is
a property of the design, not of luck.
Every other agent reads your catalog. Upstream leaves it healthier than it found it.
Apache-2.0
Built With
- datahub
- fastapi
- langgraph
- python
- react
- skills
- sse
- typescript
Log in or sign up for Devpost to join the conversation.