From 7370d38dd054d098226e46cf0235bfbee4d93b72 Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 17 Aug 2026 16:55:00 +0000 Subject: [PATCH 1/6] chore(hook-context-intelligence): silence pyright on ALL_EVENTS test import MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit test_mount_dispatcher.py imported amplifier_core.events.ALL_EVENTS without the `# type: ignore[import-not-found]` suppression the module's own source (__init__.py) already uses. ALL_EVENTS exists at runtime (real attribute); pyright just cannot resolve amplifier_core.events statically. Matches the existing in-repo convention. No behavior change. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../hook-context-intelligence/tests/test_mount_dispatcher.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/modules/hook-context-intelligence/tests/test_mount_dispatcher.py b/modules/hook-context-intelligence/tests/test_mount_dispatcher.py index 99c8e021..88cb7fc6 100644 --- a/modules/hook-context-intelligence/tests/test_mount_dispatcher.py +++ b/modules/hook-context-intelligence/tests/test_mount_dispatcher.py @@ -11,7 +11,7 @@ from typing import Any from unittest.mock import AsyncMock, MagicMock -from amplifier_core.events import ALL_EVENTS +from amplifier_core.events import ALL_EVENTS # type: ignore[import-not-found] # --------------------------------------------------------------------------- From a63fdd5daa22f79e765a00b7927f42cc8db581aa Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 17 Aug 2026 16:55:10 +0000 Subject: [PATCH 2/6] fix(hook-context-intelligence): isolate per-destination dispatcher construction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit on_session_ready built dispatchers via a bare list comprehension over active destinations. If _DestinationDispatcher.__init__ raised while constructing any one destination -- build_auth_strategy() failing for an environmental reason validate_destinations() cannot screen (e.g. the azure-identity credential chain failing to initialise for an entra destination) -- the whole comprehension aborted: set_dispatchers() was never reached and, because the kernel swallows on_session_ready exceptions, event registration never ran either. One misconfigured destination silently disabled forwarding for EVERY destination, including the healthy ones, with no diagnostic anywhere. Wrap each construction in its own try/except: skip only the failing destination, keep the rest, and always fall through to set_dispatchers() + event registration. Logged at module level (log.error, exc_info=True) because _record_forwarding_issue is an instance method on the dispatcher that failed to construct. This completes PR #85. #85 made validate_destinations() degrade per-destination at the config-validation layer so a misconfigured destination cannot take down local capture; this applies the same per-destination isolation one layer deeper -- at dispatcher construction -- which #85 left unguarded. Adds TestDispatcherConstructionFailure: on_session_ready does not raise, local capture survives, and a valid sibling destination still gets its dispatcher when another destination's auth construction fails. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../__init__.py | 70 +++++++---- .../tests/test_on_session_ready_routing.py | 117 +++++++++++++++++- 2 files changed, 163 insertions(+), 24 deletions(-) diff --git a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py index 070fe9e0..94e3856a 100644 --- a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py +++ b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py @@ -245,29 +245,53 @@ 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 -- a module-level log.error(exc_info=True) is the durable signal. + 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: + 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) --- diff --git a/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py b/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py index 1ee63c86..0e50598c 100644 --- a/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py +++ b/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py @@ -4,7 +4,7 @@ import logging from typing import Any -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -313,3 +313,118 @@ async def test_mount_does_not_raise_with_url_but_no_key( "expected a discoverable WARNING naming the missing api_key" ) await cleanup() + + +class TestDispatcherConstructionFailure: + """Issue #431 D4: dispatcher CONSTRUCTION (not config validation) blowing up + must degrade to local-only, never take down the whole fleet or local JSONL. + + validate_destinations() screens bad *config* before construction, but + _DestinationDispatcher.__init__ calls build_auth_strategy() eagerly, which + can raise for an ENVIRONMENTAL reason validation cannot foresee (e.g. the + azure-identity credential chain failing to initialise for an "entra" + destination). Before the fix this raised out of the bare list comprehension + in on_session_ready -> set_dispatchers() was never reached AND, because the + kernel swallows on_session_ready exceptions, event registration below never + ran either -- silently disabling ALL capture including local JSONL. This is + the one path where the "no data is lost" guarantee broke. + + Mirrors TestWorkingDirRequirement::test_absent_working_dir_degrades_to_local_only + for the build_auth_strategy() failure path (which had zero coverage). + """ + + async def test_build_auth_strategy_failure_degrades_to_local_only_not_raise( + self, + ) -> None: + """A single destination whose auth-strategy construction raises must NOT + propagate: on_session_ready completes and dispatchers is empty (local-only) + rather than the exception aborting the whole callback.""" + config = { + "destinations": { + "only": { + "url": "http://only:8000", + "api_key": "k", + "auth_mode": "entra", + "auth_resource": "api://only", + "include": ["**"], + }, + } + } + + def _boom(*_args: Any, **_kwargs: Any) -> Any: + raise RuntimeError("credential chain unavailable") + + with patch("context_intelligence.auth.build_auth_strategy", side_effect=_boom): + # Must NOT raise -- a raise here would be swallowed by the kernel and + # abort event registration, killing local JSONL for the session. + _, handler, cleanup = await _mount_and_ready(config) + + assert handler._dispatchers == [], ( + "construction failure must yield zero dispatchers (local-only)" + ) + await cleanup() + + async def test_local_capture_survives_dispatcher_construction_failure(self) -> None: + """The critical regression assertion the suite lacked: even when the only + destination's dispatcher construction raises, the LoggingHandler is still + registered for events -- local events.jsonl capture is NOT disabled.""" + config = { + "destinations": { + "only": { + "url": "http://only:8000", + "api_key": "k", + "auth_mode": "entra", + "auth_resource": "api://only", + "include": ["**"], + }, + } + } + + def _boom(*_args: Any, **_kwargs: Any) -> Any: + raise RuntimeError("credential chain unavailable") + + with patch("context_intelligence.auth.build_auth_strategy", side_effect=_boom): + coordinator, handler, cleanup = await _mount_and_ready(config) + + assert handler._dispatchers == [] + assert coordinator.hooks.register.called, ( + "LoggingHandler must still be registered for events -- local JSONL capture " + "must survive a dispatcher construction failure" + ) + await cleanup() + + async def test_one_bad_destination_does_not_block_good_sibling_dispatcher(self) -> None: + """Mixed fleet: one destination whose auth construction raises alongside a + valid one -- the valid destination must still receive its dispatcher. + Blast radius is the single failing destination, not the whole fleet.""" + config = { + "destinations": { + "broken": { + "url": "http://broken:8000", + "api_key": "k", + "auth_mode": "entra", + "auth_resource": "api://broken", + "include": ["**"], + }, + "good": { + "url": "http://good:8000", + "api_key": "gk", + "include": ["**"], + }, + } + } + + def _selective(*_args: Any, **kwargs: Any) -> Any: + # Only the entra destination's credential chain fails to construct; + # the static one builds normally. + if kwargs.get("auth_mode") == "entra": + raise RuntimeError("credential chain unavailable") + return MagicMock() + + with patch("context_intelligence.auth.build_auth_strategy", side_effect=_selective): + _, handler, cleanup = await _mount_and_ready(config) + + assert len(handler._dispatchers) == 1, "only the good destination should construct" + assert handler._dispatchers[0]._name == "good" + await cleanup() + await cleanup() From 8b2d58d3cad7f56a7a2809454b4856f100455dda Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 17 Aug 2026 17:06:25 +0000 Subject: [PATCH 3/6] fix(hook-context-intelligence): classify HTTP 408 as transient MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _classify_http_outcome treated only 401/429/>=500 as transient; 408 (Request Timeout) fell through to the _PERMANENT catch-all, so a timed-out request was skipped instead of retried. 408 is transient by definition -- the server gave up waiting for the request to complete; nothing about the payload, auth, or target was rejected, and RFC 7231 explicitly permits repeating the request unchanged. It belongs in the same retry-with-backoff bucket as 429 and 5xx. Add 408 to the transient set and make the enumerating comments truthful. Other status codes are unchanged: 401/429/5xx stay transient; 403/400/413/422/404/410 stay permanent. No overlap with the separate server-maintenance / Retry-After pacing work. Tests: 408 added to both TestTransientHttp parametrize lists plus an explicit test_408_is_transient_not_permanent regression guard. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../handlers/logging_handler.py | 8 ++++++-- .../tests/test_classification.py | 18 +++++++++++++++--- 2 files changed, 21 insertions(+), 5 deletions(-) diff --git a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py index 4f6a5741..eb0c7df7 100644 --- a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py +++ b/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 diff --git a/modules/hook-context-intelligence/tests/test_classification.py b/modules/hook-context-intelligence/tests/test_classification.py index aaf77693..8893e16a 100644 --- a/modules/hook-context-intelligence/tests/test_classification.py +++ b/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 From 12521066b2f4d084551ba979f28664914b1ab57b Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 17 Aug 2026 17:19:57 +0000 Subject: [PATCH 4/6] fix(hook-context-intelligence): make auth-token failures diagnosable in durable record MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When _strategy.headers() raised while producing the Authorization header, the only durable artifact was the byte-identical constant "auth token production failed", written from INSIDE the console rate-limit branch. The exception type reached only the (non-durable, rate-limited) WARNING and the full traceback only DEBUG (suppressed by default). Every distinct auth fault -- expired token, wrong audience, broker unavailable, a masked TypeError -- produced one indistinguishable durable record, and a burst within the rate-limit window collapsed to a single line. Move the durable _record_forwarding_issue write OUTSIDE the rate-limit branch so every failure is recorded, and carry type(exc).__name__ + str(exc) into the record's detail. The console WARNING/DEBUG stay rate-limited to avoid log spam. Write volume is bounded by dispatch backoff. Record schema is unchanged (only detail content); the separate HTTP-401 auth_failure path is untouched. Tests: TestAuthTokenUnavailableDurableRecord -- the durable record carries the exception type + message (and http_status is None, distinguishing it from a real 401), and two distinct faults in quick succession each get their own record rather than collapsing. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../handlers/logging_handler.py | 21 ++++++- .../tests/test_forwarding_diagnostics.py | 57 +++++++++++++++++++ 2 files changed, 75 insertions(+), 3 deletions(-) diff --git a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py index eb0c7df7..21b63f01 100644 --- a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py +++ b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py @@ -1069,6 +1069,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 @@ -1080,9 +1098,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, diff --git a/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py b/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py index 7c71c8a4..02071fbc 100644 --- a/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py +++ b/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,60 @@ 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 await d.close() From 491dacc72096eacb1bbb364c5db655f609373365 Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 17 Aug 2026 17:41:36 +0000 Subject: [PATCH 5/6] feat(hook-context-intelligence): per-session delivery_ok liveness heartbeat MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The forwarding-diagnostics sink only ever wrote on a PROBLEM (breaker_open, permanent_reject, auth_token_unavailable, auth_failure, endpoint_not_found, breaker_close, shutdown_undelivered, sustained_delivery_failure). A destination that delivered successfully wrote nothing, ever, so an empty log was ambiguous between "healthy and delivering", "never started", and "no sessions ran". Add a one-time-per-session delivery_ok record, written the first time a destination actually delivers (via a new _heartbeat_emitted flag + _emit_delivery_heartbeat helper, wired into both the normal DELIVERED branch and the breaker half-open probe-success branch). Fires at most once per dispatcher lifetime so the happy path is not flooded; best-effort, never raises. An empty log now unambiguously means "no delivery happened", not "possibly a silent outage". Tests: TestDeliveryHeartbeat (first delivery writes exactly one delivery_ok; five deliveries still produce exactly one; a never-delivering destination writes none). Also updates test_close_clean_shutdown_writes_no_durable_record to assert its true intent precisely now that the sole happy-path record is the delivery_ok heartbeat. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../handlers/logging_handler.py | 24 +++++++ .../tests/test_forwarding_diagnostics.py | 67 +++++++++++++++++++ .../test_sustained_failure_visibility.py | 13 +++- 3 files changed, 103 insertions(+), 1 deletion(-) diff --git a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py index 21b63f01..9bc5d8fb 100644 --- a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py +++ b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/handlers/logging_handler.py @@ -386,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(): @@ -707,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(): @@ -841,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.", diff --git a/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py b/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py index 02071fbc..cc97d790 100644 --- a/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py +++ b/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py @@ -280,4 +280,71 @@ async def test_distinct_failures_each_recorded_not_collapsed(self, tmp_path: Pat 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() diff --git a/modules/hook-context-intelligence/tests/test_sustained_failure_visibility.py b/modules/hook-context-intelligence/tests/test_sustained_failure_visibility.py index 3e2579f0..1b35d2ab 100644 --- a/modules/hook-context-intelligence/tests/test_sustained_failure_visibility.py +++ b/modules/hook-context-intelligence/tests/test_sustained_failure_visibility.py @@ -375,10 +375,21 @@ async def test_close_clean_shutdown_writes_no_durable_record(self, tmp_path: Pat d.enqueue("e1", {"session_id": "sess-1"}) await asyncio.wait_for(d._queue.join(), timeout=2.0) + records_before_close = _read_records(tmp_path) + with patch(LOGGER_PATH): await d.close() - assert _read_records(tmp_path) == [] + # close() on a clean shutdown must add NO durable record. The one record + # present is the D3 `delivery_ok` heartbeat, written by the worker on the + # first successful delivery (not by close()); it must not be joined by a + # shutdown_undelivered record on a clean drain. + records_after_close = _read_records(tmp_path) + assert records_after_close == records_before_close, ( + "clean close() must not add any durable record" + ) + assert [r for r in records_after_close if r["kind"] == "shutdown_undelivered"] == [] + assert [r["kind"] for r in records_after_close] == ["delivery_ok"] async def test_close_shutdown_record_includes_degraded_seconds_zero_when_never_degraded( self, tmp_path: Path From 23fea72a14eabe91763ddbd4b1f8dad70947551f Mon Sep 17 00:00:00 2001 From: colombod Date: Mon, 17 Aug 2026 20:42:17 +0000 Subject: [PATCH 6/6] fix(hook-context-intelligence): make dispatcher-construction failures durable in forwarding diagnostics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-destination construction guard reported failures only via module-level log.error(exc_info=True) -- the kernel log, not the context-intelligence forwarding-diagnostics JSONL where an operator investigating a forwarding problem looks. That was inconsistent with the rest of this change, which makes the forwarding JSONL the durable source of truth. Also write a durable dispatcher_construction_failed record via the module-level _write_forwarding_record (best-effort, never raises; does not need the failed dispatcher instance), carrying type(exc).__name__ + str(exc) in detail and a null http_status. The log.error(exc_info=True) is kept for the full traceback. Test: test_construction_failure_writes_durable_forwarding_record asserts the dropped destination leaves a dispatcher_construction_failed record whose detail carries the exception type + message. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../__init__.py | 30 +++++++++++++- .../tests/test_on_session_ready_routing.py | 40 +++++++++++++++++++ 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py b/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py index 94e3856a..5c61b401 100644 --- a/modules/hook-context-intelligence/amplifier_module_hook_context_intelligence/__init__.py +++ b/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__) @@ -258,7 +259,13 @@ async def on_session_ready(coordinator: Any) -> None: # 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 -- a module-level log.error(exc_info=True) is the durable signal. + # 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: @@ -284,7 +291,26 @@ async def on_session_ready(coordinator: Any) -> None: auth_resource=d.auth_resource, ) ) - except Exception: + 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 " diff --git a/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py b/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py index 0e50598c..2055fa24 100644 --- a/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py +++ b/modules/hook-context-intelligence/tests/test_on_session_ready_routing.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json import logging from typing import Any from unittest.mock import AsyncMock, MagicMock, patch @@ -427,4 +428,43 @@ def _selective(*_args: Any, **kwargs: Any) -> Any: assert len(handler._dispatchers) == 1, "only the good destination should construct" assert handler._dispatchers[0]._name == "good" await cleanup() + + async def test_construction_failure_writes_durable_forwarding_record( + self, tmp_path: Any + ) -> None: + """Issue #431 D4 follow-up: the dropped destination must leave a DURABLE + dispatcher_construction_failed record in forwarding-*.jsonl (not only the + kernel log), carrying the exception type + message -- visible where an + operator investigating a forwarding problem actually looks.""" + config = { + "forwarding_log_dir": str(tmp_path), + "destinations": { + "only": { + "url": "http://only:8000", + "api_key": "k", + "auth_mode": "entra", + "auth_resource": "api://only", + "include": ["**"], + }, + }, + } + + def _boom(*_args: Any, **_kwargs: Any) -> Any: + raise RuntimeError("credential chain unavailable") + + with patch("context_intelligence.auth.build_auth_strategy", side_effect=_boom): + _, handler, cleanup = await _mount_and_ready(config) + + assert handler._dispatchers == [] # failing destination dropped + files = list(tmp_path.glob("forwarding-*.jsonl")) + assert files, "expected a durable forwarding-diagnostics file to be written" + records = [json.loads(line) for f in files for line in f.read_text().splitlines()] + rec = next(r for r in records if r["kind"] == "dispatcher_construction_failed") + assert rec["destination"] == "only" + assert rec["http_status"] is None + assert "RuntimeError" in rec["detail"], f"exception type missing: {rec['detail']!r}" + assert "credential chain unavailable" in rec["detail"], ( + f"exception message missing: {rec['detail']!r}" + ) + await cleanup() await cleanup()