fix(logs): report a multiplexer outage once, with its cause

src/ccgram/handlers/polling/polling_coordinator.py

@@ -6,12 +6,12 @@
 import structlog
 from telegram.error import TelegramError
 
 from ...thread_router import chat_scope, thread_router
 from ...multiplexer import multiplexer as tmux_manager
 from ...multiplexer.base import canonical_window_id
 from ...multiplexer.reconciliation import list_windows_for_reconciliation
 from ...multiplexer.window_liveness import note_live_windows
 from ...utils import log_throttled
 from . import window_tick
 from .polling_runtime import PollingRuntime
 
@@ -25,6 +25,7 @@
 
 _BACKOFF_MIN = 2.0
 _BACKOFF_MAX = 30.0
+_UNAVAILABLE = "Status poll skipped: window listing unavailable"
 
 _LoopError = (TelegramError, OSError, RuntimeError, ValueError)
 
@@ -69,7 +70,7 @@
 async def status_poll_loop(bot: "Bot", adapter: "TelegramSessionAdapter") -> None:
     """Background task to poll terminal status for all thread-bound windows."""
     # Lazy: imports keep PTB out of the polling package's cold path.
     from ...config import config as _cfg
 
     # Lazy: PTBTelegramClient wraps the live PTB bot — resolved per-tick
     from ...telegram_client import PTBTelegramClient
@@ -89,7 +90,7 @@
         try:
             all_windows = await list_windows_for_reconciliation(tmux_manager)
             if all_windows is None:
-                logger.warning("Status poll skipped: window listing unavailable")
+                log_throttled(logger, "poll-listing", _UNAVAILABLE, level="warning")
                 await asyncio.sleep(poll_interval)
                 continue
             note_live_windows(all_windows, thread_router.all_bound_window_ids())

src/ccgram/multiplexer/agterm.py

@@ -50,6 +50,7 @@
 
 import structlog
 
