diff --git a/routes/chat_helpers.py b/routes/chat_helpers.py index 96ac103d8..eb5fcf489 100644 --- a/routes/chat_helpers.py +++ b/routes/chat_helpers.py @@ -1027,11 +1027,15 @@ async def build_chat_context( persist_user_message: bool = True, interaction_mode: str | None = None, auto_escalated: bool = False, + context_resolution=None, ) -> ChatContext: """Build the full context (preface + messages) for an LLM call. This is the shared logic between /chat and /chat_stream — preset extraction, message preprocessing, memory/RAG/web injection, compaction, normalization. + + ``context_resolution`` is the turn's already resolved context window. When + supplied, history shaping sizes against it instead of probing the endpoint. """ # Preset preset = extract_preset(chat_handler, preset_id) @@ -1225,12 +1229,20 @@ async def build_chat_context( # for every candidate. Running selected-model compaction here would mutate # session history before we know which route can answer and would make a # later larger-context candidate unable to recover discarded history. + if context_resolution is not None: + prepared_window = {"context_length": context_resolution.shaping_window} + else: + prepared_window = {} if defer_context_shaping: - context_length = get_context_length(sess.endpoint_url, sess.model) + context_length = ( + prepared_window.get("context_length") + or get_context_length(sess.endpoint_url, sess.model) + ) was_compacted = False else: messages, context_length, was_compacted = await maybe_compact( sess, sess.endpoint_url, sess.model, messages, sess.headers, owner=user, + **prepared_window, ) _before_trim_messages = len(messages) _before_trim_tokens = estimate_tokens(messages) diff --git a/routes/chat_routes.py b/routes/chat_routes.py index 09596ccbc..4ef072afa 100644 --- a/routes/chat_routes.py +++ b/routes/chat_routes.py @@ -87,6 +87,7 @@ from src.model_profiles import ( ) from src.tool_execution import AgentExecutionBridge, bind_execution_bridge from src.agent_runtime.authority import is_internal_tool_request, request_authority_for_http +from src.agent_runtime.runtime_selection import uses_compact_preview_runtime from src.turn_contract import ( FAMILY_TOOLS, bind_turn_contract, preserve_bound_editor_selected_tools, requested_capabilities, resolve_turn_contract, @@ -145,6 +146,19 @@ def _turn_contract_enabled(*, exact_tool_approval, runtime_surface, ) +def _request_privileges(request, user) -> Dict[str, Any]: + """Per-user privileges from the app's auth manager; empty when unmanaged.""" + try: + app = getattr(request, "app", None) + except (AttributeError, KeyError): + app = None + state = getattr(app, "state", None) if app is not None else None + auth_manager = getattr(state, "auth_manager", None) if state is not None else None + if not user or not auth_manager: + return {} + return auth_manager.get_privileges(user) or {} + + def _native_runtime_requires_local_browser(client_runtime_context): """Use the private browser to verify declared local HTML artifacts.""" context = client_runtime_context if isinstance(client_runtime_context, dict) else {} @@ -2893,6 +2907,38 @@ def setup_chat_routes( allowed_models=_allowed_models_for_request(request), ) + # Decide once whether this turn runs on the compact (clean v3) + # runtime. Every input is final here; the native workspace term of + # the contract policy cannot veto a requested clean route. This one + # value prepares the turn below and stamps its contract later, and + # the agent loop dispatches on that stamp. + _compact_preview_turn = uses_compact_preview_runtime( + clean_route_requested=_clean_v3_route_requested, + turn_contract_enabled=_turn_contract_enabled( + exact_tool_approval=exact_tool_approval, + runtime_surface=str((client_runtime_context or {}).get("surface") or ""), + native_workspace_contract=False, + clean_v3_route=_clean_v3_route_requested, + full_schema_route=(_effective_tool_schema_mode == "full"), + ), + agent_mode=(chat_mode == "agent"), + agent_permitted=_request_privileges( + request, effective_user(request), + ).get("can_use_agent", True), + image_generation=image_generation_session, + ) + # A compact turn resolves its typed context window once, here, with + # the session's provider credentials. History shaping below and the + # compact runtime both reuse this exact object, so the turn neither + # probes twice nor mixes the legacy untyped lookup into it. + _compact_context_resolution = None + if _compact_preview_turn: + from src.agent_runtime.context_resolution import resolve_effective_context + _compact_context_resolution = await resolve_effective_context( + sess.endpoint_url, sess.model, headers=sess.headers, + client_runtime_context=client_runtime_context, + ) + # Build shared context (stream path uses enhanced_message for context preface) ctx = await build_chat_context( sess, request, chat_handler, chat_processor, @@ -2923,6 +2969,7 @@ def setup_chat_routes( else None ), persist_user_message=not tool_approval_continuation and not is_internal_tool_request(request), + context_resolution=_compact_context_resolution, interaction_mode=chat_mode, auto_escalated=auto_escalated, ) @@ -3122,7 +3169,9 @@ def setup_chat_routes( # pasted revision request. This only offers permitted schemas; # it never requires or performs a document mutation. _turn_capabilities = _turn_capabilities | {"documents"} - _clean_v3_preview = bool(_use_turn_contract and _clean_v3_route_requested) + # Same decision that prepared the turn; it only stamps the contract + # inside the agent-contract branch below. + _clean_v3_preview = _compact_preview_turn # requested_capabilities already inherits a typed, recently executed # family for referential follow-ups. Do not additionally union stale # families into an explicit new request: that inflated regular-model @@ -3267,13 +3316,11 @@ def setup_chat_routes( }) # Enforce per-user privileges - _privs = {} # Bearer clients enter the agent loop as the sandboxed ``api`` user, # but their token is owned by the real account. Use that owner here so # a permitted TUI/WebUI client does not inherit api's default denial. _user = effective_user(request) - if _user and hasattr(request.app.state, 'auth_manager') and request.app.state.auth_manager: - _privs = request.app.state.auth_manager.get_privileges(_user) + _privs = _request_privileges(request, _user) if _privs: if not _privs.get("can_use_bash", True): disabled_tools.update(FAMILY_TOOLS["shell_files"]) @@ -4459,6 +4506,9 @@ def setup_chat_routes( client_runtime_context=client_runtime_context, thinking_mode=thinking_mode, reasoning_effort=reasoning_effort, + context_resolution=( + _compact_context_resolution if _clean_v3_preview else None + ), ): if chunk.startswith("data: ") and not chunk.startswith("data: [DONE]"): try: diff --git a/src/agent_loop.py b/src/agent_loop.py index 5de35898a..f9130a1e6 100644 --- a/src/agent_loop.py +++ b/src/agent_loop.py @@ -84,6 +84,7 @@ from src.tool_types import ToolBlock from src.turn_contract import selected_tools_for_request, with_turn_contract from src.agent_runtime.journal import propose_action, execute_action from src.agent_runtime.completion import with_completion_gate +from src.agent_runtime.runtime_selection import is_compact_preview_contract from src.teacher_escalation import with_teacher_takeover, request_teacher_takeover from src.tool_utils import _truncate, get_mcp_manager from src.agent_tools import ( @@ -20425,6 +20426,7 @@ async def stream_agent_loop( reasoning_effort: Optional[str] = None, _parent_run_id: Optional[str] = None, request_authority=MISSING_AUTHORITY, + context_resolution=None, ) -> AsyncGenerator[str, None]: """Streaming agent loop generator. @@ -20460,7 +20462,7 @@ async def stream_agent_loop( ), ) - if turn_contract is not None and turn_contract.selection_mode == 'clean_compact_v3_preview': + if is_compact_preview_contract(turn_contract): from src.clean_agent_preview import stream_preview async for chunk in stream_preview( endpoint_url=endpoint_url, model=model, messages=messages, headers=headers, @@ -20477,6 +20479,9 @@ async def stream_agent_loop( max_rounds=max_rounds, max_tool_calls=max_tool_calls, temperature=temperature, + # The route's prepared, typed context window; the compact runtime + # must not resolve it a second time. + context_resolution=context_resolution, ): yield chunk return diff --git a/src/agent_runtime/context_resolution.py b/src/agent_runtime/context_resolution.py new file mode 100644 index 000000000..8290f4837 --- /dev/null +++ b/src/agent_runtime/context_resolution.py @@ -0,0 +1,519 @@ +"""Effective model context window, resolved once per logical turn. + +A turn resolves the window it budgets against at preparation time, before any +model request, and keeps the value together with the evidence that chose it. +Terminal metrics report that stored resolution; they never start discovery. + +Evidence classes are kept apart instead of being folded into one "known" flag: + +* ``runtime_confirmed``: the serving process reported its active window + (llama.cpp ``/slots`` or ``/props``) or rejected a request of this turn with + an explicit limit. +* ``provider_advertised``: the provider's model catalog lists a window. +* ``operator_declared``: the client or operator declared a transport window. + It caps runtime or provider evidence and replaces weaker evidence. +* ``known_table``: the static ``KNOWN_CONTEXT_WINDOWS`` fallback. +* ``unknown``: nothing above is available. The value is 0, never a default. + +Any disagreement between sources is recorded as a conflict. An operator value +below a measured value is a cap, not a contradiction; an operator value above +it is a contradiction. + +Context sizing is not authority: nothing here grants or denies an operation. +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass, field, replace +from enum import Enum +import hashlib +import json +import logging +import time +from typing import Any, Mapping, Optional +from urllib.parse import urlparse + +import httpx + +logger = logging.getLogger(__name__) + +# Upper bound for all provider metadata I/O of one turn preparation. A slow +# or unreachable metadata endpoint costs at most this much before the turn +# proceeds with whatever evidence it has. +PROBE_DEADLINE_SECONDS = 3.0 +# Remote provider metadata changes rarely; failures are retried sooner so a +# transient outage does not pin a turn to weaker evidence for long. Local +# servers are always re-probed because they can restart with another window. +PROBE_CACHE_TTL_SECONDS = 600.0 +PROBE_FAILURE_TTL_SECONDS = 60.0 + +# Headers that describe the chat request body rather than the caller. +_REQUEST_ONLY_HEADERS = frozenset({"content-type", "content-length", "accept", "accept-encoding"}) + + +class ContextEvidence(str, Enum): + RUNTIME_CONFIRMED = "runtime_confirmed" + PROVIDER_ADVERTISED = "provider_advertised" + OPERATOR_DECLARED = "operator_declared" + KNOWN_TABLE = "known_table" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class ContextObservation: + evidence: ContextEvidence + value: int + source: str + + def to_dict(self) -> dict: + return {"evidence": self.evidence.value, "value": self.value, "source": self.source} + + +@dataclass(frozen=True) +class ContextConflict: + first: ContextObservation + second: ContextObservation + + def to_dict(self) -> dict: + return {"first": self.first.to_dict(), "second": self.second.to_dict()} + + +@dataclass(frozen=True) +class ContextResolution: + """The effective window of one turn and why it was chosen.""" + + effective: int + evidence: ContextEvidence + source: str + observations: tuple[ContextObservation, ...] = () + conflicts: tuple[ContextConflict, ...] = () + provider_io: bool = False + cached: bool = False + probe_errors: tuple[str, ...] = () + # The route this resolution describes. Empty for resolutions built + # directly from observations by internal callers. The URL can carry + # credentials, so it stays out of repr() and to_dict(). + endpoint_url: str = field(default="", repr=False) + model: str = "" + + @property + def mismatch(self) -> bool: + return bool(self.conflicts) + + @property + def budget_limit(self) -> int: + """Window the runtime may budget against; 0 means budget reactively.""" + return self.effective if self.evidence is not ContextEvidence.UNKNOWN else 0 + + @property + def shaping_window(self) -> int: + """Window for the legacy history compaction/trim helpers. + + Those helpers predate typed evidence and always size against some + window, using DEFAULT_CONTEXT when none is known. This only feeds them + a number; it never creates provenance for that number. + """ + if self.budget_limit: + return self.budget_limit + from src.model_context import DEFAULT_CONTEXT + return DEFAULT_CONTEXT + + def applies_to(self, endpoint_url: str, model: str) -> bool: + """Whether this resolution may be reused for the given route.""" + if not self.endpoint_url and not self.model: + return True + return self.endpoint_url == endpoint_url and self.model == model + + def observe_runtime_limit(self, limit: Any, source: str = "provider_rejection") -> "ContextResolution": + """Fold a limit the provider stated during this turn. Performs no I/O.""" + try: + value = int(limit or 0) + except (TypeError, ValueError): + return self + if value <= 0: + return self + observation = ContextObservation(ContextEvidence.RUNTIME_CONFIRMED, value, source) + if observation in self.observations: + return self + combined = combine_observations((*self.observations, observation)) + return replace( + combined, + provider_io=self.provider_io, + cached=self.cached, + probe_errors=self.probe_errors, + endpoint_url=self.endpoint_url, + model=self.model, + ) + + def to_dict(self) -> dict: + return { + "effective": self.effective, + "evidence": self.evidence.value, + "source": self.source, + "mismatch": self.mismatch, + "conflicts": [conflict.to_dict() for conflict in self.conflicts], + "observations": [observation.to_dict() for observation in self.observations], + "provider_io": self.provider_io, + "cached": self.cached, + "probe_errors": list(self.probe_errors), + } + + +UNRESOLVED_CONTEXT = ContextResolution(0, ContextEvidence.UNKNOWN, "none") + +_MEASURED = (ContextEvidence.RUNTIME_CONFIRMED, ContextEvidence.PROVIDER_ADVERTISED) + + +def _conflicts(observations: tuple[ContextObservation, ...]) -> tuple[ContextConflict, ...]: + conflicts = [] + for index, first in enumerate(observations): + for second in observations[index + 1:]: + if first.value == second.value: + continue + classes = {first.evidence, second.evidence} + if ContextEvidence.OPERATOR_DECLARED in classes: + operator, other = ( + (first, second) if first.evidence is ContextEvidence.OPERATOR_DECLARED + else (second, first) + ) + # A declared window replaces the static table and may cap a + # measured window. Only a declaration above what the runtime + # or provider supports contradicts it. + if other.evidence not in _MEASURED or operator.value < other.value: + continue + conflicts.append(ContextConflict(first, second)) + return tuple(conflicts) + + +def _strongest(observations, evidence: ContextEvidence) -> Optional[ContextObservation]: + matching = [observation for observation in observations if observation.evidence is evidence] + return min(matching, key=lambda observation: observation.value) if matching else None + + +def combine_observations(observations) -> ContextResolution: + """Choose the effective window deterministically from observations. + + The smallest runtime-confirmed value wins, else the smallest provider + value. An operator declaration caps either, and replaces the known table + or an unknown window. The known table is used only when nothing stronger + exists. No observation yields an unknown window of 0. + """ + observations = tuple(observations) + measured = ( + _strongest(observations, ContextEvidence.RUNTIME_CONFIRMED) + or _strongest(observations, ContextEvidence.PROVIDER_ADVERTISED) + ) + operator = _strongest(observations, ContextEvidence.OPERATOR_DECLARED) + if measured and operator: + chosen = operator if operator.value < measured.value else measured + else: + chosen = measured or operator or _strongest(observations, ContextEvidence.KNOWN_TABLE) + conflicts = _conflicts(observations) + if chosen is None: + return replace(UNRESOLVED_CONTEXT, observations=observations, conflicts=conflicts) + return ContextResolution( + chosen.value, chosen.evidence, chosen.source, + observations=observations, conflicts=conflicts, + ) + + +# --------------------------------------------------------------------------- +# Provider metadata probe +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class _ProbeResult: + observations: tuple[ContextObservation, ...] = () + errors: tuple[str, ...] = () + io: bool = False + + +_probe_cache: dict[tuple[str, str, str], tuple[float, _ProbeResult]] = {} + + +def clear_probe_cache() -> None: + _probe_cache.clear() + + +_DEFAULT_PORTS = {"http": 80, "https": 443} + + +def _origin(url: str) -> tuple[str, str, Optional[int]]: + parsed = urlparse(url or "") + scheme = parsed.scheme.lower() + try: + port = parsed.port + except ValueError: + return ("", "", None) + return (scheme, (parsed.hostname or "").lower(), port or _DEFAULT_PORTS.get(scheme)) + + +def _http_client(timeout: float): + # Credentials must never follow a redirect to another location. + return httpx.AsyncClient(timeout=timeout, follow_redirects=False) + + +def _provider_urls(endpoint_url: str) -> tuple[Optional[str], str]: + """Models catalog URL and the server-resolved form of the endpoint. + + Both come from the existing endpoint resolver, which may rewrite an + unresolvable host to its Tailscale address. Blocking (DNS, subprocess); + call it off the event loop. + """ + from src.endpoint_resolver import build_models_url, resolve_url + + return build_models_url(endpoint_url), resolve_url(endpoint_url) + + +def _probe_headers(trusted_origins, target_url: str, headers: Optional[Mapping[str, Any]]) -> dict: + """Forward the turn's provider credentials only to the provider's origin.""" + origin = _origin(target_url) + if not headers or not origin[1] or origin not in trusted_origins: + return {} + return { + str(name): str(value) for name, value in headers.items() + if value is not None and str(name).lower() not in _REQUEST_ONLY_HEADERS + } + + +def _auth_fingerprint(headers: Optional[Mapping[str, Any]]) -> str: + if not headers: + return "" + material = json.dumps( + sorted((str(k).lower(), str(v)) for k, v in headers.items() + if v is not None and str(k).lower() not in _REQUEST_ONLY_HEADERS), + separators=(",", ":"), + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16] + + +def _serving_base(endpoint_url: str) -> str: + # Same derivation the regular runtime uses for llama.cpp server routes. + return endpoint_url.split("/v1")[0] if "/v1" in endpoint_url else endpoint_url.rsplit("/", 1)[0] + + +async def _get_json(client, url, headers, errors, label): + try: + response = await client.get(url, headers=headers) + except httpx.TimeoutException: + errors.append(f"{label}:timeout") + return None + except httpx.TransportError: + errors.append(f"{label}:transport_error") + return None + status = getattr(response, "status_code", 0) + if not (200 <= int(status or 0) < 300): + errors.append(f"{label}:http_{status}") + return None + try: + return response.json() + except Exception: + errors.append(f"{label}:invalid_payload") + return None + + +def _positive_int(value) -> int: + if isinstance(value, bool) or not isinstance(value, (int, float)) or value <= 0: + return 0 + return int(value) + + +async def _probe(endpoint_url, model, headers, is_local, observations, errors, timeout): + from src.copilot import is_copilot_base + from src.model_context import _model_ctx_from_entry + + # Credentials go only to the configured provider's origin, or to the + # form of that same endpoint the server-owned resolver produced. + trusted = {_origin(endpoint_url)} + async with _http_client(timeout) as client: + if is_local: + base = _serving_base(endpoint_url) + slots = await _get_json( + client, f"{base}/slots", _probe_headers(trusted, f"{base}/slots", headers), + errors, "slots", + ) + n_ctx = _positive_int(slots[0].get("n_ctx")) if ( + isinstance(slots, list) and slots and isinstance(slots[0], dict) + ) else 0 + if not n_ctx: + props = await _get_json( + client, f"{base}/props", _probe_headers(trusted, f"{base}/props", headers), + errors, "props", + ) + generation = props.get("default_generation_settings") if isinstance(props, dict) else None + n_ctx = _positive_int(generation.get("n_ctx")) if isinstance(generation, dict) else 0 + source = "llamacpp_props" + else: + source = "llamacpp_slots" + if n_ctx: + observations.append( + ContextObservation(ContextEvidence.RUNTIME_CONFIRMED, n_ctx, source) + ) + + # Copilot's catalog needs headers this layer does not own; an + # unauthenticated probe only fails. Its models are table-covered. + if is_copilot_base(endpoint_url): + errors.append("models:unsupported_endpoint") + return + # URL building may resolve the host (DNS, tailscale lookup); keep that + # off the event loop and inside the probe deadline. + models_url, resolved_endpoint = await asyncio.to_thread(_provider_urls, endpoint_url) + if not models_url: + errors.append("models:unsupported_endpoint") + return + trusted.add(_origin(resolved_endpoint)) + payload = await _get_json( + client, models_url, _probe_headers(trusted, models_url, headers), + errors, "models", + ) + if payload is None: + return + entries = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(entries, list): + errors.append("models:invalid_payload") + return + wanted = model.split("/")[-1] + for entry in entries: + if not isinstance(entry, dict): + continue + entry_id = str(entry.get("id") or "") + if entry_id == model or entry_id.split("/")[-1] == wanted: + value = _model_ctx_from_entry(entry) + if value: + observations.append(ContextObservation( + ContextEvidence.PROVIDER_ADVERTISED, int(value), "models_catalog", + )) + else: + errors.append("models:no_window_listed") + return + errors.append("models:model_not_listed") + + +async def probe_provider_context( + endpoint_url: str, + model: str, + *, + headers: Optional[Mapping[str, Any]] = None, + deadline_seconds: float = PROBE_DEADLINE_SECONDS, + is_local: Optional[bool] = None, +) -> _ProbeResult: + """Query provider metadata once, bounded by ``deadline_seconds``. + + Never raises: every failure is reported as a short, secret-free error code + so a turn can continue with other evidence. + """ + observations: list[ContextObservation] = [] + errors: list[str] = [] + if is_local is None: + is_local = await _is_local(endpoint_url) + timeout = max(0.1, float(deadline_seconds)) + try: + await asyncio.wait_for( + _probe(endpoint_url, model, headers, is_local, observations, errors, timeout), + timeout=timeout, + ) + except asyncio.TimeoutError: + errors.append("deadline_exceeded") + except Exception as exc: + logger.debug("Context window probe failed: %s", type(exc).__name__) + errors.append("probe_failed") + return _ProbeResult(tuple(observations), tuple(errors), io=True) + + +async def _is_local(endpoint_url: str) -> bool: + from src.model_context import is_local_endpoint + + try: + # Reads configured endpoints from the local database on the calling + # thread, as the regular runtime does. Moving it to worker threads + # gives SQLite sessions per-thread connections the app never uses. + return bool(is_local_endpoint(endpoint_url)) + except Exception: + return False + + +async def _cached_probe(endpoint_url, model, headers, deadline_seconds, clock): + is_local = await _is_local(endpoint_url) + key = (endpoint_url, model, _auth_fingerprint(headers)) + if not is_local: + cached = _probe_cache.get(key) + if cached and cached[0] > clock(): + return cached[1], True + result = await probe_provider_context( + endpoint_url, model, headers=headers, deadline_seconds=deadline_seconds, + is_local=is_local, + ) + if not is_local: + ttl = PROBE_CACHE_TTL_SECONDS if result.observations else PROBE_FAILURE_TTL_SECONDS + _probe_cache[key] = (clock() + ttl, result) + return result, False + + +def declared_context_window(client_runtime_context: Any) -> int: + """Operator/client declared transport window, or 0.""" + if not isinstance(client_runtime_context, Mapping): + return 0 + try: + value = int(client_runtime_context.get("model_context_window") or 0) + except (TypeError, ValueError): + return 0 + return value if value > 0 else 0 + + +async def resolve_effective_context( + endpoint_url: str, + model: str, + *, + headers: Optional[Mapping[str, Any]] = None, + client_runtime_context: Any = None, + deadline_seconds: float = PROBE_DEADLINE_SECONDS, + probe: bool = True, + clock=time.monotonic, +) -> ContextResolution: + """Resolve the effective context window for one turn preparation.""" + from src.model_context import _lookup_known + + observations: list[ContextObservation] = [] + errors: tuple[str, ...] = () + provider_io = cached = False + if probe and endpoint_url and model: + result, cached = await _cached_probe( + endpoint_url, model, headers, deadline_seconds, clock, + ) + observations.extend(result.observations) + errors = result.errors + provider_io = result.io and not cached + declared = declared_context_window(client_runtime_context) + if declared: + observations.append(ContextObservation( + ContextEvidence.OPERATOR_DECLARED, declared, "client_runtime_context", + )) + known = _lookup_known(model or "") + if known: + observations.append(ContextObservation(ContextEvidence.KNOWN_TABLE, int(known), "known_table")) + resolution = combine_observations(observations) + resolution = replace( + resolution, provider_io=provider_io, cached=cached, probe_errors=errors, + endpoint_url=endpoint_url or "", model=model or "", + ) + if resolution.mismatch: + logger.info( + "Context window sources disagree for %s: %s", + model, [conflict.to_dict() for conflict in resolution.conflicts], + ) + return resolution + + +def context_metrics(resolution: Optional[ContextResolution], request_tokens: int) -> dict: + """Metrics fields derived only from a stored resolution. Performs no I/O.""" + resolution = resolution or UNRESOLVED_CONTEXT + length = resolution.budget_limit + percent = ( + min(round((request_tokens / length) * 100, 1), 100.0) + if length and request_tokens else 0 + ) + return { + "context_length": length, + "context_percent": percent, + "context_resolution": resolution.to_dict(), + } diff --git a/src/agent_runtime/runtime_selection.py b/src/agent_runtime/runtime_selection.py new file mode 100644 index 000000000..9150f070c --- /dev/null +++ b/src/agent_runtime/runtime_selection.py @@ -0,0 +1,44 @@ +"""Whether a turn runs on the compact (clean v3) preview runtime. + +The chat route decides this once, from facts known before context +preparation, and uses that one value both to prepare the turn (its typed +context resolution) and to stamp the turn contract's selection mode. The +agent loop dispatches on that stamp. Keeping both sides here, with no other +imports, means preparation and dispatch read one rule and cannot drift. + +Runtime selection is not authority: it grants or denies no operation. +""" + +COMPACT_PREVIEW_MODE = "clean_compact_v3_preview" + + +def uses_compact_preview_runtime( + *, + clean_route_requested: bool, + turn_contract_enabled: bool, + agent_mode: bool, + agent_permitted: bool, + image_generation: bool, +) -> bool: + """The single compact-runtime eligibility rule for one turn. + + ``turn_contract_enabled`` is the route's contract policy for this turn + (exact approvals, TUI surface and full-schema routes opt out). + ``agent_permitted`` is false when the user's privileges demote the turn + to plain chat; image generation sessions run their own execution path. + """ + return bool( + clean_route_requested + and turn_contract_enabled + and agent_mode + and agent_permitted + and not image_generation + ) + + +def is_compact_preview_contract(turn_contract) -> bool: + """Whether a turn contract was stamped for the compact runtime.""" + return ( + turn_contract is not None + and getattr(turn_contract, "selection_mode", None) == COMPACT_PREVIEW_MODE + ) diff --git a/src/clean_agent_preview.py b/src/clean_agent_preview.py index 4c0302eb7..8b1372f9a 100644 --- a/src/clean_agent_preview.py +++ b/src/clean_agent_preview.py @@ -21,6 +21,7 @@ import httpx import jsonschema from src.context_compactor import prune_multimodal_images, trim_for_context +from src.agent_runtime.runtime_selection import COMPACT_PREVIEW_MODE from src import agent_runs from src.agent_evidence import command_has_mutation_effect, workspace_artifact_is_usable from src.tool_capabilities import ToolEffect, ToolRunSecurityContext, capabilities_for_action @@ -47,7 +48,7 @@ from src.model_profiles import ( ) ENDPOINT_ID = 'cleanv3' -MODE = 'clean_compact_v3_preview' +MODE = COMPACT_PREVIEW_MODE class ProviderStreamError(Exception): @@ -4904,7 +4905,9 @@ async def preview_model_response(client, endpoint_url, headers, request, recover ) transport_attempts = 0 while True: - limit = recovery.get('context_limit') + # A provider-stated limit learned this turn overrides the window + # resolved at turn preparation. + limit = recovery.get('context_limit') or recovery.get('budget_limit') if limit: message_context = max(1, limit - estimate_tool_schema_tokens(request.get('tools')) @@ -4995,6 +4998,7 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac client_runtime_context=None, max_tokens=768, max_rounds=8, max_tool_calls=0, external_tool_schemas=None, temperature=0.0, + context_resolution=None, request_authority=MISSING_AUTHORITY, **ignored): from src.generation_sampling import validate_temperature temperature = validate_temperature(temperature) @@ -5249,7 +5253,18 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac static_fetch_failed_urls = set() entity_result_links = {} calendar_create_confirmation = '' - context_recovery = {} + # Resolve the effective window once, before any model request, so the + # turn budgets against it and metrics report exactly what it ran under. + # The chat route prepares it; only callers arriving without one (or with + # one for a different route) resolve here. Terminal metrics must never + # start this discovery themselves. + if context_resolution is None or not context_resolution.applies_to(endpoint_url, model): + from src.agent_runtime.context_resolution import resolve_effective_context + context_resolution = await resolve_effective_context( + endpoint_url, model, headers=headers, + client_runtime_context=client_runtime_context, + ) + context_recovery = {'budget_limit': context_resolution.budget_limit} successful_write = False editor_batch_pending = False editor_suggested_finds = [] @@ -8026,9 +8041,15 @@ async def stream_preview(*, endpoint_url, model, messages, headers, turn_contrac return elapsed = time.monotonic() - started ttft = first_token - started if first_token else None + from src.agent_runtime.context_resolution import context_metrics + # Report the stored resolution plus any limit the provider stated during + # this turn. This is pure bookkeeping; no metadata request happens here. + context_resolution = context_resolution.observe_runtime_limit( + context_recovery.get('context_limit')) yield event({'type': 'metrics', 'data': { 'email_task_scope': {**intent_accounting, 'failed': intent_scope_failed}, 'model': model, 'input_tokens': usage_in, 'output_tokens': usage_out, + **context_metrics(context_resolution, last_request_tokens), 'total_tokens': usage_in + usage_out, 'response_time': round(elapsed, 3), 'time_to_first_token': round(ttft, 3) if ttft is not None else None, 'tokens_per_second': round(usage_out / elapsed, 2) if elapsed > 0 else 0, diff --git a/src/context_compactor.py b/src/context_compactor.py index 1f9be5018..11f450323 100644 --- a/src/context_compactor.py +++ b/src/context_compactor.py @@ -481,12 +481,17 @@ async def maybe_compact( persist: bool = True, compaction_state: Optional[Dict[str, Any]] = None, deterministic: bool = False, + context_length: Optional[int] = None, ) -> tuple: """Check context usage and compact if above threshold. + ``context_length`` lets a caller that already resolved the turn's window + supply it, so this helper does not query the endpoint a second time. + Returns (messages, context_length, was_compacted). """ - context_length = get_context_length(endpoint_url, model) + if context_length is None: + context_length = get_context_length(endpoint_url, model) used = estimate_tokens(messages) pct = (used / context_length) * 100 if context_length else 0 threshold = auto_compact_threshold_percent() diff --git a/tests/conftest.py b/tests/conftest.py index 0affe928a..cdc3129d9 100644 --- a/tests/conftest.py +++ b/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 diff --git a/tests/test_context_resolution.py b/tests/test_context_resolution.py new file mode 100644 index 000000000..dcdde7bb0 --- /dev/null +++ b/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 diff --git a/tests/test_context_resolution_route.py b/tests/test_context_resolution_route.py new file mode 100644 index 000000000..648f434a3 --- /dev/null +++ b/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 diff --git a/website/configuration-reference.md b/website/configuration-reference.md index b06db2d78..60a9e972c 100644 --- a/website/configuration-reference.md +++ b/website/configuration-reference.md @@ -65,15 +65,15 @@ The source tree reads **109** `ODYSSEUS_*` variables: 79 an operator may want to | `ODYSSEUS_LOCAL_MODEL_GATE` | `'true'` | `src/llm_core.py:95` | On by default. Set 0, false, no or off to drop the gate that checks a local endpoint before routing a request to it. | | `ODYSSEUS_MISTRAL_REASONING_EFFORT` | `'high'` | `src/llm_core.py:1723` | Reasoning effort sent to Mistral thinking-capable models. The API accepts high, medium, low and none. | | `ODYSSEUS_MLX_IMAGE_VLM_MODEL` | *unset* | `scripts/mlx_image_server.py:299` | Vision-language model id for the MLX image server script. Required unless `--vlm-model` is passed on the command line. | -| `ODYSSEUS_QWEN_ROUTE_THINKING` | `'auto'` | `src/agent_loop.py:169` | Thinking policy for the Qwen routing step. An unrecognized value falls back to `auto`. | +| `ODYSSEUS_QWEN_ROUTE_THINKING` | `'auto'` | `src/agent_loop.py:170` | Thinking policy for the Qwen routing step. An unrecognized value falls back to `auto`. | ### Agent loop and tool execution | Variable | Default | Read in | What it does | |---|---|---|---| | `ODYSSEUS_DISABLE_MCP` | `''` | `src/builtin_mcp.py:89` | Truthy disables MCP entirely, as an escape hatch for compatibility problems with a server. | -| `ODYSSEUS_MAX_VISUAL_EVIDENCE_FRAMES` | `'3'` | `src/agent_loop.py:15361` | How many video frames one tool result may contribute. Clamped to 1-8. | -| `ODYSSEUS_MAX_VISUAL_EVIDENCE_IMAGES` | `'1'` | `src/agent_loop.py:15329` | How many images one tool result may contribute to the model turn. Clamped to 1-8. | +| `ODYSSEUS_MAX_VISUAL_EVIDENCE_FRAMES` | `'3'` | `src/agent_loop.py:15362` | How many video frames one tool result may contribute. Clamped to 1-8. | +| `ODYSSEUS_MAX_VISUAL_EVIDENCE_IMAGES` | `'1'` | `src/agent_loop.py:15330` | How many images one tool result may contribute to the model turn. Clamped to 1-8. | | `ODYSSEUS_MCP_ALLOWED_COMMANDS` | `''` | `src/agent_tools/admin_tools.py:140` | Security-relevant. Comma-separated allowlist of MCP launcher basenames the agent may start. Empty by default, and the deny list still wins. | | `ODYSSEUS_PYTHON_TOOL_SITE_PACKAGES` | `''` | `src/agent_tools/subprocess_tools.py:853` (+1 more) | Security-relevant. Absolute package roots, separated by the platform path separator, exposed to the sandboxed Python tool. Empty exposes none. | | `ODYSSEUS_SCRIPT_HOST` | `'localhost'` | `src/builtin_actions.py:919` | Default host for the run-script action. `localhost`, `127.0.0.1`, `local` and empty run locally; any other value runs over SSH. | @@ -195,9 +195,9 @@ Listed for completeness. Setting one of these on a real install is either a no-o | Variable | Default | Read in | What it does | |---|---|---|---| -| `ODYSSEUS_CAPTURE_MODEL_REQUESTS` | `''` | `src/agent_loop.py:3924` | Truthy writes model-request snapshots for local debugging. The marker file `/tmp/odysseus_capture_model_requests` enables the same thing. | -| `ODYSSEUS_EXPOSE_RAW_BROWSER_MCP` | `''` | `src/agent_loop.py:4129` | Truthy stops hiding the raw Playwright MCP tools from agent prompts when the private-browser tool is available. | -| `ODYSSEUS_TOOL_CONTRACT_ROOT` | `'/scripts'` | `src/clean_agent_preview.py:2183` (+1 more) | Directory holding the tool-contract scripts the clean-agent preview loads. The default is the repository's bundled scripts directory; set the variable to override it. | +| `ODYSSEUS_CAPTURE_MODEL_REQUESTS` | `''` | `src/agent_loop.py:3925` | Truthy writes model-request snapshots for local debugging. The marker file `/tmp/odysseus_capture_model_requests` enables the same thing. | +| `ODYSSEUS_EXPOSE_RAW_BROWSER_MCP` | `''` | `src/agent_loop.py:4130` | Truthy stops hiding the raw Playwright MCP tools from agent prompts when the private-browser tool is available. | +| `ODYSSEUS_TOOL_CONTRACT_ROOT` | `'/scripts'` | `src/clean_agent_preview.py:2184` (+1 more) | Directory holding the tool-contract scripts the clean-agent preview loads. The default is the repository's bundled scripts directory; set the variable to override it. | ### Email @@ -222,8 +222,8 @@ Listed for completeness. Setting one of these on a real install is either a no-o | `ODYSSEUS_QA_TEACHER_ATTEMPTS` | `'3'` | `scripts/odysseus_conversation_qa.py:370` | Retry budget for the conversation-QA teacher model call. Clamped to 1-3. | | `ODYSSEUS_QA_TEACHER_TIMEOUT` | `'120'` | `scripts/odysseus_conversation_qa.py:372` | Timeout in seconds for that call. Clamped to 15-120. | | `ODYSSEUS_RUNTIME_REVISION` | `''` | `routes/chat_helpers.py:198` (+1 more) | Revision string stamped into each captured SFT trace record, so a trace can be tied back to the build that produced it. | -| `ODYSSEUS_SFT_DISABLE_WORKSPACE_TOOLS` | `'1'` | `src/agent_loop.py:7407` | On by default. Keeps synthetic personal-assistant fixtures out of workspace mode; set 0, false, no or off to let them through. | -| `ODYSSEUS_SFT_FORCE_UTC_TIMEZONE` | `'0'` | `routes/chat_routes.py:2080` | Truthy forces `sft_` accounts to UTC for deterministic batch generation. Interactive accounts still follow the browser timezone. | +| `ODYSSEUS_SFT_DISABLE_WORKSPACE_TOOLS` | `'1'` | `src/agent_loop.py:7408` | On by default. Keeps synthetic personal-assistant fixtures out of workspace mode; set 0, false, no or off to let them through. | +| `ODYSSEUS_SFT_FORCE_UTC_TIMEZONE` | `'0'` | `routes/chat_routes.py:2094` | Truthy forces `sft_` accounts to UTC for deterministic batch generation. Interactive accounts still follow the browser timezone. | | `ODYSSEUS_SFT_TRACE_CAPTURE` | `'1'` | `routes/chat_helpers.py:161` (+1 more) | On by default, but only for owners whose name starts with `sft_`. Set 0, false, no or off to stop writing training traces. | | `ODYSSEUS_SFT_TRACE_DIR` | *unset* | `routes/chat_helpers.py:195` (+2 more) | Directory the SFT trace JSONL files are written to. Defaults to `sft_traces` under the data directory. | | `ODYSSEUS_SKIP_RUN_HINT` | *unset* | `setup.py:284` | Any non-empty value suppresses the `start the server with` hint at the end of setup. `start-macos.sh` sets it because it starts the server itself. |