272 lines
10 KiB
Python
272 lines
10 KiB
Python
|
"""Tests for TOOL/DM directive parsing, native aliases, and the
|
||
|
|
box.exec / tools.list exec ops (dynamic in-band message passing)."""
|
||
|
|
import importlib.util
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
import subprocess
|
||
|
|
import sys
|
||
|
|
import unittest
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||
|
|
|
||
|
|
|
||
|
|
def _load(mod_name, rel_path):
|
||
|
|
spec = importlib.util.spec_from_file_location(mod_name, REPO_ROOT / rel_path)
|
||
|
|
mod = importlib.util.module_from_spec(spec)
|
||
|
|
spec.loader.exec_module(mod)
|
||
|
|
return mod
|
||
|
|
|
||
|
|
|
||
|
|
harv = _load("harvester_tool_calls", "bin/response-harvester.py")
|
||
|
|
exc = _load("exec_constrained_tool_calls", "bin/exec-constrained.py")
|
||
|
|
env = _load("prompt_envelope_tool_calls", "bin/prompt_envelope.py")
|
||
|
|
|
||
|
|
|
||
|
|
class ParseToolCalls(unittest.TestCase):
|
||
|
|
def test_simple(self):
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls("[TOOL health.check {}]"),
|
||
|
|
[("health.check", {})],
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_exec_engine(self):
|
||
|
|
calls = harv.parse_tool_calls('[EXEC cron.runs {"limit": 3}]')
|
||
|
|
self.assertEqual(calls, [("cron.runs", {"limit": 3})])
|
||
|
|
|
||
|
|
def test_bracket_inside_json_survives(self):
|
||
|
|
text = '[TOOL box.exec {"action": "job-get", "arg": "a-b[0]"}]'
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls(text),
|
||
|
|
[("box.exec", {"action": "job-get", "arg": "a-b[0]"})],
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_nested_objects_and_arrays(self):
|
||
|
|
args = {"outer": {"inner": [1, 2, {"k": "v]w"}]}, "list": ["a", "b]c"]}
|
||
|
|
text = "[TOOL swarm.spawn %s]" % json.dumps(args)
|
||
|
|
self.assertEqual(harv.parse_tool_calls(text), [("swarm.spawn", args)])
|
||
|
|
|
||
|
|
def test_escaped_quotes_and_braces_in_strings(self):
|
||
|
|
args = {"prompt": 'say "{hi}" \\ done'}
|
||
|
|
text = "[TOOL followup.create %s]" % json.dumps(args)
|
||
|
|
op, got = harv.parse_tool_calls(text)[0]
|
||
|
|
self.assertEqual(op, "followup.create")
|
||
|
|
self.assertEqual(got["prompt"], args["prompt"])
|
||
|
|
|
||
|
|
def test_dm_shorthand(self):
|
||
|
|
text = '[DM {"to": "pip", "target": "pip tasks", "message": "hi [you]"}]'
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls(text),
|
||
|
|
[("dm.send", {"to": "pip", "target": "pip tasks", "message": "hi [you]"})],
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_dm_bare_form_skipped(self):
|
||
|
|
self.assertEqual(harv.parse_tool_calls("[DM hello pip]"), [])
|
||
|
|
|
||
|
|
def test_no_args(self):
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls("[TOOL cron.runs]"), [("cron.runs", {})]
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_legacy_raw_passthrough(self):
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls("[TOOL foo bar baz]"),
|
||
|
|
[("foo", {"raw": "bar baz"})],
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_broken_json_skipped(self):
|
||
|
|
self.assertEqual(harv.parse_tool_calls("[TOOL foo {bad}]"), [])
|
||
|
|
|
||
|
|
def test_fenced_block(self):
|
||
|
|
text = '```tool\n{"op": "health.check", "args": {}}\n```'
|
||
|
|
self.assertEqual(harv.parse_tool_calls(text), [("health.check", {})])
|
||
|
|
|
||
|
|
def test_dedupe_repeated_call(self):
|
||
|
|
text = "[TOOL health.check {}] ... [TOOL health.check {}]"
|
||
|
|
self.assertEqual(harv.parse_tool_calls(text), [("health.check", {})])
|
||
|
|
|
||
|
|
def test_native_aliases_applied(self):
|
||
|
|
text = '[TOOL subagent.spawn {"count": 1, "task": "t"}]'
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls(text),
|
||
|
|
[("swarm.spawn", {"count": 1, "task": "t"})],
|
||
|
|
)
|
||
|
|
text = '[TOOL cron.create {"kind": "runonce", "in_m": 5, "prompt": "p"}]'
|
||
|
|
op, args = harv.parse_tool_calls(text)[0]
|
||
|
|
self.assertEqual(op, "followup.create")
|
||
|
|
self.assertNotIn("kind", args)
|
||
|
|
self.assertEqual(args["in_m"], 5)
|
||
|
|
|
||
|
|
|
||
|
|
class NormalizeNativeCall(unittest.TestCase):
|
||
|
|
def test_dm_synonyms(self):
|
||
|
|
op, args = harv.normalize_native_call(
|
||
|
|
"dm", {"to": "pip", "thread": "pip tasks", "text": "hi"})
|
||
|
|
self.assertEqual(op, "dm.send")
|
||
|
|
self.assertEqual(args["message"], "hi")
|
||
|
|
self.assertEqual(args["target"], "pip tasks")
|
||
|
|
|
||
|
|
def test_box_synonyms(self):
|
||
|
|
op, args = harv.normalize_native_call("box", {"cmd": "fleet-status"})
|
||
|
|
self.assertEqual((op, args), ("box.exec", {"action": "fleet-status"}))
|
||
|
|
|
||
|
|
def test_tools_alias(self):
|
||
|
|
op, args = harv.normalize_native_call("tools", {})
|
||
|
|
self.assertEqual(op, "tools.list")
|
||
|
|
|
||
|
|
|
||
|
|
class FormatToolResult(unittest.TestCase):
|
||
|
|
def test_tools_list_grouping(self):
|
||
|
|
out = harv.format_tool_result_for_chat("tools.list", json.dumps({
|
||
|
|
"ok": True,
|
||
|
|
"ops": [
|
||
|
|
{"op": "health.check", "side_effecting": False},
|
||
|
|
{"op": "dm.send", "side_effecting": True},
|
||
|
|
],
|
||
|
|
}))
|
||
|
|
self.assertIn("2 tools", out)
|
||
|
|
self.assertIn("health.check", out)
|
||
|
|
self.assertIn("dm.send", out)
|
||
|
|
|
||
|
|
def test_box_exec_string_fenced(self):
|
||
|
|
out = harv.format_tool_result_for_chat("box.exec", "NODE UP")
|
||
|
|
self.assertIn("```", out)
|
||
|
|
self.assertIn("NODE UP", out)
|
||
|
|
|
||
|
|
def test_box_exec_string_truncated(self):
|
||
|
|
out = harv.format_tool_result_for_chat("box.exec", "x" * 2000)
|
||
|
|
self.assertIn("truncated", out)
|
||
|
|
self.assertLess(len(out), 1200)
|
||
|
|
|
||
|
|
def test_box_exec_json_dict_passthrough(self):
|
||
|
|
out = harv.format_tool_result_for_chat(
|
||
|
|
"box.exec", json.dumps({"ok": True, "nodes": []}))
|
||
|
|
self.assertIn("```", out)
|
||
|
|
self.assertIn('"nodes": []', out)
|
||
|
|
|
||
|
|
def test_box_exec_json_error(self):
|
||
|
|
out = harv.format_tool_result_for_chat(
|
||
|
|
"box.exec", json.dumps({"ok": False, "error": "BAD_NAME"}))
|
||
|
|
self.assertIn("BAD_NAME", out)
|
||
|
|
|
||
|
|
|
||
|
|
class BoxExecOp(unittest.TestCase):
|
||
|
|
def test_noarg_ok(self):
|
||
|
|
self.assertEqual(
|
||
|
|
exc._box_exec_validate({"action": "fleet-status"}),
|
||
|
|
{"action": "fleet-status"},
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_agent_key_tolerated(self):
|
||
|
|
self.assertEqual(
|
||
|
|
exc._box_exec_validate({"action": "unread", "agent": "646"}),
|
||
|
|
{"action": "unread"},
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_onearg_ok(self):
|
||
|
|
self.assertEqual(
|
||
|
|
exc._box_exec_validate({"action": "job-get", "arg": "abc-123"}),
|
||
|
|
{"action": "job-get", "arg": "abc-123"},
|
||
|
|
)
|
||
|
|
|
||
|
|
def test_rejects_unknown_action(self):
|
||
|
|
with self.assertRaises(exc.OpError):
|
||
|
|
exc._box_exec_validate({"action": "job-trigger"})
|
||
|
|
|
||
|
|
def test_rejects_side_effecting(self):
|
||
|
|
for action in ("vars-set", "job-delete", "timer-create", "md-write"):
|
||
|
|
with self.assertRaises(exc.OpError, msg=action):
|
||
|
|
exc._box_exec_validate({"action": action})
|
||
|
|
|
||
|
|
def test_rejects_excluded_idempotent(self):
|
||
|
|
for action in ("main-loop", "quality-validate", "git-diff", "job-next"):
|
||
|
|
with self.assertRaises(exc.OpError, msg=action):
|
||
|
|
exc._box_exec_validate({"action": action})
|
||
|
|
|
||
|
|
def test_rejects_bad_arg(self):
|
||
|
|
for bad in ("../x", "a b", "a;b", "", "x" * 200):
|
||
|
|
with self.assertRaises(exc.OpError, msg=bad):
|
||
|
|
exc._box_exec_validate({"action": "job-get", "arg": bad})
|
||
|
|
|
||
|
|
def test_rejects_arg_on_noarg_action(self):
|
||
|
|
with self.assertRaises(exc.OpError):
|
||
|
|
exc._box_exec_validate({"action": "fleet-status", "arg": "x"})
|
||
|
|
|
||
|
|
def test_build_argv(self):
|
||
|
|
argv = exc._box_exec_build({"action": "job-get", "arg": "abc"})
|
||
|
|
self.assertEqual(argv[-2:], ["job-get", "abc"])
|
||
|
|
self.assertTrue(argv[1].endswith("box-ctl.py"))
|
||
|
|
|
||
|
|
def test_registered_read_only(self):
|
||
|
|
self.assertIn("box.exec", exc.OPS)
|
||
|
|
self.assertFalse(exc.OPS["box.exec"]["side_effecting"])
|
||
|
|
self.assertIn("tools.list", exc.OPS)
|
||
|
|
self.assertFalse(exc.OPS["tools.list"]["side_effecting"])
|
||
|
|
|
||
|
|
def test_permissions_cover_agents(self):
|
||
|
|
for ident in ("muse", "pip", "646", "opm", "dev", "def"):
|
||
|
|
self.assertIn("box.exec", exc.PERMISSIONS[ident])
|
||
|
|
self.assertIn("tools.list", exc.PERMISSIONS[ident])
|
||
|
|
|
||
|
|
def test_list_ops_subcommand(self):
|
||
|
|
p = subprocess.run(
|
||
|
|
[sys.executable, str(REPO_ROOT / "bin" / "exec-constrained.py"),
|
||
|
|
"--list-ops"],
|
||
|
|
capture_output=True, text=True, timeout=30,
|
||
|
|
)
|
||
|
|
self.assertEqual(p.returncode, 0)
|
||
|
|
data = json.loads(p.stdout)
|
||
|
|
self.assertTrue(data["ok"])
|
||
|
|
names = {o["op"] for o in data["ops"]}
|
||
|
|
for want in ("box.exec", "tools.list", "dm.send", "swarm.spawn",
|
||
|
|
"health.check", "followup.create"):
|
||
|
|
self.assertIn(want, names)
|
||
|
|
|
||
|
|
|
||
|
|
class CanonicalToolPattern(unittest.TestCase):
|
||
|
|
def test_samples_match(self):
|
||
|
|
data = json.loads(
|
||
|
|
(REPO_ROOT / "lookup_internal" / "regex_patterns.json").read_text())
|
||
|
|
pat = data["patterns"]["tool_call"]["pattern"]
|
||
|
|
rx = re.compile(pat, re.S)
|
||
|
|
for s in data["patterns"]["tool_call"]["test_samples"]["valid"]:
|
||
|
|
self.assertIsNotNone(rx.search(s), s)
|
||
|
|
for s in data["patterns"]["tool_call"]["test_samples"]["invalid"]:
|
||
|
|
self.assertIsNone(rx.search(s), s)
|
||
|
|
|
||
|
|
def test_dm_sample_has_no_op(self):
|
||
|
|
data = json.loads(
|
||
|
|
(REPO_ROOT / "lookup_internal" / "regex_patterns.json").read_text())
|
||
|
|
rx = re.compile(data["patterns"]["tool_call"]["pattern"], re.S)
|
||
|
|
m = rx.search('[DM {"to": "pip"}]')
|
||
|
|
self.assertIsNotNone(m)
|
||
|
|
self.assertEqual(m.group("engine"), "DM")
|
||
|
|
self.assertIsNone(m.group("op"))
|
||
|
|
|
||
|
|
|
||
|
|
class EnvelopeRoundTrip(unittest.TestCase):
|
||
|
|
def test_wrap_advertises_new_verbs(self):
|
||
|
|
body = env.wrap("work-finder", "work-finder-1", "646",
|
||
|
|
"646 tasks", "Do the thing.")
|
||
|
|
for token in ("dm.send", "box.exec", "tools.list", "[DM {"):
|
||
|
|
self.assertIn(token, body)
|
||
|
|
|
||
|
|
def test_spawn_call_parses(self):
|
||
|
|
text = env.spawn_call("jid-1", "work-finder", "work")
|
||
|
|
op, args = harv.parse_tool_calls(text)[0]
|
||
|
|
self.assertEqual(op, "swarm.spawn")
|
||
|
|
self.assertIn("count", args)
|
||
|
|
self.assertIn("task", args)
|
||
|
|
|
||
|
|
def test_dm_call_parses(self):
|
||
|
|
text = env.dm_call("pip", "pip tasks", "hello [brackets] work")
|
||
|
|
self.assertEqual(
|
||
|
|
harv.parse_tool_calls(text),
|
||
|
|
[("dm.send", {"to": "pip", "target": "pip tasks",
|
||
|
|
"message": "hello [brackets] work"})],
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
unittest.main()
|