Merge origin/main into salvage branch (resolve AUTHOR_MAP conflict)

This commit is contained in:
Teknium
2026-06-10 23:25:54 -07:00
97 changed files with 6864 additions and 1520 deletions
@@ -0,0 +1,96 @@
"""Regression: output-only SDK fields must not leak into Anthropic request input.
Reproduces HTTP 400 `messages.N.content.M.text.parsed_output: Extra inputs are
not permitted`. Anthropic SDK response blocks carry output-only attributes
(text blocks: `parsed_output`, `citations=None`; tool_use blocks: `caller`)
that the Messages *input* schema forbids. normalize_response captured blocks
verbatim via _to_plain_data and replayed them as input → 400.
Fix: whitelist input-permitted fields per block type at three points —
normalize_response capture, _sanitize_replay_block (ordered-blocks replay), and
_convert_content_part_to_anthropic (content-list replay).
"""
import sys, os
sys.path.insert(0, os.path.expanduser("~/.hermes/hermes-agent"))
import pytest
from agent.anthropic_adapter import (
_sanitize_replay_block,
_convert_content_part_to_anthropic,
_convert_assistant_message,
)
FORBIDDEN = {"parsed_output", "caller"}
def _assert_clean(block):
"""No forbidden output-only key, and no null citations, anywhere."""
assert isinstance(block, dict)
for k in FORBIDDEN:
assert k not in block, f"forbidden field {k!r} survived: {block}"
if "citations" in block:
assert isinstance(block["citations"], list) and block["citations"], \
"citations must be a non-empty list if present (None/[] is input-invalid)"
class TestSanitizeReplayBlock:
def test_text_block_strips_parsed_output_and_null_citations(self):
poisoned = {"type": "text", "text": "hi", "parsed_output": None, "citations": None}
out = _sanitize_replay_block(poisoned)
_assert_clean(out)
assert out == {"type": "text", "text": "hi"}
def test_tool_use_strips_caller(self):
poisoned = {"type": "tool_use", "id": "toolu_1", "name": "read_file",
"input": {"path": "a"}, "caller": {"type": "agent"}}
out = _sanitize_replay_block(poisoned)
_assert_clean(out)
assert out["name"] == "read_file" and out["input"] == {"path": "a"}
def test_thinking_preserves_signature(self):
b = {"type": "thinking", "thinking": "x", "signature": "sig-AAA"}
out = _sanitize_replay_block(b)
assert out == {"type": "thinking", "thinking": "x", "signature": "sig-AAA"}
def test_text_keeps_real_citations(self):
real = [{"type": "char_location", "cited_text": "q"}]
out = _sanitize_replay_block({"type": "text", "text": "t", "citations": real})
assert out["citations"] == real
def test_unknown_type_dropped(self):
assert _sanitize_replay_block({"type": "server_tool_use", "foo": 1}) is None
class TestContentPartConversion:
def test_stored_text_block_with_parsed_output_cleaned(self):
# The exact content.N.text.parsed_output failure shape.
part = {"type": "text", "text": "hello", "parsed_output": None, "citations": None}
out = _convert_content_part_to_anthropic(part)
_assert_clean(out)
class TestAssistantReplay:
def test_interleaved_blocks_replayed_clean_and_ordered(self):
m = {
"role": "assistant",
"anthropic_content_blocks": [
{"type": "thinking", "thinking": "plan", "signature": "s1"},
{"type": "text", "text": "doing it", "parsed_output": None, "citations": None},
{"type": "tool_use", "id": "toolu_1", "name": "read_file",
"input": {"path": "a"}, "caller": {"type": "agent"}},
],
}
out = _convert_assistant_message(m)
blocks = out["content"]
# order preserved
assert [b["type"] for b in blocks] == ["thinking", "text", "tool_use"]
# every block clean
for b in blocks:
_assert_clean(b)
# signature + tool fields intact
assert blocks[0]["signature"] == "s1"
assert blocks[2]["name"] == "read_file"
if __name__ == "__main__":
raise SystemExit(pytest.main([__file__, "-v"]))
@@ -0,0 +1,314 @@
"""Regression test for the Anthropic interleaved thinking-block 400.
Reproduces: HTTP 400 ``messages.N.content.M: thinking or redacted_thinking
blocks in the latest assistant message cannot be modified. These blocks must
remain as they were in the original response.``
Root cause under test
----------------------
With adaptive / interleaved thinking (Claude 4.6+, e.g. Opus 4.8), a single
assistant turn can emit content blocks in an interleaved order::
thinking_1 (signed) · tool_use_1 · thinking_2 (signed) · tool_use_2
Anthropic signs each thinking block against the turn content that precedes it
at its position. ``thinking_2`` is signed with ``tool_use_1`` before it.
``AnthropicTransport.normalize_response`` (agent/transports/anthropic.py)
splits the turn into two *parallel* lists — ``reasoning_details`` (thinking
blocks) and ``tool_calls`` (tool_use blocks) — discarding the cross-type
ordering. ``run_agent`` stores those as separate fields on the assistant
message. On replay, ``_convert_assistant_message`` (agent/anthropic_adapter.py)
rebuilds the content as ``[all thinking][text][all tool_use]``, which reorders
``thinking_2`` ahead of ``tool_use_1``. The signature no longer matches its
original position, so Anthropic rejects the latest assistant message with the
400 above.
This test asserts that an interleaved turn round-trips through
normalize_response -> stored message -> convert_messages_to_anthropic with its
block order preserved. It FAILS on the current code (documenting the bug) and
should PASS once block ordering is preserved on replay.
"""
import json
from types import SimpleNamespace
import pytest
from agent.transports import get_transport
from agent.anthropic_adapter import convert_messages_to_anthropic
def _thinking_block(text: str, signature: str) -> SimpleNamespace:
"""A signed Anthropic thinking block, shaped like the SDK object."""
return SimpleNamespace(type="thinking", thinking=text, signature=signature)
def _tool_use_block(block_id: str, name: str, payload: dict) -> SimpleNamespace:
return SimpleNamespace(type="tool_use", id=block_id, name=name, input=payload)
def _interleaved_response() -> SimpleNamespace:
"""An assistant turn with thinking interleaved between two tool_use blocks."""
return SimpleNamespace(
content=[
_thinking_block("Plan: inspect file A first.", "sig-AAA"),
_tool_use_block("toolu_1", "read_file", {"path": "a.py"}),
_thinking_block("A looked fine; now inspect B.", "sig-BBB"),
_tool_use_block("toolu_2", "read_file", {"path": "b.py"}),
],
stop_reason="tool_use",
usage=None,
)
def _stored_assistant_message(normalized) -> dict:
"""Reconstruct the OpenAI-style assistant message the way run_agent stores it.
run_agent.py persists assistant turns as separate fields: content,
reasoning_details (from provider_data), and tool_calls. See
run_agent.py L1513-1516 and hermes_state.py.
"""
provider_data = normalized.provider_data or {}
tool_calls = []
for tc in (normalized.tool_calls or []):
tool_calls.append({
"id": tc.id,
"type": "function",
"function": {"name": tc.name, "arguments": tc.arguments},
})
msg = {
"role": "assistant",
"content": normalized.content or "",
"reasoning_details": provider_data.get("reasoning_details"),
"tool_calls": tool_calls,
}
# build_assistant_message lifts the verbatim ordered-block channel onto
# the stored message; mirror that here.
blocks = provider_data.get("anthropic_content_blocks")
if blocks:
msg["anthropic_content_blocks"] = blocks
return msg
def _original_block_order(response) -> list:
"""The (type, key) sequence of the original interleaved response."""
order = []
for b in response.content:
if b.type == "thinking":
order.append(("thinking", b.signature))
elif b.type == "tool_use":
order.append(("tool_use", b.id))
return order
def _replayed_block_order(assistant_content) -> list:
order = []
for b in assistant_content:
if not isinstance(b, dict):
continue
if b.get("type") in ("thinking", "redacted_thinking"):
order.append(("thinking", b.get("signature")))
elif b.get("type") == "tool_use":
order.append(("tool_use", b.get("id")))
return order
class TestInterleavedThinkingBlockOrder:
def test_normalize_response_loses_interleaving(self):
"""Confirm the lossy split: normalize_response stores thinking and
tool_use in independent fields with no positional linkage."""
transport = get_transport("anthropic_messages")
normalized = transport.normalize_response(_interleaved_response())
# Both thinking blocks are captured...
details = (normalized.provider_data or {}).get("reasoning_details")
assert details is not None and len(details) == 2
# ...and both tool calls...
assert normalized.tool_calls is not None and len(normalized.tool_calls) == 2
# ...but they live in separate fields. There is no single ordered
# structure recording that thinking_2 sat between the two tool calls.
# (This is the structural precondition for the reorder bug.)
def test_interleaved_order_preserved_on_replay(self):
"""The latest assistant message must replay blocks in their ORIGINAL
order, or Anthropic rejects the signed thinking blocks with a 400.
FAILS on current code: _convert_assistant_message front-loads all
thinking blocks, producing
thinking_1 · thinking_2 · tool_use_1 · tool_use_2
instead of the original
thinking_1 · tool_use_1 · thinking_2 · tool_use_2
"""
response = _interleaved_response()
original_order = _original_block_order(response)
transport = get_transport("anthropic_messages")
normalized = transport.normalize_response(response)
assistant_msg = _stored_assistant_message(normalized)
# Build a minimal conversation where this assistant turn is the LATEST
# assistant message (the one whose signed blocks are sent verbatim).
messages = [
{"role": "user", "content": "Inspect a.py and b.py."},
assistant_msg,
{"role": "tool", "tool_call_id": "toolu_1", "content": "a.py: ok"},
{"role": "tool", "tool_call_id": "toolu_2", "content": "b.py: ok"},
]
_system, anthropic_messages = convert_messages_to_anthropic(
messages,
base_url=None, # direct Anthropic
model="claude-opus-4-8", # adaptive thinking family
)
# Find the (latest) assistant message in the converted output.
assistant_out = [m for m in anthropic_messages if m.get("role") == "assistant"]
assert assistant_out, "no assistant message in converted output"
replayed_order = _replayed_block_order(assistant_out[-1]["content"])
assert replayed_order == original_order, (
"Interleaved thinking/tool_use order was not preserved on replay.\n"
f" original: {original_order}\n"
f" replayed: {replayed_order}\n"
"Anthropic signs thinking blocks against their original position; "
"reordering invalidates the signature -> HTTP 400 'thinking blocks "
"in the latest assistant message cannot be modified'."
)
def test_replay_falls_back_gracefully_without_ordered_blocks(self):
"""Without the ordered-block channel, conversion must not crash.
The channel is intentionally NOT persisted to state.db (in-memory
only): a session reloaded from disk after a crash loses the field
and falls back to reconstruction. That replay may take one HTTP 400,
which the thinking-signature recovery (#43667) absorbs by stripping
reasoning_details and retrying. This test pins the fallback shape:
conversion still produces a valid assistant message from the
parallel reasoning_details + tool_calls fields.
"""
response = _interleaved_response()
transport = get_transport("anthropic_messages")
normalized = transport.normalize_response(response)
assistant_msg = _stored_assistant_message(normalized)
# Simulate a disk reload: the in-memory-only channel is gone.
assistant_msg.pop("anthropic_content_blocks", None)
messages = [
assistant_msg,
{"role": "tool", "tool_call_id": "toolu_1", "content": "a ok"},
{"role": "tool", "tool_call_id": "toolu_2", "content": "b ok"},
]
_system, anthropic_messages = convert_messages_to_anthropic(
messages, base_url=None, model="claude-opus-4-8",
)
assistant_out = [m for m in anthropic_messages if m.get("role") == "assistant"]
assert assistant_out, "no assistant message in converted output"
content = assistant_out[-1]["content"]
assert isinstance(content, list) and content, "fallback produced empty content"
# Reconstruction keeps both tool_use blocks (answered by results).
tool_ids = [b.get("id") for b in content if isinstance(b, dict) and b.get("type") == "tool_use"]
assert set(tool_ids) == {"toolu_1", "toolu_2"}
class TestInterleavedReplayCredentialRedaction:
"""The verbatim-replay fast path must not leak un-redacted secrets.
anthropic_content_blocks captures each tool_use ``input`` from the RAW API
response (normalize_response), which is NOT credential-redacted. The
parallel tool_calls[].function.arguments IS redacted at storage time
(build_assistant_message, #19798). If the fast path replays the block's raw
input verbatim, a secret the model inlined into a tool call rides back onto
the wire — even though it is redacted everywhere else in history. The fix
re-sources tool_use input from the redacted tool_calls map by id.
"""
def test_tool_use_input_resourced_from_redacted_tool_calls(self):
REDACTED = "[REDACTED_SECRET]"
# Ordered channel: raw input carries the live secret (as captured from
# the unredacted API response).
ordered = [
{"type": "thinking", "thinking": "Call the API.", "signature": "sig-AAA"},
{
"type": "tool_use",
"id": "toolu_1",
"name": "terminal",
"input": {"command": "curl -H 'Authorization: Bearer sk-LIVE-SECRET-123'"},
},
{"type": "thinking", "thinking": "Now the second call.", "signature": "sig-BBB"},
{
"type": "tool_use",
"id": "toolu_2",
"name": "terminal",
"input": {"command": "echo done"},
},
]
# Stored tool_calls: arguments already redacted (the #19798 path).
assistant_msg = {
"role": "assistant",
"content": "",
"reasoning_details": [b for b in ordered if b["type"] == "thinking"],
"tool_calls": [
{
"id": "toolu_1",
"type": "function",
"function": {
"name": "terminal",
"arguments": json.dumps(
{"command": f"curl -H 'Authorization: Bearer {REDACTED}'"}
),
},
},
{
"id": "toolu_2",
"type": "function",
"function": {
"name": "terminal",
"arguments": json.dumps({"command": "echo done"}),
},
},
],
"anthropic_content_blocks": ordered,
}
messages = [
{"role": "user", "content": "Hit the API twice."},
assistant_msg,
{"role": "tool", "tool_call_id": "toolu_1", "content": "200 OK"},
{"role": "tool", "tool_call_id": "toolu_2", "content": "done"},
]
_system, anthropic_messages = convert_messages_to_anthropic(
messages, base_url=None, model="claude-opus-4-8",
)
assistant_out = [m for m in anthropic_messages if m.get("role") == "assistant"]
assert assistant_out, "no assistant message in converted output"
blocks = assistant_out[-1]["content"]
tool_uses = {b["id"]: b for b in blocks if b.get("type") == "tool_use"}
assert set(tool_uses) == {"toolu_1", "toolu_2"}, "tool_use blocks missing/renamed"
# The replayed input must be the REDACTED value, not the live secret.
replayed_cmd = tool_uses["toolu_1"]["input"]["command"]
assert "sk-LIVE-SECRET-123" not in replayed_cmd, (
"Un-redacted secret leaked onto the wire via the verbatim-replay "
"fast path. tool_use input must be re-sourced from the redacted "
"tool_calls map, not the raw captured block."
)
assert REDACTED in replayed_cmd
# Interleave order is still preserved (the reason the channel exists).
order = [
("thinking", b.get("signature")) if b.get("type") == "thinking"
else ("tool_use", b.get("id"))
for b in blocks if b.get("type") in ("thinking", "tool_use")
]
assert order == [
("thinking", "sig-AAA"),
("tool_use", "toolu_1"),
("thinking", "sig-BBB"),
("tool_use", "toolu_2"),
]
if __name__ == "__main__":
raise SystemExit(pytest.main([__file__, "-v"]))
+405
View File
@@ -0,0 +1,405 @@
"""Tests for agent.coding_context — RuntimeMode seam, resolver, toolset, git probe."""
import json
import subprocess
from pathlib import Path
import pytest
from agent import coding_context as cc
def _git_init(path):
env = {
"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t",
}
for args in (
["init", "-q", "-b", "main"],
["commit", "-q", "--allow-empty", "-m", "init commit"],
):
subprocess.run(["git", "-C", str(path), *args], check=True, env={**env, "HOME": str(path)})
# ── resolver ──────────────────────────────────────────────────────────────
class TestIsCodingContext:
def test_off_never_activates(self, tmp_path):
_git_init(tmp_path)
cfg = {"agent": {"coding_context": "off"}}
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is False
def test_on_forces_even_without_git(self, tmp_path):
cfg = {"agent": {"coding_context": "on"}}
assert cc.is_coding_context(platform="telegram", cwd=tmp_path, config=cfg) is True
def test_auto_requires_git_repo(self, tmp_path):
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is False
_git_init(tmp_path)
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is True
def test_auto_skips_messaging_surfaces(self, tmp_path):
_git_init(tmp_path)
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="discord", cwd=tmp_path, config=cfg) is False
assert cc.is_coding_context(platform="tui", cwd=tmp_path, config=cfg) is True
def test_default_mode_is_auto(self, tmp_path):
# Unknown/missing value normalizes to auto.
_git_init(tmp_path)
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config={}) is True
# ── toolset substitution ────────────────────────────────────────────────────
class TestCodingSelection:
def test_selects_coding_under_focus(self, tmp_path):
_git_init(tmp_path)
cfg = {"agent": {"coding_context": "focus"}}
out = cc.coding_selection(platform="cli", cwd=tmp_path, config=cfg)
assert out is not None
assert out[0] == cc.CODING_TOOLSET
def test_auto_is_prompt_only(self, tmp_path):
# Default posture must never override the user's configured toolsets —
# off-by-default toolsets are already off, and explicit opt-ins
# (image-gen, spotify, …) survive entering a code workspace.
_git_init(tmp_path)
cfg = {"agent": {"coding_context": "auto"}}
assert cc.coding_selection(platform="cli", cwd=tmp_path, config=cfg) is None
# …while the prompt posture is still active.
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is True
def test_on_is_prompt_only(self, tmp_path):
cfg = {"agent": {"coding_context": "on"}}
assert cc.coding_selection(platform="cli", cwd=tmp_path, config=cfg) is None
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is True
def test_focus_requires_workspace(self, tmp_path):
# focus inherits auto's detection gate — bare dir stays general.
cfg = {"agent": {"coding_context": "focus"}}
assert cc.coding_selection(platform="cli", cwd=tmp_path, config=cfg) is None
def test_none_when_inactive(self, tmp_path):
cfg = {"agent": {"coding_context": "off"}}
assert cc.coding_selection(platform="cli", cwd=tmp_path, config=cfg) is None
def test_coding_toolset_is_registered(self):
from toolsets import resolve_toolset
tools = resolve_toolset(cc.CODING_TOOLSET)
# Coding essentials present…
for t in ("read_file", "write_file", "patch", "search_files", "terminal", "todo"):
assert t in tools
# …and the noise is gone.
for t in ("send_message", "text_to_speech", "image_generate", "computer_use"):
assert t not in tools
# ── git/workspace probe ─────────────────────────────────────────────────────
class TestWorkspaceBlock:
def test_empty_outside_repo(self, tmp_path):
assert cc.build_coding_workspace_block(tmp_path) == ""
def test_reports_branch_and_clean_status(self, tmp_path):
_git_init(tmp_path)
block = cc.build_coding_workspace_block(tmp_path)
assert "Workspace" in block
assert f"Root: {tmp_path.resolve()}" in block or "Root:" in block
assert "Branch: main" in block
assert "Status: clean" in block
assert "init commit" in block
def test_reports_dirty_counts(self, tmp_path):
_git_init(tmp_path)
(tmp_path / "untracked.txt").write_text("hi")
block = cc.build_coding_workspace_block(tmp_path)
assert "untracked" in block
assert "clean" not in block.split("Status:")[1].splitlines()[0]
# ── project facts (verify-loop detection) ───────────────────────────────────
class TestProjectFacts:
def test_package_json_scripts_surface_verify_commands(self, tmp_path):
_git_init(tmp_path)
(tmp_path / "package.json").write_text(
json.dumps({"scripts": {"test": "vitest", "lint": "eslint .", "dev": "vite"}})
)
(tmp_path / "pnpm-lock.yaml").write_text("")
block = cc.build_coding_workspace_block(tmp_path)
assert "Project: package.json (pnpm)" in block
assert "pnpm run test" in block and "pnpm run lint" in block
# Non-verify scripts (dev servers, …) stay out of the snapshot.
assert "run dev" not in block
def test_pytest_config_and_run_tests_script(self, tmp_path):
_git_init(tmp_path)
(tmp_path / "pyproject.toml").write_text("[tool.pytest.ini_options]\n")
scripts = tmp_path / "scripts"
scripts.mkdir()
(scripts / "run_tests.sh").write_text("#!/bin/sh\n")
block = cc.build_coding_workspace_block(tmp_path)
assert "scripts/run_tests.sh" in block
assert "pytest" in block.split("Verify:")[1]
def test_makefile_verify_targets_only(self, tmp_path):
_git_init(tmp_path)
(tmp_path / "Makefile").write_text("test:\n\tgo test ./...\n\ndeploy:\n\t./deploy.sh\n")
block = cc.build_coding_workspace_block(tmp_path)
assert "make test" in block
assert "make deploy" not in block
def test_context_files_listed(self, tmp_path):
_git_init(tmp_path)
(tmp_path / "AGENTS.md").write_text("# rules")
block = cc.build_coding_workspace_block(tmp_path)
assert "Context files: AGENTS.md" in block
def test_marker_only_project_gets_snapshot_without_git(self, tmp_path):
# A non-git project (manifest only) still gets a workspace snapshot —
# just without the git lines.
(tmp_path / "package.json").write_text("{}")
block = cc.build_coding_workspace_block(tmp_path)
assert f"Root: {tmp_path.resolve()}" in block
assert "package.json" in block
assert "Branch:" not in block and "Status:" not in block
def test_malformed_package_json_is_ignored(self, tmp_path):
_git_init(tmp_path)
(tmp_path / "package.json").write_text("{not json")
block = cc.build_coding_workspace_block(tmp_path)
assert "Project: package.json" in block
assert "Verify:" not in block
# ── $HOME dotfiles guard ────────────────────────────────────────────────────
class TestHomeDotfilesGuard:
def test_dotfiles_repo_at_home_is_not_coding(self, tmp_path, monkeypatch):
home = tmp_path / "home"
home.mkdir()
_git_init(home)
monkeypatch.setattr(Path, "home", lambda: home)
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=home, config=cfg) is False
# …and a plain subdirectory of the dotfiles repo stays general too.
docs = home / "Documents"
docs.mkdir()
assert cc.is_coding_context(platform="cli", cwd=docs, config=cfg) is False
def test_marker_at_home_is_not_a_project_signal(self, tmp_path, monkeypatch):
home = tmp_path / "home"
home.mkdir()
(home / "Makefile").write_text("all:\n")
monkeypatch.setattr(Path, "home", lambda: home)
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=home, config=cfg) is False
def test_real_project_under_dotfiles_home_still_detects(self, tmp_path, monkeypatch):
home = tmp_path / "home"
home.mkdir()
_git_init(home)
monkeypatch.setattr(Path, "home", lambda: home)
proj = home / "www" / "app"
proj.mkdir(parents=True)
(proj / "package.json").write_text("{}")
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=proj, config=cfg) is True
def test_on_mode_bypasses_the_guard(self, tmp_path, monkeypatch):
home = tmp_path / "home"
home.mkdir()
monkeypatch.setattr(Path, "home", lambda: home)
cfg = {"agent": {"coding_context": "on"}}
assert cc.is_coding_context(platform="cli", cwd=home, config=cfg) is True
# ── prompt assembly integration ─────────────────────────────────────────────
class TestStatusParsing:
def test_parse_status_counts_and_branch(self):
porcelain = (
"# branch.head feature\n"
"# branch.upstream origin/feature\n"
"# branch.ab +2 -1\n"
"1 M. N... 100644 100644 100644 aaa bbb staged.py\n"
"1 .M N... 100644 100644 100644 ccc ddd modified.py\n"
"? new.py\n"
"u UU N... 1 2 3 abc def conflict.py\n"
)
branch, counts = cc._parse_status(porcelain)
assert branch["head"] == "feature"
assert branch["upstream"] == "origin/feature"
assert branch["ahead"] == "2" and branch["behind"] == "1"
assert counts["staged"] == 1
assert counts["modified"] == 1
assert counts["untracked"] == 1
assert counts["conflicts"] == 1
# ── RuntimeMode seam ────────────────────────────────────────────────────────
class TestRuntimeMode:
def test_resolves_coding_in_repo(self, tmp_path):
_git_init(tmp_path)
mode = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={})
assert mode.is_coding is True
assert mode.kind == "coding"
assert mode.profile is cc.CODING_PROFILE
def test_resolves_general_outside_workspace(self, tmp_path):
mode = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={})
assert mode.is_coding is False
assert mode.kind == "general"
# General posture pins no toolset and injects no blocks.
assert mode.toolset_selection() is None
assert mode.system_blocks() == []
def test_is_frozen(self, tmp_path):
mode = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={})
with pytest.raises(Exception):
mode.profile = cc.CODING_PROFILE # type: ignore[misc]
def test_system_blocks_include_brief_and_workspace(self, tmp_path):
_git_init(tmp_path)
mode = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={"agent": {"coding_context": "on"}})
blocks = mode.system_blocks()
assert any("coding agent" in b for b in blocks)
assert any("Workspace" in b for b in blocks)
def test_toolset_selection_gated_on_focus(self, tmp_path):
_git_init(tmp_path)
focus = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={"agent": {"coding_context": "focus"}})
sel = focus.toolset_selection()
assert sel and sel[0] == cc.CODING_TOOLSET
# auto/on resolve the coding profile but stay prompt-only.
for raw in ("auto", "on"):
mode = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={"agent": {"coding_context": raw}})
assert mode.is_coding is True
assert mode.toolset_selection() is None
# ── edit-format steering (per-model harness tuning) ──────────────────────────
class TestEditFormatSteering:
def test_family_detection(self):
assert cc._model_family("openai/gpt-5.4") == "patch"
assert cc._model_family("openai/codex-mini") == "patch"
assert cc._model_family("anthropic/claude-opus-4.8") == "replace"
assert cc._model_family("anthropic/claude-sonnet-4") == "replace"
# Gemini + open-weight coding models (RL'd on str_replace-style
# editors) steer to replace, not neutral.
for m in (
"google/gemini-3-pro", "deepseek-v3.2", "qwen3-coder",
"moonshot/kimi-k2", "zai/glm-4.6", "nousresearch/hermes-4-405b",
):
assert cc._model_family(m) == "replace"
# Unknown family and no model both fall through to neutral wording.
assert cc._model_family("acme/foo-1") is None
assert cc._model_family(None) is None
assert cc._model_family("") is None
def test_openai_family_gets_v4a_nudge(self, tmp_path):
_git_init(tmp_path)
mode = cc.resolve_runtime_mode(
platform="cli", cwd=tmp_path,
config={"agent": {"coding_context": "on"}}, model="openai/gpt-5.4",
)
brief = mode.system_blocks()[0]
assert "mode='patch'" in brief
assert "V4A" in brief
assert "write_file" in brief # new files authored, not patched
def test_anthropic_family_gets_replace_nudge(self, tmp_path):
_git_init(tmp_path)
mode = cc.resolve_runtime_mode(
platform="cli", cwd=tmp_path,
config={"agent": {"coding_context": "on"}},
model="anthropic/claude-opus-4.8",
)
brief = mode.system_blocks()[0]
assert "mode='replace'" in brief
assert "write_file" in brief # new files authored, not patched
def test_unknown_model_keeps_neutral_brief(self, tmp_path):
# No edit-format line appended — brief equals the bare profile guidance.
_git_init(tmp_path)
mode = cc.resolve_runtime_mode(
platform="cli", cwd=tmp_path,
config={"agent": {"coding_context": "on"}}, model="acme/foo-1",
)
assert mode.system_blocks()[0] == cc.CODING_AGENT_GUIDANCE
def test_no_model_keeps_neutral_brief(self, tmp_path):
_git_init(tmp_path)
mode = cc.resolve_runtime_mode(
platform="cli", cwd=tmp_path,
config={"agent": {"coding_context": "on"}},
)
assert mode.system_blocks()[0] == cc.CODING_AGENT_GUIDANCE
def test_general_posture_emits_nothing_regardless_of_model(self, tmp_path):
# Edit steering only fires inside the coding posture.
mode = cc.resolve_runtime_mode(
platform="telegram", cwd=tmp_path, config={}, model="openai/gpt-5.4",
)
assert mode.system_blocks() == []
# ── profile registry ────────────────────────────────────────────────────────
class TestProfiles:
def test_registered_profiles(self):
assert cc.get_profile("coding") is cc.CODING_PROFILE
assert cc.get_profile("general") is cc.GENERAL_PROFILE
def test_unknown_profile_falls_back_to_general(self):
assert cc.get_profile("nonsense") is cc.GENERAL_PROFILE
def test_coding_profile_shape(self):
# The coding profile declares the seams other domains read.
assert cc.CODING_PROFILE.toolset == cc.CODING_TOOLSET
assert cc.CODING_PROFILE.guidance
assert cc.CODING_PROFILE.model_hint == "coding"
# General is inert.
assert cc.GENERAL_PROFILE.toolset is None
assert cc.GENERAL_PROFILE.guidance == ""
def test_skill_pruning_scoped_to_coding_posture(self, tmp_path):
# Coding posture hides clearly-non-coding categories; coding-adjacent
# ones stay visible (deny-list semantics).
_git_init(tmp_path)
coding = cc.resolve_runtime_mode(platform="cli", cwd=tmp_path, config={})
hidden = coding.hidden_skill_categories()
assert "social-media" in hidden and "smart-home" in hidden
for kept in ("github", "devops", "software-development", "data-science"):
assert kept not in hidden
# General posture hides nothing.
general = cc.resolve_runtime_mode(
platform="telegram", cwd=tmp_path, config={}
)
assert general.hidden_skill_categories() == frozenset()
# ── detection signals ───────────────────────────────────────────────────────
class TestDetection:
@pytest.mark.parametrize("marker", ["pyproject.toml", "package.json", "go.mod", "AGENTS.md"])
def test_project_manifest_triggers_without_git(self, tmp_path, marker):
(tmp_path / marker).write_text("x")
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is True
def test_marker_in_parent_counts_from_subdir(self, tmp_path):
(tmp_path / "pyproject.toml").write_text("x")
sub = tmp_path / "src" / "pkg"
sub.mkdir(parents=True)
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=sub, config=cfg) is True
def test_bare_dir_is_not_coding(self, tmp_path):
cfg = {"agent": {"coding_context": "auto"}}
assert cc.is_coding_context(platform="cli", cwd=tmp_path, config=cfg) is False
+41
View File
@@ -12,6 +12,7 @@ from agent.display import (
set_tool_preview_max_len,
_render_inline_unified_diff,
_summarize_rendered_diff_sections,
_used_free_parallel,
render_edit_diff_with_delta,
)
@@ -171,6 +172,46 @@ class TestCuteToolMessagePreviewLength:
assert "[error]" not in line
class TestWebProviderLabel:
"""The free-path "Parallel search"/"Parallel fetch" verb labeling."""
def test_free_search_verb_is_parallel(self):
result = json.dumps({"success": True, "data": {"web": []}, "provider": "parallel"})
line = get_cute_tool_message("web_search", {"query": "hello"}, 0.1, result=result)
assert "Parallel search" in line
assert "hello" in line
def test_paid_search_verb_is_plain(self):
result = json.dumps({"success": True, "data": {"web": [{"url": "u"}]}})
line = get_cute_tool_message("web_search", {"query": "hi"}, 0.1, result=result)
assert "Parallel" not in line
assert "search" in line
def test_missing_result_verb_is_plain(self):
line = get_cute_tool_message("web_search", {"query": "hello"}, 0.1)
assert "Parallel" not in line
assert "search" in line
def test_helper_is_parallel_free_specific(self):
# Only Parallel's free MCP path marks results; nothing else does.
assert _used_free_parallel(json.dumps({"provider": "parallel"})) is True
assert _used_free_parallel(json.dumps({"provider": "exa"})) is False
assert _used_free_parallel(json.dumps({"provider": "firecrawl"})) is False
assert _used_free_parallel(json.dumps({"success": True, "data": {}})) is False
assert _used_free_parallel('not json') is False
assert _used_free_parallel(None) is False
def test_free_extract_verb_is_parallel(self):
result = json.dumps({"results": [{"url": "u", "content": "x"}], "provider": "parallel"})
line = get_cute_tool_message("web_extract", {"urls": ["https://a.test"]}, 0.1, result=result)
assert "Parallel fetch" in line
def test_paid_extract_verb_is_plain(self):
result = json.dumps({"results": [{"url": "u", "content": "x"}]})
line = get_cute_tool_message("web_extract", {"urls": ["https://a.test"]}, 0.1, result=result)
assert "Parallel" not in line
class TestEditDiffPreview:
def test_extract_edit_diff_for_patch(self):
diff = extract_edit_diff("patch", '{"success": true, "diff": "--- a/x\\n+++ b/x\\n"}')
+36
View File
@@ -276,6 +276,42 @@ class TestBuildSkillsSystemPrompt:
# "search" should appear only once per category
assert result.count("- search") == 1
def test_hidden_categories_pruned_with_note(self, monkeypatch, tmp_path):
"""Posture-driven pruning drops whole categories and discloses it."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
for cat, name in (("social-media", "tweet-stuff"), ("github", "pr-review")):
d = tmp_path / "skills" / cat / name
d.mkdir(parents=True)
(d / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: Does {name} things\n---\n"
)
result = build_skills_system_prompt(
hidden_categories=frozenset({"social-media"})
)
assert "pr-review" in result
assert "tweet-stuff" not in result
# Disclosure note so the model knows the full catalog exists.
assert "skills_list" in result
def test_hidden_categories_prune_nested_and_miss_cache_separately(
self, monkeypatch, tmp_path
):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
d = tmp_path / "skills" / "social-media" / "twitter" / "thread-writer"
d.mkdir(parents=True)
(d / "SKILL.md").write_text(
"---\nname: thread-writer\ndescription: Write threads\n---\n"
)
# Nested category ("social-media/twitter") pruned via its parent.
pruned = build_skills_system_prompt(
hidden_categories=frozenset({"social-media"})
)
assert "thread-writer" not in pruned
# Unfiltered call must not be served from the filtered cache entry.
full = build_skills_system_prompt()
assert "thread-writer" in full
def test_excludes_incompatible_platform_skills(self, monkeypatch, tmp_path):
"""Skills with platforms: [macos] should not appear on Linux."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
@@ -0,0 +1,111 @@
"""Stream read timeout must never preempt the stale-stream detector.
Reasoning models (e.g. Opus) routinely pause mid-stream for minutes during
extended thinking. The stale-stream detector is deliberately scaled up to
tolerate this (180s base, raised to 240s/300s for large contexts). The httpx
socket read timeout, however, defaulted to a flat 120s for cloud providers and
fired *first* — tearing down a healthy reasoning stream before the stale
detector (which owns retry + diagnostics) could act.
These tests pin the invariant: for a cloud provider on the default read
timeout, the httpx socket read timeout is floored at the stale-stream timeout
so it can never fire before the detector. They mirror the inline logic in
``agent/chat_completion_helpers.py`` (the real builder lives deep inside a
worker thread, so — like ``test_local_stream_timeout.py`` — the resolution is
reproduced here rather than driven end-to-end).
"""
import os
import pytest
from agent.model_metadata import is_local_endpoint
def _resolve_stale_timeout(base_url, est_tokens, stale_base=180.0):
"""Mirror of the stale-stream detector resolution."""
if stale_base == 180.0 and base_url and is_local_endpoint(base_url):
return float("inf") # detector disabled for local providers
if est_tokens > 100_000:
return max(stale_base, 300.0)
if est_tokens > 50_000:
return max(stale_base, 240.0)
return stale_base
def _resolve_read_timeout(base_url, stale_timeout, base_timeout=1800.0):
"""Mirror of the httpx socket read-timeout builder (cloud branch)."""
read_timeout = float(os.getenv("HERMES_STREAM_READ_TIMEOUT", 120.0))
if read_timeout == 120.0 and base_url and is_local_endpoint(base_url):
read_timeout = base_timeout
elif (
read_timeout == 120.0
and stale_timeout is not None
and stale_timeout != float("inf")
and stale_timeout > read_timeout
):
read_timeout = stale_timeout
return read_timeout
CLOUD_URLS = [
"https://api.githubcopilot.com",
"https://api.openai.com",
"https://openrouter.ai/api",
"https://api.anthropic.com",
]
class TestCloudReadTimeoutFloor:
@pytest.fixture(autouse=True)
def _clear_env(self):
with pytest.MonkeyPatch.context() as mp:
mp.delenv("HERMES_STREAM_READ_TIMEOUT", raising=False)
yield
@pytest.mark.parametrize("base_url", CLOUD_URLS)
@pytest.mark.parametrize("est_tokens", [0, 10_000, 60_000, 150_000])
def test_read_timeout_never_below_stale(self, base_url, est_tokens):
"""Core invariant: the socket read timeout >= the stale detector."""
stale = _resolve_stale_timeout(base_url, est_tokens)
read = _resolve_read_timeout(base_url, stale)
assert read >= stale
@pytest.mark.parametrize("base_url", CLOUD_URLS)
def test_small_context_floored_to_stale_base(self, base_url):
"""Reported case: ~120s timeouts on Copilot are raised to the 180s base."""
stale = _resolve_stale_timeout(base_url, est_tokens=37_000)
read = _resolve_read_timeout(base_url, stale)
assert read == 180.0
@pytest.mark.parametrize("base_url", CLOUD_URLS)
def test_large_context_tracks_scaled_stale(self, base_url):
"""Big contexts scale the stale detector; the read timeout follows."""
assert _resolve_read_timeout(base_url, _resolve_stale_timeout(base_url, 60_000)) == 240.0
assert _resolve_read_timeout(base_url, _resolve_stale_timeout(base_url, 150_000)) == 300.0
def test_user_override_is_respected(self):
"""An explicit HERMES_STREAM_READ_TIMEOUT is never overridden by the floor."""
with pytest.MonkeyPatch.context() as mp:
mp.setenv("HERMES_STREAM_READ_TIMEOUT", "90")
stale = _resolve_stale_timeout("https://api.githubcopilot.com", est_tokens=0)
assert _resolve_read_timeout("https://api.githubcopilot.com", stale) == 90.0
class TestLocalUnaffected:
@pytest.fixture(autouse=True)
def _clear_env(self):
with pytest.MonkeyPatch.context() as mp:
mp.delenv("HERMES_STREAM_READ_TIMEOUT", raising=False)
yield
def test_local_still_raised_to_base(self):
"""Local providers keep their existing behavior (raise to base timeout)."""
stale = _resolve_stale_timeout("http://localhost:11434", est_tokens=0)
assert stale == float("inf") # detector disabled for local
read = _resolve_read_timeout("http://localhost:11434", stale)
assert read == 1800.0 # not clamped by inf
def test_stale_none_falls_back_to_default(self):
"""If the stale value is unresolved, the read timeout keeps its default."""
assert _resolve_read_timeout("https://api.githubcopilot.com", None) == 120.0
+41
View File
@@ -55,3 +55,44 @@ class TestContextFileCwd:
def test_configured_dir_when_terminal_cwd_set(self, monkeypatch, tmp_path):
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
assert _captured_context_cwd(_make_agent()) == tmp_path
def _stable_prompt(agent):
with (
patch("run_agent.load_soul_md", return_value=""),
patch("run_agent.build_nous_subscription_prompt", return_value=""),
patch("run_agent.build_environment_hints", return_value=""),
patch("run_agent.build_context_files_prompt", return_value=""),
):
return build_system_prompt_parts(agent)["stable"]
class TestCodingContextBlock:
def test_injected_when_active(self, monkeypatch, tmp_path):
import subprocess
subprocess.run(["git", "-C", str(tmp_path), "init", "-q"], check=True)
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
agent = _make_agent(valid_tool_names=["read_file"], platform="cli")
stable = _stable_prompt(agent)
assert "coding agent" in stable
assert "Workspace" in stable
def test_absent_when_off(self, monkeypatch, tmp_path):
import subprocess
subprocess.run(["git", "-C", str(tmp_path), "init", "-q"], check=True)
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
agent = _make_agent(valid_tool_names=["read_file"], platform="cli")
# Drive the real path: force the resolved mode to "off" via config.
with patch("agent.coding_context._coding_mode", return_value="off"):
stable = _stable_prompt(agent)
assert "coding agent" not in stable
def test_absent_without_tools(self, monkeypatch, tmp_path):
import subprocess
subprocess.run(["git", "-C", str(tmp_path), "init", "-q"], check=True)
monkeypatch.setenv("TERMINAL_CWD", str(tmp_path))
agent = _make_agent(valid_tool_names=[], platform="cli")
assert "coding agent" not in _stable_prompt(agent)
-449
View File
@@ -1,449 +0,0 @@
"""Tests for per-job profile support in cron jobs.
Covers data-layer validation/storage, cronjob tool plumbing, scheduler runtime
HERMES_HOME scoping, and tick() serialization for profile jobs.
"""
from __future__ import annotations
import json
import os
import pytest
@pytest.fixture()
def isolated_cron_profile_home(tmp_path, monkeypatch):
"""Create an isolated Hermes root with a named profile and temp cron store."""
root = tmp_path / "hermes-root"
profile_home = root / "profiles" / "support"
profile_home.mkdir(parents=True)
(root / "cron").mkdir(parents=True)
monkeypatch.setenv("HERMES_HOME", str(root))
monkeypatch.setattr("cron.jobs.CRON_DIR", root / "cron")
monkeypatch.setattr("cron.jobs.JOBS_FILE", root / "cron" / "jobs.json")
monkeypatch.setattr("cron.jobs.OUTPUT_DIR", root / "cron" / "output")
return root, profile_home
class TestNormalizeProfile:
def test_none_and_empty_return_none(self, isolated_cron_profile_home):
from cron.jobs import _normalize_profile
assert _normalize_profile(None) is None
assert _normalize_profile("") is None
assert _normalize_profile(" ") is None
def test_default_profile_is_valid_and_normalized(self, isolated_cron_profile_home):
from cron.jobs import _normalize_profile
assert _normalize_profile("Default") == "default"
def test_named_profile_must_exist_and_is_normalized(self, isolated_cron_profile_home):
from cron.jobs import _normalize_profile
assert _normalize_profile("Support") == "support"
def test_invalid_profile_name_is_rejected(self, isolated_cron_profile_home):
from cron.jobs import _normalize_profile
with pytest.raises(ValueError):
_normalize_profile("invalid!")
def test_missing_named_profile_is_rejected(self, isolated_cron_profile_home):
from cron.jobs import _normalize_profile
with pytest.raises(FileNotFoundError):
_normalize_profile("missing")
class TestCreateAndUpdateJobProfile:
def test_create_stores_profile_id(self, isolated_cron_profile_home):
from cron.jobs import create_job, get_job
job = create_job(prompt="hello", schedule="every 1h", profile="Support")
stored = get_job(job["id"])
assert stored is not None
assert stored["profile"] == "support"
def test_create_without_profile_preserves_old_behaviour(self, isolated_cron_profile_home):
from cron.jobs import create_job, get_job
job = create_job(prompt="hello", schedule="every 1h")
stored = get_job(job["id"])
assert stored is not None
assert stored.get("profile") is None
def test_create_accepts_explicit_default(self, isolated_cron_profile_home):
from cron.jobs import create_job, get_job
job = create_job(prompt="hello", schedule="every 1h", profile="default")
stored = get_job(job["id"])
assert stored is not None
assert stored["profile"] == "default"
def test_update_sets_and_clears_profile(self, isolated_cron_profile_home):
from cron.jobs import create_job, get_job, update_job
job = create_job(prompt="x", schedule="every 1h")
update_job(job["id"], {"profile": "Support"})
stored = get_job(job["id"])
assert stored is not None
assert stored["profile"] == "support"
update_job(job["id"], {"profile": ""})
stored = get_job(job["id"])
assert stored is not None
assert stored["profile"] is None
def test_update_rejects_missing_profile(self, isolated_cron_profile_home):
from cron.jobs import create_job, update_job
job = create_job(prompt="x", schedule="every 1h")
with pytest.raises(FileNotFoundError):
update_job(job["id"], {"profile": "missing"})
class TestCronjobToolProfile:
def test_create_and_list_with_profile(self, isolated_cron_profile_home):
from tools.cronjob_tools import cronjob
created = json.loads(
cronjob(
action="create",
prompt="hi",
schedule="every 1h",
profile="Support",
)
)
assert created["success"] is True
assert created["job"]["profile"] == "support"
listing = json.loads(cronjob(action="list"))
assert listing["jobs"][0]["profile"] == "support"
def test_update_clears_profile_with_empty_string(self, isolated_cron_profile_home):
from tools.cronjob_tools import cronjob
created = json.loads(
cronjob(
action="create",
prompt="hi",
schedule="every 1h",
profile="Support",
)
)
updated = json.loads(
cronjob(action="update", job_id=created["job_id"], profile="")
)
assert updated["success"] is True
assert "profile" not in updated["job"]
def test_schema_advertises_profile(self):
from tools.cronjob_tools import CRONJOB_SCHEMA
assert "profile" in CRONJOB_SCHEMA["parameters"]["properties"]
desc = CRONJOB_SCHEMA["parameters"]["properties"]["profile"]["description"]
desc_lower = desc.lower()
assert "hermes profile" in desc_lower
assert "context-local" in desc_lower
assert "subprocess" in desc_lower
assert "temporarily sets hermes_home" not in desc_lower
class TestRunJobProfileContext:
@staticmethod
def _install_agent_stubs(monkeypatch, observed: dict):
import sys
import cron.scheduler as sched
class FakeAgent:
def __init__(self, **kwargs):
from hermes_constants import get_hermes_home
observed["env_home_during_init"] = os.environ.get("HERMES_HOME")
observed["profile_env_only_during_init"] = os.environ.get(
"HERMES_PROFILE_TEST_ONLY"
)
observed["profile_env_shared_during_init"] = os.environ.get(
"HERMES_PROFILE_TEST_SHARED"
)
observed["hermes_home_during_init"] = str(get_hermes_home())
observed["scheduler_home_during_init"] = str(sched._get_hermes_home())
observed["skip_context_files"] = kwargs.get("skip_context_files")
def run_conversation(self, *_a, **_kw):
from hermes_constants import get_hermes_home
observed["env_home_during_run"] = os.environ.get("HERMES_HOME")
observed["profile_env_only_during_run"] = os.environ.get(
"HERMES_PROFILE_TEST_ONLY"
)
observed["profile_env_shared_during_run"] = os.environ.get(
"HERMES_PROFILE_TEST_SHARED"
)
observed["hermes_home_during_run"] = str(get_hermes_home())
observed["scheduler_home_during_run"] = str(sched._get_hermes_home())
return {"final_response": "done", "messages": []}
def get_activity_summary(self):
return {"seconds_since_activity": 0.0}
def close(self):
observed["closed"] = True
fake_mod = type(sys)("run_agent")
fake_mod.AIAgent = FakeAgent
monkeypatch.setitem(sys.modules, "run_agent", fake_mod)
from hermes_cli import runtime_provider as runtime_provider
monkeypatch.setattr(
runtime_provider,
"resolve_runtime_provider",
lambda **_kw: {
"provider": "test",
"api_key": "test-key",
"base_url": "http://test.local",
"api_mode": "chat_completions",
},
)
monkeypatch.setattr(sched, "_build_job_prompt", lambda job, prerun_script=None: "hi")
monkeypatch.setattr(sched, "_resolve_origin", lambda job: None)
monkeypatch.setattr(sched, "_resolve_delivery_target", lambda job: None)
monkeypatch.setattr(sched, "_resolve_cron_enabled_toolsets", lambda job, cfg: None)
monkeypatch.setattr(sched, "_hermes_home", None)
monkeypatch.setenv("HERMES_CRON_TIMEOUT", "0")
import dotenv
def fake_load_dotenv(path, *_a, **_kw):
observed.setdefault("dotenv_paths", []).append(str(path))
return True
monkeypatch.setattr(dotenv, "load_dotenv", fake_load_dotenv)
def test_run_job_sets_and_restores_profile_home(
self, isolated_cron_profile_home, monkeypatch
):
import cron.scheduler as sched
root, profile_home = isolated_cron_profile_home
observed: dict = {}
self._install_agent_stubs(monkeypatch, observed)
job = {
"id": "abc",
"name": "profile-job",
"profile": "support",
"schedule_display": "manual",
}
success, _output, response, error = sched.run_job(job)
assert success is True, f"run_job failed: error={error!r} response={response!r}"
assert observed["dotenv_paths"] == [str(profile_home / ".env")]
assert observed["env_home_during_init"] == str(root)
assert observed["env_home_during_run"] == str(root)
assert observed["hermes_home_during_init"] == str(profile_home.resolve())
assert observed["hermes_home_during_run"] == str(profile_home.resolve())
assert observed["scheduler_home_during_init"] == str(profile_home.resolve())
assert observed["scheduler_home_during_run"] == str(profile_home.resolve())
assert observed["skip_context_files"] is True
assert os.environ["HERMES_HOME"] == str(root)
assert sched._get_hermes_home() == root
def test_profile_dotenv_environment_is_restored(
self, isolated_cron_profile_home, monkeypatch
):
import dotenv
import cron.scheduler as sched
root, profile_home = isolated_cron_profile_home
observed: dict = {}
self._install_agent_stubs(monkeypatch, observed)
monkeypatch.setenv("HERMES_PROFILE_TEST_SHARED", "outer")
monkeypatch.delenv("HERMES_PROFILE_TEST_ONLY", raising=False)
def fake_load_dotenv(path, *_a, **_kw):
observed.setdefault("dotenv_paths", []).append(str(path))
os.environ["HERMES_PROFILE_TEST_SHARED"] = "profile-value"
os.environ["HERMES_PROFILE_TEST_ONLY"] = "profile-only"
os.environ["HERMES_CRON_TIMEOUT"] = "123"
return True
monkeypatch.setattr(dotenv, "load_dotenv", fake_load_dotenv)
job = {
"id": "env-profile",
"name": "profile-env-job",
"profile": "support",
"schedule_display": "manual",
}
success, _output, _response, error = sched.run_job(job)
assert success is True, error
assert observed["dotenv_paths"] == [str(profile_home / ".env")]
assert observed["profile_env_only_during_init"] == "profile-only"
assert observed["profile_env_shared_during_init"] == "profile-value"
assert observed["profile_env_only_during_run"] == "profile-only"
assert observed["profile_env_shared_during_run"] == "profile-value"
assert os.environ["HERMES_PROFILE_TEST_SHARED"] == "outer"
assert "HERMES_PROFILE_TEST_ONLY" not in os.environ
assert os.environ["HERMES_CRON_TIMEOUT"] == "0"
assert os.environ["HERMES_HOME"] == str(root)
assert sched._get_hermes_home() == root
def test_no_agent_profile_uses_profile_scripts_dir_and_restores_env(
self, isolated_cron_profile_home, monkeypatch
):
import cron.scheduler as sched
root, profile_home = isolated_cron_profile_home
scripts_dir = profile_home / "scripts"
scripts_dir.mkdir(parents=True)
(scripts_dir / "print_home.py").write_text(
"import os\nprint(os.environ.get('HERMES_HOME', ''))\n",
encoding="utf-8",
)
monkeypatch.setattr(sched, "_hermes_home", None)
job = {
"id": "script1",
"name": "profile-script",
"profile": "support",
"script": "print_home.py",
"no_agent": True,
}
success, _doc, response, error = sched.run_job(job)
assert success is True, error
assert response.strip() == str(profile_home.resolve())
assert os.environ["HERMES_HOME"] == str(root)
assert sched._get_hermes_home() == root
def test_run_job_without_profile_leaves_hermes_home_untouched(
self, isolated_cron_profile_home, monkeypatch
):
import cron.scheduler as sched
root, _profile_home = isolated_cron_profile_home
observed: dict = {}
self._install_agent_stubs(monkeypatch, observed)
job = {
"id": "noprof",
"name": "no-profile-job",
"profile": None,
"schedule_display": "manual",
}
success, *_ = sched.run_job(job)
assert success is True
assert observed["hermes_home_during_init"] == str(root)
assert os.environ["HERMES_HOME"] == str(root)
def test_run_job_falls_back_on_missing_runtime_profile(
self, isolated_cron_profile_home, monkeypatch
):
import cron.scheduler as sched
root, _profile_home = isolated_cron_profile_home
observed: dict = {}
self._install_agent_stubs(monkeypatch, observed)
job = {
"id": "missing-profile",
"name": "missing-profile-job",
"profile": "missing",
"schedule_display": "manual",
}
# Should succeed with fallback, not raise
success, _output, response, error = sched.run_job(job)
assert success is True, f"run_job should fallback, not fail: error={error!r}"
# Verify it used the default home, not the missing profile
assert observed["hermes_home_during_init"] == str(root)
assert os.environ["HERMES_HOME"] == str(root)
class TestTickProfilePartition:
def test_profile_and_workdir_combined(self, isolated_cron_profile_home, monkeypatch):
"""Both profile and workdir set — verify both are applied and restored."""
import cron.scheduler as sched
root, profile_home = isolated_cron_profile_home
observed: dict = {}
TestRunJobProfileContext._install_agent_stubs(monkeypatch, observed)
fake_workdir = str(root / "myproject")
(root / "myproject").mkdir()
job = {
"id": "combo",
"name": "combo-job",
"profile": "support",
"workdir": fake_workdir,
"schedule_display": "manual",
}
success, _output, _response, error = sched.run_job(job)
assert success is True, error
assert observed["hermes_home_during_init"] == str(profile_home.resolve())
assert os.environ.get("TERMINAL_CWD", "") != fake_workdir, \
"TERMINAL_CWD should be restored after job"
assert os.environ["HERMES_HOME"] == str(root)
assert sched._get_hermes_home() == root
def test_profile_jobs_run_sequentially(self, isolated_cron_profile_home, monkeypatch):
import threading
import cron.scheduler as sched
# Two profile jobs (both sequential) + one parallel job.
profile_a = {"id": "a", "name": "A", "profile": "default"}
profile_b = {"id": "b", "name": "B", "profile": "default"}
parallel_job = {"id": "c", "name": "C", "profile": None}
monkeypatch.setattr(sched, "get_due_jobs", lambda: [profile_a, profile_b, parallel_job])
monkeypatch.setattr(sched, "advance_next_run", lambda *_a, **_kw: None)
calls: list[tuple[str, str]] = []
order_lock = threading.Lock()
def fake_run_job(job):
with order_lock:
calls.append((job["id"], threading.current_thread().name))
return True, "output", "response", None
monkeypatch.setattr(sched, "run_job", fake_run_job)
monkeypatch.setattr(sched, "save_job_output", lambda _jid, _o: None)
monkeypatch.setattr(sched, "mark_job_run", lambda *_a, **_kw: None)
monkeypatch.setattr(sched, "_deliver_result", lambda *_a, **_kw: None)
n = sched.tick(verbose=False)
assert n == 3
ids = [job_id for job_id, _thread_name in calls]
# Sequential profile jobs preserve submission order relative to each
# other (single-thread pool).
assert ids.index("a") < ids.index("b")
# Sequential (profile) jobs run on the persistent single-thread
# cron-seq pool — NOT the main thread — so a long profile job never
# blocks the ticker. Parallel jobs run on the cron-parallel pool.
for jid in ("a", "b"):
seq_thread = next(t for job_id, t in calls if job_id == jid)
assert seq_thread != threading.current_thread().name
assert seq_thread.startswith("cron-seq"), seq_thread
par_thread = next(t for job_id, t in calls if job_id == "c")
assert par_thread.startswith("cron-parallel"), par_thread
+3 -3
View File
@@ -172,10 +172,10 @@ class TestSyncMode:
class TestSequentialPool:
"""Sequential (workdir/profile) jobs use the persistent cron-seq pool.
"""Sequential (workdir) jobs use the persistent cron-seq pool.
Verifies the follow-up fix: env/context-mutating jobs no longer run inline
in the ticker thread, so a long workdir/profile job can't starve the
Verifies the follow-up fix: env-mutating jobs no longer run inline
in the ticker thread, so a long workdir job can't starve the
schedule the same way the parallel path used to.
"""
+1 -1
View File
@@ -1487,7 +1487,7 @@ class TestRunJobConfigLogging:
}
# Mock heavy post-yaml work so the test only exercises the warning
# path. Without these mocks, _run_job_impl continues into provider
# path. Without these mocks, run_job continues into provider
# resolution and MCP discovery, both of which can spawn subprocesses
# / hit the network and have caused this test to time out on CI
# (>30s wall clock) under load. See PR #33661 follow-up.
+68
View File
@@ -197,8 +197,10 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
runner, _adapter = make_restart_runner()
popen_calls = []
monkeypatch.setattr(gateway_run.sys, "platform", "linux")
monkeypatch.setattr(gateway_run, "_resolve_hermes_bin", lambda: ["/usr/bin/hermes"])
monkeypatch.setattr(gateway_run.os, "getpid", lambda: 321)
monkeypatch.setenv("_HERMES_GATEWAY", "1")
monkeypatch.setattr(shutil, "which", lambda cmd: "/usr/bin/setsid" if cmd == "setsid" else None)
def fake_popen(cmd, **kwargs):
@@ -217,6 +219,72 @@ async def test_launch_detached_restart_command_uses_setsid(monkeypatch):
assert kwargs["start_new_session"] is True
assert kwargs["stdout"] is subprocess.DEVNULL
assert kwargs["stderr"] is subprocess.DEVNULL
# The watcher must NOT inherit the gateway marker, or the CLI's
# self-restart loop guard refuses to run `hermes gateway restart`.
assert kwargs["env"].get("_HERMES_GATEWAY") is None
def test_windows_gateway_venv_imports_add_site_packages(monkeypatch, tmp_path):
venv_dir = tmp_path / "venv"
site_packages = venv_dir / "Lib" / "site-packages"
pth_extra = tmp_path / "pywin32_system32"
site_packages.mkdir(parents=True)
pth_extra.mkdir()
(site_packages / "pywin32.pth").write_text(str(pth_extra), encoding="utf-8")
project_root = str(gateway_run.Path(gateway_run.__file__).resolve().parent.parent)
monkeypatch.setattr(gateway_run.sys, "platform", "win32")
monkeypatch.setattr(gateway_run.sys, "path", ["existing"])
monkeypatch.setenv("VIRTUAL_ENV", str(venv_dir))
monkeypatch.setenv("PYTHONPATH", "already-there")
gateway_run._ensure_windows_gateway_venv_imports()
assert gateway_run.sys.path[:2] == [project_root, str(site_packages)]
assert str(pth_extra) in gateway_run.sys.path
assert gateway_run.os.environ["VIRTUAL_ENV"] == str(venv_dir.resolve())
pythonpath = gateway_run.os.environ["PYTHONPATH"].split(gateway_run.os.pathsep)
assert pythonpath[:3] == [project_root, str(site_packages), "already-there"]
@pytest.mark.asyncio
async def test_windows_detached_restart_scrubs_gateway_marker(monkeypatch, tmp_path):
runner, _adapter = make_restart_runner()
popen_calls = []
venv_dir = tmp_path / "venv"
site_packages = venv_dir / "Lib" / "site-packages"
site_packages.mkdir(parents=True)
monkeypatch.setattr(gateway_run.sys, "platform", "win32")
monkeypatch.setattr(gateway_run, "_resolve_hermes_bin", lambda: ["hermes"])
monkeypatch.setattr(gateway_run.os, "getpid", lambda: 321)
monkeypatch.setenv("_HERMES_GATEWAY", "1")
monkeypatch.setenv("VIRTUAL_ENV", str(venv_dir))
import hermes_cli._subprocess_compat as subprocess_compat
monkeypatch.setattr(
subprocess_compat,
"windows_detach_popen_kwargs",
lambda: {},
)
def fake_popen(cmd, **kwargs):
popen_calls.append((cmd, kwargs))
return MagicMock()
monkeypatch.setattr(subprocess, "Popen", fake_popen)
await runner._launch_detached_restart_command()
assert len(popen_calls) == 1
cmd, kwargs = popen_calls[0]
assert cmd[-3:] == ["hermes", "gateway", "restart"]
assert kwargs["env"].get("_HERMES_GATEWAY") is None
assert kwargs["env"]["VIRTUAL_ENV"] == str(venv_dir)
assert str(site_packages) in kwargs["env"]["PYTHONPATH"].split(gateway_run.os.pathsep)
assert kwargs["stdout"] is subprocess.DEVNULL
assert kwargs["stderr"] is subprocess.DEVNULL
# ── Shutdown notification tests ──────────────────────────────────────
+69
View File
@@ -1488,3 +1488,72 @@ async def test_terminal_progress_no_bash_block_in_verbose_mode(monkeypatch, tmp_
all_content = " ".join(call["content"] for call in adapter.sent)
all_content += " ".join(call["content"] for call in adapter.edits)
assert "```bash" not in all_content
class MultiTerminalCommandAgent:
"""Emits several consecutive terminal tool.started events, then a
different tool, then terminal again — to exercise header collapsing."""
def __init__(self, **kwargs):
self.tool_progress_callback = kwargs.get("tool_progress_callback")
self.tools = []
def run_conversation(self, message, conversation_history=None, task_id=None):
cb = self.tool_progress_callback
cb("tool.started", "terminal", "echo one", {"command": "echo one"})
cb("tool.started", "terminal", "echo two", {"command": "echo two"})
cb("tool.started", "terminal", "echo three", {"command": "echo three"})
cb("tool.started", "web_search", "query stuff", {"query": "query stuff"})
cb("tool.started", "terminal", "echo four", {"command": "echo four"})
time.sleep(0.35)
return {"final_response": "done", "messages": [], "api_calls": 1}
@pytest.mark.asyncio
async def test_consecutive_terminal_progress_collapses_headers(monkeypatch, tmp_path):
"""Back-to-back terminal calls render ONE "terminal" header followed by
adjacent code blocks; a different tool in between resets the header so the
next terminal call gets a fresh one."""
monkeypatch.setenv("HERMES_TOOL_PROGRESS_MODE", "all")
fake_dotenv = types.ModuleType("dotenv")
fake_dotenv.load_dotenv = lambda *args, **kwargs: None
monkeypatch.setitem(sys.modules, "dotenv", fake_dotenv)
fake_run_agent = types.ModuleType("run_agent")
fake_run_agent.AIAgent = MultiTerminalCommandAgent
monkeypatch.setitem(sys.modules, "run_agent", fake_run_agent)
import tools.terminal_tool # noqa: F401 - register terminal emoji
adapter = CodeBlockProgressAdapter(platform=Platform.TELEGRAM)
runner = _make_runner(adapter)
gateway_run = importlib.import_module("gateway.run")
monkeypatch.setattr(gateway_run, "_hermes_home", tmp_path)
monkeypatch.setattr(gateway_run, "_resolve_runtime_agent_kwargs", lambda: {"api_key": "***"})
source = SessionSource(
platform=Platform.TELEGRAM,
chat_id="12345",
chat_type="dm",
thread_id=None,
)
result = await runner._run_agent(
message="hello",
context_prompt="",
history=[],
source=source,
session_id="sess-terminal-consecutive",
session_key="agent:main:telegram:dm:12345",
)
assert result["final_response"] == "done"
contents = [call["content"] for call in adapter.sent] + [
call["content"] for call in adapter.edits
]
final = max(contents, key=len) if contents else ""
# All four commands present as code blocks.
for cmd in ("echo one", "echo two", "echo three", "echo four"):
assert cmd in final
# Exactly TWO terminal headers: one for the first run of three calls,
# one for the terminal call after web_search broke the streak.
assert final.count("terminal\n```") == 2
+163
View File
@@ -691,6 +691,169 @@ class TestSubcommandCompletion:
completions = _completions(SlashCommandCompleter(), "/help ")
assert completions == []
def test_tools_subcommand_completion(self):
"""`/tools ` should suggest list, disable, enable."""
completions = _completions(SlashCommandCompleter(), "/tools ")
texts = {c.text for c in completions}
assert texts == {"list", "disable", "enable"}
def test_tools_subcommand_prefix_filters(self):
completions = _completions(SlashCommandCompleter(), "/tools en")
texts = {c.text for c in completions}
assert texts == {"enable"}
def test_tools_enable_completes_toolset_names(self, monkeypatch):
"""`/tools enable ` should suggest currently-disabled toolsets."""
from hermes_cli import commands as commands_mod
# `web` is enabled, `spotify` is disabled — enabling should only offer
# the disabled ones.
monkeypatch.setattr(
"hermes_cli.tools_config._get_platform_tools",
lambda *_a, **_k: {"web", "file"},
)
monkeypatch.setattr("hermes_cli.config.load_config", lambda: {})
monkeypatch.setattr(
"hermes_cli.tools_config._get_plugin_toolset_keys",
lambda: set(),
)
completions = _completions(SlashCommandCompleter(), "/tools enable ")
texts = {c.text for c in completions}
# Should include disabled toolsets, exclude already-enabled ones.
assert "web" not in texts
assert "file" not in texts
assert "spotify" in texts
def test_tools_disable_completes_enabled_toolsets_only(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.tools_config._get_platform_tools",
lambda *_a, **_k: {"web", "file"},
)
monkeypatch.setattr("hermes_cli.config.load_config", lambda: {})
monkeypatch.setattr(
"hermes_cli.tools_config._get_plugin_toolset_keys",
lambda: set(),
)
completions = _completions(SlashCommandCompleter(), "/tools disable ")
texts = {c.text for c in completions}
# Should include enabled toolsets, exclude disabled ones.
assert texts == {"web", "file"}
def test_tools_enable_partial_filters(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.tools_config._get_platform_tools",
lambda *_a, **_k: set(),
)
monkeypatch.setattr("hermes_cli.config.load_config", lambda: {})
monkeypatch.setattr(
"hermes_cli.tools_config._get_plugin_toolset_keys",
lambda: set(),
)
completions = _completions(SlashCommandCompleter(), "/tools enable sp")
texts = {c.text for c in completions}
assert texts == {"spotify"}
def test_tools_enable_skips_already_listed(self, monkeypatch):
"""If the user already typed a name, don't suggest it again."""
monkeypatch.setattr(
"hermes_cli.tools_config._get_platform_tools",
lambda *_a, **_k: set(),
)
monkeypatch.setattr("hermes_cli.config.load_config", lambda: {})
monkeypatch.setattr(
"hermes_cli.tools_config._get_plugin_toolset_keys",
lambda: set(),
)
completions = _completions(SlashCommandCompleter(), "/tools enable spotify ")
texts = {c.text for c in completions}
assert "spotify" not in texts
def test_tools_suggests_mcp_server_prefixes(self, monkeypatch):
monkeypatch.setattr(
"hermes_cli.tools_config._get_platform_tools",
lambda *_a, **_k: set(),
)
monkeypatch.setattr(
"hermes_cli.config.load_config",
lambda: {"mcp_servers": {"github": {}, "linear": {}}},
)
monkeypatch.setattr(
"hermes_cli.tools_config._get_plugin_toolset_keys",
lambda: set(),
)
completions = _completions(SlashCommandCompleter(), "/tools enable git")
texts = {c.text for c in completions}
assert "github:" in texts
def _fake_gateway(self, monkeypatch, platforms):
"""Patch load_gateway_config with a fake whose connected platforms are
the keys of `platforms` (name -> home as None or a (chat_id, name) tuple).
"""
from types import SimpleNamespace
enums = {name: SimpleNamespace(value=name) for name in platforms}
homes = {
name: (None if home is None else SimpleNamespace(chat_id=home[0], name=home[1]))
for name, home in platforms.items()
}
fake = SimpleNamespace(
get_connected_platforms=lambda: list(enums.values()),
get_home_channel=lambda p: homes[p.value],
)
monkeypatch.setattr("gateway.config.load_gateway_config", lambda: fake)
def test_handoff_completes_connected_platforms(self, monkeypatch):
"""`/handoff ` offers connected platforms, with or without a home channel."""
self._fake_gateway(
monkeypatch,
{
"telegram": ("123", "Me"),
"discord": None, # no home channel yet -> still listed
},
)
texts = {c.text for c in _completions(SlashCommandCompleter(), "/handoff ")}
assert texts == {"telegram", "discord"}
def test_handoff_filters_by_prefix(self, monkeypatch):
self._fake_gateway(
monkeypatch,
{
"telegram": ("1", "H"),
"signal": ("2", "H"),
},
)
texts = {c.text for c in _completions(SlashCommandCompleter(), "/handoff te")}
assert texts == {"telegram"}
def test_handoff_no_completion_after_platform_chosen(self, monkeypatch):
self._fake_gateway(monkeypatch, {"telegram": ("1", "H")})
assert _completions(SlashCommandCompleter(), "/handoff telegram ") == []
def test_handoff_completion_swallows_config_errors(self, monkeypatch):
def _boom():
raise RuntimeError("no gateway config")
monkeypatch.setattr("gateway.config.load_gateway_config", _boom)
assert _completions(SlashCommandCompleter(), "/handoff ") == []
def test_personality_completes_configured_personalities(self):
"""`/personality ` lists real personalities, not just `none`.
Regression: the completer read load_config().agent.personalities, a path
that never exists, so it always came back empty. It must resolve from the
CLI config the runtime actually applies (which ships built-ins).
"""
texts = {c.text for c in _completions(SlashCommandCompleter(), "/personality ")}
assert "none" in texts
assert len(texts) > 1
# ── Ghost text (SlashCommandAutoSuggest) ────────────────────────────────
-6
View File
@@ -55,7 +55,6 @@ class TestCronCommandLifecycle:
repeat=None,
skill=None,
skills=["maps", "blogwatcher"],
profile="default",
clear_skills=False,
)
)
@@ -64,7 +63,6 @@ class TestCronCommandLifecycle:
assert updated["name"] == "Edited Job"
assert updated["prompt"] == "Revised prompt"
assert updated["schedule_display"] == "every 120m"
assert updated["profile"] == "default"
cron_command(
Namespace(
@@ -77,14 +75,12 @@ class TestCronCommandLifecycle:
repeat=None,
skill=None,
skills=None,
profile="",
clear_skills=True,
)
)
cleared = get_job(job["id"])
assert cleared["skills"] == []
assert cleared["skill"] is None
assert cleared["profile"] is None
out = capsys.readouterr().out
assert "Updated job" in out
@@ -100,7 +96,6 @@ class TestCronCommandLifecycle:
repeat=None,
skill=None,
skills=["blogwatcher", "maps"],
profile="default",
)
)
out = capsys.readouterr().out
@@ -110,7 +105,6 @@ class TestCronCommandLifecycle:
assert len(jobs) == 1
assert jobs[0]["skills"] == ["blogwatcher", "maps"]
assert jobs[0]["name"] == "Skill combo"
assert jobs[0]["profile"] == "default"
def test_list_does_not_crash_when_repeat_is_null(self, tmp_cron_dir, capsys):
"""A one-shot job can be persisted with ``"repeat": null``. `cron
@@ -47,20 +47,19 @@ def test_cron_aliases():
def test_cron_create_options():
parser = _build()
ns = parser.parse_args([
"cron", "create", "0 9 * * *", "do the thing",
"cron", "create", "0 9 * * *", "daily task prompt",
"--name", "daily", "--deliver", "origin", "--repeat", "3",
"--skill", "a", "--skill", "b", "--no-agent",
"--workdir", "/tmp/x", "--profile", "work",
"--workdir", "/tmp/x",
])
assert ns.schedule == "0 9 * * *"
assert ns.prompt == "do the thing"
assert ns.prompt == "daily task prompt"
assert ns.name == "daily"
assert ns.deliver == "origin"
assert ns.repeat == 3
assert ns.skills == ["a", "b"]
assert ns.no_agent is True
assert ns.workdir == "/tmp/x"
assert ns.profile == "work"
def test_cron_edit_no_agent_tristate():
@@ -201,6 +201,91 @@ class TestWebhookEndpoints:
r = self.client.post("/api/webhooks", json={"name": "gh", "deliver": "log"})
assert r.status_code == 400
def test_enable_platform_starts_gateway_restart(self, monkeypatch):
import hermes_cli.web_server as ws
from hermes_cli.config import load_config
ws._ACTION_PROCS.pop("gateway-restart", None)
restart_calls = []
class FakeRestartProc:
pid = 4242
def fake_spawn_action(subcommand, name):
restart_calls.append((subcommand, name))
return FakeRestartProc()
monkeypatch.setattr(ws, "_spawn_hermes_action", fake_spawn_action)
r = self.client.post("/api/webhooks/enable")
assert r.status_code == 200
assert r.json() == {
"ok": True,
"platform": "webhook",
"enabled": True,
"needs_restart": False,
"restart_started": True,
"restart_action": "gateway-restart",
"restart_pid": 4242,
}
assert restart_calls == [(["gateway", "restart"], "gateway-restart")]
assert load_config()["platforms"]["webhook"]["enabled"] is True
assert self.client.get("/api/webhooks").json()["enabled"] is True
def test_enable_platform_reports_restart_failure_after_save(self, monkeypatch):
import hermes_cli.web_server as ws
from hermes_cli.config import load_config
ws._ACTION_PROCS.pop("gateway-restart", None)
def fail_spawn_action(subcommand, name):
assert subcommand == ["gateway", "restart"]
assert name == "gateway-restart"
raise RuntimeError("supervisor unavailable")
monkeypatch.setattr(ws, "_spawn_hermes_action", fail_spawn_action)
r = self.client.post("/api/webhooks/enable")
assert r.status_code == 200
data = r.json()
assert data["ok"] is True
assert data["platform"] == "webhook"
assert data["enabled"] is True
assert data["needs_restart"] is True
assert data["restart_started"] is False
assert "supervisor unavailable" in data["restart_error"]
assert load_config()["platforms"]["webhook"]["enabled"] is True
def test_enable_platform_reuses_inflight_gateway_restart(self, monkeypatch):
import hermes_cli.web_server as ws
from hermes_cli.config import load_config
ws._ACTION_PROCS.pop("gateway-restart", None)
class FakeRunningProc:
pid = 5151
def poll(self):
return None
monkeypatch.setitem(ws._ACTION_PROCS, "gateway-restart", FakeRunningProc())
def fail_spawn_action(subcommand, name):
raise AssertionError("must not spawn a second concurrent restart")
monkeypatch.setattr(ws, "_spawn_hermes_action", fail_spawn_action)
r = self.client.post("/api/webhooks/enable")
assert r.status_code == 200
data = r.json()
assert data["needs_restart"] is False
assert data["restart_started"] is True
assert data["restart_pid"] == 5151
assert load_config()["platforms"]["webhook"]["enabled"] is True
class TestOpsEndpoints:
@pytest.fixture(autouse=True)
@@ -622,6 +707,10 @@ class TestAdminEndpointsAuthGate:
resp = self.client.get(path)
assert resp.status_code in (401, 403)
def test_webhooks_enable_post_gated(self):
resp = self.client.post("/api/webhooks/enable")
assert resp.status_code in (401, 403)
class TestUpdateCheckEndpoint:
"""``GET /api/hermes/update/check`` reports availability without applying.
@@ -953,4 +1042,3 @@ class TestToolsConfigEndpoints:
kwargs["json"] = payload
r = fn(path, **kwargs)
assert r.status_code == 401, f"{method} {path} not gated"
+13
View File
@@ -975,6 +975,19 @@ def test_toolset_has_keys_treats_no_key_providers_as_configured():
assert _toolset_has_keys("computer_use", config) is True
def test_web_no_prompt_when_usable_keyless():
"""Fresh install: web works via the free Parallel MCP, so enabling the web
toolset should not force provider setup."""
with patch("tools.web_tools.check_web_api_key", return_value=True):
assert _toolset_needs_configuration_prompt("web", {}) is False
def test_web_no_prompt_when_extract_backend_is_extract_capable():
with patch("tools.web_tools.check_web_api_key", return_value=True):
cfg = {"web": {"extract_backend": "parallel"}}
assert _toolset_needs_configuration_prompt("web", cfg) is False
def test_computer_use_needs_configuration_when_cua_driver_post_setup_pending():
"""No-key providers can still need setup when their post_setup is unsatisfied.
+40
View File
@@ -425,3 +425,43 @@ def test_tui_launch_install_uses_workspace_scope(
install_cmd = npm_calls[0]
assert "--workspace" in install_cmd
assert "ui-tui" in install_cmd
def test_make_tui_argv_omits_workspace_when_tui_has_own_lockfile(
tmp_path: Path, main_mod, monkeypatch
) -> None:
"""When ui-tui/ has its own package-lock.json, _workspace_root returns
tui_dir itself. npm install --workspace ui-tui would fail in that case
because npm cannot find a workspace named "ui-tui" inside ui-tui/.
The fix omits --workspace and runs plain npm install from tui_dir.
See #42973.
"""
tui_dir = tmp_path / "ui-tui"
tui_dir.mkdir()
(tui_dir / "package.json").write_text("{}")
# Simulate curl-install layout: tui_dir has its own lockfile
(tui_dir / "package-lock.json").write_text("{}")
# Parent also has lockfile (but _workspace_root prefers tui_dir's own)
(tmp_path / "package-lock.json").write_text("{}")
monkeypatch.delenv("TERMUX_VERSION", raising=False)
monkeypatch.setenv("PREFIX", "/usr")
monkeypatch.setattr(main_mod, "_tui_need_npm_install", lambda _root: True)
monkeypatch.setattr(main_mod.shutil, "which", lambda name: f"/bin/{name}")
calls = []
def fake_run(*args, **kwargs):
calls.append((args, kwargs))
return types.SimpleNamespace(returncode=0, stdout="", stderr="")
monkeypatch.setattr(main_mod.subprocess, "run", fake_run)
main_mod._make_tui_argv(tui_dir, tui_dev=False)
install_cmd = calls[0][0][0]
# Must NOT contain --workspace when npm_cwd == tui_dir
assert "--workspace" not in install_cmd, (
f"npm install should omit --workspace when tui_dir has its own lockfile, got: {install_cmd}"
)
assert install_cmd[:2] == ["/bin/npm", "install"]
# cwd must be tui_dir (standalone), not parent
assert calls[0][1]["cwd"] == str(tui_dir)
@@ -0,0 +1,210 @@
"""Regression tests for dashboard profile-scoped skills/toolsets management.
"Set as active" on the Profiles page only flips the sticky ``active_profile``
file (future CLI/gateway runs) it never retargets the running dashboard
process. Before the ``profile`` parameter existed, toggling a skill after
"activating" a profile silently wrote into the dashboard's own config.
These tests pin the new behavior: reads and writes land in the REQUESTED
profile's HERMES_HOME, and the dashboard's own profile stays untouched.
"""
import pytest
import yaml
def _write_skill(skills_dir, name, description="test skill"):
d = skills_dir / name
d.mkdir(parents=True, exist_ok=True)
(d / "SKILL.md").write_text(
f"---\nname: {name}\ndescription: {description}\n---\n\n# {name}\n",
encoding="utf-8",
)
@pytest.fixture
def isolated_profiles(tmp_path, monkeypatch, _isolate_hermes_home):
"""Isolated default home + one named profile, each with its own skills."""
from hermes_constants import get_hermes_home
from hermes_cli import profiles
default_home = get_hermes_home()
profiles_root = default_home / "profiles"
worker_home = profiles_root / "worker_alpha"
for home in (default_home, worker_home):
(home / "skills").mkdir(parents=True, exist_ok=True)
(home / "config.yaml").write_text("{}\n", encoding="utf-8")
_write_skill(default_home / "skills", "dashboard-skill")
_write_skill(worker_home / "skills", "worker-skill")
monkeypatch.setattr(profiles, "_get_default_hermes_home", lambda: default_home)
monkeypatch.setattr(profiles, "_get_profiles_root", lambda: profiles_root)
return {"default": default_home, "worker_alpha": worker_home}
@pytest.fixture
def client(monkeypatch, isolated_profiles):
try:
from starlette.testclient import TestClient
except ImportError:
pytest.skip("fastapi/starlette not installed")
import hermes_state
from hermes_constants import get_hermes_home
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
c = TestClient(app)
c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
return c
def _load_cfg(home):
return yaml.safe_load((home / "config.yaml").read_text()) or {}
class TestProfileScopedSkills:
def test_skills_list_scopes_to_requested_profile(self, client, isolated_profiles):
resp = client.get("/api/skills", params={"profile": "worker_alpha"})
assert resp.status_code == 200
names = {s["name"] for s in resp.json()}
assert "worker-skill" in names
assert "dashboard-skill" not in names
def test_skills_list_without_profile_uses_dashboard_home(
self, client, isolated_profiles
):
resp = client.get("/api/skills")
assert resp.status_code == 200
names = {s["name"] for s in resp.json()}
assert "dashboard-skill" in names
assert "worker-skill" not in names
def test_toggle_writes_into_target_profile_only(self, client, isolated_profiles):
resp = client.put(
"/api/skills/toggle",
json={"name": "worker-skill", "enabled": False, "profile": "worker_alpha"},
)
assert resp.status_code == 200
assert resp.json() == {"ok": True, "name": "worker-skill", "enabled": False}
worker_cfg = _load_cfg(isolated_profiles["worker_alpha"])
assert "worker-skill" in worker_cfg.get("skills", {}).get("disabled", [])
# The dashboard's own config must stay untouched — this was the bug.
default_cfg = _load_cfg(isolated_profiles["default"])
assert "worker-skill" not in default_cfg.get("skills", {}).get("disabled", [])
def test_toggle_reenable_round_trip(self, client, isolated_profiles):
for enabled in (False, True):
client.put(
"/api/skills/toggle",
json={
"name": "worker-skill",
"enabled": enabled,
"profile": "worker_alpha",
},
)
worker_cfg = _load_cfg(isolated_profiles["worker_alpha"])
assert "worker-skill" not in worker_cfg.get("skills", {}).get("disabled", [])
def test_unknown_profile_returns_404(self, client, isolated_profiles):
resp = client.get("/api/skills", params={"profile": "no_such_profile"})
assert resp.status_code == 404
def test_invalid_profile_name_returns_400(self, client, isolated_profiles):
resp = client.get("/api/skills", params={"profile": "Bad Name!"})
assert resp.status_code == 400
def test_scope_restores_module_globals(self, client, isolated_profiles):
"""The SKILLS_DIR swap is per-request; the module global must be
restored even after a scoped call (cron-style locked swap)."""
import tools.skills_tool as skills_tool
before = skills_tool.SKILLS_DIR
client.get("/api/skills", params={"profile": "worker_alpha"})
assert skills_tool.SKILLS_DIR == before
class TestProfileScopedToolsets:
def test_toolset_toggle_scopes_to_profile(self, client, isolated_profiles):
resp = client.put(
"/api/tools/toolsets/x_search",
json={"enabled": True, "profile": "worker_alpha"},
)
assert resp.status_code == 200
worker_cfg = _load_cfg(isolated_profiles["worker_alpha"])
assert "x_search" in worker_cfg.get("platform_toolsets", {}).get("cli", [])
default_cfg = _load_cfg(isolated_profiles["default"])
assert "x_search" not in default_cfg.get("platform_toolsets", {}).get("cli", [])
listing = client.get(
"/api/tools/toolsets", params={"profile": "worker_alpha"}
).json()
assert {t["name"]: t for t in listing}["x_search"]["enabled"] is True
# Unscoped listing reflects the dashboard's own (untouched) config.
listing = client.get("/api/tools/toolsets").json()
assert {t["name"]: t for t in listing}["x_search"]["enabled"] is False
def test_toolset_toggle_unknown_profile_404(self, client, isolated_profiles):
resp = client.put(
"/api/tools/toolsets/x_search",
json={"enabled": True, "profile": "ghost"},
)
assert resp.status_code == 404
class TestProfileScopedHubActions:
def test_hub_install_spawns_with_profile_flag(
self, client, isolated_profiles, monkeypatch
):
"""Hub installs must go through a fresh ``hermes -p <profile>``
subprocess the in-process scope can't reach skills_hub's
import-time SKILLS_DIR binding."""
import hermes_cli.web_server as web_server
calls = []
class _FakeProc:
pid = 4242
def _fake_spawn(subcommand, name):
calls.append((list(subcommand), name))
return _FakeProc()
monkeypatch.setattr(web_server, "_spawn_hermes_action", _fake_spawn)
resp = client.post(
"/api/skills/hub/install",
json={"identifier": "official/demo", "profile": "worker_alpha"},
)
assert resp.status_code == 200
assert calls == [
(["-p", "worker_alpha", "skills", "install", "official/demo"], "skills-install")
]
def test_hub_install_without_profile_keeps_legacy_argv(
self, client, isolated_profiles, monkeypatch
):
import hermes_cli.web_server as web_server
calls = []
class _FakeProc:
pid = 4242
monkeypatch.setattr(
web_server,
"_spawn_hermes_action",
lambda subcommand, name: calls.append(list(subcommand)) or _FakeProc(),
)
resp = client.post(
"/api/skills/hub/install", json={"identifier": "official/demo"}
)
assert resp.status_code == 200
assert calls == [["skills", "install", "official/demo"]]
def test_hub_install_unknown_profile_404(self, client, isolated_profiles):
resp = client.post(
"/api/skills/hub/install",
json={"identifier": "official/demo", "profile": "ghost"},
)
assert resp.status_code == 404
+35
View File
@@ -142,6 +142,11 @@ class TestBuildWebUISkipsWhenFresh:
def test_npm_install_uses_workspace_web_scope(self, tmp_path):
web_dir, _ = _make_web_dir(tmp_path)
# Real workspace checkout: the single lockfile lives at the root, so
# _workspace_root(web_dir) resolves to the parent and --workspace web
# scopes the install. (Without a root lockfile, web_dir IS the root and
# --workspace would be dropped — see test below and #42973.)
(tmp_path / "package-lock.json").write_text("{}", encoding="utf-8")
mock_cp = __import__("subprocess").CompletedProcess([], 0, stdout="", stderr="")
build_ok = __import__("subprocess").CompletedProcess([], 0, stdout="", stderr="")
with patch("hermes_cli.main.shutil.which", return_value="/usr/bin/npm"), \
@@ -153,6 +158,36 @@ class TestBuildWebUISkipsWhenFresh:
assert "--workspace" in install_cmd
assert install_cmd[install_cmd.index("--workspace") + 1] == "web"
def test_web_install_omits_workspace_when_web_has_own_lockfile(
self, tmp_path, monkeypatch
):
"""web/ with its own lockfile => _workspace_root returns web_dir, so
--workspace web would fail (npm can't find that workspace from inside
web/). The flag must be dropped and the install run plainly from web_dir.
Symmetric to the TUI fix in test_tui_npm_install.py. See #42973.
With web's own lockfile present at cwd, _run_npm_install_deterministic
uses ``npm ci`` (not ``npm install``).
"""
web_dir, _ = _make_web_dir(tmp_path)
(web_dir / "package-lock.json").write_text("{}", encoding="utf-8")
(tmp_path / "package-lock.json").write_text("{}", encoding="utf-8")
monkeypatch.delenv("TERMUX_VERSION", raising=False)
monkeypatch.setenv("PREFIX", "/usr")
install_cp = __import__("subprocess").CompletedProcess([], 0, stdout="", stderr="")
build_cp = __import__("subprocess").CompletedProcess([], 0, stdout="", stderr="")
with patch("hermes_cli.main.shutil.which", return_value="/usr/bin/npm"), \
patch("hermes_cli.main.subprocess.run", return_value=install_cp) as mock_run, \
patch("hermes_cli.main._run_with_idle_timeout", return_value=build_cp):
result = _build_web_ui(web_dir)
assert result is True
args, kwargs = mock_run.call_args
assert "--workspace" not in args[0]
assert args[0] == ["/usr/bin/npm", "ci", "--silent"]
assert kwargs["cwd"] == web_dir
def test_web_build_uses_idle_timeout_helper(self, tmp_path):
"""npm run build now goes through _run_with_idle_timeout (issue #33788).
@@ -0,0 +1,383 @@
"""Keyless Parallel search via the free hosted Search MCP.
Covers the transport added in ``plugins/web/parallel/provider.py`` that lets
``web_search`` work with no ``PARALLEL_API_KEY``:
- ``_mcp_headers`` Bearer attached only when a key is held
- ``_decode_mcp_envelope`` plain-JSON and SSE (``data:``) response bodies
- ``_mcp_payload`` structuredContent preferred, text-block JSON fallback, errors
- ``_mcp_web_search`` full handshake (mocked transport) standard search shape
- ``ParallelWebSearchProvider.search`` keyless path routes to the MCP
"""
from __future__ import annotations
import asyncio
import json
from unittest.mock import patch
import pytest
import plugins.web.parallel.provider as pp
# ─── _mcp_headers ──────────────────────────────────────────────────────────
class TestMcpHeaders:
def test_anonymous_has_no_authorization(self):
h = pp._mcp_headers(session_id=None, api_key=None)
assert "Authorization" not in h
assert h["Accept"] == "application/json, text/event-stream"
assert "Mcp-Session-Id" not in h
def test_user_agent_is_generic_not_hermes(self):
# Telemetry policy: no third-party usage attribution without opt-in.
# The UA must be set (not python-httpx default) but must not name
# hermes, on both the anonymous and keyed paths.
for ua in (
pp._mcp_headers(session_id=None, api_key=None)["User-Agent"],
pp._mcp_headers(session_id="sid", api_key="pk-live")["User-Agent"],
):
assert ua == f"{pp._MCP_CLIENT_NAME}/{pp._MCP_CLIENT_VERSION}"
assert "hermes" not in ua.lower()
def test_session_id_and_bearer_when_present(self):
h = pp._mcp_headers(session_id="sid-123", api_key="pk-live")
assert h["Mcp-Session-Id"] == "sid-123"
assert h["Authorization"] == "Bearer pk-live"
# ─── SSE / JSON-RPC parsing ──────────────────────────────────────────────────
class TestMcpResponseParsing:
def test_plain_json_matched_by_id(self):
body = '{"jsonrpc":"2.0","id":"abc","result":{"ok":true}}'
assert pp._mcp_response_envelope(body, "abc")["result"]["ok"] is True
def test_sse_selects_response_for_request_id_skipping_notifications(self):
# A progress notification (no id) precedes the real result; an unrelated
# response id is also present. We must pick the one matching our id.
body = (
'event: message\ndata: {"jsonrpc":"2.0","method":"notifications/progress","params":{"p":1}}\n\n'
'event: message\ndata: {"jsonrpc":"2.0","id":"other","result":{"ok":false}}\n\n'
'event: message\ndata: {"jsonrpc":"2.0","id":"req-1","result":{"ok":true}}\n\n'
)
env = pp._mcp_response_envelope(body, "req-1")
assert env["result"]["ok"] is True
def test_sse_multiline_data_concatenated(self):
body = 'data: {"jsonrpc":"2.0","id":"x",\ndata: "result":{"n":42}}\n\n'
assert pp._mcp_response_envelope(body, "x")["result"]["n"] == 42
def test_falls_back_to_last_result_when_id_absent(self):
body = '{"jsonrpc":"2.0","id":"server-chose","result":{"ok":true}}'
# request id doesn't match, but there's a single result → use it
assert pp._mcp_response_envelope(body, "mismatch")["result"]["ok"] is True
def test_empty_body(self):
assert pp._mcp_response_envelope("", "x") == {}
assert pp._mcp_response_envelope(" ", "x") == {}
def test_batched_json_array_flattened(self):
# Streamable HTTP may batch messages into a JSON array.
body = ('[{"jsonrpc":"2.0","method":"notifications/progress"},'
'{"jsonrpc":"2.0","id":"req-9","result":{"ok":true}}]')
assert pp._mcp_response_envelope(body, "req-9")["result"]["ok"] is True
def test_batched_sse_data_array_flattened(self):
body = 'data: [{"jsonrpc":"2.0","id":"a","result":{"n":1}}]\n\n'
assert pp._mcp_response_envelope(body, "a")["result"]["n"] == 1
# ─── _mcp_payload ────────────────────────────────────────────────────────────
class TestMcpPayload:
def test_prefers_structured_content(self):
env = {"result": {"structuredContent": {"results": [{"url": "u"}]},
"content": [{"type": "text", "text": "ignored"}]}}
assert pp._mcp_payload(env) == {"results": [{"url": "u"}]}
def test_parses_text_block_json(self):
inner = {"search_id": "s1", "results": [{"url": "u", "title": "t"}]}
env = {"result": {"content": [{"type": "text", "text": json.dumps(inner)}]}}
assert pp._mcp_payload(env)["search_id"] == "s1"
def test_raises_on_jsonrpc_error(self):
with pytest.raises(RuntimeError, match="Parallel MCP error"):
pp._mcp_payload({"error": {"code": -32000, "message": "boom"}})
def test_raises_on_tool_iserror(self):
with pytest.raises(RuntimeError, match="Parallel MCP tool error"):
pp._mcp_payload({"result": {"isError": True, "content": []}})
# ─── _mcp_web_search (mocked transport) ──────────────────────────────────────
class _FakeResponse:
def __init__(self, *, text="", headers=None):
self.text = text
self.headers = headers or {}
def raise_for_status(self):
return None
class _FakeClient:
"""Stands in for httpx.Client: replays init → ack → tools/call."""
def __init__(self, search_payload, init_session_id="server-sid"):
self._search_payload = search_payload
self._init_session_id = init_session_id
self.calls = []
def __enter__(self):
return self
def __exit__(self, *exc):
return False
def post(self, url, headers=None, json=None):
self.calls.append({"headers": headers, "json": json})
req = json or {}
method = req.get("method")
req_id = req.get("id")
if method == "initialize":
# Echo the request id, as the real server does.
return _FakeResponse(
text=json_dumps({"jsonrpc": "2.0", "id": req_id,
"result": {"protocolVersion": "2099-01-01"}}),
headers=(
{"mcp-session-id": self._init_session_id}
if self._init_session_id is not None
else {}
),
)
if method == "notifications/initialized":
return _FakeResponse(text="")
# tools/call
envelope = {"jsonrpc": "2.0", "id": req_id, "result": {
"content": [{"type": "text", "text": json_dumps(self._search_payload)}],
}}
return _FakeResponse(text=json_dumps(envelope))
def json_dumps(obj):
return json.dumps(obj)
class TestMcpWebSearch:
def _payload(self, n):
return {"search_id": "s", "results": [
{"url": f"https://ex/{i}", "title": f"t{i}",
"excerpts": [f"a{i}", f"b{i}"]}
for i in range(n)
]}
def test_returns_standard_shape_and_handshake(self):
fake = _FakeClient(self._payload(3))
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_search("hello", limit=5, api_key=None)
assert out["success"] is True
# Free-tier results credit Parallel.
assert "Parallel" in out["attribution"]
web = out["data"]["web"]
assert [r["position"] for r in web] == [1, 2, 3]
assert web[0]["url"] == "https://ex/0"
assert web[0]["description"] == "a0 b0" # excerpts joined
# handshake order
methods = [c["json"].get("method") for c in fake.calls]
assert methods == ["initialize", "notifications/initialized", "tools/call"]
# session id from the initialize response header is reused
assert fake.calls[-1]["headers"]["Mcp-Session-Id"] == "server-sid"
def test_stateless_server_no_session_header_not_invented(self):
# A stateless Streamable-HTTP server may omit mcp-session-id on
# initialize; we must NOT invent one (sending an unissued session id can
# get follow-up requests rejected). The follow-ups carry no header.
fake = _FakeClient(self._payload(1), init_session_id=None)
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_search("hello", limit=5, api_key=None)
assert out["success"] is True
follow_ups = [c for c in fake.calls if c["json"].get("method") != "initialize"]
assert follow_ups, "expected notifications/initialized + tools/call"
assert all("Mcp-Session-Id" not in c["headers"] for c in follow_ups)
# anonymous → no Authorization on any call
assert all("Authorization" not in c["headers"] for c in fake.calls)
# tools/call mirrors query into objective + search_queries
args = fake.calls[-1]["json"]["params"]["arguments"]
assert args["objective"] == "hello"
assert args["search_queries"] == ["hello"]
def test_limit_is_applied_client_side(self):
fake = _FakeClient(self._payload(10))
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_search("q", limit=2, api_key=None)
assert len(out["data"]["web"]) == 2
def test_bearer_attached_when_key_present(self):
fake = _FakeClient(self._payload(1))
with patch.object(pp.httpx, "Client", return_value=fake):
pp._mcp_web_search("q", limit=1, api_key="pk-live")
assert all(c["headers"]["Authorization"] == "Bearer pk-live" for c in fake.calls)
def test_negotiated_protocol_version_echoed_post_init(self):
fake = _FakeClient(self._payload(1))
with patch.object(pp.httpx, "Client", return_value=fake):
pp._mcp_web_search("q", limit=1, api_key=None)
# initialize request doesn't carry the (not-yet-negotiated) version...
assert "MCP-Protocol-Version" not in fake.calls[0]["headers"]
# ...but notifications/initialized and tools/call echo the negotiated one.
assert fake.calls[1]["headers"]["MCP-Protocol-Version"] == "2099-01-01"
assert fake.calls[-1]["headers"]["MCP-Protocol-Version"] == "2099-01-01"
# ─── provider.search keyless routing ─────────────────────────────────────────
class TestProviderKeylessSearch:
def test_search_without_key_uses_mcp(self, monkeypatch):
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
captured = {}
def _fake(query, limit, api_key):
captured.update(query=query, limit=limit, api_key=api_key)
return {"success": True, "data": {"web": []}}
monkeypatch.setattr(pp, "_mcp_web_search", _fake)
out = pp.ParallelWebSearchProvider().search("kittens", limit=4)
assert out["success"] is True
assert captured == {"query": "kittens", "limit": 4, "api_key": None}
def test_is_available_reflects_key(self, monkeypatch):
# is_available() gates the registry's active-provider walk + picker, so
# it's key-based (keyless dispatch is handled by _get_backend, not this).
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
assert pp.ParallelWebSearchProvider().is_available() is False
monkeypatch.setenv("PARALLEL_API_KEY", "k")
assert pp.ParallelWebSearchProvider().is_available() is True
# ─── web_fetch (keyless extract) ─────────────────────────────────────────────
class TestMcpWebFetch:
def _payload(self, urls):
return {"extract_id": "e1", "results": [
{"url": u, "title": f"T{i}", "publish_date": None,
"excerpts": [f"chunk-a-{i}", f"chunk-b-{i}"]}
for i, u in enumerate(urls)
]}
def test_maps_to_extract_shape(self):
urls = ["https://a.test", "https://b.test"]
fake = _FakeClient(self._payload(urls))
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_fetch(urls, api_key=None)
assert [r["url"] for r in out] == urls
assert out[0]["content"] == "chunk-a-0\n\nchunk-b-0"
assert out[0]["raw_content"] == out[0]["content"]
assert out[0]["metadata"] == {"sourceURL": "https://a.test", "title": "T0"}
# tools/call targeted web_fetch, requesting full page bodies.
args = fake.calls[-1]["json"]["params"]
assert args["name"] == "web_fetch"
assert args["arguments"]["urls"] == urls
assert args["arguments"]["full_content"] is True
assert args["arguments"]["session_id"].startswith(f"{pp._MCP_CLIENT_NAME}-")
def test_prefers_full_content_over_excerpts(self):
payload = {"results": [
{"url": "https://a.test", "title": "T",
"excerpts": ["snippet"], "full_content": "the entire page body"},
]}
fake = _FakeClient(payload)
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_fetch(["https://a.test"], api_key=None)
assert out[0]["content"] == "the entire page body"
def test_missing_url_becomes_error_entry(self):
# Server returns only one of the two requested URLs.
fake = _FakeClient(self._payload(["https://a.test"]))
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_fetch(["https://a.test", "https://missing.test"], api_key=None)
assert len(out) == 2
missing = [r for r in out if r["url"] == "https://missing.test"][0]
assert "error" in missing
assert missing["content"] == ""
def test_preserves_order_and_duplicate_inputs(self):
# MCP returns each unique URL once; output must still be one row per
# input, in order, including the duplicate.
fake = _FakeClient(self._payload(["https://a.test", "https://b.test"]))
urls = ["https://b.test", "https://a.test", "https://b.test"]
with patch.object(pp.httpx, "Client", return_value=fake):
out = pp._mcp_web_fetch(urls, api_key=None)
assert [r["url"] for r in out] == urls # one row per input, in order
assert all("error" not in r for r in out) # all three resolved
def test_extract_without_key_uses_web_fetch(self, monkeypatch):
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
captured = {}
def _fake(urls, api_key):
captured.update(urls=list(urls), api_key=api_key)
return [{"url": urls[0], "title": "", "content": "x",
"raw_content": "x", "metadata": {}}]
monkeypatch.setattr(pp, "_mcp_web_fetch", _fake)
out = asyncio.run(pp.ParallelWebSearchProvider().extract(["https://x.test"]))
assert out[0]["content"] == "x"
assert captured == {"urls": ["https://x.test"], "api_key": None}
# ─── keyed v1 REST search ────────────────────────────────────────────────────
class TestKeyedV1Search:
def test_passes_max_results_and_omits_branding(self, monkeypatch):
monkeypatch.setenv("PARALLEL_API_KEY", "pk-live")
monkeypatch.delenv("PARALLEL_SEARCH_MODE", raising=False)
captured = {}
class _Res:
def __init__(self, url):
self.url, self.title, self.excerpts = url, "T", ["x"]
class _Resp:
results = [_Res(f"https://r/{i}") for i in range(10)]
class _Client:
def search(self, **kw):
captured.update(kw)
return _Resp()
monkeypatch.setattr(pp, "_get_sync_client", lambda: _Client())
out = pp.ParallelWebSearchProvider().search("q", limit=7)
assert out["success"] is True
# honors the caller's limit via advanced_settings.max_results
assert captured["advanced_settings"] == {"max_results": 7}
assert captured["mode"] == "advanced" # v1 default
assert captured["session_id"].startswith(f"{pp._MCP_CLIENT_NAME}-") # per-call id
assert len(out["data"]["web"]) == 7 # client-side slice
# paid path: no free-tier attribution, no [Parallel] label signal
assert "attribution" not in out
assert "provider" not in out
# ─── v1 search mode mapping ──────────────────────────────────────────────────
class TestResolveSearchMode:
@pytest.mark.parametrize("env,expected", [
(None, "advanced"), # default
("advanced", "advanced"),
("basic", "basic"),
("fast", "basic"), # legacy → basic
("one-shot", "basic"), # legacy → basic
("agentic", "advanced"), # legacy → advanced
("garbage", "advanced"), # invalid → default
("BASIC", "basic"), # case-insensitive
])
def test_mode_mapping(self, monkeypatch, env, expected):
if env is None:
monkeypatch.delenv("PARALLEL_SEARCH_MODE", raising=False)
else:
monkeypatch.setenv("PARALLEL_SEARCH_MODE", env)
assert pp._resolve_search_mode() == expected
@@ -193,11 +193,16 @@ class TestIsAvailable:
assert p.is_available() is True
def test_parallel_requires_api_key(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""is_available() is key-based — it gates the registry's active-provider
walk/picker. (Keyless search/extract still work via the free MCP through
_get_backend's terminal default, independent of this flag.)
"""
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
p = get_provider("parallel")
assert p is not None
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
assert p.is_available() is False
monkeypatch.setenv("PARALLEL_API_KEY", "real")
assert p.is_available() is True
@@ -422,17 +427,33 @@ class TestErrorResponseShapes:
assert result.get("success") is False
assert "error" in result
def test_parallel_extract_returns_per_url_errors_when_unconfigured(self) -> None:
def test_parallel_extract_keyless_uses_mcp_web_fetch(
self, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Without a key, extract routes to the free MCP web_fetch tool rather
than erroring. The MCP transport is mocked so the test stays offline."""
_ensure_plugins_loaded()
from agent.web_search_registry import get_provider
import plugins.web.parallel.provider as parallel_provider
monkeypatch.delenv("PARALLEL_API_KEY", raising=False)
captured = {}
def _fake_fetch(urls, api_key):
captured["urls"] = list(urls)
captured["api_key"] = api_key
return [{"url": urls[0], "title": "Example", "content": "body",
"raw_content": "body", "metadata": {"sourceURL": urls[0]}}]
monkeypatch.setattr(parallel_provider, "_mcp_web_fetch", _fake_fetch)
p = get_provider("parallel")
assert p is not None
result = asyncio.run(p.extract(["https://example.com"]))
assert isinstance(result, list)
assert len(result) == 1
assert "error" in result[0]
assert result[0]["url"] == "https://example.com"
assert result[0]["content"] == "body"
assert captured == {"urls": ["https://example.com"], "api_key": None}
def test_firecrawl_extract_returns_per_url_errors_when_unconfigured(self) -> None:
_ensure_plugins_loaded()
+41
View File
@@ -86,6 +86,47 @@ def test_session_context_uses_session_cwd(monkeypatch, tmp_path):
server._sessions.pop(sid, None)
def test_handoff_fail_marks_only_inflight_rows(monkeypatch):
class DbContext:
def __init__(self, db):
self.db = db
def __enter__(self):
return self.db
def __exit__(self, *_args):
return False
class FakeDb:
def __init__(self, state):
self.state = state
self.failed_with = None
def get_handoff_state(self, _key):
return {"state": self.state, "platform": "telegram", "error": None}
def fail_handoff(self, _key, error):
self.failed_with = error
self.state = "failed"
sid = "rt-handoff"
server._sessions[sid] = {"session_key": "stored-handoff"}
try:
pending = FakeDb("pending")
monkeypatch.setattr(server, "_session_db", lambda _session: DbContext(pending))
result = server._methods["handoff.fail"]("r1", {"session_id": sid, "error": "timed out"})
assert result["result"] == {"failed": True, "state": "failed"}
assert pending.failed_with == "timed out"
completed = FakeDb("completed")
monkeypatch.setattr(server, "_session_db", lambda _session: DbContext(completed))
result = server._methods["handoff.fail"]("r2", {"session_id": sid, "error": "late timeout"})
assert result["result"] == {"failed": False, "state": "completed"}
assert completed.failed_with is None
finally:
server._sessions.pop(sid, None)
def test_session_context_explicit_cwd_for_ephemeral_task(monkeypatch, tmp_path):
"""Background/preview tasks use ephemeral ids absent from `_sessions`, so the
parent workspace is passed explicitly; it must pin instead of clearing back
+60 -13
View File
@@ -167,6 +167,21 @@ class TestPerCapabilityBackendSelection:
monkeypatch.setenv("TAVILY_API_KEY", "test-key")
assert web_tools._get_search_backend() == "tavily"
def test_explicit_extract_backend_honored_when_unavailable(self, monkeypatch):
"""An explicit per-capability backend is honored even with no creds, so
its setup error surfaces instead of silently rerouting to the keyless
Parallel default (which would send user URLs to a different provider)."""
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {
"extract_backend": "firecrawl",
})
for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "FIRECRAWL_GATEWAY_URL"):
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False, raising=False)
# Resolves to firecrawl (not parallel) despite firecrawl being unavailable.
assert web_tools._get_extract_backend() == "firecrawl"
def test_falls_back_to_generic_backend_when_extract_backend_empty(self, monkeypatch):
from tools import web_tools
@@ -177,7 +192,7 @@ class TestPerCapabilityBackendSelection:
monkeypatch.setenv("PARALLEL_API_KEY", "test-key")
assert web_tools._get_extract_backend() == "parallel"
def test_search_backend_ignored_when_not_available(self, monkeypatch):
def test_explicit_search_backend_honored_when_unavailable(self, monkeypatch):
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {
@@ -186,8 +201,10 @@ class TestPerCapabilityBackendSelection:
})
monkeypatch.delenv("EXA_API_KEY", raising=False)
monkeypatch.setenv("FIRECRAWL_API_KEY", "fc-key")
# Should fall back to firecrawl since exa isn't configured
assert web_tools._get_search_backend() == "firecrawl"
# The explicit per-capability choice (exa) is honored even though it's
# unavailable, so its setup error surfaces — we don't silently reroute
# to the shared backend (or the keyless Parallel default).
assert web_tools._get_search_backend() == "exa"
def test_fully_backward_compatible_with_web_backend_only(self, monkeypatch):
from tools import web_tools
@@ -291,25 +308,55 @@ class TestUnconfiguredErrorEnvelopeParity:
):
monkeypatch.delenv(k, raising=False)
def test_unconfigured_search_emits_top_level_error(self, monkeypatch):
"""``web_search_tool`` with no creds returns ``{"error": "Error searching web: ..."}``
matching main's ``tool_error()`` envelope, not a per-result shape.
def test_extract_empty_urls_does_not_raise(self, monkeypatch):
"""Regression: empty (or fully SSRF-blocked) URL sets skip the dispatch
branch; the free-Parallel flag must still be initialized so the tool
returns an error envelope instead of UnboundLocalError."""
import asyncio
from tools import web_tools
self._clear_web_creds(monkeypatch)
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
out = asyncio.run(web_tools.web_extract_tool([], "markdown"))
# The key assertion is that it returns a normal error envelope (a
# string) rather than raising UnboundLocalError.
assert isinstance(out, str)
result = json.loads(out)
assert "error" in result
def test_unconfigured_search_falls_back_to_free_parallel(self, monkeypatch):
"""``web_search_tool`` with no creds routes to Parallel's free Search
MCP rather than erroring. The MCP transport is mocked so the test
stays offline; we assert dispatch landed on parallel and returned the
standard search envelope.
"""
from tools import web_tools
import plugins.web.parallel.provider as parallel_provider
self._clear_web_creds(monkeypatch)
# Reset firecrawl client cache so the unconfigured state is re-evaluated
monkeypatch.setattr(web_tools, "_firecrawl_client", None, raising=False)
monkeypatch.setattr(web_tools, "_firecrawl_client_config", None, raising=False)
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
captured = {}
def _fake_mcp(query, limit, api_key):
captured["query"] = query
captured["api_key"] = api_key
return {
"success": True,
"data": {"web": [
{"url": "https://example.com", "title": "Example",
"description": "hit", "position": 1},
]},
}
monkeypatch.setattr(parallel_provider, "_mcp_web_search", _fake_mcp)
result = json.loads(web_tools.web_search_tool("hello world", limit=3))
assert "error" in result, f"expected top-level 'error' key, got {result}"
# ``Error searching web:`` prefix comes from web_tools' top-level except handler
assert "Error searching web:" in result["error"]
assert "FIRECRAWL_API_KEY" in result["error"]
# No per-result burying
assert "results" not in result
assert result.get("success") is True, f"expected success, got {result}"
assert result["data"]["web"][0]["url"] == "https://example.com"
# Keyless path: dispatched to parallel with no Bearer token.
assert captured == {"query": "hello world", "api_key": None}
class TestDispatchersTriggerPluginDiscovery:
+6 -2
View File
@@ -190,7 +190,11 @@ class TestDDGSBackendWiring:
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True)
assert web_tools._get_backend() == "exa"
def test_auto_detect_picks_ddgs_as_last_resort(self, monkeypatch):
def test_auto_detect_prefers_keyless_parallel_over_ddgs(self, monkeypatch):
# With no credentials, keyless Parallel is the auto-detect default even
# when the ddgs package is installed — ddgs is search-only (can't
# extract), so Parallel is preferred so both search and extract work.
# ddgs remains reachable via an explicit web.backend=ddgs.
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
for key in ("FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "PARALLEL_API_KEY",
@@ -198,7 +202,7 @@ class TestDDGSBackendWiring:
monkeypatch.delenv(key, raising=False)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: True)
assert web_tools._get_backend() == "ddgs"
assert web_tools._get_backend() == "parallel"
def test_check_web_api_key_true_when_ddgs_configured(self, monkeypatch):
from tools import web_tools
+5 -2
View File
@@ -313,7 +313,9 @@ class TestCheckWebApiKey:
)
assert web_tools.check_web_api_key() is True
def test_no_credentials_fails(self, monkeypatch):
def test_no_credentials_usable_via_free_parallel(self, monkeypatch):
"""No credentials → check_web_api_key True: the keyless Parallel free MCP
services calls, so web is usable out of the box."""
from tools import web_tools
monkeypatch.setattr(web_tools, "_load_web_config", lambda: {})
monkeypatch.delenv("FIRECRAWL_API_KEY", raising=False)
@@ -324,7 +326,8 @@ class TestCheckWebApiKey:
monkeypatch.delenv("SEARXNG_URL", raising=False)
monkeypatch.setattr(web_tools, "_is_tool_gateway_ready", lambda: False)
monkeypatch.setattr(web_tools, "check_firecrawl_api_key", lambda: False)
assert web_tools.check_web_api_key() is False
monkeypatch.setattr(web_tools, "_ddgs_package_importable", lambda: False)
assert web_tools.check_web_api_key() is True
# ---------------------------------------------------------------------------
+86 -12
View File
@@ -384,11 +384,14 @@ class TestBackendSelection:
patch.dict(os.environ, {"FIRECRAWL_API_KEY": "fc-test"}):
assert _get_backend() == "firecrawl"
def test_fallback_no_keys_defaults_to_firecrawl(self):
"""No keys, no config → 'firecrawl' (will fail at client init)."""
def test_fallback_no_keys_defaults_to_parallel(self):
"""No credentials, no config → 'parallel' (free Search MCP works
keyless). Selection is purely credential-based."""
from tools.web_tools import _get_backend
with patch("tools.web_tools._load_web_config", return_value={}):
assert _get_backend() == "firecrawl"
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
patch("tools.web_tools._ddgs_package_importable", return_value=False):
assert _get_backend() == "parallel"
def test_invalid_config_falls_through_to_fallback(self):
"""web.backend=invalid → ignored, uses key-based fallback."""
@@ -623,9 +626,74 @@ class TestCheckWebApiKey:
from tools.web_tools import check_web_api_key
assert check_web_api_key() is True
def test_no_keys_returns_false(self):
def test_no_keys_usable_via_free_parallel(self):
"""No credentials → check_web_api_key True: selection resolves to the
keyless Parallel free MCP, which genuinely services calls (web works out
of the box). check_web_api_key is a usability probe, not a key check."""
from tools.web_tools import check_web_api_key
assert check_web_api_key() is False
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
patch("tools.web_tools._ddgs_package_importable", return_value=False), \
patch.dict(os.environ, {}, clear=False):
for k in ("PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL",
"TAVILY_API_KEY", "EXA_API_KEY", "SEARXNG_URL", "BRAVE_SEARCH_API_KEY"):
os.environ.pop(k, None)
assert check_web_api_key() is True
def test_typo_extract_backend_not_masked_by_parallel(self):
"""A typo'd per-capability backend is honored (so dispatch errors)
rather than silently falling through to keyless Parallel."""
from tools.web_tools import _get_extract_backend, check_web_api_key
with patch("tools.web_tools._load_web_config",
return_value={"extract_backend": "parrallel"}):
assert _get_extract_backend() == "parrallel" # not "parallel"
assert check_web_api_key() is False # unknown → unusable
def test_keyless_parallel_unusable_when_provider_disabled(self):
"""If the bundled web-parallel provider is disabled/unregistered, the
keyless free-MCP path must NOT report web as usable otherwise setup is
skipped but web tools fail at runtime with no provider."""
from tools.web_tools import check_web_api_key
with patch("tools.web_tools._load_web_config", return_value={}), \
patch("tools.web_tools._parallel_provider_registered", return_value=False), \
patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \
patch("tools.web_tools.check_firecrawl_api_key", return_value=False), \
patch("tools.web_tools._ddgs_package_importable", return_value=False), \
patch.dict(os.environ, {}, clear=False):
for var in (
"PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL",
"TAVILY_API_KEY", "EXA_API_KEY", "BRAVE_SEARCH_API_KEY", "SEARXNG_URL",
):
os.environ.pop(var, None)
assert check_web_api_key() is False
def test_extract_autodetect_skips_search_only_for_keyless_parallel(self):
"""A search-only env credential (SEARXNG_URL) must not shadow the keyless
Parallel free-MCP extract fallback: extract auto-detect skips search-only
backends, so _get_extract_backend resolves to parallel (which can fetch),
while search auto-detect still prefers the configured searxng."""
from tools.web_tools import _get_extract_backend, _get_search_backend
with patch("tools.web_tools._load_web_config", return_value={}), \
patch.dict(os.environ, {}, clear=False):
for var in (
"PARALLEL_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL",
"TAVILY_API_KEY", "EXA_API_KEY", "BRAVE_SEARCH_API_KEY",
):
os.environ.pop(var, None)
os.environ["SEARXNG_URL"] = "http://localhost:8080"
with patch("tools.web_tools._is_tool_gateway_ready", return_value=False):
assert _get_search_backend() == "searxng"
assert _get_extract_backend() == "parallel"
def test_configured_but_unavailable_backend_reports_unusable(self):
"""An explicitly configured backend with no creds (exa, no key) →
check_web_api_key False so diagnostics flag the misconfiguration
even though the tools stay registered."""
from tools.web_tools import check_web_api_key
with patch("tools.web_tools._load_web_config", return_value={"backend": "exa"}), \
patch.dict(os.environ, {}, clear=False):
os.environ.pop("EXA_API_KEY", None)
assert check_web_api_key() is False
def test_both_keys_returns_true(self):
with patch.dict(os.environ, {
@@ -688,12 +756,18 @@ class TestCheckWebApiKey:
assert refresh_calls == []
def test_configured_backend_must_match_available_provider(self):
with patch("tools.web_tools._load_web_config", return_value={"backend": "parallel"}):
with patch("tools.web_tools._read_nous_access_token", return_value="nous-token"):
with patch.dict(os.environ, {"FIRECRAWL_GATEWAY_URL": "http://127.0.0.1:3002"}, clear=False):
from tools.web_tools import check_web_api_key
assert check_web_api_key() is False
def test_web_tools_registered_even_when_configured_backend_unavailable(self):
# Registration is unconditional (web_tools_registered) so an explicitly
# configured but unavailable backend (exa without EXA_API_KEY) keeps the
# tools registered to surface exa's setup error at call time — while the
# readiness probe (check_web_api_key) honestly reports not-configured.
from tools.web_tools import web_tools_registered, check_web_api_key
assert web_tools_registered() is True
with patch("tools.web_tools._load_web_config", return_value={"backend": "exa"}), \
patch.dict(os.environ, {}, clear=False):
os.environ.pop("EXA_API_KEY", None)
assert web_tools_registered() is True
assert check_web_api_key() is False
def test_configured_firecrawl_backend_accepts_managed_gateway(self):
with patch("tools.web_tools._load_web_config", return_value={"backend": "firecrawl"}):