sync: 188 file da Baida98/AI@c343c9b6 (2026-08-25 20:01 UTC) [deploy-all]

api/webhook.py

@@ -2,10 +2,11 @@
 import os, asyncio
 from typing import Optional
 from fastapi import APIRouter, Depends, HTTPException, Request
-from pydantic import BaseModel, field_validator
+from pydantic import BaseModel, ValidationError, field_validator
 from .state import _get_mem_manager, _get_executor, _get_planner
 from .auth_guard import require_role, AuthRole  # GAP-WEBHOOK-ADMIN-FIX
 from .task_tool_policy import build_task_tool_policy
+from .webhook_security import WebhookDeliveryStore, verify_webhook_request
 
 import logging
 _logger = logging.getLogger("api.webhook")
@@ -201,10 +202,12 @@ class PublicChatPayload(BaseModel):
 
 
 @router.post('/api/webhook/{webhook_token}')
-async def inbound_webhook(webhook_token: str, body: WebhookPayload):
-    """
-    S289 — Webhook inbound per triggering proattivo dell'agente.
-    Auth: WEBHOOK_TOKEN env var deve matchare il path token.
+async def inbound_webhook(webhook_token: str, request: Request):
+    """Inbound webhook con token path, HMAC raw-body e idempotenza fail-closed.
+
+    Il token nel path è un selettore legacy: non è sufficiente a autorizzare il
+    dispatch. JSON, policy e loop sono raggiungibili soltanto dopo firma e
+    freshness validate sul body raw.
     """
     _expected = os.getenv('WEBHOOK_TOKEN', '').strip()
     if not _expected:
@@ -212,14 +215,36 @@ async def inbound_webhook(webhook_token: str, body: WebhookPayload):
     if webhook_token != _expected:
         raise HTTPException(status_code=401, detail='Unauthorized: token non valido')
 
-    _task_policy = build_task_tool_policy(body.goal)
-    if _task_policy.literal_response:
-        return {
-            'ok': True, 'output': _task_policy.literal_response, 'engine': 'policy',
-            'goal': body.goal, 'steps': 0,
-        }
-
+    raw_body = await request.body()
+    verified = verify_webhook_request(raw_body, request.headers)
     try:
+        body = WebhookPayload.model_validate_json(raw_body)
+    except ValidationError as exc:
+        raise HTTPException(status_code=422, detail='Payload webhook non valido') from exc
+
+    # In production this is the service-role Supabase client. The security store
+    # fails closed if the durable RPC migration is unavailable.
+    from api.state import _sb as _supa
+    delivery_store = WebhookDeliveryStore(_supa)
+    claim = await delivery_store.claim(verified)
+    if claim.decision == 'replay' and claim.response is not None:
+        return claim.response
+    if claim.decision == 'conflict':
+        raise HTTPException(status_code=409, detail='X-Webhook-Id riutilizzato con payload diverso')
+    if claim.decision == 'in_progress':
+        raise HTTPException(status_code=409, detail='Webhook già in elaborazione', headers={'Retry-After': '1'})
+
+    execution_started = False
+    try:
+        _task_policy = build_task_tool_policy(body.goal)
+        if _task_policy.literal_response:
+            response = {
+                'ok': True, 'output': _task_policy.literal_response, 'engine': 'policy',
+                'goal': body.goal, 'steps': 0,
+            }
+            await delivery_store.complete(verified, response)
+            return response
+
         from agents.unified_loop import UnifiedAgentLoop
         from models.ai_client import AIClient
         client = AIClient()
@@ -238,6 +263,9 @@ async def inbound_webhook(webhook_token: str, body: WebhookPayload):
         context_str = '\n'.join(m.get('content', '') for m in body.context) if body.context else ''
         _wh_task_id = f"wh_{webhook_token[:6]}_{int(__import__('time').time()*1000)%999983:x}"
         await _tg_start(_wh_task_id, body.goal)
+        # Da qui il loop può produrre side effect: un errore conserva il claim
+        # fino al TTL invece di consentire una nuova esecuzione duplicata.
+        execution_started = True
         try:
             result = await asyncio.wait_for(
                 loop.run(goal=body.goal, context=context_str,
@@ -253,17 +281,23 @@ async def inbound_webhook(webhook_token: str, body: WebhookPayload):
             raise HTTPException(status_code=500, detail=f'Errore agente: {exc}')
         _out = result.get('output', '')
         asyncio.ensure_future(_tg_done(_wh_task_id, body.goal, _out[:500]))
-        return {
+        response = {
             'ok':     result.get('success', False),
             'output': _out,
             'engine': result.get('engine',  'unknown'),
             'goal':   body.goal,
             'steps':  len(result.get('steps', [])),
         }
-    except (HTTPException, asyncio.TimeoutError):
+        await delivery_store.complete(verified, response)
+        return response
+    except HTTPException:
+        if not execution_started:
+            await delivery_store.release(verified)
         raise
     except Exception as exc:
-        raise HTTPException(status_code=500, detail=f'Errore agente: {exc}')
+        if not execution_started:
+            await delivery_store.release(verified)
+        raise HTTPException(status_code=500, detail=f'Errore agente: {exc}') from exc
 
 
 @router.post('/api/public/chat')

api/webhook_security.py

@@ -0,0 +1,193 @@
+"""Fail-closed security primitives for the generic inbound webhook.
+
+The authenticated byte sequence is exactly ``v1.<timestamp>.<event_id>.`` + raw body.
+No JSON parsing or task dispatch is allowed before signature and freshness checks pass.
+"""
+from __future__ import annotations
+
+import asyncio
+from dataclasses import dataclass
+import hashlib
+import hmac
+import json
+import os
+import re
+import time
+from typing import Any, Mapping
+
+from fastapi import HTTPException
+
+_EVENT_ID_RE = re.compile(r"^[A-Za-z0-9._-]{16,128}$")
+_SHA256_HEX_RE = re.compile(r"^[0-9a-f]{64}$")
+
+
+@dataclass(frozen=True)
+class VerifiedWebhook:
+    event_id: str
+    timestamp: int
+    payload_sha256: str
+
+
+@dataclass(frozen=True)
+class DeliveryClaim:
+    decision: str  # claimed | replay | conflict | in_progress
+    response: dict[str, Any] | None = None
+
+
+def _env_positive_int(name: str, default: int, maximum: int) -> int:
+    try:
+        value = int(os.getenv(name, str(default)))
+    except ValueError:
+        return default
+    return max(1, min(value, maximum))
+
+
+def _signing_bytes(timestamp: str, event_id: str, raw_body: bytes) -> bytes:
+    return b"v1." + timestamp.encode("ascii") + b"." + event_id.encode("ascii") + b"." + raw_body
+
+
+def verify_webhook_request(
+    raw_body: bytes,
+    headers: Mapping[str, str],
+    *,
+    now: int | None = None,
+) -> VerifiedWebhook:
+    """Validate required headers and HMAC before parsing the JSON body."""
+    secret = os.getenv("WEBHOOK_HMAC_SECRET", "")
+    if len(secret) < 32:
+        raise HTTPException(status_code=503, detail="Webhook HMAC non configurato")
+
+    max_body = _env_positive_int("WEBHOOK_MAX_BODY_BYTES", 262_144, 1_048_576)
+    if not raw_body or len(raw_body) > max_body:
+        raise HTTPException(status_code=413, detail="Payload webhook non valido o troppo grande")
+
+    event_id = (headers.get("x-webhook-id") or "").strip()
+    timestamp_raw = (headers.get("x-webhook-timestamp") or "").strip()
+    signature = (headers.get("x-webhook-signature") or "").strip().lower()
+    if not _EVENT_ID_RE.fullmatch(event_id):
+        raise HTTPException(status_code=400, detail="X-Webhook-Id non valido")
+    if not timestamp_raw.isdigit() or len(timestamp_raw) > 12:
+        raise HTTPException(status_code=400, detail="X-Webhook-Timestamp non valido")
+    if not signature.startswith("sha256=") or not _SHA256_HEX_RE.fullmatch(signature[7:]):
+        raise HTTPException(status_code=401, detail="Firma webhook non valida")
+
+    timestamp = int(timestamp_raw)
+    current = int(time.time()) if now is None else int(now)
+    max_age = _env_positive_int("WEBHOOK_MAX_AGE_SECONDS", 300, 3_600)
+    if abs(current - timestamp) > max_age:
+        raise HTTPException(status_code=401, detail="Timestamp webhook scaduto o non valido")
+
+    expected = hmac.new(
+        secret.encode("utf-8"), _signing_bytes(timestamp_raw, event_id, raw_body), hashlib.sha256
+    ).hexdigest()
+    if not hmac.compare_digest(expected, signature[7:]):
+        raise HTTPException(status_code=401, detail="Firma webhook non valida")
+
+    return VerifiedWebhook(
+        event_id=event_id,
+        timestamp=timestamp,
+        payload_sha256=hashlib.sha256(raw_body).hexdigest(),
+    )
+
+
+class WebhookDeliveryStore:
+    """Atomic delivery state backed by Supabase RPC or explicit test-only memory."""
+
+    _memory_lock = asyncio.Lock()
+    _memory: dict[str, dict[str, Any]] = {}
+
+    def __init__(self, supabase_client: Any | None) -> None:
+        self._supabase = supabase_client
+        self._require_durable = os.getenv("WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE", "true").lower() not in {"0", "false", "no"}
+        self._ttl_seconds = _env_positive_int("WEBHOOK_IDEMPOTENCY_TTL_SECONDS", 86_400, 604_800)
+
+    @staticmethod
+    def _rpc_data(result: Any) -> dict[str, Any]:
+        data = getattr(result, "data", result)
+        if isinstance(data, list):
+            data = data[0] if data else None
+        if not isinstance(data, dict):
+            raise RuntimeError("invalid webhook idempotency RPC response")
+        return data
+
+    async def claim(self, verified: VerifiedWebhook) -> DeliveryClaim:
+        if self._supabase is not None:
+            try:
+                result = self._supabase.rpc("claim_webhook_delivery", {
+                    "p_event_id": verified.event_id,
+                    "p_payload_sha256": verified.payload_sha256,
+                    "p_ttl_seconds": self._ttl_seconds,
+                }).execute()
+                data = self._rpc_data(result)
+                decision = str(data.get("decision", ""))
+                if decision not in {"claimed", "replay", "conflict", "in_progress"}:
+                    raise RuntimeError("unknown webhook idempotency decision")
+                response = data.get("response")
+                return DeliveryClaim(decision, response if isinstance(response, dict) else None)
+            except Exception as exc:
+                raise HTTPException(status_code=503, detail="Store idempotency webhook non disponibile") from exc
+
+        if self._require_durable:
+            raise HTTPException(status_code=503, detail="Store idempotency webhook non configurato")
+
+        now = time.monotonic()
+        async with self._memory_lock:
+            expired = [key for key, value in self._memory.items() if value["expires_at"] <= now]
+            for key in expired:
+                self._memory.pop(key, None)
+            existing = self._memory.get(verified.event_id)
+            if existing is None:
+                self._memory[verified.event_id] = {
+                    "payload_sha256": verified.payload_sha256,
+                    "status": "processing",
+                    "response": None,
+                    "expires_at": now + self._ttl_seconds,
+                }
+                return DeliveryClaim("claimed")
+            if existing["payload_sha256"] != verified.payload_sha256:
+                return DeliveryClaim("conflict")
+            if existing["status"] == "completed":
+                return DeliveryClaim("replay", existing["response"])
+            return DeliveryClaim("in_progress")
+
+    async def complete(self, verified: VerifiedWebhook, response: dict[str, Any]) -> None:
+        if self._supabase is not None:
+            try:
+                self._supabase.rpc("complete_webhook_delivery", {
+                    "p_event_id": verified.event_id,
+                    "p_payload_sha256": verified.payload_sha256,
+                    "p_response": response,
+                }).execute()
+                return
+            except Exception as exc:
+                raise HTTPException(status_code=503, detail="Store idempotency webhook non disponibile") from exc
+
+        if self._require_durable:
+            raise HTTPException(status_code=503, detail="Store idempotency webhook non configurato")
+        async with self._memory_lock:
+            entry = self._memory.get(verified.event_id)
+            if entry and entry["payload_sha256"] == verified.payload_sha256:
+                entry["status"] = "completed"
+                entry["response"] = json.loads(json.dumps(response))
+
+    async def release(self, verified: VerifiedWebhook) -> None:
+        if self._supabase is not None:
+            try:
+                self._supabase.rpc("release_webhook_delivery", {
+                    "p_event_id": verified.event_id,
+                    "p_payload_sha256": verified.payload_sha256,
+                }).execute()
+                return
+            except Exception:
+                return
+        if self._require_durable:
+            return
+        async with self._memory_lock:
+            entry = self._memory.get(verified.event_id)
+            if entry and entry["payload_sha256"] == verified.payload_sha256 and entry["status"] == "processing":
+                self._memory.pop(verified.event_id, None)
+
+    @classmethod
+    async def reset_memory_for_test(cls) -> None:
+        async with cls._memory_lock:
+            cls._memory.clear()

tests/test_entrypoint_policy_contracts.py

@@ -7,12 +7,18 @@ prima di provider, notifiche, planner e tool.
 from __future__ import annotations
 
 import asyncio
+import hashlib
+import hmac
+import json
 import os
+import time
 import sys
 import types
 import unittest
 from unittest.mock import patch
 
+from fastapi import HTTPException
+
 _BACKEND = os.path.join(os.path.dirname(__file__), "..")
 if _BACKEND not in sys.path:
     sys.path.insert(0, _BACKEND)
@@ -30,6 +36,15 @@ class _Request:
         self.headers = headers or {}
 
 
+class _WebhookRequest(_Request):
+    def __init__(self, raw_body: bytes, headers: dict[str, str]) -> None:
+        super().__init__(headers)
+        self._raw_body = raw_body
+
+    async def body(self) -> bytes:
+        return self._raw_body
+
+
 class _CallbackRequest(_Request):
     async def json(self) -> dict[str, object]:
         raise AssertionError("bad secret must not parse payload")
@@ -71,9 +86,29 @@ async def _noop(*_args: object, **_kwargs: object) -> None:
     return None
 
 
+def _signed_webhook_request(
+    payload: dict[str, object],
+    *,
+    event_id: str = "evt-000000000001",
+    timestamp: int | None = None,
+    signature: str | None = None,
+) -> _WebhookRequest:
+    raw = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+    ts = str(int(time.time()) if timestamp is None else timestamp)
+    signing = b"v1." + ts.encode("ascii") + b"." + event_id.encode("ascii") + b"." + raw
+    expected = hmac.new(b"h" * 32, signing, hashlib.sha256).hexdigest()
+    return _WebhookRequest(raw, {
+        "x-webhook-id": event_id,
+        "x-webhook-timestamp": ts,
+        "x-webhook-signature": signature or f"sha256={expected}",
+    })
+
+
 class EntryPointPolicyContracts(unittest.IsolatedAsyncioTestCase):
-    def setUp(self) -> None:
+    async def asyncSetUp(self) -> None:
+        from api.webhook_security import WebhookDeliveryStore
         _RecordingLoop.calls.clear()
+        await WebhookDeliveryStore.reset_memory_for_test()
 
     async def test_telegram_callback_fails_closed_before_payload_or_outbound_work(self) -> None:
         from api import webhook
@@ -85,12 +120,19 @@ class EntryPointPolicyContracts(unittest.IsolatedAsyncioTestCase):
 
     async def test_webhook_literal_contract_returns_before_loop_or_notification(self) -> None:
         from api import webhook
+        import api.state as state
 
-        payload = webhook.WebhookPayload(goal=LITERAL_PROMPT)
-        with patch.dict(os.environ, {"WEBHOOK_TOKEN": "test-webhook-token"}, clear=False), \
+        request = _signed_webhook_request({"goal": LITERAL_PROMPT})
+        env = {
+            "WEBHOOK_TOKEN": "test-webhook-token",
+            "WEBHOOK_HMAC_SECRET": "h" * 32,
+            "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false",
+        }
+        with patch.dict(os.environ, env, clear=False), \
+             patch.object(state, "_sb", None), \
              patch.object(webhook, "_tg_start", side_effect=AssertionError("notification forbidden")), \
              patch.object(webhook, "_get_mem_manager", side_effect=AssertionError("provider forbidden")):
-            result = await webhook.inbound_webhook("test-webhook-token", payload)
+            result = await webhook.inbound_webhook("test-webhook-token", request)
 
         self.assertEqual(result["output"], "TEST_E2E_OK")
         self.assertEqual(result["steps"], 0)
@@ -114,17 +156,25 @@ class EntryPointPolicyContracts(unittest.IsolatedAsyncioTestCase):
 
     async def test_webhook_and_public_chat_propagate_no_tool_policy_to_loop(self) -> None:
         from api import webhook
+        import api.state as state
 
         fake_modules = _fake_loop_modules()
+        env = {
+            "WEBHOOK_TOKEN": "test-webhook-token",
+            "PUBLIC_API_TOKEN": "public-test-token",
+            "WEBHOOK_HMAC_SECRET": "h" * 32,
+            "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false",
+        }
         with patch.dict(sys.modules, fake_modules), \
-             patch.dict(os.environ, {"WEBHOOK_TOKEN": "test-webhook-token", "PUBLIC_API_TOKEN": "public-test-token"}, clear=False), \
+             patch.dict(os.environ, env, clear=False), \
+             patch.object(state, "_sb", None), \
              patch.object(webhook, "_get_mem_manager", return_value=object()), \
              patch.object(webhook, "_get_executor", return_value=object()), \
              patch.object(webhook, "_get_planner", return_value=object()), \
              patch.object(webhook, "_tg_start", _noop), \
              patch.object(webhook, "_tg_done", _noop):
             webhook_result = await webhook.inbound_webhook(
-                "test-webhook-token", webhook.WebhookPayload(goal=NO_TOOL_PROMPT)
+                "test-webhook-token", _signed_webhook_request({"goal": NO_TOOL_PROMPT})
             )
             public_result = await webhook.public_chat(
                 webhook.PublicChatPayload(message=NO_TOOL_PROMPT),
@@ -136,6 +186,44 @@ class EntryPointPolicyContracts(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(len(_RecordingLoop.calls), 2)
         self.assertTrue(all(call["allow_tools"] is False for call in _RecordingLoop.calls))
 
+    async def test_webhook_replay_conflict_and_expired_timestamp_are_rejected_before_dispatch(self) -> None:
+        from api import webhook
+        import api.state as state
+
+        env = {
+            "WEBHOOK_TOKEN": "test-webhook-token",
+            "WEBHOOK_HMAC_SECRET": "h" * 32,
+            "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false",
+            "WEBHOOK_MAX_AGE_SECONDS": "300",
+        }
+        first = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-replay-000001")
+        duplicate = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-replay-000001")
+        conflict = _signed_webhook_request({"goal": LITERAL_PROMPT, "context": [{"content": "different"}]}, event_id="evt-replay-000001")
+        expired = _signed_webhook_request({"goal": LITERAL_PROMPT}, event_id="evt-expired-0001", timestamp=int(time.time()) - 301)
+        with patch.dict(os.environ, env, clear=False), patch.object(state, "_sb", None):
+            first_response = await webhook.inbound_webhook("test-webhook-token", first)
+            duplicate_response = await webhook.inbound_webhook("test-webhook-token", duplicate)
+            with self.assertRaises(HTTPException) as conflict_error:
+                await webhook.inbound_webhook("test-webhook-token", conflict)
+            with self.assertRaises(HTTPException) as expired_error:
+                await webhook.inbound_webhook("test-webhook-token", expired)
+
+        self.assertEqual(first_response, duplicate_response)
+        self.assertEqual(conflict_error.exception.status_code, 409)
+        self.assertEqual(expired_error.exception.status_code, 401)
+
+    async def test_webhook_bad_signature_is_rejected_before_pydantic_parse(self) -> None:
+        from api import webhook
+
+        request = _signed_webhook_request({"goal": LITERAL_PROMPT}, signature="sha256=" + "0" * 64)
+        env = {"WEBHOOK_TOKEN": "test-webhook-token", "WEBHOOK_HMAC_SECRET": "h" * 32}
+        with patch.dict(os.environ, env, clear=False), \
+             patch.object(webhook.WebhookPayload, "model_validate_json", side_effect=AssertionError("must not parse")):
+            with self.assertRaises(HTTPException) as signature_error:
+                await webhook.inbound_webhook("test-webhook-token", request)
+
+        self.assertEqual(signature_error.exception.status_code, 401)
+
     async def test_scheduler_literal_contract_skips_provider_initialization(self) -> None:
         from api import scheduler
 

tests/test_webhook_security.py

@@ -0,0 +1,123 @@
+"""Contratti di sicurezza del webhook HMAC e della sua idempotenza."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import os
+import sys
+import unittest
+from unittest.mock import patch
+
+from fastapi import HTTPException
+
+_BACKEND = os.path.join(os.path.dirname(__file__), "..")
+if _BACKEND not in sys.path:
+    sys.path.insert(0, _BACKEND)
+
+from api.webhook_security import (  # noqa: E402
+    WebhookDeliveryStore,
+    _signing_bytes,
+    verify_webhook_request,
+)
+
+_SECRET = "s" * 32
+_EVENT_ID = "evt-security-0001"
+_NOW = 1_700_000_000
+
+
+def _headers(raw: bytes, *, event_id: str = _EVENT_ID, timestamp: int = _NOW) -> dict[str, str]:
+    timestamp_raw = str(timestamp)
+    digest = hmac.new(
+        _SECRET.encode("utf-8"),
+        _signing_bytes(timestamp_raw, event_id, raw),
+        hashlib.sha256,
+    ).hexdigest()
+    return {
+        "x-webhook-id": event_id,
+        "x-webhook-timestamp": timestamp_raw,
+        "x-webhook-signature": f"sha256={digest}",
+    }
+
+
+class WebhookSecurityTests(unittest.IsolatedAsyncioTestCase):
+    async def asyncSetUp(self) -> None:
+        await WebhookDeliveryStore.reset_memory_for_test()
+
+    def test_signature_binds_the_exact_raw_bytes_and_fresh_timestamp(self) -> None:
+        raw = b'{"goal":"safe", "context":[]}'
+        with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": _SECRET}, clear=False):
+            verified = verify_webhook_request(raw, _headers(raw), now=_NOW)
+            self.assertEqual(verified.event_id, _EVENT_ID)
+            self.assertEqual(verified.payload_sha256, hashlib.sha256(raw).hexdigest())
+            with self.assertRaises(HTTPException) as altered:
+                verify_webhook_request(b'{"goal":"safe","context":[]}', _headers(raw), now=_NOW)
+            with self.assertRaises(HTTPException) as future:
+                verify_webhook_request(raw, _headers(raw, timestamp=_NOW + 301), now=_NOW)
+
+        self.assertEqual(altered.exception.status_code, 401)
+        self.assertEqual(future.exception.status_code, 401)
+
+    def test_required_headers_and_secret_fail_closed(self) -> None:
+        raw = b'{"goal":"safe"}'
+        with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": "too-short"}, clear=False):
+            with self.assertRaises(HTTPException) as missing_secret:
+                verify_webhook_request(raw, _headers(raw), now=_NOW)
+        with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": _SECRET}, clear=False):
+            malformed = _headers(raw)
+            malformed["x-webhook-id"] = "short"
+            with self.assertRaises(HTTPException) as invalid_id:
+                verify_webhook_request(raw, malformed, now=_NOW)
+            missing_signature = _headers(raw)
+            missing_signature.pop("x-webhook-signature")
+            with self.assertRaises(HTTPException) as unsigned:
+                verify_webhook_request(raw, missing_signature, now=_NOW)
+
+        self.assertEqual(missing_secret.exception.status_code, 503)
+        self.assertEqual(invalid_id.exception.status_code, 400)
+        self.assertEqual(unsigned.exception.status_code, 401)
+
+    async def test_memory_state_machine_caches_replay_conflicts_and_releases_pre_dispatch(self) -> None:
+        raw = b'{"goal":"safe"}'
+        other_raw = b'{"goal":"other"}'
+        with patch.dict(
+            os.environ,
+            {"WEBHOOK_HMAC_SECRET": _SECRET, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "false"},
+            clear=False,
+        ):
+            verified = verify_webhook_request(raw, _headers(raw), now=_NOW)
+            conflicting = verify_webhook_request(other_raw, _headers(other_raw), now=_NOW)
+            store = WebhookDeliveryStore(None)
+            self.assertEqual((await store.claim(verified)).decision, "claimed")
+            self.assertEqual((await store.claim(verified)).decision, "in_progress")
+            self.assertEqual((await store.claim(conflicting)).decision, "conflict")
+            await store.complete(verified, {"ok": True, "output": "cached"})
+            replay = await store.claim(verified)
+            self.assertEqual(replay.decision, "replay")
+            self.assertEqual(replay.response, {"ok": True, "output": "cached"})
+
+            release_id = "evt-release-00001"
+            releasing = verify_webhook_request(raw, _headers(raw, event_id=release_id), now=_NOW)
+            self.assertEqual((await store.claim(releasing)).decision, "claimed")
+            await store.release(releasing)
+            self.assertEqual((await store.claim(releasing)).decision, "claimed")
+
+    async def test_durable_store_is_required_by_default_and_rpc_errors_are_unavailable(self) -> None:
+        raw = b'{"goal":"safe"}'
+        with patch.dict(os.environ, {"WEBHOOK_HMAC_SECRET": _SECRET, "WEBHOOK_IDEMPOTENCY_REQUIRE_DURABLE": "true"}, clear=False):
+            verified = verify_webhook_request(raw, _headers(raw), now=_NOW)
+            with self.assertRaises(HTTPException) as absent_store:
+                await WebhookDeliveryStore(None).claim(verified)
+            with self.assertRaises(HTTPException) as broken_store:
+                await WebhookDeliveryStore(_BrokenSupabase()).claim(verified)
+
+        self.assertEqual(absent_store.exception.status_code, 503)
+        self.assertEqual(broken_store.exception.status_code, 503)
+
+
+class _BrokenSupabase:
+    def rpc(self, *_args: object, **_kwargs: object) -> object:
+        raise RuntimeError("database unavailable")
+
+
+if __name__ == "__main__":
+    unittest.main()