fix(hook-context-intelligence): forwarding-path reliability & diagnostics hardening

modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py

@@ -48,6 +48,7 @@
 import fnmatch
 import logging
 from collections.abc import Callable, Coroutine
+from datetime import datetime, timezone
 from typing import Any
 
 log = logging.getLogger(__name__)
@@ -245,29 +246,78 @@ async def on_session_ready(coordinator: Any) -> None:
             active = select_active(destinations, match_key)
 
     # Build one dispatcher per ACTIVE destination (D9).
-    dispatchers = [
-        _DestinationDispatcher(
-            name=d.name,
-            url=d.url,
-            api_key=d.api_key,
-            workspace=resolver.workspace,
-            working_dir=resolver.working_dir,
-            dispatch_timeout=resolver.dispatch_timeout,
-            read_timeout=resolver.dispatch_read_timeout,
-            connect_timeout=resolver.dispatch_connect_timeout,
-            failure_threshold=resolver.dispatch_failure_threshold,
-            queue_capacity=resolver.dispatch_queue_capacity,
-            close_drain_timeout=resolver.close_drain_timeout,
-            backoff_initial=resolver.dispatch_backoff_initial,
-            backoff_max=resolver.dispatch_backoff_max,
-            backoff_jitter=resolver.dispatch_backoff_jitter,
-            storage_path=str(resolver.base_path),
-            forwarding_log_dir=resolver.forwarding_log_dir,
-            auth_mode=d.auth_mode,
-            auth_resource=d.auth_resource,
-        )
-        for d in active.values()
-    ]
+    #
+    # Each construction is isolated in its own try/except: _DestinationDispatcher.__init__
+    # calls build_auth_strategy() eagerly, which can raise for an environmental reason
+    # that validate_destinations() does not (and cannot) screen for -- e.g. the
+    # azure-identity credential chain failing to initialise for an "entra" destination.
+    # A bare list comprehension would let ONE such failure abort the whole thing: no
+    # set_dispatchers() call, and -- because the exception would propagate out of
+    # on_session_ready, which the kernel swallows (Phase 6, _session_init.py) -- the
+    # event registration below would never run either, silently disabling ALL capture
+    # including local JSONL. That is the one path where the "no data is lost" guarantee
+    # breaks. So: skip only the failing destination, keep the rest, and always fall
+    # through to set_dispatchers() + event registration. _record_forwarding_issue is an
+    # instance method on the dispatcher we just failed to construct, so there is nothing
+    # to call it on here -- instead we write a durable forwarding-diagnostics record via
+    # the module-level _write_forwarding_record (best-effort, never raises) so the failure
+    # is visible in forwarding-*.jsonl where an operator investigating a forwarding
+    # problem actually looks, not only in the kernel log. The log.error(exc_info=True)
+    # additionally carries the full traceback for deeper diagnosis.
+    from .handlers.logging_handler import _write_forwarding_record
+
+    dispatchers: list[_DestinationDispatcher] = []
+    for d in active.values():
+        try:
+            dispatchers.append(
+                _DestinationDispatcher(
+                    name=d.name,
+                    url=d.url,
+                    api_key=d.api_key,
+                    workspace=resolver.workspace,
+                    working_dir=resolver.working_dir,
+                    dispatch_timeout=resolver.dispatch_timeout,
+                    read_timeout=resolver.dispatch_read_timeout,
+                    connect_timeout=resolver.dispatch_connect_timeout,
+                    failure_threshold=resolver.dispatch_failure_threshold,
+                    queue_capacity=resolver.dispatch_queue_capacity,
+                    close_drain_timeout=resolver.close_drain_timeout,
+                    backoff_initial=resolver.dispatch_backoff_initial,
+                    backoff_max=resolver.dispatch_backoff_max,
+                    backoff_jitter=resolver.dispatch_backoff_jitter,
+                    storage_path=str(resolver.base_path),
+                    forwarding_log_dir=resolver.forwarding_log_dir,
+                    auth_mode=d.auth_mode,
+                    auth_resource=d.auth_resource,
+                )
+            )
+        except Exception as exc:
+            # Durable forwarding-diagnostics record so the dropped destination is
+            # visible in forwarding-*.jsonl (where an operator investigating a
+            # forwarding problem looks), not only in the kernel log. Best-effort:
+            # _write_forwarding_record never raises. http_status is null (this is
+            # not an HTTP response); detail carries the exception type + message.
+            _write_forwarding_record(
+                resolver.forwarding_log_dir,
+                {
+                    "ts": datetime.now(timezone.utc).isoformat(),
+                    "destination": d.name,
+                    "url": d.url,
+                    "kind": "dispatcher_construction_failed",
+                    "http_status": None,
+                    "session_id": "",
+                    "workspace": resolver.workspace or "",
+                    "detail": (f"dispatcher construction failed: {type(exc).__name__}: {exc}"),
+                },
+            )
+            # Full traceback for deeper diagnosis (console/kernel log only).
+            log.error(
+                "context-intelligence: destination %r failed to construct its dispatcher "
+                "(auth strategy init failed) -- dispatch disabled for this destination "
+                "only; local JSONL and any other configured destinations are unaffected.",
+                d.name,
+                exc_info=True,
+            )
     await logging_handler.set_dispatchers(dispatchers)
 
     # --- Fan-out log line (S2) ---

modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py

@@ -164,11 +164,15 @@ def _classify_http_outcome(status_code: int) -> str:
         # Do NOT follow — silently following an authenticated POST redirect risks
         # leaking the bearer token to a different host.
         return _PERMANENT
-    if status_code == 401 or status_code == 429 or status_code >= 500:
+    if status_code == 401 or status_code == 408 or status_code == 429 or status_code >= 500:
+        # 408 (Request Timeout) is transient by definition — the request did not
+        # arrive/complete in time; retrying with backoff is the correct response,
+        # exactly like a 5xx or a 429. It is enumerated explicitly here rather than
+        # left to fall through to the permanent catch-all below.
         return _TRANSIENT
     # 403, 400, 413, 422, 404, 410, and any other 4xx — all non-retryable, but
     # NOT all "malformed"; see the message-layer branches in _worker for the
-    # per-status cause.
+    # per-status cause. (408 is deliberately NOT here — see the transient set above.)
     return _PERMANENT
 
 
@@ -382,6 +386,28 @@ def __init__(
         # attempt so a stale value can never be inherited by an unrelated
         # outcome.
         self._auth_token_failed: bool = False
+        # Set True once the first successful delivery of this session emits a
+        # positive `delivery_ok` liveness record (D3). One heartbeat per
+        # dispatcher lifetime (per session); routine subsequent deliveries stay
+        # silent so the diagnostics file is not flooded on the happy path.
+        self._heartbeat_emitted: bool = False
+
+    def _emit_delivery_heartbeat(self) -> None:
+        """Emit a one-time positive liveness record on first successful delivery.
+
+        The forwarding-diagnostics sink otherwise only ever writes on a PROBLEM
+        (breaker_open, permanent_reject, auth_token_unavailable, ...), so an empty
+        log is structurally ambiguous between "healthy and delivering", "never
+        started", and "no sessions ran". A single `delivery_ok` record per session,
+        written the first time a destination actually delivers, disambiguates the
+        empty-log case: no record means no delivery happened, not a silent outage.
+        Fires at most once per dispatcher (per session); best-effort and never
+        raises into the delivery path.
+        """
+        if self._heartbeat_emitted:
+            return
+        self._heartbeat_emitted = True
+        self._record_forwarding_issue("delivery_ok", "first successful delivery this session")
 
     def _ensure_worker(self) -> None:
         if self._worker_task is None or self._worker_task.done():
@@ -703,6 +729,7 @@ async def _worker(self) -> None:
                     outcome = await self._post(event, payload_data)
                     if outcome == _DELIVERED:
                         self._breaker_record_delivered()
+                        self._emit_delivery_heartbeat()
                         self._degraded_warned = False
                         self._degraded_since = None
                     elif outcome == _TRANSIENT and self._is_hard_outcome():
@@ -837,6 +864,7 @@ async def _worker(self) -> None:
                     # _DELIVERED or _PERMANENT — advance to next event
                     if outcome == _DELIVERED:
                         self._breaker_record_delivered()
+                        self._emit_delivery_heartbeat()
                         if self._degraded_warned:
                             logger.info(
                                 "Reconnected to %s — resuming delivery.",
@@ -1065,6 +1093,24 @@ async def _post(self, event: str, data: dict[str, Any]) -> str:
             # credential problem instead of retrying forever in silence.
             self._last_status = None
             self._auth_token_failed = True
+            # Durable diagnostic record: written on EVERY auth-token failure, NOT
+            # gated by the console rate-limit below, and carrying the exception TYPE
+            # and MESSAGE. Previously this was a hardcoded "auth token production
+            # failed" string written from inside the rate-limit branch, so every
+            # distinct fault -- expired token, wrong audience, broker unavailable,
+            # a masked TypeError -- collapsed to one byte-identical, uninformative
+            # record (and a burst within the rate-limit window dropped all but the
+            # first). Now each fault is individually diagnosable in the forwarding
+            # JSONL. Write volume is bounded by dispatch backoff, which throttles
+            # how often _post is attempted; _record_forwarding_issue is best-effort
+            # and never raises into this path.
+            self._record_forwarding_issue(
+                "auth_token_unavailable",
+                f"auth token production failed: {type(exc).__name__}: {exc}",
+            )
+            # Console logging (WARNING + DEBUG traceback) stays rate-limited to avoid
+            # spamming the log during a sustained outage; the durable record above is
+            # the diagnostic of record and is intentionally not throttled.
             now = time.monotonic()
             if now - self._last_headers_error_log >= _LOG_RATE_LIMIT_SECONDS:
                 self._last_headers_error_log = now
@@ -1076,9 +1122,6 @@ async def _post(self, event: str, data: dict[str, Any]) -> str:
                     self._url,
                     type(exc).__name__,
                 )
-                self._record_forwarding_issue(
-                    "auth_token_unavailable", "auth token production failed"
-                )
                 logger.debug(
                     "%s auth-header production failed: %r",
                     self._name,

modules/hook-context-intelligence/tests/test_classification.py

@@ -134,9 +134,9 @@ async def test_client_closed_runtime_error_returns_delivered(self) -> None:
 
 
 class TestTransientHttp:
-    """HTTP 5xx, 429, and 401 return _TRANSIENT."""
+    """HTTP 5xx, 429, 401, and 408 return _TRANSIENT."""
 
-    @pytest.mark.parametrize("status_code", [401, 429, 500, 502, 503])
+    @pytest.mark.parametrize("status_code", [401, 408, 429, 500, 502, 503])
     async def test_transient_status_codes(self, status_code: int) -> None:
         d = _dispatcher()
         d._client = _mock_client_for_response(_mock_response(status_code))
@@ -145,7 +145,7 @@ async def test_transient_status_codes(self, status_code: int) -> None:
 
         assert result == _TRANSIENT
 
-    @pytest.mark.parametrize("status_code", [401, 429, 500, 502, 503])
+    @pytest.mark.parametrize("status_code", [401, 408, 429, 500, 502, 503])
     async def test_transient_sets_last_status(self, status_code: int) -> None:
         d = _dispatcher()
         d._client = _mock_client_for_response(_mock_response(status_code))
@@ -154,6 +154,18 @@ async def test_transient_sets_last_status(self, status_code: int) -> None:
 
         assert d._last_status == status_code
 
+    async def test_408_is_transient_not_permanent(self) -> None:
+        """Issue #431 D1: 408 Request Timeout is transient by definition and must
+        be retried, not skipped. Regression guard against it falling through the
+        transient check into the _PERMANENT catch-all (as it did before the fix)."""
+        d = _dispatcher()
+        d._client = _mock_client_for_response(_mock_response(408))
+
+        result = await d._post("test:event", {"session_id": "s1"})
+
+        assert result == _TRANSIENT
+        assert result != _PERMANENT
+
 
 # ---------------------------------------------------------------------------
 # httpx network exceptions → _TRANSIENT

modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py

@@ -17,6 +17,7 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 from amplifier_module_hook_context_intelligence.handlers.logging_handler import (
+    _TRANSIENT,
     _DestinationDispatcher,
     _write_forwarding_record,
 )
@@ -223,4 +224,127 @@ async def test_repeated_401s_skip_per_event_without_opening_breaker_short_run(
         records = [json.loads(line) for line in log_file.read_text().splitlines()]
         kinds = {r["kind"] for r in records}
         assert "breaker_open" not in kinds, f"breaker must not have opened: {kinds}"
+
+
+class TestAuthTokenUnavailableDurableRecord:
+    """Issue #431 D2: the durable auth_token_unavailable record must carry the
+    exception TYPE and MESSAGE (not a byte-identical constant), and distinct
+    failures must not be collapsed by the console rate-limit.
+
+    Drives the auth-strategy header-production failure path directly: a mocked
+    _strategy.headers() that raises is exactly what an expired `az login` /
+    broken credential chain produces at runtime (static ApiKeyAuth never raises,
+    so this path is entra-only).
+    """
+
+    async def test_record_carries_exception_type_and_message(self, tmp_path: Path) -> None:
+        d = _dispatcher(forwarding_log_dir=tmp_path)
+        d._client = MagicMock()  # headers() raises before any request is issued
+        d._strategy = MagicMock()
+        d._strategy.headers.side_effect = RuntimeError("token expired: refresh needed")
+
+        with patch(LOGGER_PATH):
+            result = await d._post("evt:x", {"session_id": "sess-9"})
+
+        assert result == _TRANSIENT  # auth failure stays on the retry path
+        log_file = tmp_path / f"forwarding-{_today_utc()}.jsonl"
+        records = [json.loads(line) for line in log_file.read_text().splitlines()]
+        rec = next(r for r in records if r["kind"] == "auth_token_unavailable")
+        # http_status None distinguishes this from a real HTTP 401 auth_failure.
+        assert rec["http_status"] is None
+        assert "RuntimeError" in rec["detail"], f"exception type missing: {rec['detail']!r}"
+        assert "token expired: refresh needed" in rec["detail"], (
+            f"exception message missing: {rec['detail']!r}"
+        )
+
+    async def test_distinct_failures_each_recorded_not_collapsed(self, tmp_path: Path) -> None:
+        """Two distinct auth faults in quick succession (within the console
+        rate-limit window) must each produce their own durable record -- the
+        durable write is no longer gated by the rate-limit branch."""
+        d = _dispatcher(forwarding_log_dir=tmp_path)
+        d._client = MagicMock()
+        d._strategy = MagicMock()
+        d._strategy.headers.side_effect = [
+            RuntimeError("expired token"),
+            ValueError("wrong audience"),
+        ]
+
+        with patch(LOGGER_PATH):
+            await d._post("e1", {"session_id": "s1"})
+            await d._post("e2", {"session_id": "s1"})
+
+        log_file = tmp_path / f"forwarding-{_today_utc()}.jsonl"
+        records = [json.loads(line) for line in log_file.read_text().splitlines()]
+        auth_recs = [r for r in records if r["kind"] == "auth_token_unavailable"]
+        assert len(auth_recs) == 2, f"expected 2 distinct durable records, got {len(auth_recs)}"
+        blob = " || ".join(r["detail"] for r in auth_recs)
+        assert "RuntimeError" in blob and "expired token" in blob
+        assert "ValueError" in blob and "wrong audience" in blob
+
+
+class TestDeliveryHeartbeat:
+    """Issue #431 D3: a destination that delivers successfully must emit a
+    positive `delivery_ok` liveness record once per session, so an EMPTY
+    forwarding log unambiguously means "no delivery happened" rather than
+    "possibly a silent outage". The sink otherwise only ever writes on problems.
+    """
+
+    async def test_first_successful_delivery_writes_heartbeat(self, tmp_path: Path) -> None:
+        d = _dispatcher(forwarding_log_dir=tmp_path)
+        d._client = _mock_client([_make_response(200)])
+        d._sleep_backoff = AsyncMock()  # type: ignore[method-assign]
+
+        with patch(LOGGER_PATH):
+            d.enqueue("e1", {"session_id": "sess-1"})
+            await asyncio.wait_for(d._queue.join(), timeout=2.0)
+
+        log_file = tmp_path / f"forwarding-{_today_utc()}.jsonl"
+        assert log_file.exists(), "a healthy delivery must still write a liveness record"
+        records = [json.loads(line) for line in log_file.read_text().splitlines()]
+        heartbeats = [r for r in records if r["kind"] == "delivery_ok"]
+        assert len(heartbeats) == 1, f"expected exactly one delivery_ok record, got {heartbeats}"
+        assert heartbeats[0]["destination"] == "test-dest"
+        await d.close()
+
+    async def test_heartbeat_emitted_only_once_per_session(self, tmp_path: Path) -> None:
+        """Multiple successful deliveries in one session produce exactly ONE
+        heartbeat -- the happy path is not flooded with liveness records."""
+        d = _dispatcher(forwarding_log_dir=tmp_path)
+        d._client = _mock_client([_make_response(200)] * 5)
+        d._sleep_backoff = AsyncMock()  # type: ignore[method-assign]
+
+        with patch(LOGGER_PATH):
+            for i in range(5):
+                d.enqueue(f"e{i}", {"session_id": "sess-1"})
+            await asyncio.wait_for(d._queue.join(), timeout=3.0)
+
+        log_file = tmp_path / f"forwarding-{_today_utc()}.jsonl"
+        records = [json.loads(line) for line in log_file.read_text().splitlines()]
+        heartbeats = [r for r in records if r["kind"] == "delivery_ok"]
+        assert len(heartbeats) == 1, (
+            f"heartbeat must fire at most once per session, got {len(heartbeats)}"
+        )
+        await d.close()
+
+    async def test_no_delivery_no_heartbeat(self, tmp_path: Path) -> None:
+        """A destination that never delivers writes no delivery_ok record -- the
+        empty/heartbeat-less case is what makes the signal meaningful."""
+        d = _dispatcher(forwarding_log_dir=tmp_path, failure_threshold=1)
+        d._client = _mock_client([_make_response(403)])  # permanent skip, never delivered
+        d._sleep_backoff = AsyncMock()  # type: ignore[method-assign]
+
+        with patch(LOGGER_PATH):
+            d.enqueue("e1", {"session_id": "sess-1"})
+            await asyncio.wait_for(d._queue.join(), timeout=2.0)
+
+        log_file = tmp_path / f"forwarding-{_today_utc()}.jsonl"
+        records = (
+            [json.loads(line) for line in log_file.read_text().splitlines()]
+            if log_file.exists()
+            else []
+        )
+        assert not [r for r in records if r["kind"] == "delivery_ok"], (
+            "no successful delivery must mean no delivery_ok heartbeat"
+        )
+        await d.close()
         await d.close()