Merge pull request #52 from pewdiepie-archdaemon/feature/runtime-context-resolution

feat(runtime): resolve compact context once at turn preparation

tests/conftest.py

@@ -201,3 +201,65 @@ def _no_leaked_module_stubs():
         "at teardown.",
         pytrace=False,
     )
+
+
+@pytest.fixture(autouse=True)
+def _no_context_window_network_probe(request):
+    """Keep the turn context-window resolver offline in tests.
+
+    Compact turns resolve their window before the first model request, and
+    most tests drive them with placeholder endpoints. Only the resolver's two
+    I/O edges are replaced: URL resolution (DNS/Tailscale lookups) and its
+    HTTP client, which records each attempted metadata request and fails it
+    as a transport error. Everything else (route wiring, caching, credential
+    scoping, evidence selection) runs for real, so an unintended extra probe
+    stays visible through the ``context_probe_ledger`` fixture.
+
+    Modules that install their own fake client opt out with a module-level
+    ``CONTEXT_PROBE_NETWORK = True``.
+    """
+    ledger = []
+    if getattr(request.module, "CONTEXT_PROBE_NETWORK", False):
+        yield ledger
+        return
+    try:
+        from src.agent_runtime import context_resolution
+    except Exception:
+        yield ledger
+        return
+
+    class _OfflineMetadataClient:
+        def __init__(self, timeout=None):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            return False
+
+        async def get(self, url, headers=None):
+            ledger.append({"url": url, "headers": dict(headers or {})})
+            raise context_resolution.httpx.ConnectError("network disabled in tests")
+
+    def _offline_provider_urls(endpoint_url):
+        base = endpoint_url.split("/v1")[0] if "/v1" in endpoint_url else endpoint_url.rstrip("/")
+        return base + "/v1/models", endpoint_url
+
+    # A private patcher keeps the shared ``monkeypatch`` fixture's teardown
+    # order unchanged for tests that check their own sys.modules hygiene.
+    patcher = pytest.MonkeyPatch()
+    patcher.setattr(context_resolution, "_http_client", _OfflineMetadataClient)
+    patcher.setattr(context_resolution, "_provider_urls", _offline_provider_urls)
+    context_resolution.clear_probe_cache()
+    try:
+        yield ledger
+    finally:
+        patcher.undo()
+        context_resolution.clear_probe_cache()
+
+
+@pytest.fixture
+def context_probe_ledger(_no_context_window_network_probe):
+    """Metadata requests the context resolver attempted during this test."""
+    return _no_context_window_network_probe

tests/test_context_resolution.py

