fix(witan): task_claim backs off between CAS retries, reports contention instead of raising raw

mcp/servers/witan/CHANGELOG.md

@@ -6,6 +6,29 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/) (pre-1.0:
 a MINOR bump may include breaking changes).
 
+## [0.17.4] - 2026-08-18
+
+### Fixed
+
+- **`task_claim` no longer leaks a raw omnigraph conflict to callers under
+  write contention.** Fixes
+  tk-task-claim-exhausts-its-3-attempt-no-backoff-cas-674414: the CAS retry
+  loop fired 3 immediate, unbacked-off attempts and, on exhaustion, re-raised
+  the raw `OmnigraphConflict` — surfacing omnigraph's internal "write
+  authority ... changed during preparation" text straight through the MCP
+  boundary whenever an unrelated write kept colliding on a hot table (most
+  often `node:Task`, written by every claim/update/close across every
+  session). Widened the retry budget to 5 attempts, added jittered backoff
+  between them, and report exhaustion as a structured `{"claimed": false,
+  "reason": "contention"}` instead.
+- **A CAS retry no longer risks resurrecting a task that closed mid-retry.**
+  `_update_task`'s merge sets `status` from the caller's `claim` dict
+  unconditionally, regardless of what its own fresh read shows, so a retry
+  that didn't revalidate first could silently revert a close (or a new
+  block) committed during the backoff window back to `in_progress`. The
+  post-conflict re-read now checks for `closed`/`blocked` and reports that
+  reason instead of ever looping back into a write that would stomp it.
+
 ## [0.17.3] - 2026-08-18
 
 ### Fixed

mcp/servers/witan/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "witan-council"
-version = "0.17.3"
+version = "0.17.4"
 description = "witan — agent memory, planning, and collaboration graph (work-coordination layer + umbrella CLI)"
 readme = "README.md"
 license = "BSD-3-Clause"
@@ -158,7 +158,7 @@ packages = ["witan"]
 "schema" = "schema"
 
 [tool.bumpversion]
-current_version = "0.17.3"
+current_version = "0.17.4"
 allow_dirty = true
 
 [[tool.bumpversion.files]]

mcp/servers/witan/tests/test_tasks.py

@@ -756,6 +756,153 @@ def flaky_change(*args, surface_conflict=False, **kwargs):
     assert server.task_get(t["slug"])["assignee"] == "agentB"
 
 
