Agentic Data Plane

// Agentic Data Plane — Core Pipeline Flow

1. DATA INGESTION EVENT
   source: PolicyAdminSystem | ClaimsSystem | BureauFeed | AgentOutput
   ├── emit: LineageEvent { event_id, source_id, schema_hash, timestamp }
   ├── trigger: ClassificationEngine.scan(data_stream)
   └── publish: audit_bus.lineage_ingestion

2. CLASSIFICATION ENGINE (in-motion scan)
   for each field in data_stream:
   ├── run: PIIDetector.classify(field)          → NONE | PII | PHI | FINANCIAL
   ├── run: InsuranceClassifier.classify(field)   → PUBLIC | INTERNAL | CONFIDENTIAL | RESTRICTED
   ├── if tier == RESTRICTED:
   │   ├── emit: GovernanceAlert { severity: HIGH, field, detected_type }
   │   └── publish: audit_bus.classification_violation
   └── persist: ClassificationRecord { field_path, tier, confidence, timestamp }

3. LINEAGE ENGINE (transformation tracking)
   on: AgentCodeExecution | DataTransformation | PipelineStep
   ├── record: LineageNode { node_id, operation, inputs[], outputs[], agent_id }
   ├── link: LineageEdge { from: input_node_id, to: output_node_id, transform_fn }
   ├── update: LineageDAG.add(node, edges)
   └── publish: audit_bus.lineage_transform

4. QUALITY VALIDATION PIPELINE
   for each dataset at checkpoint:
   ├── score: Completeness    = (non_null_fields / total_fields) * 250
   ├── score: Accuracy        = (validated_values / total_values) * 250
   ├── score: Consistency     = (schema_conforming / total_records) * 250
   ├── score: Timeliness      = freshness_factor(data_age, sla_hours) * 250
   ├── total_score            = sum(dimension_scores)  // 0–1000
   ├── if total_score < threshold:
   │   ├── emit: QualityAlert { dataset_id, score, failing_dimensions }
   │   └── publish: audit_bus.quality_violation
   └── persist: QualityRecord { dataset_id, score, dimensions, timestamp }

5. PIPELINE OBSERVABILITY
   monitor: all_active_pipelines
   ├── check: schema_drift(current_schema, baseline_schema)
   ├── check: volume_anomaly(record_count, expected_range)
   ├── check: latency_sla(processing_time, max_allowed_ms)
   ├── check: null_rate_spike(null_pct, baseline_null_pct)
   ├── if anomaly_detected:
   │   ├── emit: PipelineAlert { pipeline_id, anomaly_type, severity }
   │   └── notify: downstream_consumers[]
   └── publish: audit_bus.pipeline_health

6. LINEAGE QUERY API
   query: LineageAPI.trace(output_field_id)
   ├── traverse: LineageDAG backward from output_field_id
   ├── collect: all ancestor nodes, edges, transforms
   ├── resolve: source_systems[], agent_ids[], timestamps[]
   └── return: LineageReport { path[], sources[], transforms[], duration_ms }

7. FINANCIAL RECONCILIATION
   trigger: pipeline_run.complete
   ├── extract_stage:  compare(source_count, extracted_count, sha256(primary_keys))
   ├── transform_stage: for each calculation:
   │   ├── capture: { inputs, formula_ref, outputs } → ComputationRecord
   │   └── verify:  replay(inputs, formula) === stored_output
   ├── aggregate_stage: recompute(control_totals_from_details) === stored_aggregates
   │   └── tolerance: exact(counts), ±$0.01(currency), configurable(derived)
   ├── load_stage:    compare(staged_hash, committed_hash)
   ├── end_to_end:    compare(source_grand_totals, target_grand_totals)
   ├── if break_detected:
   │   ├── emit: ReconciliationBreak { break_id, stage, field, expected, actual, variance }
   │   ├── block: downstream_consumption
   │   └── route: break_resolution_workflow
   ├── generate: ReconciliationManifest { verdict, stages[], breaks[], signature }
   └── publish: audit_bus.data_reconciliation

8. GOVERNANCE AUDIT BUS INTEGRATION
   subscribe: governance_framework.audit_bus
   ├── forward: lineage events    → audit_trail.data_lineage
   ├── forward: quality alerts    → audit_trail.data_quality
   ├── forward: classification    → audit_trail.data_classification
   ├── forward: reconciliation    → audit_trail.data_reconciliation
   └── forward: pipeline health   → audit_trail.pipeline_observability

