From 30d2fa9e8786407f6c66f388fadccae421eeedd0 Mon Sep 17 00:00:00 2001 From: colombod Date: Wed, 26 Aug 2026 15:24:44 +0000 Subject: [PATCH] feat(hook-context-intelligence): ENOSPC disk circuit breaker + user-visible fail-loud alerts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stacked on #101 (atomic metadata writes + self-heal). Adds the resilience/UX layer on top: - ENOSPC/EDQUOT circuit breaker: skip disk writes for a capped exponential-backoff cooldown (5s..300s), then probe for recovery, instead of hammering a full disk on every event. Composes with #101 because its atomic writer re-raises OSError so this layer can classify ENOSPC. - Fail loud to the user via HookResult.user_message (bypasses the unwritable log file), severity matched to reality: PERMANENT DATA LOSS (error) when the event reached no sink (disk full and no destination / queue also full), a milder warning when still delivered to the server, and a one-shot info on recovery. - enqueue() now returns whether the event was queued so the handler distinguishes delivered from lost; network dispatch stays independent of disk state. Related: microsoft-amplifier/amplifier-support#492. 🤖 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../handlers/logging_handler.py | 229 ++++++++++++++--- .../test_logging_handler_disk_breaker.py | 239 ++++++++++++++++++ 2 files changed, 438 insertions(+), 30 deletions(-) create mode 100644 modules/hook-context-intelligence/tests/test_logging_handler_disk_breaker.py 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..de7ea50 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 @@ -7,6 +7,7 @@ from __future__ import annotations import asyncio +import errno import json import logging import random @@ -18,6 +19,7 @@ from typing import Any import httpx +from amplifier_core.models import HookResult from amplifier_module_hook_context_intelligence.upload import ( _canonical_json, @@ -25,8 +27,6 @@ build_payload, ) -from amplifier_core.models import HookResult - logger = logging.getLogger(__name__) _OPTIONAL_METADATA_FIELDS = ("agent_name", "parallel_group_id", "recipe_name", "recipe_step") @@ -51,6 +51,19 @@ _CLOSE_HARD_TIMEOUT = 5.0 _METADATA_FORMAT = "context-intelligence" _METADATA_VERSION = "1.0.0" + +#: Disk-pressure circuit breaker. When a session disk write fails with one of +#: these errnos, the whole filesystem is out of room (or over quota) -- retrying +#: on the very next event is futile and only burns syscalls and (worse) tries to +#: append warning lines to a log file that also cannot be written. So we OPEN a +#: breaker: skip disk writes for a growing cooldown, then let a single event +#: PROBE whether space has returned. This bounds the retry rate without ever +#: giving up permanently. Atomic-write recovery (temp + os.replace) lives in the +#: metadata helpers; this breaker is the layer above that keeps a full disk from +#: turning into an every-event failure loop. +_DISK_PRESSURE_ERRNOS = frozenset({errno.ENOSPC, errno.EDQUOT}) +_DISK_BACKOFF_INITIAL_SECONDS = 5.0 +_DISK_BACKOFF_MAX_SECONDS = 300.0 _CONNECT_TIMEOUT = 3.0 _READ_TIMEOUT = 3.0 _POOL_TIMEOUT = 0.5 @@ -394,9 +407,15 @@ def _ensure_worker(self) -> None: partial(_retrieve_task_exception, context=f"{self._name} dispatch worker") ) - def enqueue(self, event: str, data: dict[str, Any]) -> None: + def enqueue(self, event: str, data: dict[str, Any]) -> bool: """Enqueue an event for dispatch. HOT PATH — zero awaits, zero I/O. + Returns ``True`` if the event was queued for delivery, ``False`` if it + was dropped because the queue is full. The caller uses this to tell + "delivered to the server pipeline" apart from "dropped" — which, when + the disk is ALSO full, is the difference between a stale local log and + outright permanent data loss. + Drops on full queue (bumps _overflow_dropped counter). Never disables. Immutability contract: ``data`` MUST be treated as immutable from the @@ -410,6 +429,7 @@ def enqueue(self, event: str, data: dict[str, Any]) -> None: self._ensure_worker() try: self._queue.put_nowait((event, data)) + return True except asyncio.QueueFull: self._overflow_dropped += 1 now = time.monotonic() @@ -417,13 +437,15 @@ def enqueue(self, event: str, data: dict[str, Any]) -> None: self._last_overflow_log = now logger.warning( "%s buffer full — %d events dropped since last warning;" - " events are durable in events.jsonl." + " events are durable in events.jsonl UNLESS the disk is also full" + " (see any DISK FULL alert)." " To manually upload run: context-intelligence-upload --path %s" " (--server-url/--api-key come from flags or env/config; see --help)", self._name, self._overflow_dropped, self._storage_path, ) + return False def _record_forwarding_issue(self, kind: str, detail: str) -> None: """Write a durable forwarding-diagnostics record for this destination. @@ -1267,6 +1289,14 @@ def __init__(self, resolver: Any) -> None: self._parent_id: str = getattr(resolver, "parent_id", "") or "" self._resolve_instance_id: str = getattr(resolver, "resolve_instance_id", "") or "" self._dispatchers: list[_DestinationDispatcher] = [] + # Disk-pressure circuit breaker state (see _DISK_* constants). + # _disk_backoff_seconds == 0.0 means healthy; > 0.0 means the breaker is + # open and this is the current cooldown length. _disk_retry_at is the + # monotonic time the next probe is allowed. _last_disk_msg_at rate-limits + # the user-visible alert. + self._disk_backoff_seconds: float = 0.0 + self._disk_retry_at: float = 0.0 + self._last_disk_msg_at: float = 0.0 async def set_dispatchers(self, dispatchers: list[_DestinationDispatcher]) -> None: """Install the active per-destination dispatchers (called from on_session_ready). @@ -1285,36 +1315,27 @@ def _session_dir(self, session_id: str) -> Path: async def __call__(self, event: str, data: dict[str, Any]) -> HookResult: sanitized_data = _sanitize_for_json(data) - try: - session_id = sanitized_data.get("session_id") - if not session_id: - return HookResult(action="continue") - - session_dir = self._session_dir(session_id) - session_dir.mkdir(parents=True, exist_ok=True) - - # Lazy metadata init: create metadata.json on the very first - # event we see for a given session_id, regardless of event type. - if session_id not in self._seen_sessions: - self._seen_sessions.add(session_id) - self._ensure_metadata(session_dir, session_id, sanitized_data) - - if event in ("session:start", "session:fork"): - self._enrich_metadata_from_session_init(session_dir, session_id, sanitized_data) - elif event in ("session:end", "execution:end"): - self._finalize_metadata(session_dir, sanitized_data) - - self._append_event(session_dir, event, sanitized_data, self._workspace) - self._touch_last_event_at(session_dir, sanitized_data.get("timestamp", "")) - except Exception: - logger.warning("LoggingHandler disk write error processing %s", event, exc_info=True) + session_id = sanitized_data.get("session_id") + if not session_id: + return HookResult(action="continue") + + # Disk write is guarded by the ENOSPC breaker and NEVER raises; the + # network fan-out below runs regardless of disk state so that events + # keep flowing to the server even when the disk is full. The two + # destinations are independent on purpose. + disk_state = self._persist_to_disk(event, session_id, sanitized_data) # Fan-out to all active dispatchers — each enqueue is isolated so that - # one dispatcher's failure does not starve the others (mirrors the - # defensive disk-write block above). + # one dispatcher's failure does not starve the others. Independent of + # disk state: enqueue is a zero-I/O in-memory hot path, so a full disk + # never blocks it. We track whether the event reached AT LEAST ONE + # server pipeline, because "disk full but delivered" is a stale local + # log, while "disk full AND not delivered" is permanent data loss. + delivered_to_server = False for dispatcher in self._dispatchers: try: - dispatcher.enqueue(event, sanitized_data) + if dispatcher.enqueue(event, sanitized_data): + delivered_to_server = True except Exception: logger.warning( "LoggingHandler dispatcher enqueue failed for %s", @@ -1322,8 +1343,156 @@ async def __call__(self, event: str, data: dict[str, Any]) -> HookResult: exc_info=True, ) + # A user-visible alert MUST ride HookResult.user_message (not just the + # logger): when the disk is full the log file cannot be written either, + # so a plain logger.warning would be invisible to the user. + alert = self._build_disk_alert(disk_state, delivered_to_server=delivered_to_server) + if alert is not None: + text, level = alert + return HookResult( + action="continue", + user_message=text, + user_message_level=level, # type: ignore[arg-type] + user_message_source="context-intelligence", + ) return HookResult(action="continue") + # -- disk-write path with ENOSPC circuit breaker ------------------------ + def _persist_to_disk(self, event: str, session_id: str, data: dict[str, Any]) -> str: + """Write this event's session files, guarded by the disk-pressure breaker. + + Returns a disk-state token the caller combines with the network-dispatch + outcome to phrase the right user alert: + + * ``"ok"`` — written to disk (or nothing to report). + * ``"degraded"`` — disk full: this event was NOT written to disk. + * ``"recovered"`` — a probe just succeeded after being degraded. + + Never raises: event capture and dispatch must proceed even when the disk + cannot be written. Relies on the metadata helpers re-raising OSError + (their atomic writer cleans up its temp file and re-raises) so ENOSPC can + be classified here. + """ + now = time.monotonic() + + # Breaker OPEN: within the cooldown window, skip disk writes entirely + # rather than hammer a full filesystem on every event. + if self._disk_backoff_seconds > 0.0 and now < self._disk_retry_at: + return "degraded" + + try: + self._write_session_to_disk(event, session_id, data) + except OSError as exc: + if exc.errno in _DISK_PRESSURE_ERRNOS: + self._open_disk_breaker(now) + # Best-effort log (may not reach disk — exactly why the caller + # also surfaces a user_message that bypasses the log file). + logger.error( + "LoggingHandler disk write failed (errno %s): session data not written", + exc.errno, + ) + return "degraded" + # Non-pressure OSError (e.g. a single bad path): best-effort log, + # do NOT open the global breaker. + logger.warning("LoggingHandler disk write error processing %s", event, exc_info=True) + return "ok" + except Exception: + logger.warning("LoggingHandler disk write error processing %s", event, exc_info=True) + return "ok" + + # Success. If the breaker had been open, this was a recovery probe: + # close it and report recovery. + if self._disk_backoff_seconds > 0.0: + self._disk_backoff_seconds = 0.0 + self._disk_retry_at = 0.0 + logger.warning("LoggingHandler disk pressure cleared; session logging resumed") + return "recovered" + return "ok" + + def _write_session_to_disk(self, event: str, session_id: str, data: dict[str, Any]) -> None: + """The actual per-event disk writes. Lets OSError propagate for classification. + + The metadata helpers (``_ensure_metadata`` / ``_enrich`` / ``_finalize`` / + ``_touch_last_event_at``) already read tolerantly and write atomically, + so a corrupt file self-heals; this method adds the ENOSPC-classification + seam by letting their OSError propagate to ``_persist_to_disk``. + """ + session_dir = self._session_dir(session_id) + session_dir.mkdir(parents=True, exist_ok=True) + + # Lazy metadata init: create metadata.json on the very first event we + # see for a given session_id. Only mark the session seen AFTER the write + # succeeds, so a write that fails under disk pressure is retried on the + # next event instead of being permanently skipped. + if session_id not in self._seen_sessions: + self._ensure_metadata(session_dir, session_id, data) + self._seen_sessions.add(session_id) + + if event in ("session:start", "session:fork"): + self._enrich_metadata_from_session_init(session_dir, session_id, data) + elif event in ("session:end", "execution:end"): + self._finalize_metadata(session_dir, data) + + self._append_event(session_dir, event, data, self._workspace) + self._touch_last_event_at(session_dir, data.get("timestamp", "")) + + def _open_disk_breaker(self, now: float) -> None: + """Open or widen the disk-pressure breaker with capped exponential backoff.""" + if self._disk_backoff_seconds <= 0.0: + self._disk_backoff_seconds = _DISK_BACKOFF_INITIAL_SECONDS + else: + self._disk_backoff_seconds = min( + self._disk_backoff_seconds * 2, _DISK_BACKOFF_MAX_SECONDS + ) + self._disk_retry_at = now + self._disk_backoff_seconds + + def _build_disk_alert( + self, disk_state: str, *, delivered_to_server: bool + ) -> tuple[str, str] | None: + """Phrase the user-visible alert from the disk state AND the network outcome. + + This is where "we are losing data PERMANENTLY" is distinguished from the + far milder "local log is stale but the server still has the events": + + * disk full + event was NOT delivered to any server -> PERMANENT LOSS (error) + * disk full + event WAS delivered to the server -> degraded, recoverable (warning) + * recovered -> resumed (info) + + Error/warning alerts are rate-limited (they recur every event); the + one-shot recovery notice is not. Returns ``None`` when nothing needs + saying or the alert is rate-limited. All alerts ride + ``HookResult.user_message`` because the log file cannot be written while + the disk is full. + """ + if disk_state == "recovered": + return ( + "context-intelligence: disk space recovered — session logging has resumed.", + "info", + ) + if disk_state != "degraded": + return None + + now = time.monotonic() + if now - self._last_disk_msg_at < _LOG_RATE_LIMIT_SECONDS: + return None + self._last_disk_msg_at = now + + if delivered_to_server: + return ( + "context-intelligence: DISK FULL — local session logs (events.jsonl / " + "metadata.json) are NOT being written. Events are still being sent to the " + "server, so they are not lost yet, but free disk space to restore local " + "logging. (This alert bypasses the log file, which also cannot be written.)", + "warning", + ) + return ( + "context-intelligence: PERMANENT DATA LOSS — the disk is full AND events are " + "not reaching the server. Session events are being lost for good: NOT written " + "to events.jsonl and NOT delivered. Free disk space immediately to stop losing " + "data. (This alert bypasses the log file, which also cannot be written.)", + "error", + ) + # -- metadata lifecycle ------------------------------------------------- def _ensure_metadata( self, diff --git a/modules/hook-context-intelligence/tests/test_logging_handler_disk_breaker.py b/modules/hook-context-intelligence/tests/test_logging_handler_disk_breaker.py new file mode 100644 index 0000000..af361ca --- /dev/null +++ b/modules/hook-context-intelligence/tests/test_logging_handler_disk_breaker.py @@ -0,0 +1,239 @@ +"""Tests for the disk-pressure circuit breaker and user-visible fail-loud. + +When the disk fills, ``_touch_last_event_at`` / metadata writes / the JSONL +append all fail with ENOSPC. Two requirements drive this behaviour: + +1. Do NOT hammer a full filesystem on every event (and do not try to log the + failure to a log file that also cannot be written). Open a breaker: skip disk + writes for a growing cooldown, then let one event PROBE for recovery. +2. FAIL LOUD to the *user*, not the log file. The disk-full alert must ride + ``HookResult.user_message`` (which the orchestrator surfaces in the UI), + because ``logger.*`` output at this moment cannot reach disk. + +Verifies: +- ENOSPC opens the breaker and returns an error-level user_message. +- While the breaker is open, disk writes are skipped (no per-event hammering). +- The alert is rate-limited (not emitted on every event). +- A successful probe after cooldown closes the breaker and emits a recovery + message; normal writes resume. +- Backoff grows (capped) across repeated failures. +- A non-ENOSPC OSError does NOT open the breaker and surfaces no user_message. +- Event dispatch (network fan-out) still happens while the disk is degraded. +""" + +from __future__ import annotations + +import errno +from pathlib import Path + +import amplifier_module_hook_context_intelligence.handlers.logging_handler as mod +from amplifier_module_hook_context_intelligence.handlers.logging_handler import LoggingHandler + + +class _FakeResolver: + def __init__(self, base_path: Path, project_slug: str, workspace: str = "ws") -> None: + self.base_path = base_path + self.project_slug = project_slug + self.workspace = workspace + self.working_dir: str = "" + + def session_dir(self, session_id: str) -> Path: + return self.base_path / self.project_slug / "sessions" / session_id / "context-intelligence" + + +def _enospc(*_a, **_k): + raise OSError(errno.ENOSPC, "No space left on device") + + +class _Clock: + """Monotonic clock stub the test advances explicitly.""" + + def __init__(self) -> None: + self.t = 1000.0 + + def __call__(self) -> float: + return self.t + + +def _evt(sid: str = "s1", ts: str = "2026-01-15T10:00:00Z", **extra): + return {"session_id": sid, "timestamp": ts, **extra} + + +class TestDiskBreaker: + async def test_enospc_opens_breaker_and_alerts_user(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + result = await handler("tool:call", _evt()) + + assert result.action == "continue" + # Fail loud, to the USER, at error level — not just a log line. With no + # destination configured, disk-full means the data is gone for good. + assert result.user_message is not None + assert "PERMANENT DATA LOSS" in result.user_message + assert result.user_message_level == "error" + assert result.user_message_source == "context-intelligence" + # Breaker is now open. + assert handler._disk_backoff_seconds == mod._DISK_BACKOFF_INITIAL_SECONDS + + async def test_no_destination_disk_full_is_permanent_loss(self, tmp_path, monkeypatch) -> None: + # The user's scenario: NO dispatchers configured. A full disk then means + # events reach no sink at all — the alert must say so, at error level. + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + assert handler._dispatchers == [] # no destination + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + result = await handler("tool:call", _evt()) + + assert result.user_message_level == "error" + assert "PERMANENT DATA LOSS" in result.user_message + assert "disk is full" in result.user_message.lower() + + async def test_disk_full_but_delivered_is_warning_not_loss(self, tmp_path, monkeypatch) -> None: + # Disk full BUT a dispatcher accepted the event -> not lost, just a stale + # local log. Must be a warning, and must NOT claim permanent loss. + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + class _OkDispatcher: + def enqueue(self, event, data): + return True # accepted for delivery + + handler._dispatchers = [_OkDispatcher()] # type: ignore[list-item] + + result = await handler("tool:call", _evt()) + + assert result.user_message_level == "warning" + assert "PERMANENT DATA LOSS" not in result.user_message + assert "still being sent to the server" in result.user_message + + async def test_disk_full_and_queue_dropped_is_permanent_loss( + self, tmp_path, monkeypatch + ) -> None: + # Disk full AND the delivery queue is full (enqueue returns False) -> + # the event reached no sink -> permanent loss, at error level. + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + class _FullDispatcher: + def enqueue(self, event, data): + return False # queue full, dropped + + handler._dispatchers = [_FullDispatcher()] # type: ignore[list-item] + + result = await handler("tool:call", _evt()) + + assert result.user_message_level == "error" + assert "PERMANENT DATA LOSS" in result.user_message + + async def test_open_breaker_skips_disk_writes(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + + calls = {"n": 0} + + def _counting_enospc(*_a, **_k): + calls["n"] += 1 + raise OSError(errno.ENOSPC, "No space left on device") + + monkeypatch.setattr(handler, "_write_session_to_disk", _counting_enospc) + + await handler("tool:call", _evt()) # trips breaker (1 attempt) + assert calls["n"] == 1 + + # Next events, still inside the cooldown window: must NOT attempt a write. + clock.t += 1.0 + await handler("tool:call", _evt(ts="t2")) + clock.t += 1.0 + await handler("tool:call", _evt(ts="t3")) + assert calls["n"] == 1, "disk write was retried during the cooldown window" + + async def test_alert_is_rate_limited(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + first = await handler("tool:call", _evt()) + assert first.user_message is not None + + # Advance past the cooldown but NOT past the log rate-limit window, then + # fail again — the breaker re-opens but the user alert is suppressed. + clock.t += mod._DISK_BACKOFF_INITIAL_SECONDS + 0.1 + second = await handler("tool:call", _evt(ts="t2")) + assert second.user_message is None + + async def test_probe_recovers_and_notifies(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + await handler("tool:call", _evt()) # breaker open + assert handler._disk_backoff_seconds > 0.0 + + # Disk frees up; advance past cooldown; the probe write now succeeds. + monkeypatch.setattr(handler, "_write_session_to_disk", lambda *a, **k: None) + clock.t += mod._DISK_BACKOFF_INITIAL_SECONDS + 0.1 + result = await handler("tool:call", _evt(ts="t2")) + + assert handler._disk_backoff_seconds == 0.0 # breaker closed + assert result.user_message is not None + assert result.user_message_level == "info" + assert "recovered" in result.user_message.lower() + + async def test_backoff_grows_and_caps(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + seen = [] + for _ in range(20): + await handler("tool:call", _evt(ts="t")) + seen.append(handler._disk_backoff_seconds) + clock.t = handler._disk_retry_at + 0.001 # jump to just past each cooldown + assert seen[0] == mod._DISK_BACKOFF_INITIAL_SECONDS + assert seen[1] > seen[0] # doubled + assert max(seen) == mod._DISK_BACKOFF_MAX_SECONDS # capped, never unbounded + + async def test_non_enospc_oserror_does_not_open_breaker(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + + def _eacces(*_a, **_k): + raise OSError(errno.EACCES, "permission denied") + + monkeypatch.setattr(handler, "_write_session_to_disk", _eacces) + result = await handler("tool:call", _evt()) + + assert result.user_message is None # not a disk-full condition + assert handler._disk_backoff_seconds == 0.0 # breaker stays closed + + async def test_dispatch_still_runs_while_disk_degraded(self, tmp_path, monkeypatch) -> None: + clock = _Clock() + monkeypatch.setattr(mod.time, "monotonic", clock) + handler = LoggingHandler(_FakeResolver(tmp_path, "proj")) + monkeypatch.setattr(handler, "_write_session_to_disk", _enospc) + + enqueued = [] + + class _Dispatcher: + def enqueue(self, event, data): + enqueued.append(event) + return True + + handler._dispatchers = [_Dispatcher()] # type: ignore[list-item] + + await handler("tool:call", _evt()) + assert enqueued == ["tool:call"], "network dispatch must not depend on disk health"