diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py
index c08da0fa1..7afc106c6 100644
--- a/headroom/proxy/handlers/anthropic.py
+++ b/headroom/proxy/handlers/anthropic.py
@@ -36,7 +36,7 @@ from headroom.proxy.auth_mode import (
 from headroom.proxy.compression_decision import CompressionDecision
 from headroom.proxy.forwarded_headers import resolve_client_ip
 from headroom.proxy.handlers._debug_dump import _debug_dump_mode, _redact_debug_value
-from headroom.proxy.helpers import extract_tags
+from headroom.proxy.helpers import extract_tags, relocate_system_messages_to_top_level
 from headroom.proxy.image_isolation import run_image_compression_isolated
 from headroom.proxy.memory_decision import MemoryDecision
 from headroom.proxy.memory_query import MemoryQuery
@@ -2842,6 +2842,28 @@ class AnthropicHandlerMixin:
                 (time.perf_counter() - pre_upstream_started_at) * 1000.0,
             )
 
+            # Anthropic wire-contract guard (issue #765). Any transform or
+            # pipeline extension above may have left a ``role="system"`` entry
+            # in ``messages`` (e.g. a harness system block relocated during
+            # compression). Anthropic rejects that with a 400 ("messages.0: use
+            # the top-level 'system' parameter ..."), so relocate it back to the
+            # top-level ``system`` parameter as the last step before forwarding.
+            relocated_messages, relocated_system, system_relocated = (
+                relocate_system_messages_to_top_level(body["messages"], body.get("system"))
+            )
+            if system_relocated:
+                body["messages"] = relocated_messages
+                if relocated_system is None:
+                    body.pop("system", None)
+                else:
+                    body["system"] = relocated_system
+                body_mutation_tracker.mark_mutated("system_role_relocated")
+                logger.warning(
+                    "[%s] Relocated role=system message(s) out of messages[] into the "
+                    "top-level system parameter (Anthropic wire-contract guard, issue #765)",
+                    request_id,
+                )
+
             # Byte-faithful forwarder support (PR-A3, fixes P0-2). At this
             # point body has been through every transform (image, compression,
             # memory, tool sort, pipeline extensions). If a transform reported
diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py
index b9cce8a23..6ea06e19d 100644
--- a/headroom/proxy/helpers.py
+++ b/headroom/proxy/helpers.py
@@ -843,6 +843,82 @@ def append_text_to_latest_user_chat_message(
     return messages, 0
 
 
+# Anthropic wire contract: the system prompt lives in the top-level ``system``
+# parameter; a ``role="system"`` entry inside ``messages`` is rejected with a
+# 400 ("messages.0: use the top-level 'system' parameter ..."). ``role`` /
+# ``content`` / ``type`` are bare wire keys used throughout this module; only
+# the load-bearing values are named here.
+_ROLE_SYSTEM = "system"
+_TEXT_BLOCK_TYPE = "text"
+
+
+def _system_message_to_blocks(message: dict[str, Any]) -> list[Any]:
+    """Convert a ``role="system"`` message into Anthropic system content blocks."""
+    content = message.get("content")
+    if isinstance(content, str):
+        return [{"type": _TEXT_BLOCK_TYPE, "text": content}] if content else []
+    if isinstance(content, list):
+        blocks: list[Any] = []
+        for block in content:
+            if isinstance(block, dict):
+                blocks.append(block)
+            elif isinstance(block, str) and block:
+                blocks.append({"type": _TEXT_BLOCK_TYPE, "text": block})
+        return blocks
+    return []
+
+
+def relocate_system_messages_to_top_level(
+    messages: list[dict[str, Any]],
+    system: Any,
+) -> tuple[list[dict[str, Any]], Any, bool]:
+    """Move any ``role="system"`` entries out of ``messages`` into ``system``.
+
+    Anthropic's Messages API rejects a ``system`` role inside ``messages`` with
+    HTTP 400 ("messages.0: use the top-level 'system' parameter for the initial
+    system prompt"). Internal transforms / pipeline extensions can leave a stray
+    system message in the list (e.g. a relocated harness system block during
+    compression). This is the Anthropic forwarder's last line of defense: it
+    guarantees the forwarded body never violates the wire contract, regardless
+    of which transform introduced the entry.
+
+    The relocated content is appended after any existing top-level ``system``
+    so wire order (system prompt, then conversation) is preserved and no content
+    is dropped.
+
+    Returns ``(clean_messages, new_system, changed)``. When no system-role
+    message is present the inputs pass through unchanged (``changed=False``) so
+    the common path is untouched.
+    """
+    system_indices = {
+        i for i, m in enumerate(messages) if isinstance(m, dict) and m.get("role") == _ROLE_SYSTEM
+    }
+    if not system_indices:
+        return messages, system, False
+
+    relocated_blocks: list[Any] = []
+    for i in sorted(system_indices):
+        relocated_blocks.extend(_system_message_to_blocks(messages[i]))
+
+    clean_messages = [m for i, m in enumerate(messages) if i not in system_indices]
+
+    if not relocated_blocks:
+        # System message(s) carried no content — drop the empty entries only.
+        return clean_messages, system, True
+
+    if system is None or system == "" or system == []:
+        new_system: Any = relocated_blocks
+    elif isinstance(system, str):
+        new_system = [{"type": _TEXT_BLOCK_TYPE, "text": system}, *relocated_blocks]
+    elif isinstance(system, list):
+        new_system = [*system, *relocated_blocks]
+    else:
+        # Unexpected shape — wrap rather than drop (safety-first: never lose content).
+        new_system = [system, *relocated_blocks]
+
+    return clean_messages, new_system, True
+
+
 def append_text_to_latest_user_input_item(
     body_input: list[dict[str, Any]],
     context_text: str,
diff --git a/tests/_gemini_live.py b/tests/_gemini_live.py
new file mode 100644
index 000000000..482814917
--- /dev/null
+++ b/tests/_gemini_live.py
@@ -0,0 +1,24 @@
+"""Helpers for optional live Gemini tests."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+GEMINI_QUOTA_STATUS = 429
+_GEMINI_QUOTA_MARKERS = (
+    "quota",
+    "rate limit",
+    "rate-limit",
+    "too many requests",
+)
+
+
+def skip_if_gemini_quota_exhausted(response: Any) -> None:
+    """Skip live Gemini tests when the configured key has no available quota."""
+    if getattr(response, "status_code", None) != GEMINI_QUOTA_STATUS:
+        return
+    body = getattr(response, "text", "") or ""
+    if any(marker in body.lower() for marker in _GEMINI_QUOTA_MARKERS):
+        pytest.skip("Gemini live API quota/rate limit exhausted")
diff --git a/tests/test_cli_doctor.py b/tests/test_cli_doctor.py
index c93a6d9a4..6b4837418 100644
--- a/tests/test_cli_doctor.py
+++ b/tests/test_cli_doctor.py
@@ -546,7 +546,16 @@ class TestDoctorCommand:
         monkeypatch.setattr(doctor_mod, "codex_config_path", lambda: tmp_path / "config.toml")
         monkeypatch.setattr(doctor_mod, "savings_path", lambda: tmp_path / "savings.json")
         monkeypatch.setattr(doctor_mod, "list_manifests", lambda: [])
-        for var in ("ANTHROPIC_BASE_URL", "OPENAI_BASE_URL", "HEADROOM_PORT"):
+        for var in (
+            "ANTHROPIC_API_KEY",
+            "ANTHROPIC_AUTH_TOKEN",
+            "ANTHROPIC_BASE_URL",
+            "CLAUDE_CODE_USE_BEDROCK",
+            "CLAUDE_CODE_USE_FOUNDRY",
+            "CLAUDE_CODE_USE_VERTEX",
+            "OPENAI_BASE_URL",
+            "HEADROOM_PORT",
+        ):
             monkeypatch.delenv(var, raising=False)
         return tmp_path
 
@@ -576,6 +585,7 @@ class TestDoctorCommand:
     def test_remote_control_warning_exits_1(self, runner, isolated, monkeypatch):
         monkeypatch.setattr(doctor_mod, "probe_json", self._probe(LIVEZ_OK, STATS_OK))
         monkeypatch.setattr(doctor_mod, "get_version", lambda: "0.26.0")
+        monkeypatch.setattr(doctor_mod, "detect_claude_code_version", lambda: None)
         (isolated / "settings.json").write_text(
             json.dumps({"env": {"ANTHROPIC_BASE_URL": "http://127.0.0.1:8787"}}),
             encoding="utf-8",
diff --git a/tests/test_proxy_gemini_integration.py b/tests/test_proxy_gemini_integration.py
index 0aba8d09d..4582a64dd 100644
--- a/tests/test_proxy_gemini_integration.py
+++ b/tests/test_proxy_gemini_integration.py
@@ -23,6 +23,7 @@ pytest.importorskip("httpx")
 from fastapi.testclient import TestClient  # noqa: E402
 
 from headroom.proxy.server import ProxyConfig, create_app  # noqa: E402
+from tests._gemini_live import skip_if_gemini_quota_exhausted  # noqa: E402
 
 GEMINI_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/openai"
 
@@ -63,6 +64,7 @@ class TestGeminiChatCompletions:
                 ],
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -93,6 +95,7 @@ class TestGeminiChatCompletions:
                 ],
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -110,6 +113,7 @@ class TestGeminiChatCompletions:
                 "messages": [{"role": "user", "content": "Count from 1 to 3."}],
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
 
         # Parse SSE stream
@@ -154,6 +158,7 @@ class TestGeminiChatCompletions:
                 "tool_choice": "auto",
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -182,6 +187,7 @@ class TestGeminiChatCompletions:
                 "response_format": {"type": "json_object"},
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -198,6 +204,7 @@ class TestGeminiModels:
     def test_list_models(self, gemini_client, api_key):
         """Can list available models."""
         response = gemini_client.get("/v1/models", headers={"Authorization": f"Bearer {api_key}"})
+        skip_if_gemini_quota_exhausted(response)
         # This goes through passthrough handler
         assert response.status_code == 200
         data = response.json()
@@ -211,11 +218,12 @@ class TestProxyStats:
     def test_stats_track_requests(self, gemini_client, api_key):
         """Proxy stats track Gemini requests."""
         # Make a request
-        gemini_client.post(
+        response = gemini_client.post(
             "/v1/chat/completions",
             headers={"Authorization": f"Bearer {api_key}"},
             json={"model": "gemini-2.0-flash", "messages": [{"role": "user", "content": "Hi"}]},
         )
+        skip_if_gemini_quota_exhausted(response)
 
         # Check stats
         stats_response = gemini_client.get("/stats")
diff --git a/tests/test_proxy_gemini_native_integration.py b/tests/test_proxy_gemini_native_integration.py
index ca05a56af..270c335ff 100644
--- a/tests/test_proxy_gemini_native_integration.py
+++ b/tests/test_proxy_gemini_native_integration.py
@@ -23,6 +23,7 @@ pytest.importorskip("httpx")
 from fastapi.testclient import TestClient  # noqa: E402
 
 from headroom.proxy.server import ProxyConfig, create_app  # noqa: E402
+from tests._gemini_live import skip_if_gemini_quota_exhausted  # noqa: E402
 
 
 @pytest.fixture
@@ -54,6 +55,7 @@ class TestGeminiNativeGenerateContent:
             f"/v1beta/models/gemini-2.0-flash:generateContent?key={api_key}",
             json={"contents": [{"parts": [{"text": "What is 2+2? Reply with just the number."}]}]},
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -78,6 +80,7 @@ class TestGeminiNativeGenerateContent:
                 "systemInstruction": {"parts": [{"text": "Always respond with exactly one word."}]},
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
         text = data["candidates"][0]["content"]["parts"][0]["text"]
@@ -96,6 +99,7 @@ class TestGeminiNativeGenerateContent:
                 ]
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
         text = data["candidates"][0]["content"]["parts"][0]["text"].lower()
@@ -126,6 +130,7 @@ class TestGeminiNativeGenerateContent:
                 ],
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -150,6 +155,7 @@ class TestGeminiNativeGenerateContent:
                 "generationConfig": {"maxOutputTokens": 50, "temperature": 0.1},
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
         # Response should be limited by maxOutputTokens
@@ -178,6 +184,7 @@ class TestGeminiNativeCompression:
                 ]
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
         text = data["candidates"][0]["content"]["parts"][0]["text"]
@@ -204,6 +211,7 @@ class TestGeminiNativeCompression:
                 ]
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         # The request should succeed - user messages are protected from compression
 
@@ -214,10 +222,11 @@ class TestGeminiNativeStats:
     def test_stats_track_gemini_provider(self, gemini_native_client, api_key):
         """Stats show requests under 'gemini' provider."""
         # Make a request
-        gemini_native_client.post(
+        response = gemini_native_client.post(
             f"/v1beta/models/gemini-2.0-flash:generateContent?key={api_key}",
             json={"contents": [{"parts": [{"text": "Hi"}]}]},
         )
+        skip_if_gemini_quota_exhausted(response)
 
         stats = gemini_native_client.get("/stats").json()
         assert "gemini" in stats["requests"]["by_provider"]
@@ -225,10 +234,11 @@ class TestGeminiNativeStats:
 
     def test_stats_track_model(self, gemini_native_client, api_key):
         """Stats track the specific model used."""
-        gemini_native_client.post(
+        response = gemini_native_client.post(
             f"/v1beta/models/gemini-2.0-flash:generateContent?key={api_key}",
             json={"contents": [{"parts": [{"text": "Hi"}]}]},
         )
+        skip_if_gemini_quota_exhausted(response)
 
         stats = gemini_native_client.get("/stats").json()
         assert "gemini-2.0-flash" in stats["requests"]["by_model"]
@@ -258,6 +268,7 @@ class TestGeminiNativeErrorHandling:
         response = gemini_native_client.post(
             f"/v1beta/models/gemini-2.0-flash:generateContent?key={api_key}", json={"contents": []}
         )
+        skip_if_gemini_quota_exhausted(response)
         # Should either return error or handle gracefully
         assert response.status_code in [200, 400]
 
@@ -272,6 +283,7 @@ class TestGeminiNativeHeaderAuth:
             headers={"x-goog-api-key": api_key},
             json={"contents": [{"parts": [{"text": "Hi"}]}]},
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
 
 
@@ -284,6 +296,7 @@ class TestGeminiNativeCountTokens:
             f"/v1beta/models/gemini-2.0-flash:countTokens?key={api_key}",
             json={"contents": [{"parts": [{"text": "Hello, world!"}]}]},
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -301,6 +314,7 @@ class TestGeminiNativeCountTokens:
                 "systemInstruction": {"parts": [{"text": "You are a helpful assistant."}]},
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         # Note: systemInstruction may not be supported by countTokens in all versions
         assert response.status_code in [200, 400]
         if response.status_code == 200:
@@ -332,6 +346,7 @@ class TestGeminiNativeCountTokens:
                 ]
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
 
@@ -357,6 +372,7 @@ class TestGeminiNativeCountTokens:
                 ]
             },
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
         assert "totalTokens" in data
@@ -369,6 +385,7 @@ class TestGeminiNativeCountTokens:
             headers={"x-goog-api-key": api_key},
             json={"contents": [{"parts": [{"text": "Hello"}]}]},
         )
+        skip_if_gemini_quota_exhausted(response)
         assert response.status_code == 200
         data = response.json()
         assert "totalTokens" in data
diff --git a/tests/test_proxy_handler_helpers.py b/tests/test_proxy_handler_helpers.py
index 7fb8f442e..a11e437b3 100644
--- a/tests/test_proxy_handler_helpers.py
+++ b/tests/test_proxy_handler_helpers.py
@@ -17,7 +17,10 @@ from headroom.proxy.handlers.openai import (
     _passthrough_usage_from_json,
     _prefers_http1_passthrough,
 )
-from headroom.proxy.helpers import _headroom_bypass_enabled
+from headroom.proxy.helpers import (
+    _headroom_bypass_enabled,
+    relocate_system_messages_to_top_level,
+)
 from headroom.proxy.server import HeadroomProxy
 
 
@@ -268,6 +271,46 @@ def test_openai_handler_prefix_helpers_cover_edge_cases() -> None:
     assert changed == 1
 
 
+def test_relocate_system_messages_moves_stray_system_into_top_level() -> None:
+    # Issue #765: compression relocated the harness system block into
+    # messages[0] as a role="system" entry, which Anthropic rejects with a 400.
+    # The forwarder guard must move it back to the top-level `system` parameter.
+    messages = [
+        {"role": "system", "content": "You are a harness."},
+        {"role": "user", "content": "hi"},
+    ]
+    clean, system, changed = relocate_system_messages_to_top_level(messages, None)
+
+    assert changed is True
+    # No role="system" entry may survive in messages[] — that is the wire-contract violation.
+    assert all(m.get("role") != "system" for m in clean)
+    assert clean == [{"role": "user", "content": "hi"}]
+    # The relocated content lands in the top-level system parameter.
+    assert system == [{"type": "text", "text": "You are a harness."}]
+
+
+def test_relocate_system_messages_appends_to_existing_system() -> None:
+    messages = [
+        {"role": "system", "content": [{"type": "text", "text": "B"}]},
+        {"role": "user", "content": "hi"},
+    ]
+    clean, system, changed = relocate_system_messages_to_top_level(messages, "A")
+
+    assert changed is True
+    assert clean == [{"role": "user", "content": "hi"}]
+    # Existing system first, relocated content after — wire order preserved.
+    assert system == [{"type": "text", "text": "A"}, {"type": "text", "text": "B"}]
+
+
+def test_relocate_system_messages_noop_without_system_entry() -> None:
+    messages = [{"role": "user", "content": "hi"}]
+    clean, system, changed = relocate_system_messages_to_top_level(messages, "A")
+
+    assert changed is False
+    assert clean is messages
+    assert system == "A"
+
+
 def test_headroom_bypass_helper_is_transport_neutral() -> None:
     assert _headroom_bypass_enabled({"x-headroom-bypass": "true"}) is True
     assert _headroom_bypass_enabled({"x-headroom-bypass": " TRUE "}) is True
diff --git a/tests/test_realignment_live_multi_turn.py b/tests/test_realignment_live_multi_turn.py
index b3ff95f2e..8608a51c8 100644
--- a/tests/test_realignment_live_multi_turn.py
+++ b/tests/test_realignment_live_multi_turn.py
@@ -48,6 +48,7 @@ from fastapi.testclient import TestClient  # noqa: E402
 
 from headroom.proxy.server import ProxyConfig, create_app  # noqa: E402
 from tests._dotenv import autouse_apply_env, load_env_overrides  # noqa: E402
+from tests._gemini_live import skip_if_gemini_quota_exhausted  # noqa: E402
 
 # ---------------------------------------------------------------------------
 # Module-level config
@@ -639,6 +640,7 @@ def test_gemini_multi_turn_through_proxy(proxy_client: TestClient) -> None:
     ]
 
     resp1 = proxy_client.post(url, json={"contents": contents})
+    skip_if_gemini_quota_exhausted(resp1)
     assert resp1.status_code == 200, resp1.text
     data1 = resp1.json()
     text1 = data1["candidates"][0]["content"]["parts"][0]["text"]
@@ -648,6 +650,7 @@ def test_gemini_multi_turn_through_proxy(proxy_client: TestClient) -> None:
     contents.append({"role": "user", "parts": [{"text": "Now reply with the single word WORLD."}]})
 
     resp2 = proxy_client.post(url, json={"contents": contents})
+    skip_if_gemini_quota_exhausted(resp2)
     assert resp2.status_code == 200, resp2.text
     data2 = resp2.json()
     text2 = data2["candidates"][0]["content"]["parts"][0]["text"]
diff --git a/tests/test_ssl_context.py b/tests/test_ssl_context.py
index dda61f854..9b78ae874 100644
--- a/tests/test_ssl_context.py
+++ b/tests/test_ssl_context.py
@@ -49,6 +49,7 @@ Tfx2hBGZ0UogmREaXFi099rmaueZ0HIBn51b3kYqc7of5TI0fHwSHF4GdXXs2OZi
 kF9agIt8Q8t/2kviMn2roInGTwTyPYOEQV0m
 -----END CERTIFICATE-----
 """
+_TEST_CA_COUNT = 1
 
 
 @pytest.fixture()
@@ -70,6 +71,10 @@ def _clean_env(monkeypatch):
         monkeypatch.delenv(var, raising=False)
 
 
+def _default_x509_ca_count() -> int:
+    return ssl.create_default_context().cert_store_stats()["x509_ca"]
+
+
 class FakeSSLContext:
     def __init__(self, verify_flags: int = 0) -> None:
         self.verify_flags = verify_flags
@@ -136,9 +141,10 @@ class TestFindCaBundleWithValidPem:
         ctx = find_ca_bundle()
         assert isinstance(ctx, ssl.SSLContext)
         stats = ctx.cert_store_stats()
-        # The default trust store has dozens of CAs; if only the test cert
-        # were loaded (replacement), x509_ca would be 1.
-        assert stats["x509_ca"] > 1
+        # Additive loading preserves whatever the runner's default trust store
+        # contains. Some minimal CI images have a tiny or empty default store, so
+        # compare against the local baseline instead of assuming "dozens" of CAs.
+        assert stats["x509_ca"] >= _default_x509_ca_count() + _TEST_CA_COUNT
 
 
 class TestFindCaBundlePriority:
@@ -265,8 +271,9 @@ class TestBuildHttpxVerify:
             assert ctx.verify_flags & strict_flag == 0
         # Still a real verifying context — NOT verify=False.
         assert ctx.verify_mode == ssl.CERT_REQUIRED
-        # Default trust store retained (additive, not a 1-cert replacement).
-        assert ctx.cert_store_stats()["x509_ca"] > 1
+        # Default trust store retained for this runner, not replaced by a custom
+        # one-cert bundle.
+        assert ctx.cert_store_stats()["x509_ca"] == _default_x509_ca_count()
 
     def test_custom_ca_takes_precedence_over_toggle(self, monkeypatch, ca_pem_file):
         """A configured CA bundle wins; the result is that bundle's context."""