From 8e1db41041d755dd59568422ee89bfc4f8b458ca Mon Sep 17 00:00:00 2001 From: milnerrad <302432023+milnerrad@users.noreply.github.com> Date: Wed, 26 Aug 2026 11:15:57 +0800 Subject: [PATCH] fix(gateway): redeliver transient failures after reconnect --- gateway/delivery_ledger.py | 192 ++++++++++- gateway/platforms/base.py | 35 +- gateway/run.py | 180 +++++++++-- tests/gateway/test_delivery_ledger.py | 306 +++++++++++++++++- .../gateway/test_delivery_ledger_producer.py | 30 +- .../test_multiplex_adapter_registry.py | 15 + tests/gateway/test_platform_reconnect.py | 4 + 7 files changed, 726 insertions(+), 36 deletions(-) diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index a0b5c61783..16f4633076 100644 --- a/gateway/delivery_ledger.py +++ b/gateway/delivery_ledger.py @@ -17,10 +17,12 @@ bounded retention). The gateway writes three checkpoints around the send: mark_failed() state='failed' on a definitive rejection On startup, ``sweep_recoverable()`` claims rows whose owning process is -dead and hands them to the gateway for redelivery. Crash semantics are -explicit about ambiguity (the contract review of the earlier -delivery-outbox attempt, #61790, closed it for silently resending -ambiguous sends): +dead and hands them to the gateway for redelivery. After a platform adapter +reconnects without a process restart, ``sweep_failed_for_runtime()`` may claim +only the same live process's explicitly allowlisted transient failures. Crash +semantics are explicit about ambiguity (the contract review of the earlier +delivery-outbox attempt, #61790, closed it for silently resending ambiguous +sends): - ``pending`` — the send never started: redeliver plainly, no dup risk. - ``attempting`` — crashed mid-await: the platform MAY already have the @@ -70,6 +72,20 @@ RECOVERED_MARKER = ( "so this may be a duplicate:\n\n" ) +# Runtime recovery uses a distinct marker because no gateway restart occurred. +# Keep the ambiguity explicit: a network rejection normally means the platform +# did not accept the message, but an acknowledgement can be lost independently. +RECONNECTED_MARKER = ( + "♻️ Recovered reply — the messaging platform reconnected after the original " + "delivery failed, so this may be a duplicate:\n\n" +) + +# Runtime replay is deliberately fail-closed. Only errors whose send contract +# proves they are transient reconnect failures belong here; permanent rejects +# (blocked bot, bad auth, missing chat) must not be retried merely because an +# adapter reconnected. +_RUNTIME_RETRYABLE_ERRORS = frozenset({"send_path_degraded"}) + def _db_path(): return get_hermes_home() / "state.db" @@ -107,9 +123,22 @@ def _initialize_schema(conn: sqlite3.Connection) -> None: updated_at REAL NOT NULL, owner_pid INTEGER, owner_started_at INTEGER, - last_error TEXT + last_error TEXT, + adapter_profile TEXT )""" ) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(delivery_obligations)") + } + if "adapter_profile" not in columns: + try: + conn.execute( + "ALTER TABLE delivery_obligations ADD COLUMN adapter_profile TEXT" + ) + except sqlite3.OperationalError as exc: + # Concurrent first-use connections can both observe the old schema. + if "duplicate column" not in str(exc).lower(): + raise @contextmanager @@ -209,20 +238,22 @@ def record_obligation( chat_id: str, thread_id: Optional[str], content: str, + adapter_profile: Optional[str] = None, ) -> None: """Record a final response as owed to the platform (state='pending').""" now = time.time() + stored_profile = str(adapter_profile).strip() if adapter_profile else "default" pid, started = _owner_stamp() with _DB_LOCK, _transaction() as conn: conn.execute( """INSERT OR REPLACE INTO delivery_obligations (obligation_id, session_key, platform, chat_id, thread_id, content, state, attempts, created_at, updated_at, - owner_pid, owner_started_at) - VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?)""", + owner_pid, owner_started_at, adapter_profile) + VALUES (?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?, ?, ?, ?)""", (obligation_id, session_key, platform, str(chat_id), str(thread_id) if thread_id else None, content, now, now, - pid, started), + pid, started, stored_profile), ) _prune() @@ -239,6 +270,32 @@ def mark_failed(obligation_id: str, error: str = "") -> None: _update_state(obligation_id, "failed", error=error) +def release_runtime_claim(obligation_id: str, error: str = "") -> bool: + """Return an unsent runtime claim to ``failed`` without spending an attempt. + + Runtime recovery claims before clearing ``resume_pending`` so that two + reconnect paths cannot send the same row. If the session flag cannot be + cleared, no platform send was attempted and the claim must not consume the + bounded redelivery budget. Release is fail-closed to the exact current + process instance and the ``attempting`` state. + """ + pid, started = _owner_stamp() + if started is None: + return False + with _DB_LOCK, _transaction() as conn: + cursor = conn.execute( + """UPDATE delivery_obligations + SET state='failed', attempts=CASE + WHEN attempts > 0 THEN attempts - 1 ELSE 0 END, + updated_at=?, last_error=? + WHERE obligation_id=? AND state='attempting' + AND owner_pid IS ? AND owner_started_at IS ?""", + (time.time(), error[:500] if error else None, + obligation_id, pid, started), + ) + return bool(cursor.rowcount) + + def _update_state(obligation_id: str, state: str, error: str = "") -> None: with _DB_LOCK, _transaction() as conn: conn.execute( @@ -253,6 +310,7 @@ def sweep_recoverable( now: Optional[float] = None, *, deliverable_platforms: Optional[set] = None, + deliverable_targets: Optional[set] = None, ) -> List[Dict[str, Any]]: """Claim undelivered rows owned by dead processes; return them for redelivery. @@ -269,6 +327,10 @@ def sweep_recoverable( that failed to connect would otherwise burn one attempt per boot and hit the cap having never been sent once. Rows for absent platforms are left untouched for a later boot; the stale cutoff still bounds them. + + ``deliverable_targets`` further scopes multiplexed gateways by exact + ``(platform, adapter_profile)`` identity, preventing one connected bot from + spending another disconnected bot's retry budget. """ now = now if now is not None else time.time() pid, started = _owner_stamp() @@ -277,12 +339,13 @@ def sweep_recoverable( rows = conn.execute( """SELECT obligation_id, session_key, platform, chat_id, thread_id, content, state, attempts, created_at, - owner_pid, owner_started_at + owner_pid, owner_started_at, adapter_profile FROM delivery_obligations WHERE state IN ('pending', 'attempting', 'failed')""" ).fetchall() for (oid, session_key, platform, chat_id, thread_id, content, state, - attempts, created_at, owner_pid, owner_started_at) in rows: + attempts, created_at, owner_pid, owner_started_at, + adapter_profile) in rows: if _owner_alive(owner_pid, owner_started_at): continue # a live gateway still owns this row if attempts >= MAX_ATTEMPTS or (now - created_at) > STALE_AFTER_SECONDS: @@ -299,6 +362,11 @@ def sweep_recoverable( # No adapter for this platform this boot — the caller cannot # send, so claiming would spend an attempt on a no-op. continue + if ( + deliverable_targets is not None + and (platform, adapter_profile) not in deliverable_targets + ): + continue cursor = conn.execute( """UPDATE delivery_obligations SET owner_pid=?, owner_started_at=?, attempts=attempts+1, @@ -317,6 +385,110 @@ def sweep_recoverable( # pending = send never started, redeliver plainly; # attempting/failed = ambiguous or rejected, carry marker. "needs_marker": state != "pending", + "profile": adapter_profile, + "attempts": attempts + 1, + }) + return claimed + + +def sweep_failed_for_runtime( + platform: str, + now: Optional[float] = None, + *, + profile: Optional[str] = None, +) -> List[Dict[str, Any]]: + """Claim this process's reconnect-retryable failed rows for one adapter. + + ``profile`` scopes multiplexed gateways to the bot identity that actually + owned the failed send; ``None`` means the primary/default adapter. The + persisted adapter owner is independent of the routed session namespace. + + Startup recovery intentionally ignores rows owned by a live gateway. That + protects concurrent processes, but it also means a final response rejected + with ``send_path_degraded`` remains stranded when only the platform adapter + reconnects. This runtime sweep closes that gap without weakening ownership: + + - only rows stamped to this exact process instance are eligible; + - only explicitly allowlisted transient errors are eligible; + - attempts/staleness bounds match startup recovery; + - every update is guarded by the prior owner stamp and ``failed`` state. + + Unowned rows and rows owned by another process are left untouched for the + normal startup/dead-owner sweep. Claimed rows always carry the reconnect + marker because the failed send's acknowledgement is not safe to infer. + """ + now = now if now is not None else time.time() + pid, started = _owner_stamp() + if started is None: + # PID equality alone cannot distinguish this process from a stale row + # left by an earlier process incarnation after PID reuse. Runtime replay + # is optional recovery, so fail closed when the process fingerprint is + # unavailable; startup recovery remains the durable fallback. + return [] + claimed: List[Dict[str, Any]] = [] + with _DB_LOCK, _transaction() as conn: + rows = conn.execute( + """SELECT obligation_id, session_key, platform, chat_id, thread_id, + content, attempts, created_at, owner_pid, + owner_started_at, last_error, adapter_profile + FROM delivery_obligations + WHERE state='failed' AND platform=?""", + (platform,), + ).fetchall() + for ( + oid, + session_key, + row_platform, + chat_id, + thread_id, + content, + attempts, + created_at, + owner_pid, + owner_started_at, + last_error, + adapter_profile, + ) in rows: + expected_profile = ( + "default" if not profile or profile == "default" else str(profile) + ) + if adapter_profile != expected_profile: + continue + # Runtime reconnect recovery may act only on its own rows. Exact + # process-start matching prevents PID reuse from stealing work. + if owner_pid != pid or owner_started_at != started: + continue + if str(last_error or "").strip().lower() not in _RUNTIME_RETRYABLE_ERRORS: + continue + owner_guard = (oid, owner_pid, owner_started_at) + if attempts >= MAX_ATTEMPTS or (now - created_at) > STALE_AFTER_SECONDS: + conn.execute( + """UPDATE delivery_obligations + SET state='abandoned', updated_at=? + WHERE obligation_id=? AND state='failed' + AND owner_pid IS ? AND owner_started_at IS ?""", + (now, *owner_guard), + ) + continue + cursor = conn.execute( + """UPDATE delivery_obligations + SET state='attempting', attempts=attempts+1, updated_at=? + WHERE obligation_id=? AND state='failed' + AND owner_pid IS ? AND owner_started_at IS ?""", + (now, *owner_guard), + ) + if cursor.rowcount: + claimed.append({ + "obligation_id": oid, + "session_key": session_key, + "platform": row_platform, + "chat_id": chat_id, + "thread_id": thread_id, + "content": content, + "needs_marker": True, + "marker": RECONNECTED_MARKER, + "profile": adapter_profile, + "runtime_recovery": True, "attempts": attempts + 1, }) return claimed diff --git a/gateway/platforms/base.py b/gateway/platforms/base.py index aa0c83d128..e990255794 100644 --- a/gateway/platforms/base.py +++ b/gateway/platforms/base.py @@ -6684,6 +6684,9 @@ class BasePlatformAdapter(ABC): chat_id=event.source.chat_id, thread_id=getattr(event.source, "thread_id", None), content=text_content, + adapter_profile=getattr( + delivery_adapter, "_owner_profile", None + ), ) await asyncio.to_thread(mark_attempting, _obligation_id) except Exception: @@ -6706,11 +6709,41 @@ class BasePlatformAdapter(ABC): if getattr(result, "success", False): await asyncio.to_thread(mark_delivered, _obligation_id) else: + _delivery_error = str( + getattr(result, "error", "") or "" + ) await asyncio.to_thread( mark_failed, _obligation_id, - str(getattr(result, "error", "") or ""), + _delivery_error, ) + # A replacement can finish reconnecting before + # this in-flight failure reaches mark_failed. In + # that ordering the watcher's sweep found no row. + # Signal a second transactional sweep only when a + # new live adapter is already installed; atomic + # claiming makes concurrent signals idempotent. + if _delivery_error == "send_path_degraded": + _live_adapter = self._final_delivery_adapter( + event.source + ) + _runtime_redeliver = getattr( + getattr(self, "gateway_runner", None), + "_redeliver_failed_obligations_for_platform", + None, + ) + if ( + _live_adapter is not delivery_adapter + and callable(_runtime_redeliver) + ): + await _runtime_redeliver( + event.source.platform, + profile=getattr( + delivery_adapter, + "_owner_profile", + None, + ), + ) except Exception: logger.debug( "delivery ledger update failed", exc_info=True diff --git a/gateway/run.py b/gateway/run.py index e56aaaa5d5..b219097a03 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -12047,6 +12047,35 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew task, "background boot-path send failed after gate release: see traceback" ) + async def _clear_resume_pending_for_claimed_obligations( + self, claimed: list, *, require_success: bool = False + ) -> list: + """Clear resume flags and return rows safe to redeliver. + + Startup recovery preserves its historical best-effort behavior. Runtime + reconnect recovery is stricter: if the session-store write fails, the + corresponding response must not be sent because the same agent turn + could otherwise be resumed immediately afterward. + """ + sendable = [] + for row in claimed: + session_key = row.get("session_key") or "" + if not session_key: + sendable.append(row) + continue + try: + await self.async_session_store.clear_resume_pending(session_key) + except Exception: + logger.debug( + "clear_resume_pending failed for %s", session_key, + exc_info=True, + ) + if not require_success: + sendable.append(row) + else: + sendable.append(row) + return sendable + async def _claim_pending_obligations(self) -> list: """Claim recoverable delivery-ledger rows and clear their ``resume_pending`` flags. Pure DB work — no network sends. @@ -12072,14 +12101,31 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew if not await asyncio.to_thread(ledger_enabled): return [] - # Only claim rows we can actually send this boot: self.adapters - # holds a platform only after its connect() succeeded, and each - # claim spends one of the row's three redelivery attempts. - _deliverable = { - getattr(p, "value", str(p)) for p in self.adapters + # Only claim rows whose exact transport owner is connected this + # boot. A multiplexed gateway can host several bot identities for + # one platform; platform-only filtering would spend a disconnected + # bot's retry budget merely because another bot is online. + _profile_adapters = getattr(self, "_profile_adapters", None) or {} + _deliverable_targets = { + (getattr(p, "value", str(p)), "default") for p in self.adapters } + # Legacy rows predate adapter_profile. They are unambiguous only in + # a non-multiplexed gateway; fail closed when multiple bot identities + # share the process. + if not _profile_adapters: + _deliverable_targets.update( + (getattr(p, "value", str(p)), None) for p in self.adapters + ) + for _profile, _adapters in _profile_adapters.items(): + _deliverable_targets.update( + (getattr(p, "value", str(p)), _profile) for p in _adapters + ) + _deliverable = {platform for platform, _ in _deliverable_targets} claimed = await asyncio.to_thread( - sweep_recoverable, None, deliverable_platforms=_deliverable + sweep_recoverable, + None, + deliverable_platforms=_deliverable, + deliverable_targets=_deliverable_targets, ) except Exception: logger.debug("delivery ledger sweep failed", exc_info=True) @@ -12091,17 +12137,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew # send. Claiming already spent one of the row's redelivery attempts — # the answer is in the ledger, so the resume path must never re-run # these turns (#91969). - for row in claimed: - session_key = row.get("session_key") or "" - if not session_key: - continue - try: - await self.async_session_store.clear_resume_pending(session_key) - except Exception: - logger.debug( - "clear_resume_pending failed for %s", session_key, - exc_info=True, - ) + await self._clear_resume_pending_for_claimed_obligations(claimed) return claimed async def _redeliver_claimed_obligations(self, claimed: list) -> int: @@ -12119,6 +12155,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew RECOVERED_MARKER, mark_delivered, mark_failed, + release_runtime_claim, ) except Exception: logger.debug("delivery ledger import failed", exc_info=True) @@ -12134,14 +12171,36 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew row["obligation_id"], row.get("platform"), ) continue - adapter = self.adapters.get(platform) + if "profile" in row: + adapter = self._authorization_adapter( + platform, row.get("profile") + ) + else: + # Startup rows preserve the historical default-adapter route. + adapter = self.adapters.get(platform) if adapter is None: - # Platform not connected this boot — leave the row claimed; - # attempts cap + stale cutoff bound the retries on later boots. + # Runtime claims have not reached a transport yet. If the + # reconnect vanished before dispatch, release the claim without + # spending an attempt so the next reconnect can retry it. + if row.get("runtime_recovery"): + try: + await asyncio.to_thread( + release_runtime_claim, + row["obligation_id"], + "send_path_degraded", + ) + except Exception: + logger.debug( + "failed to release undispatched runtime obligation %s", + row["obligation_id"], + exc_info=True, + ) + # Startup claims preserve their historical state; attempts cap + # + stale cutoff bound later retries. continue content = row["content"] if row.get("needs_marker"): - content = RECOVERED_MARKER + content + content = row.get("marker", RECOVERED_MARKER) + content metadata = ( {"thread_id": row["thread_id"]} if row.get("thread_id") else None ) @@ -12191,6 +12250,67 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew await self._claim_pending_obligations() ) + async def _redeliver_failed_obligations_for_platform( + self, + platform: Platform, + *, + profile: Optional[str] = None, + ) -> int: + """Replay one adapter identity's transient failures after reconnect. + + The startup sweep cannot claim live-owner rows by design. A platform + adapter can reconnect without the gateway process exiting, however, so + ``send_path_degraded`` responses otherwise remain failed until the next + process restart. Claiming, resume clearing, and sending stay best-effort + and reuse the startup redelivery path's attempt and ambiguity contract. + """ + try: + from gateway.delivery_ledger import ( + ledger_enabled, + release_runtime_claim, + sweep_failed_for_runtime, + ) + + if not await asyncio.to_thread(ledger_enabled): + return 0 + claimed = await asyncio.to_thread( + sweep_failed_for_runtime, + platform.value, + profile=profile, + ) + except Exception: + logger.debug( + "runtime delivery ledger sweep failed after %s reconnect", + platform.value, + exc_info=True, + ) + return 0 + if not claimed: + return 0 + + # Clear before any send so the reconnect path cannot both redeliver an + # already-produced answer and schedule the same agent turn for resume. + sendable = await self._clear_resume_pending_for_claimed_obligations( + claimed, require_success=True + ) + sendable_ids = {row["obligation_id"] for row in sendable} + for row in claimed: + if row["obligation_id"] in sendable_ids: + continue + try: + await asyncio.to_thread( + release_runtime_claim, + row["obligation_id"], + "send_path_degraded", + ) + except Exception: + logger.debug( + "failed to release runtime delivery claim %s", + row["obligation_id"], + exc_info=True, + ) + return await self._redeliver_claimed_obligations(sendable) + def _schedule_resume_pending_sessions(self, platform=None) -> int: """Auto-continue fresh restart-interrupted sessions after startup. @@ -14600,6 +14720,21 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew ) logger.info("✓ %s reconnected successfully", platform.value) + # Final responses rejected while this adapter was down + # are still owned by this live process, so startup + # recovery cannot claim them. Replay the explicitly + # transient subset now that the platform is usable. + try: + await self._redeliver_failed_obligations_for_platform( + platform + ) + except Exception: + logger.debug( + "failed-obligation redelivery after %s reconnect failed", + platform.value, + exc_info=True, + ) + # Rebuild channel directory with the new adapter try: from gateway.channel_directory import build_channel_directory @@ -15717,6 +15852,9 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew platform.value, profile_name, ) + await self._redeliver_failed_obligations_for_platform( + platform, profile=profile_name + ) return # A newer reconnect already won the slot while this # attempt was awaiting connect; do not replace it. diff --git a/tests/gateway/test_delivery_ledger.py b/tests/gateway/test_delivery_ledger.py index dd4bc52130..f699be1764 100644 --- a/tests/gateway/test_delivery_ledger.py +++ b/tests/gateway/test_delivery_ledger.py @@ -9,8 +9,10 @@ id stability, and the startup redelivery sweep's contract: - poison rows abandon at the attempts cap / stale cutoff """ -import time +import os +import sqlite3 import threading +import time from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -36,18 +38,23 @@ def _record(oid="ob-1", session_key="agent:main:slack:channel:C1", **kw): chat_id=kw.get("chat_id", "C1"), thread_id=kw.get("thread_id", "171.001"), content=kw.get("content", "the final answer"), + adapter_profile=kw.get("adapter_profile"), ) def _row(oid): with dl._connect() as conn: r = conn.execute( - """SELECT state, attempts, owner_pid, content + """SELECT state, attempts, owner_pid, content, last_error FROM delivery_obligations WHERE obligation_id=?""", (oid,), ).fetchone() return None if r is None else { - "state": r[0], "attempts": r[1], "owner_pid": r[2], "content": r[3], + "state": r[0], + "attempts": r[1], + "owner_pid": r[2], + "content": r[3], + "last_error": r[4], } @@ -88,6 +95,37 @@ def _orphan(oid): ) +class TestSchemaMigration: + def test_adds_adapter_profile_to_existing_ledger(self): + conn = sqlite3.connect(dl._db_path()) + try: + conn.execute( + """CREATE TABLE delivery_obligations ( + obligation_id TEXT PRIMARY KEY, + session_key TEXT NOT NULL, + platform TEXT NOT NULL, + chat_id TEXT NOT NULL, + thread_id TEXT, + content TEXT NOT NULL, + state TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + owner_pid INTEGER, + owner_started_at INTEGER, + last_error TEXT + )""" + ) + dl._initialize_schema(conn) + columns = { + row[1] for row in conn.execute("PRAGMA table_info(delivery_obligations)") + } + finally: + conn.close() + + assert "adapter_profile" in columns + + class TestStateMachine: def test_record_starts_pending(self): _record() @@ -123,6 +161,146 @@ class TestSweep: assert dl.sweep_recoverable() == [] +class TestRuntimeFailedSweep: + """A live gateway may reclaim only its own transient reconnect failures.""" + + def test_claims_current_process_send_path_degraded_row(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + + claimed = dl.sweep_failed_for_runtime("telegram") + + assert len(claimed) == 1 + assert claimed[0]["needs_marker"] is True + assert claimed[0]["attempts"] == 1 + assert _row("ob-1")["state"] == "attempting" + + def test_permanent_failure_is_not_claimed(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "Forbidden: bot was blocked by the user") + + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-1")["state"] == "failed" + assert _row("ob-1")["attempts"] == 0 + + def test_claim_is_platform_scoped_and_not_reclaimed_while_attempting(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + _record( + oid="ob-2", + session_key="agent:main:slack:channel:C2", + platform="slack", + chat_id="C2", + ) + dl.mark_failed("ob-2", "send_path_degraded") + + claimed = dl.sweep_failed_for_runtime("telegram") + + assert [row["obligation_id"] for row in claimed] == ["ob-1"] + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-2")["state"] == "failed" + assert _row("ob-2")["attempts"] == 0 + + def test_other_live_owner_is_not_claimed_or_abandoned(self, monkeypatch): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET owner_pid=?, " + "owner_started_at=?, attempts=? WHERE obligation_id=?", + (12345, 101, dl.MAX_ATTEMPTS, "ob-1"), + ) + monkeypatch.setattr(dl, "_owner_stamp", lambda: (54321, 202)) + + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-1")["state"] == "failed" + assert _row("ob-1")["attempts"] == dl.MAX_ATTEMPTS + + def test_unowned_row_is_not_claimed(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET owner_pid=NULL, " + "owner_started_at=NULL WHERE obligation_id=?", + ("ob-1",), + ) + + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-1")["state"] == "failed" + + def test_missing_current_process_start_stamp_fails_closed(self, monkeypatch): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET owner_started_at=NULL " + "WHERE obligation_id=?", + ("ob-1",), + ) + monkeypatch.setattr(dl, "_owner_stamp", lambda: (os.getpid(), None)) + + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-1")["state"] == "failed" + + def test_same_pid_with_different_start_stamp_is_not_claimed(self, monkeypatch): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET owner_pid=?, owner_started_at=? " + "WHERE obligation_id=?", + (os.getpid(), 101, "ob-1"), + ) + monkeypatch.setattr(dl, "_owner_stamp", lambda: (os.getpid(), 202)) + + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-1")["state"] == "failed" + + def test_profile_scope_never_claims_another_bot_identity(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + _record( + oid="ob-2", + session_key="agent:reviewer:telegram:dm:C2", + platform="telegram", + chat_id="C2", + adapter_profile="reviewer", + ) + dl.mark_failed("ob-2", "send_path_degraded") + + claimed = dl.sweep_failed_for_runtime("telegram", profile="reviewer") + + assert [row["obligation_id"] for row in claimed] == ["ob-2"] + assert claimed[0]["profile"] == "reviewer" + assert _row("ob-1")["state"] == "failed" + + def test_current_owner_row_at_attempt_cap_is_abandoned(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET attempts=? WHERE obligation_id=?", + (dl.MAX_ATTEMPTS, "ob-1"), + ) + + assert dl.sweep_failed_for_runtime("telegram") == [] + assert _row("ob-1")["state"] == "abandoned" + + def test_current_owner_stale_row_is_abandoned(self): + _record(platform="telegram") + dl.mark_failed("ob-1", "send_path_degraded") + now = time.time() + with dl._connect() as conn: + conn.execute( + "UPDATE delivery_obligations SET created_at=? WHERE obligation_id=?", + (now - dl.STALE_AFTER_SECONDS - 1, "ob-1"), + ) + + assert dl.sweep_failed_for_runtime("telegram", now=now) == [] + assert _row("ob-1")["state"] == "abandoned" + + class TestPrune: def test_old_delivered_rows_pruned(self): _record() @@ -152,6 +330,8 @@ class TestGatewayRedeliverySweep: runner = object.__new__(GatewayRunner) runner.adapters = {Platform.SLACK: adapter} if adapter else {} + runner._profile_adapters = {} + runner._active_profile_name = lambda: "default" _store = MagicMock() _store.clear_resume_pending = AsyncMock() _store._store = None @@ -185,6 +365,45 @@ class TestGatewayRedeliverySweep: "agent:main:slack:channel:C1" ) + @pytest.mark.asyncio + async def test_startup_redelivery_uses_persisted_transport_owner(self): + from gateway.config import Platform + + _record( + session_key="agent:routed-profile:slack:channel:C1", + adapter_profile="credential-owner", + ) + _orphan("ob-1") + default_adapter = self._adapter() + owner_adapter = self._adapter() + runner = self._runner(default_adapter) + runner._profile_adapters = { + "credential-owner": {Platform.SLACK: owner_adapter} + } + + n = await runner._redeliver_pending_obligations() + + assert n == 1 + default_adapter.send.assert_not_awaited() + owner_adapter.send.assert_awaited_once() + + @pytest.mark.asyncio + async def test_startup_does_not_claim_disconnected_transport_owner(self): + _record( + session_key="agent:routed-profile:slack:channel:C1", + adapter_profile="credential-owner", + ) + _orphan("ob-1") + default_adapter = self._adapter() + runner = self._runner(default_adapter) + + n = await runner._redeliver_pending_obligations() + + assert n == 0 + default_adapter.send.assert_not_awaited() + assert _row("ob-1")["state"] == "pending" + assert _row("ob-1")["attempts"] == 0 + @pytest.mark.asyncio async def test_attempting_redelivers_with_marker(self): _record() @@ -199,6 +418,87 @@ class TestGatewayRedeliverySweep: assert sent["content"].startswith(dl.RECOVERED_MARKER) assert sent["content"].endswith("the final answer") + @pytest.mark.asyncio + async def test_runtime_failed_redelivery_clears_resume_before_send(self): + from gateway.config import Platform + + _record(platform="slack") + dl.mark_failed("ob-1", "send_path_degraded") + adapter = self._adapter() + runner = self._runner(adapter) + + n = await runner._redeliver_failed_obligations_for_platform(Platform.SLACK) + + assert n == 1 + runner._async_session_store.clear_resume_pending.assert_awaited_once_with( + "agent:main:slack:channel:C1" + ) + assert adapter.send.await_count == 1 + assert adapter.send.call_args.kwargs["content"].startswith( + dl.RECONNECTED_MARKER + ) + assert _row("ob-1")["state"] == "delivered" + + @pytest.mark.asyncio + async def test_runtime_profile_redelivery_uses_matching_bot_adapter(self): + from gateway.config import Platform + + _record( + session_key="agent:reviewer:slack:channel:C1", + platform="slack", + adapter_profile="reviewer", + ) + dl.mark_failed("ob-1", "send_path_degraded") + default_adapter = self._adapter() + reviewer_adapter = self._adapter() + runner = self._runner(default_adapter) + runner._profile_adapters = { + "reviewer": {Platform.SLACK: reviewer_adapter} + } + + n = await runner._redeliver_failed_obligations_for_platform( + Platform.SLACK, profile="reviewer" + ) + + assert n == 1 + default_adapter.send.assert_not_awaited() + reviewer_adapter.send.assert_awaited_once() + + @pytest.mark.asyncio + async def test_runtime_missing_adapter_releases_unsent_claim(self): + from gateway.config import Platform + + _record(platform="slack") + dl.mark_failed("ob-1", "send_path_degraded") + runner = self._runner() + + n = await runner._redeliver_failed_obligations_for_platform(Platform.SLACK) + + assert n == 0 + assert _row("ob-1")["state"] == "failed" + assert _row("ob-1")["attempts"] == 0 + assert _row("ob-1")["last_error"] == "send_path_degraded" + + @pytest.mark.asyncio + async def test_runtime_clear_failure_does_not_send_or_lose_retry(self): + from gateway.config import Platform + + _record(platform="slack") + dl.mark_failed("ob-1", "send_path_degraded") + adapter = self._adapter() + runner = self._runner(adapter) + runner._async_session_store.clear_resume_pending.side_effect = RuntimeError( + "session store unavailable" + ) + + n = await runner._redeliver_failed_obligations_for_platform(Platform.SLACK) + + assert n == 0 + adapter.send.assert_not_awaited() + assert _row("ob-1")["state"] == "failed" + assert _row("ob-1")["attempts"] == 0 + assert _row("ob-1")["last_error"] == "send_path_degraded" + @pytest.mark.parametrize( ("send_success", "ledger_method"), [(True, "mark_delivered"), (False, "mark_failed")], diff --git a/tests/gateway/test_delivery_ledger_producer.py b/tests/gateway/test_delivery_ledger_producer.py index 1071d9f36e..3f339dc61d 100644 --- a/tests/gateway/test_delivery_ledger_producer.py +++ b/tests/gateway/test_delivery_ledger_producer.py @@ -62,7 +62,8 @@ def _event(text="hello agent"): def _rows(): with dl._connect() as conn: return conn.execute( - "SELECT obligation_id, state, content FROM delivery_obligations" + """SELECT obligation_id, state, content, adapter_profile + FROM delivery_obligations""" ).fetchall() @@ -122,6 +123,33 @@ class TestProducerHook: assert len(rows) == 1 assert rows[0][1] == "failed" + @pytest.mark.asyncio + async def test_late_transient_failure_signals_reconnected_runner(self): + """A replacement installed mid-send must trigger another ledger sweep.""" + adapter = _Adapter() + adapter._owner_profile = "reviewer" + replacement = _Adapter() + replacement._owner_profile = "reviewer" + runner = MagicMock() + runner._adapter_for_source.side_effect = [adapter, replacement] + runner._redeliver_failed_obligations_for_platform = AsyncMock(return_value=1) + adapter.gateway_runner = runner + adapter.send = AsyncMock( + return_value=SendResult( + success=False, + error="send_path_degraded", + retryable=True, + ) + ) + + await _run(adapter, _event()) + + assert _rows()[0][1] == "failed" + assert _rows()[0][3] == "reviewer" + runner._redeliver_failed_obligations_for_platform.assert_awaited_once_with( + Platform.SLACK, profile="reviewer" + ) + @pytest.mark.asyncio async def test_slow_ledger_record_does_not_block_event_loop(self): diff --git a/tests/gateway/test_multiplex_adapter_registry.py b/tests/gateway/test_multiplex_adapter_registry.py index d048d9a0ea..3d0c196cbf 100644 --- a/tests/gateway/test_multiplex_adapter_registry.py +++ b/tests/gateway/test_multiplex_adapter_registry.py @@ -178,6 +178,7 @@ def _secondary_recovery_runner(*, running=True): runner._make_adapter_auth_check = lambda platform, profile_name=None: object() runner._adapter_disconnect_timeout_secs = lambda: 0 runner._sync_voice_mode_state_to_adapter = lambda adapter: None + runner._redeliver_failed_obligations_for_platform = AsyncMock(return_value=0) return runner @@ -228,6 +229,15 @@ class TestSecondaryProfileFatalRecovery: return True monkeypatch.setattr(runner, "_connect_adapter_with_timeout", connect) + redelivery_homes = [] + + async def redeliver(platform, *, profile=None): + from hermes_constants import get_hermes_home + + redelivery_homes.append(Path(get_hermes_home())) + return 0 + + runner._redeliver_failed_obligations_for_platform.side_effect = redeliver await runner._handle_profile_adapter_fatal_error( "reviewer", Platform.DISCORD, stale ) @@ -238,8 +248,13 @@ class TestSecondaryProfileFatalRecovery: assert len(tasks) == 1 await tasks[0] assert runner._profile_adapters["reviewer"][Platform.DISCORD] is replacement + runner._redeliver_failed_obligations_for_platform.assert_awaited_once_with( + Platform.DISCORD, profile="reviewer" + ) assert scoped_homes assert all(path == Path("/profiles/reviewer") for path in scoped_homes) + assert redelivery_homes + assert all(path != Path("/profiles/reviewer") for path in redelivery_homes) @pytest.mark.asyncio diff --git a/tests/gateway/test_platform_reconnect.py b/tests/gateway/test_platform_reconnect.py index e0c4e4a955..75029072c8 100644 --- a/tests/gateway/test_platform_reconnect.py +++ b/tests/gateway/test_platform_reconnect.py @@ -227,6 +227,7 @@ class TestPlatformReconnectWatcher: """ runner = _make_runner() runner._sync_voice_mode_state_to_adapter = MagicMock() + runner._redeliver_failed_obligations_for_platform = AsyncMock(return_value=1) runner._schedule_resume_pending_sessions = MagicMock(return_value=1) platform_config = PlatformConfig(enabled=True, token="test") @@ -258,6 +259,9 @@ class TestPlatformReconnectWatcher: await run_one_iteration() assert Platform.TELEGRAM in runner.adapters + runner._redeliver_failed_obligations_for_platform.assert_awaited_once_with( + Platform.TELEGRAM + ) runner._schedule_resume_pending_sessions.assert_called_once_with( platform=Platform.TELEGRAM )