diff --git a/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/cli.py b/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/cli.py index 4dd31a4e..6e1c9068 100644 --- a/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/cli.py +++ b/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/cli.py @@ -351,6 +351,34 @@ def _build_parser() -> argparse.ArgumentParser: ), ) + parser.add_argument( + "--max-retries", + type=int, + default=5, + metavar="N", + dest="max_retries", + help=( + "Number of ADDITIONAL retries after the first attempt for transient " + "failures (connection errors, timeouts, HTTP 5xx, HTTP 429) using " + "exponential backoff (default: 5, i.e. up to 6 attempts per event). " + "Permanent failures (4xx other than 429, and DNS/TLS errors) fail " + "immediately. Set 0 to disable retries." + ), + ) + + parser.add_argument( + "--timeout", + type=float, + default=None, + metavar="SECONDS", + dest="timeout_s", + help=( + "Read/write HTTP timeout in seconds (default: 30). Increase for a slow " + "or variable link with large event payloads; the connect timeout is " + "unaffected." + ), + ) + # Auth flags parser.add_argument( "--auth-mode", @@ -499,6 +527,8 @@ def main() -> None: event_delay_s=args.event_delay_ms / 1000.0, auth_strategy=auth_strategy, replay=not args.no_replay, + max_retries=args.max_retries, + timeout_s=args.timeout_s, ) # 7. Write result JSON to stdout diff --git a/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/uploader.py b/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/uploader.py index 9d9aaf31..f3654df5 100644 --- a/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/uploader.py +++ b/modules/tool-context-intelligence-upload/amplifier_module_tool_context_intelligence_upload/uploader.py @@ -10,6 +10,9 @@ from __future__ import annotations import json +import random +import socket +import ssl import sys import time from pathlib import Path @@ -97,6 +100,89 @@ def _count_lines(file_path: Path) -> int: return count +# --------------------------------------------------------------------------- +# Retry / backoff (issue #338 — bounded exponential backoff on transient errors) +# --------------------------------------------------------------------------- +#: Backoff schedule mirrors the live-forwarding hook's proven values +#: (hook-context-intelligence logging_handler: initial 1.0s, cap 30.0s, jitter). +#: Kept as module constants rather than run_upload parameters — there is no +#: second caller that needs to tune them, so widening the signature would be +#: speculative generality. +_BACKOFF_INITIAL_S = 1.0 +_BACKOFF_MAX_S = 30.0 +_BACKOFF_JITTER = True +#: Default number of *additional* retries after the first attempt. Total +#: attempts per event = max_retries + 1 (default 5 -> 6 attempts). +_DEFAULT_MAX_RETRIES = 5 + + +def _is_fatal_transport_error(exc: BaseException) -> bool: + """True if a transport-level error is PERMANENT (never succeeds on retry). + + ``httpx.ConnectError`` covers both genuinely transient faults (connection + reset, refused) and permanent ones — DNS resolution failure (``socket.gaierror``, + e.g. a mistyped ``--server-url``) and TLS/certificate failure (``ssl.SSLError``, + e.g. an untrusted cert). Retrying a permanent transport fault just burns the + whole backoff budget on a guaranteed-dead destination, so we classify those as + fail-fast. The underlying OS error is wrapped by httpx, so walk the + cause/context chain to find it. + """ + seen: set[int] = set() + cur: BaseException | None = exc + while cur is not None and id(cur) not in seen: + seen.add(id(cur)) + if isinstance(cur, (socket.gaierror, ssl.SSLError)): + return True + cur = cur.__cause__ or cur.__context__ + return False + + +def _is_transient_status(status_code: int) -> bool: + """Return True if *status_code* is a transient failure worth retrying. + + Transient = HTTP 429 (throttled) or any 5xx (server-side). Everything else + non-2xx is permanent. + + NOTE — deliberately ADAPTED from the live-forwarding hook's classifier + (``_classify_http_outcome`` treats 401 as transient): for the batch + uploader, per-request token refresh (see :func:`run_upload`) already handles + token expiry, so a 401 that survives a fresh token is a *real* auth fault + that retrying cannot fix. 401 is therefore PERMANENT here — fail fast, loud. + """ + return status_code == 429 or status_code >= 500 + + +def _backoff_delay(retry_index: int) -> float: + """Exponential backoff for *retry_index* (0 = first retry), capped + jittered.""" + delay = min(_BACKOFF_INITIAL_S * (2**retry_index), _BACKOFF_MAX_S) + if _BACKOFF_JITTER: + # Full-ish jitter in [50%, 100%] of the computed delay — spreads retries + # without ever collapsing the backoff to ~0. + delay = delay * (0.5 + random.random() * 0.5) + return delay + + +def _retry_after_or_backoff(response: httpx.Response, retry_index: int) -> float: + """Honor a numeric ``Retry-After`` header (clamped to the cap); else backoff. + + APIM commonly answers 429/503 with ``Retry-After``. Ignoring it either + under-waits (keeps getting throttled) or over-waits. Only the delta-seconds + form is honored; the HTTP-date form falls back to exponential backoff. + """ + try: + raw = response.headers.get("Retry-After") + except (AttributeError, TypeError): # defensive: non-httpx mock response + raw = None + if raw: + try: + secs = float(raw) + except (ValueError, TypeError): + secs = -1.0 + if secs >= 0: + return min(secs, _BACKOFF_MAX_S) + return _backoff_delay(retry_index) + + def run_upload( sessions: list[tuple[Path, dict[str, Any]]], server_url: str, @@ -106,6 +192,8 @@ def run_upload( *, auth_strategy: AuthStrategy | None = None, replay: bool = True, + max_retries: int = _DEFAULT_MAX_RETRIES, + timeout_s: float | None = None, ) -> UploadResult: """Replay all events from *sessions* to the server. @@ -148,14 +236,23 @@ def run_upload( auth_strategy = ApiKeyAuth(api_key) endpoint = f"{server_url}/events" - timeout = httpx.Timeout(connect=5.0, read=30.0, write=30.0, pool=5.0) - headers = auth_strategy.headers() + # timeout_s (when set via --timeout) tunes the read/write timeout — the dial an + # operator reaches for on a slow/variable link with large event payloads. connect + # stays short (a slow *connect* is a real problem, not a payload-size one). + rw_timeout = timeout_s if timeout_s is not None else 30.0 + timeout = httpx.Timeout(connect=5.0, read=rw_timeout, write=rw_timeout, pool=5.0) query_params: dict[str, str] | None = {"replay": "true"} if replay else None + # max_retries is the number of ADDITIONAL attempts after the first; guard + # against negatives so there is always at least one POST (never a silent skip). + max_retries = max(0, max_retries) total_events_uploaded = 0 total_sessions_uploaded = 0 - with httpx.Client(headers=headers, timeout=timeout) as client: + # NOTE (issue #338): auth headers are fetched PER attempt inside the loop + # (not baked into the client here), so a long run that crosses the Entra + # token-expiry boundary transparently picks up a refreshed bearer token. + with httpx.Client(timeout=timeout) as client: for session_dir, metadata in sessions: session_id: str = metadata["session_id"] events_file = session_dir / "events.jsonl" @@ -202,36 +299,111 @@ def run_upload( payload = build_payload(event, workspace, data) - try: - response = client.post(endpoint, json=payload, params=query_params) - except httpx.HTTPError as exc: - tracker.mark_failed( - session_id=session_id, - event_index=event_index, - http_status=0, - error=str(exc), - ) - return UploadResult( - success=False, - sessions_uploaded=total_sessions_uploaded, - events_uploaded=total_events_uploaded, - error=str(exc), - failed_at={ - "session_id": session_id, - "event_index": event_index, - "http_status": 0, - }, - ) - - if response.status_code < 200 or response.status_code >= 300: + # --- POST with bounded retry + exponential backoff (issue #338) --- + # Transient failures (connection errors, timeouts, 5xx, 429) are + # retried up to *max_retries* times; permanent failures (4xx other + # than 429, 3xx) and an exhausted retry budget fail loud exactly as + # before. tracker.event_sent()/mark_failed() fire ONCE per event on + # the terminal outcome — never per attempt (so progress.json never + # flips to 'failed' mid-retry, and sent-counts never over-count). + retry_index = 0 + while True: + # Fetch the auth header for THIS attempt so a retry that crosses + # the token-expiry margin transparently gets a refreshed token. + # headers() can raise (unusable API key -> ValueError; credential + # failure -> azure error) — neither is an httpx.HTTPError, so guard + # it explicitly and fail loud rather than crash the run mid-batch. + try: + request_headers = auth_strategy.headers() + except Exception as exc: # noqa: BLE001 - auth failure must fail loud, not crash + error_msg = f"auth header error: {exc}" + tracker.mark_failed( + session_id=session_id, + event_index=event_index, + http_status=0, + error=error_msg, + ) + return UploadResult( + success=False, + sessions_uploaded=total_sessions_uploaded, + events_uploaded=total_events_uploaded, + error=error_msg, + failed_at={ + "session_id": session_id, + "event_index": event_index, + "http_status": 0, + }, + ) + + try: + response = client.post( + endpoint, + json=payload, + params=query_params, + headers=request_headers, + ) + except httpx.HTTPError as exc: + # Transport-level error. Genuinely transient ones (connection + # reset, timeout) are retried; PERMANENT ones (DNS resolution, + # TLS/cert failure) never succeed on retry, so fail them fast + # instead of burning the whole backoff budget on a dead host. + if not _is_fatal_transport_error(exc) and retry_index < max_retries: + delay = _backoff_delay(retry_index) + print( + f"WARNING: transient network error uploading session " + f"{session_id!r} event {event_index} " + f"(attempt {retry_index + 1}/{max_retries + 1}): {exc}; " + f"retrying in {delay:.1f}s", + file=sys.stderr, + ) + time.sleep(delay) + retry_index += 1 + continue + tracker.mark_failed( + session_id=session_id, + event_index=event_index, + http_status=0, + error=str(exc), + ) + return UploadResult( + success=False, + sessions_uploaded=total_sessions_uploaded, + events_uploaded=total_events_uploaded, + error=str(exc), + failed_at={ + "session_id": session_id, + "event_index": event_index, + "http_status": 0, + }, + ) + + status_code = response.status_code + if 200 <= status_code < 300: + break # delivered + + # Non-2xx: retry only transient statuses, and only while budget remains. + if _is_transient_status(status_code) and retry_index < max_retries: + delay = _retry_after_or_backoff(response, retry_index) + print( + f"WARNING: HTTP {status_code} uploading session " + f"{session_id!r} event {event_index} " + f"(attempt {retry_index + 1}/{max_retries + 1}); " + f"retrying in {delay:.1f}s", + file=sys.stderr, + ) + time.sleep(delay) + retry_index += 1 + continue + + # Permanent failure, or transient budget exhausted — fail loud. body = response.text[:200].strip() if response.text else "" - error_msg = f"HTTP {response.status_code} from {endpoint}" + ( + error_msg = f"HTTP {status_code} from {endpoint}" + ( f": {body}" if body else "" ) tracker.mark_failed( session_id=session_id, event_index=event_index, - http_status=response.status_code, + http_status=status_code, error=error_msg, ) return UploadResult( @@ -242,7 +414,7 @@ def run_upload( failed_at={ "session_id": session_id, "event_index": event_index, - "http_status": response.status_code, + "http_status": status_code, }, ) diff --git a/modules/tool-context-intelligence-upload/tests/test_auth_wiring.py b/modules/tool-context-intelligence-upload/tests/test_auth_wiring.py index d5a999e0..45f2360d 100644 --- a/modules/tool-context-intelligence-upload/tests/test_auth_wiring.py +++ b/modules/tool-context-intelligence-upload/tests/test_auth_wiring.py @@ -106,8 +106,10 @@ def test_injected_strategy_headers_used_not_api_key(self, tmp_path: Path) -> Non auth_strategy=strategy, ) - _, init_kwargs = mock_cls.call_args - assert init_kwargs["headers"] == {"Authorization": "Bearer my-entra-token"} + # Issue #338: headers are sent PER REQUEST (so token refresh can fire + # mid-run), not baked into the httpx.Client constructor. + post_kwargs = mock_client.post.call_args.kwargs + assert post_kwargs["headers"] == {"Authorization": "Bearer my-entra-token"} def test_no_strategy_derives_api_key_auth(self, tmp_path: Path) -> None: """When auth_strategy is None, ApiKeyAuth(api_key) is derived — backward compat.""" @@ -128,11 +130,17 @@ def test_no_strategy_derives_api_key_auth(self, tmp_path: Path) -> None: tracker=tracker, ) - _, init_kwargs = mock_cls.call_args - assert init_kwargs["headers"] == {"Authorization": "Bearer legacy-key"} + # Issue #338: header is sent per request, not baked into the client. + post_kwargs = mock_client.post.call_args.kwargs + assert post_kwargs["headers"] == {"Authorization": "Bearer legacy-key"} - def test_strategy_headers_called_once_at_client_construction(self, tmp_path: Path) -> None: - """The strategy's headers() is called once, at httpx.Client construction.""" + def test_strategy_headers_fetched_per_request(self, tmp_path: Path) -> None: + """Issue #338: the strategy's headers() is fetched PER request, not baked once. + + The Authorization header must be attached to each ``client.post`` call (so a + long run can pick up a refreshed token mid-flight), NOT passed to the + ``httpx.Client`` constructor where it would freeze for the whole run. + """ from context_intelligence.auth import ApiKeyAuth from amplifier_module_tool_context_intelligence_upload.uploader import run_upload @@ -154,10 +162,13 @@ def test_strategy_headers_called_once_at_client_construction(self, tmp_path: Pat auth_strategy=strategy, ) - # httpx.Client is constructed once; headers kwarg should contain strategy headers - assert mock_cls.call_count == 1 - _, kwargs = mock_cls.call_args - assert kwargs.get("headers") == {"Authorization": "Bearer test-static"} + # The header is NOT baked into the client constructor... + _, init_kwargs = mock_cls.call_args + assert "headers" not in init_kwargs or init_kwargs.get("headers") is None + # ...it is attached to every POST (3 events -> 3 posts, each carrying it). + assert mock_client.post.call_count == 3 + for call in mock_client.post.call_args_list: + assert call.kwargs.get("headers") == {"Authorization": "Bearer test-static"} # --------------------------------------------------------------------------- diff --git a/modules/tool-context-intelligence-upload/tests/test_retry_and_refresh.py b/modules/tool-context-intelligence-upload/tests/test_retry_and_refresh.py new file mode 100644 index 00000000..2fb2d970 --- /dev/null +++ b/modules/tool-context-intelligence-upload/tests/test_retry_and_refresh.py @@ -0,0 +1,594 @@ +"""Regression tests for issue #338 — mid-run Entra token refresh + bounded retry. + +These tests exercise the ACTUAL behaviour the fix introduces (per-request auth +header, transient-vs-permanent classification, bounded exponential backoff, +Retry-After, fail-loud on auth errors, tracker-once semantics) rather than just +asserting mock plumbing. ``time.sleep`` is patched throughout so backoff never +sleeps for real. +""" + +from __future__ import annotations + +import json +import time +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, call, patch + +import httpx +import pytest + +from amplifier_module_tool_context_intelligence_upload.uploader import ( + _is_transient_status, + run_upload, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _write_session( + tmp_path: Path, session_id: str, events: list[dict[str, Any]] +) -> tuple[Path, dict[str, Any]]: + session_dir = tmp_path / f"session-{session_id}" + session_dir.mkdir(parents=True, exist_ok=True) + metadata = {"session_id": session_id, "format": "context-intelligence"} + (session_dir / "metadata.json").write_text(json.dumps(metadata), encoding="utf-8") + (session_dir / "events.jsonl").write_text( + "\n".join(json.dumps(e) for e in events), encoding="utf-8" + ) + return session_dir, metadata + + +def _make_events(n: int) -> list[dict[str, Any]]: + return [{"event": f"e-{i}", "workspace": "ws", "data": {"index": i}} for i in range(n)] + + +class _Resp: + """Minimal stand-in for httpx.Response (status + text + headers).""" + + def __init__( + self, status_code: int, text: str = "", headers: dict[str, str] | None = None + ) -> None: + self.status_code = status_code + self.text = text + self.headers = headers or {} + + +class ConstAuth: + """Auth strategy returning a constant bearer header.""" + + def __init__(self, token: str = "tok") -> None: + self._t = token + self.calls = 0 + + def headers(self) -> dict[str, str]: + self.calls += 1 + return {"Authorization": f"Bearer {self._t}"} + + +class RaisingAuth: + """Auth strategy whose headers() always raises (unusable key / cred failure).""" + + def __init__(self, exc: Exception) -> None: + self._exc = exc + + def headers(self) -> dict[str, str]: + raise self._exc + + +class _FakeToken: + def __init__(self, token: str, expires_on: float) -> None: + self.token = token + self.expires_on = expires_on + + +class _ExpiringCredential: + """Credential returning a fresh, near-expiry token on every get_token(). + + ``expires_on`` sits inside EntraTokenAuth's 300s safety margin, so every + ``headers()`` call is forced down the refresh path — exactly the mid-run + token-rotation scenario from the bug report. + """ + + def __init__(self) -> None: + self.n = 0 + + def get_token(self, *scopes: str, **kwargs: Any) -> _FakeToken: + self.n += 1 + return _FakeToken(f"tok-{self.n}", expires_on=time.time() + 100.0) + + +# --------------------------------------------------------------------------- +# Bug 1 — per-request token refresh actually happens mid-run +# --------------------------------------------------------------------------- + + +class TestPerRequestTokenRefresh: + def test_expired_token_is_refreshed_between_posts(self, tmp_path: Path) -> None: + """A near-expiry Entra token is re-fetched per POST; later POST carries a NEW bearer.""" + from context_intelligence.auth import EntraTokenAuth + + cred = _ExpiringCredential() + strategy = EntraTokenAuth(cred, "api://resource") + + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(2)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=strategy, + ) + + # Credential was re-invoked per request (genuine refresh, not a cached header). + assert cred.n == 2 + first_hdr = mock_client.post.call_args_list[0].kwargs["headers"] + second_hdr = mock_client.post.call_args_list[1].kwargs["headers"] + assert first_hdr == {"Authorization": "Bearer tok-1"} + assert second_hdr == {"Authorization": "Bearer tok-2"} + + def test_header_not_baked_into_client(self, tmp_path: Path) -> None: + """The httpx.Client is constructed WITHOUT a frozen Authorization header.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth("abc"), + ) + + _, init_kwargs = mock_cls.call_args + assert "headers" not in init_kwargs or init_kwargs.get("headers") is None + + +# --------------------------------------------------------------------------- +# Bug 2 — bounded retry with exponential backoff +# --------------------------------------------------------------------------- + + +class TestRetryBackoff: + def test_transient_sequence_retried_then_succeeds(self, tmp_path: Path) -> None: + """A ReadTimeout, then 503, then 200 → event succeeds; tracker fires ONCE.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + outcomes = [httpx.ReadTimeout("read timed out"), _Resp(503), _Resp(200)] + + def side_effect(url: str, **kwargs: Any) -> Any: + item = outcomes.pop(0) + if isinstance(item, Exception): + raise item + return item + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.side_effect = side_effect + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + ) + + assert result.success is True + assert result.events_uploaded == 1 + assert mock_client.post.call_count == 3 + # tracker.event_sent fires exactly once (terminal success), never per attempt. + assert tracker.event_sent.call_count == 1 + tracker.mark_failed.assert_not_called() + # Two transient failures → two backoff sleeps. + assert mock_sleep.call_count == 2 + + def test_persistent_transient_fails_after_exactly_max_retries(self, tmp_path: Path) -> None: + """Unrelenting 503 → fail after max_retries+1 attempts; mark_failed once.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls, patch("time.sleep"): + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(503, text="down") + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + max_retries=3, + ) + + assert result.success is False + assert result.failed_at is not None + assert result.failed_at["http_status"] == 503 + # 3 retries + 1 initial attempt = 4 POSTs. + assert mock_client.post.call_count == 4 + assert tracker.mark_failed.call_count == 1 + tracker.event_sent.assert_not_called() + + def test_permanent_4xx_fails_immediately_zero_retries(self, tmp_path: Path) -> None: + """A 403 is permanent → single attempt, no backoff sleep.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(403, text="forbidden") + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + max_retries=5, + ) + + assert result.success is False + assert result.failed_at is not None + assert result.failed_at["http_status"] == 403 + assert mock_client.post.call_count == 1 + mock_sleep.assert_not_called() + + def test_401_is_permanent_not_retried(self, tmp_path: Path) -> None: + """401 is deliberately PERMANENT for the uploader (refresh owns token expiry).""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(401, text="Unauthorized") + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + max_retries=5, + ) + + assert result.success is False + assert result.failed_at is not None + assert result.failed_at["http_status"] == 401 + assert mock_client.post.call_count == 1 + mock_sleep.assert_not_called() + + def test_429_honors_retry_after(self, tmp_path: Path) -> None: + """A 429 with numeric Retry-After sleeps that many seconds (clamped), then succeeds.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + outcomes = [_Resp(429, headers={"Retry-After": "2"}), _Resp(200)] + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.side_effect = lambda url, **kw: outcomes.pop(0) + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + ) + + assert result.success is True + assert mock_client.post.call_count == 2 + # Retry-After: 2 honored verbatim (min(2, 30) == 2.0). + assert call(2.0) in mock_sleep.call_args_list + + def test_max_retries_zero_still_posts_once(self, tmp_path: Path) -> None: + """max_retries=0 → exactly ONE POST attempt (no silent skip of the event).""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls, patch("time.sleep"): + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + max_retries=0, + ) + + assert result.success is True + assert mock_client.post.call_count == 1 + assert result.events_uploaded == 1 + + +# --------------------------------------------------------------------------- +# Auth-error guard — headers() raising must fail LOUD, never crash the run +# --------------------------------------------------------------------------- + + +class TestAuthHeaderErrorGuard: + def test_valueerror_from_headers_fails_loud_not_crash(self, tmp_path: Path) -> None: + """ApiKeyAuth-style ValueError from headers() → UploadResult(success=False), no crash.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls, patch("time.sleep"): + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + # Must return a result, not raise. + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=RaisingAuth(ValueError("api key is unusable")), + ) + + assert result.success is False + assert result.error is not None and "auth header error" in result.error + assert result.failed_at is not None + assert result.failed_at["http_status"] == 0 + # No POST was ever attempted; failure recorded once; run did not crash. + mock_client.post.assert_not_called() + assert tracker.mark_failed.call_count == 1 + tracker.event_sent.assert_not_called() + + def test_arbitrary_exception_from_headers_is_caught(self, tmp_path: Path) -> None: + """A non-ValueError credential failure (e.g. azure error) is also caught, not raised.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + class _AzureError(Exception): + pass + + with patch("httpx.Client") as mock_cls, patch("time.sleep"): + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=RaisingAuth(_AzureError("credential unavailable")), + ) + + assert result.success is False + assert "auth header error" in (result.error or "") + + +# --------------------------------------------------------------------------- +# Classifier unit — the adapted (uploader-specific) transient rule +# --------------------------------------------------------------------------- + + +class TestClassifier: + @pytest.mark.parametrize("status", [500, 502, 503, 504, 429]) + def test_transient_statuses(self, status: int) -> None: + assert _is_transient_status(status) is True + + @pytest.mark.parametrize("status", [400, 401, 403, 404, 409, 413, 422, 301, 302]) + def test_permanent_statuses(self, status: int) -> None: + assert _is_transient_status(status) is False + + +# --------------------------------------------------------------------------- +# Follow-up #1 — configurable read/write timeout (--timeout) +# --------------------------------------------------------------------------- + + +class TestConfigurableTimeout: + def test_default_timeout_is_30s(self, tmp_path: Path) -> None: + """With no timeout_s, the httpx.Client read/write timeout stays at 30s.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + ) + + timeout = mock_cls.call_args.kwargs["timeout"] + assert timeout.read == 30.0 + assert timeout.write == 30.0 + # connect stays short regardless. + assert timeout.connect == 5.0 + + def test_timeout_s_overrides_read_and_write(self, tmp_path: Path) -> None: + """timeout_s=120 sets read+write to 120s; connect is left short.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + with patch("httpx.Client") as mock_cls: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.return_value = _Resp(200) + + run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + timeout_s=120.0, + ) + + timeout = mock_cls.call_args.kwargs["timeout"] + assert timeout.read == 120.0 + assert timeout.write == 120.0 + assert timeout.connect == 5.0 + + +# --------------------------------------------------------------------------- +# Follow-up #2 — cause-classify ConnectError (DNS/TLS fatal = fail fast) +# --------------------------------------------------------------------------- + + +class TestFatalTransportClassification: + def test_dns_failure_is_not_retried(self, tmp_path: Path) -> None: + """A ConnectError caused by DNS resolution (gaierror) fails fast — one attempt.""" + import socket + + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + def raise_dns(url: str, **kwargs: Any) -> Any: + exc = httpx.ConnectError("name resolution failed") + exc.__cause__ = socket.gaierror(-2, "Name or service not known") + raise exc + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.side_effect = raise_dns + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://nonexistent.invalid", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + max_retries=5, + ) + + assert result.success is False + assert result.failed_at is not None + assert result.failed_at["http_status"] == 0 + # Fatal transport error → NO retries, NO backoff. + assert mock_client.post.call_count == 1 + mock_sleep.assert_not_called() + + def test_tls_failure_is_not_retried(self, tmp_path: Path) -> None: + """A ConnectError caused by a TLS/cert error (ssl.SSLError) fails fast.""" + import ssl + + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + def raise_tls(url: str, **kwargs: Any) -> Any: + exc = httpx.ConnectError("certificate verify failed") + exc.__cause__ = ssl.SSLError("CERTIFICATE_VERIFY_FAILED") + raise exc + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.side_effect = raise_tls + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + max_retries=5, + ) + + assert result.success is False + assert mock_client.post.call_count == 1 + mock_sleep.assert_not_called() + + def test_plain_connection_error_is_still_retried(self, tmp_path: Path) -> None: + """A ConnectError with NO DNS/TLS cause (e.g. reset) remains transient — retried then succeeds.""" + session_dir, metadata = _write_session(tmp_path, "s1", _make_events(1)) + tracker = MagicMock() + + outcomes = [httpx.ConnectError("connection reset by peer"), _Resp(200)] + + def side_effect(url: str, **kwargs: Any) -> Any: + item = outcomes.pop(0) + if isinstance(item, Exception): + raise item + return item + + with patch("httpx.Client") as mock_cls, patch("time.sleep") as mock_sleep: + mock_client = MagicMock() + mock_cls.return_value.__enter__.return_value = mock_client + mock_client.post.side_effect = side_effect + + result = run_upload( + sessions=[(session_dir, metadata)], + server_url="https://server", + api_key="", + tracker=tracker, + auth_strategy=ConstAuth(), + ) + + assert result.success is True + assert mock_client.post.call_count == 2 + assert mock_sleep.call_count == 1 + + +# --------------------------------------------------------------------------- +# CLI flag wiring — --max-retries and --timeout +# --------------------------------------------------------------------------- + + +class TestCliRobustnessFlags: + def test_max_retries_default_is_5(self) -> None: + from amplifier_module_tool_context_intelligence_upload.cli import _build_parser + + args = _build_parser().parse_args( + ["--path", "/tmp", "--server-url", "http://s", "--api-key", "k"] + ) + assert args.max_retries == 5 + + def test_max_retries_accepts_value(self) -> None: + from amplifier_module_tool_context_intelligence_upload.cli import _build_parser + + args = _build_parser().parse_args( + ["--path", "/tmp", "--server-url", "http://s", "--api-key", "k", "--max-retries", "0"] + ) + assert args.max_retries == 0 + + def test_timeout_default_is_none(self) -> None: + from amplifier_module_tool_context_intelligence_upload.cli import _build_parser + + args = _build_parser().parse_args( + ["--path", "/tmp", "--server-url", "http://s", "--api-key", "k"] + ) + assert args.timeout_s is None + + def test_timeout_accepts_value(self) -> None: + from amplifier_module_tool_context_intelligence_upload.cli import _build_parser + + args = _build_parser().parse_args( + ["--path", "/tmp", "--server-url", "http://s", "--api-key", "k", "--timeout", "120"] + ) + assert args.timeout_s == 120.0 diff --git a/modules/tool-context-intelligence-upload/tests/test_uploader.py b/modules/tool-context-intelligence-upload/tests/test_uploader.py index eed1bd9f..6f459334 100644 --- a/modules/tool-context-intelligence-upload/tests/test_uploader.py +++ b/modules/tool-context-intelligence-upload/tests/test_uploader.py @@ -177,8 +177,13 @@ def test_multiple_sessions_uploaded_in_order(self, tmp_path: Path) -> None: class TestUploadStopOnFailure: """Tests for run_upload — stop on failure scenarios.""" - def test_stops_on_503(self, tmp_path: Path) -> None: - """Fail on 3rd call: events_uploaded=2, call_count=3.""" + def test_stops_on_permanent_403(self, tmp_path: Path) -> None: + """A permanent 4xx (403) on the 3rd call aborts immediately: uploaded=2, calls=3. + + (Issue #338: a *transient* status like 503 is now retried; this test uses a + permanent status to prove the whole-run-stops-on-failure behaviour is preserved + for errors retrying cannot fix.) + """ events = _make_events(5) session_dir, metadata = _write_session(tmp_path, "abc", events) sessions = [(session_dir, metadata)] @@ -189,7 +194,7 @@ def test_stops_on_503(self, tmp_path: Path) -> None: def side_effect(url: str, **kwargs: Any) -> MagicMock: post_calls.append(1) if len(post_calls) == 3: - return _mock_response(503) + return _mock_response(403) return _mock_response(200) with patch("httpx.Client") as mock_client_cls: @@ -204,7 +209,11 @@ def side_effect(url: str, **kwargs: Any) -> MagicMock: assert mock_client.post.call_count == 3 def test_failed_at_populated_correctly(self, tmp_path: Path) -> None: - """status='failed', failed_at has session_id, event_index=2, http_status=503.""" + """status='failed', failed_at has session_id, event_index=2, http_status=403. + + Uses a permanent 403 (not 503, which is now retried) so the run aborts at the + failing event with no retry noise. + """ events = _make_events(5) session_dir, metadata = _write_session(tmp_path, "my-session", events) sessions = [(session_dir, metadata)] @@ -215,7 +224,7 @@ def test_failed_at_populated_correctly(self, tmp_path: Path) -> None: def side_effect(url: str, **kwargs: Any) -> MagicMock: post_calls.append(1) if len(post_calls) == 3: - return _mock_response(503) + return _mock_response(403) return _mock_response(200) with patch("httpx.Client") as mock_client_cls: @@ -230,7 +239,7 @@ def side_effect(url: str, **kwargs: Any) -> MagicMock: assert result.failed_at is not None assert result.failed_at["session_id"] == "my-session" assert result.failed_at["event_index"] == 2 - assert result.failed_at["http_status"] == 503 + assert result.failed_at["http_status"] == 403 # --------------------------------------------------------------------------- @@ -332,8 +341,10 @@ def test_authorization_header_set(self, tmp_path: Path) -> None: run_upload(sessions, "https://server", "sk-my-key", tracker) - _, kwargs = mock_client_cls.call_args - headers = kwargs.get("headers", {}) + # Issue #338: the Authorization header is now sent PER REQUEST (so token + # refresh can fire mid-run), not baked into the httpx.Client constructor. + post_kwargs = mock_client.post.call_args.kwargs + headers = post_kwargs.get("headers", {}) assert headers.get("Authorization") == "Bearer sk-my-key" def test_default_sends_replay_true_query_param(self, tmp_path: Path) -> None: @@ -457,7 +468,9 @@ def test_network_error_returns_failure_with_http_status_zero(self, tmp_path: Pat mock_client_cls.return_value.__enter__.return_value = mock_client mock_client.post.side_effect = httpx.HTTPError("Connection refused") - result = run_upload(sessions, "https://server", "api-key", tracker) + # max_retries=0 → single attempt, so a persistent transport error fails + # immediately (no backoff sleeps in the unit test). + result = run_upload(sessions, "https://server", "api-key", tracker, max_retries=0) assert result.success is False assert result.error == "Connection refused" @@ -609,7 +622,9 @@ def test_500_error_message_includes_server_body(self, tmp_path: Path) -> None: mock_response.text = "Internal Server Error" mock_client.post.return_value = mock_response - result = run_upload(sessions, "https://server/", "key", tracker) + # 500 is transient (retried); max_retries=0 → single attempt so the test + # asserts the terminal error message without incurring backoff sleeps. + result = run_upload(sessions, "https://server/", "key", tracker, max_retries=0) assert result.success is False assert result.error is not None