A case decision was approved.

Actor: reviewer_123
Action: APPROVE_DECISION
Time: 2026-07-05T09:21:33Z

Decision record depends on:
- case profile projection v12
- sanctions screening result v5
- evidence document metadata v8
- policy rule table version 2026.06

The risk classification was generated by rule_engine@sha256:...
using source events up to LSN 42/BF9E2A10,
policy bundle 2026.06.14,
and external registry snapshot 2026-07-01.
Validation passed: checksum=..., input_count=..., exception_count=0.

name: regulatory_case_summary
owner: enforcement-data-platform
classification: internal-restricted
source_of_truth: false
canonical_source:
  - operational.case
  - operational.case_status_transition
  - operational.evidence_document
freshness_sla: P15M
rebuildable: true
retention_policy: CASE_RETENTION_7Y
lineage_level:
  dataset: required
  column: required
  row: required_for_decision_fields
quality_gates:
  - row_count_reconciliation
  - orphan_case_check
  - status_transition_validity
consumers:
  - case_dashboard
  - monthly_enforcement_report
  - evidence_export_service

CREATE TABLE data_asset (
  asset_id            uuid PRIMARY KEY,
  asset_type          text NOT NULL CHECK (asset_type IN (
    'database', 'schema', 'table', 'column', 'topic', 'file', 'report', 'api', 'model', 'dashboard'
  )),
  asset_urn           text NOT NULL UNIQUE,
  display_name        text NOT NULL,
  owner_team          text NOT NULL,
  classification      text NOT NULL,
  system_name         text NOT NULL,
  created_at          timestamptz NOT NULL DEFAULT now(),
  retired_at          timestamptz
);

CREATE TABLE data_job (
  job_id              uuid PRIMARY KEY,
  job_urn             text NOT NULL UNIQUE,
  job_type            text NOT NULL CHECK (job_type IN (
    'etl', 'cdc', 'api_command', 'migration', 'backfill', 'report_build', 'ml_inference'
  )),
  owner_team          text NOT NULL,
  code_repository     text,
  description         text,
  created_at          timestamptz NOT NULL DEFAULT now(),
  retired_at          timestamptz
);

