From ce934d00a97c678aeddd0066de099b78adbd4b16 Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Mon, 4 May 2026 12:11:51 +0800 Subject: [PATCH] feat(connector): unify replay lifecycle --- connector/platforms/feishu_webhook.py | 16 +- connector/platforms/kakao_webhook.py | 60 +++- connector/platforms/slack_webhook.py | 51 +++- connector/platforms/telegram_polling.py | 58 +++- services/connector_callback_contract.py | 91 +++++- services/connector_replay_lifecycle.py | 266 ++++++++++++++++++ tests/test_connector_callback_contract.py | 63 +++++ tests/test_f44_kakao_bundle.py | 17 ++ tests/test_f59_slack_interactions.py | 39 +++ tests/test_f69_feishu_callbacks.py | 27 ++ tests/test_f72_telegram_topic_delivery.py | 80 +++++- tests/test_r189_connector_replay_lifecycle.py | 73 +++++ 12 files changed, 809 insertions(+), 32 deletions(-) create mode 100644 services/connector_replay_lifecycle.py create mode 100644 tests/test_r189_connector_replay_lifecycle.py diff --git a/connector/platforms/feishu_webhook.py b/connector/platforms/feishu_webhook.py index a219d0f..9596dae 100644 --- a/connector/platforms/feishu_webhook.py +++ b/connector/platforms/feishu_webhook.py @@ -775,9 +775,19 @@ class FeishuWebhookServer: if decision.requires_approval else request.text ) - contract.acknowledge_request(envelope_dict.get("request_id", "")) - response = await self.router.handle(request) - contract.complete_request(envelope_dict.get("request_id", "")) + request_id = str(envelope_dict.get("request_id", "") or "") + contract.acknowledge_request(request_id) + try: + response = await self.router.handle(request) + except Exception: + # IMPORTANT: failures before route completion remain retryable. + # After router.handle returns, the action may already have side effects, + # so completion failures must not release the claim for rerouting. + contract.release_request_retryable( + request_id, reason="feishu_callback_failed_before_commit" + ) + raise + contract.complete_request(request_id) response_text = str(getattr(response, "text", "") or "").strip() or ( "Action processed." ) diff --git a/connector/platforms/kakao_webhook.py b/connector/platforms/kakao_webhook.py index 65a2982..dec9162 100644 --- a/connector/platforms/kakao_webhook.py +++ b/connector/platforms/kakao_webhook.py @@ -28,6 +28,11 @@ from ..contract import CommandRequest, CommandResponse from ..router import CommandRouter from ..security_profile import AllowlistPolicy, ReplayGuard +try: + from services.connector_replay_lifecycle import ConnectorReplayLifecycle +except ImportError: # pragma: no cover + ConnectorReplayLifecycle = None # type: ignore + logger = logging.getLogger(__name__) @@ -99,6 +104,13 @@ class KakaoWebhookServer: window_sec=self.REPLAY_WINDOW_SEC, max_entries=self.NONCE_CACHE_SIZE, ) + if ConnectorReplayLifecycle is None: # pragma: no cover + self._replay_lifecycle = None + else: + self._replay_lifecycle = ConnectorReplayLifecycle( + ttl_sec=self.REPLAY_WINDOW_SEC, + max_entries=self.NONCE_CACHE_SIZE, + ) # S32: Allowlist (soft-deny via AllowlistPolicy primitive) self._user_allowlist = AllowlistPolicy(config.kakao_allowed_users, strict=False) @@ -165,10 +177,24 @@ class KakaoWebhookServer: # We use a hash of the body bytes as the "nonce" for deduplication. # This prevents re-transmitting the exact same request. content_hash = hashlib.sha256(body_bytes).hexdigest() - if not self._replay_guard.check_and_record(content_hash): - logger.warning(f"Replay rejected for Kakao hash: {content_hash}") - # Return 200 to stop Kakao retries - return _make_response(web, status=200, text="OK") + lifecycle_key = f"kakao:webhook:{content_hash}" + if self._replay_lifecycle is None: # pragma: no cover + if not self._replay_guard.check_and_record(content_hash): + logger.warning(f"Replay rejected for Kakao hash: {content_hash}") + return _make_response(web, status=200, text="OK") + else: + claim = self._replay_lifecycle.claim( + lifecycle_key, + metadata={"platform": "kakao"}, + ) + if not claim.accepted: + logger.warning( + "Replay rejected for Kakao hash: %s code=%s state=%s", + content_hash, + claim.code, + claim.record.state, + ) + return _make_response(web, status=200, text="OK") # Normalization # userRequest.user.id is the opaque user ID (botUserKey) @@ -180,6 +206,10 @@ class KakaoWebhookServer: if not sender_id: # Not a valid user request (maybe a ping?) + if self._replay_lifecycle is not None: + self._replay_lifecycle.fail_terminal( + lifecycle_key, reason="invalid_payload_no_user_id" + ) return self._build_error_response("Invalid Payload: No User ID") # S32: Allowlist @@ -208,6 +238,17 @@ class KakaoWebhookServer: try: resp = await self.router.handle(req) + except Exception as e: + # IMPORTANT: router failures happen before Kakao response delivery and + # must remain retryable; successful router returns are never rerouted. + if self._replay_lifecycle is not None: + self._replay_lifecycle.release_retryable( + lifecycle_key, reason="kakao_router_failed_before_commit" + ) + logger.exception(f"Error handling Kakao command: {e}") + return self._build_error_response("Internal Error") + + try: # IMPORTANT: # Router mocks in unit tests may return non-string `.text` values. # Normalize defensively to avoid turning a valid routing flow into @@ -230,13 +271,20 @@ class KakaoWebhookServer: # Skipping complex media upload for F44 scope unless specifically required. if resp_text or buttons: - return self._build_response(resp_text, quick_replies=buttons) + response = self._build_response(resp_text, quick_replies=buttons) else: # No response content (e.g. valid command but no output intended?) # Kakao requires *some* response payload or it treats as error. # We'll return a simple valid JSON to ack. - return self._build_response("Command processed.") + response = self._build_response("Command processed.") + if self._replay_lifecycle is not None: + self._replay_lifecycle.commit_success(lifecycle_key, reason="routed") + return response except Exception as e: + if self._replay_lifecycle is not None: + self._replay_lifecycle.fail_terminal( + lifecycle_key, reason="kakao_response_build_failed" + ) logger.exception(f"Error handling Kakao command: {e}") return self._build_error_response("Internal Error") diff --git a/connector/platforms/slack_webhook.py b/connector/platforms/slack_webhook.py index e44b6b5..ca9978d 100644 --- a/connector/platforms/slack_webhook.py +++ b/connector/platforms/slack_webhook.py @@ -42,6 +42,11 @@ from ..router import CommandRouter from ..security_profile import AllowlistPolicy, ReplayGuard from .slack_installation_manager import SlackInstallationManager +try: + from services.connector_replay_lifecycle import ConnectorReplayLifecycle +except ImportError: # pragma: no cover + ConnectorReplayLifecycle = None # type: ignore + logger = logging.getLogger(__name__) _SLACK_INTERACTION_TYPES = frozenset( @@ -218,6 +223,13 @@ class SlackWebhookServer: window_sec=self.REPLAY_WINDOW_SEC, max_entries=self.NONCE_CACHE_SIZE, ) + if ConnectorReplayLifecycle is None: # pragma: no cover + self._interaction_lifecycle = None + else: + self._interaction_lifecycle = ConnectorReplayLifecycle( + ttl_sec=self.REPLAY_WINDOW_SEC, + max_entries=self.NONCE_CACHE_SIZE, + ) # S67: Allowlists (fail-closed when configured) self._user_allowlist = AllowlistPolicy(config.slack_allowed_users, strict=False) @@ -671,8 +683,29 @@ class SlackWebhookServer: return False replay_key = self._interaction_replay_key(payload, request) - if not self._replay_guard.check_and_record(replay_key): - logger.debug("Slack duplicate interaction %s (accepted, no-op)", replay_key) + if self._interaction_lifecycle is None: # pragma: no cover + if not self._replay_guard.check_and_record(replay_key): + logger.debug( + "Slack duplicate interaction %s (accepted, no-op)", replay_key + ) + return False + claim = None + else: + claim = self._interaction_lifecycle.claim( + replay_key, + metadata={ + "platform": "slack", + "workspace_id": request.workspace_id, + "interaction_type": str(payload.get("type", "") or ""), + }, + ) + if claim is not None and not claim.accepted: + logger.debug( + "Slack duplicate interaction %s state=%s code=%s (accepted, no-op)", + replay_key, + claim.record.state, + claim.code, + ) return False # IMPORTANT: interactive run-like payloads must be routed through the same @@ -683,7 +716,19 @@ class SlackWebhookServer: ): request.text = _force_approval_command(request.text) - response = await self.router.handle(request) + try: + response = await self.router.handle(request) + except Exception: + # IMPORTANT: only failures before router completion are retryable. + # Once router.handle returns, duplicate user actions must not reroute. + if self._interaction_lifecycle is not None: + self._interaction_lifecycle.release_retryable( + replay_key, reason="slack_interaction_failed_before_commit" + ) + raise + + if self._interaction_lifecycle is not None: + self._interaction_lifecycle.commit_success(replay_key, reason="routed") response_text = str(getattr(response, "text", "") or "").strip() response_buttons = getattr(response, "buttons", []) or [] if response_text or response_buttons: diff --git a/connector/platforms/telegram_polling.py b/connector/platforms/telegram_polling.py index 020dd3b..9415084 100644 --- a/connector/platforms/telegram_polling.py +++ b/connector/platforms/telegram_polling.py @@ -9,6 +9,8 @@ import re import time from typing import Optional +from services.connector_replay_lifecycle import ConnectorReplayLifecycle + from ..config import ConnectorConfig from ..contract import CommandRequest, CommandResponse from ..router import CommandRouter @@ -52,6 +54,10 @@ class TelegramPolling: # Remediation: Load offset from persistent state self.offset = self.state_store.get_offset("telegram") self.session = None + self._update_lifecycle = ConnectorReplayLifecycle( + ttl_sec=300, + max_entries=5000, + ) async def start(self): aiohttp = _import_aiohttp() @@ -108,15 +114,42 @@ class TelegramPolling: if self.config.debug and not updates: logger.debug("Telegram poll OK (no updates). offset=%s", self.offset) for update in updates: - next_offset = update["update_id"] + 1 - if next_offset > self.offset: - self.offset = next_offset - # Remediation: Persist offset - self.state_store.set_offset("telegram", self.offset) + update_id = update["update_id"] + lifecycle_key = f"telegram:update:{update_id}" + claim = self._update_lifecycle.claim( + lifecycle_key, + metadata={"platform": "telegram"}, + ) + if not claim.accepted: + logger.debug( + "Telegram duplicate update_id=%s code=%s state=%s", + update_id, + claim.code, + claim.record.state, + ) + if claim.code == "duplicate_after_success": + self._commit_offset(update_id + 1) + continue - await self._process_update(update) + processed = await self._process_update(update) + if processed: + self._update_lifecycle.commit_success( + lifecycle_key, reason="processed" + ) + self._commit_offset(update_id + 1) + else: + # IMPORTANT: keep failed-before-delivery updates retryable. + # Advancing the Telegram offset here would drop the update. + self._update_lifecycle.release_retryable( + lifecycle_key, reason="telegram_update_failed_before_commit" + ) - async def _process_update(self, update: dict): + def _commit_offset(self, next_offset: int) -> None: + if next_offset > self.offset: + self.offset = next_offset + self.state_store.set_offset("telegram", self.offset) + + async def _process_update(self, update: dict) -> bool: # Telegram update shapes vary by chat type and sender mode. # - Normal groups/DMs: `message` # - Edited messages: `edited_message` @@ -133,7 +166,7 @@ class TelegramPolling: or update.get("edited_channel_post") ) if not message or "text" not in message: - return + return True chat_id = message["chat"]["id"] # `from` may be missing for channel posts; `sender_chat` is used for anonymous admins. @@ -174,7 +207,7 @@ class TelegramPolling: try: resp = await self.router.handle(req) - await self._send_response( + return await self._send_response( chat_id, resp, delivery_context=( @@ -183,7 +216,7 @@ class TelegramPolling: ) except Exception as e: logger.exception(f"Error handling command: {e}") - await self._send_response( + return await self._send_response( chat_id, CommandResponse(text="[Error] Internal processing error."), delivery_context=( @@ -221,7 +254,7 @@ class TelegramPolling: chat_id: int, resp: CommandResponse, delivery_context: Optional[dict] = None, - ): + ) -> bool: url = f"{self.base_url}/sendMessage" payload = { "chat_id": chat_id, @@ -241,8 +274,11 @@ class TelegramPolling: logger.error( f"Failed to send Telegram response: {r.status} {await r.text()}" ) + return False + return True except Exception as e: logger.error(f"Telegram send exception: {e}") + return False async def send_image( self, diff --git a/services/connector_callback_contract.py b/services/connector_callback_contract.py index 8884ceb..7d69820 100644 --- a/services/connector_callback_contract.py +++ b/services/connector_callback_contract.py @@ -11,7 +11,12 @@ from typing import Any, Dict, Optional from connector.config import CommandClass from connector.security_profile import ReplayGuard -from connector.transport_contract import CallbackContract, CallbackError, CallbackRecord +from connector.transport_contract import ( + CallbackContract, + CallbackError, + CallbackRecord, + CallbackState, +) try: from .audit import emit_audit_event @@ -20,6 +25,7 @@ try: InstallationResolution, get_connector_installation_registry, ) + from .connector_replay_lifecycle import ConnectorReplayLifecycle except ImportError: from services.audit import emit_audit_event # type: ignore from services.connector_installation_registry import ( # type: ignore @@ -27,6 +33,9 @@ except ImportError: InstallationResolution, get_connector_installation_registry, ) + from services.connector_replay_lifecycle import ( + ConnectorReplayLifecycle, # type: ignore + ) try: from .tenant_context import DEFAULT_TENANT_ID, get_current_tenant_id @@ -134,6 +143,7 @@ class ConnectorCallbackContract: *, installation_registry: Optional[ConnectorInstallationRegistry] = None, replay_guard: Optional[ReplayGuard] = None, + replay_lifecycle: Optional[ConnectorReplayLifecycle] = None, callback_contract: Optional[CallbackContract] = None, action_policy_map: Optional[Dict[str, str]] = None, timestamp_drift_sec: int = DEFAULT_CALLBACK_TIMESTAMP_DRIFT_SEC, @@ -145,6 +155,9 @@ class ConnectorCallbackContract: self._replay_guard = replay_guard or ReplayGuard( window_sec=300, max_entries=5000 ) + self._replay_lifecycle = replay_lifecycle or ConnectorReplayLifecycle( + ttl_sec=300, max_entries=5000 + ) self._callback_contract = callback_contract or CallbackContract() self._action_policy_map = dict(action_policy_map or {}) self._timestamp_drift_sec = max(1, int(timestamp_drift_sec)) @@ -244,6 +257,13 @@ class ConnectorCallbackContract: details=details, ) + @staticmethod + def _request_lifecycle_key(envelope: InteractiveCallbackEnvelope) -> str: + return ( + f"interactive:{envelope.workspace_id}:" + f"{envelope.action_type}:{envelope.request_id}" + ) + def evaluate( self, *, @@ -302,11 +322,19 @@ class ConnectorCallbackContract: ) return decision - if self._replay_guard.is_duplicate(envelope.request_id): + lifecycle_key = self._request_lifecycle_key(envelope) + claim = self._replay_lifecycle.claim( + lifecycle_key, + metadata={ + "workspace_id": envelope.workspace_id, + "action_type": envelope.action_type, + }, + ) + if not claim.accepted: decision = CallbackDecision( ok=False, decision_code=CallbackDecisionCode.REJECT_REPLAY.value, - message="request_id_replay", + message=claim.code, ) self._audit_decision( platform=platform, envelope=envelope, decision=decision @@ -320,6 +348,9 @@ class ConnectorCallbackContract: ) if not resolution.ok or resolution.installation is None: decision = self._map_installation_reject(resolution) + self._replay_lifecycle.fail_terminal( + lifecycle_key, reason=decision.decision_code + ) self._audit_decision( platform=platform, envelope=envelope, decision=decision ) @@ -333,6 +364,9 @@ class ConnectorCallbackContract: installation_id=resolution.installation.installation_id, message="unknown_action_type", ) + self._replay_lifecycle.fail_terminal( + lifecycle_key, reason=decision.decision_code + ) self._audit_decision( platform=platform, envelope=envelope, decision=decision ) @@ -393,6 +427,9 @@ class ConnectorCallbackContract: installation_id=resolution.installation.installation_id, message="admin_required", ) + self._replay_lifecycle.fail_terminal( + lifecycle_key, reason=decision.decision_code + ) else: decision = CallbackDecision( ok=False, @@ -400,6 +437,9 @@ class ConnectorCallbackContract: installation_id=resolution.installation.installation_id, message="unsupported_command_class", ) + self._replay_lifecycle.fail_terminal( + lifecycle_key, reason=decision.decision_code + ) self._audit_decision(platform=platform, envelope=envelope, decision=decision) return decision @@ -416,4 +456,47 @@ class ConnectorCallbackContract: record = self.get_record(request_id) if record is None: raise CallbackError(f"Callback not found for request_id={request_id}") - return self._callback_contract.deliver(record.callback_id) + delivered = self._callback_contract.deliver(record.callback_id) + workspace_id = str(delivered.metadata.get("workspace_id", "") or "") + action_type = str(delivered.metadata.get("action_type", "") or "") + if workspace_id and action_type: + self._replay_lifecycle.commit_success( + f"interactive:{workspace_id}:{action_type}:{request_id}", + reason="completed", + ) + return delivered + + def release_request_retryable( + self, request_id: str, *, reason: str = "retryable_failure" + ) -> None: + record = self.get_record(request_id) + if record is None: + return + if record.state != CallbackState.DELIVERED.value: + # IMPORTANT: retryable release must make CallbackContract.create() + # allocate a fresh record on the next accepted claim. Keeping the + # old acknowledged/pending record would fail the second ack path. + record.state = CallbackState.FAILED.value + workspace_id = str(record.metadata.get("workspace_id", "") or "") + action_type = str(record.metadata.get("action_type", "") or "") + if workspace_id and action_type: + self._replay_lifecycle.release_retryable( + f"interactive:{workspace_id}:{action_type}:{request_id}", + reason=reason, + ) + + def fail_request_terminal( + self, request_id: str, *, reason: str = "terminal_failure" + ) -> None: + record = self.get_record(request_id) + if record is None: + return + if record.state != CallbackState.DELIVERED.value: + record.state = CallbackState.FAILED.value + workspace_id = str(record.metadata.get("workspace_id", "") or "") + action_type = str(record.metadata.get("action_type", "") or "") + if workspace_id and action_type: + self._replay_lifecycle.fail_terminal( + f"interactive:{workspace_id}:{action_type}:{request_id}", + reason=reason, + ) diff --git a/services/connector_replay_lifecycle.py b/services/connector_replay_lifecycle.py new file mode 100644 index 0000000..92a9ced --- /dev/null +++ b/services/connector_replay_lifecycle.py @@ -0,0 +1,266 @@ +""" +Shared connector replay/dedupe lifecycle. + +This complements the simple sliding-window ReplayGuard with explicit state +transitions for connector actions that can fail before delivery and should be +retryable without allowing duplicate execution after success. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass, field +from enum import Enum +from threading import RLock +from typing import Any, Dict, Optional + +DEFAULT_REPLAY_LIFECYCLE_TTL_SEC = 300 +DEFAULT_REPLAY_LIFECYCLE_MAX_ENTRIES = 5000 + + +class ReplayLifecycleState(str, Enum): + CLAIMED = "claimed" + RETRYABLE_FAILURE = "retryable_failure" + DELIVERED = "delivered" + TERMINAL_FAILURE = "terminal_failure" + + +class ReplayClaimCode(str, Enum): + CLAIMED = "claimed" + RETRY_CLAIMED = "retry_claimed" + DUPLICATE_IN_FLIGHT = "duplicate_in_flight" + DUPLICATE_AFTER_SUCCESS = "duplicate_after_success" + DUPLICATE_AFTER_TERMINAL_FAILURE = "duplicate_after_terminal_failure" + + +@dataclass +class ReplayLifecycleRecord: + key: str + state: str = ReplayLifecycleState.CLAIMED.value + created_at: float = field(default_factory=time.time) + updated_at: float = field(default_factory=time.time) + expires_at: float = 0.0 + claim_count: int = 1 + last_reason: str = "" + metadata: Dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "key": self.key, + "state": self.state, + "created_at": self.created_at, + "updated_at": self.updated_at, + "expires_at": self.expires_at, + "claim_count": self.claim_count, + "last_reason": self.last_reason, + "metadata": dict(self.metadata), + } + + +@dataclass(frozen=True) +class ReplayClaimResult: + accepted: bool + code: str + record: ReplayLifecycleRecord + + def to_dict(self) -> Dict[str, Any]: + data = self.record.to_dict() + data.update({"accepted": self.accepted, "code": self.code}) + return data + + +class ConnectorReplayLifecycle: + """Bounded in-memory replay lifecycle for connector events/actions.""" + + def __init__( + self, + *, + ttl_sec: int = DEFAULT_REPLAY_LIFECYCLE_TTL_SEC, + max_entries: int = DEFAULT_REPLAY_LIFECYCLE_MAX_ENTRIES, + ) -> None: + self._ttl_sec = max(1, int(ttl_sec)) + self._max_entries = max(1, int(max_entries)) + self._records: Dict[str, ReplayLifecycleRecord] = {} + self._lock = RLock() + + @property + def ttl_sec(self) -> int: + return self._ttl_sec + + def claim( + self, + key: str, + *, + metadata: Optional[Dict[str, Any]] = None, + now: Optional[float] = None, + ) -> ReplayClaimResult: + normalized_key = self._normalize_key(key) + current_time = time.time() if now is None else float(now) + with self._lock: + self._evict_expired_locked(current_time) + existing = self._records.get(normalized_key) + if existing is None: + record = self._new_record( + normalized_key, current_time, metadata=metadata + ) + self._records[normalized_key] = record + self._enforce_cap_locked() + return ReplayClaimResult( + accepted=True, + code=ReplayClaimCode.CLAIMED.value, + record=record, + ) + + if existing.state == ReplayLifecycleState.RETRYABLE_FAILURE.value: + existing.state = ReplayLifecycleState.CLAIMED.value + existing.updated_at = current_time + existing.expires_at = current_time + self._ttl_sec + existing.claim_count += 1 + existing.last_reason = "" + if metadata: + existing.metadata.update(metadata) + return ReplayClaimResult( + accepted=True, + code=ReplayClaimCode.RETRY_CLAIMED.value, + record=existing, + ) + + if existing.state == ReplayLifecycleState.CLAIMED.value: + code = ReplayClaimCode.DUPLICATE_IN_FLIGHT.value + elif existing.state == ReplayLifecycleState.DELIVERED.value: + code = ReplayClaimCode.DUPLICATE_AFTER_SUCCESS.value + else: + code = ReplayClaimCode.DUPLICATE_AFTER_TERMINAL_FAILURE.value + existing.updated_at = current_time + return ReplayClaimResult(accepted=False, code=code, record=existing) + + def release_retryable( + self, + key: str, + *, + reason: str = "", + now: Optional[float] = None, + ) -> Optional[ReplayLifecycleRecord]: + return self._transition( + key, + ReplayLifecycleState.RETRYABLE_FAILURE.value, + reason=reason, + now=now, + allowed_from={ReplayLifecycleState.CLAIMED.value}, + ) + + def commit_success( + self, + key: str, + *, + reason: str = "", + now: Optional[float] = None, + ) -> Optional[ReplayLifecycleRecord]: + return self._transition( + key, + ReplayLifecycleState.DELIVERED.value, + reason=reason, + now=now, + allowed_from={ + ReplayLifecycleState.CLAIMED.value, + ReplayLifecycleState.RETRYABLE_FAILURE.value, + }, + ) + + def fail_terminal( + self, + key: str, + *, + reason: str = "", + now: Optional[float] = None, + ) -> Optional[ReplayLifecycleRecord]: + return self._transition( + key, + ReplayLifecycleState.TERMINAL_FAILURE.value, + reason=reason, + now=now, + allowed_from={ + ReplayLifecycleState.CLAIMED.value, + ReplayLifecycleState.RETRYABLE_FAILURE.value, + }, + ) + + def get( + self, key: str, *, now: Optional[float] = None + ) -> Optional[ReplayLifecycleRecord]: + normalized_key = self._normalize_key(key) + current_time = time.time() if now is None else float(now) + with self._lock: + self._evict_expired_locked(current_time) + return self._records.get(normalized_key) + + def clear(self) -> None: + with self._lock: + self._records.clear() + + @property + def size(self) -> int: + with self._lock: + self._evict_expired_locked(time.time()) + return len(self._records) + + def _transition( + self, + key: str, + state: str, + *, + reason: str = "", + now: Optional[float] = None, + allowed_from: set[str], + ) -> Optional[ReplayLifecycleRecord]: + normalized_key = self._normalize_key(key) + current_time = time.time() if now is None else float(now) + with self._lock: + self._evict_expired_locked(current_time) + record = self._records.get(normalized_key) + if record is None or record.state not in allowed_from: + return record + record.state = state + record.updated_at = current_time + record.expires_at = current_time + self._ttl_sec + record.last_reason = str(reason or "") + return record + + def _new_record( + self, + key: str, + now: float, + *, + metadata: Optional[Dict[str, Any]] = None, + ) -> ReplayLifecycleRecord: + return ReplayLifecycleRecord( + key=key, + state=ReplayLifecycleState.CLAIMED.value, + created_at=now, + updated_at=now, + expires_at=now + self._ttl_sec, + claim_count=1, + metadata=dict(metadata or {}), + ) + + def _evict_expired_locked(self, now: float) -> None: + expired = [ + key for key, record in self._records.items() if record.expires_at <= now + ] + for key in expired: + del self._records[key] + + def _enforce_cap_locked(self) -> None: + if len(self._records) <= self._max_entries: + return + excess = len(self._records) - self._max_entries + oldest = sorted(self._records.items(), key=lambda item: item[1].updated_at) + for key, _ in oldest[:excess]: + del self._records[key] + + @staticmethod + def _normalize_key(key: str) -> str: + normalized = str(key or "").strip() + if not normalized: + raise ValueError("replay lifecycle key must be non-empty") + return normalized diff --git a/tests/test_connector_callback_contract.py b/tests/test_connector_callback_contract.py index f7c1ddc..1ccce52 100644 --- a/tests/test_connector_callback_contract.py +++ b/tests/test_connector_callback_contract.py @@ -290,6 +290,69 @@ class TestConnectorCallbackContract(unittest.TestCase): delivered = self.contract.complete_request("req-ack") self.assertEqual(delivered.state, "delivered") + def test_duplicate_after_completion_rejected_without_new_record(self): + payload = self._payload() + envelope = self.contract.build_envelope( + request_id="req-complete-duplicate", + workspace_id="T1", + action_type="action.status", + payload=payload, + ) + first = self.contract.evaluate( + platform="slack", + envelope_dict=envelope.__dict__, + payload=payload, + actor=CallbackActorContext(), + ) + self.contract.acknowledge_request("req-complete-duplicate") + self.contract.complete_request("req-complete-duplicate") + + duplicate = self.contract.evaluate( + platform="slack", + envelope_dict=envelope.__dict__, + payload=payload, + actor=CallbackActorContext(), + ) + + self.assertTrue(first.ok) + self.assertFalse(duplicate.ok) + self.assertEqual( + duplicate.decision_code, CallbackDecisionCode.REJECT_REPLAY.value + ) + self.assertEqual(duplicate.message, "duplicate_after_success") + + def test_retryable_release_allows_second_evaluate_and_ack(self): + payload = self._payload() + envelope = self.contract.build_envelope( + request_id="req-retry-release", + workspace_id="T1", + action_type="action.status", + payload=payload, + ) + first = self.contract.evaluate( + platform="slack", + envelope_dict=envelope.__dict__, + payload=payload, + actor=CallbackActorContext(), + ) + self.contract.acknowledge_request("req-retry-release") + self.contract.release_request_retryable( + "req-retry-release", reason="send_failed_before_delivery" + ) + + second = self.contract.evaluate( + platform="slack", + envelope_dict=envelope.__dict__, + payload=payload, + actor=CallbackActorContext(), + ) + acked = self.contract.acknowledge_request("req-retry-release") + + self.assertTrue(first.ok) + self.assertTrue(second.ok) + self.assertEqual(second.decision_code, CallbackDecisionCode.ACCEPT_PUBLIC.value) + self.assertEqual(acked.state, "acknowledged") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_f44_kakao_bundle.py b/tests/test_f44_kakao_bundle.py index a8f4f37..3ec45bc 100644 --- a/tests/test_f44_kakao_bundle.py +++ b/tests/test_f44_kakao_bundle.py @@ -77,6 +77,23 @@ class TestKakaoAdapter(unittest.IsolatedAsyncioTestCase): self.assertEqual(server.router.handle.call_count, 1) self.assertEqual(resp.status, 200) + async def test_router_failure_releases_payload_for_retry(self): + server = self._make_server(allowed_users=["u123"]) + server.router.handle = AsyncMock( + side_effect=[ + RuntimeError("temporary router failure"), + MagicMock(text="Recovered"), + ] + ) + payload = self._make_payload(text="retryable request") + + first = await server.handle_webhook(self._make_mock_request(payload)) + second = await server.handle_webhook(self._make_mock_request(payload)) + + self.assertEqual(first.status, 200) + self.assertEqual(second.status, 200) + self.assertEqual(server.router.handle.await_count, 2) + async def test_allowlist_soft_deny(self): """S32: Untrusted user -> Logged but routed (soft deny).""" server = self._make_server(allowed_users=["trusted"]) diff --git a/tests/test_f59_slack_interactions.py b/tests/test_f59_slack_interactions.py index 92c3ef1..8ef7be4 100644 --- a/tests/test_f59_slack_interactions.py +++ b/tests/test_f59_slack_interactions.py @@ -184,6 +184,45 @@ class TestF59SlackInteractions(unittest.IsolatedAsyncioTestCase): self.assertEqual(response2.status, 200) server.router.handle.assert_called_once() + async def test_failed_block_action_releases_retryable_claim(self): + server = _make_server() + server.router.handle = AsyncMock( + side_effect=[ + RuntimeError("temporary send failure"), + CommandResponse(text=""), + ] + ) + payload = _block_action_payload( + value="/status", action_id="retry", trigger_id="t-retry" + ) + req1 = _build_interaction_request(payload) + req2 = _build_interaction_request(payload) + + first = await server.handle_interaction(req1) + second = await server.handle_interaction(req2) + + self.assertEqual(first.status, 500) + self.assertEqual(second.status, 200) + self.assertEqual(server.router.handle.await_count, 2) + + async def test_reply_failure_after_routing_does_not_reroute_duplicate(self): + server = _make_server() + server.router.handle = AsyncMock(return_value=CommandResponse(text="notify")) + server._send_reply = AsyncMock(side_effect=RuntimeError("reply failed")) + payload = _block_action_payload( + value="/status", action_id="reply-failed", trigger_id="t-reply" + ) + req1 = _build_interaction_request(payload) + req2 = _build_interaction_request(payload) + + first = await server.handle_interaction(req1) + second = await server.handle_interaction(req2) + + self.assertEqual(first.status, 500) + self.assertEqual(second.status, 200) + self.assertEqual(server.router.handle.await_count, 1) + self.assertEqual(server._send_reply.await_count, 1) + async def test_untrusted_run_action_is_forced_to_approval(self): server = _make_server(trusted=False, admin=False) req = _build_interaction_request( diff --git a/tests/test_f69_feishu_callbacks.py b/tests/test_f69_feishu_callbacks.py index 97ba7ae..e78412c 100644 --- a/tests/test_f69_feishu_callbacks.py +++ b/tests/test_f69_feishu_callbacks.py @@ -178,6 +178,33 @@ class TestF69FeishuCallbacks(unittest.IsolatedAsyncioTestCase): self.assertTrue(second["duplicate"]) self.assertEqual(self.router.handle.await_count, 1) + async def test_admin_callback_router_failure_can_retry(self): + self.router._is_admin = MagicMock(return_value=True) + self.router._is_trusted = MagicMock(return_value=True) + self.router.handle = AsyncMock( + side_effect=[ + RuntimeError("temporary route failure"), + CommandResponse(text="OK"), + ] + ) + body = self._callback_body( + button={ + "label": "Approve apr_1", + "value": "/approve apr_1", + "action_type": "approval.approve", + "approval_id": "apr_1", + "style": "primary", + } + ) + + with self.assertRaisesRegex(RuntimeError, "temporary route failure"): + await self.server.process_callback_payload(body) + retried = await self.server.process_callback_payload(body) + + self.assertTrue(retried["ok"]) + self.assertEqual(retried["decision_code"], "cb_accept_admin") + self.assertEqual(self.router.handle.await_count, 2) + async def test_run_callback_degrades_to_approval_for_untrusted_actor(self): self.router._is_admin = MagicMock(return_value=False) self.router._is_trusted = MagicMock(return_value=False) diff --git a/tests/test_f72_telegram_topic_delivery.py b/tests/test_f72_telegram_topic_delivery.py index e656b23..236af19 100644 --- a/tests/test_f72_telegram_topic_delivery.py +++ b/tests/test_f72_telegram_topic_delivery.py @@ -1,3 +1,5 @@ +import os +import tempfile import unittest from unittest.mock import AsyncMock @@ -7,10 +9,16 @@ from connector.platforms.telegram_polling import TelegramPolling class _FakeResponse: - status = 200 + def __init__(self, *, status=200, text="ok", json_data=None): + self.status = status + self._text = text + self._json_data = json_data async def text(self): - return "ok" + return self._text + + async def json(self): + return self._json_data if self._json_data is not None else {"ok": True} class _FakePostContext: @@ -25,12 +33,26 @@ class _FakePostContext: class _FakeSession: - def __init__(self): + def __init__(self, *, post_status=200): self.posts = [] + self.post_status = post_status def post(self, url, **kwargs): self.posts.append((url, kwargs)) - return _FakePostContext(_FakeResponse()) + return _FakePostContext(_FakeResponse(status=self.post_status)) + + +class _FakePollingSession(_FakeSession): + def __init__(self, updates, *, post_status=200): + super().__init__(post_status=post_status) + self.updates = updates + self.gets = [] + + def get(self, url, **kwargs): + self.gets.append((url, kwargs)) + return _FakePostContext( + _FakeResponse(json_data={"ok": True, "result": self.updates}) + ) class _FakeRouter: @@ -50,10 +72,12 @@ def _form_field_value(form_data, name): class TestTelegramTopicDelivery(unittest.IsolatedAsyncioTestCase): - def _server(self): + def _server(self, *, state_path=None): cfg = ConnectorConfig() cfg.telegram_bot_token = "token" cfg.telegram_allowed_chats = [-100123] + if state_path: + cfg.state_path = state_path router = _FakeRouter() server = TelegramPolling(cfg, router) server.session = _FakeSession() @@ -79,6 +103,52 @@ class TestTelegramTopicDelivery(unittest.IsolatedAsyncioTestCase): _url, kwargs = server.session.posts[-1] self.assertEqual(kwargs["json"]["message_thread_id"], 456) + async def test_poll_commits_offset_after_successful_update_delivery(self): + with tempfile.TemporaryDirectory() as temp_dir: + server, router = self._server( + state_path=os.path.join(temp_dir, "state.json") + ) + update = { + "update_id": 8, + "message": { + "message_id": 77, + "chat": {"id": -100123, "type": "supergroup"}, + "from": {"id": 42, "username": "alice"}, + "text": "/status", + }, + } + server.session = _FakePollingSession([update]) + + await server._poll_once() + + self.assertEqual(server.offset, 9) + self.assertEqual(len(router.requests), 1) + + async def test_poll_releases_update_when_response_delivery_fails(self): + with tempfile.TemporaryDirectory() as temp_dir: + server, router = self._server( + state_path=os.path.join(temp_dir, "state.json") + ) + update = { + "update_id": 8, + "message": { + "message_id": 77, + "chat": {"id": -100123, "type": "supergroup"}, + "from": {"id": 42, "username": "alice"}, + "text": "/status", + }, + } + server.session = _FakePollingSession([update], post_status=500) + + await server._poll_once() + self.assertEqual(server.offset, 0) + + server.session = _FakePollingSession([update], post_status=200) + await server._poll_once() + + self.assertEqual(server.offset, 9) + self.assertEqual(len(router.requests), 2) + async def test_send_message_includes_valid_thread_id(self): server, _router = self._server() diff --git a/tests/test_r189_connector_replay_lifecycle.py b/tests/test_r189_connector_replay_lifecycle.py new file mode 100644 index 0000000..449f8c1 --- /dev/null +++ b/tests/test_r189_connector_replay_lifecycle.py @@ -0,0 +1,73 @@ +import unittest + +from services.connector_replay_lifecycle import ( + ConnectorReplayLifecycle, + ReplayClaimCode, + ReplayLifecycleState, +) + + +class TestConnectorReplayLifecycle(unittest.TestCase): + def test_duplicate_in_flight_rejected(self): + lifecycle = ConnectorReplayLifecycle(ttl_sec=30) + + first = lifecycle.claim("telegram:update:1", metadata={"platform": "telegram"}) + second = lifecycle.claim("telegram:update:1") + + self.assertTrue(first.accepted) + self.assertEqual(first.code, ReplayClaimCode.CLAIMED.value) + self.assertFalse(second.accepted) + self.assertEqual(second.code, ReplayClaimCode.DUPLICATE_IN_FLIGHT.value) + self.assertEqual(second.record.state, ReplayLifecycleState.CLAIMED.value) + + def test_retryable_release_allows_reclaim(self): + lifecycle = ConnectorReplayLifecycle(ttl_sec=30) + + first = lifecycle.claim("slack:interaction:1") + lifecycle.release_retryable("slack:interaction:1", reason="send_failed") + second = lifecycle.claim("slack:interaction:1") + + self.assertTrue(first.accepted) + self.assertTrue(second.accepted) + self.assertEqual(second.code, ReplayClaimCode.RETRY_CLAIMED.value) + self.assertEqual(second.record.claim_count, 2) + self.assertEqual(second.record.state, ReplayLifecycleState.CLAIMED.value) + + def test_success_commit_is_terminal_for_duplicates(self): + lifecycle = ConnectorReplayLifecycle(ttl_sec=30) + + lifecycle.claim("feishu:callback:1") + lifecycle.commit_success("feishu:callback:1", reason="delivered") + duplicate = lifecycle.claim("feishu:callback:1") + + self.assertFalse(duplicate.accepted) + self.assertEqual(duplicate.code, ReplayClaimCode.DUPLICATE_AFTER_SUCCESS.value) + self.assertEqual(duplicate.record.state, ReplayLifecycleState.DELIVERED.value) + + def test_terminal_failure_does_not_reclaim(self): + lifecycle = ConnectorReplayLifecycle(ttl_sec=30) + + lifecycle.claim("kakao:webhook:1") + lifecycle.fail_terminal("kakao:webhook:1", reason="invalid_policy") + duplicate = lifecycle.claim("kakao:webhook:1") + + self.assertFalse(duplicate.accepted) + self.assertEqual( + duplicate.code, ReplayClaimCode.DUPLICATE_AFTER_TERMINAL_FAILURE.value + ) + self.assertEqual( + duplicate.record.state, ReplayLifecycleState.TERMINAL_FAILURE.value + ) + + def test_expired_claim_can_be_reclaimed(self): + lifecycle = ConnectorReplayLifecycle(ttl_sec=5) + + lifecycle.claim("whatsapp:webhook:1", now=10.0) + reclaimed = lifecycle.claim("whatsapp:webhook:1", now=16.0) + + self.assertTrue(reclaimed.accepted) + self.assertEqual(reclaimed.code, ReplayClaimCode.CLAIMED.value) + + +if __name__ == "__main__": + unittest.main()