fix(cli): honor bounded Retry-After for Hosted ingest 429s

docs/CLI.md

@@ -89,6 +89,14 @@ the freed name; if no finding reproduces, restore the backup. `mutiny run` has n
 
 **Hosted sync (M-PR8C):** After local campaign completion, CLI POSTs to `/api/ingest/v1/*` with `schema_version=1` and `redaction.applied=true`. Bearer from `MUTINY_API_TOKEN` when set. Local success + sync failure → exit **3** (local result remains authoritative). Missing `api_url` with `--hosted` → exit 2. Config URL alone never selects Hosted.
 
+**Rate limits:** Each ingest POST retries HTTP `429` up to twice (three attempts
+total), reusing the identical sanitized payload and IDs. `Retry-After` may be
+integer seconds or an HTTP date; each delay is capped at five seconds, with a
+one-second fallback for missing/malformed headers and zero for elapsed delays.
+Other HTTP errors and transport failures are not retried. After exhaustion, the
+existing sync failure/exit behavior and sanitized pending file are preserved.
+See [ingest retry semantics](./HOSTED_INGESTION.md#bounded-rate-limit-retries-43).
+
 **Destination policy:** By default, Hosted sync accepts only HTTP(S) URLs whose
 host is `localhost` or a literal loopback IP (`127.0.0.0/8` or `::1`). Other hosts
 require `--allow-remote-hosted` on the command line, even with `--hosted-url` or

docs/HOSTED_INGESTION.md

@@ -250,7 +250,7 @@ CLI still reports local success/failure as today
         ↓
 warn on stderr; write .mutiny/hosted-pending/<campaign_id>.json (sanitized)
         ↓
-retry manually; `mutiny sync` does not exist — no automatic retry in M-PR8C
+manual recovery after bounded 429 retries; `mutiny sync` does not exist
 ```
 
 **Rules:**
@@ -261,6 +261,27 @@ retry manually; `mutiny sync` does not exist — no automatic retry in M-PR8C
 4. No Hosted round-trip inside Core evaluation loops.
 5. `MUTINY_DISABLE_SECRET_REDACTION=1` refuses Hosted upload (sync exit 3 after local success); local persistence still allowed.
 
+### Bounded rate-limit retries (#43)
+
+The CLI retries each idempotent ingest POST only on HTTP `429`: at most **two
+retries** after the initial attempt (**three attempts total**). Campaign open,
+event/artifact batches, completion, regressions, and the test-run client method
+share this behavior. The request's already-redacted payload and identifiers are
+reused unchanged; payloads and timestamps are not regenerated between attempts.
+
+`Retry-After` accepts integer seconds or an HTTP date. Each delay is clamped to
+**0–5 seconds**; past dates and negative delays wait zero seconds. Missing or
+malformed headers use **one second**. Scheduled waiting is at most **10 seconds
+per POST**, in addition to the existing HTTP timeout for each attempt. No sleep
+occurs after the last failed attempt.
+
+Non-429 responses and transport errors are not retried. Exhausted 429s use the
+existing failure path: local results remain authoritative, local success plus
+sync failure exits `3`, and sanitized pending JSON is saved when enabled.
+Replaying pending files and a `mutiny sync` command remain future work; these
+immediate retries do not add a background sync queue. Redaction, destination
+validation, bearer authentication, and redirect refusal remain required.
+
 ---
 
 ## 12. Versioning
@@ -471,5 +492,5 @@ Preserved by M-PR8E (customer Hosted exec retired; demo + ingest retained):
 ### Current limitations (post M-PR8E)
 
 - `mutiny test --hosted` not wired (payload builder only) — remaining CLI sync work / later
-- Automatic retry / `mutiny sync` command deferred (pending JSON written on sync failure)
+- Pending-file replay / `mutiny sync` command deferred; immediate HTTP 429 retries are bounded as described above (pending JSON written on final sync failure)
 - Web cannot distinguish local-success + sync-failure from a missing Hosted row: sync failures do not fully ingest, so Hosted never claims those runs were received. A terminal `failed` status on an ingested `execution_mode=local_cli` campaign means the CLI reported campaign failure, not Hosted sync failure.

packages/mutiny_cli/src/mutiny_cli/hosted_sync.py

@@ -14,9 +14,11 @@
 import json
 import logging
 import os
+import time
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
+from email.utils import parsedate_to_datetime
 from pathlib import Path
 from typing import Any, Literal
 from urllib.parse import urljoin
@@ -31,6 +33,9 @@
 INGEST_SCHEMA_VERSION = 1
 EXECUTION_MODE = "local_cli"
 DEFAULT_TIMEOUT_S = 30.0
+MAX_RATE_LIMIT_RETRIES = 2
+MAX_RETRY_AFTER_S = 5.0
+DEFAULT_RETRY_AFTER_S = 1.0
 
 # Local campaign OK but Hosted sync failed / refused (distinct from local fail=1).
 SYNC_FAILED_EXIT = 3
@@ -484,13 +489,30 @@ def write_pending_bundle(bundle: LocalRunBundle, payloads: dict[str, Any]) -> Pa
         "kind": "hosted_pending_sync",
         "campaign_id": bundle.campaign_id,
         "created_at": _utcnow(),
-        "note": "Sanitized ingest payloads; not executable. Retry is future work.",
+        "note": "Sanitized ingest payloads; not executable. Pending-file replay is future work.",
         "payloads": payloads,
     }
     path.write_text(json.dumps(doc, indent=2), encoding="utf-8")
     return path
 
 
+def _retry_after_seconds(value: str | None) -> float:
+    """Parse delta-seconds or an HTTP date, with a bounded fallback delay."""
+    if value is None:
+        return DEFAULT_RETRY_AFTER_S
+    try:
+        delay = float(int(value))
+    except (ValueError, OverflowError):
+        try:
+            when = parsedate_to_datetime(value)
+            if when.tzinfo is None:
+                when = when.replace(tzinfo=timezone.utc)
+            delay = (when - datetime.now(timezone.utc)).total_seconds()
+        except (ValueError, TypeError, OverflowError):
+            return DEFAULT_RETRY_AFTER_S
+    return min(MAX_RETRY_AFTER_S, max(0.0, delay))
+
+
 class HostedIngestClient:
     """Thin httpx client for ``/api/ingest/v1/*`` (timeouts; no token logging)."""
 
@@ -553,41 +575,38 @@ def _raise_for_response(self, response: Any, *, action: str) -> None:
                 msg = f"{msg}: {detail[:160]}"
         raise HostedSyncError(msg, error_code=code, http_status=status)
 
-    def open_campaign(self, payload: dict[str, Any]) -> dict[str, Any]:
+    def _post(
+        self, path: str, payload: dict[str, Any], *, action: str
+    ) -> dict[str, Any]:
         with self._client() as client:
-            r = client.post("/api/ingest/v1/campaigns", json=payload)
-            self._raise_for_response(r, action="campaign open")
+            for attempt in range(MAX_RATE_LIMIT_RETRIES + 1):
+                r = client.post(path, json=payload)
+                if r.status_code != 429 or attempt == MAX_RATE_LIMIT_RETRIES:
+                    break
+                time.sleep(_retry_after_seconds(r.headers.get("Retry-After")))
+            self._raise_for_response(r, action=action)
             return r.json()
 
+    def open_campaign(self, payload: dict[str, Any]) -> dict[str, Any]:
+        return self._post("/api/ingest/v1/campaigns", payload, action="campaign open")
+
     def post_batch(self, campaign_id: str, payload: dict[str, Any]) -> dict[str, Any]:
-        with self._client() as client:
-            r = client.post(
-                f"/api/ingest/v1/campaigns/{campaign_id}/batch", json=payload
-            )
-            self._raise_for_response(r, action="batch ingest")
-            return r.json()
+        return self._post(
+            f"/api/ingest/v1/campaigns/{campaign_id}/batch", payload, action="batch ingest"
+        )
 
     def complete_campaign(
         self, campaign_id: str, payload: dict[str, Any]
     ) -> dict[str, Any]:
-        with self._client() as client:
-            r = client.post(
-                f"/api/ingest/v1/campaigns/{campaign_id}/complete", json=payload
-            )
-            self._raise_for_response(r, action="campaign complete")
-            return r.json()
+        return self._post(
+            f"/api/ingest/v1/campaigns/{campaign_id}/complete", payload, action="campaign complete"
+        )
 
     def post_regression(self, payload: dict[str, Any]) -> dict[str, Any]:
-        with self._client() as client:
-            r = client.post("/api/ingest/v1/regressions", json=payload)
-            self._raise_for_response(r, action="regression ingest")
-            return r.json()
+        return self._post("/api/ingest/v1/regressions", payload, action="regression ingest")
 
     def post_test_run(self, payload: dict[str, Any]) -> dict[str, Any]:
-        with self._client() as client:
-            r = client.post("/api/ingest/v1/test-runs", json=payload)
-            self._raise_for_response(r, action="test-run ingest")
-            return r.json()
+        return self._post("/api/ingest/v1/test-runs", payload, action="test-run ingest")
 
 
 def sync_local_campaign(

tests/unit/test_cli_hosted_sync.py

@@ -1078,3 +1078,183 @@ def test_30_sync_module_has_no_exec_gadgets() -> None:
         "eval(",
     ):
         assert needle not in src
+
+
+@pytest.fixture
+def rate_limited_client(monkeypatch):
+    import httpx
+
+    requests = []
+    sleeps = []
+    responses = []
+    client = HostedIngestClient("http://127.0.0.1:8000", headers={"Authorization": f"Bearer {FAKE_TOKEN}"})
+
+    def handle(request):
+        requests.append(request)
+        response = responses.pop(0)
+        if isinstance(response, Exception):
+            raise response
+        return response
+
+    monkeypatch.delenv("MUTINY_DISABLE_SECRET_REDACTION", raising=False)
+    monkeypatch.setattr(hosted_sync.time, "sleep", sleeps.append)
+    monkeypatch.setattr(client, "_client", lambda: httpx.Client(
+        base_url=client.api_url, headers=client.headers, follow_redirects=False,
+        transport=httpx.MockTransport(handle),
+    ))
+    return client, requests, sleeps, responses
+
+
+@pytest.mark.parametrize("method,path", [
+    ("open_campaign", "/api/ingest/v1/campaigns"),
+    ("post_batch", "/api/ingest/v1/campaigns/c1/batch"),
+    ("complete_campaign", "/api/ingest/v1/campaigns/c1/complete"),
+    ("post_regression", "/api/ingest/v1/regressions"),
+    ("post_test_run", "/api/ingest/v1/test-runs"),
+])
+def test_ingest_posts_retry_429_with_identical_payload(rate_limited_client, method, path, caplog):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    responses.extend([
+        httpx.Response(429, headers={"Retry-After": "2"}),
+        httpx.Response(200, json={"ok": True}),
+    ])
+    payload = {"schema_version": 1, "redaction": {"applied": True}, "event_id": "stable-id"}
+    args = ("c1", payload) if method in {"post_batch", "complete_campaign"} else (payload,)
+    assert getattr(client, method)(*args) == {"ok": True}
+    assert sleeps == [2.0]
+    assert len(requests) == 2
+    assert all(request.url.path == path for request in requests)
+    assert requests[0].content == requests[1].content
+    assert all(request.headers["Authorization"] == f"Bearer {FAKE_TOKEN}" for request in requests)
+    assert FAKE_TOKEN not in caplog.text
+
+
+@pytest.mark.parametrize("value,expected", [
+    (None, 1.0), ("", 1.0), ("invalid", 1.0), ("NaN", 1.0), ("inf", 1.0),
+    ("-10", 0.0), ("0", 0.0), ("999999", 5.0),
+])
+def test_retry_delay_is_bounded_with_safe_fallback(rate_limited_client, value, expected):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    responses.extend([
+        httpx.Response(429, headers={} if value is None else {"Retry-After": value}),
+        httpx.Response(200, json={}),
+    ])
+    client.open_campaign({})
+    assert sleeps == [expected]
+    assert len(requests) == 2
+
+
+@pytest.mark.parametrize("offset,expected", [(3, 3.0), (-10, 0.0), (60, 5.0)])
+def test_retry_after_http_date(rate_limited_client, monkeypatch, offset, expected):
+    import httpx
+    from datetime import datetime, timedelta, timezone
+    from email.utils import format_datetime
+
+    now = datetime(2026, 11, 11, 12, 0, tzinfo=timezone.utc)
+
+    class Clock(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return now
+
+    monkeypatch.setattr(hosted_sync, "datetime", Clock)
+    client, _, sleeps, responses = rate_limited_client
+    responses.extend([
+        httpx.Response(429, headers={"Retry-After": format_datetime(now + timedelta(seconds=offset), usegmt=True)}),
+        httpx.Response(200, json={}),
+    ])
+    client.open_campaign({})
+    assert sleeps == [expected]
+
+
+def test_repeated_429_exhausts_retries_and_keeps_pending_bundle(rate_limited_client, tmp_path):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    responses.extend(httpx.Response(
+        429, headers={"Retry-After": "999"},
+        json={"error": {"code": "rate_limit_exceeded", "message": "wait"}},
+    ) for _ in range(3))
+    outcome = sync_local_campaign(
+        _bundle(tmp_path, campaign_id="limited", secret_in_trace=True),
+        api_url=client.api_url, client=client,
+    )
+    assert not outcome.ok
+    assert outcome.http_status == 429
+    assert outcome.error_code == "rate_limit_exceeded"
+    assert len(requests) == 3
+    assert sleeps == [5.0, 5.0]
+    assert outcome.pending_path.is_file()
+    saved = outcome.pending_path.read_text(encoding="utf-8")
+    assert SECRET_VALUE not in saved
+    assert FAKE_TOKEN not in saved
+    assert json.loads(saved)["payloads"]["batch"]["redaction"]["applied"] is True
+
+
+@pytest.mark.parametrize("status", [400, 401, 403, 409, 413, 500])
+def test_non_429_errors_do_not_retry(rate_limited_client, status):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    responses.append(httpx.Response(status, json={"error": {"code": "original_error"}}))
+    with pytest.raises(HostedSyncError) as caught:
+        client.open_campaign({})
+    assert caught.value.http_status == status
+    assert caught.value.error_code == "original_error"
+    assert len(requests) == 1
+    assert sleeps == []
+
+
+def test_retry_stops_on_a_different_error(rate_limited_client):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    responses.extend([httpx.Response(429, headers={"Retry-After": "1"}), httpx.Response(401, json={})])
+    with pytest.raises(HostedSyncError) as caught:
+        client.open_campaign({})
+    assert caught.value.http_status == 401
+    assert len(requests) == 2
+    assert sleeps == [1.0]
+
+
+def test_transport_errors_are_not_retried(rate_limited_client):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    responses.append(httpx.ReadTimeout("offline timeout"))
+    with pytest.raises(httpx.ReadTimeout):
+        client.open_campaign({})
+    assert len(requests) == 1
+    assert sleeps == []
+
+
+def test_sync_recovers_from_429_without_regenerating_or_unredacting_payloads(rate_limited_client, tmp_path):
+    import httpx
+
+    client, requests, sleeps, responses = rate_limited_client
+    for _ in range(4):
+        responses.extend([httpx.Response(429, headers={"Retry-After": "0"}), httpx.Response(200, json={})])
+    outcome = sync_local_campaign(
+        _bundle(tmp_path, violated=True, secret_in_trace=True, with_regression=True),
+        api_url=client.api_url, client=client,
+    )
+    assert outcome.ok
+    assert sleeps == [0.0] * 4
+    assert len(requests) == 8
+    for index in range(0, len(requests), 2):
+        assert requests[index].content == requests[index + 1].content
+        assert SECRET_VALUE.encode() not in requests[index].content
+        assert json.loads(requests[index].content)["redaction"]["applied"] is True
+
+
+def test_retry_support_does_not_bypass_disabled_redaction(rate_limited_client, tmp_path, monkeypatch):
+    client, requests, sleeps, _ = rate_limited_client
+    monkeypatch.setenv("MUTINY_DISABLE_SECRET_REDACTION", "1")
+    outcome = sync_local_campaign(_bundle(tmp_path), api_url=client.api_url, client=client)
+    assert not outcome.ok
+    assert outcome.error_code == "redaction_disabled"
+    assert requests == sleeps == []