CREATE TABLE data_lineage_edge (
  lineage_edge_id     uuid PRIMARY KEY,
  upstream_asset_id   uuid NOT NULL REFERENCES data_asset(asset_id),
  downstream_asset_id uuid NOT NULL REFERENCES data_asset(asset_id),
  job_id              uuid REFERENCES data_job(job_id),
  transformation_type text NOT NULL CHECK (transformation_type IN (
    'copy', 'filter', 'join', 'aggregate', 'derive', 'mask', 'tokenize', 'enrich', 'project', 'export'
  )),
  lineage_level       text NOT NULL CHECK (lineage_level IN ('system', 'dataset', 'column', 'row', 'value')),
  confidence          text NOT NULL CHECK (confidence IN ('declared', 'observed', 'verified')),
  valid_from          timestamptz NOT NULL DEFAULT now(),
  valid_to            timestamptz,
  metadata            jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX idx_lineage_upstream ON data_lineage_edge(upstream_asset_id);
CREATE INDEX idx_lineage_downstream ON data_lineage_edge(downstream_asset_id);
CREATE INDEX idx_lineage_job ON data_lineage_edge(job_id);

postgres://prod-core/enforcement/public.case
postgres://prod-core/enforcement/public.case.status
kafka://prod/enforcement.case-events.v1
s3://regulated-lake/bronze/case_events/dt=2026-07-05/
dashboard://superset/enforcement-case-overview
report://regulatory/monthly-enforcement-summary/v2
api://case-service/GET /cases/{caseId}
model://risk-classifier/v3

case_table
summary_report
source_file

postgres://prod-core/enforcement/public.case
report://regulator-x/monthly-enforcement-summary/v2
sftp://agency-y/imports/registry_snapshot_2026_07_01.csv

INSERT INTO data_lineage_edge (
  lineage_edge_id,
  upstream_asset_id,
  downstream_asset_id,
  job_id,
  transformation_type,
  lineage_level,
  confidence,
  metadata
)
VALUES (
  gen_random_uuid(),
  :case_table_asset_id,
  :case_summary_table_asset_id,
  :case_summary_job_id,
  'aggregate',
  'dataset',
  'verified',
  jsonb_build_object(
    'refresh_mode', 'incremental',
    'freshness_sla', 'P15M',
    'watermark_column', 'updated_at'
  )
);

CREATE TABLE column_lineage_edge (
  column_lineage_edge_id uuid PRIMARY KEY,
  source_column_asset_id uuid NOT NULL REFERENCES data_asset(asset_id),
  target_column_asset_id uuid NOT NULL REFERENCES data_asset(asset_id),
  job_id                 uuid REFERENCES data_job(job_id),
  expression             text,
  transformation_type    text NOT NULL,
  confidence             text NOT NULL CHECK (confidence IN ('declared', 'observed', 'verified')),
  valid_from             timestamptz NOT NULL DEFAULT now(),
  valid_to               timestamptz
);

CREATE INDEX idx_column_lineage_source ON column_lineage_edge(source_column_asset_id);
CREATE INDEX idx_column_lineage_target ON column_lineage_edge(target_column_asset_id);

target: warehouse.case_fact.days_to_close
sources:
  - operational.case.opened_at
  - operational.case.closed_at
expression: date_diff('day', opened_at, closed_at)
semantic_contract: calendar days from case opening to terminal closure
null_semantics: null when case is not terminal
version: 2

CREATE TABLE provenance_run (
  run_id             uuid PRIMARY KEY,
  job_id             uuid NOT NULL REFERENCES data_job(job_id),
  run_status         text NOT NULL CHECK (run_status IN ('running', 'succeeded', 'failed', 'cancelled')),
  started_at         timestamptz NOT NULL,
  completed_at       timestamptz,
  code_version       text NOT NULL,
  config_version     text,
  rule_version       text,
  input_watermark    text,
  output_watermark   text,
  actor_type         text NOT NULL CHECK (actor_type IN ('scheduler', 'service', 'user', 'external_system')),
  actor_id           text NOT NULL,
  metadata           jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE TABLE entity_provenance (
  entity_provenance_id uuid PRIMARY KEY,
  run_id               uuid NOT NULL REFERENCES provenance_run(run_id),
  output_asset_id      uuid NOT NULL REFERENCES data_asset(asset_id),
  output_entity_key    text NOT NULL,
  input_asset_id       uuid NOT NULL REFERENCES data_asset(asset_id),
  input_entity_key     text NOT NULL,
  relationship_type    text NOT NULL CHECK (relationship_type IN (
    'copied_from', 'derived_from', 'aggregated_from', 'joined_with', 'validated_against', 'enriched_by'
  )),
  captured_at          timestamptz NOT NULL DEFAULT now(),
  metadata             jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE INDEX idx_entity_prov_output ON entity_provenance(output_asset_id, output_entity_key);
CREATE INDEX idx_entity_prov_input ON entity_provenance(input_asset_id, input_entity_key);
CREATE INDEX idx_entity_prov_run ON entity_provenance(run_id);

Output: risk_assessment(case_id=CASE-123, version=8)
Derived from:
- case(case_id=CASE-123, version=27)
- evidence_document(document_id=DOC-991, checksum=...)
- screening_result(screening_id=SCR-77, version=5)
- policy_bundle(policy_id=POLICY-2026-06, checksum=...)
Generated by:
- job: risk-classifier
- code: git sha abc123
- rule version: 2026.06.14
- run: 7df4...

CREATE TABLE decision_record (
  decision_id          uuid PRIMARY KEY,
  case_id              uuid NOT NULL,
  decision_type        text NOT NULL,
  decision_status      text NOT NULL,
  decision_outcome     text NOT NULL,
  decided_by_actor_id  text NOT NULL,
  decided_at           timestamptz NOT NULL,
  rationale            text NOT NULL,
  evidence_package_id  uuid,
  provenance_run_id    uuid REFERENCES provenance_run(run_id),
  source_snapshot_ref  text NOT NULL,
  policy_version       text NOT NULL,
  created_at           timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE report_build (
  report_build_id      uuid PRIMARY KEY,
  report_name          text NOT NULL,
  report_version       text NOT NULL,
  period_start         date NOT NULL,
  period_end           date NOT NULL,
  built_at             timestamptz NOT NULL DEFAULT now(),
  built_by_job_id      uuid NOT NULL REFERENCES data_job(job_id),
  source_snapshot_ref  text NOT NULL,
  code_version         text NOT NULL,
  config_version       text NOT NULL,
  row_count            bigint NOT NULL,
  output_checksum      text NOT NULL,
  validation_status    text NOT NULL CHECK (validation_status IN ('passed', 'failed', 'waived')),
  storage_uri          text NOT NULL
);

{
  "eventType": "LINEAGE_RUN_COMPLETED",
  "runId": "7df4b8d7-7a75-4a0f-9f9b-3b6b5d8b0d1a",
  "job": {
    "name": "case-summary-builder",
    "version": "git:abc123",
    "owner": "enforcement-data-platform"
  },
  "inputs": [
    {
      "assetUrn": "postgres://prod-core/enforcement/public.case",
      "snapshotRef": "lsn:42/BF9E2A10",
      "rowCount": 982331
    },
    {
      "assetUrn": "postgres://prod-core/enforcement/public.case_status_transition",
      "snapshotRef": "lsn:42/BF9E2A10",
      "rowCount": 3341220
    }
  ],
  "outputs": [
    {
      "assetUrn": "postgres://analytics/enforcement/public.case_summary",
      "rowCount": 982331,
      "checksum": "sha256:..."
    }
  ],
  "startedAt": "2026-07-05T01:00:00Z",
  "completedAt": "2026-07-05T01:04:12Z",
  "status": "succeeded"
}

WITH RECURSIVE downstream AS (
  SELECT
    le.upstream_asset_id,
    le.downstream_asset_id,
    1 AS depth,
    ARRAY[le.upstream_asset_id, le.downstream_asset_id] AS path
  FROM data_lineage_edge le
  WHERE le.upstream_asset_id = :asset_id
    AND le.valid_to IS NULL

  UNION ALL

  SELECT
    d.downstream_asset_id,
    le.downstream_asset_id,
    d.depth + 1,
    d.path || le.downstream_asset_id
  FROM downstream d
  JOIN data_lineage_edge le
    ON le.upstream_asset_id = d.downstream_asset_id
   AND le.valid_to IS NULL
  WHERE NOT le.downstream_asset_id = ANY(d.path)
)
SELECT
  d.depth,
  a.asset_urn,
  a.asset_type,
  a.owner_team,
  a.classification
FROM downstream d
JOIN data_asset a ON a.asset_id = d.downstream_asset_id
ORDER BY d.depth, a.asset_urn;

WITH RECURSIVE upstream AS (
  SELECT
    le.downstream_asset_id,
    le.upstream_asset_id,
    1 AS depth,
    ARRAY[le.downstream_asset_id, le.upstream_asset_id] AS path
  FROM data_lineage_edge le
  WHERE le.downstream_asset_id = :report_asset_id
    AND le.valid_to IS NULL

  UNION ALL

  SELECT
    u.upstream_asset_id,
    le.upstream_asset_id,
    u.depth + 1,
    u.path || le.upstream_asset_id
  FROM upstream u
  JOIN data_lineage_edge le
    ON le.downstream_asset_id = u.upstream_asset_id
   AND le.valid_to IS NULL
  WHERE NOT le.upstream_asset_id = ANY(u.path)
)
SELECT
  u.depth,
  a.asset_urn,
  a.asset_type,
  a.owner_team
FROM upstream u
JOIN data_asset a ON a.asset_id = u.upstream_asset_id
ORDER BY u.depth, a.asset_urn;

SELECT DISTINCT
  consumer.asset_urn AS consumer_urn,
  consumer.owner_team
FROM column_lineage_edge cle
JOIN data_asset target_col
  ON target_col.asset_id = cle.target_column_asset_id
JOIN data_lineage_edge le
  ON le.upstream_asset_id = target_col.asset_id
JOIN data_asset consumer
  ON consumer.asset_id = le.downstream_asset_id
WHERE cle.source_column_asset_id = :source_column_asset_id
  AND cle.valid_to IS NULL
  AND le.valid_to IS NULL;

Monthly closed case count is 12% lower than expected.

CREATE TABLE data_quality_result (
  quality_result_id uuid PRIMARY KEY,
  run_id            uuid NOT NULL REFERENCES provenance_run(run_id),
  asset_id          uuid NOT NULL REFERENCES data_asset(asset_id),
  rule_name         text NOT NULL,
  rule_version      text NOT NULL,
  status            text NOT NULL CHECK (status IN ('passed', 'failed', 'warning', 'waived')),
  checked_at        timestamptz NOT NULL DEFAULT now(),
  observed_value    numeric,
  expected_value    numeric,
  details           jsonb NOT NULL DEFAULT '{}'::jsonb
);

ALTER TABLE data_asset
ADD COLUMN contains_pii boolean NOT NULL DEFAULT false,
ADD COLUMN pii_categories text[] NOT NULL DEFAULT '{}',
ADD COLUMN privacy_transform text;

CREATE TABLE model_inference_provenance (
  inference_id        uuid PRIMARY KEY,
  model_name          text NOT NULL,
  model_version       text NOT NULL,
  feature_set_version text NOT NULL,
  input_asset_id      uuid NOT NULL REFERENCES data_asset(asset_id),
  input_entity_key    text NOT NULL,
  output_asset_id     uuid NOT NULL REFERENCES data_asset(asset_id),
  output_entity_key   text NOT NULL,
  inference_time      timestamptz NOT NULL,
  code_version        text NOT NULL,
  config_version      text NOT NULL,
  input_checksum      text NOT NULL,
  output_checksum     text NOT NULL,
  explanation_ref     text,
  metadata            jsonb NOT NULL DEFAULT '{}'::jsonb
);

CREATE TABLE lineage_event (
  lineage_event_id uuid PRIMARY KEY,
  event_type       text NOT NULL,
  event_time       timestamptz NOT NULL,
  emitted_by       text NOT NULL,
  correlation_id   text,
  payload          jsonb NOT NULL,
  payload_hash     text NOT NULL,
  supersedes_event_id uuid REFERENCES lineage_event(lineage_event_id)
);

lineage_coverage = assets_with_owner_and_upstream / total_production_assets
verified_lineage_coverage = verified_edges / total_edges
stale_lineage_count = edges_not_confirmed_within_policy_window
unknown_consumer_count = production_queries_from_unregistered_clients

schema_change:
  asset: postgres://prod-core/enforcement/public.case.status
  change_type: semantic_change
  impact_analysis_required: true
  downstream_assets:
    - postgres://analytics/enforcement/public.case_summary.current_status
    - dashboard://superset/enforcement-case-overview
    - report://regulatory/monthly-enforcement-summary/v2
  consumer_approval_required: true
  backward_compatible: true

Because risk_score was HIGH.

CASE-123 was escalated by rule bundle POLICY-2026-06.14,
rule RISK_ESCALATION_07,
at 2026-07-05T09:21:33Z.
The rule evaluated these inputs:
- case_profile version 27
- evidence_document DOC-991 checksum sha256:...
- sanctions_screening SCR-77 version 5
- prior_enforcement_history snapshot LSN 42/BF9E2A10
The escalation was generated by service risk-engine version git:abc123.
Validation passed and the transition was recorded in case_status_transition id ...
The decision context is preserved in evidence_package EPK-...

CREATE TABLE case_decision_input_ref (
  decision_input_ref_id uuid PRIMARY KEY,
  decision_id           uuid NOT NULL REFERENCES decision_record(decision_id),
  source_asset_urn      text NOT NULL,
  source_entity_key     text NOT NULL,
  source_version        text NOT NULL,
  source_checksum       text,
  input_role            text NOT NULL,
  captured_at           timestamptz NOT NULL DEFAULT now()
);