Build a production-ready Agentic Data Plane for insurance software systems generated by autonomous AI agents. This is a data governance infrastructure layer that runs beneath agent-generated systems, providing lineage tracking, quality validation, pipeline observability, continuous data classification, and financial audit-grade source-to-target reconciliation. Target environment: Node.js 20+ backend with TypeScript, PostgreSQL for lineage storage, Redis for real-time event streaming, and a React + D3.js lineage dashboard.

## Core Components to Build

### 1. Data Lineage Engine
Create a lineage tracking service at `src/lineage/LineageEngine.ts`:

- Implement a directed acyclic graph (DAG) data structure for data provenance using an adjacency list with `LineageNode` and `LineageEdge` types
- `LineageNode`: { node_id: string, operation: string, agent_id: string, session_id: string, inputs: string[], outputs: string[], transform_fn: string, schema_hash: string, timestamp: ISO8601, metadata: Record }
- `LineageEdge`: { from: string, to: string, field_map: Record, transform_applied: string }
- Implement `LineageEngine.record(event: LineageEvent): Promise` — appends to both in-memory DAG and PostgreSQL persistence
- Implement `LineageEngine.trace(output_field_id: string): Promise` — backward traversal from output to all source nodes, returning the complete path with < 500ms SLA
- Implement `LineageEngine.query(filters: LineageQueryFilters): Promise` — supports filtering by time range, agent_id, source_system, and output_destination
- Persist all lineage nodes and edges in PostgreSQL tables `lineage_nodes` and `lineage_edges` with appropriate indexes on node_id, agent_id, and timestamp
- Emit all lineage events to the governance audit bus (Redis Streams topic `audit:data:lineage`)
- Enforce 7-year retention for standard data, 10-year for Restricted-tier; implement `LineageEngine.purge(before: Date, tier: DataTier)` with dual-authorization requirement (two officer tokens required)

### 2. Quality Validation Pipeline
Create `src/quality/QualityValidator.ts`:

- Implement four scoring dimensions, each 0–250 points, summing to a 0–1000 total:
  - Completeness: count non-null required fields / total required fields, weighted by field criticality
  - Accuracy: validate values against reference datasets (NAIC codes, ISO codes, VIN patterns, ZIP+4 format)
  - Consistency: check cross-field business rules (effective_date < expiry_date, premium > 0, deductible <= limit, valid state/LOB combinations)
  - Timeliness: freshness_score = max(0, 250 * (1 - data_age_hours / sla_hours))
- Implement `QualityValidator.score(dataset: Dataset, config: QualityConfig): Promise`
- `QualityScore`: { dataset_id, total_score, completeness, accuracy, consistency, timeliness, failing_checks: FailingCheck[], timestamp }
- Implement threshold enforcement: if total_score < config.blocking_threshold (default 700), reject promotion with `QualityBlockError` and publish `audit:data:quality` event
- Implement trend tracking: store score history in PostgreSQL `quality_scores` table; compute 7-day and 30-day rolling averages; emit `QUALITY_DEBT_ALERT` when 30-day trend shows > 5% decline before blocking threshold is reached
- Expose `QualityValidator.report(dataset_id: string, from: Date, to: Date): Promise` for regulatory report generation

### 3. Pipeline Observability Layer
Create `src/observability/PipelineObserver.ts`:

- Register pipelines with `PipelineObserver.register(pipeline: PipelineDefinition)` — stores baseline schema, expected volume range, null rate baselines, and SLA config
- Implement real-time monitoring via `PipelineObserver.observe(pipeline_id: string, batch: DataBatch): Promise`:
  - Schema drift: deep-compare current schema to baseline; detect field additions, removals, type changes, and required/optional status changes
  - Volume anomaly: compare batch record count to rolling 30-day baseline; flag if deviation > 2σ
  - Null rate spike: compute per-field null rates; flag any field with > 10% increase from baseline
  - SLA breach: measure processing time; flag if exceeds `pipeline.max_processing_ms`
- On anomaly detection: emit `PipelineAlert` to Redis Streams topic `audit:data:pipeline`; notify all registered downstream consumers via webhook with retry (3 attempts, exponential backoff)
- Consumers acknowledge via `PipelineObserver.acknowledge(pipeline_id: string, consumer_id: string, alert_id: string)` — unacknowledged alerts after configurable TTL escalate to manual review queue

### 4. Continuous Classification Engine
Create `src/classification/ClassificationEngine.ts`:

- Implement field-level classification scanning for data in motion
- Build pattern-matching + ML hybrid classifiers for:
  - PII: SSN (XXX-XX-XXXX pattern + Luhn-variant check), DOB, name+address combos, email, phone
  - PHI (HIPAA): medical record numbers, diagnosis codes (ICD-10 format), treatment dates co-located with patient identifiers, provider NPI numbers
  - Financial: payment card numbers (Luhn validation), bank account numbers, routing numbers
  - Insurance-specific NAIC fields: NAIC company codes, FEIN/TIN in insurance context, adjuster license numbers, DOI filing reference numbers
- Each classifier returns `ClassificationResult { field_path, detected_tier: DataTier, detected_type: string, confidence: number (0.0-1.0), evidence: string[] }`
- Classifications with confidence < 0.75 are flagged `NEEDS_REVIEW` rather than auto-enforced
- Implement `ClassificationEngine.scan(data_stream: AsyncIterable): AsyncIterable` — streaming API, max 150ms latency per record
- Publish all classification results to `audit:data:classification` Redis Stream
- Implement override API: `ClassificationEngine.override(result_id: string, new_tier: DataTier, rationale: string, officer_token: string)` — records override to audit bus with officer identity

### 5. Lineage Query REST + GraphQL API
Create `src/api/LineageAPI.ts` using Express + Apollo Server:

REST endpoints:
- `GET /api/v1/lineage/trace/:output_field_id` — full backward trace to sources
- `GET /api/v1/lineage/sources/:source_id/outputs` — all outputs derived from a source
- `GET /api/v1/lineage/datasets/:dataset_id/history` — transformation history
- `GET /api/v1/lineage/events?from=&to=&agent_id=&pipeline_id=` — time-range event query
- `GET /api/v1/lineage/report/:policy_id` — regulatory lineage report for a policy record
- `GET /api/v1/quality/datasets/:dataset_id/trend?days=30` — quality trend report
- `POST /api/v1/reports/regulatory` — generate formatted regulatory report (PDF/JSON)

GraphQL schema: expose `lineageTrace`, `lineageSearch`, `qualityScore`, `classificationHistory`, `pipelineHealth` queries with full type definitions.

All API responses must include `X-Query-Time-Ms` header. Authentication via JWT with role-based access (viewer, analyst, compliance_officer, admin).

### 6. Lineage Dashboard
Create `src/dashboard/` as a React + TypeScript SPA:

- Use D3.js for DAG visualization; render nodes as rectangles with color-coded tiers (T0=gray, T1=blue, T2=orange, T3=red)
- Support interactive features: node click to expand/collapse; edge hover to show transform details; field-level drill-down panel
- Search bar: query by output field name, dataset ID, time range; results highlight matching subgraph
- Performance: 1,000-node DAG must render in < 3 seconds using WebWorker for layout computation
- Quality panel: show sparkline of 30-day quality trend per selected dataset
- Pipeline health panel: show last 24h of pipeline observations with anomaly indicators
- Classification coverage: show heatmap of classification tier distribution across active pipelines

### 7. Governance Audit Bus Integration
Create `src/audit/AuditBusClient.ts`:

- Wrap Redis Streams with typed publish methods for each event type
- Implement `AuditBusClient.publish(event: AuditEvent): Promise` with retry logic (3 attempts) and dead-letter queue for failed publishes
- Sign all events with HMAC-SHA256 using the service's private key before publishing — consumers can verify integrity
- Implement hash-chained event IDs: each event's ID incorporates a hash of the previous event in the stream, providing tamper evidence
- Subscribe methods: `AuditBusClient.subscribe(topic: string, handler: EventHandler): () => void` — returns unsubscribe function

### 8. Financial Reconciliation Engine
Create `src/reconciliation/ReconciliationEngine.ts`:

- Implement `ReconciliationEngine.reconcile(pipelineRunId: string): Promise<ReconciliationManifest>` — compares source and target at every stage
- At extraction: compare source record count vs extracted count, SHA-256 of sort-ordered primary keys
- At transformation: capture inputs + formula reference + outputs as `ComputationRecord`, support deterministic replay via `ReconciliationEngine.replay(recordId: string)`
- At aggregation: independently recompute control totals from detail records, compare against stored aggregates with configurable tolerance (exact for counts, ±$0.01 for currency)
- At loading: compare staged record count/hash vs committed count/hash in target system
- End-to-end: compare source grand totals vs target grand totals using independently computed values
- `ReconciliationManifest`: { run_id, timestamp, source_system, target_system, stages: StageResult[], record_counts: { source, target, matched, unmatched, orphaned }, control_totals: ControlTotal[], verdict: 'PASS' | 'PASS_WITH_EXCEPTIONS' | 'FAIL', breaks: Break[], signature: Ed25519Signature }
- Break resolution: `ReconciliationEngine.resolveBreak(breakId: string, rootCause: string): Promise<void>` — records resolution with lineage trace
- Continuous mode: auto-reconcile on every pipeline run, publish results to audit bus topic `audit:data:reconciliation`
- Report generation: `ReconciliationEngine.generateReport(manifestId: string, format: 'pdf' | 'json'): Promise<Buffer>`
- Point-in-time snapshots: immutable, queryable by date range, minimum 7-year retention for insurance compliance
- Reconciliation dashboard: current status, historical trends, break frequency, MTTR per pipeline
- SLA: reconciliation of 1M-record pipeline completes within 60 seconds

## Database Schema (PostgreSQL)

Create migration files in `migrations/`:

```sql
-- Lineage nodes
CREATE TABLE lineage_nodes (
  node_id UUID PRIMARY KEY,
  operation VARCHAR(100) NOT NULL,
  agent_id VARCHAR(100) NOT NULL,
  session_id VARCHAR(100) NOT NULL,
  schema_hash VARCHAR(64),
  tier VARCHAR(20),
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  metadata JSONB
);
CREATE INDEX idx_lineage_nodes_agent ON lineage_nodes(agent_id);
CREATE INDEX idx_lineage_nodes_created ON lineage_nodes(created_at);

-- Lineage edges
CREATE TABLE lineage_edges (
  edge_id UUID PRIMARY KEY,
  from_node UUID REFERENCES lineage_nodes(node_id),
  to_node UUID REFERENCES lineage_nodes(node_id),
  field_map JSONB,
  transform_applied TEXT,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_lineage_edges_from ON lineage_edges(from_node);
CREATE INDEX idx_lineage_edges_to ON lineage_edges(to_node);

-- Quality scores
CREATE TABLE quality_scores (
  score_id UUID PRIMARY KEY,
  dataset_id VARCHAR(200) NOT NULL,
  pipeline_id VARCHAR(200),
  total_score SMALLINT NOT NULL,
  completeness SMALLINT,
  accuracy SMALLINT,
  consistency SMALLINT,
  timeliness SMALLINT,
  failing_checks JSONB,
  scored_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_quality_dataset ON quality_scores(dataset_id, scored_at DESC);

-- Classification records
CREATE TABLE classification_records (
  record_id UUID PRIMARY KEY,
  field_path TEXT NOT NULL,
  pipeline_id VARCHAR(200),
  detected_tier VARCHAR(20) NOT NULL,
  detected_type VARCHAR(100),
  confidence NUMERIC(4,3),
  status VARCHAR(20) DEFAULT 'AUTO',
  override_rationale TEXT,
  override_officer VARCHAR(100),
  detected_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_classification_pipeline ON classification_records(pipeline_id, detected_at DESC);
```

## Configuration
Use environment variables for all configuration. Provide a `.env.example` with all required variables:
- `DATABASE_URL`, `REDIS_URL`, `AUDIT_BUS_SERVICE_KEY`
- `QUALITY_BLOCKING_THRESHOLD` (default: 700)
- `CLASSIFICATION_MIN_CONFIDENCE` (default: 0.75)
- `LINEAGE_RETENTION_YEARS_STANDARD` (default: 7)
- `LINEAGE_RETENTION_YEARS_RESTRICTED` (default: 10)
- `PIPELINE_ANOMALY_SIGMA_THRESHOLD` (default: 2.0)

## Testing Requirements
- Unit tests for all validators, classifiers, and lineage traversal logic (Jest)
- Integration tests for API endpoints using supertest with a test database
- Performance test: lineage trace on 1M-node DAG must complete in < 500ms
- Classification accuracy test: run built-in test corpus of 500 labeled insurance records; assert > 98.5% detection rate for T3 fields

## Deliverables
1. Complete TypeScript source with no stubs or TODOs
2. Docker Compose file with PostgreSQL, Redis, API service, and dashboard
3. Database migration files
4. API documentation (OpenAPI 3.0 YAML)
5. README with setup instructions, environment variable reference, and regulatory compliance notes
6. Test suite with > 90% coverage on core components

Do not use any hardcoded secrets. All sensitive configuration via environment variables. No arbitrary code execution from user input in the lineage or classification APIs.