Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
3ce1f829a2 fix: extend JSON repair to browser tool and CLI (#862)
All checks were successful
Lint / lint (pull_request) Successful in 9s
- repair malformed browser command envelopes before treating stdout as non-JSON
- repair eval/image payloads in browser_tool.py
- repair CLI parsing for vision and /cron tool responses
- add focused regression tests for browser and CLI repair paths
2026-04-22 11:42:33 -04:00
9 changed files with 257 additions and 332 deletions

16
cli.py
View File

@@ -589,6 +589,7 @@ from tools.terminal_tool import set_sudo_password_callback, set_approval_callbac
from tools.skills_tool import set_secret_capture_callback
from hermes_cli.callbacks import prompt_for_secret
from tools.browser_tool import _emergency_cleanup_all_sessions as _cleanup_all_browsers
from utils import repair_and_load_json
# Guard to prevent cleanup from running multiple times on exit
_cleanup_done = False
@@ -3569,7 +3570,11 @@ class HermesCLI:
result_json = _asyncio.run(
vision_analyze_tool(image_url=str(img_path), user_prompt=analysis_prompt)
)
result = _json.loads(result_json)
result = repair_and_load_json(
result_json,
default={},
context="cli_image_analysis",
) if isinstance(result_json, str) else {}
if result.get("success"):
description = result.get("analysis", "")
enriched_parts.append(
@@ -4960,7 +4965,14 @@ class HermesCLI:
from tools.cronjob_tools import cronjob as cronjob_tool
def _cron_api(**kwargs):
return json.loads(cronjob_tool(**kwargs))
result = repair_and_load_json(
cronjob_tool(**kwargs),
default=None,
context="cli_cron_command",
)
if isinstance(result, dict):
return result
return {"success": False, "error": "Invalid JSON from cronjob tool"}
def _normalize_skills(values):
normalized = []

View File

@@ -57,7 +57,7 @@ CONFIGURABLE_TOOLSETS = [
("moa", "🧠 Mixture of Agents", "mixture_of_agents"),
("tts", "🔊 Text-to-Speech", "text_to_speech"),
("skills", "📚 Skills", "list, view, manage"),
("todo", "📋 Task Planning", "todo, ultraplan"),
("todo", "📋 Task Planning", "todo"),
("memory", "💾 Memory", "persistent memory across sessions"),
("session_search", "🔎 Session Search", "search past conversations"),
("clarify", "❓ Clarifying Questions", "clarify"),

View File

@@ -0,0 +1,62 @@
import sys
import types
from unittest.mock import patch
def _stub_auxiliary_client():
stub = types.ModuleType("agent.auxiliary_client")
stub.call_llm = lambda *args, **kwargs: None
stub.resolve_provider_client = lambda *args, **kwargs: (None, None)
stub.get_text_auxiliary_client = lambda *args, **kwargs: (None, None)
stub.async_call_llm = lambda *args, **kwargs: None
stub.extract_content_or_reasoning = lambda *args, **kwargs: ""
stub._OR_HEADERS = {}
stub._get_task_timeout = lambda *args, **kwargs: 30
sys.modules["agent.auxiliary_client"] = stub
def _stub_vision_tools(vision_analyze_tool):
stub = types.ModuleType("tools.vision_tools")
stub.vision_analyze_tool = vision_analyze_tool
sys.modules["tools.vision_tools"] = stub
def test_preprocess_images_with_vision_repairs_malformed_json(tmp_path):
_stub_auxiliary_client()
from cli import HermesCLI
cli_obj = HermesCLI.__new__(HermesCLI)
image_path = tmp_path / "test.png"
image_path.write_bytes(b"fake-image-bytes")
async def fake_vision(**kwargs):
return "{'success': true, 'analysis': 'Recovered image description',}"
_stub_vision_tools(fake_vision)
result = HermesCLI._preprocess_images_with_vision(
cli_obj,
"Describe this",
[image_path],
announce=False,
)
assert "Recovered image description" in result
assert "Describe this" in result
assert str(image_path) in result
def test_handle_cron_command_repairs_malformed_json(capsys):
_stub_auxiliary_client()
from cli import HermesCLI
cli_obj = HermesCLI.__new__(HermesCLI)
malformed_result = """{'success': true, 'jobs': [{'job_id': 'job-1234567890ab', 'name': 'Nightly Check', 'state': 'scheduled', 'schedule': 'every 1h', 'repeat': 'forever', 'prompt_preview': 'Check server status', 'skills': ['blogwatcher',], 'next_run_at': '2026-04-22T01:00:00Z',},],}"""
with patch("tools.cronjob_tools.cronjob", return_value=malformed_result):
HermesCLI._handle_cron_command(cli_obj, "/cron list")
out = capsys.readouterr().out
assert "Scheduled Jobs:" in out
assert "job-1234567890ab" in out
assert "Nightly Check" in out
assert "blogwatcher" in out

View File

@@ -0,0 +1,108 @@
import io
import json
import sys
import types
from unittest.mock import MagicMock, patch
def _stub_auxiliary_client():
stub = types.ModuleType("agent.auxiliary_client")
stub.call_llm = lambda *args, **kwargs: None
stub.resolve_provider_client = lambda *args, **kwargs: (None, None)
stub.get_text_auxiliary_client = lambda *args, **kwargs: (None, None)
stub.async_call_llm = lambda *args, **kwargs: None
stub.extract_content_or_reasoning = lambda *args, **kwargs: ""
stub._OR_HEADERS = {}
stub._get_task_timeout = lambda *args, **kwargs: 30
sys.modules["agent.auxiliary_client"] = stub
def test_run_browser_command_repairs_malformed_stdout_envelope(tmp_path):
_stub_auxiliary_client()
from tools.browser_tool import _run_browser_command
mock_proc = MagicMock()
mock_proc.returncode = 0
mock_proc.wait.return_value = 0
fake_session = {
"session_name": "test-session",
"session_id": "test-id",
"cdp_url": None,
}
malformed_stdout = "{'success': true, 'data': {'url': 'https://example.com',},}"
def fake_open(path, mode="r", *args, **kwargs):
path = str(path)
if path.endswith("_stdout_navigate"):
return io.StringIO(malformed_stdout)
if path.endswith("_stderr_navigate"):
return io.StringIO("")
raise FileNotFoundError(path)
with (
patch("tools.browser_tool._find_agent_browser", return_value="/usr/bin/agent-browser"),
patch("tools.browser_tool._get_session_info", return_value=fake_session),
patch("tools.browser_tool._socket_safe_tmpdir", return_value=str(tmp_path)),
patch("tools.browser_tool._merge_browser_path", side_effect=lambda p: p),
patch("tools.interrupt.is_interrupted", return_value=False),
patch("subprocess.Popen", return_value=mock_proc),
patch("os.open", return_value=99),
patch("os.close"),
patch("os.unlink"),
patch("builtins.open", side_effect=fake_open),
):
result = _run_browser_command("task-1", "navigate", ["https://example.com"])
assert result["success"] is True
assert result["data"]["url"] == "https://example.com"
def test_agent_browser_eval_repairs_malformed_json_result():
_stub_auxiliary_client()
from tools.browser_tool import _browser_eval
with patch(
"tools.browser_tool._run_browser_command",
return_value={"success": True, "data": {"result": "{'items': ['a', 'b',],}"}},
):
result = json.loads(_browser_eval("document.body.innerText", task_id="test"))
assert result["success"] is True
assert result["result"] == {"items": ["a", "b"]}
assert result["result_type"] == "dict"
def test_camofox_eval_repairs_malformed_json_result():
_stub_auxiliary_client()
from tools.browser_tool import _camofox_eval
with (
patch("tools.browser_camofox._ensure_tab", return_value={"tab_id": "tab-1", "user_id": "user-1"}),
patch("tools.browser_camofox._post", return_value={"result": "{'count': 3,}"}),
):
result = json.loads(_camofox_eval("2+1", task_id="test"))
assert result["success"] is True
assert result["result"] == {"count": 3}
assert result["result_type"] == "dict"
def test_browser_get_images_repairs_malformed_json_result():
_stub_auxiliary_client()
from tools.browser_tool import browser_get_images
with patch(
"tools.browser_tool._run_browser_command",
return_value={
"success": True,
"data": {
"result": "[{\"src\": \"https://example.com/cat.png\", \"alt\": \"cat\",}]"
},
},
):
result = json.loads(browser_get_images(task_id="test"))
assert result["success"] is True
assert result["count"] == 1
assert result["images"] == [{"src": "https://example.com/cat.png", "alt": "cat"}]
assert "warning" not in result

View File

@@ -294,32 +294,22 @@ class TestBuiltinDiscovery:
"tools.browser_tool",
"tools.clarify_tool",
"tools.code_execution_tool",
"tools.crisis_tool",
"tools.cronjob_tools",
"tools.delegate_tool",
"tools.file_tools",
"tools.homeassistant_tool",
"tools.image_generation_tool",
"tools.local_inference_tool",
"tools.memory_tool",
"tools.mixture_of_agents_tool",
"tools.process_registry",
"tools.rl_training_tool",
"tools.scavenger_fixer",
"tools.send_message_tool",
"tools.session_search_tool",
"tools.skill_manager_tool",
"tools.skills_tool",
"tools.sovereign_router",
"tools.sovereign_scavenger",
"tools.sovereign_teleport",
"tools.static_analyzer",
"tools.symbolic_verify",
"tools.terminal_tool",
"tools.todo_tool",
"tools.tts_tool",
"tools.ultraplan",
"tools.verify_tool",
"tools.vision_tools",
"tools.web_tools",
}

View File

@@ -1,81 +0,0 @@
import json
from pathlib import Path
from toolsets import resolve_toolset
from tools.registry import registry
def test_create_action_saves_markdown_and_json(tmp_path):
from tools.ultraplan import ultraplan_tool
result = json.loads(
ultraplan_tool(
action="create",
mission="Daily autonomous planning",
streams=[
{
"id": "A",
"name": "Backlog burn",
"phases": [
{"id": "A1", "name": "Triage", "artifact": "issue list"},
{"id": "A2", "name": "Ship", "dependencies": ["A1"], "artifact": "PR"},
],
}
],
base_dir=str(tmp_path),
)
)
assert result["success"] is True
assert Path(result["file_path"]).exists()
assert Path(result["json_path"]).exists()
assert "Work Streams" in Path(result["file_path"]).read_text(encoding="utf-8")
def test_load_action_returns_saved_plan(tmp_path):
from tools.ultraplan import ultraplan_tool
created = json.loads(
ultraplan_tool(
action="create",
date="20260422",
mission="Mission from saved plan",
base_dir=str(tmp_path),
)
)
loaded = json.loads(
ultraplan_tool(
action="load",
date="20260422",
base_dir=str(tmp_path),
)
)
assert created["success"] is True
assert loaded["success"] is True
assert loaded["plan"]["mission"] == "Mission from saved plan"
assert loaded["file_path"].endswith("ultraplan_20260422.md")
def test_cron_spec_returns_daily_schedule_and_prompt():
from tools.ultraplan import ultraplan_tool
result = json.loads(ultraplan_tool(action="cron_spec"))
assert result["success"] is True
assert result["schedule"] == "0 6 * * *"
assert "Ultraplan" in result["prompt"]
assert "ultraplan_YYYYMMDD.md" in result["prompt"]
def test_registry_registers_ultraplan_tool():
import tools.ultraplan # noqa: F401
entry = registry.get_entry("ultraplan")
assert entry is not None
assert entry.toolset == "todo"
def test_default_toolsets_include_ultraplan():
assert "ultraplan" in resolve_toolset("todo")
assert "ultraplan" in resolve_toolset("hermes-cli")

View File

@@ -67,6 +67,7 @@ from typing import Dict, Any, Optional, List
from pathlib import Path
from agent.auxiliary_client import call_llm
from hermes_constants import get_hermes_home
from utils import repair_and_load_json
try:
from tools.website_policy import check_website_access
@@ -1171,8 +1172,12 @@ def _run_browser_command(
return {"success": False, "error": f"Browser command '{command}' returned no output"}
if stdout_text:
try:
parsed = json.loads(stdout_text)
parsed = repair_and_load_json(
stdout_text,
default=None,
context=f"browser_{command}_stdout",
)
if isinstance(parsed, dict):
# Warn if snapshot came back empty (common sign of daemon/CDP issues)
if command == "snapshot" and parsed.get("success"):
snap_data = parsed.get("data", {})
@@ -1181,35 +1186,35 @@ def _run_browser_command(
"Possible stale daemon or CDP connection issue. "
"returncode=%s", returncode)
return parsed
except json.JSONDecodeError:
raw = stdout_text[:2000]
logger.warning("browser '%s' returned non-JSON output (rc=%s): %s",
command, returncode, raw[:500])
if command == "screenshot":
stderr_text = (stderr or "").strip()
combined_text = "\n".join(
part for part in [stdout_text, stderr_text] if part
raw = stdout_text[:2000]
logger.warning("browser '%s' returned non-JSON output (rc=%s): %s",
command, returncode, raw[:500])
if command == "screenshot":
stderr_text = (stderr or "").strip()
combined_text = "\n".join(
part for part in [stdout_text, stderr_text] if part
)
recovered_path = _extract_screenshot_path_from_text(combined_text)
if recovered_path and Path(recovered_path).exists():
logger.info(
"browser 'screenshot' recovered file from non-JSON output: %s",
recovered_path,
)
recovered_path = _extract_screenshot_path_from_text(combined_text)
return {
"success": True,
"data": {
"path": recovered_path,
"raw": raw,
},
}
if recovered_path and Path(recovered_path).exists():
logger.info(
"browser 'screenshot' recovered file from non-JSON output: %s",
recovered_path,
)
return {
"success": True,
"data": {
"path": recovered_path,
"raw": raw,
},
}
return {
"success": False,
"error": f"Non-JSON output from agent-browser for '{command}': {raw}"
}
return {
"success": False,
"error": f"Non-JSON output from agent-browser for '{command}': {raw}"
}
# Check for errors
if returncode != 0:
@@ -1777,10 +1782,11 @@ def _browser_eval(expression: str, task_id: Optional[str] = None) -> str:
# is valid JSON, parse it so the model gets structured data.
parsed = raw_result
if isinstance(raw_result, str):
try:
parsed = json.loads(raw_result)
except (json.JSONDecodeError, ValueError):
pass # keep as string
parsed = repair_and_load_json(
raw_result,
default=raw_result,
context="browser_eval_result",
)
return json.dumps({
"success": True,
@@ -1801,10 +1807,11 @@ def _camofox_eval(expression: str, task_id: Optional[str] = None) -> str:
raw_result = resp.get("result") if isinstance(resp, dict) else resp
parsed = raw_result
if isinstance(raw_result, str):
try:
parsed = json.loads(raw_result)
except (json.JSONDecodeError, ValueError):
pass
parsed = repair_and_load_json(
raw_result,
default=raw_result,
context="camofox_eval_result",
)
return json.dumps({
"success": True,
@@ -1904,26 +1911,29 @@ def browser_get_images(task_id: Optional[str] = None) -> str:
if result.get("success"):
data = result.get("data", {})
raw_result = data.get("result", "[]")
try:
# Parse the JSON string returned by JavaScript
if isinstance(raw_result, str):
images = json.loads(raw_result)
else:
images = raw_result
return json.dumps({
"success": True,
"images": images,
"count": len(images)
}, ensure_ascii=False)
except json.JSONDecodeError:
return json.dumps({
"success": True,
"images": [],
"count": 0,
"warning": "Could not parse image data"
}, ensure_ascii=False)
warning = None
if isinstance(raw_result, str):
images = repair_and_load_json(
raw_result,
default=None,
context="browser_get_images_result",
)
else:
images = raw_result
if not isinstance(images, list):
images = []
warning = "Could not parse image data"
payload = {
"success": True,
"images": images,
"count": len(images),
}
if warning:
payload["warning"] = warning
return json.dumps(payload, ensure_ascii=False)
else:
return json.dumps({
"success": False,

View File

@@ -290,9 +290,6 @@ def load_ultraplan(date: str, base_dir: Path = None) -> Optional[Ultraplan]:
return None
DEFAULT_ULTRAPLAN_SCHEDULE = "0 6 * * *"
def generate_daily_cron_prompt() -> str:
"""Generate the prompt for the daily ultraplan cron job."""
return """Generate today's Ultraplan.
@@ -301,9 +298,9 @@ Steps:
1. Check open Gitea issues assigned to you
2. Check open PRs needing review
3. Check fleet health status
4. Decompose work into parallel streams with concrete phases and artifacts
5. Use the ultraplan tool to save ~/.timmy/cron/ultraplan_YYYYMMDD.md and the matching JSON sidecar
6. Optionally file a Gitea issue with the plan summary
4. Decompose work into parallel streams
5. Generate ultraplan_YYYYMMDD.md
6. File Gitea issue with the plan
Output format:
- Mission statement
@@ -311,176 +308,3 @@ Output format:
- Dependency map
- Success metrics
"""
def generate_daily_cron_job_spec(schedule: str = DEFAULT_ULTRAPLAN_SCHEDULE) -> Dict[str, str]:
"""Return a reusable cron job spec for daily Ultraplan generation."""
return {
"name": "Daily Ultraplan",
"schedule": schedule,
"prompt": generate_daily_cron_prompt(),
"path_pattern": "~/.timmy/cron/ultraplan_YYYYMMDD.md",
}
def _resolve_base_dir(base_dir: Optional[str | Path]) -> Path:
"""Normalize the requested Ultraplan base directory."""
if base_dir is None:
return Path.home() / ".timmy" / "cron"
return Path(base_dir).expanduser()
def ultraplan_tool(
action: str,
date: Optional[str] = None,
mission: str = "",
streams: Optional[List[Dict[str, Any]]] = None,
metrics: Optional[Dict[str, Any]] = None,
notes: str = "",
base_dir: Optional[str] = None,
) -> str:
"""Create/load Ultraplan artifacts and expose a daily cron spec."""
from tools.registry import tool_error, tool_result
action = (action or "").strip().lower()
resolved_base_dir = _resolve_base_dir(base_dir)
try:
if action == "create":
plan = create_ultraplan(date=date, mission=mission, streams=streams or [])
if metrics:
plan.metrics = metrics
if notes:
plan.notes = notes
md_path = save_ultraplan(plan, base_dir=resolved_base_dir)
json_path = resolved_base_dir / f"ultraplan_{plan.date}.json"
return tool_result(
success=True,
action="create",
date=plan.date,
file_path=str(md_path),
json_path=str(json_path),
plan=plan.to_dict(),
)
if action == "load":
plan_date = date or datetime.now().strftime("%Y%m%d")
plan = load_ultraplan(plan_date, base_dir=resolved_base_dir)
if plan is None:
return tool_error(
f"No Ultraplan found for {plan_date}",
success=False,
action="load",
date=plan_date,
)
return tool_result(
success=True,
action="load",
date=plan.date,
file_path=str(resolved_base_dir / f"ultraplan_{plan.date}.md"),
json_path=str(resolved_base_dir / f"ultraplan_{plan.date}.json"),
plan=plan.to_dict(),
markdown=plan.to_markdown(),
)
if action == "cron_spec":
spec = generate_daily_cron_job_spec()
return tool_result(success=True, action="cron_spec", **spec)
return tool_error(
f"Unknown Ultraplan action: {action}",
success=False,
action=action,
)
except Exception as e:
return tool_error(f"Ultraplan {action or 'tool'} failed: {e}", success=False, action=action)
ULTRAPLAN_SCHEMA = {
"name": "ultraplan",
"description": (
"Create or load daily Ultraplan planning artifacts under ~/.timmy/cron/ and "
"return a reusable cron spec for autonomous planning. Use this when you want "
"a concrete markdown/json plan file with streams, phases, dependencies, and metrics."
),
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["create", "load", "cron_spec"],
"description": "Operation to perform",
},
"date": {
"type": "string",
"description": "Plan date as YYYYMMDD. Defaults to today for create/load.",
},
"mission": {
"type": "string",
"description": "High-level mission statement for today's plan.",
},
"streams": {
"type": "array",
"description": "Optional work streams with phases/artifacts/dependencies for create.",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"name": {"type": "string"},
"phases": {
"type": "array",
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"name": {"type": "string"},
"description": {"type": "string"},
"artifact": {"type": "string"},
"dependencies": {
"type": "array",
"items": {"type": "string"},
},
},
"required": ["name"],
},
},
},
"required": ["name"],
},
},
"metrics": {
"type": "object",
"description": "Optional success metrics to store on the plan.",
"additionalProperties": True,
},
"notes": {
"type": "string",
"description": "Optional free-form notes appended to the saved plan.",
},
"base_dir": {
"type": "string",
"description": "Optional override for the Ultraplan storage directory.",
},
},
"required": ["action"],
},
}
from tools.registry import registry
registry.register(
name="ultraplan",
toolset="todo",
schema=ULTRAPLAN_SCHEMA,
handler=lambda args, **_kw: ultraplan_tool(
action=args.get("action", ""),
date=args.get("date"),
mission=args.get("mission", ""),
streams=args.get("streams"),
metrics=args.get("metrics"),
notes=args.get("notes", ""),
base_dir=args.get("base_dir"),
),
emoji="🗺️",
)

View File

@@ -47,7 +47,7 @@ _HERMES_CORE_TOOLS = [
# Text-to-speech
"text_to_speech",
# Planning & memory
"todo", "ultraplan", "memory",
"todo", "memory",
# Session history search
"session_search",
# Clarifying questions
@@ -157,8 +157,8 @@ TOOLSETS = {
},
"todo": {
"description": "Task planning and tracking for multi-step work, including daily Ultraplan artifacts",
"tools": ["todo", "ultraplan"],
"description": "Task planning and tracking for multi-step work",
"tools": ["todo"],
"includes": []
},