+from ..utils import log_throttled
 from .agterm_panes import pane_sessions, split_session_id
 from .base import (
     AgentStatus,
@@ -191,7 +192,7 @@
         return False
     # Lazy: providers reach back into the multiplexer seam, so a module-level
     # import here would close the loop.
     from ..providers.process_detection import classify_provider_from_argv
 
     provider = classify_provider_from_argv([str(part) for part in foreground])
     return bool(provider) and provider != "shell"
@@ -465,11 +466,26 @@
         """
         window_ids = await self._open_window_ids()
         if window_ids is None:
+            log_throttled(
+                logger,
+                "agterm-window-list",
+                "agterm window list failed; window listing unconfirmed",
+                cooldown=60.0,
+                level="warning",
+            )
             return None
         workspaces: list[dict] = []
         for window_id in window_ids:
             result = await self._call(["tree", "--window", window_id])
             if result is None:
+                log_throttled(
+                    logger,
+                    f"agterm-tree:{window_id}",
+                    "agterm tree read failed for window %s; window listing unconfirmed",
+                    window_id,
+                    cooldown=60.0,
+                    level="warning",
+                )
                 return None
             tree = result.get("tree")
             if not isinstance(tree, dict):
@@ -501,7 +517,7 @@
             command = str(foreground[0])
         elif isinstance(shell := session.get("foregroundShell"), str):
             # Lazy: the providers registry imports the multiplexer during setup.
             from ..providers.shell_infra import KNOWN_SHELLS
 
             if shell in KNOWN_SHELLS:
                 command = shell

src/ccgram/session_monitor.py

@@ -26,7 +26,7 @@
 
 from telegram.error import TelegramError
 
 from .config import config
 from .delivery_contract import (
     DeliveryReceipt,
     activate_delivery_receipt,
@@ -35,10 +35,10 @@
     settled_prefix,
     settled_run_offset,
 )
 from .event_reader import read_new_events
 from .idle_tracker import IdleTracker
 from .monitor_state import BacklogSkipIntent, MonitorState, TrackedSession
 from .providers import get_provider_for_window, registry  # noqa: F401 (used by test patches)
 from .session_map import (
     acknowledge_replay_from_start,
     observed_provider,
@@ -47,18 +47,18 @@
     session_map_prefix,
     strip_session_map_prefix,
 )
 from .window_state_ports import identity_state
 from .session_lifecycle import session_lifecycle
 from .multiplexer import multiplexer as tmux_manager
 from .multiplexer.base import canonical_window_id
 from .multiplexer.reconciliation import list_windows_for_reconciliation
 from .multiplexer.window_liveness import note_live_windows
 from .multiplexer.topic_mapping import is_agent_topic_window
 from .monitor_events import NewMessage, NewWindowEvent, SessionInfo
 from .transcript_reader import TranscriptReader, render_transcript_batch
 from .session_protocol.events import EventHub
 from .session_protocol.messages import TranscriptBatch
-from .utils import task_done_callback
+from .utils import log_throttled, task_done_callback
 
 import json
 
@@ -127,7 +127,7 @@
         # Lazy: providers.base imports HookEvent and gets imported back
         # through tmux_manager → providers; keep at call site.
         # Lazy: HookEvent pulled by hook dispatch path; defer until that path runs
         from .providers.base import HookEvent
 
         self._hook_event_callback: Callable[[HookEvent], Awaitable[None]] | None = None
 
@@ -791,7 +791,7 @@
         # Lazy: thread_router is wired into session_manager which imports
         # session_monitor; hoisting forms a startup cycle.
         # Lazy: proxies wired by SessionManager constructor
         from .thread_router import thread_router
 
         for window_id, details in result.changed_windows.items():
             if not thread_router.has_window(window_id):
@@ -865,7 +865,7 @@
             return
         # Lazy: thread_router is wired into session_manager which imports
         # session_monitor; hoisting forms a startup cycle.
         from .thread_router import thread_router
 
         caps = tmux_manager.capabilities
         known_lookup = _adoption_lookup(known_window_ids)
@@ -918,7 +918,7 @@
             return
         # Lazy: thread_router is wired into session_manager which imports
         # session_monitor; hoisting forms a startup cycle.
         from .thread_router import thread_router
 
         bound_window_ids = {wid for _, _, wid in thread_router.iter_thread_bindings()}
         bound_lookup = _adoption_lookup(bound_window_ids)
@@ -1035,8 +1035,13 @@
                     else {w.window_id for w in all_windows if w.topic_eligible}
                 )
                 if all_windows is None:
-                    logger.warning(
-                        "Multiplexer listing unavailable; skipping window reconciliation"
+                    log_throttled(
+                        logger,
+                        "monitor-window-reconciliation-listing",
+                        "Multiplexer listing unavailable; "
+                        "skipping window reconciliation",
+                        cooldown=60.0,
+                        level="warning",
                     )
                 else:
                     # Before anything keys off these ids, let the backend
@@ -1049,7 +1054,7 @@
                     from .session import session_manager as _sm
 
                     # Lazy: thread routing imports monitor-facing helpers.
                     from .thread_router import thread_router
 
                     _sm.reconcile_window_aliases(all_windows)
                     note_live_windows(all_windows, thread_router.all_bound_window_ids())

src/ccgram/session_protocol/runtime.py

@@ -12,7 +12,7 @@
 from pathlib import Path
 from typing import IO
 
 from ..event_reader import read_new_events
 from ..multiplexer.base import Multiplexer
 from ..providers.base import HookEvent
 from .backend import MultiplexerSessionBackend
@@ -57,15 +57,27 @@
             logger.error("Protocol hook queue full; affected approvals require restart")
 
     async def poll(self, prefix: str) -> None:
+        unavailable = False
         while True:
             try:
                 await self.service.refresh()
+                if unavailable:
+                    unavailable = False
+                    logger.info("Session protocol source recovered")
                 while not self.hooks.empty():
                     event, sessions = self.hooks.get_nowait()
                     await self.service.observe_hook(event, prefix)
                     self.service.resolve_hook(sessions)
-            except ProtocolError, OSError:
-                logger.warning("Session protocol source unavailable", exc_info=True)
+            except (ProtocolError, OSError) as exc:
+                # A multiplexer that is briefly unreachable or restarting is
+                # expected, and the next pass retries. Log the incident's start
+                # and end instead of one traceback per second: a sustained
+                # outage buried real errors under thousands of identical stacks.
+                if not unavailable:
+                    unavailable = True
+                    logger.warning(
+                        "Session protocol source unavailable: %s", exc, exc_info=True
+                    )
             await asyncio.sleep(1)
 
     async def close(self) -> None:

tests/ccgram/handlers/polling/test_polling_coordinator.py

@@ -156,6 +156,9 @@ async def _stop_sleep(_delay: float) -> None:
         ctx.mocks["tick_window"].assert_not_called()
         ctx.mocks["run_periodic"].assert_not_called()
         ctx.mocks["run_lifecycle"].assert_not_called()
+        # Every tick hits the same outage; the skip notice is throttled per key.
+        ctx.mocks["log_throttled"].assert_called_once()
+        assert ctx.mocks["log_throttled"].call_args.args[1] == "poll-listing"
 
 
 class TestStatusPollLoopPassesWindowLookup:

tests/ccgram/session_protocol/test_runtime_outage_logging.py

@@ -0,0 +1,64 @@
+"""A backend outage must log its edges, not once per retry.
+
+The poll loop retries every second. Logging a full traceback for every failed
+attempt buried the journal under identical stacks during a one-minute
+multiplexer hiccup and hid the errors around it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from tempfile import TemporaryFile
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from ccgram.session_protocol.models import ProtocolError
+from ccgram.session_protocol.runtime import ProtocolRuntime
+
+
+def _sleeper(iterations: int) -> AsyncMock:
+    """Sleep *iterations* times, then cancel the loop like a shutdown would."""
+    return AsyncMock(side_effect=[None] * iterations + [asyncio.CancelledError()])
+
+
+async def test_repeated_outage_logs_one_warning(service, backend, store, monkeypatch):
+    backend.unavailable = True
+    logger = MagicMock()
+    with TemporaryFile() as lock:
+        monkeypatch.setattr("ccgram.session_protocol.runtime.logger", logger)
+        monkeypatch.setattr(
+            "ccgram.session_protocol.runtime.asyncio.sleep", _sleeper(4)
+        )
+        runtime = ProtocolRuntime(store, service, lock)
+        with pytest.raises(asyncio.CancelledError):
+            await runtime.poll("")
+    assert logger.warning.call_count == 1
+    message, error = logger.warning.call_args.args
+    assert "source unavailable" in message
+    assert "offline" in str(error)
+
+
+async def test_recovery_logs_once_after_an_outage(service, store, monkeypatch):
+    attempts = {"count": 0}
+
+    async def refresh() -> None:
+        attempts["count"] += 1
+        if attempts["count"] == 1:
+            raise ProtocolError(503, "backend_unavailable", "offline")
+
+    monkeypatch.setattr(service, "refresh", refresh)
+    logger = MagicMock()
+    with TemporaryFile() as lock:
+        monkeypatch.setattr("ccgram.session_protocol.runtime.logger", logger)
+        monkeypatch.setattr(
+            "ccgram.session_protocol.runtime.asyncio.sleep", _sleeper(3)
+        )
+        runtime = ProtocolRuntime(store, service, lock)
+        with pytest.raises(asyncio.CancelledError):
+            await runtime.poll("")
+    assert logger.warning.call_count == 1
+    recovered = [
+        call for call in logger.info.call_args_list if "recovered" in call.args[0]
+    ]
+    assert len(recovered) == 1

tests/ccgram/test_agterm_backend.py

@@ -15,11 +15,14 @@
 import asyncio
 import json
 from collections.abc import Sequence
+from unittest.mock import MagicMock
 
 import pytest
 
+from ccgram.multiplexer import agterm as agterm_module
 from ccgram.multiplexer.agterm import AgtermManager, _key_to_bytes
 from ccgram.multiplexer.base import AgentStatus, Multiplexer
+from ccgram.utils import log_throttle_reset
 
 
 class FakeAgtermctl:
@@ -482,6 +485,44 @@ async def test_non_json_output_is_a_failure() -> None:
     assert await _manager(fake).list_windows() == []
 
 
+async def test_a_failed_window_list_names_its_cause_once(monkeypatch) -> None:
+    """The listing failure must say which agterm call failed, and say it once.
+
+    Several pollers ask for this listing every second, so an unreachable
+    backend would otherwise repeat the same line forever.
+    """
+    log_throttle_reset("agterm-")
+    logger = MagicMock()
+    monkeypatch.setattr(agterm_module, "logger", logger)
+    fake = FakeAgtermctl().on("window", "list", rc=1, err="agterm not running")
+
+    manager = _manager(fake)
+    assert await manager.list_windows_for_reconciliation() is None
+    assert await manager.list_windows_for_reconciliation() is None
+
+    assert logger.warning.call_count == 1
+    assert "window list failed" in logger.warning.call_args.args[0]
+
+
+async def test_a_failed_tree_read_names_the_window(monkeypatch) -> None:
+    log_throttle_reset("agterm-")
+    logger = MagicMock()
+    monkeypatch.setattr(agterm_module, "logger", logger)
+    fake = (
+        FakeAgtermctl()
+        .on("window", "list", out=_windows())
+        .on("tree", rc=1, err="window not open")
+    )
+
+    manager = _manager(fake)
+    assert await manager.list_windows_for_reconciliation() is None
+
+    assert logger.warning.call_count == 1
+    message = logger.warning.call_args.args[0] % logger.warning.call_args.args[1:]
+    assert "tree read failed" in message
+    assert WINDOW_A in message
+
+
 async def test_send_to_a_vanished_session_reports_failure() -> None:
     """agterm rejects an unknown --target itself, so there is no preflight.
 

tests/ccgram/test_session_monitor.py

@@ -3650,6 +3650,35 @@ async def spy(raw=None, *, adoptable_window_ids):
             "the loop must hand over the eligible subset, not every live window"
         )
 
+    async def test_an_unavailable_listing_goes_through_the_throttle(
+        self, monitor: SessionMonitor, monkeypatch
+    ) -> None:
+        notices: list[tuple[str, str]] = []
+
+        def spy(log, key, msg, *args, **kwargs):
+            notices.append((key, msg))
+
+        async def stop(raw=None, *, adoptable_window_ids):
+            monitor._running = False
+            return {}
+
+        monkeypatch.setattr(
+            "ccgram.session_monitor.list_windows_for_reconciliation",
+            AsyncMock(return_value=None),
+        )
+        monkeypatch.setattr("ccgram.session_monitor.log_throttled", spy)
+        monkeypatch.setattr(
+            "ccgram.session_monitor.read_session_map_raw", AsyncMock(return_value={})
+        )
+        monkeypatch.setattr(monitor, "_read_hook_events", AsyncMock())
+        monkeypatch.setattr(monitor, "_cleanup_all_stale_sessions", AsyncMock())
+        monkeypatch.setattr(monitor, "_detect_and_cleanup_changes", stop)
+        monitor._running = True
+
+        await monitor._monitor_loop()
+
+        assert [key for key, _ in notices] == ["monitor-window-reconciliation-listing"]
+
 
 class TestAdoptionIdentityFoldsCaseInTheMonitor:
     """Session-map keys are written by the hook; the adoption set comes from