ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

fix(gateway): redeliver transient failures after reconnect

Browse Source
This commit is contained in:
milnerrad
2026-08-26 11:15:57 +08:00
committed by Teknium
parent b455abe0b3
commit 8e1db41041
7 changed files with 726 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/delivery_ledger.py
+182 -10
View File
@@ -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
gateway/platforms/base.py
+34 -1
View File
@@ -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
gateway/run.py
+159 -21
View File
@@ -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.
tests/gateway/test_delivery_ledger.py
+303 -3
View File
@@ -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")],
tests/gateway/test_delivery_ledger_producer.py
+29 -1
View File
@@ -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):
tests/gateway/test_multiplex_adapter_registry.py
+15
View File
@@ -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
tests/gateway/test_platform_reconnect.py
+4
View File
@@ -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
)