fix: claim-then-post run-level failure dedup; stop permanent suppression

agent/completion.py

@@ -7,9 +7,12 @@ short failure reply to the originating channel, so a run that died on a server
 recycle or hit a limit never leaves the user in silence.
 
 This decouples "the user gets an answer" from "the agent remembered to reply."
-The reply is idempotent per run when the webhook includes a run id. Older or
-manual payloads without a run id fall back to legacy thread-level idempotence so
-missing ids degrade dedupe instead of silencing failure replies.
+The reply is idempotent per run: the run id (or, for manual payloads without
+one, a run-distinguishing ``updated_at``/``created_at`` marker) is claimed in a
+bounded per-thread list *before* the post (claim-then-post), so a retried or
+concurrent completion webhook can't double-post. A payload that carries nothing
+to distinguish one run from another posts without deduping — a rare duplicate is
+preferred over silencing a later, genuinely different failed run.
 """
 
 from __future__ import annotations
@@ -17,6 +20,7 @@ from __future__ import annotations
 import hmac
 import logging
 import os
+from collections.abc import Awaitable, Callable
 from typing import Any
 
 from .utils.dashboard_links import dashboard_thread_url
@@ -33,11 +37,15 @@ logger = logging.getLogger(__name__)
 # follow-up halts the prior run (status "interrupted") while its replacement
 # carries on — that's healthy, not a failure worth a "couldn't finish" reply.
 _TERMINAL_FAILURE_STATUSES = frozenset({"error", "timeout"})
-_FAILURE_REPLY_FLAG = "failure_reply_posted"
 _FAILURE_REPLY_RUN_ID = "failure_reply_posted_run_id"
 _FAILURE_REPLY_RUN_IDS = "failure_reply_posted_run_ids"
 _MAX_FAILURE_REPLY_RUN_IDS = 20
 
+
+class _ClaimFailed(Exception):
+    """Raised when the dedup key couldn't be claimed, so we skip the post."""
+
+
 # Shared-secret bearer token proving a /webhooks/run-complete call came from our
 # own dispatch (which appends ?token= when this is set) rather than from an
 # attacker hitting the public route. Fail closed when unset: the route rejects
@@ -78,8 +86,19 @@ def _failure_text(status: str, dashboard_url: str | None = None) -> str:
     return text
 
 
-async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status: str) -> bool:
-    """Post a failure reply to the run's originating channel. Best-effort."""
+async def _post_failure_reply(
+    thread_id: str,
+    metadata: dict[str, Any],
+    status: str,
+    *,
+    claim: Callable[[], Awaitable[None]],
+) -> bool:
+    """Post a failure reply to the run's originating channel. Best-effort.
+
+    ``claim`` is awaited immediately before the network post (claim-then-post),
+    only on a branch that actually delivers, so a retried/concurrent webhook
+    can't double-post and threads with no reply channel never burn the claim.
+    """
     source = metadata.get("source")
     ctx = metadata.get("source_context")
     ctx = ctx if isinstance(ctx, dict) else {}
@@ -91,6 +110,7 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
             channel_id = slack_thread.get("channel_id")
             thread_ts = slack_thread.get("thread_ts")
             if channel_id and thread_ts:
+                await claim()
                 slack_text = _failure_text(status, dashboard_thread_url(thread_id))
                 return await post_slack_thread_reply(channel_id, thread_ts, slack_text)
         return False
@@ -100,6 +120,7 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
         if isinstance(linear_issue, dict):
             issue_id = linear_issue.get("id")
             if issue_id:
+                await claim()
                 return await comment_on_linear_issue(issue_id, text)
         return False
 
@@ -113,6 +134,7 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
         if isinstance(repo_config, dict) and isinstance(number, int):
             token = await get_github_app_installation_token()
             if token:
+                await claim()
                 return await post_github_comment(repo_config, number, text, token=token)
         return False
 
