Merge branch 'simp/r3-05-w6' into simp/r3-05
This commit is contained in:
Teknium
+146
-356
@@ -6,6 +6,7 @@ as one prompt, collects text chunks, and returns the minimal OpenAI-client shape
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import contextlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@@ -32,19 +33,34 @@ from tools.environments.local import hermes_subprocess_env
|
||||
ACP_MARKER_BASE_URL = "acp://copilot"
|
||||
logger = logging.getLogger(__name__)
|
||||
_DEFAULT_TIMEOUT_SECONDS = 900.0
|
||||
|
||||
# Stderr fingerprint of the deprecated `gh copilot` extension. Require BOTH the
|
||||
# product name AND a deprecation marker: the NEW `@github/copilot` CLI (repo
|
||||
# github/copilot-cli) legitimately mentions "copilot-cli" in its own banners.
|
||||
# Stderr fingerprint of the deprecated `gh copilot` extension. Require BOTH the product name
|
||||
# AND a deprecation marker: the NEW `@github/copilot` CLI legitimately mentions "copilot-cli".
|
||||
_DEPRECATION_REQUIRED = ("gh-copilot",)
|
||||
_DEPRECATION_MARKERS = ("has been deprecated", "no commands will be executed")
|
||||
|
||||
_ROLE_LABELS = {"system": "System", "user": "User", "assistant": "Assistant", "tool": "Tool", "context": "Context"}
|
||||
|
||||
# Probe verdicts per binary path so the ~50ms --help cost is paid once per
|
||||
# process. Only definitive True/False is cached; an inconclusive probe is not,
|
||||
# so a CLI installed mid-session is picked up.
|
||||
# Probe verdicts per binary path (~50ms --help paid once per process). Only definitive
|
||||
# True/False is cached, so a CLI installed mid-session is picked up.
|
||||
_ACP_PROBE_CACHE: dict[str, bool] = {}
|
||||
_PROMPT_PREAMBLE = (
|
||||
"You are being used as the active ACP agent backend for Hermes.",
|
||||
"Use ACP capabilities to complete tasks.",
|
||||
"IMPORTANT: If you take an action with a tool, you MUST output tool calls using <tool_call>{...}</tool_call> blocks with JSON exactly in OpenAI function-call shape.",
|
||||
"If no tool is needed, answer normally.",
|
||||
)
|
||||
_INITIALIZE_PARAMS = {
|
||||
"protocolVersion": 1,
|
||||
"clientCapabilities": {"fs": {"readTextFile": True, "writeTextFile": True}},
|
||||
"clientInfo": {"name": "hermes-agent", "title": "Hermes Agent", "version": "0.0.0"},
|
||||
}
|
||||
_DEPRECATED_CLI_ERROR = (
|
||||
"Hermes ACP mode requires the NEW GitHub Copilot CLI (github.com/github/copilot-cli), but the binary it just "
|
||||
"spawned is the deprecated `gh copilot` extension.\n\n"
|
||||
"Install the new CLI:\n npm install -g @github/copilot\n # then verify with: copilot --help\n\n"
|
||||
"If `copilot` already resolves to the new CLI but you still see this,\npoint Hermes at it explicitly:\n"
|
||||
" export HERMES_COPILOT_ACP_COMMAND=/path/to/new/copilot\n\n"
|
||||
"Alternative: use the `copilot` provider (no ACP, hits the Copilot API\ndirectly with a Copilot subscription "
|
||||
"token) via `hermes setup`.\n\nOriginal error:\n"
|
||||
)
|
||||
|
||||
|
||||
def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
|
||||
@@ -54,73 +70,57 @@ def _is_gh_copilot_deprecation_message(stderr_text: str) -> bool:
|
||||
|
||||
|
||||
def _resolve_command() -> str:
|
||||
return (
|
||||
os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip()
|
||||
or os.getenv("COPILOT_CLI_PATH", "").strip()
|
||||
or "copilot"
|
||||
)
|
||||
return os.getenv("HERMES_COPILOT_ACP_COMMAND", "").strip() or os.getenv("COPILOT_CLI_PATH", "").strip() or "copilot"
|
||||
|
||||
|
||||
def _resolve_args() -> list[str]:
|
||||
raw = os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()
|
||||
return shlex.split(raw) if raw else ["--acp", "--stdio"]
|
||||
return shlex.split(os.getenv("HERMES_COPILOT_ACP_ARGS", "").strip()) or ["--acp", "--stdio"]
|
||||
|
||||
|
||||
def _acp_supported(command: str, args: list[str]) -> bool | None:
|
||||
"""Tri-state probe: does ``command`` accept ``--acp``?
|
||||
|
||||
A CLI without the flag (older releases, Claude Code v2.x) exits 1 with
|
||||
``error: unknown option '--acp'`` and the parent then waits the full
|
||||
child timeout for stdout that never arrives. True = help advertises --acp;
|
||||
False = help ran cleanly without it (caller fast-fails); None = inconclusive
|
||||
(binary missing / --help failed), caller falls through to the normal spawn error.
|
||||
Only probes when ``--acp`` is among ``args`` — a custom transport is the operator's business.
|
||||
"""
|
||||
"""Tri-state ``--acp`` probe (a CLI without the flag exits 1 and the parent would wait the
|
||||
full child timeout for stdout that never arrives). True = help advertises --acp; False =
|
||||
help ran cleanly without it (caller fast-fails); None = inconclusive (binary missing /
|
||||
--help failed → normal spawn error). Skipped when ``--acp`` is not in ``args`` (custom transport)."""
|
||||
if "--acp" not in args:
|
||||
return True
|
||||
cached = _ACP_PROBE_CACHE.get(command)
|
||||
if cached is not None:
|
||||
if (cached := _ACP_PROBE_CACHE.get(command)) is not None:
|
||||
return cached
|
||||
try:
|
||||
probe = subprocess.run(
|
||||
[command, "--help"], capture_output=True, text=True, encoding="utf-8", errors="replace",
|
||||
timeout=5, stdin=subprocess.DEVNULL,
|
||||
[command, "--help"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=5,
|
||||
stdin=subprocess.DEVNULL,
|
||||
)
|
||||
except (FileNotFoundError, subprocess.TimeoutExpired, OSError):
|
||||
return None
|
||||
if probe.returncode != 0:
|
||||
return None
|
||||
# ``--acp`` as a flag token; tolerate spacing and ``[--acp]`` variants.
|
||||
verdict = bool(re.search(r"(?:^|[\s\[])--acp(?:[\s=\],]|$)", probe.stdout, re.MULTILINE))
|
||||
_ACP_PROBE_CACHE[command] = verdict
|
||||
verdict = _ACP_PROBE_CACHE[command] = bool(re.search(r"(?:^|[\s\[])--acp(?:[\s=\],]|$)", probe.stdout, re.MULTILINE))
|
||||
return verdict
|
||||
|
||||
|
||||
def _resolve_home_dir() -> str:
|
||||
"""Return a stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
|
||||
home = os.environ.get("HOME", "").strip()
|
||||
if home:
|
||||
"""Stable HOME for child ACP processes; /tmp as a last resort so the child never starts HOME-less."""
|
||||
if home := os.environ.get("HOME", "").strip():
|
||||
return home
|
||||
expanded = os.path.expanduser("~")
|
||||
if expanded and expanded != "~":
|
||||
if (expanded := os.path.expanduser("~")) and expanded != "~":
|
||||
return expanded
|
||||
try:
|
||||
import pwd
|
||||
|
||||
resolved = pwd.getpwuid(os.getuid()).pw_dir.strip() # windows-footgun: ok — POSIX fallback inside try/except (pwd import fails on Windows)
|
||||
if resolved:
|
||||
return resolved
|
||||
return pwd.getpwuid(os.getuid()).pw_dir.strip() or "/tmp" # windows-footgun: ok — POSIX fallback inside try/except (pwd import fails on Windows)
|
||||
except Exception:
|
||||
pass
|
||||
return "/tmp"
|
||||
return "/tmp"
|
||||
|
||||
|
||||
def _build_subprocess_env() -> dict[str, str]:
|
||||
# Copilot ACP drives a model and legitimately needs LLM provider credentials;
|
||||
# the central helper still strips Tier-1 secrets (bot tokens, GitHub auth, infra).
|
||||
from hermes_constants import apply_subprocess_home_env
|
||||
|
||||
# Copilot ACP drives a model and needs LLM provider credentials; the central helper still
|
||||
# strips Tier-1 secrets (bot tokens, GitHub auth, infra).
|
||||
env = hermes_subprocess_env(inherit_credentials=True)
|
||||
env["HOME"] = _resolve_home_dir()
|
||||
from hermes_constants import apply_subprocess_home_env
|
||||
apply_subprocess_home_env(env)
|
||||
return env
|
||||
|
||||
@@ -133,103 +133,42 @@ def _jsonrpc_error(message_id: Any, code: int, message: str) -> dict[str, Any]:
|
||||
return {"jsonrpc": "2.0", "id": message_id, "error": {"code": code, "message": message}}
|
||||
|
||||
|
||||
def _permission_denied(message_id: Any) -> dict[str, Any]:
|
||||
return _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})
|
||||
def _enabled_ids(entries: Any, key: str) -> set[str]:
|
||||
"""Ids of ``entries`` (dicts) whose ``_meta.copilotEnablement`` is not ``disabled``."""
|
||||
return {str(e.get(key) or "").strip() for e in (entries or []) if isinstance(e, dict)
|
||||
and str((e.get("_meta") or {}).get("copilotEnablement") or "").strip().lower() != "disabled"}
|
||||
|
||||
|
||||
def _model_selection_request(
|
||||
session: dict[str, Any], requested_model: str
|
||||
) -> tuple[str, dict[str, str]] | None:
|
||||
"""Return the ACP request that selects ``requested_model`` for ``session``.
|
||||
|
||||
Prefer stable v1 ``session/set_config_option``. Fall back to Copilot's
|
||||
pre-stabilization ``session/set_model`` extension only when no model
|
||||
config option is advertised. A reported model list is authoritative:
|
||||
unknown and policy-disabled ids return None instead of being sent.
|
||||
"""
|
||||
def _model_selection_request(session: dict[str, Any], requested_model: str) -> tuple[str, dict[str, str]] | None:
|
||||
"""ACP request selecting ``requested_model`` for ``session``: stable v1
|
||||
``session/set_config_option``, else Copilot's pre-stabilization ``session/set_model``
|
||||
when no model config option is advertised. A reported model list is authoritative:
|
||||
unknown and policy-disabled ids return None instead of being sent."""
|
||||
session_id = str(session.get("sessionId") or "").strip()
|
||||
requested_model = str(requested_model or "").strip()
|
||||
if not session_id or not requested_model or requested_model == "copilot-acp":
|
||||
return None
|
||||
|
||||
config_options = [
|
||||
o for o in (session.get("configOptions") or []) if isinstance(o, dict)
|
||||
]
|
||||
model_option = next(
|
||||
(
|
||||
o for o in config_options
|
||||
if o.get("category") == "model" or o.get("id") == "model"
|
||||
),
|
||||
None,
|
||||
)
|
||||
if model_option is not None:
|
||||
enabled_values = {
|
||||
str(o.get("value") or "").strip()
|
||||
for o in (model_option.get("options") or [])
|
||||
if isinstance(o, dict)
|
||||
and str(
|
||||
((o.get("_meta") or {}).get("copilotEnablement")) or ""
|
||||
).strip().lower() != "disabled"
|
||||
}
|
||||
if requested_model not in enabled_values:
|
||||
options = [o for o in (session.get("configOptions") or []) if isinstance(o, dict) and "model" in (o.get("category"), o.get("id"))]
|
||||
if options:
|
||||
if requested_model not in _enabled_ids(options[0].get("options"), "value"):
|
||||
return None
|
||||
return (
|
||||
"session/set_config_option",
|
||||
{
|
||||
"sessionId": session_id,
|
||||
"configId": str(model_option.get("id") or "model"),
|
||||
"value": requested_model,
|
||||
},
|
||||
)
|
||||
|
||||
advertised = [
|
||||
m
|
||||
for m in ((session.get("models") or {}).get("availableModels") or [])
|
||||
if isinstance(m, dict)
|
||||
]
|
||||
available = {
|
||||
str(m.get("modelId") or "").strip()
|
||||
for m in advertised
|
||||
if str(
|
||||
((m.get("_meta") or {}).get("copilotEnablement")) or ""
|
||||
).strip().lower() != "disabled"
|
||||
}
|
||||
if available and requested_model not in available:
|
||||
return None
|
||||
return (
|
||||
"session/set_model",
|
||||
{"sessionId": session_id, "modelId": requested_model},
|
||||
)
|
||||
return "session/set_config_option", {"sessionId": session_id, "configId": str(options[0].get("id") or "model"), "value": requested_model}
|
||||
available = _enabled_ids((session.get("models") or {}).get("availableModels"), "modelId")
|
||||
return None if available and requested_model not in available else ("session/set_model", {"sessionId": session_id, "modelId": requested_model})
|
||||
|
||||
|
||||
def _format_messages_as_prompt(
|
||||
messages: list[dict[str, Any]],
|
||||
model: str | None = None,
|
||||
tools: list[dict[str, Any]] | None = None,
|
||||
tool_choice: Any = None,
|
||||
messages: list[dict[str, Any]], model: str | None = None, tools: list[dict[str, Any]] | None = None, tool_choice: Any = None,
|
||||
) -> str:
|
||||
sections: list[str] = [
|
||||
"You are being used as the active ACP agent backend for Hermes.",
|
||||
"Use ACP capabilities to complete tasks.",
|
||||
"IMPORTANT: If you take an action with a tool, you MUST output tool calls using <tool_call>{...}</tool_call> blocks with JSON exactly in OpenAI function-call shape.",
|
||||
"If no tool is needed, answer normally.",
|
||||
]
|
||||
# Deliberately no "requested model" line: the model is applied for real via ACP
|
||||
# session/set_model; a prompt-text mention makes a substituted backend model
|
||||
# FALSELY self-identify as the requested one. Identity comes from the backend.
|
||||
# Copilot has no tools of its own that collide with Hermes', so forward the whole toolset.
|
||||
sections.extend(_render_tool_bridge_sections(tools, tool_choice))
|
||||
|
||||
# Deliberately no "requested model" line: the model is applied for real via ACP session/set_model;
|
||||
# a prompt-text mention makes a substituted backend model FALSELY self-identify as the requested
|
||||
# one. Copilot has no tools of its own that collide with Hermes', so forward the whole toolset.
|
||||
sections: list[str] = [*_PROMPT_PREAMBLE, *_render_tool_bridge_sections(tools, tool_choice)]
|
||||
transcript: list[str] = []
|
||||
for message in messages:
|
||||
if not isinstance(message, dict):
|
||||
continue
|
||||
for message in (m for m in messages if isinstance(m, dict)):
|
||||
role = str(message.get("role") or "unknown").strip().lower()
|
||||
if role not in _ROLE_LABELS:
|
||||
role = "context"
|
||||
rendered = _render_message_content(message.get("content"))
|
||||
if rendered:
|
||||
transcript.append(f"{_ROLE_LABELS[role]}:\n{rendered}")
|
||||
if rendered := _render_message_content(message.get("content")):
|
||||
transcript.append(f"{_ROLE_LABELS.get(role, 'Context')}:\n{rendered}")
|
||||
if transcript:
|
||||
sections.append("Conversation transcript:\n\n" + "\n\n".join(transcript))
|
||||
sections.append("Continue the conversation from the latest user request.")
|
||||
@@ -239,33 +178,21 @@ def _format_messages_as_prompt(
|
||||
def _render_message_content(content: Any) -> str:
|
||||
if content is None:
|
||||
return ""
|
||||
if isinstance(content, str):
|
||||
return content.strip()
|
||||
if isinstance(content, dict):
|
||||
if "text" in content:
|
||||
return str(content.get("text") or "").strip()
|
||||
if isinstance(content.get("content"), str):
|
||||
return str(content.get("content") or "").strip()
|
||||
return json.dumps(content, ensure_ascii=True)
|
||||
return content["content"].strip() if isinstance(content.get("content"), str) else json.dumps(content, ensure_ascii=True)
|
||||
if isinstance(content, list):
|
||||
parts: list[str] = []
|
||||
for item in content:
|
||||
if isinstance(item, str):
|
||||
parts.append(item)
|
||||
elif isinstance(item, dict):
|
||||
text = item.get("text")
|
||||
if isinstance(text, str) and text.strip():
|
||||
parts.append(text.strip())
|
||||
parts = [item if isinstance(item, str) else item["text"].strip() for item in content if isinstance(item, str)
|
||||
or (isinstance(item, dict) and isinstance(item.get("text"), str) and item["text"].strip())]
|
||||
return "\n".join(parts).strip()
|
||||
return str(content).strip()
|
||||
|
||||
|
||||
def _ensure_path_within_cwd(path_text: str, cwd: str) -> Path:
|
||||
candidate = Path(path_text)
|
||||
if not candidate.is_absolute():
|
||||
if not Path(path_text).is_absolute():
|
||||
raise PermissionError("ACP file-system paths must be absolute.")
|
||||
resolved = candidate.resolve()
|
||||
root = Path(cwd).resolve()
|
||||
resolved, root = Path(path_text).resolve(), Path(cwd).resolve()
|
||||
try:
|
||||
resolved.relative_to(root)
|
||||
except ValueError as exc:
|
||||
@@ -275,48 +202,33 @@ def _ensure_path_within_cwd(path_text: str, cwd: str) -> Path:
|
||||
|
||||
def _effective_timeout(timeout: Any) -> float:
|
||||
"""Normalise a float or httpx.Timeout-like object to wall-clock seconds (largest component wins)."""
|
||||
if timeout is None:
|
||||
return _DEFAULT_TIMEOUT_SECONDS
|
||||
if isinstance(timeout, (int, float)):
|
||||
return float(timeout)
|
||||
_candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")]
|
||||
_numeric = [float(v) for v in _candidates if isinstance(v, (int, float))]
|
||||
return max(_numeric) if _numeric else _DEFAULT_TIMEOUT_SECONDS
|
||||
candidates = [getattr(timeout, attr, None) for attr in ("read", "write", "connect", "pool", "timeout")]
|
||||
return max((float(v) for v in candidates if isinstance(v, (int, float))), default=_DEFAULT_TIMEOUT_SECONDS)
|
||||
|
||||
|
||||
def _fs_read_text_file(params: dict[str, Any], cwd: str) -> Any:
|
||||
path = _ensure_path_within_cwd(str(params.get("path") or ""), cwd)
|
||||
block_error = get_read_block_error(str(path))
|
||||
if block_error:
|
||||
if block_error := get_read_block_error(str(path)):
|
||||
raise PermissionError(block_error)
|
||||
try:
|
||||
content = path.read_text(encoding="utf-8")
|
||||
except FileNotFoundError:
|
||||
content = ""
|
||||
line = params.get("line")
|
||||
limit = params.get("limit")
|
||||
line, limit = params.get("line"), params.get("limit")
|
||||
if isinstance(line, int) and line > 1:
|
||||
lines = content.splitlines(keepends=True)
|
||||
start = line - 1
|
||||
end = start + limit if isinstance(limit, int) and limit > 0 else None
|
||||
content = "".join(lines[start:end])
|
||||
if content:
|
||||
content = redact_sensitive_text(content, force=True)
|
||||
return {"content": content}
|
||||
end = line - 1 + limit if isinstance(limit, int) and limit > 0 else None
|
||||
content = "".join(content.splitlines(keepends=True)[line - 1:end])
|
||||
return {"content": redact_sensitive_text(content, force=True) if content else content}
|
||||
|
||||
|
||||
def _fs_write_text_file(params: dict[str, Any], cwd: str) -> Any:
|
||||
path = _ensure_path_within_cwd(str(params.get("path") or ""), cwd)
|
||||
denied = get_write_denied_error(str(path))
|
||||
if denied:
|
||||
if denied := get_write_denied_error(str(path)):
|
||||
raise PermissionError(denied)
|
||||
# Approval-gated paths (e.g. ~/.ssh/config) are only soft-gated for interactive
|
||||
# tools, but the ACP shim has no human channel to confirm — fail closed.
|
||||
if is_write_approval_required(str(path)):
|
||||
raise PermissionError(
|
||||
f"Write denied: '{path}' requires interactive approval "
|
||||
"and cannot be written through the ACP file bridge."
|
||||
)
|
||||
if is_write_approval_required(str(path)): # soft-gated for interactive tools; the ACP shim has no human channel → fail closed
|
||||
raise PermissionError(f"Write denied: '{path}' requires interactive approval and cannot be written through the ACP file bridge.")
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(str(params.get("content") or ""), encoding="utf-8")
|
||||
return None
|
||||
@@ -325,138 +237,80 @@ def _fs_write_text_file(params: dict[str, Any], cwd: str) -> Any:
|
||||
_FS_HANDLERS = {"fs/read_text_file": _fs_read_text_file, "fs/write_text_file": _fs_write_text_file}
|
||||
|
||||
|
||||
class _ACPChatCompletions:
|
||||
def __init__(self, client: "CopilotACPClient"):
|
||||
self._client = client
|
||||
|
||||
def create(self, **kwargs: Any) -> Any:
|
||||
return self._client._create_chat_completion(**kwargs)
|
||||
|
||||
|
||||
class _ACPChatNamespace:
|
||||
def __init__(self, client: "CopilotACPClient"):
|
||||
self.completions = _ACPChatCompletions(client)
|
||||
|
||||
|
||||
class CopilotACPClient:
|
||||
"""Minimal OpenAI-client-compatible facade for Copilot ACP."""
|
||||
|
||||
# Declared for agent/auxiliary_client.py: this shim drives an ACP subprocess
|
||||
# over stdio, so it is already a complete client (never re-dispatch it
|
||||
# through a wire adapter) and is safe to use from async code as-is.
|
||||
# Declared for agent/auxiliary_client.py: this shim drives an ACP subprocess over stdio, so it is
|
||||
# already a complete client (never re-dispatch through a wire adapter) and async-safe as-is.
|
||||
HERMES_SKIP_TRANSPORT_WRAP = True
|
||||
HERMES_SKIP_ASYNC_WRAP = True
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
api_key: str | None = None,
|
||||
base_url: str | None = None,
|
||||
default_headers: dict[str, str] | None = None,
|
||||
acp_command: str | None = None,
|
||||
acp_args: list[str] | None = None,
|
||||
acp_cwd: str | None = None,
|
||||
command: str | None = None,
|
||||
args: list[str] | None = None,
|
||||
**_: Any,
|
||||
self, *, api_key: str | None = None, base_url: str | None = None, default_headers: dict[str, str] | None = None,
|
||||
acp_command: str | None = None, acp_args: list[str] | None = None, acp_cwd: str | None = None, command: str | None = None,
|
||||
args: list[str] | None = None, **_: Any,
|
||||
):
|
||||
self.api_key = api_key or "copilot-acp"
|
||||
self.base_url = base_url or ACP_MARKER_BASE_URL
|
||||
self.api_key, self.base_url = api_key or "copilot-acp", base_url or ACP_MARKER_BASE_URL
|
||||
self._default_headers = dict(default_headers or {})
|
||||
self._acp_command = acp_command or command or _resolve_command()
|
||||
self._acp_args = list(acp_args or args or _resolve_args())
|
||||
self._acp_cwd = str(Path(acp_cwd or os.getcwd()).resolve())
|
||||
self.chat = _ACPChatNamespace(self)
|
||||
self.is_closed = False
|
||||
self._active_process: subprocess.Popen[str] | None = None
|
||||
self.chat = SimpleNamespace(completions=SimpleNamespace(create=self._create_chat_completion))
|
||||
self.is_closed, self._active_process = False, None
|
||||
self._active_process_lock = threading.Lock()
|
||||
|
||||
def close(self) -> None:
|
||||
with self._active_process_lock:
|
||||
proc = self._active_process
|
||||
self._active_process = None
|
||||
proc, self._active_process = self._active_process, None
|
||||
self.is_closed = True
|
||||
if proc is None:
|
||||
return
|
||||
try:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=2)
|
||||
if proc is not None:
|
||||
proc.terminate()
|
||||
proc.wait(timeout=2)
|
||||
except Exception:
|
||||
try:
|
||||
with contextlib.suppress(Exception):
|
||||
proc.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _create_chat_completion(
|
||||
self,
|
||||
*,
|
||||
model: str | None = None,
|
||||
messages: list[dict[str, Any]] | None = None,
|
||||
timeout: float | None = None,
|
||||
tools: list[dict[str, Any]] | None = None,
|
||||
tool_choice: Any = None,
|
||||
stream: bool = False,
|
||||
**_: Any,
|
||||
self, *, model: str | None = None, messages: list[dict[str, Any]] | None = None, timeout: float | None = None,
|
||||
tools: list[dict[str, Any]] | None = None, tool_choice: Any = None, stream: bool = False, **_: Any,
|
||||
) -> Any:
|
||||
prompt_text = _format_messages_as_prompt(messages or [], model=model, tools=tools, tool_choice=tool_choice)
|
||||
response_text, reasoning_text = self._run_prompt(
|
||||
prompt_text, timeout_seconds=_effective_timeout(timeout), model=model
|
||||
)
|
||||
response_text, reasoning = self._run_prompt(prompt_text, timeout_seconds=_effective_timeout(timeout), model=model)
|
||||
tool_calls, cleaned_text = _extract_tool_calls_from_text(response_text)
|
||||
|
||||
usage = SimpleNamespace(
|
||||
prompt_tokens=0,
|
||||
completion_tokens=0,
|
||||
total_tokens=0,
|
||||
prompt_tokens_details=SimpleNamespace(cached_tokens=0),
|
||||
)
|
||||
assistant_message = SimpleNamespace(
|
||||
content=cleaned_text,
|
||||
tool_calls=tool_calls,
|
||||
reasoning=reasoning_text or None,
|
||||
reasoning_content=reasoning_text or None,
|
||||
message = SimpleNamespace(
|
||||
content=cleaned_text, tool_calls=tool_calls, reasoning=reasoning or None, reasoning_content=reasoning or None,
|
||||
reasoning_details=None,
|
||||
)
|
||||
choice = SimpleNamespace(message=assistant_message, finish_reason="tool_calls" if tool_calls else "stop")
|
||||
completion = SimpleNamespace(choices=[choice], usage=usage, model=model or "copilot-acp")
|
||||
completion = SimpleNamespace(
|
||||
choices=[SimpleNamespace(message=message, finish_reason="tool_calls" if tool_calls else "stop")],
|
||||
usage=SimpleNamespace(prompt_tokens=0, completion_tokens=0, total_tokens=0, prompt_tokens_details=SimpleNamespace(cached_tokens=0)),
|
||||
model=model or "copilot-acp",
|
||||
)
|
||||
return _completion_to_stream_chunks(completion) if stream else completion
|
||||
|
||||
def _spawn(self) -> subprocess.Popen[str]:
|
||||
# Fast-fail when the CLI rejects --acp: without the probe the parent waits
|
||||
# the full child timeout for stdout that never arrives. ``None`` falls
|
||||
# through to the spawn, which raises the established start error.
|
||||
# Fast-fail when the CLI rejects --acp (else the parent waits the full child timeout for stdout that
|
||||
# never arrives). ``None`` falls through to the spawn's established start error.
|
||||
if _acp_supported(self._acp_command, self._acp_args) is False:
|
||||
preview = " ".join(self._acp_args[:3]) if self._acp_args else "(none)"
|
||||
raise RuntimeError(
|
||||
f"ACP transport not supported by '{self._acp_command}': "
|
||||
f"`{preview}` is rejected as an unknown option. "
|
||||
f"This usually means the CLI is an older release (e.g. "
|
||||
f"Claude Code v2.x) or a different tool than expected. "
|
||||
f"Either install a CLI that ships with --acp support "
|
||||
f"(e.g. `@github/copilot` late 2025+), or set "
|
||||
f"HERMES_COPILOT_ACP_COMMAND / HERMES_COPILOT_ACP_ARGS "
|
||||
f"to a working pair."
|
||||
f"ACP transport not supported by '{self._acp_command}': `{preview}` is rejected as an unknown option. This "
|
||||
"usually means the CLI is an older release (e.g. Claude Code v2.x) or a different tool than expected. Either "
|
||||
"install a CLI that ships with --acp support (e.g. `@github/copilot` late 2025+), or set "
|
||||
"HERMES_COPILOT_ACP_COMMAND / HERMES_COPILOT_ACP_ARGS to a working pair."
|
||||
)
|
||||
try:
|
||||
# Hide the console the child would flash on Windows; stdio pipes stay intact.
|
||||
from hermes_cli._subprocess_compat import windows_hide_flags
|
||||
from hermes_cli._subprocess_compat import windows_hide_flags # hide the Windows console flash; pipes intact
|
||||
|
||||
proc = subprocess.Popen(
|
||||
[self._acp_command] + self._acp_args,
|
||||
stdin=subprocess.PIPE,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True, encoding='utf-8', errors='replace',
|
||||
bufsize=1,
|
||||
cwd=self._acp_cwd,
|
||||
env=_build_subprocess_env(),
|
||||
[self._acp_command] + self._acp_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
||||
text=True, encoding='utf-8', errors='replace', bufsize=1, cwd=self._acp_cwd, env=_build_subprocess_env(),
|
||||
creationflags=windows_hide_flags(),
|
||||
)
|
||||
except FileNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
f"Could not start Copilot ACP command '{self._acp_command}'. "
|
||||
"Install GitHub Copilot CLI or set HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH."
|
||||
) from exc
|
||||
raise RuntimeError(f"Could not start Copilot ACP command '{self._acp_command}'. Install GitHub Copilot CLI or set "
|
||||
"HERMES_COPILOT_ACP_COMMAND/COPILOT_CLI_PATH.") from exc
|
||||
if proc.stdin is None or proc.stdout is None:
|
||||
proc.kill()
|
||||
raise RuntimeError("Copilot ACP process did not expose stdin/stdout pipes.")
|
||||
@@ -466,165 +320,101 @@ class CopilotACPClient:
|
||||
return proc
|
||||
|
||||
def _run_prompt(self, prompt_text: str, *, timeout_seconds: float, model: str | None = None) -> tuple[str, str]:
|
||||
# The CLI's `--model` spawn flag is deliberately NOT used: `copilot --acp`
|
||||
# validates it (unknown id aborts the spawn) but ignores it for the session.
|
||||
# The model is applied after session/new via ACP-native model selection.
|
||||
# The CLI's `--model` spawn flag is deliberately NOT used: `copilot --acp` validates it (unknown id
|
||||
# aborts the spawn) but ignores it for the session; the model is applied after session/new instead.
|
||||
requested_model = str(model or "").strip()
|
||||
proc = self._spawn()
|
||||
inbox: queue.Queue[dict[str, Any]] = queue.Queue()
|
||||
stderr_tail: deque[str] = deque(maxlen=40)
|
||||
|
||||
def _stdout_reader() -> None:
|
||||
for line in proc.stdout:
|
||||
try:
|
||||
inbox.put(json.loads(line))
|
||||
except Exception:
|
||||
inbox.put({"raw": line.rstrip("\n")})
|
||||
def _decode(line: str) -> dict[str, Any]:
|
||||
try:
|
||||
return json.loads(line)
|
||||
except Exception:
|
||||
return {"raw": line.rstrip("\n")}
|
||||
|
||||
def _stderr_reader() -> None:
|
||||
if proc.stderr is None:
|
||||
return
|
||||
for line in proc.stderr:
|
||||
stderr_tail.append(line.rstrip("\n"))
|
||||
def _pump(stream, sink) -> None:
|
||||
for line in stream or ():
|
||||
sink(line)
|
||||
|
||||
threading.Thread(target=_stdout_reader, daemon=True).start()
|
||||
threading.Thread(target=_stderr_reader, daemon=True).start()
|
||||
next_id = 0
|
||||
threading.Thread(target=_pump, args=(proc.stdout, lambda line: inbox.put(_decode(line))), daemon=True).start()
|
||||
threading.Thread(target=_pump, args=(proc.stderr, lambda line: stderr_tail.append(line.rstrip("\n"))), daemon=True).start()
|
||||
request_ids = iter(range(1, 1 << 62))
|
||||
|
||||
def _request(method: str, params: dict[str, Any], *, text_parts: list[str] | None = None, reasoning_parts: list[str] | None = None) -> Any:
|
||||
nonlocal next_id
|
||||
next_id += 1
|
||||
request_id = next_id
|
||||
request_id = next(request_ids)
|
||||
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": request_id, "method": method, "params": params}) + "\n")
|
||||
proc.stdin.flush()
|
||||
|
||||
deadline = time.monotonic() + timeout_seconds
|
||||
while time.monotonic() < deadline:
|
||||
if proc.poll() is not None:
|
||||
break
|
||||
while time.monotonic() < deadline and proc.poll() is None:
|
||||
try:
|
||||
msg = inbox.get(timeout=0.1)
|
||||
except queue.Empty:
|
||||
continue
|
||||
if self._handle_server_message(
|
||||
msg, process=proc, cwd=self._acp_cwd, text_parts=text_parts, reasoning_parts=reasoning_parts
|
||||
):
|
||||
continue
|
||||
if msg.get("id") != request_id:
|
||||
) or msg.get("id") != request_id:
|
||||
continue
|
||||
if "error" in msg:
|
||||
err = msg.get("error") or {}
|
||||
raise RuntimeError(f"Copilot ACP {method} failed: {err.get('message') or err}")
|
||||
return msg.get("result")
|
||||
|
||||
stderr_text = "\n".join(stderr_tail).strip()
|
||||
if proc.poll() is not None and stderr_text:
|
||||
if _is_gh_copilot_deprecation_message(stderr_text):
|
||||
raise RuntimeError(
|
||||
"Hermes ACP mode requires the NEW GitHub Copilot CLI "
|
||||
"(github.com/github/copilot-cli), but the binary it just "
|
||||
"spawned is the deprecated `gh copilot` extension.\n\n"
|
||||
"Install the new CLI:\n"
|
||||
" npm install -g @github/copilot\n"
|
||||
" # then verify with: copilot --help\n\n"
|
||||
"If `copilot` already resolves to the new CLI but you still see this,\n"
|
||||
"point Hermes at it explicitly:\n"
|
||||
" export HERMES_COPILOT_ACP_COMMAND=/path/to/new/copilot\n\n"
|
||||
"Alternative: use the `copilot` provider (no ACP, hits the Copilot API\n"
|
||||
"directly with a Copilot subscription token) via `hermes setup`.\n\n"
|
||||
f"Original error:\n{stderr_text}"
|
||||
)
|
||||
raise RuntimeError(_DEPRECATED_CLI_ERROR + stderr_text)
|
||||
raise RuntimeError(f"Copilot ACP process exited early: {stderr_text}")
|
||||
raise TimeoutError(f"Timed out waiting for Copilot ACP response to {method}.")
|
||||
|
||||
try:
|
||||
_request(
|
||||
"initialize",
|
||||
{
|
||||
"protocolVersion": 1,
|
||||
"clientCapabilities": {"fs": {"readTextFile": True, "writeTextFile": True}},
|
||||
"clientInfo": {"name": "hermes-agent", "title": "Hermes Agent", "version": "0.0.0"},
|
||||
},
|
||||
)
|
||||
_request("initialize", _INITIALIZE_PARAMS)
|
||||
session = _request("session/new", {"cwd": self._acp_cwd, "mcpServers": []}) or {}
|
||||
session_id = str(session.get("sessionId") or "").strip()
|
||||
if not session_id:
|
||||
raise RuntimeError("Copilot ACP did not return a sessionId.")
|
||||
|
||||
# Prefer the stable ACP v1 session-config API (category="model" select
|
||||
# option + session/set_config_option); session/set_model is the fallback.
|
||||
if requested_model and requested_model != "copilot-acp":
|
||||
try:
|
||||
selection = _model_selection_request(session, requested_model)
|
||||
if selection is not None:
|
||||
method, params = selection
|
||||
_request(method, params)
|
||||
if (selection := _model_selection_request(session, requested_model)) is not None:
|
||||
_request(*selection)
|
||||
else:
|
||||
logger.warning(
|
||||
"Copilot ACP does not offer model %r; using the "
|
||||
"session default.",
|
||||
requested_model,
|
||||
)
|
||||
logger.warning("Copilot ACP does not offer model %r; using the session default.", requested_model)
|
||||
except Exception as exc:
|
||||
logger.warning(
|
||||
"Copilot ACP model selection for %r failed; continuing "
|
||||
"with the session default: %s",
|
||||
requested_model,
|
||||
exc,
|
||||
)
|
||||
|
||||
logger.warning("Copilot ACP model selection for %r failed; continuing with the session default: %s", requested_model, exc)
|
||||
text_parts: list[str] = []
|
||||
reasoning_parts: list[str] = []
|
||||
_request(
|
||||
"session/prompt",
|
||||
{"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]},
|
||||
text_parts=text_parts,
|
||||
reasoning_parts=reasoning_parts,
|
||||
)
|
||||
prompt = {"sessionId": session_id, "prompt": [{"type": "text", "text": prompt_text}]}
|
||||
_request("session/prompt", prompt, text_parts=text_parts, reasoning_parts=reasoning_parts)
|
||||
return "".join(text_parts), "".join(reasoning_parts)
|
||||
finally:
|
||||
self.close()
|
||||
|
||||
def _handle_server_message(
|
||||
self,
|
||||
msg: dict[str, Any],
|
||||
*,
|
||||
process: subprocess.Popen[str],
|
||||
cwd: str,
|
||||
text_parts: list[str] | None,
|
||||
reasoning_parts: list[str] | None,
|
||||
self, msg: dict[str, Any], *, process: subprocess.Popen[str], cwd: str, text_parts: list[str] | None, reasoning_parts: list[str] | None,
|
||||
) -> bool:
|
||||
"""Consume a server->client message; True when handled (notification or request answered)."""
|
||||
method = msg.get("method")
|
||||
if not isinstance(method, str):
|
||||
return False
|
||||
|
||||
if method == "session/update":
|
||||
update = (msg.get("params") or {}).get("update") or {}
|
||||
kind = str(update.get("sessionUpdate") or "").strip()
|
||||
content = update.get("content") or {}
|
||||
chunk_text = str(content.get("text") or "") if isinstance(content, dict) else ""
|
||||
if kind == "agent_message_chunk" and chunk_text and text_parts is not None:
|
||||
text_parts.append(chunk_text)
|
||||
elif kind == "agent_thought_chunk" and chunk_text and reasoning_parts is not None:
|
||||
reasoning_parts.append(chunk_text)
|
||||
sinks = {"agent_message_chunk": text_parts, "agent_thought_chunk": reasoning_parts}
|
||||
if chunk_text and (sink := sinks.get(str(update.get("sessionUpdate") or "").strip())) is not None:
|
||||
sink.append(chunk_text)
|
||||
return True
|
||||
|
||||
if process.stdin is None:
|
||||
return True
|
||||
|
||||
message_id = msg.get("id")
|
||||
params = msg.get("params") or {}
|
||||
if method == "session/request_permission":
|
||||
response = _permission_denied(message_id)
|
||||
response = _jsonrpc_result(message_id, {"outcome": {"outcome": "cancelled"}})
|
||||
elif method in _FS_HANDLERS:
|
||||
try:
|
||||
response = _jsonrpc_result(message_id, _FS_HANDLERS[method](params, cwd))
|
||||
response = _jsonrpc_result(message_id, _FS_HANDLERS[method](msg.get("params") or {}, cwd))
|
||||
except Exception as exc:
|
||||
response = _jsonrpc_error(message_id, -32602, str(exc))
|
||||
else:
|
||||
response = _jsonrpc_error(message_id, -32601, f"ACP client method '{method}' is not supported by Hermes yet.")
|
||||
|
||||
process.stdin.write(json.dumps(response) + "\n")
|
||||
process.stdin.flush()
|
||||
return True
|
||||
|
||||
+154
-294
@@ -1,76 +1,47 @@
|
||||
"""Credits tracking for Nous inference API responses: parses x-nous-credits-*
|
||||
(and optional x-nous-tool-pool-*) headers into a validated CreditsState, with
|
||||
depletion detection (paid_access), subscription-cap used_fraction, and warn-once
|
||||
schema-version gating. The hardened parser used by all live consumers.
|
||||
|
||||
Header schema (x-nous-credits-*; each *-micros balance has a *-usd twin holding
|
||||
the server's formatted USD string):
|
||||
version contract/schema version
|
||||
remaining-micros/-usd total remaining balance
|
||||
subscription-micros/-usd subscription balance (SIGNED; may be negative/debt)
|
||||
subscription-limit-* subscription cap (PAIRED/optional)
|
||||
rollover-micros rolled-over balance
|
||||
purchased-micros/-usd purchased balance
|
||||
denominator-kind "subscription_cap" | "none"
|
||||
paid-access "true" | "false" (STRING!)
|
||||
disabled-reason reason string (header omitted when null)
|
||||
as-of-ms server-side timestamp (ms epoch)
|
||||
Tool-pool headers use a SEPARATE prefix: x-nous-tool-pool-micros (balance) and
|
||||
x-nous-tool-pool-gated-off ("true" | "false" STRING!).
|
||||
|
||||
Money is handled as micros ints only; *_usd values are preserved verbatim as
|
||||
the raw strings the server sent (never re-parsed to float).
|
||||
"""
|
||||
"""Nous credits: parse ``x-nous-credits-*`` / ``x-nous-tool-pool-*`` response
|
||||
headers into a validated CreditsState (depletion = paid_access, subscription-cap
|
||||
used_fraction, warn-once schema-version gating) and drive the notice policy.
|
||||
Header contract: see ``_HEADER_FIELDS``. Money is micros ints only; ``*_usd``
|
||||
strings are preserved verbatim (never re-parsed to float)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
from typing import Any, Mapping, Optional
|
||||
|
||||
from utils import is_truthy_value
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Warn-once latch: emit the version-unsupported warning at most once per process.
|
||||
_version_warning_emitted: bool = False
|
||||
|
||||
# Valid denominator kinds (exhaustive set from the API contract).
|
||||
_version_warning_emitted: bool = False # warn-once latch (per process)
|
||||
_VALID_DENOMINATOR_KINDS = frozenset({"subscription_cap", "none"})
|
||||
|
||||
# USD format: optional leading minus, one-or-more digits, dot, exactly 2 digits.
|
||||
_USD_RE = re.compile(r"^-?\d+\.\d{2}$")
|
||||
|
||||
_SENTINEL = object() # singleton sentinel for "parse failed"
|
||||
_USD_RE = re.compile(r"^-?\d+\.\d{2}$") # optional minus, digits, exactly 2 decimals
|
||||
_SENTINEL = object() # "parse failed"
|
||||
|
||||
|
||||
def _safe_int(value: Any) -> Any:
|
||||
"""Exact int (money-safe) or ``_SENTINEL``. ``int()`` directly, NOT ``int(float())``
|
||||
(precision loss above 2**53 corrupts money); float-shaped strings fail."""
|
||||
if value is None:
|
||||
return _SENTINEL
|
||||
try:
|
||||
return int(str(value))
|
||||
return _SENTINEL if value is None else int(str(value))
|
||||
except (TypeError, ValueError):
|
||||
return _SENTINEL
|
||||
|
||||
|
||||
def _validate_usd(value: Optional[str]) -> bool:
|
||||
"""Return True iff value is a non-None string matching ^-?\\d+\\.\\d{2}$."""
|
||||
return value is not None and bool(_USD_RE.match(value))
|
||||
|
||||
|
||||
@dataclass
|
||||
class CreditsState:
|
||||
"""Full credits state parsed from x-nous-credits-* response headers."""
|
||||
"""Credits state parsed from x-nous-credits-* response headers."""
|
||||
|
||||
version: int = 0
|
||||
remaining_micros: int = 0
|
||||
remaining_usd: str = ""
|
||||
subscription_micros: int = 0 # SIGNED — may be negative (debt). ONLY field allowed negative.
|
||||
subscription_micros: int = 0 # SIGNED — the ONLY field allowed negative (debt)
|
||||
subscription_usd: str = ""
|
||||
subscription_limit_micros: Optional[int] = None # PAIRED + OPTIONAL (only when subscription_cap)
|
||||
subscription_limit_usd: Optional[str] = None
|
||||
@@ -83,7 +54,7 @@ class CreditsState:
|
||||
paid_access: bool = True # depletion keys off THIS == False, NEVER remaining==0
|
||||
disabled_reason: Optional[str] = None # header omitted entirely when null
|
||||
as_of_ms: int = 0
|
||||
captured_at: float = 0.0 # time.time() when this was captured
|
||||
captured_at: float = 0.0 # time.time() when captured
|
||||
from_header: bool = False # True only when populated by parse_credits_headers()
|
||||
|
||||
@property
|
||||
@@ -96,51 +67,44 @@ class CreditsState:
|
||||
|
||||
@property
|
||||
def depleted(self) -> bool:
|
||||
"""Keyed off ``paid_access == False`` ONLY — never ``remaining_micros == 0``,
|
||||
a false positive when the balance is zero but access is live (renewal pending)."""
|
||||
"""``paid_access == False`` ONLY — ``remaining_micros == 0`` is a false positive
|
||||
when the balance is zero but access is live (renewal pending)."""
|
||||
return not self.paid_access
|
||||
|
||||
@property
|
||||
def used_fraction(self) -> Optional[float]:
|
||||
"""Fraction of the subscription cap consumed, in [0.0, 1.0]; None without a
|
||||
computable denominator. Guarded on the LIMIT FIELD (the real denominator),
|
||||
not ``denominator_kind`` (metadata)."""
|
||||
if not isinstance(self.subscription_limit_micros, int) or self.subscription_limit_micros <= 0:
|
||||
"""Fraction of the subscription cap consumed in [0.0, 1.0]; None without a computable
|
||||
denominator. Guarded on the LIMIT FIELD (the real denominator), not ``denominator_kind``."""
|
||||
lim = self.subscription_limit_micros
|
||||
if not isinstance(lim, int) or lim <= 0:
|
||||
return None
|
||||
used = self.subscription_limit_micros - self.subscription_micros
|
||||
return max(0.0, min(1.0, used / self.subscription_limit_micros))
|
||||
return max(0.0, min(1.0, (lim - self.subscription_micros) / lim))
|
||||
|
||||
|
||||
# ── Credits policy constants. Switching notices sticky→TTL later also needs a
|
||||
# paired *_TTL_MS per notice kind (AgentNotice has the field; not plumbed yet).
|
||||
CREDITS_NOTICE_KIND = "sticky" # v1: credits notices are sticky
|
||||
CREDITS_RESTORED_TTL_MS = 8000 # the only TTL notice in v1 (depletion-recovery confirmation)
|
||||
|
||||
# Usage-gauge bands (ascending): (threshold_fraction, level, label_pct). One
|
||||
# escalating line showing the HIGHEST band reached (50 → 75 → 90); crossing up
|
||||
# replaces it, recovering steps it down. The policy derives everything from it.
|
||||
# escalating line shows the HIGHEST band reached; climbing replaces it, recovery steps down.
|
||||
CREDITS_USAGE_BANDS: tuple[tuple[float, str, int], ...] = ((0.50, "info", 50), (0.75, "warn", 75), (0.90, "warn", 90))
|
||||
CREDITS_USAGE_KEY = "credits.usage" # single key for the escalating usage notice
|
||||
|
||||
CREDITS_USAGE_KEY = "credits.usage"
|
||||
# Min subscription balance counting as "grant not yet spent" for the grant_spent
|
||||
# gate. 1¢: portal-seeded states (float dollars → micros) can carry sub-cent
|
||||
# residue where headers report 0 — without the floor such a seed opens the gate
|
||||
# and the first header re-creates the at-open nag.
|
||||
# gate. 1¢: portal-seeded states (float dollars → micros) can carry sub-cent residue
|
||||
# where headers report 0 — without the floor a seed opens the gate and the first
|
||||
# header re-creates the at-open nag.
|
||||
GRANT_UNSPENT_MIN_MICROS = 10_000
|
||||
|
||||
|
||||
def new_credits_latch() -> dict:
|
||||
"""Fresh notice latch for :func:`evaluate_credits_notices`. Every producer must
|
||||
build it here so a new gate key lands everywhere at once."""
|
||||
"""Fresh notice latch for :func:`evaluate_credits_notices`; every producer builds it here so a new gate key lands everywhere."""
|
||||
return {"active": set(), "seen_below_90": False, "usage_band": None, "seen_grant_unspent": False}
|
||||
|
||||
|
||||
@dataclass
|
||||
class AgentNotice:
|
||||
"""Driver-agnostic out-of-band notice, fired via ``AIAgent.notice_callback``
|
||||
(cleared via ``notice_clear_callback``); each driver renders its own way.
|
||||
``kind``/``ttl_ms`` stay expressive so a future config can switch v1's
|
||||
sticky credits notices to TTL without touching the policy."""
|
||||
"""Driver-agnostic out-of-band notice (``AIAgent.notice_callback`` / ``notice_clear_callback``); each driver
|
||||
renders its own way. ``kind``/``ttl_ms`` stay expressive so a future config can switch v1's sticky notices to TTL."""
|
||||
|
||||
text: str
|
||||
level: str = "info" # info | warn | error | success
|
||||
@@ -155,18 +119,11 @@ def _sticky_notice(text: str, level: str, key: str) -> AgentNotice:
|
||||
|
||||
|
||||
def is_free_tier_model(model: str, base_url: str = "") -> bool:
|
||||
"""Return True when *model* is a Nous free-tier model, using ONLY local data.
|
||||
|
||||
Zero-network signals: (1) ``:free`` suffix — canonical Nous free SKU marker;
|
||||
(2) ``stealth/`` prefix — stealth-preview SKUs are free without the suffix
|
||||
(naming-convention trust: a PAID ``stealth/`` model would wrongly suppress
|
||||
the banner); (3) a PEEK into ``hermes_cli.models``' pricing cache (filled by
|
||||
the model picker; a miss never fetches — gateway sessions never run the
|
||||
picker, so there only 1-2 apply).
|
||||
|
||||
Fail-open to False (depleted notice still shows): a wrong warning is
|
||||
recoverable noise; hiding it on a paid model masks a real block.
|
||||
"""
|
||||
"""True when *model* is a Nous free-tier model, using ONLY local data: (1) ``:free`` suffix — canonical
|
||||
Nous free SKU marker; (2) ``stealth/`` prefix — stealth-preview SKUs are free without the suffix
|
||||
(naming-convention trust: a PAID ``stealth/`` model would wrongly suppress the banner); (3) a PEEK into
|
||||
``hermes_cli.models``' pricing cache (filled by the model picker; a miss never fetches). Fail-open to
|
||||
False (depleted notice still shows): a wrong warning is recoverable noise; hiding it masks a real block."""
|
||||
if not model:
|
||||
return False
|
||||
if model.endswith(":free") or model.startswith("stealth/"):
|
||||
@@ -176,84 +133,55 @@ def is_free_tier_model(model: str, base_url: str = "") -> bool:
|
||||
try:
|
||||
from hermes_cli.models import _is_model_free, peek_cached_pricing
|
||||
|
||||
# peek_cached_pricing owns the /v1-suffix and auth-state key details.
|
||||
pricing = peek_cached_pricing(base_url)
|
||||
if not pricing:
|
||||
return False
|
||||
return _is_model_free(model, pricing)
|
||||
pricing = peek_cached_pricing(base_url) # owns the /v1-suffix and auth-state key details
|
||||
return bool(pricing) and _is_model_free(model, pricing)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def evaluate_credits_notices(
|
||||
state: CreditsState, latch: dict, *, model_is_free: bool = False,
|
||||
) -> tuple[list[AgentNotice], list[str]]:
|
||||
"""Reconcile credits notices against the latch (see :func:`new_credits_latch`).
|
||||
Mutates ``latch`` IN PLACE. Pure — no I/O, no agent/run_agent imports.
|
||||
|
||||
``model_is_free`` (see :func:`is_free_tier_model`) suppresses
|
||||
``credits.depleted`` — a depleted account on a free model keeps inferencing,
|
||||
so the banner is noise. Suppression does NOT emit "restored"; that fires
|
||||
only on a genuine ``paid_access`` flip back to True.
|
||||
|
||||
Returns ``(to_show, to_clear)``; caller emits to_clear FIRST, then to_show.
|
||||
"""
|
||||
def evaluate_credits_notices(state: CreditsState, latch: dict, *, model_is_free: bool = False) -> tuple[list[AgentNotice], list[str]]:
|
||||
"""Reconcile credits notices against the latch (see :func:`new_credits_latch`); mutates ``latch`` IN
|
||||
PLACE. Pure — no I/O, no agent/run_agent imports. ``model_is_free`` suppresses ``credits.depleted`` (a
|
||||
depleted account on a free model keeps inferencing) WITHOUT emitting "restored" — that fires only on a
|
||||
genuine ``paid_access`` flip back to True. Returns ``(to_show, to_clear)``; caller emits to_clear FIRST."""
|
||||
to_show: list[AgentNotice] = []
|
||||
to_clear: list[str] = []
|
||||
uf = state.used_fraction
|
||||
|
||||
uf, active = state.used_fraction, latch["active"]
|
||||
# Crossing latch: band notices fire only once uf was observed below the LOWEST
|
||||
# band, so a session opening mid-range doesn't fire on its first observation
|
||||
# (the cold-start seed primes this when it WANTS an open-high warning).
|
||||
if uf is not None and uf < CREDITS_USAGE_BANDS[0][0]:
|
||||
latch["seen_below_90"] = True
|
||||
|
||||
# Grant-spent gate: fires only after this session OBSERVED the grant unspent
|
||||
# (≥1¢). Opening at grant-spent is a steady STATE (/usage carries it), not an
|
||||
# event. Unlike seen_below_90, seeds must NOT prime this gate.
|
||||
if uf is not None and uf < 1.0 and state.subscription_micros >= GRANT_UNSPENT_MIN_MICROS:
|
||||
latch["seen_grant_unspent"] = True
|
||||
active = latch["active"]
|
||||
|
||||
# Highest band reached (ascending → last match wins); None below all.
|
||||
# Highest band reached (ascending → last match wins); None below all. Top-up
|
||||
# suppression: with purchased credits the cap gauge is the wrong denominator
|
||||
# ("90% used" on $50 of top-up is noise; it used to stick PERMANENTLY beside
|
||||
# grant_spent at >=100%) — grant_spent covers the cap-reached case, and a
|
||||
# mid-session top-up flips current_band → None so the clear path removes the line.
|
||||
current_band: Optional[tuple[float, str, int]] = None
|
||||
if uf is not None:
|
||||
for band in CREDITS_USAGE_BANDS:
|
||||
if uf >= band[0]:
|
||||
current_band = band
|
||||
# Top-up suppression: with purchased credits the cap gauge is the wrong
|
||||
# denominator ("90% used" on $50 of top-up is noise; it used to stick
|
||||
# PERMANENTLY beside grant_spent at >=100%). grant_spent below covers the
|
||||
# cap-reached case; a mid-session top-up flips current_band → None and the
|
||||
# clear path removes the band line.
|
||||
if state.purchased_micros > 0:
|
||||
current_band = None
|
||||
grant_cond = (
|
||||
state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
|
||||
)
|
||||
depleted_cond = not state.paid_access
|
||||
if uf is not None and state.purchased_micros <= 0:
|
||||
current_band = next((b for b in reversed(CREDITS_USAGE_BANDS) if uf >= b[0]), None)
|
||||
|
||||
# ── usage gauge: highest crossed band only; replace on band change (climb or
|
||||
# step-down); clear below the lowest band or when the denominator vanishes.
|
||||
shown_band = latch.get("usage_band")
|
||||
target_band = current_band[2] if (current_band and latch["seen_below_90"]) else None
|
||||
if target_band != shown_band:
|
||||
if target_band != latch.get("usage_band"):
|
||||
if CREDITS_USAGE_KEY in active:
|
||||
to_clear.append(CREDITS_USAGE_KEY)
|
||||
active.discard(CREDITS_USAGE_KEY)
|
||||
if target_band is not None:
|
||||
# Absolute dollars used (a bare "N%" is only meaningful against a Nous
|
||||
# cap): cap − remaining in micros, clamped [0, cap]; "$?" if a producer
|
||||
# set the limit without its *_usd. Re-emits on band change only.
|
||||
_cap_usd = state.subscription_limit_usd or "?"
|
||||
_level = current_band[1] # type: ignore[index] (current_band set when target_band set)
|
||||
_lim = state.subscription_limit_micros or 0
|
||||
_used_micros = max(0, min(_lim, _lim - state.subscription_micros))
|
||||
_used_usd = f"{_used_micros / 1_000_000:.2f}" if _lim else "?"
|
||||
_glyph = "⚠" if _level == "warn" else "•"
|
||||
to_show.append(
|
||||
_sticky_notice(f"{_glyph} You've used ${_used_usd} of your ${_cap_usd} cap", _level, CREDITS_USAGE_KEY)
|
||||
)
|
||||
# Absolute dollars used (a bare "N%" is only meaningful against a Nous cap): cap − remaining,
|
||||
# clamped [0, cap]; "$?" if a producer set the limit without its *_usd. Re-emits on band change only.
|
||||
level = current_band[1] # type: ignore[index] (current_band set when target_band set)
|
||||
lim = state.subscription_limit_micros or 0
|
||||
used_usd = f"{max(0, min(lim, lim - state.subscription_micros)) / 1_000_000:.2f}" if lim else "?"
|
||||
text = f"{'⚠' if level == 'warn' else '•'} You've used ${used_usd} of your ${state.subscription_limit_usd or '?'} cap"
|
||||
to_show.append(_sticky_notice(text, level, CREDITS_USAGE_KEY))
|
||||
active.add(CREDITS_USAGE_KEY)
|
||||
latch["usage_band"] = target_band
|
||||
|
||||
@@ -261,10 +189,11 @@ def evaluate_credits_notices(
|
||||
# announcement per crossing. A header flicker (uf → None → 1.0) clears the
|
||||
# line but cannot re-announce; only a renewal re-opening the gate (fresh ≥1¢
|
||||
# observation) arms the next. .get(): default closed for hand-built latches.
|
||||
grant_cond = (
|
||||
state.denominator_kind == "subscription_cap" and uf is not None and uf >= 1.0 and state.purchased_micros > 0
|
||||
)
|
||||
if grant_cond and "credits.grant_spent" not in active and latch.get("seen_grant_unspent", False):
|
||||
to_show.append(
|
||||
_sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent")
|
||||
)
|
||||
to_show.append(_sticky_notice(f"• Grant spent · ${state.purchased_usd} top-up left", "info", "credits.grant_spent"))
|
||||
active.add("credits.grant_spent")
|
||||
latch["seen_grant_unspent"] = False
|
||||
elif "credits.grant_spent" in active and not grant_cond:
|
||||
@@ -272,6 +201,7 @@ def evaluate_credits_notices(
|
||||
active.discard("credits.grant_spent")
|
||||
|
||||
# ── depleted: suppressed while the model is free (inference still works).
|
||||
depleted_cond = not state.paid_access
|
||||
show_depleted = depleted_cond and not model_is_free
|
||||
if show_depleted and "credits.depleted" not in active:
|
||||
to_show.append(_sticky_notice("✕ Credit access paused · run /topup to top up", "error", "credits.depleted"))
|
||||
@@ -279,9 +209,7 @@ def evaluate_credits_notices(
|
||||
elif "credits.depleted" in active and not show_depleted:
|
||||
to_clear.append("credits.depleted")
|
||||
active.discard("credits.depleted")
|
||||
if not depleted_cond:
|
||||
# Genuine recovery only — switching to a free model while still
|
||||
# depleted must NOT claim access was restored.
|
||||
if not depleted_cond: # genuine recovery only — a free-model switch while depleted is NOT "restored"
|
||||
to_show.append(AgentNotice(
|
||||
text="✓ Credit access restored", level="success", kind="ttl",
|
||||
ttl_ms=CREDITS_RESTORED_TTL_MS, key="credits.restored", id="credits.restored",
|
||||
@@ -289,41 +217,47 @@ def evaluate_credits_notices(
|
||||
return (to_show, to_clear)
|
||||
|
||||
|
||||
# (field, header, signed) — required micros fields; only subscription may be negative.
|
||||
_MICROS_FIELDS: tuple[tuple[str, str, bool], ...] = (
|
||||
("remaining_micros", "x-nous-credits-remaining-micros", False),
|
||||
("subscription_micros", "x-nous-credits-subscription-micros", True),
|
||||
("rollover_micros", "x-nous-credits-rollover-micros", False),
|
||||
("purchased_micros", "x-nous-credits-purchased-micros", False), ("as_of_ms", "x-nous-credits-as-of-ms", False),
|
||||
)
|
||||
_USD_FIELDS: tuple[tuple[str, str], ...] = (
|
||||
("remaining_usd", "x-nous-credits-remaining-usd"), ("subscription_usd", "x-nous-credits-subscription-usd"),
|
||||
("purchased_usd", "x-nous-credits-purchased-usd"),
|
||||
)
|
||||
# (field, header, default-when-absent) — "true"/"false" (case-insensitive) STRING flags.
|
||||
_BOOL_FIELDS: tuple[tuple[str, str, bool], ...] = (
|
||||
("paid_access", "x-nous-credits-paid-access", True), # absent → fail-open (assume access)
|
||||
("tool_pool_gated_off", "x-nous-tool-pool-gated-off", False),
|
||||
# Header contract: (field, kind[, default-when-absent]); a field is REQUIRED unless it has a default.
|
||||
# Header name = ``x-nous-credits-<field>`` (``x-nous-<field>`` for tool_pool_*), underscores → dashes.
|
||||
# micros: int >= 0 ("signed": may be negative); usd: the server's formatted string ^-?\d+\.\d{2}$
|
||||
# (never re-parsed); bool: "true"/"false" STRING. Handled inline: subscription-limit-* (PAIRED/optional),
|
||||
# denominator-kind ("subscription_cap" | "none"), disabled-reason (omitted when null).
|
||||
_HEADER_FIELDS: tuple[tuple, ...] = (
|
||||
("remaining_micros", "micros"), ("subscription_micros", "signed"), ("rollover_micros", "micros"),
|
||||
("purchased_micros", "micros"), ("as_of_ms", "micros"), ("tool_pool_micros", "micros", 0),
|
||||
("remaining_usd", "usd"), ("subscription_usd", "usd"), ("purchased_usd", "usd"),
|
||||
("paid_access", "bool", True), # absent → fail-open (assume access)
|
||||
("tool_pool_gated_off", "bool", False),
|
||||
)
|
||||
|
||||
|
||||
def _header_name(field: str) -> str:
|
||||
return "x-nous-" + ("" if field.startswith("tool_pool_") else "credits-") + field.replace("_", "-")
|
||||
|
||||
|
||||
def _parse_field(kind: str, raw: Optional[str], default: Any = _SENTINEL) -> Any:
|
||||
"""One header value → field value; ``default`` when absent, ``_SENTINEL`` on a contract violation."""
|
||||
if raw is None:
|
||||
return default
|
||||
if kind in ("micros", "signed"):
|
||||
val = _safe_int(raw)
|
||||
return _SENTINEL if val is _SENTINEL or (kind == "micros" and val < 0) else val
|
||||
if kind == "usd":
|
||||
return raw if _USD_RE.match(raw) else _SENTINEL
|
||||
flag = raw.strip().lower()
|
||||
return _SENTINEL if flag not in ("true", "false") else flag == "true"
|
||||
|
||||
|
||||
def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Optional[CreditsState]:
|
||||
"""Parse x-nous-credits-* (and x-nous-tool-pool-*) headers into a CreditsState.
|
||||
|
||||
Returns None (miss) on ANY of: no ``x-nous-credits-version`` header; version
|
||||
!= 1 (> 1 also warns once); a ``*_micros``/``as_of_ms`` field non-integer or
|
||||
negative (subscription excepted); a ``*_usd`` not matching
|
||||
``^-?\\d+\\.\\d{2}$``; ``denominator_kind`` outside {"subscription_cap",
|
||||
"none"}; ``paid_access``/``tool_pool_gated_off`` not exactly "true"/"false";
|
||||
any unexpected exception.
|
||||
|
||||
Fail-open on the subscription_limit pair: a half-pair (only -micros or only
|
||||
-usd) is treated as both-absent — the parse STILL SUCCEEDS with both None.
|
||||
"""
|
||||
None (miss) on ANY of: no version header; version != 1 (> 1 also warns once);
|
||||
a required field violating ``_HEADER_FIELDS``; unknown ``denominator_kind``;
|
||||
any unexpected exception. Fail-open on the subscription_limit pair: a
|
||||
half-pair (only -micros or only -usd) parses as both-absent (both None)."""
|
||||
global _version_warning_emitted
|
||||
try:
|
||||
# Cheap probe before the lowercase copy (header names are case-insensitive):
|
||||
# bail when the version header is absent — the hot path for non-Nous providers.
|
||||
# Cheap probe before the lowercase copy (header names are case-insensitive): bail when the
|
||||
# version header is absent — the hot path for non-Nous providers.
|
||||
if not any(k.lower() == "x-nous-credits-version" for k in headers):
|
||||
return None
|
||||
lowered = {k.lower(): v for k, v in headers.items()}
|
||||
@@ -335,80 +269,39 @@ def parse_credits_headers(headers: Mapping[str, str], provider: str = "") -> Opt
|
||||
_version_warning_emitted = True
|
||||
logger.warning("credits header version %d unsupported, ignoring — update Hermes", version_val)
|
||||
return None
|
||||
fields: dict[str, Any] = {}
|
||||
for name, key, signed in _MICROS_FIELDS:
|
||||
val = _safe_int(lowered.get(key))
|
||||
if val is _SENTINEL or (not signed and val < 0):
|
||||
return None
|
||||
fields[name] = val
|
||||
|
||||
# tool_pool_micros is OPTIONAL: absent → 0; present-but-invalid → miss.
|
||||
_tp_raw = lowered.get("x-nous-tool-pool-micros")
|
||||
_tp_val = 0 if _tp_raw is None else _safe_int(_tp_raw)
|
||||
if _tp_val is _SENTINEL or _tp_val < 0:
|
||||
return None
|
||||
fields["tool_pool_micros"] = _tp_val
|
||||
for name, key in _USD_FIELDS:
|
||||
val = lowered.get(key, "")
|
||||
if not _validate_usd(val):
|
||||
return None
|
||||
fields[name] = val
|
||||
|
||||
# subscription_limit_* PAIRED + OPTIONAL: both present → validate both
|
||||
# (any invalid → miss); half-pair or both absent → both None, parse continues.
|
||||
sub_limit_micros_raw = lowered.get("x-nous-credits-subscription-limit-micros")
|
||||
sub_limit_usd_raw = lowered.get("x-nous-credits-subscription-limit-usd")
|
||||
if sub_limit_micros_raw is not None and sub_limit_usd_raw is not None:
|
||||
lm = _safe_int(sub_limit_micros_raw)
|
||||
if lm is _SENTINEL or lm < 0 or not _validate_usd(sub_limit_usd_raw):
|
||||
return None
|
||||
fields["subscription_limit_micros"] = lm
|
||||
fields["subscription_limit_usd"] = sub_limit_usd_raw
|
||||
fields: dict[str, Any] = {
|
||||
name: _parse_field(kind, lowered.get(_header_name(name)), *default) for name, kind, *default in _HEADER_FIELDS
|
||||
}
|
||||
lim_micros_raw = lowered.get("x-nous-credits-subscription-limit-micros")
|
||||
lim_usd_raw = lowered.get("x-nous-credits-subscription-limit-usd")
|
||||
if lim_micros_raw is not None and lim_usd_raw is not None:
|
||||
fields["subscription_limit_micros"] = _parse_field("micros", lim_micros_raw)
|
||||
fields["subscription_limit_usd"] = _parse_field("usd", lim_usd_raw)
|
||||
denominator_kind = lowered.get("x-nous-credits-denominator-kind", "none")
|
||||
if denominator_kind not in _VALID_DENOMINATOR_KINDS:
|
||||
if _SENTINEL in fields.values() or denominator_kind not in _VALID_DENOMINATOR_KINDS:
|
||||
return None
|
||||
for name, key, default in _BOOL_FIELDS:
|
||||
if key not in lowered:
|
||||
fields[name] = default
|
||||
continue
|
||||
raw = lowered[key].strip().lower()
|
||||
if raw not in ("true", "false"):
|
||||
return None
|
||||
fields[name] = raw == "true"
|
||||
return CreditsState(
|
||||
version=version_val,
|
||||
denominator_kind=denominator_kind,
|
||||
disabled_reason=lowered.get("x-nous-credits-disabled-reason"), # None if absent (omitted when null)
|
||||
captured_at=time.time(),
|
||||
from_header=True,
|
||||
**fields,
|
||||
)
|
||||
except Exception:
|
||||
# Fail-open → miss; breadcrumb distinguishes a parser regression from a
|
||||
# legitimate no-headers response.
|
||||
disabled_reason = lowered.get("x-nous-credits-disabled-reason") # None if absent (omitted when null)
|
||||
return CreditsState(version=version_val, denominator_kind=denominator_kind, disabled_reason=disabled_reason,
|
||||
captured_at=time.time(), from_header=True, **fields)
|
||||
except Exception: # fail-open → miss; the breadcrumb distinguishes a parser regression from a no-headers response
|
||||
logger.debug("credits ▸ parse_credits_headers raised (fail-open miss)", exc_info=True)
|
||||
return None
|
||||
|
||||
|
||||
# ── Dev fixtures (HERMES_DEV_CREDITS_FIXTURE): throwaway scaffolding to trigger
|
||||
# any notice state without real spend. Value is a state NAME or a FILE PATH whose
|
||||
# contents are a name (re-read every turn → `echo depleted > /tmp/cf` flips live).
|
||||
# Drives the per-turn capture/notice path, the cold-start seed, and /usage.
|
||||
# ── Dev fixtures (HERMES_DEV_CREDITS_FIXTURE): throwaway scaffolding to trigger any notice state
|
||||
# without real spend. Value is a state NAME or a FILE PATH whose contents are a name (re-read every
|
||||
# turn → `echo depleted > /tmp/cf` flips live). Drives per-turn notices, the cold-start seed, and /usage.
|
||||
def _fixture(remaining: str, subscription: str, limit: Optional[str] = None, purchased: Optional[str] = None,
|
||||
*, paid: bool = True, reason: Optional[str] = None) -> dict:
|
||||
"""Fixture spec from *_usd strings; micros derived exactly (Decimal)."""
|
||||
from decimal import Decimal
|
||||
d: dict = {}
|
||||
for field, usd in (("remaining", remaining), ("subscription", subscription),
|
||||
("subscription_limit", limit), ("purchased", purchased)):
|
||||
for field, usd in (("remaining", remaining), ("subscription", subscription), ("subscription_limit", limit), ("purchased", purchased)):
|
||||
if usd is not None:
|
||||
d[f"{field}_micros"], d[f"{field}_usd"] = int(Decimal(usd) * 1_000_000), usd
|
||||
if limit is not None:
|
||||
d["denominator_kind"] = "subscription_cap"
|
||||
d["paid_access"] = paid
|
||||
if reason is not None:
|
||||
d["disabled_reason"] = reason
|
||||
return d
|
||||
return d if reason is None else {**d, "disabled_reason": reason}
|
||||
|
||||
|
||||
_DEV_FIXTURES: dict[str, dict] = {
|
||||
@@ -426,59 +319,44 @@ _DEV_FIXTURES: dict[str, dict] = {
|
||||
|
||||
|
||||
def dev_fixture_credits_state() -> Optional[CreditsState]:
|
||||
"""Return a fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None
|
||||
(unknown name / "clear" / "none" / unset → None).
|
||||
|
||||
Hard prod-leak guard: applies ONLY when HERMES_DEV_CREDITS is also on, so a
|
||||
stray fixture env var can never surface fabricated balances on a real account.
|
||||
"""
|
||||
if not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")):
|
||||
"""Fixture CreditsState for HERMES_DEV_CREDITS_FIXTURE, or None (unknown name / unset). Prod-leak guard:
|
||||
applies ONLY when HERMES_DEV_CREDITS is also on, so a stray fixture env var never surfaces fabricated balances."""
|
||||
name = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
|
||||
if not name or not is_truthy_value(os.environ.get("HERMES_DEV_CREDITS")):
|
||||
return None
|
||||
raw = os.environ.get("HERMES_DEV_CREDITS_FIXTURE", "").strip()
|
||||
if not raw:
|
||||
return None
|
||||
name = raw
|
||||
if os.path.sep in raw or "/" in raw: # looks like a path → read the name from the file
|
||||
if os.path.sep in name or "/" in name: # looks like a path → read the name from the file
|
||||
try:
|
||||
with open(raw, "r", encoding="utf-8") as fh:
|
||||
with open(name, "r", encoding="utf-8") as fh:
|
||||
name = fh.read().strip()
|
||||
except OSError:
|
||||
return None
|
||||
spec = _DEV_FIXTURES.get(name.lower())
|
||||
if not spec:
|
||||
if not (spec := _DEV_FIXTURES.get(name.lower())):
|
||||
return None
|
||||
# Stamp what the REAL parser always guarantees so a fixture is field-identical
|
||||
# to a parse_credits_headers() result (differential test): version 1, and a
|
||||
# valid purchased_usd (a zero-top-up account still carries "0.00").
|
||||
merged = {"version": 1, "purchased_usd": "0.00", **spec}
|
||||
return CreditsState(**merged, from_header=True, captured_at=time.time())
|
||||
# Stamp what the REAL parser always guarantees so a fixture is field-identical to a
|
||||
# parse_credits_headers() result: version 1 and a valid purchased_usd (zero top-up = "0.00").
|
||||
return CreditsState(**{"version": 1, "purchased_usd": "0.00", **spec}, from_header=True, captured_at=time.time())
|
||||
|
||||
|
||||
def _credits_state_from_account(info) -> Optional[CreditsState]:
|
||||
"""Map a NousPortalAccountInfo into a header-shaped CreditsState for the seed.
|
||||
Float account dollars → micros plus a DISPLAY *_usd (formatting account floats
|
||||
is allowed; parsing a server *_usd is not). Fail-open → None."""
|
||||
"""Map a NousPortalAccountInfo into a header-shaped CreditsState for the seed. Float account dollars →
|
||||
micros plus a DISPLAY *_usd (formatting account floats is allowed; parsing a server *_usd is not). Fail-open → None."""
|
||||
try:
|
||||
_acc = getattr(info, "paid_service_access_info", None)
|
||||
_sub = getattr(info, "subscription", None)
|
||||
acc = getattr(info, "paid_service_access_info", None)
|
||||
sub = getattr(info, "subscription", None)
|
||||
|
||||
def _money(dollars) -> tuple[int, str]: # (micros, display usd); (0, "") when absent
|
||||
if isinstance(dollars, (int, float)):
|
||||
return int(round(dollars * 1_000_000)), f"{dollars:.2f}"
|
||||
return 0, ""
|
||||
_remaining = _money(getattr(_acc, "total_usable_credits", None))
|
||||
_sub_rem = _money(getattr(_acc, "subscription_credits_remaining", None))
|
||||
_purchased = _money(getattr(_acc, "purchased_credits_remaining", None))
|
||||
_monthly = getattr(_sub, "monthly_credits", None)
|
||||
_cap = _money(_monthly) if isinstance(_monthly, (int, float)) and _monthly > 0 else (None, None)
|
||||
_paid = getattr(info, "paid_service_access", None)
|
||||
return (int(round(dollars * 1_000_000)), f"{dollars:.2f}") if isinstance(dollars, (int, float)) else (0, "")
|
||||
fields: dict[str, Any] = {}
|
||||
for prefix, attr in (("remaining", "total_usable_credits"), ("subscription", "subscription_credits_remaining"),
|
||||
("purchased", "purchased_credits_remaining")):
|
||||
fields[f"{prefix}_micros"], fields[f"{prefix}_usd"] = _money(getattr(acc, attr, None))
|
||||
monthly = getattr(sub, "monthly_credits", None)
|
||||
cap = _money(monthly) if isinstance(monthly, (int, float)) and monthly > 0 else (None, None)
|
||||
paid = getattr(info, "paid_service_access", None)
|
||||
return CreditsState(
|
||||
remaining_micros=_remaining[0], remaining_usd=_remaining[1], subscription_micros=_sub_rem[0],
|
||||
subscription_usd=_sub_rem[1], subscription_limit_micros=_cap[0], subscription_limit_usd=_cap[1],
|
||||
purchased_micros=_purchased[0], purchased_usd=_purchased[1],
|
||||
rollover_micros=_money(getattr(_sub, "rollover_credits", None))[0],
|
||||
denominator_kind="subscription_cap" if _cap[0] is not None else "none",
|
||||
paid_access=_paid if isinstance(_paid, bool) else True, from_header=False, captured_at=time.time(),
|
||||
**fields, subscription_limit_micros=cap[0], subscription_limit_usd=cap[1], from_header=False, captured_at=time.time(),
|
||||
rollover_micros=_money(getattr(sub, "rollover_credits", None))[0], paid_access=paid if isinstance(paid, bool) else True,
|
||||
denominator_kind="subscription_cap" if cap[0] is not None else "none",
|
||||
)
|
||||
except Exception:
|
||||
logger.debug("credits ▸ seed account→state mapping failed", exc_info=True)
|
||||
@@ -486,63 +364,45 @@ def _credits_state_from_account(info) -> Optional[CreditsState]:
|
||||
|
||||
|
||||
def _hydrate_seed_state(agent, state) -> None:
|
||||
"""Install a seed CreditsState on the agent and fire the notice policy once.
|
||||
Primes the crossing gate: the cold-start snapshot IS the first observation, so
|
||||
a session opening in a band warns immediately. Safe from a worker thread."""
|
||||
"""Install a seed CreditsState on the agent and fire the notice policy once. Primes the crossing gate:
|
||||
the cold-start snapshot IS the first observation, so a session opening in a band warns immediately."""
|
||||
agent._credits_state = state
|
||||
if getattr(agent, "_credits_session_start_micros", None) is None:
|
||||
agent._credits_session_start_micros = state.remaining_micros
|
||||
_latch = getattr(agent, "_credits_latch", None)
|
||||
if isinstance(_latch, dict) and state.used_fraction is not None:
|
||||
# Prime ONLY seen_below_90. Never prime seen_grant_unspent: a seed
|
||||
# observing grant-spent is a steady state; priming revives the nag.
|
||||
_latch["seen_below_90"] = True
|
||||
emit = getattr(agent, "_emit_credits_notices", None)
|
||||
if callable(emit):
|
||||
latch = getattr(agent, "_credits_latch", None)
|
||||
if isinstance(latch, dict) and state.used_fraction is not None:
|
||||
latch["seen_below_90"] = True # ONLY this gate — priming seen_grant_unspent would revive the steady-state nag
|
||||
if callable(emit := getattr(agent, "_emit_credits_notices", None)):
|
||||
emit()
|
||||
|
||||
|
||||
def seed_credits_at_session_start(agent) -> bool:
|
||||
"""Hydrate agent._credits_state from the portal account (or a dev fixture) and
|
||||
fire the notice policy so warnings show at session OPEN. Shared by the
|
||||
TUI/desktop build ("ready") and first-turn setup (plain-CLI fallback).
|
||||
Idempotent once a seed or real header populated _credits_state.
|
||||
|
||||
Returns True iff it seeded this call. Never raises — credits must never block startup.
|
||||
"""
|
||||
"""Hydrate agent._credits_state from the portal account (or a dev fixture) and fire the notice policy so
|
||||
warnings show at session OPEN (TUI/desktop "ready" and plain-CLI first-turn setup). Idempotent once a seed
|
||||
or real header populated _credits_state. Returns True iff it seeded this call. Never raises."""
|
||||
try:
|
||||
if getattr(agent, "provider", "") != "nous":
|
||||
return False
|
||||
if getattr(agent, "_credits_state", None) is not None:
|
||||
if getattr(agent, "provider", "") != "nous" or getattr(agent, "_credits_state", None) is not None:
|
||||
return False
|
||||
try:
|
||||
fixture = dev_fixture_credits_state()
|
||||
except Exception:
|
||||
fixture = None
|
||||
if fixture is not None:
|
||||
# Synchronous: a fixture is instant, and tests rely on the state +
|
||||
# notice landing before this returns.
|
||||
if fixture is not None: # synchronous: instant, and tests rely on state + notice landing before return
|
||||
_hydrate_seed_state(agent, fixture)
|
||||
return True
|
||||
|
||||
# FIRE-AND-FORGET: a slow portal must never delay "ready". The daemon
|
||||
# thread re-checks idempotency (a live header may land first).
|
||||
import threading
|
||||
|
||||
def _bg_seed() -> None:
|
||||
def _bg_seed() -> None: # FIRE-AND-FORGET: a slow portal must never delay "ready"
|
||||
try:
|
||||
from hermes_cli.nous_account import get_nous_portal_account_info
|
||||
info = get_nous_portal_account_info(force_fresh=True)
|
||||
if getattr(agent, "_credits_state", None) is not None:
|
||||
return # a live inference header beat us — don't clobber it
|
||||
state = _credits_state_from_account(info)
|
||||
if state is not None:
|
||||
if (state := _credits_state_from_account(info)) is not None:
|
||||
_hydrate_seed_state(agent, state)
|
||||
except Exception:
|
||||
logger.debug("credits ▸ session-start seed (background) failed", exc_info=True)
|
||||
threading.Thread(target=_bg_seed, name="credits-seed", daemon=True).start()
|
||||
return True
|
||||
except Exception:
|
||||
# Innermost log across all call sites so a dead seed is diagnosable.
|
||||
logger.debug("credits ▸ session-start seed failed (fail-open)", exc_info=True)
|
||||
logger.debug("credits ▸ session-start seed failed (fail-open)", exc_info=True) # innermost log: diagnosable dead seed
|
||||
return False
|
||||
|
||||
+9
-14
@@ -1,12 +1,11 @@
|
||||
"""Bounded fast-mode windows (``/fast auto`` and ``/fast cold``).
|
||||
|
||||
``agent.service_tier`` is ``None`` (normal), ``"priority"`` (static fast, pinned
|
||||
into ``agent.request_overrides`` at build time), ``"auto"`` (every user turn opens
|
||||
a window of ``agent.fast_auto_seconds``) or ``"cold"`` (only a session's first
|
||||
turn, no prior history, opens it). The provider's fast override is layered onto
|
||||
the request kwargs only while the window is open. Only per-request params
|
||||
(``service_tier`` / ``speed``) vary — system prompt, tools, and messages are
|
||||
untouched, so the prompt cache survives the boundary.
|
||||
``agent.service_tier``: ``None`` (normal), ``"priority"`` (static fast, pinned into
|
||||
``agent.request_overrides`` at build time), ``"auto"`` (every user turn opens a
|
||||
window of ``agent.fast_auto_seconds``) or ``"cold"`` (only a session's first turn,
|
||||
no prior history, opens it). The provider's fast override is layered onto request
|
||||
kwargs only while the window is open; only per-request params (``service_tier`` /
|
||||
``speed``) vary, so the prompt cache survives the boundary.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -39,17 +38,13 @@ def begin_turn(agent: Any, conversation_history: Any) -> None:
|
||||
def effective_request_overrides(agent: Any) -> dict[str, Any]:
|
||||
"""``agent.request_overrides`` plus the fast override while the window is open."""
|
||||
overrides = dict(getattr(agent, "request_overrides", None) or {})
|
||||
if getattr(agent, "service_tier", None) not in BOUNDED_MODES or time.monotonic() >= getattr(
|
||||
agent, "_fast_until", 0.0
|
||||
):
|
||||
if getattr(agent, "service_tier", None) not in BOUNDED_MODES or time.monotonic() >= getattr(agent, "_fast_until", 0.0):
|
||||
return overrides
|
||||
from hermes_cli.models import resolve_fast_mode_overrides
|
||||
base_url = getattr(agent, "base_url", None)
|
||||
if getattr(agent, "api_mode", None) == "anthropic_messages":
|
||||
base_url = getattr(agent, "_anthropic_base_url", None) or base_url
|
||||
fast = resolve_fast_mode_overrides(
|
||||
getattr(agent, "model", None), provider=getattr(agent, "provider", None), base_url=base_url
|
||||
overrides.update(
|
||||
resolve_fast_mode_overrides(getattr(agent, "model", None), provider=getattr(agent, "provider", None), base_url=base_url) or {}
|
||||
)
|
||||
if fast:
|
||||
overrides.update(fast)
|
||||
return overrides
|
||||
|
||||
+377
-573
File diff suppressed because it is too large
Load Diff
+16
-27
@@ -5,37 +5,30 @@ from __future__ import annotations
|
||||
import math
|
||||
from typing import Any, Dict
|
||||
|
||||
# Gemini's ``FunctionDeclaration.parameters`` accepts the ``Schema`` object,
|
||||
# only a subset of OpenAPI 3.0 / JSON Schema; everything else is stripped.
|
||||
# Gemini's ``FunctionDeclaration.parameters`` accepts only a subset of OpenAPI 3.0 /
|
||||
# JSON Schema (the ``Schema`` object); everything else is stripped.
|
||||
_GEMINI_SCHEMA_ALLOWED_KEYS = {
|
||||
"type", "format", "title", "description", "nullable", "enum",
|
||||
"maxItems", "minItems", "properties", "required", "minProperties",
|
||||
"maxProperties", "minLength", "maxLength", "pattern", "example",
|
||||
"anyOf", "propertyOrdering", "default", "items", "minimum", "maximum",
|
||||
"type", "format", "title", "description", "nullable", "enum", "maxItems", "minItems", "properties", "required",
|
||||
"minProperties", "maxProperties", "minLength", "maxLength", "pattern", "example", "anyOf", "propertyOrdering",
|
||||
"default", "items", "minimum", "maximum",
|
||||
}
|
||||
|
||||
|
||||
def _stringify_enum_value(item: Any) -> Any:
|
||||
"""Gemini-safe string for a scalar enum entry, or None to drop it."""
|
||||
if isinstance(item, str):
|
||||
return item
|
||||
if isinstance(item, bool):
|
||||
return "true" if item else "false"
|
||||
if isinstance(item, (int, float)) and math.isfinite(item):
|
||||
return str(item)
|
||||
return None
|
||||
return item if isinstance(item, str) else None
|
||||
|
||||
|
||||
def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
|
||||
"""Return a Gemini-compatible copy of a tool parameter schema.
|
||||
|
||||
Keeps only the documented Gemini subset (drops e.g. ``$schema`` /
|
||||
``additionalProperties``) and recursively sanitizes nested
|
||||
``properties`` / ``items`` / ``anyOf`` definitions.
|
||||
"""
|
||||
"""Gemini-compatible copy of a tool parameter schema: keeps only the documented subset
|
||||
(drops e.g. ``$schema`` / ``additionalProperties``) and recursively sanitizes nested
|
||||
``properties`` / ``items`` / ``anyOf``."""
|
||||
if not isinstance(schema, dict):
|
||||
return {}
|
||||
|
||||
cleaned: Dict[str, Any] = {}
|
||||
for key, value in schema.items():
|
||||
if key not in _GEMINI_SCHEMA_ALLOWED_KEYS:
|
||||
@@ -53,31 +46,27 @@ def sanitize_gemini_schema(schema: Any) -> Dict[str, Any]:
|
||||
|
||||
# Gemini requires every ``enum`` entry to be a string even for
|
||||
# integer/number/boolean types; the declared type stays intact and Gemini
|
||||
# still emits typed tool arguments at runtime.
|
||||
# still emits typed tool arguments at runtime. dict.fromkeys = ordered dedupe.
|
||||
enum_val = cleaned.get("enum")
|
||||
if isinstance(enum_val, list) and cleaned.get("type") in {"integer", "number", "boolean"}:
|
||||
# dict.fromkeys = order-preserving dedupe.
|
||||
stringified = list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None))
|
||||
if stringified:
|
||||
if stringified := list(dict.fromkeys(v for v in map(_stringify_enum_value, enum_val) if v is not None)):
|
||||
cleaned["enum"] = stringified
|
||||
else:
|
||||
cleaned.pop("enum", None)
|
||||
|
||||
# Gemini validates ``required`` strictly against the same node's
|
||||
# ``properties`` (HTTP 400 "property is not defined"), and one bad tool
|
||||
# schema fails the ENTIRE request. MCP servers routinely emit ``required``
|
||||
# without ``properties``, so keep only names that exist here; the tool
|
||||
# handler still validates required fields at execution time.
|
||||
# Gemini validates ``required`` strictly against the same node's ``properties`` (HTTP 400
|
||||
# "property is not defined") and one bad tool schema fails the ENTIRE request. MCP servers
|
||||
# routinely emit ``required`` without ``properties``, so keep only names that exist here;
|
||||
# the tool handler still validates required fields at execution time.
|
||||
required_val = cleaned.get("required")
|
||||
if isinstance(required_val, list):
|
||||
props_val = cleaned.get("properties")
|
||||
prop_names = set(props_val.keys()) if isinstance(props_val, dict) else set()
|
||||
prop_names = set(props_val) if isinstance(props_val, dict) else set()
|
||||
valid_required = [name for name in required_val if isinstance(name, str) and name in prop_names]
|
||||
if not valid_required:
|
||||
cleaned.pop("required", None)
|
||||
elif len(valid_required) != len(required_val):
|
||||
cleaned["required"] = valid_required
|
||||
|
||||
return cleaned
|
||||
|
||||
|
||||
|
||||
+6
-10
@@ -1,10 +1,8 @@
|
||||
"""Best-effort early import of the OpenAI SDK's native streaming parser.
|
||||
|
||||
On some Windows installs ``jiter``'s native extension imports fine from the venv
|
||||
but fails when first imported later inside the threaded streaming path. Loading
|
||||
it once at agent-package import avoids that while preserving the SDK's normal
|
||||
error path for genuinely broken installs.
|
||||
"""
|
||||
"""Best-effort early import of the OpenAI SDK's native streaming parser: on some
|
||||
Windows installs ``jiter``'s native extension imports fine from the venv but fails
|
||||
when first imported inside the threaded streaming path. Loading it once at
|
||||
agent-package import avoids that while keeping the SDK's normal error path for
|
||||
genuinely broken installs."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -15,7 +13,6 @@ _JITER_PRELOAD_ERROR: Exception | None = None
|
||||
|
||||
|
||||
def preload_jiter_native_extension() -> bool:
|
||||
"""Import jiter's native extension early if it is available."""
|
||||
global _JITER_PRELOADED, _JITER_PRELOAD_ERROR
|
||||
if _JITER_PRELOADED:
|
||||
return True
|
||||
@@ -25,8 +22,7 @@ def preload_jiter_native_extension() -> bool:
|
||||
except Exception as exc:
|
||||
_JITER_PRELOAD_ERROR = exc
|
||||
return False
|
||||
_JITER_PRELOADED = True
|
||||
_JITER_PRELOAD_ERROR = None
|
||||
_JITER_PRELOADED, _JITER_PRELOAD_ERROR = True, None
|
||||
return True
|
||||
|
||||
|
||||
|
||||
+7
-10
@@ -1,28 +1,25 @@
|
||||
"""Lazy forwarding helpers shared by the AIAgent façade and its mixins.
|
||||
|
||||
Targets are resolved on every call, so ``patch("<module>.<name>")`` in tests still intercepts and the
|
||||
heavy agent modules stay off the ``run_agent`` import path.
|
||||
Targets are resolved on every call, so ``patch("<module>.<name>")`` in tests still
|
||||
intercepts and the heavy agent modules stay off the ``run_agent`` import path.
|
||||
"""
|
||||
import importlib
|
||||
|
||||
|
||||
def lazy_attr(module: str, name: str):
|
||||
"""Resolve ``module.name`` at call time (keeps heavy/cyclic imports off the run_agent import path)."""
|
||||
"""Resolve ``module.name`` at call time."""
|
||||
return getattr(importlib.import_module(module), name)
|
||||
|
||||
|
||||
def forward(module: str, name: str, *, static: bool = False):
|
||||
"""Build an AIAgent method that lazily forwards to ``module.name`` (``target(self, *args, **kwargs)``).
|
||||
|
||||
The target is resolved on every call, so ``patch("<module>.<name>")`` in tests still intercepts and
|
||||
run_agent keeps its import-time cost. ``static=True`` drops ``self``.
|
||||
"""
|
||||
"""Build an AIAgent method that lazily forwards to ``module.name``
|
||||
(``target(self, *args, **kwargs)``; ``static=True`` drops ``self``)."""
|
||||
if static:
|
||||
def forwarder(*args, **kwargs):
|
||||
return getattr(importlib.import_module(module), name)(*args, **kwargs)
|
||||
return lazy_attr(module, name)(*args, **kwargs)
|
||||
else:
|
||||
def forwarder(self, *args, **kwargs):
|
||||
return getattr(importlib.import_module(module), name)(self, *args, **kwargs)
|
||||
return lazy_attr(module, name)(self, *args, **kwargs)
|
||||
forwarder.__name__ = forwarder.__qualname__ = name
|
||||
forwarder.__doc__ = f"Forwarder — see ``{module}.{name}``."
|
||||
return staticmethod(forwarder) if static else forwarder
|
||||
|
||||
+17
-30
@@ -1,41 +1,30 @@
|
||||
"""LM Studio reasoning-effort resolution shared by the chat-completions
|
||||
transport and run_agent's iteration-limit summary path.
|
||||
|
||||
LM Studio publishes per-model ``capabilities.reasoning.allowed_options``
|
||||
(``["off","on"]`` for toggle models, ``["off","minimal","low"]`` for graduated
|
||||
ones). We map the user's ``reasoning_config`` onto LM Studio's OpenAI-compatible
|
||||
vocabulary, then clamp against the model's allowed set so the server doesn't 400.
|
||||
"""
|
||||
"""LM Studio reasoning-effort resolution (chat-completions transport + run_agent's
|
||||
iteration-limit summary path). LM Studio publishes per-model
|
||||
``capabilities.reasoning.allowed_options`` (``["off","on"]`` for toggle models,
|
||||
``["off","minimal","low"]`` for graduated ones); the user's ``reasoning_config`` is
|
||||
mapped onto LM Studio's vocabulary, then clamped to the allowed set so the server
|
||||
doesn't 400."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import List, Optional
|
||||
|
||||
# Top-level reasoning_effort values LM Studio's OpenAI-compatible endpoint accepts.
|
||||
_LM_VALID_EFFORTS = {"none", "minimal", "low", "medium", "high", "xhigh"}
|
||||
|
||||
# Toggle-style models publish allowed_options as ["off","on"]; map onto the
|
||||
# request vocabulary. Also applied to the published allowed_options themselves.
|
||||
# Toggle vocabulary → request vocabulary; also applied to published allowed_options.
|
||||
_LM_EFFORT_ALIASES = {"off": "none", "on": "medium"}
|
||||
|
||||
# Hermes' ladder grew past LM Studio's vocabulary ("max", "ultra"). Without this
|
||||
# ceiling clamp they miss _LM_VALID_EFFORTS, keep the "medium" default and are
|
||||
# conflated with unparseable input — asking for more yields less than "xhigh".
|
||||
# Kept separate from _LM_EFFORT_ALIASES, which must not rewrite allowed_options.
|
||||
# Hermes' ladder grew past LM Studio's vocabulary ("max", "ultra"); without this
|
||||
# ceiling clamp they'd fall to the "medium" default (more yields less than "xhigh").
|
||||
# Separate from _LM_EFFORT_ALIASES, which must not rewrite allowed_options.
|
||||
_LM_EFFORT_CLAMP = {"max": "xhigh", "ultra": "xhigh"}
|
||||
|
||||
|
||||
def resolve_lmstudio_effort(
|
||||
reasoning_config: Optional[dict],
|
||||
allowed_options: Optional[List[str]],
|
||||
) -> Optional[str]:
|
||||
"""Return the ``reasoning_effort`` to send to LM Studio, or ``None``.
|
||||
|
||||
``None`` means "omit the field": the user picked a level the model can't
|
||||
honor, so LM Studio falls back to the model's declared default rather than
|
||||
a silently substituted effort. Falsy ``allowed_options`` (probe failed)
|
||||
skips clamping and sends the resolved effort anyway.
|
||||
"""
|
||||
def resolve_lmstudio_effort(reasoning_config: Optional[dict], allowed_options: Optional[List[str]]) -> Optional[str]:
|
||||
"""Return the ``reasoning_effort`` to send to LM Studio, or ``None`` = omit the
|
||||
field (the user picked a level the model can't honor, so LM Studio falls back
|
||||
to the model's declared default rather than a silently substituted effort).
|
||||
Falsy ``allowed_options`` (probe failed) skips clamping."""
|
||||
effort = "medium"
|
||||
if reasoning_config and isinstance(reasoning_config, dict):
|
||||
if reasoning_config.get("enabled") is False:
|
||||
@@ -46,8 +35,6 @@ def resolve_lmstudio_effort(
|
||||
raw = _LM_EFFORT_CLAMP.get(raw, raw)
|
||||
if raw in _LM_VALID_EFFORTS:
|
||||
effort = raw
|
||||
if allowed_options:
|
||||
allowed = {_LM_EFFORT_ALIASES.get(opt, opt) for opt in allowed_options}
|
||||
if effort not in allowed:
|
||||
return None
|
||||
if allowed_options and effort not in {_LM_EFFORT_ALIASES.get(opt, opt) for opt in allowed_options}:
|
||||
return None
|
||||
return effort
|
||||
|
||||
Reference in New Issue
Block a user