diff --git a/agent/agent_runtime_helpers.py b/agent/agent_runtime_helpers.py index 97f78b9143c..092a87da95e 100644 --- a/agent/agent_runtime_helpers.py +++ b/agent/agent_runtime_helpers.py @@ -3401,75 +3401,139 @@ def reapply_reasoning_echo_for_provider(agent, api_messages: list) -> int: return changed +def _iter_httpx_pool_objects(http_client: Any): + """Yield httpcore pool objects reachable from an httpx client. + + Hermes' keepalive client (#10324 / ``_build_keepalive_http_client``) and + any ``HTTP(S)_PROXY`` configuration put live connections on *mounted* + transports (``client._mounts``), not only on the default + ``client._transport``. Walking the default transport alone makes + ``force_close_tcp_sockets`` return 0 while a stream is still mid-recv — + the interrupt logs success and the provider keeps burning the slot + (#72975). + """ + seen_pools: set[int] = set() + + def _emit(pool: Any): + if pool is None: + return + marker = id(pool) + if marker in seen_pools: + return + seen_pools.add(marker) + yield pool + + def _pools_for_transport(transport: Any): + if transport is None: + return + # Normal httpx.HTTPTransport / HTTPProxy-as-transport: connections + # live under ``_pool``. HTTPProxy itself *is* a ConnectionPool and + # may be mounted directly — then ``_connections`` is on the + # transport. + pool = getattr(transport, "_pool", None) + if pool is not None: + yield from _emit(pool) + return + if getattr(transport, "_connections", None) is not None: + yield from _emit(transport) + + try: + yield from _pools_for_transport(getattr(http_client, "_transport", None)) + mounts = getattr(http_client, "_mounts", None) or {} + for _pattern, mounted in list(mounts.items()): + yield from _pools_for_transport(mounted) + except Exception: + return + + +def _connection_candidates(conn: Any): + """Walk nested ``_connection`` wrappers (proxy tunnel → HTTP11/2).""" + seen: set[int] = set() + stack = [conn] + while stack: + candidate = stack.pop() + if candidate is None: + continue + marker = id(candidate) + if marker in seen: + continue + seen.add(marker) + yield candidate + inner = getattr(candidate, "_connection", None) + if inner is not None and id(inner) not in seen: + stack.append(inner) + + def _iter_pool_sockets(client: Any): """Yield raw sockets reachable from an OpenAI/httpx client pool. httpcore 1.x stores the concrete HTTP11/HTTP2 connection under ``conn._connection``; older versions exposed stream attributes directly - on the pool entry. Keep the traversal defensive because these are private - transport internals and vary across httpx/httpcore releases. + on the pool entry. Proxy tunnels wrap another layer + (``TunnelHTTPConnection`` / ``ForwardHTTPConnection``). Keep the + traversal defensive because these are private transport internals and + vary across httpx/httpcore releases. + + Also walks ``httpx`` mount transports — see ``_iter_httpx_pool_objects``. """ try: http_client = getattr(client, "_client", None) if http_client is None: - return - transport = getattr(http_client, "_transport", None) - if transport is None: - return - pool = getattr(transport, "_pool", None) - if pool is None: - return + # Some SDK wrappers *are* the httpx client (or expose the pool + # directly). Fall through so mount-aware discovery still runs. + http_client = client + pools = list(_iter_httpx_pool_objects(http_client)) + except Exception: + return + + if not pools: + return + + seen: set[int] = set() + for pool in pools: connections = ( getattr(pool, "_connections", None) or getattr(pool, "_pool", None) or [] ) - except Exception: - return - - seen: set[int] = set() - for conn in list(connections): - candidates = [conn] - inner = getattr(conn, "_connection", None) - if inner is not None: - candidates.append(inner) - for candidate in candidates: - stream = ( - getattr(candidate, "_network_stream", None) - or getattr(candidate, "_stream", None) - ) - if stream is None: - continue - sock = getattr(stream, "_sock", None) - if sock is None: - get_extra_info = getattr(stream, "get_extra_info", None) - if callable(get_extra_info): - try: - sock = get_extra_info("socket") - except Exception: - sock = None - if sock is None: - wrapped = getattr(stream, "stream", None) - if wrapped is not None: - sock = getattr(wrapped, "_sock", None) - if sock is None: - # anyio-backed streams expose the raw socket through - # SocketAttribute.raw_socket when available. - wrapped = getattr(stream, "_stream", None) - extra = getattr(wrapped, "extra", None) - if callable(extra): - try: - from anyio.abc import SocketAttribute - sock = extra(SocketAttribute.raw_socket) - except Exception: - sock = None - if sock is None: - continue - marker = id(sock) - if marker in seen: - continue - seen.add(marker) - yield sock + for conn in list(connections): + for candidate in _connection_candidates(conn): + stream = ( + getattr(candidate, "_network_stream", None) + or getattr(candidate, "_stream", None) + ) + if stream is None: + continue + sock = getattr(stream, "_sock", None) + if sock is None: + get_extra_info = getattr(stream, "get_extra_info", None) + if callable(get_extra_info): + try: + sock = get_extra_info("socket") + except Exception: + sock = None + if sock is None: + wrapped = getattr(stream, "stream", None) + if wrapped is not None: + sock = getattr(wrapped, "_sock", None) + if sock is None: + # anyio-backed streams expose the raw socket through + # SocketAttribute.raw_socket when available. + wrapped = getattr(stream, "_stream", None) + extra = getattr(wrapped, "extra", None) + if callable(extra): + try: + from anyio.abc import SocketAttribute + sock = extra(SocketAttribute.raw_socket) + except Exception: + sock = None + if sock is None: + continue + marker = id(sock) + if marker in seen: + continue + seen.add(marker) + yield sock def cleanup_dead_connections(agent) -> bool: diff --git a/agent/chat_completion_helpers.py b/agent/chat_completion_helpers.py index 7ce13665212..fac42cc3cf7 100644 --- a/agent/chat_completion_helpers.py +++ b/agent/chat_completion_helpers.py @@ -526,10 +526,16 @@ def direct_api_call(agent, api_kwargs: dict): def _abort_active_request(reason: str) -> None: """Abort the inline request from a watchdog/interrupt thread.""" + # Abort while still holding the holder lock: the instant it is + # released, the inline finally may pop + cache the client for reuse + # and the NEXT call check it out — a late abort would then poison + # the slot and shut down an innocent in-flight request's sockets + # (same atomicity contract as _close_request_client_once in the + # interruptible variants; the abort itself never blocks). with request_client_lock: request_client = request_client_holder["client"] - if request_client is not None: - agent._abort_request_openai_client(request_client, reason=reason) + if request_client is not None: + agent._abort_request_openai_client(request_client, reason=reason) def _make_client(reason: str, kind: str = "openai"): # direct_api_call only runs for OpenAI-wire chat_completions cron @@ -542,6 +548,10 @@ def direct_api_call(agent, api_kwargs: dict): agent._active_request_abort = _abort_active_request return client + # Only a clean return may report the reuse reason (request_complete): + # after an error or interrupt the wire client is really closed so the + # retry builds a fresh pool (see _REQUEST_CLIENT_REUSE_REASONS). + succeeded = False try: response = _dispatch_nonstreaming_api_request( agent, api_kwargs, make_client=_make_client @@ -554,6 +564,7 @@ def direct_api_call(agent, api_kwargs: dict): if getattr(agent, "_interrupt_requested", False): raise InterruptedError("Agent interrupted during API call") _reset_stale_streak(agent) + succeeded = True return response finally: if getattr(agent, "_active_request_abort", None) is _abort_active_request: @@ -562,7 +573,10 @@ def direct_api_call(agent, api_kwargs: dict): request_client = request_client_holder["client"] request_client_holder["client"] = None if request_client is not None: - agent._close_request_openai_client(request_client, reason="request_complete") + agent._close_request_openai_client( + request_client, + reason="request_complete" if succeeded else "request_error_cleanup", + ) def interruptible_api_call(agent, api_kwargs: dict): @@ -640,20 +654,28 @@ def interruptible_api_call(agent, api_kwargs: dict): and owner_tid is not None and owner_tid != threading.get_ident() ) - if not stranger_thread: - # Owning thread (or no recorded owner) → pop and fully close. - request_client_holder["client"] = None - request_client_holder["owner_tid"] = None + if stranger_thread: + # Abort while still holding the holder lock: the instant it + # is released, the worker's finally may pop + cache the client + # for reuse and the NEXT call check it out — an abort landing + # after that would poison the slot and shut down an innocent + # in-flight request's sockets. The abort itself never blocks + # (socket shutdown + slot poison), so holding the lock across + # it only delays the racing pop, never the data path. + if request_client_kind.get("value", "openai") == "anthropic_messages": + agent._abort_request_anthropic_client( + request_client, reason=reason + ) + else: + agent._abort_request_openai_client(request_client, reason=reason) + return + # Owning thread (or no recorded owner) → pop and fully close. + request_client_holder["client"] = None + request_client_holder["owner_tid"] = None if request_client is None: return - kind = request_client_kind.get("value", "openai") - if kind == "anthropic_messages": - if stranger_thread: - agent._abort_request_anthropic_client(request_client, reason=reason) - else: - agent._close_request_anthropic_client(request_client, reason=reason) - elif stranger_thread: - agent._abort_request_openai_client(request_client, reason=reason) + if request_client_kind.get("value", "openai") == "anthropic_messages": + agent._close_request_anthropic_client(request_client, reason=reason) else: agent._close_request_openai_client(request_client, reason=reason) @@ -690,7 +712,15 @@ def interruptible_api_call(agent, api_kwargs: dict): return result["error"] = e finally: - _close_request_client_once("request_complete") + # Reuse reason only on a clean response; any other outcome — + # error, or the cancel-swallow return above (which leaves both + # result slots None) — really closes so the next attempt builds + # a fresh pool (see _REQUEST_CLIENT_REUSE_REASONS). + _close_request_client_once( + "request_complete" + if result["response"] is not None + else "request_error_cleanup" + ) # ── Stale-call timeout (mirrors streaming stale detector) ──────── # Non-streaming calls return nothing until the full response is @@ -2757,20 +2787,27 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= and owner_tid is not None and owner_tid != threading.get_ident() ) - if not stranger_thread: - request_client_holder["client"] = None - request_client_holder["owner_tid"] = None + if stranger_thread: + # Abort under the holder lock — see the non-streaming variant + # for why the holder read and the abort must be atomic (a late + # abort would otherwise hit the NEXT request's checkout). + if request_client_kind.get("value", "openai") == "anthropic_messages": + agent._abort_request_anthropic_client( + request_client, reason=reason + ) + else: + agent._abort_request_openai_client(request_client, reason=reason) + return + request_client_holder["client"] = None + request_client_holder["owner_tid"] = None if request_client is None: return + # Stranger threads returned under the lock above, so only the owner + # (or an any-thread-safe stream handle) reaches the close dispatch. if request_kind == "stream": _close_request_stream_handle(request_client, reason) elif request_kind == "anthropic_messages": - if stranger_thread: - agent._abort_request_anthropic_client(request_client, reason=reason) - else: - agent._close_request_anthropic_client(request_client, reason=reason) - elif stranger_thread: - agent._abort_request_openai_client(request_client, reason=reason) + agent._close_request_anthropic_client(request_client, reason=reason) else: agent._close_request_openai_client(request_client, reason=reason) @@ -2945,6 +2982,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= _diag = agent._stream_diag_init() request_client_holder["diag"] = _diag _writer_token = {"value": None} + attempt_request_client = {"value": None} def _open_stream(next_api_kwargs: dict[str, Any]): stream_kwargs = { @@ -2966,6 +3004,7 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= api_kwargs=stream_kwargs, ) ) + attempt_request_client["value"] = request_client last_chunk_time["t"] = time.time() agent._touch_activity("waiting for provider response (streaming)") return request_client.chat.completions.create(**stream_kwargs) @@ -3081,6 +3120,26 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= pass if agent._interrupt_requested: + # Abandoning a half-read SSE response leaves its connection + # permanently checked out of the httpx pool — and the partial + # response built below makes the worker's finally report a + # reuse-reason close, which would cache the client together + # with the leaked connection (each interrupt leaking one more + # until the pool exhausts). Close the stream here, on the + # owning thread, so the connection is released first. + try: + stream.close() + except Exception: + # Connection may still be checked out — poison the slot so + # the finally's close really closes the pool instead of + # caching it (owner-thread abort: shutdown is safe, and the + # FD release still happens in the finally below). + request_client = attempt_request_client["value"] + if request_client is not None: + agent._abort_request_openai_client( + request_client, + reason="interrupt_stream_close_failed", + ) break if not _stream_attempt_is_active(stream_attempt_id): @@ -3892,7 +3951,14 @@ def interruptible_streaming_api_call(agent, api_kwargs: dict, *, on_first_delta= return finally: _close_managed_stream() - _close_request_client_once("stream_request_complete") + # Reuse reason only on a clean stream; any other outcome (error, + # cancel-swallow) really closes so the next attempt builds a + # fresh pool (see _REQUEST_CLIENT_REUSE_REASONS). + _close_request_client_once( + "stream_request_complete" + if result["response"] is not None + else "stream_error_cleanup" + ) # Provider-configured stale timeout takes priority over env default. _cfg_stale = get_provider_stale_timeout(agent.provider, agent.model) diff --git a/agent/codex_runtime.py b/agent/codex_runtime.py index 0c5bb7b7050..0e0b87b2196 100644 --- a/agent/codex_runtime.py +++ b/agent/codex_runtime.py @@ -74,7 +74,10 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]: try: if not agent._session_db_created: agent._ensure_db_session() - agent._session_db.update_token_counts( + # Enqueued for the SessionDB background writer — keeps the + # per-call accounting write off the turn thread (see + # conversation_loop's queue_token_counts call). + agent._session_db.queue_token_counts( agent.session_id, model=agent.model, billing_provider=agent.provider, @@ -154,7 +157,8 @@ def _record_codex_app_server_usage(agent, turn) -> dict[str, Any]: try: if not agent._session_db_created: agent._ensure_db_session() - agent._session_db.update_token_counts( + # Enqueued for the SessionDB background writer (see above). + agent._session_db.queue_token_counts( agent.session_id, input_tokens=canonical_usage.input_tokens, output_tokens=canonical_usage.output_tokens, @@ -1399,7 +1403,20 @@ def run_codex_stream(agent, api_kwargs: dict, client: Any = None, on_first_delta try: close_fn() except Exception: - pass + # A failed close can leave this response's connection + # checked out of the httpx pool while the caller's finally + # reports a reuse-reason close (e.g. interrupt_check broke + # the event loop with collected output) — caching the + # client with the leaked connection. Poison the slot so + # that close really closes the pool (owner-thread abort; + # mirrors the chat-streaming interrupt-break handling). + # ``client is None`` means the shared primary client, + # which is never reuse-cached and must not have its + # sockets force-shut here. + if client is not None: + agent._abort_request_openai_client( + active_client, reason="codex_stream_close_failed" + ) def run_codex_create_stream_fallback(agent, api_kwargs: dict, client: Any = None): diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 73fa1e36f21..4837925ce62 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -1354,6 +1354,125 @@ class ContextCompressor(ContextEngine): previous = telemetry.get("aux_call_duration_ms") or 0 telemetry["aux_call_duration_ms"] = previous + max(0, int(duration_ms)) + def _emit_init_summary_once(self) -> None: + """Emit the informative startup line once, on first resolution. + + Deferred out of ``__init__`` (#32221): the line reports resolved token + budgets, so emitting it there would force the synchronous + ``get_model_context_length()`` probe during construction. Reads via + the properties below are safe here because + ``_resolved_context_length`` is already set. + """ + if not getattr(self, "_log_init_summary", False): + return + self._log_init_summary = False + logger.info( + "Context compressor initialized: model=%s context_length=%d " + "threshold=%d (%.0f%%) target_ratio=%.0f%% tail_budget=%d " + "provider=%s base_url=%s", + self.model, self._resolved_context_length, self.threshold_tokens, + self.threshold_percent * 100, self.summary_target_ratio * 100, + self.tail_token_budget, + self.provider or "none", self.base_url or "none", + ) + + def _resolve_context_length(self) -> int: + """Resolve and cache the model's context length on first access.""" + if self._resolved_context_length is None: + self._resolved_context_length = get_model_context_length( + self.model, + base_url=self.base_url, + api_key=self.api_key, + config_context_length=self._config_context_length, + provider=self.provider, + ) + # Small-context threshold floor: models under 512K trigger at + # >=75% so compaction doesn't fire with half the window still + # free. Raise-only; must run AFTER context_length is resolved + # and BEFORE threshold_tokens is derived (deferred here from + # __init__ along with the resolution itself, #32221). + # _base_threshold_percent already has the per-model override + # applied, so the floor stacks on top of it. + self.threshold_percent = self._effective_threshold_percent( + self._resolved_context_length, self._base_threshold_percent, + ) + self._emit_init_summary_once() + return self._resolved_context_length + + @property + def context_length(self) -> int: + return self._resolve_context_length() + + @context_length.setter + def context_length(self, value: int) -> None: + # No-op guard: repeated assignment of the SAME window (e.g. the codex + # app-server usage callback re-reports the window on every response) + # must not invalidate the derived budgets — that would wipe runtime + # corrections applied directly to threshold_tokens/tail_token_budget + # (see conversation_compression's aux-context threshold sync), which + # persisted on main's eager-init behavior. + if value == getattr(self, "_resolved_context_length", None): + return + self._resolved_context_length = value + # Re-apply the small-context floor (raise-only) for the genuinely new + # window so the invalidated budgets below recompute coherently — + # percent and tokens must derive from the same window. Skipped on + # bare test instances built via object.__new__ that never ran + # __init__ (no _base_threshold_percent). + _base = getattr(self, "_base_threshold_percent", None) + if _base is not None: + self.threshold_percent = self._effective_threshold_percent( + value, _base, + ) + self._threshold_tokens = None + self._tail_token_budget = None + self._max_summary_tokens = None + self._emit_init_summary_once() + + @property + def threshold_tokens(self) -> int: + if self._threshold_tokens is None: + # Resolve the window FIRST (may apply the small-context floor to + # threshold_percent as a side effect) so the percent read below + # is the floored value regardless of argument evaluation order. + _ctx = self.context_length + # Floor: never compress below MINIMUM_CONTEXT_LENGTH tokens even + # if the percentage would suggest a lower value (#14690 handles + # the degenerate small-window case inside the helper). + self._threshold_tokens = self._compute_threshold_tokens( + _ctx, self.threshold_percent, self.max_tokens, + ) + # Apply absolute token cap (compression.threshold_tokens) — + # takes the lower of the ratio-based threshold and the cap. + self._apply_threshold_tokens_cap() + return self._threshold_tokens + + @threshold_tokens.setter + def threshold_tokens(self, value: int) -> None: + self._threshold_tokens = value + + @property + def tail_token_budget(self) -> int: + if self._tail_token_budget is None: + self._tail_token_budget = int(self.threshold_tokens * self.summary_target_ratio) + return self._tail_token_budget + + @tail_token_budget.setter + def tail_token_budget(self, value: int) -> None: + self._tail_token_budget = value + + @property + def max_summary_tokens(self) -> int: + if self._max_summary_tokens is None: + self._max_summary_tokens = min( + int(self.context_length * 0.05), _SUMMARY_TOKENS_CEILING, + ) + return self._max_summary_tokens + + @max_summary_tokens.setter + def max_summary_tokens(self, value: int) -> None: + self._max_summary_tokens = value + def on_session_end(self, session_id: str, messages: List[Dict[str, Any]]) -> None: """Clear all per-session compaction state at a real session boundary. @@ -1988,56 +2107,30 @@ class ContextCompressor(ContextEngine): # deterministic "summary unavailable" handoff and drop the middle window. self.abort_on_summary_failure = abort_on_summary_failure - self.context_length = get_model_context_length( - model, base_url=base_url, api_key=api_key, - config_context_length=config_context_length, - provider=provider, - ) - # Small-context threshold floor: models under 512K trigger at >=75% - # so compaction doesn't fire with half the window still free (the - # incompressible floor makes 50%-triggered compaction thrash on - # 128K-262K models). Raise-only; must run AFTER context_length is - # resolved and BEFORE threshold_tokens is derived. The pre-floor - # value is kept so update_model() can re-derive for a new window - # (switching small -> large must drop back to the configured value). - # Note: _base_threshold_percent already has the per-model override - # applied, so the floor stacks on top of any model-specific threshold. + # Defer context-length resolution to first access (#32221): + # get_model_context_length() can issue a synchronous /models HTTP + # probe, which must not block AIAgent construction. The small-context + # threshold floor and the absolute threshold cap both need the + # resolved window, so they are applied on first resolution (see + # _resolve_context_length / the threshold_tokens property) instead + # of here. update_model() re-derives the floor for a new window from + # _config_threshold_percent (the raw config value snapshotted above), + # so switching small -> large correctly drops back to the configured + # value. + self._config_context_length = config_context_length self._configured_threshold_percent = self.threshold_percent - self.threshold_percent = self._effective_threshold_percent( - self.context_length, self._base_threshold_percent, - ) - threshold_percent = self.threshold_percent - # Floor: never compress below MINIMUM_CONTEXT_LENGTH tokens even if - # the percentage would suggest a lower value. This prevents premature - # compression on large-context models at 50% while keeping the % sane - # for models right at the minimum. _compute_threshold_tokens also - # guards the degenerate case where the floor would equal/exceed the - # window (small models), so auto-compression can still fire (#14690). - self.threshold_tokens = self._compute_threshold_tokens( - self.context_length, threshold_percent, self.max_tokens, - ) - # Apply absolute token cap (compression.threshold_tokens) — takes - # the lower of the ratio-based threshold and the cap. - self._apply_threshold_tokens_cap() + self._resolved_context_length: int | None = None + self._threshold_tokens: int | None = None + self._tail_token_budget: int | None = None + self._max_summary_tokens: int | None = None self.compression_count = 0 - # Derive token budgets: ratio is relative to the threshold, not total context - target_tokens = int(self.threshold_tokens * self.summary_target_ratio) - self.tail_token_budget = target_tokens - self.max_summary_tokens = min( - int(self.context_length * 0.05), _SUMMARY_TOKENS_CEILING, - ) - - if not quiet_mode: - logger.info( - "Context compressor initialized: model=%s context_length=%d " - "threshold=%d (%.0f%%) target_ratio=%.0f%% tail_budget=%d " - "provider=%s base_url=%s", - model, self.context_length, self.threshold_tokens, - threshold_percent * 100, self.summary_target_ratio * 100, - self.tail_token_budget, - provider or "none", base_url or "none", - ) + # The "initialized" log reports resolved token budgets, which would + # force the deferred get_model_context_length() probe to run inside + # __init__ and re-introduce the exact synchronous blocking this change + # removes (#32221). Emit it on first context-length resolution instead + # so construction stays non-blocking on every path (not just quiet). + self._log_init_summary = not quiet_mode self._context_probed = False # True after a step-down from context error self.last_prompt_tokens = 0 diff --git a/agent/context_references.py b/agent/context_references.py index 8981aa472f5..ab370a5a592 100644 --- a/agent/context_references.py +++ b/agent/context_references.py @@ -213,8 +213,12 @@ async def preprocess_context_references_async( f"@ context injection warning: {injected_tokens} tokens exceeds the 25% soft limit ({soft_limit})." ) - stripped = _remove_reference_tokens(message, refs) - final = stripped + # Leave the `@file:`/`@folder:` tokens where the user typed them. The token + # IS the reference, not scaffolding around it: clients render each one as an + # inline chip, so stripping them left a sentence with a hole in it ("review + # and ship") and made the desktop re-derive the refs from the attached block + # to show them as a detached list above the prose. + final = message if warnings: final = f"{final}\n\n--- Context Warnings ---\n" + "\n".join(f"- {warning}" for warning in warnings) if blocks: @@ -473,19 +477,6 @@ def _parse_file_reference_value(value: str) -> tuple[str, int | None, int | None return _strip_reference_wrappers(value), None, None -def _remove_reference_tokens(message: str, refs: list[ContextReference]) -> str: - pieces: list[str] = [] - cursor = 0 - for ref in refs: - pieces.append(message[cursor:ref.start]) - cursor = ref.end - pieces.append(message[cursor:]) - text = "".join(pieces) - text = re.sub(r"\s{2,}", " ", text) - text = re.sub(r"\s+([,.;:!?])", r"\1", text) - return text.strip() - - def _is_binary_file(path: Path) -> bool: mime, _ = mimetypes.guess_type(path.name) if mime and not mime.startswith("text/") and not any( diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index 506181f2d39..8308b618da7 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -2721,16 +2721,28 @@ def try_shrink_image_parts_in_messages( media_type = "image/jpeg" return f"data:{media_type};base64,{data}" - def _write_data_url_to_source(source: dict, data_url: str) -> None: + def _write_data_url_to_source(source: dict, data_url: str) -> dict: + """Return a NEW source dict carrying the re-encoded payload. + + Copy-on-write: content parts on the per-call ``api_messages`` list may + be shared references into the persistent conversation history (the + per-message copy is shallow, and cache decoration only deep-copies the + marked messages). Mutating the existing dict would rewrite the stored + transcript with the degraded image — so the caller replaces the part, + never edits it in place. + """ header, _, data = data_url.partition(",") media_type = "image/jpeg" if header.startswith("data:"): candidate = header[len("data:"):].split(";", 1)[0].strip() if candidate.startswith("image/"): media_type = candidate - source["type"] = "base64" - source["media_type"] = media_type - source["data"] = data + return { + **source, + "type": "base64", + "media_type": media_type, + "data": data, + } for msg in api_messages: if not isinstance(msg, dict): @@ -2738,7 +2750,13 @@ def try_shrink_image_parts_in_messages( content = msg.get("content") if not isinstance(content, list): continue - for part in content: + # Copy-on-write per message: never mutate part/source dicts in place — + # they can alias the stored conversation history (see + # _write_data_url_to_source). Build a replacement content list on the + # first shrunken part and reassign msg["content"] (a top-level write on + # the per-call message copy, which never reaches history). + new_content: list | None = None + for part_idx, part in enumerate(content): if not isinstance(part, dict): continue ptype = part.get("type") @@ -2747,7 +2765,12 @@ def try_shrink_image_parts_in_messages( url = _source_to_data_url(source) resized, unshrinkable = _shrink_data_url(url or "") if resized and isinstance(source, dict): - _write_data_url_to_source(source, resized) + if new_content is None: + new_content = list(content) + new_content[part_idx] = { + **part, + "source": _write_data_url_to_source(source, resized), + } changed_count += 1 elif unshrinkable: unshrinkable_oversized += 1 @@ -2761,17 +2784,26 @@ def try_shrink_image_parts_in_messages( url = image_value.get("url", "") resized, unshrinkable = _shrink_data_url(url) if resized: - image_value["url"] = resized + if new_content is None: + new_content = list(content) + new_content[part_idx] = { + **part, + "image_url": {**image_value, "url": resized}, + } changed_count += 1 elif unshrinkable: unshrinkable_oversized += 1 elif isinstance(image_value, str): resized, unshrinkable = _shrink_data_url(image_value) if resized: - part["image_url"] = resized + if new_content is None: + new_content = list(content) + new_content[part_idx] = {**part, "image_url": resized} changed_count += 1 elif unshrinkable: unshrinkable_oversized += 1 + if new_content is not None: + msg["content"] = new_content if changed_count: logger.info( diff --git a/agent/conversation_loop.py b/agent/conversation_loop.py index d03a2cfa3d2..06504616bac 100644 --- a/agent/conversation_loop.py +++ b/agent/conversation_loop.py @@ -140,6 +140,19 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text incomplete by definition; the model regenerates it on the retried turn. If a future path needs to preserve interrupted thinking, carry it in a provider-gated reasoning *field*, never in content. + INVARIANT — the scaffolding is provider-replay text, not transcript text. + ``[This response was interrupted by a user correction.]`` and its + ``Visible response before the interruption:`` header exist so the MODEL + understands its own reply was cut off. They are not prose the user wrote + or the agent said. Persisting them into ``content`` painted the raw + machinery as an assistant bubble on every reload (and merged it into the + preceding tool-call bubble), which is what made a steered transcript + unreadable. Carry the scaffolded form in the ``api_content`` sidecar -- + the exact bytes replayed to the provider -- and keep ``content`` clean. + When nothing was on screen there is no clean form at all, so the row is + marked ``display_kind="hidden"``: still replayed to the model, dropped by + every transcript surface (desktop, TUI, CLI resume), exactly like the + compaction-reference rows. """ visible = agent._strip_think_blocks( getattr(agent, "_current_streamed_assistant_text", "") or "" @@ -162,9 +175,22 @@ def _apply_active_turn_redirect(agent: Any, messages: List[Dict[str, Any]], text f"{checkpoint}\n\n" f"{text}" ) - messages.append({"role": "user", "content": correction}) + # Transcript shows the user's own words; the provider replays the + # scaffolded form so it still sees the interrupted context. + messages.append( + {"role": "user", "content": text, "api_content": correction} + ) else: - messages.append({"role": "assistant", "content": checkpoint}) + entry: Dict[str, Any] = { + "role": "assistant", + "content": visible or checkpoint, + "api_content": checkpoint, + } + if not visible: + # Nothing reached the screen — this row carries no assistant prose + # at all, only the cut-off notice for the model. + entry["display_kind"] = "hidden" + messages.append(entry) messages.append({"role": "user", "content": text}) agent._current_streamed_assistant_text = "" @@ -1067,6 +1093,8 @@ def run_conversation( stream_callback: Optional[callable] = None, persist_user_message: Optional[Any] = None, persist_user_timestamp: Optional[float] = None, + persist_user_display_kind: Optional[str] = None, + persist_user_display_metadata: Optional[Dict[str, Any]] = None, moa_config: Optional[dict[str, Any]] = None, ) -> Dict[str, Any]: """ @@ -1085,6 +1113,13 @@ def run_conversation( synthetic prefixes. persist_user_timestamp: Optional platform event timestamp to store as metadata on that persisted user message. + persist_user_display_kind: Optional presentation type for a + synthesized user turn (``auto_continue``, ``model_switch``, …). + Display-only: transcript surfaces render the row as a timeline + event instead of a user bubble, while the model still receives + the message unchanged. + persist_user_display_metadata: Optional payload for that event + (e.g. a delegation's task count). or queuing follow-up prefetch work. Returns: @@ -1127,6 +1162,8 @@ def run_conversation( stream_callback, persist_user_message, persist_user_timestamp, + persist_user_display_kind=persist_user_display_kind, + persist_user_display_metadata=persist_user_display_metadata, restore_or_build_system_prompt=_restore_or_build_system_prompt, install_safe_stdio=_install_safe_stdio, sanitize_surrogates=_sanitize_surrogates, @@ -2509,6 +2546,17 @@ def run_conversation( _backoff_touch_counter = 0 while time.time() < sleep_end: if agent._interrupt_requested: + # A redirect uses the interrupt machinery to cancel + # only the live request. Aborting the retry here + # with clear_interrupt() would DESTROY the pending + # correction and kill the turn with "Operation + # interrupted" — the exact mid-stream steer loss + # users hit when a redirect lands during provider + # backoff. Rebuild from the correction instead, + # mirroring the InterruptedError handler. + if agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + break agent._vprint(f"{agent.log_prefix}⚡ Interrupt detected during retry wait, aborting.", force=True) _interrupt_text = f"Operation interrupted during retry ({_failure_hint}, attempt {retry_count}/{max_retries})." close_interrupted_tool_sequence(messages, _interrupt_text) @@ -2530,6 +2578,8 @@ def run_conversation( f"retry backoff ({retry_count}/{max_retries}), " f"{int(sleep_end - time.time())}s remaining" ) + if _retry.restart_with_redirected_messages: + break # rebuild this iteration from the correction continue # Retry the API call agent._turn_received_provider_response = True @@ -3189,7 +3239,12 @@ def run_conversation( _cost_delta = (_cost_delta or 0.0) + float(_moa_ref_cost) except (TypeError, ValueError): # pragma: no cover pass - agent._session_db.update_token_counts( + # Enqueued, not written: the background writer + # applies the delta off the turn thread (a cold + # state.db UPDATE here stalled the tool loop for + # up to hundreds of ms per API call). Drained at + # turn finalize via _persist_session. + agent._session_db.queue_token_counts( agent.session_id, input_tokens=canonical_usage.input_tokens, output_tokens=canonical_usage.output_tokens, @@ -4028,6 +4083,12 @@ def run_conversation( # Check for interrupt before deciding to retry if agent._interrupt_requested: + # Preserve a pending redirect (mid-stream correction): the + # user is steering, not stopping. Rebuild the turn from the + # correction instead of aborting with a dead-end interrupt. + if agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + break agent._vprint(f"{agent.log_prefix}⚡ Interrupt detected during error handling, aborting retries.", force=True) _interrupt_text = f"Operation interrupted: handling API error ({error_type}: {agent._clean_error_message(str(api_error))})." close_interrupted_tool_sequence(messages, _interrupt_text) @@ -5264,6 +5325,12 @@ def run_conversation( _backoff_touch_counter = 0 while time.time() < sleep_end: if agent._interrupt_requested: + # Same preserve-redirect rule as the retry-wait above: + # a steering correction must survive backoff, not die + # as "Operation interrupted". + if agent.clear_interrupt(preserve_redirect=True): + _retry.restart_with_redirected_messages = True + break agent._vprint(f"{agent.log_prefix}⚡ Interrupt detected during retry wait, aborting.", force=True) _interrupt_text = f"Operation interrupted: retrying API call after error (retry {retry_count}/{max_retries})." close_interrupted_tool_sequence(messages, _interrupt_text) @@ -5285,6 +5352,11 @@ def run_conversation( f"error retry backoff ({retry_count}/{max_retries}), " f"{int(sleep_end - time.time())}s remaining" ) + if _retry.restart_with_redirected_messages: + # Leave the retry loop — the check right below rebuilds this + # iteration from the correction instead of re-firing the + # stale request. + break if _retry.restart_with_redirected_messages: # The cancelled request produced no valid assistant item. Reuse the diff --git a/agent/insights.py b/agent/insights.py index 086150c279e..9d148a15446 100644 --- a/agent/insights.py +++ b/agent/insights.py @@ -113,6 +113,13 @@ class InsightsEngine: """ cutoff = time.time() - (days * 86400) + # Token/cost totals may still sit on the SessionDB's async + # accounting queue; drain so the report reflects exact counters. + # (self.db may be a raw sqlite3 connection in tests — guard.) + flush = getattr(self.db, "flush_token_counts", None) + if callable(flush): + flush() + # Gather raw data sessions = self._get_sessions(cutoff, source) tool_usage = self._get_tool_usage(cutoff, source) diff --git a/agent/prompt_caching.py b/agent/prompt_caching.py index 2e606cd377c..1a0326c79df 100644 --- a/agent/prompt_caching.py +++ b/agent/prompt_caching.py @@ -187,17 +187,18 @@ def apply_anthropic_cache_control( is retained. Returns: - Deep copy of messages with cache_control breakpoints injected. + Shallow copy of message list with selective deep copies of modified messages. """ - messages = copy.deepcopy(api_messages) - if not messages: - return messages + if not api_messages: + return api_messages + messages = list(api_messages) marker = _build_marker(cache_ttl) breakpoints_used = 0 if messages[0].get("role") == "system": + messages[0] = copy.deepcopy(messages[0]) breakpoints_used = _apply_system_cache_markers( messages[0], marker, @@ -213,6 +214,7 @@ def apply_anthropic_cache_control( and _can_carry_marker(messages[i], native_anthropic=native_anthropic) ] for idx in non_sys[-remaining:]: + messages[idx] = copy.deepcopy(messages[idx]) _apply_cache_marker(messages[idx], marker, native_anthropic=native_anthropic) return messages diff --git a/agent/reasoning_timeouts.py b/agent/reasoning_timeouts.py index 9c5fc202015..da7fcd2fcbe 100644 --- a/agent/reasoning_timeouts.py +++ b/agent/reasoning_timeouts.py @@ -146,19 +146,18 @@ _REASONING_STALE_TIMEOUT_FLOORS: tuple[tuple[str, int], ...] = ( # so we accept that community forks inheriting the same prefix are # treated as reasoning models (a reasonable default — the upstream # gateway timing is the same). -_PATTERN_CACHE: dict[str, re.Pattern[str]] = {} - - -def _get_pattern(slug: str) -> re.Pattern[str]: - compiled = _PATTERN_CACHE.get(slug) - if compiled is None: - compiled = re.compile( - r"^" - + re.escape(slug) - + r"(?:$|[\-._])" - ) - _PATTERN_CACHE[slug] = compiled - return compiled +# Pre-compile all patterns at module load time to avoid per-call regex +# compilation and thread-safety issues with the mutable _PATTERN_CACHE. +# The list is built once at import and never mutated afterwards, so it is +# safe for free-threaded Python 3.13+ without any locking. The slug is kept +# in each entry for debuggability (log/inspection), even though _match_any +# only consumes floor + pattern. +_SORTED_REASONING_FLOORS: list[tuple[str, float, re.Pattern[str]]] = [ + (slug, floor, re.compile(r"^" + re.escape(slug) + r"(?:$|[\-._])")) + for slug, floor in sorted( + _REASONING_STALE_TIMEOUT_FLOORS, key=lambda kv: -len(kv[0]) + ) +] def _match_any(model_lower: str) -> Optional[float]: @@ -169,13 +168,8 @@ def _match_any(model_lower: str) -> Optional[float]: order is irrelevant: longest slug wins (so ``o3-mini`` beats ``o3`` on a model like ``openai/o3-mini``). """ - # Sort by slug length descending so longer / more-specific slugs - # win on shared prefixes (o3-mini beats o3). - sorted_floors = sorted( - _REASONING_STALE_TIMEOUT_FLOORS, key=lambda kv: -len(kv[0]) - ) - for slug, floor in sorted_floors: - if _get_pattern(slug).search(model_lower): + for _slug, floor, pattern in _SORTED_REASONING_FLOORS: + if pattern.search(model_lower): return float(floor) return None diff --git a/agent/relay_llm.py b/agent/relay_llm.py index 3e3f54d4dfd..4508fce9414 100644 --- a/agent/relay_llm.py +++ b/agent/relay_llm.py @@ -332,6 +332,7 @@ class ManagedLlmStream(Iterator[Any]): self._stream: Any = None self._raw_stream_resource: Any = None self._closed = False + self._close_error: BaseException | None = None self._callback_error: BaseException | None = None self._logical: tuple[relay_runtime.RelayTurnContext, Any, str] | None = None self._defer_logical_completion = defer_logical_completion @@ -426,7 +427,11 @@ class ManagedLlmStream(Iterator[Any]): finally: close = getattr(raw_stream, "close", None) if callable(close): - run_callback(close) + try: + run_callback(close) + except BaseException as exc: + self._close_error = exc + raise def observe_chunk(chunk: Any) -> None: if self._on_chunk is not None: @@ -494,10 +499,10 @@ class ManagedLlmStream(Iterator[Any]): try: chunk = next(self._stream) except StopIteration: - self.close() + self._close(logical_outcome="cancelled") raise if self._accept_chunk is not None and not self._accept_chunk(chunk): - self.close() + self._close(logical_outcome="cancelled") raise StopIteration return chunk @@ -512,7 +517,7 @@ class ManagedLlmStream(Iterator[Any]): if not self._defer_logical_completion: _complete_logical(self._logical, outcome="success") self._logical = None - self.close() + self._close(logical_outcome="cancelled") raise StopIteration from None except BaseException as exc: callback_error = self._callback_error @@ -552,6 +557,10 @@ class ManagedLlmStream(Iterator[Any]): def close(self) -> None: """Close an explicitly abandoned stream and cancel its logical call.""" self._close(logical_outcome="cancelled") + close_error = self._close_error + self._close_error = None + if close_error is not None: + raise close_error def _preserve_pending_provider_chunks(self) -> None: """Switch a failed Relay stream to its undelivered provider chunks.""" @@ -601,7 +610,9 @@ class ManagedLlmStream(Iterator[Any]): if callable(close): try: close() - except Exception: + except Exception as exc: + if self._close_error is None: + self._close_error = exc logger.debug( "Provider stream cleanup failed", exc_info=True, @@ -618,15 +629,16 @@ class ManagedLlmStream(Iterator[Any]): try: loop.run_until_complete(close_stream()) - except Exception: - pass + except Exception as exc: + if self._close_error is None: + self._close_error = exc if not self._defer_logical_completion: _complete_logical(self._logical, outcome=logical_outcome) self._logical = None loop.close() def __del__(self) -> None: - self.close() + self._close(logical_outcome="cancelled") class AnthropicStreamAccumulator: diff --git a/agent/skill_commands.py b/agent/skill_commands.py index 294ca2b1754..3f1156a8592 100644 --- a/agent/skill_commands.py +++ b/agent/skill_commands.py @@ -97,7 +97,7 @@ def extract_user_instruction_from_skill_message(content: Any) -> Optional[str]: return None -def describe_skill_invocation(content: Any) -> Optional[str]: +def describe_skill_invocation(content: Any, separator: str = " — ") -> Optional[str]: """Render a slash-skill-expanded turn the way the user typed it. The expanded message embeds the whole skill body, so any surface that @@ -109,6 +109,10 @@ def describe_skill_invocation(content: Any) -> Optional[str]: Returns ``"/work — fix the title leak"``, or ``"/work"`` for a bare invocation, or ``None`` when *content* is not skill scaffolding (the caller should then summarize it as an ordinary message). + + *separator* joins the command and the instruction. Previews use the + default em dash; pass ``" "`` for the literal invocation the user typed, + which is what chat transcripts render. """ if not isinstance(content, str) or not content.startswith(_SKILL_INVOCATION_PREFIX): return None @@ -127,7 +131,7 @@ def describe_skill_invocation(content: Any) -> Optional[str]: instruction = instruction.split(SKILL_EXCERPT_JOINT)[0] instruction = " ".join(instruction.split()) if instruction: - return f"{label} — {instruction}" if name else instruction + return f"{label}{separator}{instruction}" if name else instruction return label if name else None diff --git a/agent/turn_context.py b/agent/turn_context.py index 5a006894887..e080d6a5d96 100644 --- a/agent/turn_context.py +++ b/agent/turn_context.py @@ -336,6 +336,8 @@ def build_turn_context( persist_user_message: Optional[Any], persist_user_timestamp: Optional[float] = None, *, + persist_user_display_kind: Optional[str] = None, + persist_user_display_metadata: Optional[Dict[str, Any]] = None, restore_or_build_system_prompt, install_safe_stdio, sanitize_surrogates, @@ -542,6 +544,19 @@ def build_turn_context( # Add the current user message after the prompt/session setup has made # close persistence safe. The handoff above preserves any marker already # stamped by an earlier close flush. + # + # A synthesized turn (auto-continue recovery note, delegation completion) + # declares how it should READ in a transcript. Stamp that on the live + # message so the crash persist below writes the row already typed. Typing + # it after the turn instead leaves the row untyped for the whole run — and + # forever if the turn crashes — so the raw system note paints as a user + # bubble. The model still receives role/content unchanged; the api_messages + # build strips both fields from every outgoing copy. + if persist_user_display_kind: + user_msg["display_kind"] = persist_user_display_kind + if persist_user_display_metadata: + user_msg["display_metadata"] = persist_user_display_metadata + messages.append(user_msg) current_turn_user_idx = len(messages) - 1 agent._persist_user_message_idx = current_turn_user_idx diff --git a/apps/desktop/e2e/fixtures.ts b/apps/desktop/e2e/fixtures.ts index 3427d42c9bf..fee8e419703 100644 --- a/apps/desktop/e2e/fixtures.ts +++ b/apps/desktop/e2e/fixtures.ts @@ -120,6 +120,17 @@ export function createSandbox(prefix: string): Sandbox { 'utf8', ) + // Pin Chromium actual-size zoom (level 0) for the suite. Fresh installs + // ship DEFAULT_ZOOM_LEVEL at the Appearance 90% preset, but Playwright + // click hit-testing and the committed visual baselines were calibrated at + // 100%. Without this file every sandbox would inherit the product default + // and fail pointer interception + snapshot diffs. + fs.writeFileSync( + path.join(userDataDir, 'zoom-state.json'), + JSON.stringify({ zoomLevel: 0 }, null, 2), + 'utf8', + ) + return { root, hermesHome, diff --git a/apps/desktop/electron/dev-cdp.test.ts b/apps/desktop/electron/dev-cdp.test.ts new file mode 100644 index 00000000000..f86aa9db304 --- /dev/null +++ b/apps/desktop/electron/dev-cdp.test.ts @@ -0,0 +1,94 @@ +/** + * Tests for electron/dev-cdp.ts. + * + * Run with: npx vitest run --project electron electron/dev-cdp.test.ts + */ + +import assert from 'node:assert/strict' + +import { test } from 'vitest' + +import { DEFAULT_PORT, describeDevCdpDecision, resolveDevCdpPort } from './dev-cdp' + +const DEV_SERVER = 'http://127.0.0.1:5174' + +/** The ordinary `npm run dev` / `hgui` run. */ +const devRun = { env: {}, isPackaged: false, devServer: DEV_SERVER } + +test('a dev-server run opens the default port with no opt-in', () => { + assert.deepEqual(resolveDevCdpPort(devRun), { port: DEFAULT_PORT, reason: null }) +}) + +test('the default matches what the scripts/ tooling reaches for', () => { + // scripts/eval.mjs and scripts/perf/lib/cdp.mjs both default here; if this + // drifts, `node scripts/eval.mjs ...` stops finding a live renderer. + assert.equal(DEFAULT_PORT, 9222) +}) + +test('a packaged build never opens the port, however loudly the env asks', () => { + const decision = resolveDevCdpPort({ ...devRun, env: { HERMES_DESKTOP_CDP_PORT: '9222' }, isPackaged: true }) + + assert.deepEqual(decision, { port: null, reason: 'packaged' }) +}) + +test('packaged is checked before every other gate', () => { + // Belt-and-suspenders: dev server present, valid port requested, still shut. + for (const value of ['9222', '', 'off', 'garbage']) { + const decision = resolveDevCdpPort({ + env: { HERMES_DESKTOP_CDP_PORT: value }, + isPackaged: true, + devServer: DEV_SERVER + }) + + assert.equal(decision.port, null, `expected packaged to refuse ${JSON.stringify(value)}`) + assert.equal(decision.reason, 'packaged') + } +}) + +test('an unpackaged dist run (no dev server) does not qualify', () => { + // `electron .` against dist/ is how the packaged app gets smoke tested; it + // should behave like the packaged app, not like a source-tree dev run. + assert.deepEqual(resolveDevCdpPort({ ...devRun, devServer: undefined }), { port: null, reason: 'no-dev-server' }) +}) + +test('the port is overridable', () => { + assert.equal(resolveDevCdpPort({ ...devRun, env: { HERMES_DESKTOP_CDP_PORT: '9333' } }).port, 9333) +}) + +test('tolerates surrounding whitespace on the override', () => { + assert.equal(resolveDevCdpPort({ ...devRun, env: { HERMES_DESKTOP_CDP_PORT: ' 9333 ' } }).port, 9333) +}) + +test('can be switched off on a dev run', () => { + for (const value of ['0', 'off', 'OFF', 'false', 'no']) { + const decision = resolveDevCdpPort({ ...devRun, env: { HERMES_DESKTOP_CDP_PORT: value } }) + + assert.equal(decision.port, null, `expected ${JSON.stringify(value)} to close the port`) + assert.equal(decision.reason, 'opted-out') + } +}) + +test('refuses ports that are not usable integers', () => { + for (const value of ['80', '-1', '70000', 'yes', '9222.5', '92 22']) { + const decision = resolveDevCdpPort({ ...devRun, env: { HERMES_DESKTOP_CDP_PORT: value } }) + + assert.equal(decision.port, null, `expected ${JSON.stringify(value)} to be refused`) + assert.equal(decision.reason, 'invalid-port') + } +}) + +test('explains itself when an explicit setting was not honoured', () => { + // A typo'd port or a deliberate opt-out should say so — silently doing + // something other than what the env asked for is the bad failure mode. + for (const value of ['garbage', 'off']) { + const decision = resolveDevCdpPort({ ...devRun, env: { HERMES_DESKTOP_CDP_PORT: value } }) + + assert.ok(describeDevCdpDecision(decision), `expected an explanation for ${JSON.stringify(value)}`) + } +}) + +test('stays quiet when the port opened, or is closed by design', () => { + assert.equal(describeDevCdpDecision(resolveDevCdpPort(devRun)), null) + assert.equal(describeDevCdpDecision(resolveDevCdpPort({ ...devRun, isPackaged: true })), null) + assert.equal(describeDevCdpDecision(resolveDevCdpPort({ ...devRun, devServer: undefined })), null) +}) diff --git a/apps/desktop/electron/dev-cdp.ts b/apps/desktop/electron/dev-cdp.ts new file mode 100644 index 00000000000..c5c4e3ef353 --- /dev/null +++ b/apps/desktop/electron/dev-cdp.ts @@ -0,0 +1,108 @@ +/** + * Dev Chrome DevTools Protocol exposure for the desktop renderer. + * + * The renderer is a Chromium page, so `--remote-debugging-port` turns it into + * something the repo's existing CDP tooling (`scripts/eval.mjs`, + * `scripts/perf/lib/cdp.mjs`, the `diag-*` / `probe-*` family) can attach to + * and read the live DOM from. Every one of those scripts already defaults to + * 9222, so a dev-server run opens 9222 and they just work. + * + * If you are running a dev server you are already executing arbitrary local + * JS — vite's module graph and every postinstall in node_modules — so a + * loopback debugging port does not meaningfully widen that. `perf:serve` + * already opens one unconditionally. What must never happen is a *packaged* + * app exposing it, which is the one hard gate here. + * + * - packaged build → always closed, whatever the env says. + * - no HERMES_DESKTOP_DEV_SERVER → closed (an unpackaged `electron .` against + * dist/ is how the packaged app gets smoke tested; it should behave like + * the packaged app). + * - otherwise → open on 9222, or HERMES_DESKTOP_CDP_PORT. + * + * `HERMES_DESKTOP_CDP_PORT=off` (or `0` / `false`) opts out for anyone who + * wants the port closed on a dev run. + * + * The port binds to loopback (Chromium's default) and the address is + * deliberately not configurable: there is no reason to expose a renderer + * debugger off-host, and offering the knob invites someone to try. + */ + +/** Why the port is closed, for a one-line log the developer can act on. */ +type ClosedReason = 'packaged' | 'no-dev-server' | 'opted-out' | 'invalid-port' + +type DevCdpDecision = { port: number; reason: null } | { port: null; reason: ClosedReason } + +type DevCdpInput = { + env: Record + isPackaged: boolean + devServer: string | undefined +} + +/** What every script under scripts/ already reaches for. */ +const DEFAULT_PORT = 9222 + +// Below 1024 needs privileges on most platforms; 65535 is the ceiling. +const MIN_PORT = 1024 +const MAX_PORT = 65535 + +const OPT_OUT = new Set(['0', 'off', 'false', 'no']) + +/** + * Decide whether this run may expose a renderer debugging port, and on which + * port. Pure: every input is passed in, so the gate is testable without an + * Electron app or a real environment. + */ +function resolveDevCdpPort({ env, isPackaged, devServer }: DevCdpInput): DevCdpDecision { + // Packaged wins over everything. Checked first so no combination of + // environment variables can talk a shipped build into opening the port. + if (isPackaged) { + return { port: null, reason: 'packaged' } + } + + // A dev server means a source-tree run (`npm run dev` / `hgui`). + if (!devServer) { + return { port: null, reason: 'no-dev-server' } + } + + const requested = (env.HERMES_DESKTOP_CDP_PORT ?? '').trim() + + if (!requested) { + return { port: DEFAULT_PORT, reason: null } + } + + if (OPT_OUT.has(requested.toLowerCase())) { + return { port: null, reason: 'opted-out' } + } + + const port = Number(requested) + + if (!Number.isInteger(port) || port < MIN_PORT || port > MAX_PORT) { + return { port: null, reason: 'invalid-port' } + } + + return { port, reason: null } +} + +/** One-line explanation for a closed port, or null when it opened. */ +function describeDevCdpDecision(decision: DevCdpDecision): string | null { + switch (decision.reason) { + case null: + return null + + case 'invalid-port': + return `HERMES_DESKTOP_CDP_PORT is not a valid port (expected an integer ${MIN_PORT}-${MAX_PORT}, or "off"); renderer debugging is disabled.` + + case 'opted-out': + return 'renderer debugging disabled by HERMES_DESKTOP_CDP_PORT.' + + // Packaged and dist-run builds are closed by design — the common case, not + // worth a line of startup noise. + case 'packaged': + + case 'no-dev-server': + return null + } +} + +export { DEFAULT_PORT, describeDevCdpDecision, resolveDevCdpPort } +export type { DevCdpDecision } diff --git a/apps/desktop/electron/hardening.test.ts b/apps/desktop/electron/hardening.test.ts index 1a5852f720a..4b41962e0cd 100644 --- a/apps/desktop/electron/hardening.test.ts +++ b/apps/desktop/electron/hardening.test.ts @@ -7,6 +7,9 @@ import { pathToFileURL } from 'node:url' import { test } from 'vitest' import { + clampDataUrlReadMaxMb, + DATA_URL_READ_DEFAULT_MAX_MB, + dataUrlReadMaxBytesFromMb, DEFAULT_FETCH_TIMEOUT_MS, encryptDesktopSecret, resolveDirectoryForIpc, @@ -24,6 +27,14 @@ async function rejectsWithCode(promise, code: string) { }) } +test('clampDataUrlReadMaxMb defaults and bounds the attach size preference', () => { + assert.equal(clampDataUrlReadMaxMb(undefined), DATA_URL_READ_DEFAULT_MAX_MB) + assert.equal(clampDataUrlReadMaxMb(0), 1) + assert.equal(clampDataUrlReadMaxMb(256), 256) + assert.equal(clampDataUrlReadMaxMb(99999), 4096) + assert.equal(dataUrlReadMaxBytesFromMb(16), 16 * 1024 * 1024) +}) + test('resolveTimeoutMs falls back to defaults and accepts overrides', () => { assert.equal(resolveTimeoutMs(undefined), DEFAULT_FETCH_TIMEOUT_MS) assert.equal(resolveTimeoutMs(0), DEFAULT_FETCH_TIMEOUT_MS) diff --git a/apps/desktop/electron/hardening.ts b/apps/desktop/electron/hardening.ts index 2d6b5331001..de15a5dca5e 100644 --- a/apps/desktop/electron/hardening.ts +++ b/apps/desktop/electron/hardening.ts @@ -4,9 +4,30 @@ import path from 'node:path' import { fileURLToPath } from 'node:url' const DEFAULT_FETCH_TIMEOUT_MS = 15_000 -const DATA_URL_READ_MAX_BYTES = 16 * 1024 * 1024 +// Default / floor / ceiling for Desktop's data-URL file load (composer attach, +// image preview, etc.). The whole file is base64-buffered in main, so this is +// a memory guard — not a model limit. Settings → Chat takes a free-form MB +// value; 16 MB ships as default. The ceiling is only a typo guard (very large +// values can OOM / crash the app). +const DATA_URL_READ_DEFAULT_MAX_MB = 16 +const DATA_URL_READ_MIN_MAX_MB = 1 +const DATA_URL_READ_MAX_MAX_MB = 4096 const TEXT_PREVIEW_SOURCE_MAX_BYTES = 64 * 1024 * 1024 +function clampDataUrlReadMaxMb(value) { + const parsed = Number(value) + + if (!Number.isFinite(parsed)) { + return DATA_URL_READ_DEFAULT_MAX_MB + } + + return Math.min(DATA_URL_READ_MAX_MAX_MB, Math.max(DATA_URL_READ_MIN_MAX_MB, Math.round(parsed))) +} + +function dataUrlReadMaxBytesFromMb(maxMb) { + return clampDataUrlReadMaxMb(maxMb) * 1024 * 1024 +} + const SAFE_ENV_SUFFIXES = new Set(['dist', 'example', 'sample', 'template']) const SENSITIVE_EXTENSIONS = new Set(['.kdbx', '.p12', '.pem', '.pfx']) @@ -304,7 +325,11 @@ async function resolveReadableFileForIpc( } export { - DATA_URL_READ_MAX_BYTES, + clampDataUrlReadMaxMb, + DATA_URL_READ_DEFAULT_MAX_MB, + DATA_URL_READ_MAX_MAX_MB, + DATA_URL_READ_MIN_MAX_MB, + dataUrlReadMaxBytesFromMb, DEFAULT_FETCH_TIMEOUT_MS, encryptDesktopSecret, rejectUnsafePathSyntax, diff --git a/apps/desktop/electron/main.ts b/apps/desktop/electron/main.ts index 7ac24c57f8e..dfb4747f039 100644 --- a/apps/desktop/electron/main.ts +++ b/apps/desktop/electron/main.ts @@ -81,6 +81,7 @@ import { shouldRemoveAppBundle, uninstallArgsForMode } from './desktop-uninstall' +import { describeDevCdpDecision, resolveDevCdpPort } from './dev-cdp' import { installEmbedReferer } from './embed-referer' import { createEventDeduper } from './event-dedupe' import { findGitBash as _findGitBash } from './find-git-bash' @@ -114,7 +115,9 @@ import { switchBranch } from './git-worktree-ops' import { - DATA_URL_READ_MAX_BYTES, + clampDataUrlReadMaxMb, + DATA_URL_READ_DEFAULT_MAX_MB, + dataUrlReadMaxBytesFromMb, DEFAULT_FETCH_TIMEOUT_MS, encryptDesktopSecret as encryptDesktopSecretStrict, resolveReadableFileForIpc, @@ -272,6 +275,29 @@ if (REMOTE_DISPLAY_REASON) { ) } +// Renderer debugging port. On for dev-server runs (`hgui` / `npm run dev`) so +// the CDP tooling in scripts/ can attach; never for a packaged build — see +// electron/dev-cdp.ts. Must run before app `ready` like the switches above; +// Chromium binds it at launch. +const DEV_CDP = resolveDevCdpPort({ env: process.env, isPackaged: IS_PACKAGED, devServer: DEV_SERVER }) + +if (DEV_CDP.port) { + app.commandLine.appendSwitch('remote-debugging-port', String(DEV_CDP.port)) + // Loopback only. Chromium already defaults to 127.0.0.1, but say it out loud + // so a future edit can't widen it by omission. + app.commandLine.appendSwitch('remote-debugging-address', '127.0.0.1') + console.log( + `[hermes] renderer debugging on http://127.0.0.1:${DEV_CDP.port} — anything that can reach it ` + + 'can run code in the renderer. HERMES_DESKTOP_CDP_PORT=off to disable.' + ) +} else { + const why = describeDevCdpDecision(DEV_CDP) + + if (why) { + console.warn(`[hermes] ${why}`) + } +} + // WSLg: Chromium blocklists the Mesa vGPU → software compositing → typing lag. // /dev/dxg means a real GPU is available; un-blocklist it. Skipped when a remote // display already forced software (SSH'd-into-WSL). @@ -936,7 +962,7 @@ app.setAboutPanelOptions({ // Custom scheme for streaming local media (video/audio) into the renderer. // Reading large media through `readFileDataUrl` failed: it base64-loads the -// whole file into memory and is hard-capped at DATA_URL_READ_MAX_BYTES (16 MB), +// whole file into memory and is hard-capped (default 16 MB, Settings → Chat), // so any non-trivial video silently refused to load. Streaming via a protocol // handler removes the size cap and gives the