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 070fe9e..5c61b40 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__) @@ -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) --- 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 4f6a574..9bc5d8f 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 @@ -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, diff --git a/modules/hook-context-intelligence/tests/test_classification.py b/modules/hook-context-intelligence/tests/test_classification.py index aaf7769..8893e16 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 diff --git a/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py b/modules/hook-context-intelligence/tests/test_forwarding_diagnostics.py index 7c71c8a..cc97d79 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,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() diff --git a/modules/hook-context-intelligence/tests/test_mount_dispatcher.py b/modules/hook-context-intelligence/tests/test_mount_dispatcher.py index 99c8e02..88cb7fc 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] # --------------------------------------------------------------------------- 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 1ee63c8..2055fa2 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,9 +2,10 @@ from __future__ import annotations +import json import logging from typing import Any -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -313,3 +314,157 @@ 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() + + 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() 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 3e2579f..1b35d2a 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