@@ -120,6 +142,25 @@ async def _post_failure_reply(thread_id: str, metadata: dict[str, Any], status:
     return False
 
 
+def _dedup_key(payload: dict[str, Any]) -> str | None:
+    """A run-distinguishing key for dedupe, or None when nothing distinguishes runs.
+
+    Prefers the platform's ``run_id`` (always present on real completion
+    webhooks). Manual/legacy payloads without one fall back to a marker built
+    from ``updated_at``/``created_at`` so a retry of the *same* run still dedupes
+    while a *different* failed run on the same thread still gets its reply — the
+    old sticky per-thread boolean silenced every later run forever (SR160-03).
+    """
+    raw = payload.get("run_id")
+    if isinstance(raw, str) and raw:
+        return raw
+    for marker in ("updated_at", "created_at"):
+        value = payload.get(marker)
+        if isinstance(value, str) and value:
+            return f"{marker}:{value}"
+    return None
+
+
 def _posted_failure_run_ids(metadata: dict[str, Any]) -> list[str]:
     raw = metadata.get(_FAILURE_REPLY_RUN_IDS)
     ids = [item for item in raw if isinstance(item, str) and item] if isinstance(raw, list) else []
@@ -129,13 +170,11 @@ def _posted_failure_run_ids(metadata: dict[str, Any]) -> list[str]:
     return ids
 
 
-def _failure_reply_metadata(metadata: dict[str, Any], run_id: str | None) -> dict[str, Any]:
-    if run_id is None:
-        return {_FAILURE_REPLY_FLAG: True}
-    ids = [item for item in _posted_failure_run_ids(metadata) if item != run_id]
-    ids.append(run_id)
+def _failure_reply_metadata(metadata: dict[str, Any], key: str) -> dict[str, Any]:
+    ids = [item for item in _posted_failure_run_ids(metadata) if item != key]
+    ids.append(key)
     return {
-        _FAILURE_REPLY_RUN_ID: run_id,
+        _FAILURE_REPLY_RUN_ID: key,
         _FAILURE_REPLY_RUN_IDS: ids[-_MAX_FAILURE_REPLY_RUN_IDS:],
     }
 
@@ -148,8 +187,6 @@ async def handle_run_completion(payload: dict[str, Any]) -> dict[str, str]:
     """
     status = payload.get("status")
     thread_id = payload.get("thread_id")
-    raw_run_id = payload.get("run_id")
-    run_id = raw_run_id if isinstance(raw_run_id, str) and raw_run_id else None
     if not isinstance(thread_id, str) or not thread_id:
         return {"status": "ignored", "reason": "missing thread_id"}
     if status not in _TERMINAL_FAILURE_STATUSES:
@@ -164,24 +201,35 @@ async def handle_run_completion(payload: dict[str, Any]) -> dict[str, str]:
 
     metadata = thread.get("metadata") if isinstance(thread, dict) else None
     metadata = metadata if isinstance(metadata, dict) else {}
-    if run_id is None:
-        # Payloads without run ids fall back to the old per-thread flag; run-scoped
-        # dedupe intentionally does not read it so future runs can still report.
-        if metadata.get(_FAILURE_REPLY_FLAG):
-            return {"status": "ignored", "reason": "failure reply already posted"}
-    elif run_id in _posted_failure_run_ids(metadata):
+    key = _dedup_key(payload)
+    if key is not None and key in _posted_failure_run_ids(metadata):
         return {"status": "ignored", "reason": "failure reply already posted for run"}
 
-    posted = await _post_failure_reply(thread_id, metadata, status)
-    if not posted:
-        return {"status": "ignored", "reason": "no reply posted"}
+    # Claim-then-post: record the dedup key immediately before the actual post
+    # (via the claim callback) so a retried/concurrent completion webhook can't
+    # double-post (SR160-02). The key is only claimed on a branch that delivers,
+    # so a thread with no reply channel never burns it. If the claim itself fails
+    # we raise so we skip the post, leaving the key unclaimed for a later retry
+    # rather than reporting a clean success that invites a duplicate. A payload
+    # with no distinguishing key can't be recorded; it posts un-deduped rather
+    # than being permanently suppressed.
+    async def _claim() -> None:
+        if key is None:
+            return
+        try:
+            await client.threads.update(
+                thread_id=thread_id,
+                metadata=_failure_reply_metadata(metadata, key),
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("run-complete: could not flag thread %s", thread_id, exc_info=True)
+            raise _ClaimFailed from exc
 
     try:
-        await client.threads.update(
-            thread_id=thread_id,
-            metadata=_failure_reply_metadata(metadata, run_id),
-        )
-    except Exception:  # noqa: BLE001
-        logger.warning("run-complete: could not flag thread %s", thread_id, exc_info=True)
+        posted = await _post_failure_reply(thread_id, metadata, status, claim=_claim)
+    except _ClaimFailed:
+        return {"status": "error", "reason": "could not claim failure reply"}
+    if not posted:
+        return {"status": "ignored", "reason": "no reply posted"}
     logger.info("Posted failure reply for thread %s (status=%s)", thread_id, status)
     return {"status": "ok", "reason": "failure reply posted"}

tests/test_completion_webhook.py

@@ -9,20 +9,27 @@ from agent import completion
 
 
 class _FakeThreads:
-    def __init__(self, metadata: dict[str, Any]) -> None:
+    def __init__(self, metadata: dict[str, Any], *, fail_updates: int = 0) -> None:
         self._metadata = metadata
         self.updates: list[dict[str, Any]] = []
+        self._fail_updates = fail_updates
 
     async def get(self, thread_id: str) -> dict[str, Any]:
-        return {"thread_id": thread_id, "metadata": self._metadata}
+        # Reflect prior claims so a duplicate/retried delivery sees them, exactly
+        # as the platform persists thread metadata across webhook deliveries.
+        return {"thread_id": thread_id, "metadata": dict(self._metadata)}
 
     async def update(self, *, thread_id: str, metadata: dict[str, Any]) -> None:
+        if self._fail_updates > 0:
+            self._fail_updates -= 1
+            raise RuntimeError("simulated metadata write failure")
         self.updates.append(metadata)
+        self._metadata.update(metadata)
 
 
 class _FakeClient:
-    def __init__(self, metadata: dict[str, Any]) -> None:
-        self.threads = _FakeThreads(metadata)
+    def __init__(self, metadata: dict[str, Any], *, fail_updates: int = 0) -> None:
+        self.threads = _FakeThreads(metadata, fail_updates=fail_updates)
 
 
 def _slack_metadata() -> dict[str, Any]:
@@ -138,9 +145,12 @@ async def test_missing_thread_id_is_ignored() -> None:
 
 
 @pytest.mark.asyncio
-async def test_missing_run_id_falls_back_to_thread_level_dedupe(
+async def test_no_run_id_and_no_marker_posts_without_dedupe(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    # A payload carrying nothing to distinguish one run from another posts
+    # un-deduped (no sticky flag written) rather than being suppressed. It
+    # prefers a rare duplicate over permanent silence (SR160-03).
     client = _FakeClient(_slack_metadata())
     monkeypatch.setattr(completion, "langgraph_client", lambda: client)
     reply = AsyncMock(return_value=True)
@@ -150,13 +160,15 @@ async def test_missing_run_id_falls_back_to_thread_level_dedupe(
 
     assert result["status"] == "ok"
     reply.assert_awaited_once()
-    assert client.threads.updates == [{"failure_reply_posted": True}]
+    assert client.threads.updates == []
 
 
 @pytest.mark.asyncio
-async def test_missing_run_id_respects_thread_level_dedupe(
+async def test_legacy_sticky_flag_does_not_suppress_new_failure(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
+    # An old per-thread sticky boolean left by a prior code version must not
+    # permanently silence a later failed run (SR160-03).
     metadata = _slack_metadata()
     metadata["failure_reply_posted"] = True
     client = _FakeClient(metadata)
@@ -164,11 +176,123 @@ async def test_missing_run_id_respects_thread_level_dedupe(
     reply = AsyncMock(return_value=True)
     monkeypatch.setattr(completion, "post_slack_thread_reply", reply)
 
-    result = await completion.handle_run_completion({"thread_id": "t1", "status": "error"})
+    result = await completion.handle_run_completion(
+        {"thread_id": "t1", "run_id": "run-9", "status": "error"}
+    )
 
-    assert result["status"] == "ignored"
+    assert result["status"] == "ok"
+    reply.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_no_run_id_dedupes_on_updated_at_marker(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Without a run_id the same run's retry (identical updated_at) dedupes...
+    client = _FakeClient(_slack_metadata())
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+    reply = AsyncMock(return_value=True)
+    monkeypatch.setattr(completion, "post_slack_thread_reply", reply)
+
+    first = await completion.handle_run_completion(
+        {"thread_id": "t1", "status": "error", "updated_at": "2026-07-09T00:00:00Z"}
+    )
+    retry = await completion.handle_run_completion(
+        {"thread_id": "t1", "status": "error", "updated_at": "2026-07-09T00:00:00Z"}
+    )
+
+    assert first["status"] == "ok"
+    assert retry["status"] == "ignored"
+    reply.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_no_run_id_different_runs_each_reply(monkeypatch: pytest.MonkeyPatch) -> None:
+    # ...while two genuinely different failed runs (distinct updated_at) on one
+    # thread each get their reply — the crux of SR160-03.
+    client = _FakeClient(_slack_metadata())
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+    reply = AsyncMock(return_value=True)
+    monkeypatch.setattr(completion, "post_slack_thread_reply", reply)
+
+    first = await completion.handle_run_completion(
+        {"thread_id": "t1", "status": "error", "updated_at": "2026-07-09T00:00:00Z"}
+    )
+    second = await completion.handle_run_completion(
+        {"thread_id": "t1", "status": "timeout", "updated_at": "2026-07-09T01:00:00Z"}
+    )
+
+    assert first["status"] == "ok"
+    assert second["status"] == "ok"
+    assert reply.await_count == 2
+
+
+@pytest.mark.asyncio
+async def test_claim_written_before_post(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Claim-then-post: the dedup key is recorded before the network reply fires.
+    client = _FakeClient(_slack_metadata())
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+    order: list[str] = []
+
+    async def _update(*, thread_id: str, metadata: dict[str, Any]) -> None:
+        order.append("claim")
+        client.threads.updates.append(metadata)
+        client.threads._metadata.update(metadata)
+
+    async def _reply(*args: Any, **kwargs: Any) -> bool:
+        order.append("post")
+        return True
+
+    monkeypatch.setattr(client.threads, "update", _update)
+    monkeypatch.setattr(completion, "post_slack_thread_reply", _reply)
+
+    result = await completion.handle_run_completion(
+        {"thread_id": "t1", "run_id": "run-1", "status": "error"}
+    )
+
+    assert result["status"] == "ok"
+    assert order == ["claim", "post"]
+
+
+@pytest.mark.asyncio
+async def test_claim_write_failure_does_not_post_and_retry_posts_once(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # SR160-02: if the claim write throws, we do NOT post and report an error so
+    # the retry re-attempts — the retry then claims and posts exactly once,
+    # rather than the old flow double-posting on retry.
+    client = _FakeClient(_slack_metadata(), fail_updates=1)
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+    reply = AsyncMock(return_value=True)
+    monkeypatch.setattr(completion, "post_slack_thread_reply", reply)
+
+    first = await completion.handle_run_completion(
+        {"thread_id": "t1", "run_id": "run-1", "status": "error"}
+    )
+    assert first["status"] == "error"
     reply.assert_not_called()
-    assert client.threads.updates == []
+
+    retry = await completion.handle_run_completion(
+        {"thread_id": "t1", "run_id": "run-1", "status": "error"}
+    )
+    assert retry["status"] == "ok"
+    reply.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_duplicate_delivery_posts_exactly_once(monkeypatch: pytest.MonkeyPatch) -> None:
+    # A retried/duplicate delivery for one run_id posts exactly once because the
+    # first delivery's claim is persisted before the second reads metadata.
+    client = _FakeClient(_slack_metadata())
+    monkeypatch.setattr(completion, "langgraph_client", lambda: client)
+    reply = AsyncMock(return_value=True)
+    monkeypatch.setattr(completion, "post_slack_thread_reply", reply)
+
+    payload = {"thread_id": "t1", "run_id": "run-1", "status": "error"}
+    first = await completion.handle_run_completion(dict(payload))
+    second = await completion.handle_run_completion(dict(payload))
+
+    assert first["status"] == "ok"
+    assert second["status"] == "ignored"
+    reply.assert_awaited_once()
 
 
 @pytest.mark.asyncio