+@requires_omnigraph
+def test_claim_conflict_does_not_resurrect_a_closed_task(server, monkeypatch):
+    """A close committed during the CAS retry window must not be reverted back
+    to in_progress by the next retry attempt. `_update_task` merges `status`
+    from `claim` unconditionally regardless of what its own fresh read shows
+    (see its docstring), so a retry that does not revalidate claimability
+    first would silently resurrect a closed task. Review finding on the PR
+    for tk-task-claim-exhausts-its-3-attempt-no-backoff-cas-674414."""
+    from witan import graph as graph_mod
+    from witan import server as srv
+
+    t = server.task_create(title="closed-mid-claim", description="x")
+    real_change = srv.client.change
+    calls = {"n": 0}
+
+    async def no_sleep(_seconds):
+        return None
+
+    monkeypatch.setattr(srv.anyio, "sleep", no_sleep)
+
+    def close_then_conflict(*args, surface_conflict=False, **kwargs):
+        if surface_conflict and calls["n"] == 0:
+            calls["n"] += 1
+            # the task is closed (by a rival, or via task_close) in the
+            # window between our read and our write
+            srv._update_task(
+                t["slug"], {"status": "closed", "closed_at": srv.now_iso()}
+            )
+            raise graph_mod.OmnigraphConflict("stale view")
+        if surface_conflict:
+            raise AssertionError("must not retry the write once the task is closed")
+        return real_change(*args, surface_conflict=surface_conflict, **kwargs)
+
+    monkeypatch.setattr(srv.client, "change", close_then_conflict)
+
+    res = server.task_claim(t["slug"], assignee="agentA")
+
+    assert res == {"slug": t["slug"], "claimed": False, "reason": "closed"}
+    assert calls["n"] == 1
+    assert server.task_get(t["slug"])["status"] == "closed"
+
+
+@requires_omnigraph
+def test_claim_conflict_does_not_reopen_a_blocked_task(server, monkeypatch):
+    """Same regression as the closed-task case, for `blocked`."""
+    from witan import graph as graph_mod
+    from witan import server as srv
+
+    t = server.task_create(title="blocked-mid-claim", description="x")
+    real_change = srv.client.change
+    calls = {"n": 0}
+
+    async def no_sleep(_seconds):
+        return None
+
+    monkeypatch.setattr(srv.anyio, "sleep", no_sleep)
+
+    def block_then_conflict(*args, surface_conflict=False, **kwargs):
+        if surface_conflict and calls["n"] == 0:
+            calls["n"] += 1
+            srv._update_task(t["slug"], {"status": "blocked"})
+            raise graph_mod.OmnigraphConflict("stale view")
+        if surface_conflict:
+            raise AssertionError("must not retry the write once the task is blocked")
+        return real_change(*args, surface_conflict=surface_conflict, **kwargs)
+
+    monkeypatch.setattr(srv.client, "change", block_then_conflict)
+
+    res = server.task_claim(t["slug"], assignee="agentA")
+
+    assert res == {"slug": t["slug"], "claimed": False, "reason": "blocked"}
+    assert calls["n"] == 1
+    assert server.task_get(t["slug"])["status"] == "blocked"
+
+
+@requires_omnigraph
+def test_claim_exhausted_conflicts_report_contention_not_raise(server, monkeypatch):
+    """Every retry attempt hits an OCC conflict from unrelated writes elsewhere
+    on the graph — no rival ever actually holds the task. task_claim must
+    exhaust its retry budget and report a structured ``{"claimed": false,
+    "reason": "contention"}`` — not leak the raw `OmnigraphConflict` (the
+    omnigraph "write authority ... changed during preparation" prose) to the
+    caller. See tk-task-claim-exhausts-its-3-attempt-no-backoff-cas-674414."""
+    from witan import graph as graph_mod
+    from witan import server as srv
+
+    t = server.task_create(title="perpetually-contended", description="x")
+    calls = {"n": 0}
+
+    async def no_sleep(_seconds):
+        return None
+
+    monkeypatch.setattr(srv.anyio, "sleep", no_sleep)
+
+    def always_conflict(*args, surface_conflict=False, **kwargs):
+        if surface_conflict:
+            calls["n"] += 1
+            raise graph_mod.OmnigraphConflict(
+                "write authority 'table_head:node:Task' changed during preparation"
+            )
+        raise AssertionError("unconditional write attempted mid-claim")
+
+    monkeypatch.setattr(srv.client, "change", always_conflict)
+
+    res = server.task_claim(t["slug"], assignee="agentA")
+
+    assert res["claimed"] is False
+    assert res["reason"] == "contention"
+    assert calls["n"] == srv._CLAIM_MAX_ATTEMPTS
+    # left exactly as it started, not half-claimed
+    assert server.task_get(t["slug"])["status"] == "open"
+
+
+@requires_omnigraph
+def test_claim_retries_back_off_between_attempts(server, monkeypatch):
+    """A retry after an unrelated conflict must wait, not immediately re-fire —
+    three back-to-back attempts have no chance against multi-second write
+    contention. See tk-task-claim-exhausts-its-3-attempt-no-backoff-cas-674414."""
+    from witan import graph as graph_mod
+    from witan import server as srv
+
+    t = server.task_create(title="briefly-contended", description="x")
+    real_change = srv.client.change
+    calls = {"n": 0}
+    sleeps = []
+
+    async def recording_sleep(seconds):
+        sleeps.append(seconds)
+
+    monkeypatch.setattr(srv.anyio, "sleep", recording_sleep)
+
+    def flaky_once(*args, surface_conflict=False, **kwargs):
+        if surface_conflict and calls["n"] == 0:
+            calls["n"] += 1
+            raise graph_mod.OmnigraphConflict("stale view")
+        return real_change(*args, surface_conflict=surface_conflict, **kwargs)
+
+    monkeypatch.setattr(srv.client, "change", flaky_once)
+
+    res = server.task_claim(t["slug"], assignee="agentA")
+
+    assert res["claimed"] is True
+    # attempt 1's backoff: base delay plus up to 10% jitter (see _claim_backoff)
+    assert len(sleeps) == 1
+    assert srv._CLAIM_BACKOFF_BASE <= sleeps[0] <= srv._CLAIM_BACKOFF_BASE * 1.1
+
+
 # ── conditional claims (omnigraph #470 compare-and-swap) ─────────────
 
 

mcp/servers/witan/witan/server.py

@@ -6,6 +6,7 @@
 import json
 import math
 import os
+import random
 import re
 import shutil
 import subprocess
@@ -695,9 +696,31 @@ def _claim_remedy(slug: str, held_by: str, lease_started_at: str | None) -> str:
 
 # Bounded re-tries for the best-effort CAS claim loop: on each surfaced
 # optimistic-concurrency conflict we re-read and either bail (a rival won) or
