"""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()