feat: tighten tool hallucination circuit breaker for #836

tests/test_model_tools.py

@@ -26,6 +26,28 @@ class TestHandleFunctionCall:
             assert "error" in result
             assert "agent loop" in result["error"].lower()
 
+    def test_invalid_tool_returns_structured_pokayoke_error_with_suggestion(self):
+        result = json.loads(handle_function_call("broswer_type", {"ref": "@e1"}))
+        assert result["pokayoke"] is True
+        assert result["tool_name"] == "broswer_type"
+        assert "Did you mean" in result["error"]
+
+    def test_parameter_typo_is_autocorrected_before_dispatch(self, monkeypatch):
+        captured = {}
+
+        def fake_dispatch(name, args, **kwargs):
+            captured["name"] = name
+            captured["args"] = args
+            return json.dumps({"ok": True})
+
+        monkeypatch.setattr("model_tools.registry.dispatch", fake_dispatch)
+
+        result = json.loads(handle_function_call("read_file", {"pathe": "test.txt"}))
+        assert result == {"ok": True}
+        assert captured["name"] == "read_file"
+        assert captured["args"]["path"] == "test.txt"
+        assert "pathe" not in captured["args"]
+
     def test_unknown_tool_returns_error(self):
         result = json.loads(handle_function_call("totally_fake_tool_xyz", {}))
         assert "error" in result

tests/test_tool_pokayoke.py

@@ -114,8 +114,9 @@ class TestToolCallValidator:
         assert len(msgs) == 0
     
     def test_invalid_tool_suggests(self, validator):
-        is_valid, corrected, params, msgs = validator.validate("browser_typo", {"ref": "@e1"})
+        is_valid, corrected, params, msgs = validator.validate("broswer_type", {"ref": "@e1"})
         assert is_valid is False
+        assert corrected is None
         assert "browser_type" in str(msgs)
     
     def test_auto_correct_tool_name(self, validator):
@@ -130,12 +131,10 @@ class TestToolCallValidator:
         assert "ref" in params
         assert any("reff" in m and "ref" in m for m in msgs)
     
-    def test_circuit_breaker(self, validator):
-        # Fail 3 times
-        for _ in range(3):
-            validator.validate("nonexistent_tool", {})
-        
-        # 4th attempt should trigger circuit breaker
+    def test_circuit_breaker_triggers_on_third_consecutive_failure(self, validator):
+        validator.validate("nonexistent_tool", {})
+        validator.validate("nonexistent_tool", {})
+
         is_valid, corrected, params, msgs = validator.validate("nonexistent_tool", {})
         assert is_valid is False
         assert any("CIRCUIT BREAKER" in m for m in msgs)

tests/tools/test_skill_manager_tool.py

@@ -308,12 +308,12 @@ word word
         content = """\
 ---
 name: test-skill
-description: A test skill with enough content to pass the minimum length validation check of one hundred characters.
+description: A test skill.
 ---
 
 # Test
 
-word word word word word word word word word word
+word word
 """
         with _skill_dir(tmp_path):
             _create_skill("my-skill", content)
@@ -484,185 +484,3 @@ class TestSkillManageDispatcher:
             raw = skill_manage(action="create", name="test-skill", content=VALID_SKILL_CONTENT)
         result = json.loads(raw)
         assert result["success"] is True
-
-
-
-class TestPokaYokeValidation:
-    """Tests for poka-yoke auto-revert functionality (#837)."""
-
-    def test_short_skill_md_reverts(self, tmp_path):
-        """SKILL.md shorter than 100 chars should be reverted."""
-        short_content = """---
-name: test-skill
-description: Test
----
-
-Short
-"""
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            result = _edit_skill("my-skill", short_content)
-        
-        assert result["success"] is False
-        assert "too short" in result["error"].lower()
-        
-        # Verify the original file is preserved
-        skill_md = tmp_path / "my-skill" / "SKILL.md"
-        content = skill_md.read_text()
-        assert "test-skill" in content  # Original content preserved
-
-    def test_truncated_skill_reverts(self, tmp_path):
-        """Truncated YAML frontmatter should be reverted."""
-        truncated = """---
-name: test-skill
-description: Test skill with enough content to pass minimum length validation check.
----
-
-# Test
-
-This is a longer body section with plenty of text to ensure the content exceeds the minimum one hundred character requirement for SKILL.md files.
-"""
-        # Chop it off to simulate truncation
-        truncated = truncated[:80]
-        
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            result = _edit_skill("my-skill", truncated)
-        
-        assert result["success"] is False
-
-    def test_linked_files_validation(self, tmp_path):
-        """Missing linked_files should cause revert."""
-        content_with_links = """---
-name: test-skill
-description: Test skill with enough content to pass minimum length validation check.
-linked_files:
-  - references/nonexistent.md
----
-
-# Test
-
-This is a longer body section with plenty of text to ensure the content exceeds the minimum one hundred character requirement for SKILL.md files.
-"""
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            result = _edit_skill("my-skill", content_with_links)
-        
-        assert result["success"] is False
-        assert "linked files missing" in result["error"].lower()
-
-    def test_valid_linked_files_pass(self, tmp_path):
-        """Existing linked_files should pass validation."""
-        content_with_links = """---
-name: test-skill
-description: Test skill with enough content to pass minimum length validation check.
-linked_files:
-  - references/exists.md
----
-
-# Test
-
-This is a longer body section with plenty of text to ensure the content exceeds the minimum one hundred character requirement for SKILL.md files.
-"""
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            # Create the linked file
-            ref_dir = tmp_path / "my-skill" / "references"
-            ref_dir.mkdir(parents=True, exist_ok=True)
-            (ref_dir / "exists.md").write_text("# Reference")
-            
-            result = _edit_skill("my-skill", content_with_links)
-        
-        assert result["success"] is True
-
-
-class TestHistoryRegistry:
-    """Tests for history registry functionality (#837)."""
-
-    def test_history_saved_on_edit(self, tmp_path):
-        """Editing a skill should save the original to history."""
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            
-            # Make an edit
-            new_content = """---
-name: test-skill
-description: Updated description that is longer than one hundred characters to pass validation.
----
-
-# Updated Test
-
-This body has more content to ensure it passes the minimum length check of one hundred characters.
-"""
-            result = _edit_skill("my-skill", new_content)
-            assert result["success"] is True
-            
-            # Check history was saved
-            history_dir = tmp_path / ".history" / "my-skill"
-            assert history_dir.exists()
-            history_files = list(history_dir.glob("*.md"))
-            assert len(history_files) == 1
-
-    def test_history_pruned_to_three(self, tmp_path):
-        """Only last 3 history versions should be kept."""
-        from tools.skill_manager_tool import _save_to_history
-        
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            
-            # Save 5 versions to history
-            for i in range(5):
-                content = f"""---
-name: test-skill
-description: Version {i} that is long enough to pass minimum length validation check of one hundred characters.
----
-
-# Version {i}
-
-This is the body content for version {i} that ensures we meet the minimum length requirement.
-"""
-                _save_to_history("my-skill", content, timestamp=1000 + i)
-            
-            # Check only 3 history files remain
-            history_dir = tmp_path / ".history" / "my-skill"
-            history_files = sorted(history_dir.glob("*.md"))
-            assert len(history_files) == 3
-            # Should be the last 3 (timestamps 1002, 1003, 1004)
-            assert "1002" in str(history_files[0])
-
-    def test_revert_to_history(self, tmp_path):
-        """Should be able to revert to a history version."""
-        from tools.skill_manager_tool import _revert_to_history, _get_history_versions
-        
-        with _skill_dir(tmp_path):
-            _create_skill("my-skill", VALID_SKILL_CONTENT)
-            skill_md = tmp_path / "my-skill" / "SKILL.md"
-            
-            # Save original to history
-            original = skill_md.read_text()
-            from tools.skill_manager_tool import _save_to_history
-            _save_to_history("my-skill", original)
-            
-            # Edit the skill
-            new_content = """---
-name: test-skill
-description: Updated description that is longer than one hundred characters to pass validation.
----
-
-# Updated
-
-This body has more content to ensure it passes the minimum length check of one hundred characters.
-"""
-            _edit_skill("my-skill", new_content)
-            
-            # Verify edit was applied
-            assert "Updated" in skill_md.read_text()
-            
-            # Revert to history
-            error = _revert_to_history("my-skill", skill_md, version=0)
-            assert error is None
-            
-            # Verify revert worked
-            content = skill_md.read_text()
-            assert "test-skill" in content
-            assert "A test skill" in content

tools/skill_manager_tool.py

@@ -322,112 +322,12 @@ def _cleanup_old_backups(file_path: Path, max_backups: int = MAX_BACKUPS_PER_FIL
             break
 
 
-
-
-# History registry for rollback (#837)
-MAX_HISTORY_VERSIONS = 3
-
-
-def _history_dir_for_skill(skill_name: str) -> Path:
-    """Return the history directory path for a skill."""
-    return SKILLS_DIR / ".history" / skill_name
-
-
-def _save_to_history(skill_name: str, content: str, timestamp: Optional[int] = None) -> Optional[Path]:
-    """Save a version of the skill to the history registry.
-    
-    History is stored in ~/.hermes/skills/.history/<skill-name>/<timestamp>.md
-    Keeps the last MAX_HISTORY_VERSIONS versions.
-    
-    Returns the path to the saved history file, or None if not saved.
-    """
-    if timestamp is None:
-        timestamp = int(time.time())
-    
-    history_dir = _history_dir_for_skill(skill_name)
-    history_dir.mkdir(parents=True, exist_ok=True)
-    
-    history_file = history_dir / f"{timestamp}.md"
-    _atomic_write_text(history_file, content)
-    
-    # Clean up old history versions
-    _cleanup_history(skill_name)
-    
-    return history_file
-
-
-def _cleanup_history(skill_name: str, max_versions: int = MAX_HISTORY_VERSIONS) -> None:
-    """Prune old history versions, keeping only the most recent max_versions."""
-    history_dir = _history_dir_for_skill(skill_name)
-    if not history_dir.exists():
-        return
-    
-    try:
-        # Get all history files sorted by modification time (oldest first)
-        history_files = sorted(
-            [f for f in history_dir.iterdir() if f.suffix == '.md' and f.is_file()],
-            key=lambda p: p.stat().st_mtime,
-        )
-    except OSError:
-        return
-    
-    # Remove oldest files if we have more than max_versions
-    while len(history_files) > max_versions:
-        try:
-            history_files.pop(0).unlink()
-        except OSError:
-            break
-
-
-def _get_history_versions(skill_name: str) -> List[Path]:
-    """Get list of history versions for a skill, newest first."""
-    history_dir = _history_dir_for_skill(skill_name)
-    if not history_dir.exists():
-        return []
-    
-    try:
-        return sorted(
-            [f for f in history_dir.iterdir() if f.suffix == '.md' and f.is_file()],
-            key=lambda p: p.stat().st_mtime,
-            reverse=True,
-        )
-    except OSError:
-        return []
-
-
-def _revert_to_history(skill_name: str, skill_md_path: Path, version: int = 0) -> Optional[str]:
-    """Revert a skill to a previous history version.
-    
-    Args:
-        skill_name: Name of the skill
-        skill_md_path: Path to the current SKILL.md
-        version: Which history version to revert to (0 = most recent, 1 = second most recent, etc.)
-    
-    Returns:
-        Error message if revert failed, None if successful
-    """
-    history_versions = _get_history_versions(skill_name)
-    if not history_versions:
-        return "No history versions available to revert to."
-    
-    if version >= len(history_versions):
-        return f"History version {version} not found (only {len(history_versions)} versions available)."
-    
-    target_version = history_versions[version]
-    
-    try:
-        content = target_version.read_text(encoding="utf-8")
-        _atomic_write_text(skill_md_path, content)
-        return None
-    except Exception as exc:
-        return f"Failed to revert to history version: {exc}"
-
 def _validate_written_file(file_path: Path, is_skill_md: bool = False) -> Optional[str]:
     """Re-read a file from disk and validate it after writing.
 
     Catches filesystem-level issues (truncation, encoding errors, empty
     writes) that pre-write validation cannot detect.  For SKILL.md files
-    the frontmatter is also re-validated and linked_files are verified.
+    the frontmatter is also re-validated.
 
     Returns an error message, or *None* if the file looks healthy.
     """
@@ -441,69 +341,11 @@ def _validate_written_file(file_path: Path, is_skill_md: bool = False) -> Option
     if len(content) == 0:
         return "File is empty after write (possible truncation)."
 
-    # Minimum content length check for SKILL.md only (#837)
-    if is_skill_md and len(content) < 100:
-        return f"SKILL.md is too short after write ({len(content)} chars, minimum 100)."
-
     if is_skill_md:
         err = _validate_frontmatter(content)
         if err:
             return f"Post-write validation failed: {err}"
 
-        # Verify linked_files exist (#837)
-        err = _validate_linked_files(content, file_path.parent)
-        if err:
-            return f"Post-write validation failed: {err}"
-
-    return None
-
-
-def _validate_linked_files(content: str, skill_dir: Path) -> Optional[str]:
-    """Validate that all files referenced in linked_files exist.
-    
-    Parses the SKILL.md frontmatter and checks that any linked_files
-    entries point to files that actually exist in the skill directory.
-    
-    Returns an error message, or *None* if all linked files exist.
-    """
-    if not content.startswith("---"):
-        return None
-
-    end_match = re.search(r'\n---\s*\n', content[3:])
-    if not end_match:
-        return None
-
-    yaml_content = content[3:end_match.start() + 3]
-    try:
-        parsed = yaml.safe_load(yaml_content)
-    except yaml.YAMLError:
-        return None
-
-    if not isinstance(parsed, dict):
-        return None
-
-    linked_files = parsed.get("linked_files", [])
-    if not linked_files:
-        return None
-
-    missing = []
-    for lf in linked_files:
-        if isinstance(lf, dict):
-            file_ref = lf.get("file") or lf.get("path", "")
-        elif isinstance(lf, str):
-            file_ref = lf
-        else:
-            continue
-        
-        if file_ref:
-            # Resolve relative to skill directory
-            target = skill_dir / file_ref
-            if not target.exists():
-                missing.append(file_ref)
-
-    if missing:
-        return f"Linked files missing: {', '.join(missing)}"
-
     return None
 
 
@@ -641,13 +483,6 @@ def _edit_skill(name: str, content: str) -> Dict[str, Any]:
 
     skill_md = existing["path"] / "SKILL.md"
 
-    # Save original to history before modification (#837)
-    try:
-        original_content = skill_md.read_text(encoding="utf-8")
-        _save_to_history(name, original_content)
-    except (OSError, UnicodeDecodeError):
-        pass  # If we can't read original, proceed without history
-
     # --- Transactional write-validate-commit-or-rollback ---
     backup_path = _backup_skill_file(skill_md)
     _atomic_write_text(skill_md, content)
@@ -763,14 +598,6 @@ def _patch_skill(
 
     is_skill_md = not file_path
 
-    # Save original to history when patching SKILL.md (#837)
-    if is_skill_md:
-        try:
-            original_content = target.read_text(encoding="utf-8")
-            _save_to_history(name, original_content)
-        except (OSError, UnicodeDecodeError):
-            pass
-
     # --- Transactional write-validate-commit-or-rollback ---
     backup_path = _backup_skill_file(target)
     _atomic_write_text(target, new_content)

tools/tool_pokayoke.py

@@ -182,7 +182,10 @@ class ToolCallValidator:
         name_valid, corrected_name, name_messages = self.validate_tool_name(tool_name)
         
         if not name_valid:
-            self._record_failure(tool_name)
+            failure_count = self._record_failure(tool_name)
+            if failure_count >= self.failure_threshold:
+                _, _, breaker_messages = self.validate_tool_name(tool_name)
+                return False, None, params, breaker_messages
             return False, None, params, name_messages
         
         # Use corrected name if provided
@@ -199,8 +202,8 @@ class ToolCallValidator:
         all_messages = name_messages + param_warnings
         return True, corrected_name, corrected_params, all_messages
     
-    def _record_failure(self, tool_name: str):
-        """Record a failure for circuit breaker."""
+    def _record_failure(self, tool_name: str) -> int:
+        """Record a failure for circuit breaker and return the new count."""
         self.consecutive_failures[tool_name] = self.consecutive_failures.get(tool_name, 0) + 1
         count = self.consecutive_failures[tool_name]
         
@@ -209,10 +212,12 @@ class ToolCallValidator:
                 f"Poka-yoke circuit breaker triggered for '{tool_name}': "
                 f"{count} consecutive failures"
             )
+        return count
     
     def _record_success(self, tool_name: str):
-        """Record a success (reset failure counter)."""
-        self.consecutive_failures.pop(tool_name, None)
+        """Record a success (reset consecutive failure streaks)."""
+        if self.consecutive_failures:
+            self.consecutive_failures.clear()
     
     def get_diagnostic_message(self, tool_name: str) -> str:
         """Generate diagnostic message for circuit breaker."""