From 873c2d31eaf27ec454ce7482cc8a95974fa1ca2e Mon Sep 17 00:00:00 2001 From: Adam Moussa Date: Thu, 9 Jul 2026 17:08:26 -0400 Subject: [PATCH] fix: claim-then-post run-level failure dedup; stop permanent suppression --- agent/completion.py | 106 ++++++++++++++++------- tests/test_completion_webhook.py | 144 ++++++++++++++++++++++++++++--- 2 files changed, 211 insertions(+), 39 deletions(-) diff --git a/agent/completion.py b/agent/completion.py index 1df81c0c..f2dfee9d 100644 --- a/agent/completion.py +++ b/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"} diff --git a/tests/test_completion_webhook.py b/tests/test_completion_webhook.py index 5aa773ba..867ee84a 100644 --- a/tests/test_completion_webhook.py +++ b/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