refactor(agent): compact prompt_builder siblings — collapse guard ladders, inline single-use helpers, trim docstrings (emitted text byte-identical)

This commit is contained in:
Teknium
2026-09-02 18:22:01 -07:00
parent 62c7da915c
commit afce1b22db
7 changed files with 140 additions and 250 deletions
+14 -16
View File
@@ -10,11 +10,10 @@ from agent.redact import redact_sensitive_text
def describe_compression_lock_skip(lock_signal: Any) -> str:
"""User-facing text for a manual /compress skipped by the compression lock.
``lock_signal`` is ``agent._compression_skipped_due_to_lock`` (or the TUI's
``CompressionLockHeld.holder``): a holder string when another compressor
CONFIRMED holds the lock, else ``True``/``None``. A failed acquire is NOT
proof another compression is running (``try_acquire_compression_lock``
swallows ``sqlite3.Error``), so the two cases are worded differently.
``lock_signal`` is a holder string when another compressor CONFIRMED holds
the lock, else ``True``/``None``. A failed acquire is NOT proof another
compression is running (``try_acquire_compression_lock`` swallows
``sqlite3.Error``), so the two cases are worded differently.
"""
if isinstance(lock_signal, str) and lock_signal.strip():
return (
@@ -28,10 +27,6 @@ def describe_compression_lock_skip(lock_signal: Any) -> str:
)
def _state_flag(state: Any, name: str) -> bool:
return state is not None and getattr(state, name, False) is True
def summarize_manual_compression(
before_messages: Sequence[dict[str, Any]],
after_messages: Sequence[dict[str, Any]],
@@ -40,14 +35,18 @@ def summarize_manual_compression(
*,
compression_state: Any = None,
) -> dict[str, Any]:
"""Return consistent user-facing feedback for manual compression."""
"""Consistent user-facing feedback (headline, token line, optional note) for manual compression."""
before_count = len(before_messages)
after_count = len(after_messages)
noop = list(after_messages) == list(before_messages)
aborted = _state_flag(compression_state, "_last_compress_aborted")
refused_would_grow = _state_flag(compression_state, "_last_compress_refused_would_grow")
fallback_used = _state_flag(compression_state, "_last_summary_fallback_used")
failure_reason = getattr(compression_state, "_last_summary_error", None) if compression_state is not None else None
def flag(name: str) -> bool:
return getattr(compression_state, name, False) is True
aborted = flag("_last_compress_aborted")
refused_would_grow = flag("_last_compress_refused_would_grow")
fallback_used = flag("_last_summary_fallback_used")
failure_reason = getattr(compression_state, "_last_summary_error", None)
if not isinstance(failure_reason, str) or not failure_reason.strip():
failure_reason = None
@@ -85,8 +84,7 @@ def summarize_manual_compression(
if failure_reason and (aborted or fallback_used):
# Crosses a user-facing UI boundary: never let a disabled global redaction
# preference expose credentials embedded in provider exception text.
safe_reason = redact_sensitive_text(failure_reason.strip(), force=True)
note = f"{note} Reason: {safe_reason}"
note = f"{note} Reason: {redact_sensitive_text(failure_reason.strip(), force=True)}"
return {
"noop": noop,
+18 -33
View File
@@ -2,14 +2,11 @@
Models pad tables assuming one cell per character; CJK glyphs and most emoji
take two, so body rows drift right on real terminals. This rebuilds padding
with ``wcwidth.wcswidth`` while preserving pipes/dashes so the table still reads
as plain text in ``strip``/unrendered modes (Rich already aligns CJK itself).
with ``wcwidth.wcswidth`` while preserving pipes/dashes (Rich already aligns CJK).
Deliberately conservative: only contiguous ``| ... |`` blocks with a divider are
rewritten; everything else passes through; single-line/mid-stream fragments are
left alone (callers buffer rows and flush complete blocks). ``wcwidth`` returns
``-1`` for some emoji+variation-selector sequences (``⚠️``); those clamp to 0 —
a 1-cell drift on that glyph beats widening every table that contains one.
rewritten; single-line/mid-stream fragments pass through (callers buffer rows and
flush complete blocks). ``wcswidth`` returns ``-1`` for some emoji+variation-selector
sequences (``⚠️``); those clamp to 0 — a 1-cell drift beats widening every table.
"""
from __future__ import annotations
@@ -33,8 +30,7 @@ _MIN_COL_WIDTH = 3 # matches the divider's minimum dash run.
def _disp_width(s: str) -> int:
"""``wcswidth`` clamped to >= 0 (it returns -1 for control/unknown sequences)."""
w = wcswidth(s)
return w if w > 0 else 0
return max(wcswidth(s), 0)
def _pad_to_width(s: str, target: int) -> str:
@@ -43,11 +39,7 @@ def _pad_to_width(s: str, target: int) -> str:
def split_table_row(row: str) -> List[str]:
"""Split ``| a | b | c |`` into ``["a", "b", "c"]`` with trims."""
s = row.strip()
if s.startswith("|"):
s = s[1:]
if s.endswith("|"):
s = s[:-1]
s = row.strip().removeprefix("|").removesuffix("|")
return [c.strip() for c in s.split("|")]
@@ -61,16 +53,12 @@ def looks_like_table_row(row: str) -> bool:
"""True when ``row`` could plausibly be a markdown table row.
Intentionally permissive for streaming callers deciding whether to buffer a
line: the realigner only rewrites divider-backed blocks, so a false positive
at most delays printing one line. A leading pipe is the strongest signal;
without it we accept >= 2 pipes so models that omit the leading pipe still match.
line (a false positive at most delays printing one line). A leading pipe is
the strongest signal; without it we accept >= 2 pipes so models that omit
the leading pipe still match.
"""
if "|" not in row:
return False
stripped = row.strip()
if not stripped:
return False
return stripped.startswith("|") or stripped.count("|") >= 2
return bool(stripped) and (stripped.startswith("|") or stripped.count("|") >= 2)
def _render_block(rows: List[List[str]], available_width: int | None = None) -> List[str]:
@@ -106,8 +94,7 @@ def _hard_break(word: str, w: int) -> List[str]:
cw = _disp_width(ch) or 1
if bw + cw > w and buf:
out.append(buf)
buf = ch
bw = cw
buf, bw = ch, cw
else:
buf += ch
bw += cw
@@ -119,7 +106,7 @@ def _hard_break(word: str, w: int) -> List[str]:
def _wrap_to_width(text: str, width: int) -> List[str]:
"""Soft-wrap ``text`` at word boundaries to ``width`` display cells.
Words wider than ``width`` are hard-broken. Empty input yields a single
Words wider than ``width`` are hard-broken. Empty input yields a single
empty string so the caller's row count stays predictable.
"""
if width <= 0 or not text:
@@ -167,8 +154,7 @@ def _render_vertical(rows: List[List[str]], ncols: int, available_width: int) ->
return []
headers = rows[0] + [""] * (ncols - len(rows[0]))
labels = [h or f"Column {i + 1}" for i, h in enumerate(headers)]
sep_width = max(20, min(40, available_width - 2)) if available_width else 30
separator = "─" * sep_width
separator = "─" * (max(20, min(40, available_width - 2)) if available_width else 30)
indent = " "
cont_budget = max(10, available_width - _disp_width(indent))
@@ -184,11 +170,10 @@ def _render_vertical(rows: List[List[str]], ncols: int, available_width: int) ->
continue
wrapped = _wrap_to_width(value, max(10, available_width - _disp_width(label) - 2))
out.append(f"{label}: {wrapped[0]}")
if len(wrapped) > 1:
# Re-flow continuation text at the wider continuation budget.
for cl in _wrap_to_width(" ".join(wrapped[1:]), cont_budget):
if cl.strip():
out.append(f"{indent}{cl}")
# Re-flow continuation text at the wider continuation budget.
for cl in _wrap_to_width(" ".join(wrapped[1:]), cont_budget) if len(wrapped) > 1 else ():
if cl.strip():
out.append(f"{indent}{cl}")
return out
@@ -217,7 +202,7 @@ def realign_markdown_tables(text: str, available_width: int | None = None) -> st
if not is_table_divider(lines[j]):
body.append(split_table_row(lines[j]))
j += 1
if any(c for c in header) or body:
if any(header) or body:
out.extend(_render_block([header] + body, available_width))
i = j
continue
+17 -32
View File
@@ -1,9 +1,8 @@
"""Contextual first-touch onboarding hints.
Each hint is shown once per install the *first* time a user hits a behavior
fork (message-while-running, first long tool, ...), tracked in ``config.yaml``
under ``onboarding.seen.<flag>``. Kept tiny and dependency-free so both the CLI
and gateway can import it.
fork, tracked in ``config.yaml`` under ``onboarding.seen.<flag>``. Kept tiny and
dependency-free so both the CLI and gateway can import it.
"""
from __future__ import annotations
@@ -22,10 +21,8 @@ OPENCLAW_RESIDUE_FLAG = "openclaw_residue_cleanup"
PROFILE_BUILD_FLAG = "profile_build_offered"
# ── Hint content ──────────────────────────────────────────────────────────
# Busy-input hints are keyed by the effective busy_input_mode that was just
# applied so the message matches reality; "interrupt" is the default branch.
_BUSY_INPUT_HINTS_GATEWAY = {
"queue": (
"💡 First-time tip — I queued your message instead of interrupting. "
@@ -115,36 +112,33 @@ def openclaw_residue_hint_cli() -> str:
def detect_openclaw_residue(home: Optional[Path] = None) -> bool:
"""True if ``$HOME/.openclaw`` is a directory (pure check; ``home`` override for tests)."""
base = home or Path.home()
"""True if ``$HOME/.openclaw`` is a directory (``home`` override for tests)."""
try:
return (base / ".openclaw").is_dir()
return ((home or Path.home()) / ".openclaw").is_dir()
except OSError:
return False
# ── Onboarding profile-build path (opt-in, consent-gated) ─────────────────
def _onboarding_section(config: Mapping[str, Any]) -> Mapping[str, Any]:
onboarding = config.get("onboarding") if isinstance(config, Mapping) else None
return onboarding if isinstance(onboarding, Mapping) else {}
def profile_build_mode(config: Mapping[str, Any]) -> str:
"""``config.onboarding.profile_build``: ``"off"`` never offers; anything else -> ``"ask"`` (offer on first contact).
"""``config.onboarding.profile_build``: ``"off"`` never offers; anything else -> ``"ask"``.
This only governs whether the offer is made; lookups inside the flow are
Only governs whether the offer is made; lookups inside the flow are
consented to separately in conversation.
"""
onboarding = config.get("onboarding") if isinstance(config, Mapping) else None
if not isinstance(onboarding, Mapping):
return "ask"
mode = onboarding.get("profile_build")
if isinstance(mode, str) and mode.strip().lower() == "off":
return "off"
return "ask"
mode = _onboarding_section(config).get("profile_build")
return "off" if isinstance(mode, str) and mode.strip().lower() == "off" else "ask"
def profile_build_directive() -> str:
"""System-note directive appended to the very first message ever.
Runs a short opt-in profile-build flow persisting to the user-profile memory
store; phrased so the agent ASKS before any lookup and never silently reads
Short opt-in profile-build flow persisting to the user-profile memory store;
phrased so the agent ASKS before any lookup and never silently reads
connected accounts.
"""
return (
@@ -167,19 +161,10 @@ def profile_build_directive() -> str:
)
# ── State read / write ────────────────────────────────────────────────────
def _get_seen_dict(config: Mapping[str, Any]) -> Mapping[str, Any]:
onboarding = config.get("onboarding") if isinstance(config, Mapping) else None
if not isinstance(onboarding, Mapping):
return {}
seen = onboarding.get("seen")
return seen if isinstance(seen, Mapping) else {}
def is_seen(config: Mapping[str, Any], flag: str) -> bool:
"""Return True if the user has already been shown this first-touch hint."""
return bool(_get_seen_dict(config).get(flag))
"""True if the user has already been shown this first-touch hint."""
seen = _onboarding_section(config).get("seen")
return bool(seen.get(flag)) if isinstance(seen, Mapping) else False
def mark_seen(config_path: Path, flag: str) -> bool:
+13 -24
View File
@@ -1,12 +1,12 @@
"""Shared one-off LLM requests for non-conversational helpers.
A "one-shot" is a single stateless model call outside any conversation: it
never touches session history or prompt caching and returns plain text (commit
messages, rename suggestions, summaries). Call with explicit
``instructions``/``user_input`` or a registered ``template`` + ``variables`` so
prompt engineering stays consistent across CLI/TUI/desktop. Model selection
rides :func:`agent.auxiliary_client.call_llm`: ``main_runtime`` inherits the
live session's provider/model, else ``task`` resolves a cheap backend.
A "one-shot" is a single stateless model call outside any conversation (commit
messages, rename suggestions, summaries): it never touches session history or
prompt caching. Call with explicit ``instructions``/``user_input`` or a registered
``template`` + ``variables`` so prompt engineering stays consistent across
CLI/TUI/desktop. Model selection rides :func:`agent.auxiliary_client.call_llm`:
``main_runtime`` inherits the live session's provider/model, else ``task``
resolves a cheap backend.
"""
import logging
@@ -23,9 +23,7 @@ PromptTemplate = Callable[[Dict[str, Any]], Tuple[str, str]]
def _truncate(text: str, limit: int) -> str:
text = text or ""
if len(text) <= limit:
return text
return text[:limit].rstrip() + "\n…(truncated)"
return text if len(text) <= limit else text[:limit].rstrip() + "\n…(truncated)"
_COMMIT_INSTRUCTIONS = (
@@ -47,7 +45,6 @@ _COMMIT_INSTRUCTIONS = (
def _commit_message_template(variables: Dict[str, Any]) -> Tuple[str, str]:
diff = _truncate(str(variables.get("diff") or ""), 12000)
recent = _truncate(str(variables.get("recent_commits") or ""), 1500)
parts = []
if recent.strip():
parts.append(
@@ -55,7 +52,6 @@ def _commit_message_template(variables: Dict[str, Any]) -> Tuple[str, str]:
f"{recent}"
)
parts.append("Diff to describe:\n" + (diff or "(no textual diff available)"))
# "Regenerate" must yield something new even on greedy/server-pinned
# temperature models; a nonce isn't enough, so hand back the previous
# message and require a genuinely different one.
@@ -67,7 +63,6 @@ def _commit_message_template(variables: Dict[str, Any]) -> Tuple[str, str]:
"reasonable, a different emphasis or scope framing) — do not repeat "
f"it:\n{avoid}"
)
return _COMMIT_INSTRUCTIONS, "\n\n".join(parts)
@@ -104,15 +99,11 @@ def run_oneshot(
"""
if template:
instructions, user_input = render_template(template, variables)
if not (instructions or "").strip() and not (user_input or "").strip():
has_instructions = bool((instructions or "").strip())
if not has_instructions and not (user_input or "").strip():
raise ValueError("run_oneshot requires a template or instructions/user_input")
messages = []
if (instructions or "").strip():
messages.append({"role": "system", "content": instructions})
messages = [{"role": "system", "content": instructions}] if has_instructions else []
messages.append({"role": "user", "content": user_input or ""})
response = call_llm(
task=task,
messages=messages,
@@ -121,9 +112,7 @@ def run_oneshot(
timeout=timeout,
main_runtime=main_runtime,
)
text = (extract_content_or_reasoning(response) or "").strip()
return _strip_code_fence(text)
return _strip_code_fence((extract_content_or_reasoning(response) or "").strip())
def _strip_code_fence(text: str) -> str:
@@ -131,6 +120,6 @@ def _strip_code_fence(text: str) -> str:
if not text.startswith("```"):
return text
lines = text.splitlines()
if len(lines) >= 2 and lines[0].startswith("```") and lines[-1].strip() == "```":
if len(lines) >= 2 and lines[-1].strip() == "```":
return "\n".join(lines[1:-1]).strip()
return text
+10 -23
View File
@@ -1,16 +1,13 @@
#!/usr/bin/env python3
"""``/plan`` — build the plan-mode prompt: a saved markdown implementation plan, no execution.
A first-class built-in (the former bundled skill fell off capped Telegram/Discord
command menus). No engine, no model-tool footprint: every surface feeds
:func:`build_plan_prompt` to the agent as a normal turn, like ``/learn`` and
``/init``, so system prompt and history stay untouched (prompt-cache safe).
A first-class built-in with no engine and no model-tool footprint: every surface
feeds :func:`build_plan_prompt` to the agent as a normal turn (like ``/learn``),
so system prompt and history stay untouched (prompt-cache safe).
"""
from __future__ import annotations
# Plan-mode ground rules + authoring craft, distilled from the retired bundled
# skill (writing-craft adapted from obra/superpowers).
# Ground rules + authoring craft (writing-craft adapted from obra/superpowers).
_PLAN_MODE_RULES = """\
For this turn, you are in PLAN MODE — planning only.
@@ -63,20 +60,10 @@ Interaction style:
def build_plan_prompt(task: str = "") -> str:
"""Build the plan-mode prompt; empty *task* asks the agent to infer it from conversation context."""
task = (task or "").strip()
if task:
task_block = f"Task to plan:\n{task}\n"
else:
task_block = (
"No explicit task was given with /plan — infer the task from the "
"current conversation context (the thing we have been discussing "
"or working toward). If the conversation does not imply a task, "
"ask a brief clarifying question.\n"
)
return (
"[/plan — plan mode]\n\n"
+ _PLAN_MODE_RULES
+ "\n"
+ task_block
+ "\n"
+ _PLAN_CRAFT
task_block = f"Task to plan:\n{task}\n" if task else (
"No explicit task was given with /plan — infer the task from the "
"current conversation context (the thing we have been discussing "
"or working toward). If the conversation does not imply a task, "
"ask a brief clarifying question.\n"
)
return "[/plan — plan mode]\n\n" + _PLAN_MODE_RULES + "\n" + task_block + "\n" + _PLAN_CRAFT
+31 -52
View File
@@ -2,15 +2,10 @@
Every Hermes request to the Nous Portal (main loop, auxiliary client, fallback
paths) must carry the same product-attribution tags, sent in OpenAI-compatible
``extra_body['tags']``::
["product=hermes-agent", "client=hermes-client-v<__version__>"]
One helper instead of inlined literals: the call sites drifted apart before,
and tests can assert one tag list everywhere. The version is read live from
``hermes_cli.__version__`` (the release script bumps that single string) — do
NOT pre-compute it as a module constant in consumers; it can change at runtime
(editable installs, hot reload).
``extra_body['tags']``: ``["product=hermes-agent", "client=hermes-client-v<__version__>"]``.
The version is read live from ``hermes_cli.__version__`` — do NOT pre-compute it
as a module constant in consumers; it can change at runtime (editable installs,
hot reload).
"""
from __future__ import annotations
@@ -19,32 +14,25 @@ from contextvars import ContextVar
from typing import List, Optional
# Ambient conversation id (ATTRIBUTION value, sent as ``conversation=<id>``).
# The agent loop publishes it at turn entry; the dozens of auxiliary call
# sites funnelling through ``auxiliary_client.call_llm`` (no session handle)
# pick it up via ``nous_portal_tags()`` instead of threading a session_id
# parameter everywhere. A ContextVar, not a module global, so concurrent agents
# in one process (gateway sessions, delegate subagents) never see each other's
# id; ``tools.thread_context.propagate_context_to_thread`` workers inherit it,
# bare threads capture it at spawn time.
_conversation_id: ContextVar[Optional[str]] = ContextVar(
"nous_portal_conversation_id", default=None
)
# The agent loop publishes it at turn entry; auxiliary call sites funnelling
# through ``auxiliary_client.call_llm`` (no session handle) pick it up via
# ``nous_portal_tags()``. A ContextVar so concurrent agents in one process never
# see each other's id; ``propagate_context_to_thread`` workers inherit it.
_conversation_id: ContextVar[Optional[str]] = ContextVar("nous_portal_conversation_id", default=None)
# Ambient affinity scope (ROUTING value): OpenRouter's sticky ``session_id``,
# Nous Portal's sticky key and xAI's ``x-grok-conv-id`` pin a conversation to
# one backend/prompt cache. Usually equal to the conversation id, but a host
# that mints one physical session per RESPONSE must route on the key it
# declared for the whole chat (``prompt_cache_scope.declared_conversation_scope``).
# Only that declared value is published; unset means consumers fall back to the
# conversation id, so delegate trees keep sharing their parent's sticky key.
_affinity_scope: ContextVar[Optional[str]] = ContextVar(
"hermes_affinity_scope", default=None
)
# Ambient affinity scope (ROUTING value): OpenRouter's sticky ``session_id``, Nous
# Portal's sticky key and xAI's ``x-grok-conv-id`` pin a conversation to one
# backend/prompt cache. Usually equal to the conversation id, but a host that mints
# one physical session per RESPONSE must route on the key it declared for the whole
# chat (``prompt_cache_scope.declared_conversation_scope``). Only that declared value
# is published; unset means consumers fall back to the conversation id, so delegate
# trees keep sharing their parent's sticky key.
_affinity_scope: ContextVar[Optional[str]] = ContextVar("hermes_affinity_scope", default=None)
def _reset_var(var: ContextVar, token) -> None:
"""Reset ``var``; a token from another Context (reset on a different
thread) falls back to clearing rather than raising in cleanup paths."""
"""Reset ``var``; a token from another Context (reset on a different thread)
falls back to clearing rather than raising in cleanup paths."""
try:
var.reset(token)
except Exception:
@@ -62,16 +50,14 @@ def reset_affinity_scope(token) -> None:
def get_affinity_scope() -> Optional[str]:
"""Return the declared routing/affinity scope, or ``None`` when unset."""
return _affinity_scope.get()
def set_conversation_context(conversation_id: Optional[str]):
"""Publish the active conversation id for ambient Portal tagging.
"""Publish the active conversation id for ambient Portal tagging; returns the token.
Called by the agent loop at turn entry with the session-lineage ROOT id
(so the tag survives context-compression rotation). ``None`` clears.
Returns the ContextVar token for ``reset_conversation_context``.
Called by the agent loop at turn entry with the session-lineage ROOT id (so
the tag survives context-compression rotation). ``None`` clears.
"""
return _conversation_id.set(conversation_id or None)
@@ -82,36 +68,29 @@ def reset_conversation_context(token) -> None:
def get_conversation_context() -> Optional[str]:
"""Return the ambient conversation id, or ``None`` when unset."""
return _conversation_id.get()
def _hermes_version() -> str:
"""Current Hermes release version; ``"unknown"`` if hermes_cli is unimportable."""
def hermes_client_tag() -> str:
"""``client=hermes-client-v<MAJOR>.<MINOR>.<PATCH>`` ("unknown" if hermes_cli is unimportable)."""
try:
from hermes_cli import __version__
return __version__
except Exception:
return "unknown"
def hermes_client_tag() -> str:
"""``client=hermes-client-v<MAJOR>.<MINOR>.<PATCH>``."""
return f"client=hermes-client-v{_hermes_version()}"
__version__ = "unknown"
return f"client=hermes-client-v{__version__}"
def conversation_tag(session_id: str) -> str:
"""``conversation=<session_id>`` — high-cardinality, so only appended when
a session id is actually available, never in the always-on base set."""
"""``conversation=<session_id>`` — high-cardinality, so only appended when a
session id is actually available, never in the always-on base set."""
return f"conversation={session_id}"
def nous_portal_tags(session_id: str | None = None) -> List[str]:
"""Return a fresh list of the canonical Nous Portal tags.
"""Fresh list of the canonical Nous Portal tags.
The ambient conversation context (lineage ROOT id published by the agent
loop) wins over the explicit ``session_id``, which remains a fallback for
callers outside any agent turn; with neither, the two-tag base set.
The ambient conversation context (lineage ROOT id) wins over the explicit
``session_id``, a fallback for callers outside any agent turn.
"""
tags = ["product=hermes-agent", hermes_client_tag()]
effective = get_conversation_context() or session_id
+37 -70
View File
@@ -1,9 +1,9 @@
"""Process-level bootstrap helpers for ``run_agent``.
Lazy OpenAI SDK import (``_load_openai_cls`` / ``_OpenAIProxy``, preserving
``isinstance`` and ``patch("run_agent.OpenAI")`` patterns), crash-resistant
stdio (``_SafeWriter``), env-only HTTP proxy resolution, and Codex dual-stack
(Happy Eyeballs) connection racing. ``run_agent`` re-exports every name.
Lazy OpenAI SDK import (``_OpenAIProxy`` keeps ``isinstance`` and
``patch("run_agent.OpenAI")`` working), crash-resistant stdio (``_SafeWriter``),
env-only HTTP proxy resolution, and Codex dual-stack (Happy Eyeballs)
connection racing. ``run_agent`` re-exports every name.
"""
from __future__ import annotations
@@ -25,26 +25,20 @@ _HAPPY_EYEBALLS_DELAY_SECONDS = 0.25
def _interleave_addrinfos(addrinfos: list[tuple]) -> list[tuple]:
"""Interleave resolved address families while preserving resolver order."""
"""Round-robin the resolved address families (deduped), preserving resolver order within each."""
queues: dict[int, list[tuple]] = {}
family_order: list[int] = []
seen: set[tuple] = set()
for addrinfo in addrinfos:
family, socktype, proto, _canonname, sockaddr = addrinfo
marker = (family, socktype, proto, sockaddr)
if marker in seen:
continue
seen.add(marker)
if family not in queues:
queues[family] = []
family_order.append(family)
queues[family].append(addrinfo)
if marker not in seen:
seen.add(marker)
queues.setdefault(family, []).append(addrinfo)
interleaved: list[tuple] = []
while any(queues.values()):
for family in family_order:
if queues[family]:
interleaved.append(queues[family].pop(0))
for queue in queues.values():
if queue:
interleaved.append(queue.pop(0))
return interleaved
@@ -81,17 +75,13 @@ def _happy_eyeballs_create_connection(
candidate = socket.socket(family, socktype, proto)
try:
if source_address is not None:
local_infos = socket.getaddrinfo(
source_address[0], source_address[1], family=family, type=socktype
)
local_infos = socket.getaddrinfo(source_address[0], source_address[1], family=family, type=socktype)
if not local_infos:
raise OSError(
f"getaddrinfo returned no local {family} address for {source_address[0]}"
)
raise OSError(f"getaddrinfo returned no local {family} address for {source_address[0]}")
candidate.bind(local_infos[0][4])
candidate.setblocking(False)
result = candidate.connect_ex(sockaddr)
if result == 0 or result == errno.EISCONN:
if result in (0, errno.EISCONN):
return candidate
if result not in in_progress:
raise OSError(result, os.strerror(result))
@@ -109,9 +99,8 @@ def _happy_eyeballs_create_connection(
raise socket.timeout("timed out")
if pending and now >= next_launch:
addrinfo = pending.pop(0)
try:
winner = start_attempt(addrinfo)
winner = start_attempt(pending.pop(0))
except OSError as exc:
last_error = exc
if not active:
@@ -126,8 +115,7 @@ def _happy_eyeballs_create_connection(
until_launch = max(0.0, next_launch - now)
wait_timeout = until_launch if wait_timeout is None else min(wait_timeout, until_launch)
events = selector.select(wait_timeout)
for key, _mask in events:
for key, _mask in selector.select(wait_timeout):
candidate = key.fileobj
error_code = candidate.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
selector.unregister(candidate)
@@ -143,9 +131,7 @@ def _happy_eyeballs_create_connection(
next_launch = time.monotonic()
if winner is None:
if last_error is not None:
raise last_error
raise OSError(f"Could not connect to {host}:{port}")
raise last_error if last_error is not None else OSError(f"Could not connect to {host}:{port}")
try:
selector.unregister(winner)
@@ -206,19 +192,15 @@ class _HappyEyeballsSyncBackend:
def _uses_codex_cloud_transport(base_url: str) -> bool:
return (
base_url_hostname(base_url).lower() == "chatgpt.com"
and "/backend-api/codex" in str(base_url).lower()
)
return base_url_hostname(base_url).lower() == "chatgpt.com" and "/backend-api/codex" in str(base_url).lower()
def _enable_happy_eyeballs(transport, skip_pool_types: tuple = ()) -> None:
"""Install the racing backend on one httpx transport.
Reaches into private ``transport._pool._network_backend`` (httpcore is
pinned 1.0.x); hasattr-guarded so an incompatible httpcore degrades to the
default serial backend instead of crashing. Pools of ``skip_pool_types``
(proxies) are left alone.
Reaches into private ``transport._pool._network_backend`` (httpcore pinned
1.0.x); hasattr-guarded so an incompatible httpcore degrades to the default
serial backend. Pools of ``skip_pool_types`` (proxies) are left alone.
"""
pool = getattr(transport, "_pool", None)
if pool is None or not hasattr(pool, "_network_backend"):
@@ -229,11 +211,11 @@ def _enable_happy_eyeballs(transport, skip_pool_types: tuple = ()) -> None:
def enable_happy_eyeballs_on_client(client) -> None:
"""Install the racing backend on every direct transport of a ready-built httpx.Client
(for callers that build clients inline, e.g. Codex OAuth/device-login in hermes_cli.auth).
"""Install the racing backend on every direct transport of a ready-built httpx.Client.
Proxy-backed pools are skipped (TCP connect goes to the proxy host) and
async clients need nothing (anyio already races per RFC 8305). Best-effort.
For callers that build clients inline (Codex OAuth/device-login). Proxy-backed
pools are skipped (TCP connect goes to the proxy host); async clients need
nothing (anyio already races per RFC 8305). Best-effort.
"""
try:
import httpcore
@@ -243,9 +225,7 @@ def enable_happy_eyeballs_on_client(client) -> None:
)
except Exception:
return
transports = [getattr(client, "_transport", None)]
transports.extend((getattr(client, "_mounts", None) or {}).values())
transports = [getattr(client, "_transport", None), *(getattr(client, "_mounts", None) or {}).values()]
for transport in transports:
_enable_happy_eyeballs(transport, proxy_pool_types)
@@ -314,8 +294,7 @@ class _SafeWriter:
def _get_proxy_from_env() -> Optional[str]:
"""First configured proxy URL from HTTPS_PROXY / HTTP_PROXY / ALL_PROXY (any case), or None."""
for key in ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY",
"https_proxy", "http_proxy", "all_proxy"):
for key in ("HTTPS_PROXY", "HTTP_PROXY", "ALL_PROXY", "https_proxy", "http_proxy", "all_proxy"):
value = os.environ.get(key, "").strip()
if value:
return normalize_proxy_url(value)
@@ -323,56 +302,44 @@ def _get_proxy_from_env() -> Optional[str]:
def _get_proxy_for_base_url(base_url: Optional[str]) -> Optional[str]:
"""Return an env-configured proxy unless NO_PROXY excludes this base URL."""
"""Env-configured proxy unless NO_PROXY excludes this base URL."""
proxy = _get_proxy_from_env()
if not proxy or not base_url:
return proxy
host = base_url_hostname(base_url)
host = base_url_hostname(base_url) if proxy and base_url else ""
if not host:
return proxy
try:
if urllib.request.proxy_bypass_environment(host):
return None
except Exception:
pass
return proxy
def build_keepalive_http_client(base_url: str = "", *, async_mode: bool = False, verify: Any = True) -> Optional[Any]:
"""Build an httpx client for OpenAI SDK calls with env-only proxy policy.
"""httpx client for OpenAI SDK calls with env-only proxy policy (None on failure).
Explicit no-proxy mounts disable httpx's ``trust_env`` path so macOS system
proxies (which omit the ExceptionsList) are never applied. ``keepalive_expiry``
reaps idle connections before reverse proxies' 30-60 s timeouts (a custom
socket_options transport broke streaming and stripped TCP_NODELAY). ``verify``
lets auxiliary calls honor the same ``ssl_ca_cert``/``ssl_verify``/``HERMES_CA_BUNDLE``
as the main client; it goes on the client AND the mounts, since a mounted
transport owns its SSL context.
goes on the client AND the mounts, since a mounted transport owns its SSL context.
"""
try:
import httpx
proxy = _get_proxy_for_base_url(base_url)
limits = httpx.Limits(max_keepalive_connections=20, max_connections=100, keepalive_expiry=20.0)
# Generous read=None for SSE streaming endpoints.
timeout = httpx.Timeout(connect=15.0, read=None, write=15.0, pool=10.0)
timeout = httpx.Timeout(connect=15.0, read=None, write=15.0, pool=10.0) # read=None for SSE streaming
transport_cls = httpx.AsyncHTTPTransport if async_mode else httpx.HTTPTransport
client_cls = httpx.AsyncClient if async_mode else httpx.Client
mounts = {}
mounts = None
if proxy is None:
http_transport = transport_cls(verify=verify)
https_transport = transport_cls(verify=verify)
mounts = {"http://": transport_cls(verify=verify), "https://": transport_cls(verify=verify)}
# Async transports race natively (anyio happy_eyeballs_delay=0.25).
if not async_mode and _uses_codex_cloud_transport(base_url):
_enable_happy_eyeballs(http_transport)
_enable_happy_eyeballs(https_transport)
mounts = {"http://": http_transport, "https://": https_transport}
return client_cls(limits=limits, timeout=timeout, proxy=proxy, mounts=mounts or None, verify=verify)
for transport in mounts.values():
_enable_happy_eyeballs(transport)
return client_cls(limits=limits, timeout=timeout, proxy=proxy, mounts=mounts, verify=verify)
except Exception:
return None