admin/ComfyUI-OpenClaw
1
0
Fork 0
mirror of https://github.com/rookiestar28/ComfyUI-OpenClaw.git synced 2026-08-14 00:48:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(connector): unify replay lifecycle

Browse Source
This commit is contained in:
rookiestar28
2026-05-04 12:11:51 +08:00
parent f6d92b2d60
commit ce934d00a9
12 changed files with 809 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connector/platforms/feishu_webhook.py
+13 -3
View File
@@ -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."
)
connector/platforms/kakao_webhook.py
+54 -6
View File
@@ -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")
connector/platforms/slack_webhook.py
+48 -3
View File
@@ -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:
connector/platforms/telegram_polling.py
+47 -11
View File
@@ -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,
services/connector_callback_contract.py
+87 -4
View File
@@ -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,
)
services/connector_replay_lifecycle.py
+266
View File
@@ -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
tests/test_connector_callback_contract.py
+63
View File
@@ -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()
tests/test_f44_kakao_bundle.py
+17
View File
@@ -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"])
tests/test_f59_slack_interactions.py
+39
View File
@@ -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(
tests/test_f69_feishu_callbacks.py
+27
View File
@@ -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)
tests/test_f72_telegram_topic_delivery.py
+75 -5
View File
@@ -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()
tests/test_r189_connector_replay_lifecycle.py
+73
View File
@@ -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()