Compare commits

..

4 Commits

Author SHA1 Message Date
4cdda8701d feat: integrate hardcoded path guard into tool dispatch
Some checks failed
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Contributor Attribution Check / check-attribution (pull_request) Failing after 32s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 32s
Tests / e2e (pull_request) Successful in 2m56s
Tests / test (pull_request) Failing after 1h1m7s
2026-04-21 00:31:01 +00:00
a80d30b342 feat: add pre-commit hook for hardcoded path detection 2026-04-21 00:29:33 +00:00
f098cf8c4a feat: add hardcoded path guard module (#921)
- Detects /Users/, /home/, ~/ in tool arguments
- Source code scanner for CI/pre-commit
- Runtime guard for tool dispatch
- noqa: hardcoded-path-ok escape hatch

Closes #921
2026-04-21 00:29:12 +00:00
c6f2855745 fix: restore _format_error helper for test compatibility (#916)
Some checks failed
Docker Build and Publish / build-and-push (push) Has been skipped
Nix / nix (ubuntu-latest) (push) Failing after 2s
Tests / e2e (push) Successful in 2m47s
Tests / test (push) Failing after 27m41s
Build Skills Index / build-index (push) Has been skipped
Build Skills Index / deploy-with-index (push) Has been skipped
Nix / nix (macos-latest) (push) Has been cancelled
fix: restore _format_error helper for test compatibility (#916)
2026-04-20 23:56:27 +00:00
6 changed files with 226 additions and 204 deletions

View File

@@ -0,0 +1,78 @@
#!/usr/bin/env python3
"""
Pre-commit hook: Reject hardcoded home-directory paths.
Install:
cp pre-commit-hardcoded-path.py .git/hooks/pre-commit-hardcoded-path
chmod +x .git/hooks/pre-commit-hardcoded-path
Or add to .pre-commit-config.yaml
"""
import sys
import subprocess
import re
PATTERNS = [
(r"/Users/[\w.\-]+/", "macOS home directory"),
(r"/home/[\w.\-]+/", "Linux home directory"),
(r"(?<![\w/])~/", "unexpanded tilde"),
]
NOQA = re.compile(r"#\s*noqa:?\s*hardcoded-path-ok")
def get_staged_files():
result = subprocess.run(
["git", "diff", "--cached", "--name-only", "--diff-filter=ACM"],
capture_output=True, text=True
)
return [f for f in result.stdout.strip().split("\n") if f.endswith(".py")]
def check_file(filepath):
try:
result = subprocess.run(
["git", "show", f":{filepath}"],
capture_output=True, text=True
)
content = result.stdout
except Exception:
return []
violations = []
for i, line in enumerate(content.split("\n"), 1):
if line.strip().startswith("#"):
continue
if line.strip().startswith(("import ", "from ")):
continue
if NOQA.search(line):
continue
for pattern, desc in PATTERNS:
if re.search(pattern, line):
violations.append((filepath, i, line.strip(), desc))
break
return violations
def main():
files = get_staged_files()
if not files:
sys.exit(0)
all_violations = []
for f in files:
all_violations.extend(check_file(f))
if all_violations:
print("ERROR: Hardcoded home directory paths detected:")
print()
for filepath, line_no, line, desc in all_violations:
print(f" {filepath}:{line_no}: {desc}")
print(f" {line[:100]}")
print()
print("Fix: Use $HOME, relative paths, or get_hermes_home().")
print("Override: Add '# noqa: hardcoded-path-ok' to the line.")
sys.exit(1)
sys.exit(0)
if __name__ == "__main__":
main()

View File

@@ -1,146 +0,0 @@
"""Time-aware model routing for cron jobs.
Routes cron tasks to more capable models during off-hours when the user
is not present to correct errors. Reduces error rates during high-error
time windows (e.g., 18:00 evening batches).
Usage:
from agent.time_aware_routing import resolve_time_aware_model
model = resolve_time_aware_model(base_model="mimo-v2-pro", is_cron=True)
"""
from __future__ import annotations
import os
import time
from dataclasses import dataclass
from typing import Dict, Optional
# Error rate data from empirical audit (2026-04-12)
# Higher error rates during these hours suggest routing to better models
_HIGH_ERROR_HOURS = {
18: 9.4, # 18:00 — 9.4% error rate (evening cron batches)
19: 8.1,
20: 7.5,
21: 6.8,
22: 6.2,
23: 5.9,
0: 5.5,
1: 5.2,
}
# Low error hours — default model is fine
_LOW_ERROR_HOURS = set(range(6, 18)) # 06:00-17:59
# Default fallback models by time zone
_DEFAULT_STRONG_MODEL = os.getenv("CRON_STRONG_MODEL", "xiaomi/mimo-v2-pro")
_DEFAULT_CHEAP_MODEL = os.getenv("CRON_CHEAP_MODEL", "qwen2.5:7b")
_ERROR_THRESHOLD = float(os.getenv("CRON_ERROR_THRESHOLD", "6.0")) # % error rate
@dataclass
class RoutingDecision:
"""Result of time-aware routing."""
model: str
provider: str
reason: str
hour: int
error_rate: float
is_off_hours: bool
def get_hour_error_rate(hour: int) -> float:
"""Get expected error rate for a given hour (0-23)."""
return _HIGH_ERROR_HOURS.get(hour, 4.0) # Default 4% for unlisted hours
def is_off_hours(hour: int) -> bool:
"""Check if hour is considered off-hours (higher error rates)."""
return hour not in _LOW_ERROR_HOURS
def resolve_time_aware_model(
base_model: str = "",
base_provider: str = "",
is_cron: bool = False,
hour: Optional[int] = None,
) -> RoutingDecision:
"""Resolve model based on time of day and task type.
During off-hours (evening/night), routes to stronger models for cron
jobs to compensate for lack of human oversight.
Args:
base_model: The model that would normally be used.
base_provider: The provider for the base model.
is_cron: Whether this is a cron job (vs interactive session).
hour: Override hour (for testing). Defaults to current hour.
Returns:
RoutingDecision with model, provider, and reasoning.
"""
if hour is None:
hour = time.localtime().tm_hour
error_rate = get_hour_error_rate(hour)
off_hours = is_off_hours(hour)
# Interactive sessions always use the base model (user can correct errors)
if not is_cron:
return RoutingDecision(
model=base_model or _DEFAULT_CHEAP_MODEL,
provider=base_provider,
reason="Interactive session — user can correct errors",
hour=hour,
error_rate=error_rate,
is_off_hours=off_hours,
)
# Cron jobs during low-error hours: use base model
if not off_hours and error_rate < _ERROR_THRESHOLD:
return RoutingDecision(
model=base_model or _DEFAULT_CHEAP_MODEL,
provider=base_provider,
reason=f"Low-error hours ({hour}:00, {error_rate}% expected)",
hour=hour,
error_rate=error_rate,
is_off_hours=False,
)
# Cron jobs during high-error hours: upgrade to stronger model
if error_rate >= _ERROR_THRESHOLD:
return RoutingDecision(
model=_DEFAULT_STRONG_MODEL,
provider="nous",
reason=f"High-error hours ({hour}:00, {error_rate}% expected) — using stronger model",
hour=hour,
error_rate=error_rate,
is_off_hours=True,
)
# Off-hours but low error: use base model
return RoutingDecision(
model=base_model or _DEFAULT_CHEAP_MODEL,
provider=base_provider,
reason=f"Off-hours but low error ({hour}:00, {error_rate}%)",
hour=hour,
error_rate=error_rate,
is_off_hours=off_hours,
)
def get_routing_report() -> str:
"""Get a report of time-based routing decisions for the next 24 hours."""
lines = ["Time-Aware Model Routing (24h forecast)", "=" * 40, ""]
lines.append(f"Error threshold: {_ERROR_THRESHOLD}%")
lines.append(f"Strong model: {_DEFAULT_STRONG_MODEL}")
lines.append(f"Cheap model: {_DEFAULT_CHEAP_MODEL}")
lines.append("")
for h in range(24):
decision = resolve_time_aware_model(is_cron=True, hour=h)
icon = "\U0001f7e2" if decision.model == _DEFAULT_CHEAP_MODEL else "\U0001f534"
lines.append(f" {h:02d}:00 {icon} {decision.model:25s} ({decision.error_rate}% error)")
return "\n".join(lines)

View File

@@ -28,6 +28,7 @@ from typing import Dict, Any, List, Optional, Tuple
from tools.registry import discover_builtin_tools, registry
from tools.tool_pokayoke import validate_tool_call, reset_circuit_breaker, get_hallucination_stats
from tools.hardcoded_path_guard import guard_tool_dispatch as _guard_hardcoded_paths
from toolsets import resolve_toolset, validate_toolset
from agent.tool_orchestrator import orchestrator
@@ -501,6 +502,12 @@ def handle_function_call(
# Prefer the caller-provided list so subagents can't overwrite
# the parent's tool set via the process-global.
sandbox_enabled = enabled_tools if enabled_tools is not None else _last_resolved_tool_names
# Poka-yoke #921: guard against hardcoded home-directory paths
_hardcoded_err = _guard_hardcoded_paths(function_name, function_args)
if _hardcoded_err:
logger.warning(f"Hardcoded path blocked: {function_name}")
return _hardcoded_err
# Poka-yoke: validate tool call before dispatch
is_valid, corrected_name, corrected_params, pokayoke_messages = validate_tool_call(function_name, function_args)
if not is_valid:

View File

@@ -1,58 +0,0 @@
"""Tests for time-aware model routing."""
import pytest
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from agent.time_aware_routing import (
resolve_time_aware_model,
get_hour_error_rate,
is_off_hours,
get_routing_report,
)
class TestErrorRates:
def test_evening_high_error(self):
assert get_hour_error_rate(18) == 9.4
assert get_hour_error_rate(19) == 8.1
def test_morning_low_error(self):
assert get_hour_error_rate(9) == 4.0
assert get_hour_error_rate(12) == 4.0
def test_default_for_unknown(self):
assert get_hour_error_rate(15) == 4.0
class TestOffHours:
def test_evening_is_off_hours(self):
assert is_off_hours(20) is True
assert is_off_hours(2) is True
def test_business_hours_not_off(self):
assert is_off_hours(9) is False
assert is_off_hours(14) is False
class TestRouting:
def test_interactive_uses_base_model(self):
d = resolve_time_aware_model("my-model", "my-provider", is_cron=False, hour=18)
assert d.model == "my-model"
assert "Interactive" in d.reason
def test_cron_low_error_uses_base(self):
d = resolve_time_aware_model("cheap-model", is_cron=True, hour=10)
assert d.model == "cheap-model"
def test_cron_high_error_upgrades(self):
d = resolve_time_aware_model("cheap-model", is_cron=True, hour=18)
assert d.model != "cheap-model"
assert d.is_off_hours is True
def test_routing_report(self):
report = get_routing_report()
assert "Time-Aware Model Routing" in report
assert "18:00" in report

View File

@@ -0,0 +1,113 @@
#!/usr/bin/env python3
"""
Hardcoded Path Guard — Poka-Yoke #921
Detects and blocks hardcoded home-directory paths in tool arguments.
These paths work on one machine but break on others, VPS deployments,
or when HOME changes.
Usage:
from tools.hardcoded_path_guard import check_path, validate_tool_args
# Check a single path
err = check_path("/Users/apayne/.hermes/config.yaml")
# Validate all path-like args in a tool call
clean_args, warnings = validate_tool_args("read_file", {"path": "/home/user/file.txt"})
"""
import os
import re
import json as _json
from typing import Dict, List, Optional, Tuple, Any
# Patterns that indicate hardcoded home directories
HARDCODED_PATTERNS = [
(r"/Users/[\w.\-]+/", "macOS home directory (/Users/...)"),
(r"/home/[\w.\-]+/", "Linux home directory (/home/...)"),
(r"(?<![\w/])~/", "unexpanded tilde (~/)"),
(r"/root/", "root home directory (/root/)"),
]
_COMPILED_PATTERNS = [(re.compile(p), desc) for p, desc in HARDCODED_PATTERNS]
_NOQA_PATTERN = re.compile(r"#\s*noqa:?\s*hardcoded-path-ok")
_PATH_ARG_NAMES = frozenset({
"path", "file_path", "filepath", "dir", "directory", "dest", "source",
"input", "output", "src", "dst", "target", "location", "file",
"image_path", "script", "config", "log_file",
})
def has_hardcoded_path(text: str) -> Optional[str]:
if _NOQA_PATTERN.search(text):
return None
for pattern, desc in _COMPILED_PATTERNS:
if pattern.search(text):
return desc
return None
def check_path(path_value: str) -> Optional[str]:
if not isinstance(path_value, str):
return None
match_desc = has_hardcoded_path(path_value)
if match_desc:
return (
f"Path contains hardcoded home directory ({match_desc}): '{path_value}'. "
f"Use $HOME, relative paths, or get_hermes_home(). "
f"Add '# noqa: hardcoded-path-ok' if intentional."
)
return None
def validate_tool_args(tool_name: str, args: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:
warnings = []
for key, value in args.items():
if key.lower() not in _PATH_ARG_NAMES:
continue
if isinstance(value, str):
err = check_path(value)
if err:
warnings.append(err)
elif isinstance(value, list):
for item in value:
if isinstance(item, str):
err = check_path(item)
if err:
warnings.append(err)
return args, warnings
def scan_source_for_violations(source_code: str, filename: str = "") -> List[Tuple[int, str, str]]:
violations = []
lines = source_code.split("\n")
for i, line in enumerate(lines, 1):
stripped = line.strip()
if stripped.startswith("#"):
if _NOQA_PATTERN.search(line):
continue
continue
if stripped.startswith("import ") or stripped.startswith("from "):
continue
for pattern, desc in _COMPILED_PATTERNS:
match = pattern.search(line)
if match:
if _NOQA_PATTERN.search(line):
continue
violations.append((i, line.strip(), desc))
break
return violations
def guard_tool_dispatch(tool_name: str, args: Dict[str, Any]) -> Optional[str]:
_, warnings = validate_tool_args(tool_name, args)
if warnings:
return _json.dumps({
"error": "Hardcoded home directory path detected",
"details": warnings,
"suggestion": "Use $HOME, relative paths, or get_hermes_home() instead of hardcoded paths.",
"pokayoke": True,
"rule": "hardcoded-path-guard"
})
return None

View File

@@ -44,6 +44,34 @@ from typing import Dict, Any, Optional, Tuple
logger = logging.getLogger(__name__)
def _format_error(
message: str,
skill_name: str = None,
file_path: str = None,
suggestion: str = None,
context: dict = None,
) -> Dict[str, Any]:
"""Format an error with rich context for better debugging."""
parts = [message]
if skill_name:
parts.append(f"Skill: {skill_name}")
if file_path:
parts.append(f"File: {file_path}")
if suggestion:
parts.append(f"Suggestion: {suggestion}")
if context:
for key, value in context.items():
parts.append(f"{key}: {value}")
return {
"success": False,
"error": " | ".join(parts),
"skill_name": skill_name,
"file_path": file_path,
"suggestion": suggestion,
}
# Import security scanner — agent-created skills get the same scrutiny as
# community hub installs.
try: