Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
a581d03a2b | ||
|
|
69b30152b4 |
@@ -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"),
|
||||
("todo", "📋 Task Planning", "todo, ultraplan"),
|
||||
("memory", "💾 Memory", "persistent memory across sessions"),
|
||||
("session_search", "🔎 Session Search", "search past conversations"),
|
||||
("clarify", "❓ Clarifying Questions", "clarify"),
|
||||
|
||||
@@ -148,184 +148,3 @@ class TestStrategyNameSurfaced:
|
||||
assert count == 0
|
||||
assert strategy is None
|
||||
assert err is not None
|
||||
|
||||
|
||||
class TestEscapeDriftGuard:
|
||||
"""Tests for the escape-drift guard that catches bash/JSON serialization
|
||||
artifacts where an apostrophe gets prefixed with a spurious backslash
|
||||
in tool-call transport.
|
||||
"""
|
||||
|
||||
def test_drift_blocked_apostrophe(self):
|
||||
"""File has ', old_string and new_string both have \\' — classic
|
||||
tool-call drift. Guard must block with a helpful error instead of
|
||||
writing \\' literals into source code."""
|
||||
content = "x = \"hello there\"\n"
|
||||
# Simulate transport-corrupted old_string and new_string where an
|
||||
# apostrophe-like context got prefixed with a backslash. The content
|
||||
# itself has no apostrophe, but both strings do — matching via
|
||||
# whitespace/anchor strategies would otherwise succeed.
|
||||
old_string = "x = \"hello there\" # don\\'t edit\n"
|
||||
new_string = "x = \"hi there\" # don\\'t edit\n"
|
||||
# This particular pair won't match anything, so it exits via
|
||||
# no-match path. Build a case where a non-exact strategy DOES match.
|
||||
content = "line\n x = 1\nline"
|
||||
old_string = "line\n x = \\'a\\'\nline"
|
||||
new_string = "line\n x = \\'b\\'\nline"
|
||||
new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
|
||||
assert count == 0
|
||||
assert err is not None and "Escape-drift" in err
|
||||
assert "backslash" in err.lower()
|
||||
assert new == content # file untouched
|
||||
|
||||
def test_drift_blocked_double_quote(self):
|
||||
"""Same idea but with \\" drift instead of \\'."""
|
||||
content = 'line\n x = 1\nline'
|
||||
old_string = 'line\n x = \\"a\\"\nline'
|
||||
new_string = 'line\n x = \\"b\\"\nline'
|
||||
new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
|
||||
assert count == 0
|
||||
assert err is not None and "Escape-drift" in err
|
||||
|
||||
def test_drift_allowed_when_file_genuinely_has_backslash_escapes(self):
|
||||
"""If the file already contains \\' (e.g. inside an existing escaped
|
||||
string), the model is legitimately preserving it. Guard must NOT
|
||||
fire."""
|
||||
content = "line\n x = \\'a\\'\nline"
|
||||
old_string = "line\n x = \\'a\\'\nline"
|
||||
new_string = "line\n x = \\'b\\'\nline"
|
||||
new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
|
||||
assert err is None
|
||||
assert count == 1
|
||||
assert "\\'b\\'" in new
|
||||
|
||||
def test_drift_allowed_on_exact_match(self):
|
||||
"""Exact matches bypass the drift guard entirely — if the file
|
||||
really contains the exact bytes old_string specified, it's not
|
||||
drift."""
|
||||
content = "hello \\'world\\'"
|
||||
new, count, strategy, err = fuzzy_find_and_replace(
|
||||
content, "hello \\'world\\'", "hello \\'there\\'"
|
||||
)
|
||||
assert err is None
|
||||
assert count == 1
|
||||
assert strategy == "exact"
|
||||
|
||||
def test_drift_allowed_when_adding_escaped_strings(self):
|
||||
"""Model is adding new content with \\' that wasn't in the original.
|
||||
old_string has no \\', so guard doesn't fire."""
|
||||
content = "line1\nline2\nline3"
|
||||
old_string = "line1\nline2\nline3"
|
||||
new_string = "line1\nprint(\\'added\\')\nline2\nline3"
|
||||
new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
|
||||
assert err is None
|
||||
assert count == 1
|
||||
assert "\\'added\\'" in new
|
||||
|
||||
def test_no_drift_check_when_new_string_lacks_suspect_chars(self):
|
||||
"""Fast-path: if new_string has no \\' or \\", guard must not
|
||||
fire even on fuzzy match."""
|
||||
content = "def foo():\n pass" # extra space ignored by line_trimmed
|
||||
old_string = "def foo():\n pass"
|
||||
new_string = "def bar():\n return 1"
|
||||
new, count, strategy, err = fuzzy_find_and_replace(content, old_string, new_string)
|
||||
assert err is None
|
||||
assert count == 1
|
||||
|
||||
|
||||
class TestFindClosestLines:
|
||||
def setup_method(self):
|
||||
from tools.fuzzy_match import find_closest_lines
|
||||
self.find_closest_lines = find_closest_lines
|
||||
|
||||
def test_finds_similar_line(self):
|
||||
content = "def foo():\n pass\ndef bar():\n return 1\n"
|
||||
result = self.find_closest_lines("def baz():", content)
|
||||
assert "def foo" in result or "def bar" in result
|
||||
|
||||
def test_returns_empty_for_no_match(self):
|
||||
content = "completely different content here"
|
||||
result = self.find_closest_lines("xyzzy_no_match_possible_!!!", content)
|
||||
assert result == ""
|
||||
|
||||
def test_returns_empty_for_empty_inputs(self):
|
||||
assert self.find_closest_lines("", "some content") == ""
|
||||
assert self.find_closest_lines("old string", "") == ""
|
||||
|
||||
def test_includes_context_lines(self):
|
||||
content = "line1\nline2\ndef target():\n pass\nline5\n"
|
||||
result = self.find_closest_lines("def target():", content)
|
||||
assert "target" in result
|
||||
|
||||
def test_includes_line_numbers(self):
|
||||
content = "line1\nline2\ndef foo():\n pass\n"
|
||||
result = self.find_closest_lines("def foo():", content)
|
||||
# Should include line numbers in format "N| content"
|
||||
assert "|" in result
|
||||
|
||||
|
||||
class TestFormatNoMatchHint:
|
||||
"""Gating tests for format_no_match_hint — the shared helper that decides
|
||||
whether a 'Did you mean?' snippet should be appended to an error.
|
||||
"""
|
||||
|
||||
def setup_method(self):
|
||||
from tools.fuzzy_match import format_no_match_hint
|
||||
self.fmt = format_no_match_hint
|
||||
|
||||
def test_fires_on_could_not_find_with_match(self):
|
||||
"""Classic no-match: similar content exists → hint fires."""
|
||||
content = "def foo():\n pass\ndef bar():\n pass\n"
|
||||
result = self.fmt(
|
||||
"Could not find a match for old_string in the file",
|
||||
0, "def baz():", content,
|
||||
)
|
||||
assert "Did you mean" in result
|
||||
assert "foo" in result or "bar" in result
|
||||
|
||||
def test_silent_on_ambiguous_match_error(self):
|
||||
"""'Found N matches' is not a missing-match failure — no hint."""
|
||||
content = "aaa bbb aaa\n"
|
||||
result = self.fmt(
|
||||
"Found 2 matches for old_string. Provide more context to make it unique, or use replace_all=True.",
|
||||
0, "aaa", content,
|
||||
)
|
||||
assert result == ""
|
||||
|
||||
def test_silent_on_escape_drift_error(self):
|
||||
"""Escape-drift errors are intentional blocks — hint would mislead."""
|
||||
content = "x = 1\n"
|
||||
result = self.fmt(
|
||||
"Escape-drift detected: old_string and new_string contain the literal sequence '\\\\''...",
|
||||
0, "x = \\'1\\'", content,
|
||||
)
|
||||
assert result == ""
|
||||
|
||||
def test_silent_on_identical_strings(self):
|
||||
"""old_string == new_string — hint irrelevant."""
|
||||
result = self.fmt(
|
||||
"old_string and new_string are identical",
|
||||
0, "foo", "foo bar\n",
|
||||
)
|
||||
assert result == ""
|
||||
|
||||
def test_silent_when_match_count_nonzero(self):
|
||||
"""If match succeeded, we shouldn't be in the error path — defense in depth."""
|
||||
result = self.fmt(
|
||||
"Could not find a match for old_string in the file",
|
||||
1, "foo", "foo bar\n",
|
||||
)
|
||||
assert result == ""
|
||||
|
||||
def test_silent_on_none_error(self):
|
||||
"""No error at all — no hint."""
|
||||
result = self.fmt(None, 0, "foo", "bar\n")
|
||||
assert result == ""
|
||||
|
||||
def test_silent_when_no_similar_content(self):
|
||||
"""Even for a valid no-match error, skip hint when nothing similar exists."""
|
||||
result = self.fmt(
|
||||
"Could not find a match for old_string in the file",
|
||||
0, "totally_unique_xyzzy_qux", "abc\nxyz\n",
|
||||
)
|
||||
assert result == ""
|
||||
|
||||
@@ -1,114 +0,0 @@
|
||||
import json
|
||||
import os
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import tools.skill_manager_tool as skill_manager_tool
|
||||
from tools.file_tools import patch_tool
|
||||
from tools.skill_manager_tool import _create_skill, _patch_skill
|
||||
|
||||
|
||||
def _disable_patch_tool_guards(monkeypatch):
|
||||
monkeypatch.setattr("tools.file_tools._check_sensitive_path", lambda _path: None)
|
||||
monkeypatch.setattr("tools.file_tools._check_file_staleness", lambda _path, _task_id: None)
|
||||
monkeypatch.setattr("tools.file_tools._log_and_check_conflict", lambda _path, _task_id, _action: None)
|
||||
|
||||
|
||||
def test_patch_tool_replace_no_match_shows_rich_hint_without_legacy_hint(tmp_path, monkeypatch):
|
||||
_disable_patch_tool_guards(monkeypatch)
|
||||
sample = tmp_path / "sample.py"
|
||||
sample.write_text("def foo():\n return 1\n\ndef bar():\n return 2\n", encoding="utf-8")
|
||||
|
||||
raw = patch_tool(
|
||||
mode="replace",
|
||||
path=str(sample),
|
||||
old_string="def barycentric():",
|
||||
new_string="def barycentric_new():",
|
||||
task_id="qa960-replace-rich-hint",
|
||||
)
|
||||
|
||||
result = json.loads(raw)
|
||||
assert result["success"] is False
|
||||
assert "Could not find a match" in result["error"]
|
||||
assert "Did you mean one of these sections?" in result["error"]
|
||||
assert "def bar():" in result["error"] or "def foo():" in result["error"]
|
||||
assert "[Hint:" not in raw
|
||||
|
||||
|
||||
def test_patch_tool_replace_ambiguous_error_does_not_show_did_you_mean(tmp_path, monkeypatch):
|
||||
_disable_patch_tool_guards(monkeypatch)
|
||||
sample = tmp_path / "sample.py"
|
||||
sample.write_text("aaa\nbbb\naaa\n", encoding="utf-8")
|
||||
|
||||
raw = patch_tool(
|
||||
mode="replace",
|
||||
path=str(sample),
|
||||
old_string="aaa",
|
||||
new_string="ccc",
|
||||
task_id="qa960-replace-ambiguous",
|
||||
)
|
||||
|
||||
result = json.loads(raw)
|
||||
assert result["success"] is False
|
||||
assert "Found 2 matches" in result["error"]
|
||||
assert "Did you mean one of these sections?" not in result["error"]
|
||||
assert "[Hint:" not in raw
|
||||
|
||||
|
||||
def test_patch_tool_v4a_no_match_shows_rich_hint(tmp_path, monkeypatch):
|
||||
_disable_patch_tool_guards(monkeypatch)
|
||||
sample = tmp_path / "sample.py"
|
||||
sample.write_text("def foo():\n return 1\n", encoding="utf-8")
|
||||
|
||||
patch = textwrap.dedent(
|
||||
f"""\
|
||||
*** Begin Patch
|
||||
*** Update File: {sample}
|
||||
@@
|
||||
-def barycentric():
|
||||
+def barycentric_new():
|
||||
*** End Patch
|
||||
"""
|
||||
)
|
||||
|
||||
raw = patch_tool(mode="patch", patch=patch, task_id="qa960-v4a-rich-hint")
|
||||
result = json.loads(raw)
|
||||
assert result["success"] is False
|
||||
assert "Patch validation failed" in result["error"]
|
||||
assert "Did you mean one of these sections?" in result["error"]
|
||||
assert "def foo():" in result["error"]
|
||||
|
||||
|
||||
def test_skill_patch_no_match_shows_rich_hint(tmp_path, monkeypatch):
|
||||
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
|
||||
skills_dir = tmp_path / "skills"
|
||||
skills_dir.mkdir(parents=True, exist_ok=True)
|
||||
monkeypatch.setattr(skill_manager_tool, "SKILLS_DIR", skills_dir)
|
||||
monkeypatch.setattr(skill_manager_tool, "_security_scan_skill", lambda _skill_dir: None)
|
||||
|
||||
_create_skill(
|
||||
"qa-skill",
|
||||
textwrap.dedent(
|
||||
"""\
|
||||
---
|
||||
name: qa-skill
|
||||
description: test
|
||||
---
|
||||
|
||||
Step 1: Do the thing.
|
||||
Step 2: Verify the thing.
|
||||
"""
|
||||
),
|
||||
)
|
||||
|
||||
result = _patch_skill(
|
||||
"qa-skill",
|
||||
"Step 1: Do the production rollout.",
|
||||
"Step 1: Updated.",
|
||||
)
|
||||
|
||||
assert result["success"] is False
|
||||
assert "Could not find a match" in result["error"]
|
||||
assert "Did you mean one of these sections?" in result["error"]
|
||||
assert "Step 1: Do the thing." in result["error"]
|
||||
assert "file_preview" in result
|
||||
@@ -294,22 +294,32 @@ 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",
|
||||
}
|
||||
|
||||
81
tests/tools/test_ultraplan_tool.py
Normal file
81
tests/tools/test_ultraplan_tool.py
Normal file
@@ -0,0 +1,81 @@
|
||||
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")
|
||||
@@ -757,14 +757,12 @@ class ShellFileOperations(FileOperations):
|
||||
content, old_string, new_string, replace_all
|
||||
)
|
||||
|
||||
if error or match_count == 0:
|
||||
err_msg = error or f"Could not find match for old_string in {path}"
|
||||
try:
|
||||
from tools.fuzzy_match import format_no_match_hint
|
||||
err_msg += format_no_match_hint(err_msg, match_count, old_string, content)
|
||||
except Exception:
|
||||
pass
|
||||
return PatchResult(error=err_msg)
|
||||
if error:
|
||||
return PatchResult(error=error)
|
||||
|
||||
if match_count == 0:
|
||||
return PatchResult(error=f"Could not find match for old_string in {path}")
|
||||
|
||||
# Write back
|
||||
write_result = self.write_file(path, new_content)
|
||||
if write_result.error:
|
||||
|
||||
@@ -8,7 +8,6 @@ import os
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, Optional
|
||||
from tools.binary_extensions import has_binary_extension
|
||||
from tools.file_operations import ShellFileOperations
|
||||
from agent.redact import redact_sensitive_text
|
||||
@@ -691,11 +690,8 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
|
||||
result_json = json.dumps(result_dict, ensure_ascii=False)
|
||||
# Hint when old_string not found — saves iterations where the agent
|
||||
# retries with stale content instead of re-reading the file.
|
||||
# Suppressed when patch_replace already attached a rich "Did you mean?"
|
||||
# snippet (which is strictly more useful than the generic hint).
|
||||
if result_dict.get("error") and "Could not find" in str(result_dict["error"]):
|
||||
if "Did you mean one of these sections?" not in str(result_dict["error"]):
|
||||
result_json += "\n\n[Hint: old_string not found. Use read_file to verify the current content, or search_files to locate the text.]"
|
||||
result_json += "\n\n[Hint: old_string not found. Use read_file to verify the current content, or search_files to locate the text.]"
|
||||
return result_json
|
||||
except Exception as e:
|
||||
return tool_error(str(e))
|
||||
|
||||
@@ -93,21 +93,6 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
|
||||
f"Provide more context to make it unique, or use replace_all=True."
|
||||
)
|
||||
|
||||
# Escape-drift guard: when the matched strategy is NOT `exact`,
|
||||
# we matched via some form of normalization. If new_string
|
||||
# contains shell/JSON-style escape sequences (\\' or \\") that
|
||||
# would be written literally into the file but the matched
|
||||
# region of the file has no such sequences, this is almost
|
||||
# certainly tool-call serialization drift — the model typed
|
||||
# an apostrophe/quote and the transport added a stray
|
||||
# backslash. Writing new_string as-is would corrupt the file.
|
||||
# Block with a helpful error so the model re-reads and retries
|
||||
# instead of the caller silently persisting garbage (or not).
|
||||
if strategy_name != "exact":
|
||||
drift_err = _detect_escape_drift(content, matches, old_string, new_string)
|
||||
if drift_err:
|
||||
return content, 0, None, drift_err
|
||||
|
||||
# Perform replacement
|
||||
new_content = _apply_replacements(content, matches, new_string)
|
||||
return new_content, len(matches), strategy_name, None
|
||||
@@ -116,46 +101,6 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
|
||||
return content, 0, None, "Could not find a match for old_string in the file"
|
||||
|
||||
|
||||
def _detect_escape_drift(content: str, matches: List[Tuple[int, int]],
|
||||
old_string: str, new_string: str) -> Optional[str]:
|
||||
"""Detect tool-call escape-drift artifacts in new_string.
|
||||
|
||||
Looks for ``\\'`` or ``\\"`` sequences that are present in both
|
||||
old_string and new_string (i.e. the model copy-pasted them as "context"
|
||||
it intended to preserve) but don't exist in the matched region of the
|
||||
file. That pattern indicates the transport layer inserted spurious
|
||||
shell-style escapes around apostrophes or quotes — writing new_string
|
||||
verbatim would literally insert ``\\'`` into source code.
|
||||
|
||||
Returns an error string if drift is detected, None otherwise.
|
||||
"""
|
||||
# Cheap pre-check: bail out unless new_string actually contains a
|
||||
# suspect escape sequence. This keeps the guard free for all the
|
||||
# common, correct cases.
|
||||
if "\\'" not in new_string and '\\"' not in new_string:
|
||||
return None
|
||||
|
||||
# Aggregate matched regions of the file — that's what new_string will
|
||||
# replace. If the suspect escapes are present there already, the
|
||||
# model is genuinely preserving them (valid for some languages /
|
||||
# escaped strings); accept the patch.
|
||||
matched_regions = "".join(content[start:end] for start, end in matches)
|
||||
|
||||
for suspect in ("\\'", '\\"'):
|
||||
if suspect in new_string and suspect in old_string and suspect not in matched_regions:
|
||||
plain = suspect[1] # "'" or '"'
|
||||
return (
|
||||
f"Escape-drift detected: old_string and new_string contain "
|
||||
f"the literal sequence {suspect!r} but the matched region of "
|
||||
f"the file does not. This is almost always a tool-call "
|
||||
f"serialization artifact where an apostrophe or quote got "
|
||||
f"prefixed with a spurious backslash. Re-read the file with "
|
||||
f"read_file and pass old_string/new_string without "
|
||||
f"backslash-escaping {plain!r} characters."
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def _apply_replacements(content: str, matches: List[Tuple[int, int]], new_string: str) -> str:
|
||||
"""
|
||||
Apply replacements at the given positions.
|
||||
@@ -619,86 +564,3 @@ def _map_normalized_positions(original: str, normalized: str,
|
||||
original_matches.append((orig_start, min(orig_end, len(original))))
|
||||
|
||||
return original_matches
|
||||
|
||||
|
||||
def find_closest_lines(old_string: str, content: str, context_lines: int = 2, max_results: int = 3) -> str:
|
||||
"""Find lines in content most similar to old_string for "did you mean?" feedback.
|
||||
|
||||
Returns a formatted string showing the closest matching lines with context,
|
||||
or empty string if no useful match is found.
|
||||
"""
|
||||
if not old_string or not content:
|
||||
return ""
|
||||
|
||||
old_lines = old_string.splitlines()
|
||||
content_lines = content.splitlines()
|
||||
|
||||
if not old_lines or not content_lines:
|
||||
return ""
|
||||
|
||||
# Use first line of old_string as anchor for search
|
||||
anchor = old_lines[0].strip()
|
||||
if not anchor:
|
||||
# Try second line if first is blank
|
||||
candidates = [l.strip() for l in old_lines if l.strip()]
|
||||
if not candidates:
|
||||
return ""
|
||||
anchor = candidates[0]
|
||||
|
||||
# Score each line in content by similarity to anchor
|
||||
scored = []
|
||||
for i, line in enumerate(content_lines):
|
||||
stripped = line.strip()
|
||||
if not stripped:
|
||||
continue
|
||||
ratio = SequenceMatcher(None, anchor, stripped).ratio()
|
||||
if ratio > 0.3:
|
||||
scored.append((ratio, i))
|
||||
|
||||
if not scored:
|
||||
return ""
|
||||
|
||||
# Take top matches
|
||||
scored.sort(key=lambda x: -x[0])
|
||||
top = scored[:max_results]
|
||||
|
||||
parts = []
|
||||
seen_ranges = set()
|
||||
for _, line_idx in top:
|
||||
start = max(0, line_idx - context_lines)
|
||||
end = min(len(content_lines), line_idx + len(old_lines) + context_lines)
|
||||
key = (start, end)
|
||||
if key in seen_ranges:
|
||||
continue
|
||||
seen_ranges.add(key)
|
||||
snippet = "\n".join(
|
||||
f"{start + j + 1:4d}| {content_lines[start + j]}"
|
||||
for j in range(end - start)
|
||||
)
|
||||
parts.append(snippet)
|
||||
|
||||
if not parts:
|
||||
return ""
|
||||
|
||||
return "\n---\n".join(parts)
|
||||
|
||||
|
||||
def format_no_match_hint(error: Optional[str], match_count: int,
|
||||
old_string: str, content: str) -> str:
|
||||
"""Return a '\\n\\nDid you mean...' snippet for plain no-match errors.
|
||||
|
||||
Gated so the hint only fires for actual "old_string not found" failures.
|
||||
Ambiguous-match ("Found N matches"), escape-drift, and identical-strings
|
||||
errors all have ``match_count == 0`` but a "did you mean?" snippet would
|
||||
be misleading — those failed for unrelated reasons.
|
||||
|
||||
Returns an empty string when there's nothing useful to append.
|
||||
"""
|
||||
if match_count != 0:
|
||||
return ""
|
||||
if not error or not error.startswith("Could not find"):
|
||||
return ""
|
||||
hint = find_closest_lines(old_string, content)
|
||||
if not hint:
|
||||
return ""
|
||||
return "\n\nDid you mean one of these sections?\n" + hint
|
||||
|
||||
@@ -290,16 +290,10 @@ def _validate_operations(
|
||||
)
|
||||
if count == 0:
|
||||
label = f"'{hunk.context_hint}'" if hunk.context_hint else "(no hint)"
|
||||
msg = (
|
||||
errors.append(
|
||||
f"{op.file_path}: hunk {label} not found"
|
||||
+ (f" — {match_error}" if match_error else "")
|
||||
)
|
||||
try:
|
||||
from tools.fuzzy_match import format_no_match_hint
|
||||
msg += format_no_match_hint(match_error, count, search_pattern, simulated)
|
||||
except Exception:
|
||||
pass
|
||||
errors.append(msg)
|
||||
else:
|
||||
# Advance simulation so subsequent hunks validate correctly.
|
||||
# Reuse the result from the call above — no second fuzzy run.
|
||||
@@ -543,13 +537,7 @@ def _apply_update(op: PatchOperation, file_ops: Any) -> Tuple[bool, str]:
|
||||
error = None
|
||||
|
||||
if error:
|
||||
err_msg = f"Could not apply hunk: {error}"
|
||||
try:
|
||||
from tools.fuzzy_match import format_no_match_hint
|
||||
err_msg += format_no_match_hint(error, 0, search_pattern, new_content)
|
||||
except Exception:
|
||||
pass
|
||||
return False, err_msg
|
||||
return False, f"Could not apply hunk: {error}"
|
||||
else:
|
||||
# Addition-only hunk (no context or removed lines).
|
||||
# Insert at the location indicated by the context hint, or at end of file.
|
||||
|
||||
@@ -575,15 +575,9 @@ def _patch_skill(
|
||||
if match_error:
|
||||
# Show a short preview of the file so the model can self-correct
|
||||
preview = content[:500] + ("..." if len(content) > 500 else "")
|
||||
err_msg = match_error
|
||||
try:
|
||||
from tools.fuzzy_match import format_no_match_hint
|
||||
err_msg += format_no_match_hint(match_error, match_count, old_string, content)
|
||||
except Exception:
|
||||
pass
|
||||
return {
|
||||
"success": False,
|
||||
"error": err_msg,
|
||||
"error": match_error,
|
||||
"file_preview": preview,
|
||||
}
|
||||
|
||||
|
||||
@@ -290,6 +290,9 @@ 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.
|
||||
@@ -298,9 +301,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
|
||||
5. Generate ultraplan_YYYYMMDD.md
|
||||
6. File Gitea issue with the plan
|
||||
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
|
||||
|
||||
Output format:
|
||||
- Mission statement
|
||||
@@ -308,3 +311,176 @@ 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="🗺️",
|
||||
)
|
||||
|
||||
@@ -47,7 +47,7 @@ _HERMES_CORE_TOOLS = [
|
||||
# Text-to-speech
|
||||
"text_to_speech",
|
||||
# Planning & memory
|
||||
"todo", "memory",
|
||||
"todo", "ultraplan", "memory",
|
||||
# Session history search
|
||||
"session_search",
|
||||
# Clarifying questions
|
||||
@@ -157,8 +157,8 @@ TOOLSETS = {
|
||||
},
|
||||
|
||||
"todo": {
|
||||
"description": "Task planning and tracking for multi-step work",
|
||||
"tools": ["todo"],
|
||||
"description": "Task planning and tracking for multi-step work, including daily Ultraplan artifacts",
|
||||
"tools": ["todo", "ultraplan"],
|
||||
"includes": []
|
||||
},
|
||||
|
||||
|
||||
Reference in New Issue
Block a user