@@ -0,0 +1,621 @@
+"""Turn-preparation context window resolution and its compact-runtime use."""
+import asyncio
+from types import SimpleNamespace
+import json
+
+import httpx
+import pytest
+
+import src.model_context as model_context
+from src.agent_runtime import context_resolution as cr
+from src.agent_runtime.context_resolution import (
+    ContextEvidence,
+    ContextObservation,
+    UNRESOLVED_CONTEXT,
+    combine_observations,
+    context_metrics,
+    resolve_effective_context,
+)
+from src.clean_agent_preview import stream_preview
+from src.tool_policy import ToolPolicy
+from src.turn_contract import resolve_full_inventory_contract
+
+REMOTE = "http://provider.test/v1/chat/completions"
+LOCAL = "http://127.0.0.1:8080/v1/chat/completions"
+AUTH = {"Authorization": "Bearer secret-token", "Content-Type": "application/json"}
+
+# Opt out of the conftest guard; every test here installs a fake HTTP client.
+CONTEXT_PROBE_NETWORK = True
+
+
+def _obs(evidence, value, source="test"):
+    return ContextObservation(evidence, value, source)
+
+
+class _Response:
+    def __init__(self, status=200, payload=None):
+        self.status_code = status
+        self._payload = payload
+
+    def json(self):
+        if isinstance(self._payload, Exception):
+            raise self._payload
+        return self._payload
+
+
+class _ModelStream:
+    def __init__(self, lines, status=200, text=""):
+        self.status_code = status
+        self.text = text
+        self._lines = lines
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    async def aread(self):
+        return self.text.encode()
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise AssertionError(f"unexpected provider status {self.status_code}")
+
+    async def aiter_lines(self):
+        for line in self._lines:
+            yield line
+
+
+def _answer_lines(usage=None):
+    lines = ["data: " + json.dumps({"choices": [{"delta": {"content": "Done."}}]})]
+    if usage is not None:
+        lines.append("data: " + json.dumps({"choices": [{"delta": {}}], "usage": usage}))
+    lines.append("data: [DONE]")
+    return lines
+
+
+class _FakeNetwork:
+    """One fake HTTP surface for both metadata GETs and model streams.
+
+    The compact runtime and the probe share ``httpx.AsyncClient``; recording
+    every call in order proves which phase performed which I/O.
+    """
+
+    def __init__(self, routes=None, *, stream_responses=None, get_delay=0.0, get_error=None):
+        self.routes = routes or {}
+        self.stream_responses = list(stream_responses or [])
+        self.get_delay = get_delay
+        self.get_error = get_error
+        self.calls = []
+        self.requests = []
+        self.client_kwargs = []
+
+    def client_factory(self):
+        network = self
+
+        class Client:
+            def __init__(self, **kwargs):
+                network.client_kwargs.append(kwargs)
+
+            async def __aenter__(self):
+                return self
+
+            async def __aexit__(self, *args):
+                return False
+
+            async def get(self, url, headers=None):
+                network.calls.append(("get", url, dict(headers or {})))
+                if network.get_delay:
+                    await asyncio.sleep(network.get_delay)
+                if network.get_error is not None:
+                    raise network.get_error
+                for suffix, response in network.routes.items():
+                    if url.endswith(suffix):
+                        return response
+                return _Response(404, None)
+
+            def stream(self, method, url, headers=None, json=None):
+                network.calls.append(("stream", url, dict(headers or {})))
+                network.requests.append(json)
+                if network.stream_responses:
+                    return network.stream_responses.pop(0)
+                return _ModelStream(_answer_lines({"prompt_tokens": 2048, "completion_tokens": 3}))
+
+        return Client
+
+
+@pytest.fixture(autouse=True)
+def _fresh_cache(monkeypatch):
+    cr.clear_probe_cache()
+    monkeypatch.setattr(model_context, "is_local_endpoint", lambda url: "127.0.0.1" in url)
+    # The real URL builder may resolve hosts; these tests stay off the network.
+    monkeypatch.setattr(
+        "src.endpoint_resolver.build_models_url", lambda base: base.split("/v1")[0] + "/v1/models",
+    )
+    monkeypatch.setattr("src.endpoint_resolver.resolve_url", lambda url: url)
+    yield
+    cr.clear_probe_cache()
+
+
+def _install(monkeypatch, network):
+    monkeypatch.setattr(cr.httpx, "AsyncClient", network.client_factory())
+
+    def legacy_get(url, *args, **kwargs):
+        # The legacy model_context probe is synchronous; record it so a
+        # terminal-metrics probe through that path is caught as well.
+        network.calls.append(("sync_get", url, dict(kwargs.get("headers") or {})))
+        return _Response(404, None)
+
+    monkeypatch.setattr(cr.httpx, "get", legacy_get)
+    return network
+
+
+def _catalog(model, **fields):
+    return _Response(200, {"data": [{"id": model, **fields}]})
+
+
+# ---------------------------------------------------------------------------
+# Pure combination rules
+# ---------------------------------------------------------------------------
+
+def test_operator_declared_window_replaces_known_table_without_contradiction():
+    resolution = combine_observations([
+        _obs(ContextEvidence.KNOWN_TABLE, 128000),
+        _obs(ContextEvidence.OPERATOR_DECLARED, 200000),
+    ])
+    assert resolution.effective == 200000
+    assert resolution.evidence is ContextEvidence.OPERATOR_DECLARED
+    assert not resolution.mismatch
+
+
+def test_operator_declared_window_caps_measured_window():
+    resolution = combine_observations([
+        _obs(ContextEvidence.PROVIDER_ADVERTISED, 131072),
+        _obs(ContextEvidence.OPERATOR_DECLARED, 32768),
+    ])
+    assert (resolution.effective, resolution.evidence) == (32768, ContextEvidence.OPERATOR_DECLARED)
+    # A tighter declared transport limit is a cap, not a contradiction.
+    assert not resolution.mismatch
+
+
+def test_operator_declaration_above_provider_is_contradiction_and_provider_wins():
+    resolution = combine_observations([
+        _obs(ContextEvidence.PROVIDER_ADVERTISED, 8192),
+        _obs(ContextEvidence.OPERATOR_DECLARED, 32768),
+    ])
+    assert (resolution.effective, resolution.evidence) == (8192, ContextEvidence.PROVIDER_ADVERTISED)
+    assert resolution.mismatch
+    assert {c.evidence for c in (resolution.conflicts[0].first, resolution.conflicts[0].second)} == {
+        ContextEvidence.PROVIDER_ADVERTISED, ContextEvidence.OPERATOR_DECLARED,
+    }
+
+
+def test_provider_advertised_beats_known_table_and_disagreement_is_visible():
+    resolution = combine_observations([
+        _obs(ContextEvidence.PROVIDER_ADVERTISED, 8192),
+        _obs(ContextEvidence.KNOWN_TABLE, 131072),
+    ])
+    # The legacy probe takes max(api, table) for cloud endpoints. A static
+    # table is weaker evidence and must not override the provider silently.
+    assert (resolution.effective, resolution.evidence) == (8192, ContextEvidence.PROVIDER_ADVERTISED)
+    assert resolution.mismatch
+
+
+def test_runtime_confirmed_beats_provider_advertised_and_records_mismatch():
+    resolution = combine_observations([
+        _obs(ContextEvidence.PROVIDER_ADVERTISED, 32768),
+        _obs(ContextEvidence.RUNTIME_CONFIRMED, 16384),
+    ])
+    assert (resolution.effective, resolution.evidence) == (16384, ContextEvidence.RUNTIME_CONFIRMED)
+    assert resolution.mismatch
+
+
+def test_no_evidence_is_unknown_zero_not_a_default():
+    resolution = combine_observations([])
+    assert resolution.effective == 0
+    assert resolution.evidence is ContextEvidence.UNKNOWN
+    assert resolution.budget_limit == 0
+    assert context_metrics(resolution, 500)["context_length"] == 0
+
+
+def test_runtime_limit_observation_is_pure_and_lowers_effective_window():
+    base = combine_observations([_obs(ContextEvidence.PROVIDER_ADVERTISED, 8192)])
+    updated = base.observe_runtime_limit(4096)
+    assert (updated.effective, updated.evidence, updated.source) == (
+        4096, ContextEvidence.RUNTIME_CONFIRMED, "provider_rejection",
+    )
+    assert updated.mismatch
+    assert base.observe_runtime_limit(None) is base
+    assert base.observe_runtime_limit(0) is base
+
+
+def test_context_metrics_reports_percent_against_stored_window():
+    resolution = combine_observations([_obs(ContextEvidence.KNOWN_TABLE, 8000)])
+    metrics = context_metrics(resolution, 2000)
+    assert metrics["context_length"] == 8000
+    assert metrics["context_percent"] == 25.0
+    assert metrics["context_resolution"]["evidence"] == "known_table"
+    assert metrics["context_resolution"]["mismatch"] is False
+
+
+# ---------------------------------------------------------------------------
+# Provider probe
+# ---------------------------------------------------------------------------
+
+@pytest.mark.asyncio
+async def test_provider_advertised_window_uses_turn_credentials(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _catalog("acme-model", max_model_len=8192)}))
+    resolution = await resolve_effective_context(REMOTE, "acme-model", headers=AUTH)
+    assert (resolution.effective, resolution.evidence, resolution.source) == (
+        8192, ContextEvidence.PROVIDER_ADVERTISED, "models_catalog",
+    )
+    assert resolution.provider_io and not resolution.cached
+    [(kind, url, headers)] = network.calls
+    assert kind == "get" and url.endswith("/models")
+    assert headers == {"Authorization": "Bearer secret-token"}
+
+
+@pytest.mark.asyncio
+async def test_credentials_are_not_forwarded_to_another_origin(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _catalog("acme-model", max_model_len=8192)}))
+    monkeypatch.setattr(
+        "src.endpoint_resolver.build_models_url", lambda base: "http://catalog.elsewhere.test/v1/models",
+    )
+    await resolve_effective_context(REMOTE, "acme-model", headers=AUTH)
+    assert network.calls[0][2] == {}
+
+
+@pytest.mark.asyncio
+async def test_credentials_follow_the_resolved_provider_host(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _catalog("acme-model", max_model_len=8192)}))
+    monkeypatch.setattr(
+        "src.endpoint_resolver.build_models_url", lambda base: "http://100.64.0.9/v1/models",
+    )
+    monkeypatch.setattr(
+        "src.endpoint_resolver.resolve_url", lambda url: url.replace("provider.test", "100.64.0.9"),
+    )
+    await resolve_effective_context(REMOTE, "acme-model", headers=AUTH)
+    assert network.calls[0][2] == {"Authorization": "Bearer secret-token"}
+
+
+@pytest.mark.asyncio
+async def test_slow_url_resolution_is_bounded_by_the_probe_deadline(monkeypatch):
+    import time as _time
+
+    _install(monkeypatch, _FakeNetwork())
+
+    def slow_models_url(base):
+        _time.sleep(0.5)
+        return base.split("/v1")[0] + "/v1/models"
+
+    monkeypatch.setattr("src.endpoint_resolver.build_models_url", slow_models_url)
+    loop = asyncio.get_running_loop()
+    started = loop.time()
+    resolution = await resolve_effective_context(REMOTE, "gpt-4o", deadline_seconds=0.05)
+    assert loop.time() - started < 0.4
+    assert resolution.probe_errors == ("deadline_exceeded",)
+    assert resolution.evidence is ContextEvidence.KNOWN_TABLE
+
+
+@pytest.mark.asyncio
+async def test_known_table_fallback_when_provider_lists_no_window(monkeypatch):
+    _install(monkeypatch, _FakeNetwork({"/models": _catalog("gpt-4o-mini")}))
+    resolution = await resolve_effective_context(REMOTE, "gpt-4o-mini", headers=AUTH)
+    assert (resolution.effective, resolution.evidence) == (128000, ContextEvidence.KNOWN_TABLE)
+    assert "models:no_window_listed" in resolution.probe_errors
+
+
+@pytest.mark.asyncio
+async def test_unavailable_metadata_and_unknown_model_is_unknown(monkeypatch):
+    _install(monkeypatch, _FakeNetwork({"/models": _Response(503, None)}))
+    resolution = await resolve_effective_context(REMOTE, "mystery-model", headers=AUTH)
+    assert resolution.evidence is ContextEvidence.UNKNOWN
+    assert resolution.effective == 0
+    assert resolution.probe_errors == ("models:http_503",)
+
+
+@pytest.mark.asyncio
+async def test_rejected_credentials_are_reported_without_leaking_them(monkeypatch):
+    _install(monkeypatch, _FakeNetwork({"/models": _Response(401, None)}))
+    resolution = await resolve_effective_context(REMOTE, "gpt-4o", headers=AUTH)
+    assert resolution.evidence is ContextEvidence.KNOWN_TABLE
+    assert resolution.probe_errors == ("models:http_401",)
+    assert "secret-token" not in json.dumps(resolution.to_dict())
+
+
+@pytest.mark.asyncio
+async def test_provider_timeout_is_bounded_and_falls_back(monkeypatch):
+    _install(monkeypatch, _FakeNetwork(get_delay=5.0))
+    loop = asyncio.get_running_loop()
+    started = loop.time()
+    resolution = await resolve_effective_context(
+        REMOTE, "gpt-4o", headers=AUTH, deadline_seconds=0.05,
+    )
+    assert loop.time() - started < 1.0
+    assert "deadline_exceeded" in resolution.probe_errors
+    assert (resolution.effective, resolution.evidence) == (128000, ContextEvidence.KNOWN_TABLE)
+
+
+@pytest.mark.asyncio
+async def test_provider_transport_failure_never_raises(monkeypatch):
+    _install(monkeypatch, _FakeNetwork(get_error=httpx.ConnectError("refused")))
+    resolution = await resolve_effective_context(
+        REMOTE, "mystery-model", client_runtime_context={"model_context_window": 16384},
+    )
+    assert resolution.probe_errors == ("models:transport_error",)
+    assert (resolution.effective, resolution.evidence) == (16384, ContextEvidence.OPERATOR_DECLARED)
+
+
+@pytest.mark.asyncio
+async def test_local_runtime_confirmed_slots_and_provider_mismatch(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({
+        "/slots": _Response(200, [{"n_ctx": 16384}]),
+        "/models": _catalog("local-model", max_model_len=32768),
+    }))
+    resolution = await resolve_effective_context(LOCAL, "local-model")
+    assert (resolution.effective, resolution.evidence, resolution.source) == (
+        16384, ContextEvidence.RUNTIME_CONFIRMED, "llamacpp_slots",
+    )
+    assert resolution.mismatch
+    assert [url.rsplit("/", 1)[-1] for _, url, _ in network.calls] == ["slots", "models"]
+
+
+@pytest.mark.asyncio
+async def test_local_runtime_confirmed_props_when_slots_disabled(monkeypatch):
+    _install(monkeypatch, _FakeNetwork({
+        "/slots": _Response(501, None),
+        "/props": _Response(200, {"default_generation_settings": {"n_ctx": 8192}}),
+        "/models": _catalog("local-model"),
+    }))
+    resolution = await resolve_effective_context(LOCAL, "local-model")
+    assert (resolution.effective, resolution.source) == (8192, "llamacpp_props")
+    assert not resolution.mismatch
+
+
+@pytest.mark.asyncio
+async def test_remote_resolution_is_cached_per_credentials(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _catalog("acme-model", max_model_len=8192)}))
+    first = await resolve_effective_context(REMOTE, "acme-model", headers=AUTH)
+    second = await resolve_effective_context(REMOTE, "acme-model", headers=AUTH)
+    assert first.effective == second.effective == 8192
+    assert second.cached and not second.provider_io
+    assert len(network.calls) == 1
+    await resolve_effective_context(REMOTE, "acme-model", headers={"Authorization": "Bearer other"})
+    assert len(network.calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_failed_remote_probe_expires_sooner(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _Response(503, None)}))
+    now = [1000.0]
+    await resolve_effective_context(REMOTE, "acme-model", clock=lambda: now[0])
+    await resolve_effective_context(REMOTE, "acme-model", clock=lambda: now[0])
+    assert len(network.calls) == 1
+    now[0] += cr.PROBE_FAILURE_TTL_SECONDS + 1
+    await resolve_effective_context(REMOTE, "acme-model", clock=lambda: now[0])
+    assert len(network.calls) == 2
+
+
+@pytest.mark.asyncio
+async def test_local_resolution_is_not_cached(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/slots": _Response(200, [{"n_ctx": 4096}])}))
+    await resolve_effective_context(LOCAL, "local-model")
+    await resolve_effective_context(LOCAL, "local-model")
+    assert sum(1 for kind, url, _ in network.calls if url.endswith("/slots")) == 2
+
+
+# ---------------------------------------------------------------------------
+# Compact runtime integration
+# ---------------------------------------------------------------------------
+
+async def _run_preview(messages=None, **kwargs):
+    contract = resolve_full_inventory_contract(schemas=[], policy=ToolPolicy())
+    raw = [chunk async for chunk in stream_preview(
+        endpoint_url=kwargs.pop("endpoint_url", REMOTE), model=kwargs.pop("model", "acme-model"),
+        messages=messages or [{"role": "user", "content": "hi"}],
+        headers=kwargs.pop("headers", AUTH),
+        turn_contract=contract, session_id="test", owner="test",
+        disabled_tools=set(), tool_policy=ToolPolicy(), **kwargs,
+    )]
+    assert raw[-1] == "data: [DONE]\n\n"
+    events = [json.loads(chunk[6:]) for chunk in raw if chunk.startswith("data: {")]
+    return next(event["data"] for event in events if event.get("type") == "metrics")
+
+
+@pytest.mark.asyncio
+async def test_compact_turn_resolves_once_before_model_and_metrics_do_no_discovery(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _catalog("acme-model", max_model_len=8192)}))
+    metrics = await _run_preview()
+    kinds = [kind for kind, _, _ in network.calls]
+    first_stream = kinds.index("stream")
+    # All discovery precedes the first model request; nothing after it,
+    # in particular nothing between the last model byte and [DONE].
+    assert kinds[:first_stream] == ["get"]
+    assert kinds[first_stream:] == ["stream"]
+    assert metrics["context_length"] == 8192
+    assert metrics["context_percent"] == 25.0
+    assert metrics["context_resolution"]["evidence"] == "provider_advertised"
+    assert metrics["context_resolution"]["provider_io"] is True
+
+
+@pytest.mark.asyncio
+async def test_compact_turn_with_supplied_resolution_performs_no_metadata_io(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork())
+    supplied = combine_observations([_obs(ContextEvidence.OPERATOR_DECLARED, 4096, "client_runtime_context")])
+    metrics = await _run_preview(context_resolution=supplied)
+    assert [kind for kind, _, _ in network.calls] == ["stream"]
+    assert metrics["context_length"] == 4096
+    assert metrics["context_resolution"] == supplied.to_dict()
+
+
+@pytest.mark.asyncio
+async def test_compact_metrics_report_unknown_window_as_zero(monkeypatch):
+    _install(monkeypatch, _FakeNetwork({"/models": _Response(404, None)}))
+    metrics = await _run_preview(model="mystery-model")
+    assert metrics["context_length"] == 0
+    assert metrics["context_percent"] == 0
+    assert metrics["context_resolution"]["evidence"] == "unknown"
+    assert metrics["context_resolution"]["probe_errors"] == ["models:http_404"]
+
+
+@pytest.mark.asyncio
+async def test_compact_runtime_budgets_against_resolved_window(monkeypatch):
+    turns = []
+    for index in range(6):
+        turns.append({"role": "user", "content": f"question {index} " + "word " * 280})
+        turns.append({"role": "assistant", "content": f"answer {index} " + "word " * 280})
+    session = SimpleNamespace(history=turns)
+
+    unbudgeted = _install(monkeypatch, _FakeNetwork())
+    await _run_preview(history_session=session, context_resolution=UNRESOLVED_CONTEXT)
+    budgeted = _install(monkeypatch, _FakeNetwork())
+    resolved = combine_observations([_obs(ContextEvidence.PROVIDER_ADVERTISED, 4096, "models_catalog")])
+    await _run_preview(history_session=session, context_resolution=resolved)
+
+    full = unbudgeted.requests[0]["messages"]
+    trimmed = budgeted.requests[0]["messages"]
+    # An unknown window keeps the previous reactive-only behavior.
+    assert len(full) > len(trimmed)
+    assert model_context.estimate_tokens(trimmed) < 4096
+    assert trimmed[-1]["content"] == "hi"
+
+
+@pytest.mark.asyncio
+async def test_compact_metrics_fold_provider_stated_limit_without_new_discovery(monkeypatch):
+    rejection = (
+        "This model's maximum context length is 4096 tokens. However, you requested "
+        "5000 tokens (4800 in the messages, 200 in the completion)."
+    )
+    network = _install(monkeypatch, _FakeNetwork(
+        {"/models": _catalog("acme-model", max_model_len=8192)},
+        stream_responses=[
+            _ModelStream([], status=400, text=json.dumps({"error": {"message": rejection}})),
+            _ModelStream(_answer_lines({"prompt_tokens": 1024, "completion_tokens": 3})),
+        ],
+    ))
+    metrics = await _run_preview()
+    assert [kind for kind, _, _ in network.calls] == ["get", "stream", "stream"]
+    resolution = metrics["context_resolution"]
+    assert metrics["context_length"] == 4096
+    assert (resolution["evidence"], resolution["source"]) == ("runtime_confirmed", "provider_rejection")
+    assert resolution["mismatch"] is True
+
+
+@pytest.mark.asyncio
+async def test_compact_turn_proceeds_when_metadata_probe_times_out(monkeypatch):
+    _install(monkeypatch, _FakeNetwork(get_error=httpx.ReadTimeout("slow catalog")))
+    metrics = await _run_preview(
+        model="gpt-4o", client_runtime_context={"model_context_window": 65536},
+    )
+    resolution = metrics["context_resolution"]
+    assert resolution["probe_errors"] == ["models:timeout"]
+    assert metrics["context_length"] == 65536
+    assert resolution["evidence"] == "operator_declared"
+
+
+# ---------------------------------------------------------------------------
+# Credential scoping (adversarial)
+# ---------------------------------------------------------------------------
+
+def _models_at(monkeypatch, models_url, resolved=None):
+    monkeypatch.setattr("src.endpoint_resolver.build_models_url", lambda base: models_url)
+    monkeypatch.setattr(
+        "src.endpoint_resolver.resolve_url", lambda url: url if resolved is None else resolved,
+    )
+
+
+async def _forwarded_headers(monkeypatch, endpoint_url, models_url, resolved=None):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _catalog("acme-model", max_model_len=8192)}))
+    _models_at(monkeypatch, models_url, resolved)
+    await resolve_effective_context(endpoint_url, "acme-model", headers=AUTH)
+    [(kind, url, headers)] = network.calls
+    assert url == models_url
+    return headers
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("models_url", [
+    "https://provider.test/v1/models",
+    "https://PROVIDER.test:443/v1/models",
+])
+async def test_credentials_reach_the_configured_provider_origin(monkeypatch, models_url):
+    headers = await _forwarded_headers(
+        monkeypatch, "https://provider.test/v1/chat/completions", models_url,
+    )
+    assert headers == {"Authorization": "Bearer secret-token"}
+
+
+@pytest.mark.asyncio
+async def test_credentials_reach_only_the_server_resolved_form_of_the_provider(monkeypatch):
+    endpoint = "http://gpu-box:8000/v1/chat/completions"
+    resolved = "http://100.64.0.9:8000/v1/chat/completions"
+    assert await _forwarded_headers(
+        monkeypatch, endpoint, "http://100.64.0.9:8000/v1/models", resolved,
+    ) == {"Authorization": "Bearer secret-token"}
+    cr.clear_probe_cache()
+    # The same address is not trusted when the server resolver did not
+    # produce it for this endpoint.
+    assert await _forwarded_headers(
+        monkeypatch, endpoint, "http://100.64.0.9:8000/v1/models",
+    ) == {}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("models_url", [
+    "http://provider.test/v1/models",               # scheme downgrade
+    "https://provider.test:8443/v1/models",         # other port
+    "https://provider.test.evil.example/v1/models", # lookalike host
+    "https://evil.example/provider.test/v1/models", # host in path
+    "https://provider.test@evil.example/v1/models", # host in userinfo
+    "/v1/models",                                   # no origin at all
+])
+async def test_unrelated_models_url_receives_no_credentials(monkeypatch, models_url):
+    assert await _forwarded_headers(
+        monkeypatch, "https://provider.test/v1/chat/completions", models_url,
+    ) == {}
+
+
+@pytest.mark.asyncio
+async def test_probe_client_never_follows_redirects(monkeypatch):
+    network = _install(monkeypatch, _FakeNetwork({"/models": _Response(302, None)}))
+    resolution = await resolve_effective_context(REMOTE, "acme-model", headers=AUTH)
+    assert network.client_kwargs and all(
+        kwargs.get("follow_redirects") is False for kwargs in network.client_kwargs
+    )
+    assert resolution.probe_errors == ("models:http_302",)
+    assert len(network.calls) == 1
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("failure", [
+    _FakeNetwork(get_error=httpx.ConnectError(
+        "connect to https://user:pw-secret@provider.test/v1/models?api_key=query-secret failed")),
+    _FakeNetwork(get_error=httpx.ReadTimeout("Bearer secret-token timed out")),
+    _FakeNetwork({"/models": _Response(401, None)}),
+    _FakeNetwork({"/models": _Response(200, ValueError("api_key=query-secret"))}),
+    _FakeNetwork(get_error=RuntimeError("Authorization: Bearer secret-token")),
+])
+async def test_probe_errors_never_expose_credentials_or_urls(monkeypatch, caplog, failure):
+    import logging
+
+    caplog.set_level(logging.DEBUG)
+    endpoint = "https://user:pw-secret@provider.test/v1/chat/completions?api_key=query-secret"
+    _install(monkeypatch, failure)
+    resolution = await resolve_effective_context(endpoint, "acme-model", headers=AUTH)
+    assert resolution.probe_errors
+    exposed = json.dumps(resolution.to_dict()) + caplog.text + json.dumps(
+        context_metrics(resolution, 10),
+    )
+    for secret in ("secret-token", "pw-secret", "query-secret", "provider.test", "Authorization"):
+        assert secret not in exposed
+
+
+@pytest.mark.asyncio
+async def test_bound_endpoint_url_stays_out_of_repr_and_metrics(monkeypatch):
+    _install(monkeypatch, _FakeNetwork({"/models": _Response(503, None)}))
+    endpoint = "https://user:pw-secret@provider.test/v1/chat/completions?api_key=query-secret"
+    resolution = await resolve_effective_context(endpoint, "acme-model", headers=AUTH)
+    assert resolution.applies_to(endpoint, "acme-model")
+    for rendered in (repr(resolution), str(resolution), json.dumps(resolution.to_dict())):
+        assert "pw-secret" not in rendered and "query-secret" not in rendered

