Conductor Data Pipeline.
A governed data pipeline that extracts, transforms, masks, and loads data under JSON-schema contracts — with consistent tokenization, PROV-AGENT lineage, post-mask residual-PII scanning, and blocking human-approval gates.
Install
Transcribed from the repository README; not yet executed from a clean environment:
git clone https://github.com/bulletproofsoftware-ai/bulletproof-conductor-data-pipeline.git cd bulletproof-conductor-data-pipeline python3.12 -m venv .venv && source .venv/bin/activate pip install -r requirements.txt python -m pytest # 891 tests, no external services required
bulletproof-conductor-data-pipeline/docs/media/1Problem Statement
Moving data through a pipeline is easy; proving afterwards what moved, how it was masked, and who authorised it is not. When the data contains PII, that gap becomes a regulatory liability — there is no record connecting an extraction to the contract that permitted it or the person who approved it.
Masking is usually applied per-column and per-job, which quietly breaks analysis. The same customer name masked independently in a structured column and in a free-text note produces two unrelated tokens, so joins fail and the masked dataset loses its utility.
Worst of all, masking is rarely verified. A policy is configured, the job reports success, and nobody rescans the output to confirm no personal data survived. This pipeline treats each of those as an enforced gate rather than a convention: contracts are signed by a data steward, masking is validated after the fact, and sensitive operations block on human approval.
2Architecture
The pipeline exposes no HTTP API. As the compose file states, the MCP tool layer is the only interface for agents — eight tools registered in tools/tool_registry.py, each carrying a governance classification that determines what gating applies.
Classified tool layer
Eight MCP tools (data_connect, data_extract, data_transform, data_mask, data_load, data_profile, data_contract_validate, data_lineage_query) are registered with CLASSIFICATION_STANDARD, CLASSIFICATION_ELEVATED, or CLASSIFICATION_ELEVATED_HUMAN. Elevated tools write an AuditEntry.
Steward gate
contracts/steward_gate.py blocks execution until a contract is signed by a steward identity, checking PIPELINE_EXISTS, CONTRACT_EXISTS, STEWARD_VALID, NOT_STALE (30 days), and INTEGRITY_VALID. The module documents that there is no bypass mechanism.
Human approval
gates/human_approval.py mints single-use 256-bit tokens bound to a pipeline execution and contract version, expiring after 24 hours. Approval is required when classification is confidential or restricted and the operation is a mask or load.
Dual-store lineage
lineage/emitter.py writes PROV-AGENT events to both Qdrant and Postgres. For confidential or restricted data both writes must succeed or the pipeline blocks with LineageWriteError; for lower classifications a lineage gap is logged as a warning.
Post-mask validation
gates/pii_validator.py runs after masking, sampling up to 100 rows per table and rescanning sensitive columns through Presidio, then validating that values match expected tokenized, format-preserving, or redacted patterns.
Supporting Machinery
- Consistent tokenization — a masking seed drives deterministic tokens so the same entity maps to the same token across structured columns and free text, preserving joins on masked data.
- Contract validation — JSON Schema draft 2020-12 contracts describe columns, PII types, and governance metadata, with
contracts/schema_drift_detector.pycatching upstream shape changes. - Quality assertions —
quality/parses assertions and executes them through DuckDB, alongside a transform engine handling joins, filters, derives, and aggregates. - Gate registry —
gates/gate_registry.pydefines blocking versus warning modes across nine pipeline triggers, writing verdicts into conductor state. - GDPR Article 30 —
compliance/gdpr_article30.pyderives a processing record from lineage and contract fields rather than from a separately maintained document. - Credential hygiene —
tools/credential_resolver.pyresolves secrets at runtime from Vault; credentials are never embedded in pipeline definitions.
3Requirements
Requirements reflect the gates and modules implemented in the repository.
| ID | Requirement |
|---|---|
| REQ-DPL-001 | Expose all pipeline operations as classified MCP tools with no public HTTP surface. |
| REQ-DPL-002 | Require a steward-signed, non-stale, integrity-valid contract before execution. |
| REQ-DPL-003 | Validate pipeline and contract documents against JSON Schema draft 2020-12. |
| REQ-DPL-004 | Block mask and load operations on confidential or restricted data pending human approval. |
| REQ-DPL-005 | Issue single-use approval tokens bound to execution and contract version, expiring in 24h. |
| REQ-DPL-006 | Apply consistent tokenization across structured and unstructured columns. |
| REQ-DPL-007 | Emit PROV-AGENT lineage to both Qdrant and Postgres. |
| REQ-DPL-008 | Fail closed on lineage write failure for confidential and restricted classifications. |
| REQ-DPL-009 | Rescan masked output for residual PII and validate token formats. |
| REQ-DPL-010 | Detect schema drift against the agreed contract. |
| REQ-DPL-011 | Resolve credentials at runtime from a secret store, never from pipeline definitions. |
| REQ-DPL-012 | Derive GDPR Article 30 processing records from lineage and contract metadata. |
4Interfaces
Agents drive the pipeline entirely through MCP tools. Schemas define the documents those tools consume.
| Interface | Purpose |
|---|---|
| data_connect | Establish a governed source or destination connection |
| data_extract | Extract records under an approved contract |
| data_transform | Apply joins, filters, derives, and aggregates |
| data_mask | Apply masking and anonymization policy (elevated) |
| data_load | Load into a destination (elevated) |
| data_profile | Profile columns and classify candidate PII |
| data_contract_validate | Validate a contract document |
| data_lineage_query | Query emitted lineage events |
| pipeline.schema.json / contract.schema.json | Pipeline and contract document schemas |
| masking-policy.schema.json / lineage-event.schema.json | Masking policy and lineage event schemas |
5Integration Points
The pipeline composes several external services, each with a specific governance role.
- Microsoft Presidio — named-entity recognition for PII detection, used both to classify and to re-verify after masking.
- HashiCorp Vault — runtime credential resolution, keeping secrets out of pipeline definitions.
- Qdrant and PostgreSQL — the dual lineage stores whose combined success is required for sensitive classifications.
- Airbyte — extraction and loading connectors, running as server and worker containers.
- OpenTelemetry — trace emission alongside lineage events.
- Conductor — data phases insert between architecture and implementation, coordinated through
conductor-state.json.
6Repository
Modules are organised by governance concern, with policy and schema separated from execution.
| Path | Purpose |
|---|---|
| tools/ | The eight MCP tools and the classified tool registry |
| gates/ | Human approval, PII validation, gate registry, key rotation |
| contracts/ | Steward gate, schema drift detection, artifact integrity |
| lineage/ | Qdrant, Postgres, and OpenTelemetry emitters |
| quality/ | Assertion parser and DuckDB execution engine |
| compliance/ | GDPR Article 30 record derivation |
| schemas/ | Pipeline, contract, masking policy, and lineage event schemas |
| agents/ | Data engineer and data steward agent definitions |
| tests/ | 36 test modules across the gate and contract logic |
7Implementation Notes
Operational limits and deployment constraints worth knowing before running it.
- Data phases activate only at STANDARD tier or above; the basic tier disables them entirely.
- The deployment targets roughly 5.3GB of RAM across six containers, against a stated 6GB ceiling.
- Approval tokens are single-use and expire in 24 hours; an expired token requires a fresh approval rather than a renewal.
- Steward review is architecturally mandatory — the module documents that no bypass exists.
${VARIABLE}placeholders pass schema validation as plain strings, so substitution errors surface at runtime rather than validation time.- The masking engine ships as its own container image:
masking_engine/carries the Dockerfile, requirements, and the Presidio NER client (app/ner/presidio_client.py) thatgates/pii_validator.pyimports for post-mask rescanning. - CI runs
pytest -qas a hard gate — a failing test fails the build. 36 test modules cover the gate and contract logic.