fix(herdr): ensure unique agent name via sha1 truncation and align mock errors

- Replace legacy s[:16]-s[-15:] truncation with 8-char SHA-1 hash suffix in sanitize.py and lib.sh to prevent workspace name collisions
- Align mock Herdr error messages in tests/conftest.py to real Herdr 0.8.0 output format
- Add missing required/invalid_agent_name to early abort regex in lib.sh
- Remove legacy heuristics in lib.sh has-session and unify mock agent lookups
- Add unit tests in tests/test_sanitize_and_mock_errors.py (256/256 passed)
This commit is contained in:
2026-08-15 09:00:48 +09:00
parent 301ff5bb1f
commit 14b9de14fc
5 changed files with 165 additions and 27 deletions
+46 -13
View File
@@ -1,10 +1,16 @@
import os
import sys
import shutil
import json
import uuid
import sqlite3
from pathlib import Path
import pytest
SKILLS_DIR = str(Path(__file__).resolve().parent.parent / ".agents" / "skills")
if SKILLS_DIR not in sys.path:
sys.path.insert(0, SKILLS_DIR)
@pytest.fixture
def mam_sandbox(tmp_path, monkeypatch):
"""
@@ -160,6 +166,24 @@ def save_state():
# Save calls immediately so they persist even if we exit early or error out
save_state()
def _sanitize_agent_name(name):
if not name:
return "agent"
s = str(name).lower()
import re, hashlib
s = re.sub(r'[^a-z0-9_-]', '-', s)
if not s or not s[0].isalpha():
s = "x-" + s
if len(s) > 32:
h = hashlib.sha1(s.encode('utf-8')).hexdigest()[:8]
s = f"{s[:23]}-{h}"
return s
def _match_agent(k, target):
if not k or not target:
return False
return k == target or _sanitize_agent_name(k) == target or _sanitize_agent_name(target) == k or _sanitize_agent_name(k) == _sanitize_agent_name(target)
args = sys.argv[1:]
while args and args[0] in ("-L", "--server", "-s", "--session"):
if len(args) > 1:
@@ -399,17 +423,26 @@ elif cmd1 == "agent":
i += 1
if unknown_flags:
sys.stderr.write("error: unexpected argument '" + str(unknown_flags[0]) + "' found\\\\n\\\\nUsage: herdr agent start <NAME> --kind <KIND> --pane <ID> [OPTIONS] [-- [AGENT_ARG]...]\\\\n")
sys.stderr.write("unknown option: " + str(unknown_flags[0]) + "\\n")
sys.exit(1)
if not agent_type or not pane:
sys.stderr.write("error: the following required arguments were not provided:\\\\n --kind <KIND>\\\\n --pane <ID>\\\\n")
missing_arg = "--pane" if not pane else "--kind"
sys.stderr.write("missing required " + missing_arg + "\\n")
sys.exit(1)
# Name validation: ^[a-z][a-z0-9_-]{0,31}$
import re
if not re.match(r'^[a-z][a-z0-9_-]{0,31}$', name):
print(f"error: invalid value '{name}' for '<NAME>': agent name must start with a lowercase letter and contain only lowercase letters, digits, '-' or '_' (1-32 characters)", file=sys.stderr)
import json
err_payload = {
"id": "cli:agent:start",
"error": {
"code": "invalid_agent_name",
"message": "agent name must start with a lowercase letter and contain only lowercase letters, digits, '-' or '_' (1-32 characters)"
}
}
sys.stderr.write(json.dumps(err_payload) + "\\n")
sys.exit(1)
# TUI Welcome Tokens definition to prevent TUI readiness check timeout
@@ -540,7 +573,7 @@ elif cmd1 == "agent":
agents = state.get("agents", {})
matched_k = None
for k in agents:
if k == name or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == name) or (len(name) > 32 and f"{name[:16]}-{name[-15:]}"[:32] == k):
if _match_agent(k, name):
matched_k = k
break
if matched_k:
@@ -565,7 +598,7 @@ elif cmd1 == "agent":
}))
sys.exit(0)
else:
sys.stderr.write("Agent " + name + " not found\\\\n")
sys.stderr.write("Agent " + name + " not found\\n")
sys.exit(1)
elif cmd2 == "read":
if len(args) < 3:
@@ -574,7 +607,7 @@ elif cmd1 == "agent":
agents = state.get("agents", {})
matched_k = None
for k in agents:
if k == name or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == name) or (len(name) > 32 and f"{name[:16]}-{name[-15:]}"[:32] == k):
if _match_agent(k, name):
matched_k = k
break
if matched_k:
@@ -582,7 +615,7 @@ elif cmd1 == "agent":
print(buffer_content)
sys.exit(0)
else:
sys.stderr.write("Agent " + name + " not found\\\\n")
sys.stderr.write("Agent " + name + " not found\\n")
sys.exit(1)
elif cmd2 == "prompt":
if len(args) < 4:
@@ -592,12 +625,12 @@ elif cmd1 == "agent":
agents = state.get("agents", {})
matched_k = None
for k in agents:
if k == name or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == name) or (len(name) > 32 and f"{name[:16]}-{name[-15:]}"[:32] == k):
if _match_agent(k, name):
matched_k = k
break
if matched_k:
agents[matched_k]["sent_text"] = agents[matched_k].get("sent_text", "") + text
agents[matched_k]["buffer"] = agents[matched_k].get("buffer", "") + "\\\\n" + text + "\\\\n\\\\nesc to interrupt"
agents[matched_k]["buffer"] = agents[matched_k].get("buffer", "") + "\\n" + text + "\\n\\nesc to interrupt"
state["agents"] = agents
save_state()
print(json.dumps({"id": "cli:agent:prompt", "result": {"type": "ok"}}))
@@ -610,7 +643,7 @@ elif cmd1 == "agent":
agents = state.get("agents", {})
matched_k = None
for k in agents:
if k == name or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == name) or (len(name) > 32 and f"{name[:16]}-{name[-15:]}"[:32] == k):
if _match_agent(k, name):
matched_k = k
break
if matched_k:
@@ -716,7 +749,7 @@ elif cmd1 == "list-panes":
agents = state.get("agents", {})
matched_k = None
for k in agents:
if k == session_target or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == session_target) or (len(session_target) > 32 and f"{session_target[:16]}-{session_target[-15:]}"[:32] == k):
if _match_agent(k, session_target):
matched_k = k
break
if matched_k:
@@ -744,7 +777,7 @@ elif cmd1 == "has-session":
elif len(args) > 1:
sess_target = args[1]
agents = state.get("agents", {})
if any(k == sess_target or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == sess_target) or (len(sess_target) > 32 and f"{sess_target[:16]}-{sess_target[-15:]}"[:32] == k) for k in agents):
if any(_match_agent(k, sess_target) for k in agents):
sys.exit(0)
else:
sys.exit(1)
@@ -756,7 +789,7 @@ elif cmd1 == "kill-session":
elif len(args) > 1:
sess_target = args[1]
agents = state.get("agents", {})
to_del = [k for k in agents if k == sess_target or (len(k) > 32 and f"{k[:16]}-{k[-15:]}"[:32] == sess_target) or (len(sess_target) > 32 and f"{sess_target[:16]}-{sess_target[-15:]}"[:32] == k)]
to_del = [k for k in agents if _match_agent(k, sess_target)]
for k in to_del:
del agents[k]
state["agents"] = agents
+84
View File
@@ -0,0 +1,84 @@
import subprocess
import hashlib
import json
import sys
import pytest
from lib_py.agents.sanitize import sanitize_herdr_agent_name
def test_sanitize_sibling_workspace_non_collision():
"""
Verifies that sibling workspaces with identical prefix and suffix
do NOT collapse into the same 32-character sanitized herdr agent name.
"""
workspaces = [
"canary-projects-educative-export-tools-creator-claude",
"canary-projects-getting-started-a2a-creator-claude",
"canary-projects-multi-agent-mux-creator-claude",
"canary-projects-pu-riverpod-cookbook-creator-claude",
]
sanitized = [sanitize_herdr_agent_name(w) for w in workspaces]
# 1. All lengths <= 32
for s in sanitized:
assert len(s) <= 32, f"Length exceeds 32: {s}"
assert len(s) == 32, f"Expected 32 chars for long names: {s}"
# 2. All names are unique (0 collisions)
assert len(set(sanitized)) == len(workspaces), f"Collisions detected: {sanitized}"
def test_sanitize_bash_python_parity(mam_sandbox):
"""
Verifies that bash _sanitize_herdr_agent_name and Python sanitize_herdr_agent_name
produce byte-for-byte identical output across various edge cases.
"""
test_cases = [
"canary-projects-multi-agent-mux-creator-claude",
"canary-projects-educative-export-tools-creator-claude",
"canary-projects-getting-started-a2a-creator-claude",
"canary-projects-pu-riverpod-cookbook-creator-claude",
"short-name",
"UPPER_CASE_123.dots",
"123_starts_digit",
"_underscore_leading",
"",
"exact-32-chars-long-name-1234567",
"exact-33-chars-long-name-12345678",
]
lib_path = mam_sandbox / ".agents" / "skills" / "lib.sh"
for case in test_cases:
py_res = sanitize_herdr_agent_name(case)
cmd = f"""
source {lib_path}
_sanitize_herdr_agent_name "{case}"
"""
bash_res = subprocess.check_output(["bash", "-c", cmd]).decode().strip()
assert py_res == bash_res, f"Mismatch for '{case}': Python={py_res} vs Bash={bash_res}"
def test_mock_herdr_error_formatting_and_abort(mock_herdr, mock_agents):
"""
Verifies that mock herdr error output matches real Herdr 0.8.0 format
and triggers lib.sh early abort regex.
"""
import re
abort_regex = re.compile(r"^usage:|unknown option|unknown flag|missing required|invalid_agent_name|^error:", re.IGNORECASE)
mock_bin = str(mock_agents / "herdr")
# 1. Unknown flag
res_unknown = subprocess.run([mock_bin, "agent", "start", "testagent", "--kind", "claude", "--pane", "w1:p1", "--env", "FOO=BAR"], capture_output=True, text=True)
assert res_unknown.returncode != 0
assert abort_regex.search(res_unknown.stderr) is not None, f"Stderr did not match abort regex: {res_unknown.stderr}"
assert "unknown option: --env" in res_unknown.stderr
# 2. Missing required --pane
res_missing = subprocess.run([mock_bin, "agent", "start", "testagent", "--kind", "claude"], capture_output=True, text=True)
assert res_missing.returncode != 0
assert abort_regex.search(res_missing.stderr) is not None, f"Stderr did not match abort regex: {res_missing.stderr}"
assert "missing required --pane" in res_missing.stderr
# 3. Invalid agent name
res_badname = subprocess.run([mock_bin, "agent", "start", "BAD_NAME", "--kind", "claude", "--pane", "w1:p1"], capture_output=True, text=True)
assert res_badname.returncode != 0
assert abort_regex.search(res_badname.stderr) is not None, f"Stderr did not match abort regex: {res_badname.stderr}"
assert "invalid_agent_name" in res_badname.stderr
+3 -1
View File
@@ -54,7 +54,8 @@ def test_create_session_full(mam_sandbox, mock_herdr, mock_agents):
assert len(agents) == 1, "There should be exactly one registered agent in the mock herdr state."
session_name = list(agents.keys())[0]
assert session_name.endswith("-creator-claude")
from lib_py.agents.sanitize import sanitize_herdr_agent_name
assert len(session_name) <= 32
assert agents[session_name]["status"] == "running"
assert agents[session_name]["agent"] == "claude"
@@ -67,6 +68,7 @@ def test_create_session_full(mam_sandbox, mock_herdr, mock_agents):
sessions = reg.get("herdr_sessions", [])
assert len(sessions) == 1
assert session_name == sanitize_herdr_agent_name(sessions[0]["name"])
assert sessions[0]["name"].endswith("-creator-claude")
assert sessions[0]["status"] == "running"
assert sessions[0]["role"] == "Creator"