box/tests/test_tool_calls.py

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