-# re-attempt the claim. Small, since a claim conflicting this many times in a row
-# without a rival taking it is pathological.
-_CLAIM_MAX_ATTEMPTS = 3
+# re-attempt the claim. Was 3 with no backoff between attempts — three
+# back-to-back immediate re-attempts have no chance against the contention
+# windows tk-the-write-gate-is-sized-against-a-3-45s-solo-wri-73fc2b measured
+# (loaded writes taking 17-44s), so an unrelated conflict on a hot table (e.g.
+# node:Task, written by every claim/update/close across every session) would
+# exhaust this budget and escape as a raw OmnigraphConflict — see
+# tk-task-claim-exhausts-its-3-attempt-no-backoff-cas-674414. Widened and
+# paired with a short jittered backoff below; still deliberately short next to
+# the 30s call deadline, since a real fix for multi-second contention is the
+# write-gate/batching work, not a longer spin here.
+_CLAIM_MAX_ATTEMPTS = 5
+_CLAIM_BACKOFF_BASE = 0.25
+_CLAIM_BACKOFF_CAP = 3.0
+
+
+def _claim_backoff(attempt: int) -> float:
+    """Jittered exponential backoff between CAS retry attempts.
+
+    Jitter matters here specifically because the trigger case is a burst of
+    concurrent claimers on the same task/table — lockstep exponential backoff
+    would just have them collide again on the next attempt.
+    """
+    delay = _CLAIM_BACKOFF_BASE * (2 ** (attempt - 1))
+    jitter = random.uniform(0, 0.1 * delay)
+    return min(delay + jitter, _CLAIM_BACKOFF_CAP)
 
 # Bounds for task_claim's post-write verification catch-up loop (see the
 # comment at the call site) — sized above the largest staleness gap actually
@@ -4947,7 +4970,9 @@ async def task_claim(
     The coordination primitive for parallel/multi-user agents — call it before
     starting a ready task so others see it is taken. Returns ``{"claimed": true,
     …}`` on success, or ``{"claimed": false, "reason": …}`` when the task is
-    closed, still blocked, or held by someone else. The lease (``claimed_at``)
+    closed, still blocked, held by someone else (``"held"``/``"lost_race"``),
+    or the graph is too busy to complete the CAS after retrying
+    (``"contention"`` — safe and expected to retry). The lease (``claimed_at``)
     expires if the holder never closes/releases, making the task reclaimable
     (see ``task_ready``); re-calling renews it. Pass ``force`` to steal a live
     claim.
@@ -5076,10 +5101,25 @@ async def task_claim(
             # A concurrent writer committed between our read and our write.
             rows = await _offload(client.read, "read.gq", "get_task", {"slug": slug})
             fresh = rows[0] if rows else {}
+            fresh_status = fresh.get("status")
+            # ★ REVALIDATE CLAIMABILITY BEFORE EVER RETRYING, not just before
+            # giving up. `_update_task`'s merge sets `status` from `claim`
+            # unconditionally (see its docstring) — it does not care what the
+            # fresh row it reads says — so a retry that goes ahead while the
+            # task is now closed/blocked would silently resurrect it to
+            # in_progress rather than conflict again. That window used to be
+            # a bare network round-trip; the backoff below (and the wider
+            # attempt budget) makes it wide enough to matter. Bail with the
+            # real reason instead of looping back into a write that would
+            # stomp it.
+            if fresh_status == "closed":
+                return {"slug": slug, "claimed": False, "reason": "closed"}
+            if fresh_status == "blocked":
+                return {"slug": slug, "claimed": False, "reason": "blocked"}
             rival = fresh.get("assignee")
             fresh_lease_started_at = fresh.get("claimed_at") or fresh.get("updated_at")
             if (
-                fresh.get("status") == "in_progress"
+                fresh_status == "in_progress"
                 and rival != holder
                 and not _lease_expired(fresh_lease_started_at)
                 and not force
@@ -5094,10 +5134,39 @@ async def task_claim(
                         slug, rival or _UNKNOWN_HOLDER, fresh_lease_started_at
                     ),
                 }
-            # The conflict was unrelated (or the rival's lease has lapsed) —
-            # retry the claim now that the manifest has advanced.
+            # The task itself still looks claimable, so the conflict came
+            # from something else on the branch (an unrelated write, a rival
+            # claim already released, a same-holder update, or — with
+            # `force=True` — a still-live rival we intend to steal anyway).
+            # The branch-head precondition does not tell us which; back off
+            # and retry now that the manifest has advanced.
             if attempt + 1 == _CLAIM_MAX_ATTEMPTS:
-                raise
+                # Exhausted the budget without ever seeing a rival hold the
+                # task or the task become closed/blocked. Report it as a
+                # retryable condition rather than leaking the raw omnigraph
+                # "write authority ... changed during preparation" text to
+                # the caller (tk-task-claim-exhausts-its-3-attempt-no-backoff-
+                # cas-674414).
+                logger.warning(
+                    "witan.task_claim.contention_exhausted",
+                    task_slug=slug,
+                    holder=holder,
+                    attempts=_CLAIM_MAX_ATTEMPTS,
+                )
+                return {
+                    "slug": slug,
+                    "claimed": False,
+                    "reason": "contention",
+                    "remedy": (
+                        f"{_CLAIM_MAX_ATTEMPTS} claim attempts each hit a "
+                        "write conflict on the graph without the task itself "
+                        "becoming closed, blocked, or held by a rival — most "
+                        "likely heavy write load elsewhere on the graph, "
+                        "though the branch-head precondition cannot fully "
+                        "rule out a same-task race. Retry task_claim."
+                    ),
+                }
+            await anyio.sleep(_claim_backoff(attempt + 1))
 
     # Post-write verification: with no store-level CAS, a rival's claim could
     # still have landed last. Re-read and confirm we actually hold it before