tests/test_context_resolution_route.py

@@ -0,0 +1,534 @@
+"""One typed context resolution per compact chat turn, end to end.
+
+These tests keep the conftest offline guard active on purpose: the real route,
+agent loop and compact runtime run, only the resolver's HTTP/DNS edges are
+offline, and ``context_probe_ledger`` records every metadata request.
+"""
+from dataclasses import replace
+import json
+
+import pytest
+
+import src.model_context as model_context
+from src.agent_runtime import context_resolution as cr
+from src.agent_runtime.context_resolution import (
+    ContextEvidence,
+    ContextObservation,
+    combine_observations,
+)
+from tests.test_foreground_model_routing import _RouteRequest, _chat_stream_endpoint
+
+COMPACT_MODEL = "odysseus-qwen3.5-tools-pre-heretic"
+
+
+class _ModelStream:
+    def __init__(self, lines, status=200, text=""):
+        self.status_code = status
+        self.text = text
+        self._lines = lines
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *args):
+        return False
+
+    async def aread(self):
+        return self.text.encode()
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise AssertionError(f"unexpected provider status {self.status_code}")
+
+    async def aiter_lines(self):
+        for line in self._lines:
+            yield line
+
+
+def _answer(prompt_tokens=1024):
+    return _ModelStream([
+        "data: " + json.dumps({"choices": [{"delta": {"content": "Hello."}}]}),
+        "data: " + json.dumps({"choices": [{"delta": {}}],
+                               "usage": {"prompt_tokens": prompt_tokens, "completion_tokens": 2}}),
+        "data: [DONE]",
+    ])
+
+
+def _install_model(monkeypatch, *responses):
+    """Fake provider for chat completions only; metadata goes to the ledger."""
+    import src.clean_agent_preview as preview
+
+    queue = list(responses)
+    sent = []
+
+    class Client:
+        def __init__(self, **kwargs):
+            pass
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            return False
+
+        def stream(self, method, url, headers=None, json=None):
+            sent.append(json)
+            return queue.pop(0) if queue else _answer()
+
+    monkeypatch.setattr(preview.httpx, "AsyncClient", Client)
+    return sent
+
+
+def _forbid_legacy_probe(monkeypatch):
+    calls = []
+
+    def legacy(*args, **kwargs):
+        calls.append(args)
+        raise AssertionError("legacy context probe used on a compact turn")
+
+    monkeypatch.setattr(model_context, "_query_context_length", legacy)
+    model_context._context_cache.clear()
+    return calls
+
+
+def _spy(monkeypatch):
+    """Count resolutions and capture what the compact runtime received/emitted."""
+    import src.clean_agent_preview as preview
+
+    seen = {"resolutions": [], "preview_kwargs": [], "preview_chunks": []}
+    real_resolve = cr.resolve_effective_context
+
+    async def counting_resolve(*args, **kwargs):
+        result = await real_resolve(*args, **kwargs)
+        seen["resolutions"].append(result)
+        return result
+
+    real_preview = preview.stream_preview
+
+    async def recording_preview(**kwargs):
+        seen["preview_kwargs"].append(kwargs)
+        async for chunk in real_preview(**kwargs):
+            seen["preview_chunks"].append(chunk)
+            yield chunk
+
+    monkeypatch.setattr(cr, "resolve_effective_context", counting_resolve)
+    monkeypatch.setattr(preview, "stream_preview", recording_preview)
+    return seen
+
+
+def _metrics(chunks):
+    for chunk in chunks:
+        if chunk.startswith("data: {"):
+            event = json.loads(chunk[6:])
+            if event.get("type") == "metrics":
+                return event["data"]
+    raise AssertionError("no metrics event")
+
+
+async def _drive_route(monkeypatch, *, model=COMPACT_MODEL, message="hello"):
+    from routes import chat_routes
+    import src.agent_loop as agent_loop
+
+    captured = {}
+    endpoint = _chat_stream_endpoint(
+        monkeypatch, "agent", captured, capture_context=True, session_model=model,
+    )
+    monkeypatch.setattr(
+        chat_routes, "coerce_message_and_session", lambda *args, **kwargs: (message, "session-1"),
+    )
+    # The real agent loop, compact dispatch and compact runtime run below.
+    monkeypatch.setattr(chat_routes, "stream_agent_loop", agent_loop.stream_agent_loop)
+    request = _RouteRequest("agent")
+    request._form.update({"message": message, "compare_mode": "false"})
+    response = await endpoint(request)
+    body = [chunk async for chunk in response.body_iterator]
+    return captured, body
+
+
+@pytest.mark.asyncio
+async def test_compact_chat_route_resolves_once_and_reuses_the_exact_object(
+    monkeypatch, context_probe_ledger,
+):
+    legacy = _forbid_legacy_probe(monkeypatch)
+    seen = _spy(monkeypatch)
+    sent = _install_model(monkeypatch)
+
+    captured, _ = await _drive_route(monkeypatch)
+
+    # One typed resolution, one metadata request, no legacy lookup.
+    [resolution] = seen["resolutions"]
+    assert len(context_probe_ledger) == 1
+    assert context_probe_ledger[0]["url"] == "https://selected.example/v1/models"
+    # The route used the session's provider credentials.
+    assert context_probe_ledger[0]["headers"] == {"Authorization": "Bearer selected"}
+    assert legacy == []
+    assert sent, "the compact runtime never reached the model"
+
+    # The exact object crosses route -> build_chat_context and
+    # route -> stream_agent_loop -> stream_preview.
+    assert captured["build_context"]["context_resolution"] is resolution
+    [preview_kwargs] = seen["preview_kwargs"]
+    assert preview_kwargs["context_resolution"] is resolution
+    assert resolution.applies_to("https://selected.example/v1", COMPACT_MODEL)
+
+    # Metrics report that same resolution; the offline probe failed, so the
+    # trusted table supplies the window.
+    metrics = _metrics(seen["preview_chunks"])
+    assert metrics["context_resolution"] == resolution.to_dict()
+    assert metrics["context_length"] == 131072
+    assert metrics["context_resolution"]["evidence"] == "known_table"
+    assert seen["preview_chunks"][-1] == "data: [DONE]\n\n"
+
+
+@pytest.mark.asyncio
+async def test_compact_route_metrics_fold_a_provider_limit_without_a_second_probe(
+    monkeypatch, context_probe_ledger,
+):
+    _forbid_legacy_probe(monkeypatch)
+    seen = _spy(monkeypatch)
+    rejection = (
+        "This model's maximum context length is 4096 tokens. However, you requested "
+        "5000 tokens (4800 in the messages, 200 in the completion)."
+    )
+    _install_model(
+        monkeypatch,
+        _ModelStream([], status=400, text=json.dumps({"error": {"message": rejection}})),
+        _answer(),
+    )
+
+    await _drive_route(monkeypatch)
+
+    [resolution] = seen["resolutions"]
+    assert len(context_probe_ledger) == 1
+    metrics = _metrics(seen["preview_chunks"])
+    reported = metrics["context_resolution"]
+    assert metrics["context_length"] == 4096
+    assert (reported["evidence"], reported["source"]) == ("runtime_confirmed", "provider_rejection")
+    # Everything else is the prepared resolution, unchanged.
+    assert reported == resolution.observe_runtime_limit(4096).to_dict()
+    assert reported["observations"][:-1] == resolution.to_dict()["observations"]
+
+
+@pytest.mark.asyncio
+async def test_regular_model_route_does_not_prepare_a_compact_resolution(
+    monkeypatch, context_probe_ledger,
+):
+    from routes import chat_routes
+
+    seen = _spy(monkeypatch)
+    captured = {}
+    endpoint = _chat_stream_endpoint(monkeypatch, "agent", captured, capture_context=True)
+
+    async def capture_agent(*args, **kwargs):
+        captured["agent_kwargs"] = kwargs
+        yield "data: [DONE]\n\n"
+
+    monkeypatch.setattr(chat_routes, "stream_agent_loop", capture_agent)
+    request = _RouteRequest("agent")
+    request._form.update({"message": "hello", "compare_mode": "false"})
+    response = await endpoint(request)
+    async for _ in response.body_iterator:
+        pass
+
+    assert seen["resolutions"] == []
+    assert context_probe_ledger == []
+    assert captured["build_context"]["context_resolution"] is None
+    assert captured["agent_kwargs"].get("context_resolution") is None
+
+
+@pytest.mark.asyncio
+async def test_bare_context_length_never_becomes_typed_provenance(monkeypatch, context_probe_ledger):
+    """A legacy integer handed to the agent loop is not evidence of anything."""
+    import src.agent_loop as agent_loop
+    from src.tool_policy import ToolPolicy
+    from src.turn_contract import resolve_full_inventory_contract
+    from src.clean_agent_preview import MODE
+
+    seen = _spy(monkeypatch)
+    _install_model(monkeypatch)
+    contract = replace(
+        resolve_full_inventory_contract(schemas=[], policy=ToolPolicy()), selection_mode=MODE,
+    )
+    chunks = [chunk async for chunk in agent_loop.stream_agent_loop(
+        "https://selected.example/v1", "mystery-model", [{"role": "user", "content": "hello"}],
+        headers={"Authorization": "Bearer selected"}, context_length=4096,
+        turn_contract=contract, session_id="s", owner="alice",
+        tool_policy=ToolPolicy(), disabled_tools=set(),
+    )]
+
+    # A direct caller without a prepared resolution still resolves safely, once.
+    [resolution] = seen["resolutions"]
+    assert len(context_probe_ledger) == 1
+    metrics = _metrics(chunks)
+    assert metrics["context_length"] == 0
+    assert metrics["context_resolution"]["evidence"] == "unknown"
+    assert all(
+        observation["value"] != 4096
+        for observation in metrics["context_resolution"]["observations"]
+    )
+    assert resolution.to_dict() == metrics["context_resolution"]
+
+
+@pytest.mark.asyncio
+async def test_stream_preview_rejects_a_resolution_prepared_for_another_route(
+    monkeypatch, context_probe_ledger,
+):
+    import src.clean_agent_preview as preview
+    from src.tool_policy import ToolPolicy
+    from src.turn_contract import resolve_full_inventory_contract
+
+    _install_model(monkeypatch)
+    foreign = combine_observations([
+        ContextObservation(ContextEvidence.PROVIDER_ADVERTISED, 2048, "models_catalog"),
+    ])
+    foreign = replace(foreign, endpoint_url="https://other.example/v1", model="other-model")
+    chunks = [chunk async for chunk in preview.stream_preview(
+        endpoint_url="https://selected.example/v1", model="gpt-4o",
+        messages=[{"role": "user", "content": "hello"}], headers={},
+        turn_contract=resolve_full_inventory_contract(schemas=[], policy=ToolPolicy()),
+        session_id="s", owner="alice", disabled_tools=set(), tool_policy=ToolPolicy(),
+        context_resolution=foreign,
+    )]
+    metrics = _metrics(chunks)
+    assert len(context_probe_ledger) == 1
+    assert metrics["context_length"] == 128000
+    assert metrics["context_resolution"]["evidence"] == "known_table"
+
+
+@pytest.mark.asyncio
+async def test_supplied_resolution_means_no_probe_in_stream_preview(monkeypatch, context_probe_ledger):
+    import src.clean_agent_preview as preview
+    from src.tool_policy import ToolPolicy
+    from src.turn_contract import resolve_full_inventory_contract
+
+    _install_model(monkeypatch)
+    prepared = await cr.resolve_effective_context(
+        "https://selected.example/v1", "gpt-4o", headers={"Authorization": "Bearer selected"},
+    )
+    assert len(context_probe_ledger) == 1
+    chunks = [chunk async for chunk in preview.stream_preview(
+        endpoint_url="https://selected.example/v1", model="gpt-4o",
+        messages=[{"role": "user", "content": "hello"}], headers={"Authorization": "Bearer selected"},
+        turn_contract=resolve_full_inventory_contract(schemas=[], policy=ToolPolicy()),
+        session_id="s", owner="alice", disabled_tools=set(), tool_policy=ToolPolicy(),
+        context_resolution=prepared,
+    )]
+    # Neither preparation nor terminal metrics probed again.
+    assert len(context_probe_ledger) == 1
+    assert _metrics(chunks)["context_resolution"] == prepared.to_dict()
+    assert chunks[-1] == "data: [DONE]\n\n"
+
+
+# ---------------------------------------------------------------------------
+# build_chat_context consumes the prepared resolution
+# ---------------------------------------------------------------------------
+
+def _context_harness(monkeypatch):
+    from tests.test_kv_cache_invalidation_2927 import _build_context_harness, _install_chat_helpers_stubs
+
+    chat_helpers = _install_chat_helpers_stubs(monkeypatch)
+    sess, request, chat_handler, chat_processor = _build_context_harness(monkeypatch, chat_helpers, history=[])
+    legacy = []
+    monkeypatch.setattr(
+        chat_helpers, "get_context_length",
+        lambda *args: legacy.append(args) or 8192,
+    )
+    compactions = []
+
+    async def recording_maybe_compact(sess, endpoint_url, model, messages, headers, owner=None, **kwargs):
+        compactions.append(kwargs)
+        return messages, kwargs.get("context_length", 8192), False
+
+    monkeypatch.setattr(chat_helpers, "maybe_compact", recording_maybe_compact)
+    return chat_helpers, (sess, request, chat_handler, chat_processor), legacy, compactions
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("defer", [False, True])
+async def test_build_chat_context_shapes_with_the_prepared_resolution(monkeypatch, defer):
+    chat_helpers, (sess, request, handler, processor), legacy, compactions = _context_harness(monkeypatch)
+    prepared = combine_observations([
+        ContextObservation(ContextEvidence.PROVIDER_ADVERTISED, 32768, "models_catalog"),
+    ])
+    ctx = await chat_helpers.build_chat_context(
+        sess=sess, request=request, chat_handler=handler, chat_processor=processor,
+        message="hello", session_id="s", defer_context_shaping=defer,
+        context_resolution=prepared,
+    )
+    assert legacy == []
+    assert ctx.context_length == 32768
+    assert compactions == ([] if defer else [{"context_length": 32768}])
+
+
+@pytest.mark.asyncio
+async def test_build_chat_context_unknown_resolution_shapes_with_legacy_default(monkeypatch):
+    chat_helpers, (sess, request, handler, processor), legacy, compactions = _context_harness(monkeypatch)
+    ctx = await chat_helpers.build_chat_context(
+        sess=sess, request=request, chat_handler=handler, chat_processor=processor,
+        message="hello", session_id="s", context_resolution=combine_observations([]),
+    )
+    # Shaping still needs a number, but no probe and no provenance is created.
+    assert legacy == []
+    assert ctx.context_length == model_context.DEFAULT_CONTEXT
+    assert compactions == [{"context_length": model_context.DEFAULT_CONTEXT}]
+
+
+@pytest.mark.asyncio
+async def test_build_chat_context_without_resolution_keeps_legacy_lookup(monkeypatch):
+    chat_helpers, (sess, request, handler, processor), legacy, compactions = _context_harness(monkeypatch)
+    await chat_helpers.build_chat_context(
+        sess=sess, request=request, chat_handler=handler, chat_processor=processor,
+        message="hello", session_id="s", defer_context_shaping=True,
+    )
+    assert len(legacy) == 1
+    assert compactions == []
+
+
+@pytest.mark.asyncio
+async def test_offline_guard_replaces_only_io_edges(context_probe_ledger):
+    """The conftest guard must not mask the resolver itself."""
+    headers = {"Authorization": "Bearer selected", "Content-Type": "application/json"}
+    first = await cr.resolve_effective_context("https://selected.example/v1", "gpt-4o", headers=headers)
+    second = await cr.resolve_effective_context("https://selected.example/v1", "gpt-4o", headers=headers)
+    # Real request construction and credential scoping ran...
+    assert context_probe_ledger == [{
+        "url": "https://selected.example/v1/models",
+        "headers": {"Authorization": "Bearer selected"},
+    }]
+    # ...as did real error mapping, evidence selection and caching.
+    assert first.probe_errors == ("models:transport_error",)
+    assert (first.evidence, first.effective) == (ContextEvidence.KNOWN_TABLE, 128000)
+    assert second.cached and not second.provider_io
+
+
+# ---------------------------------------------------------------------------
+# Route preparation and agent-loop dispatch share one compact decision
+# ---------------------------------------------------------------------------
+
+class _RegularPath(Exception):
+    pass
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "case, model, mode, privileges, surface, configured, image_generation, expected, regular_loop",
+    [
+        ("compact_agent", COMPACT_MODEL, "agent", None, None, "", False, True, False),
+        ("compact_chat_escalates", COMPACT_MODEL, "chat", None, None, "", False, True, False),
+        ("regular_model", "selected-model", "agent", None, None, "", False, False, True),
+        ("configured_compact", "selected-model", "agent", None, None, "compact", False, True, False),
+        ("configured_full", COMPACT_MODEL, "agent", None, None, "full", False, False, True),
+        # Plain chat and image generation leave before the agent loop.
+        ("agent_privilege_denied", COMPACT_MODEL, "agent", {"can_use_agent": False}, None, "", False, False, False),
+        ("agent_privilege_granted", COMPACT_MODEL, "agent", {"can_use_agent": True}, None, "", False, True, False),
+        ("tui_surface", COMPACT_MODEL, "agent", None, "odysseus-tui", "", False, False, True),
+        ("image_generation", COMPACT_MODEL, "agent", None, None, "", True, False, False),
+    ],
+)
+async def test_route_preparation_and_compact_dispatch_cannot_diverge(
+    monkeypatch, context_probe_ledger,
+    case, model, mode, privileges, surface, configured, image_generation, expected, regular_loop,
+):
+    from routes import chat_routes
+    import src.agent_loop as agent_loop
+
+    seen = _spy(monkeypatch)
+    _install_model(monkeypatch)
+    captured = {}
+    endpoint = _chat_stream_endpoint(
+        monkeypatch, mode, captured, capture_context=True, session_model=model,
+    )
+    monkeypatch.setattr(
+        chat_routes, "coerce_message_and_session", lambda *args, **kwargs: ("hello", "session-1"),
+    )
+    monkeypatch.setattr(
+        chat_routes, "_configured_model_tool_surface", lambda *args, **kwargs: configured,
+    )
+    monkeypatch.setattr(
+        chat_routes, "_is_image_generation_session", lambda *args, **kwargs: image_generation,
+    )
+    # Real agent loop: the compact branch reaches the recorded stream_preview;
+    # the regular branch stops at its first step.
+    regular = []
+
+    def stop_regular(*args, **kwargs):
+        regular.append(True)
+        raise _RegularPath()
+
+    monkeypatch.setattr(agent_loop, "_contract_allows_single_action_terminal", stop_regular)
+    monkeypatch.setattr(chat_routes, "stream_agent_loop", agent_loop.stream_agent_loop)
+
+    request = _RouteRequest(mode, privileges=privileges)
+    request._form.update({"message": "hello", "compare_mode": "false"})
+    if surface:
+        request._form["client_runtime_context"] = json.dumps({"surface": surface})
+    response = await endpoint(request)
+    try:
+        async for _ in response.body_iterator:
+            pass
+    except _RegularPath:
+        pass
+
+    prepared = captured["build_context"].get("context_resolution")
+    dispatched_compact = bool(seen["preview_kwargs"])
+    assert (prepared is not None) == dispatched_compact == expected, case
+    if expected:
+        # One resolution, prepared by the route and reused by dispatch.
+        assert len(seen["resolutions"]) == 1 and len(context_probe_ledger) == 1
+        assert seen["preview_kwargs"][0]["context_resolution"] is prepared
+    else:
+        assert seen["resolutions"] == [] and context_probe_ledger == []
+    # The case really reached the dispatch point it claims to exercise.
+    assert bool(regular) == regular_loop, case
+
+
+def test_compact_selection_rule_and_contract_stamp():
+    from src.agent_runtime.runtime_selection import (
+        COMPACT_PREVIEW_MODE, is_compact_preview_contract, uses_compact_preview_runtime,
+    )
+    from src.clean_agent_preview import MODE
+    from routes.chat_routes import _turn_contract_enabled
+    from types import SimpleNamespace
+
+    facts = dict(
+        clean_route_requested=True, turn_contract_enabled=True,
+        agent_mode=True, agent_permitted=True, image_generation=False,
+    )
+    assert uses_compact_preview_runtime(**facts)
+    for name, value in (
+        ("clean_route_requested", False), ("turn_contract_enabled", False),
+        ("agent_mode", False), ("agent_permitted", False), ("image_generation", True),
+    ):
+        assert not uses_compact_preview_runtime(**{**facts, name: value}), name
+    # An exact tool approval opts the turn out through the contract policy.
+    assert not uses_compact_preview_runtime(**{**facts, "turn_contract_enabled": _turn_contract_enabled(
+        exact_tool_approval=object(), runtime_surface="", native_workspace_contract=False,
+        clean_v3_route=True,
+    )})
+    # The route stamps contracts with the same constant the loop checks.
+    assert MODE == COMPACT_PREVIEW_MODE
+    assert is_compact_preview_contract(SimpleNamespace(selection_mode=MODE))
+    assert not is_compact_preview_contract(SimpleNamespace(selection_mode="routed"))
+    assert not is_compact_preview_contract(None)
+
+
+def test_synthetic_request_without_app_scope_does_not_gain_agent_privilege():
+    """A synthetic Request without scope['app'] must not crash or gain privileges."""
+    from starlette.requests import Request
+    from routes.chat_routes import _request_privileges
+    from src.agent_runtime.runtime_selection import uses_compact_preview_runtime
+
+    req = Request({"type": "http", "method": "POST", "path": "/api/chat_stream", "headers": []})
+    privs = _request_privileges(req, "alice")
+    assert privs == {}
+    assert privs.get("can_use_agent") is None
+
+    # Plain chat without app/auth state must not qualify for compact agent preview
+    selected = uses_compact_preview_runtime(
+        clean_route_requested=True,
+        turn_contract_enabled=True,
+        agent_mode=False,
+        agent_permitted=privs.get("can_use_agent", True),
+        image_generation=False,
+    )
+    assert not selected