ilds an adjacency list from the manifest and performs breadth-first traversal to compute transitive closures. BFS guarantees shortest-path detection for model principals, which simplifies taint classification.
3. Deterministic Ordering: All node expansions and principal collections use sorted iteration. This eliminates non-determinism from hash map ordering or filesystem traversal, ensuring byte-identical STDOUT across runs.
4. Role-Based Enforcement: The engine distinguishes between authorization and context roles. Model-tainted signals in context generate informational warnings. The same taint in authorization triggers a hard failure, enforcing the core security invariant.
Implementation (TypeScript)
The following implementation replaces the original Python script with a modular TypeScript architecture. It uses explicit interfaces, a graph builder, and a visitor-based validator. The logic remains functionally equivalent but demonstrates a production-ready structure with type safety and explicit error boundaries.
import { readFileSync } from 'fs';
import { exit } from 'process';
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// Type Definitions
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
type PrincipalKind = 'human' | 'external' | 'model';
type FeatureRole = 'authorization' | 'context';
type TaintClass = 'WORLD_ANCHORED' | 'MODEL_AUTHORED' | 'MODEL_LAUNDERED';
interface StoreSpec {
written_by: string[];
}
interface GateFeature {
name: string;
reads: string;
role: FeatureRole;
}
interface Manifest {
stores: Record<string, StoreSpec>;
gate_features: GateFeature[];
}
interface ValidationResult {
feature: string;
role: FeatureRole;
taint: TaintClass | null;
hasFeedbackLoop: boolean;
pathToModel: string[] | null;
}
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// Core Engine
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
class ProvenanceValidator {
private graph: Map<string, string[]> = new Map();
private principals: Set<string> = new Set();
private results: ValidationResult[] = [];
constructor(private manifest: Manifest) {
this.buildGraph();
}
private buildGraph(): void {
for (const [store, spec] of Object.entries(this.manifest.stores)) {
this.graph.set(store, [...spec.written_by].sort());
}
}
private isPrincipal(node: string): boolean {
return node.includes(':');
}
private getKind(node: string): PrincipalKind | null {
if (!this.isPrincipal(node)) return null;
const kind = node.split(':')[0] as PrincipalKind;
return ['human', 'external', 'model'].includes(kind) ? kind : null;
}
private computeClosure(startNode: string): { principals: Set<string>; visited: Set<string> } {
const principals = new Set<string>();
const visited = new Set<string>();
const queue: string[] = [...this.graph.get(startNode) || []];
while (queue.length > 0) {
const current = queue.shift()!;
if (this.isPrincipal(current)) {
principals.add(current);
} else if (!visited.has(current)) {
visited.add(current);
const next = this.graph.get(current) || [];
queue.push(...next);
}
}
return { principals, visited };
}
private findModelPath(startNode: string): string[] | null {
const visited = new Set<string>();
const queue: string[][] = [[startNode]];
while (queue.length > 0) {
const path = queue.shift()!;
const current = path[path.length - 1];
const writers = this.graph.get(current) || [];
for (const writer of writers) {
if (this.isPrincipal(writer)) {
if (this.getKind(writer) === 'model') {
return [...path, writer];
}
} else if (!visited.has(writer)) {
visited.add(writer);
queue.push([...path, writer]);
}
}
}
return null;
}
private detectFeedbackLoop(startNode: string): boolean {
const path = new Set<string>();
const stack: string[] = [startNode];
while (stack.length > 0) {
const current = stack.pop()!;
if (path.has(current)) return true;
path.add(current);
const writers = this.graph.get(current) || [];
for (const writer of writers) {
if (!this.isPrincipal(writer)) {
stack.push(writer);
}
}
}
return false;
}
public validate(): ValidationResult[] {
this.results = [];
for (const feature of this.manifest.gate_features) {
const closure = this.computeClosure(feature.reads);
const modelPath = this.findModelPath(feature.reads);
const hasLoop = this.detectFeedbackLoop(feature.reads);
let taint: TaintClass | null = null;
if (modelPath) {
taint = modelPath.length === 2 ? 'MODEL_AUTHORED' : 'MODEL_LAUNDERED';
}
this.results.push({
feature: feature.name,
role: feature.role,
taint,
hasFeedbackLoop: hasLoop,
pathToModel: modelPath
});
}
return this.results;
}
public getExitCode(): number {
const authViolations = this.results.filter(
r => r.role === 'authorization' && r.taint !== null
);
return authViolations.length > 0 ? 1 : 0;
}
}
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
// CLI Entry Point
// βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
function main(): void {
const filePath = process.argv[2];
if (!filePath) {
console.error('ERROR: manifest path required');
exit(2);
}
let raw: string;
try {
raw = readFileSync(filePath, 'utf-8');
} catch {
console.error('ERROR: cannot read manifest file');
exit(2);
}
let data: unknown;
try {
data = JSON.parse(raw);
} catch {
console.error('ERROR: invalid JSON format');
exit(2);
}
if (typeof data !== 'object' || data === null || !('stores' in data) || !('gate_features' in data)) {
console.error('ERROR: manifest missing required sections');
exit(2);
}
const manifest = data as Manifest;
const validator = new ProvenanceValidator(manifest);
const results = validator.validate();
console.log('=== Provenance Validation Report ===');
for (const r of results) {
const status = r.taint ? `[${r.taint}]` : '[WORLD_ANCHORED]';
const loopFlag = r.hasFeedbackLoop ? ' β FEEDBACK_LOOP' : '';
console.log(`${r.feature} (${r.role}): ${status}${loopFlag}`);
if (r.pathToModel) {
console.log(` β Model path: ${r.pathToModel.join(' β ')}`);
}
}
const exitCode = validator.getExitCode();
console.log(`\nExit code: ${exitCode}`);
exit(exitCode);
}
main();
Why This Architecture Works
The class-based structure isolates graph construction, traversal, and validation logic. This separation enables unit testing of individual components without mocking filesystem I/O. The BFS-based path finder guarantees deterministic shortest-path detection, which is critical for distinguishing MODEL_AUTHORED (direct write) from MODEL_LAUNDERED (indirect write). The feedback loop detector uses a depth-first approach with path tracking, catching recursive data pipelines that could cause authorization drift over time.
TypeScript's strict typing prevents manifest schema drift from causing silent failures. The PrincipalKind and FeatureRole unions enforce compile-time validation of expected values. Runtime checks remain for JSON parsing and file I/O, but the core engine operates on guaranteed structures.
Pitfall Guide
1. Name-Based Trust Assumption
Explanation: Engineers assume a signal's name reflects its origin. sender_trust appears human-curated, but the underlying store may be populated by a batch inference job.
Fix: Never trust naming conventions. Require explicit written_by declarations for every store. Validate provenance before evaluating signal semantics.
2. Shallow Dependency Tracing
Explanation: Checking only direct writers misses laundered model outputs. A reputation table might be written by aggregation_service, which itself reads from a model output store.
Fix: Compute the full transitive closure. Use BFS or DFS to traverse all upstream dependencies until reaching principal leaves. Classify based on the complete closure, not the immediate parent.
3. Ignoring Feedback Loops
Explanation: When a model's output influences a store that later feeds back into the model's training or inference context, authorization drift occurs. The gate validates actions based on signals the model helped create.
Fix: Implement cycle detection in the write graph. Flag any FEEDBACK_LOOP condition where a model principal appears in a reachable cycle. Treat feedback loops as critical violations for authorization signals.
4. Hardcoding Principal Kinds
Explanation: Assuming only human, external, and model exist limits extensibility. New integrations (e.g., service:payment_gateway, bot:monitoring_agent) break rigid validation logic.
Fix: Define principal kinds as an explicit enum or union type. Allow custom kinds but enforce a strict validation rule: only human and external qualify as world-anchored. Reject unknown kinds during manifest parsing.
5. Treating Context Taint as Safe
Explanation: Model-tainted signals in context roles are often ignored. However, context signals can influence downstream authorization logic through heuristic weighting or LLM reasoning.
Fix: Log all model-tainted features regardless of role. Enforce strict separation: authorization signals must be world-anchored. context signals may carry model taint but require explicit documentation and monitoring.
6. Manifest Drift from Runtime
Explanation: The declared write map diverges from actual data pipeline behavior. A store marked as human:sre_approver might be silently overwritten by a cron job running a model.
Fix: Integrate manifest validation into CI/CD pipelines. Pair static linting with runtime data lineage tools (e.g., OpenLineage, Marquez). Alert on discrepancies between declared and observed writers.
7. Missing Cycle Detection in Path Finders
Explanation: Breadth-first search without visited tracking can infinite-loop on cyclic graphs, causing CI hangs or stack overflows.
Fix: Maintain a visited set during traversal. Skip nodes already processed. For feedback loop detection, use path tracking instead of global visited sets to identify cycles specific to each feature's closure.
Production Bundle
Action Checklist
Decision Matrix
| Scenario | Recommended Approach | Why | Cost Impact |
|---|
| Pre-merge validation | Static provenance lint | Deterministic, zero network, catches pipeline drift early | Low (CI compute) |
| Runtime authorization | Policy engine + signal validation | Catches dynamic data changes, handles live traffic | Medium (latency overhead) |
| Compliance auditing | Manifest versioning + lineage tools | Provides immutable trail, satisfies regulatory requirements | Medium (tooling integration) |
| Rapid prototyping | Name-based trust + manual review | Faster iteration, acceptable for non-critical paths | Low (high risk) |
| Production hardening | Provenance lint + feedback loop detection | Prevents model drift, enforces strict authorization boundaries | Low (CI gate) |
Configuration Template
{
"stores": {
"human_approvals": {
"written_by": ["human:security_team"]
},
"bank_transactions": {
"written_by": ["external:stripe_feed"]
},
"risk_aggregator": {
"written_by": ["human_approvals", "bank_transactions"]
},
"model_scoring": {
"written_by": ["model:fraud_classifier_v2"]
},
"final_trust": {
"written_by": ["risk_aggregator"]
}
},
"gate_features": [
{
"name": "transaction_trust",
"reads": "final_trust",
"role": "authorization"
},
{
"name": "fraud_probability",
"reads": "model_scoring",
"role": "context"
}
]
}
Quick Start Guide
- Save the validator: Copy the TypeScript implementation into
provenance-validator.ts. Install dependencies: npm init -y && npm install typescript @types/node.
- Create a manifest: Write a
manifest.json file declaring your stores and gate features using the template above. Ensure every written_by array contains valid principal kinds or store references.
- Compile and run: Execute
npx tsc provenance-validator.ts --outDir dist && node dist/provenance-validator.js manifest.json. Review the STDOUT report. Exit code 0 indicates clean authorization signals. Exit code 1 flags model-tainted authorization features. Exit code 2 indicates manifest syntax errors.
- Integrate into CI: Add the command to your pipeline configuration. Fail builds on non-zero exit codes. Archive STDOUT logs for audit trails. Update manifests whenever data pipelines change writers or add intermediate stores.