feat(box): enable real-time agent tool triggers and capabilities expansion

bin/response-harvester.py

@@ -28,6 +28,7 @@ import hashlib
 import json
 import os
 import re
+import ssl
 import subprocess
 import sys
 import time
@@ -105,6 +106,86 @@ def iter_verb_markers(text):
         yield m.group(1), m.group(2).strip()
 
 
+def parse_tool_calls(text):
+    """
+    Extract structured tool/exec calls from assistant messages.
+    Supports:
+      1. [TOOL <op> <json_args>] or [EXEC <op> <json_args>]
+      2. ```box / ```tool / ```exec JSON blocks
+    """
+    calls = []
+    for m in re.finditer(r"\[(?:TOOL|EXEC)\s+([a-zA-Z0-9_.-]+)(?:\s+(.*?))?\]", text or ""):
+        op = m.group(1).strip()
+        raw_args = (m.group(2) or "").strip()
+        args = {}
+        if raw_args:
+            try:
+                args = json.loads(raw_args)
+            except Exception:
+                args = {"raw": raw_args}
+        calls.append((op, args))
+
+    for m in re.finditer(r"```(?:box|tool|exec)\s*\n(.*?)```", text or "", re.DOTALL):
+        block = m.group(1).strip()
+        try:
+            d = json.loads(block)
+            if isinstance(d, dict) and "op" in d:
+                calls.append((d["op"], d.get("args", {})))
+        except Exception:
+            pass
+    return calls
+
+
+def execute_agent_tool(agent, op, args, timeout=30):
+    """
+    Execute tool call via local exec-constrained HTTPS daemon over Tailscale.
+    Returns (ok, result_or_error_string).
+    """
+    token = None
+    token_file = os.path.expanduser(f"~/.exec-tokens/{agent}")
+    if not os.path.exists(token_file):
+        token_file = os.path.expanduser("~/.exec-server-token")
+    if os.path.exists(token_file):
+        try:
+            with open(token_file) as f:
+                token = f.read().strip()
+        except Exception:
+            pass
+
+    url = "https://100.123.153.75:8444/exec"
+    payload = json.dumps({"op": op, "args": args}).encode("utf-8")
+    headers = {
+        "Content-Type": "application/json",
+        "User-Agent": f"Box-Harvester/{agent}",
+    }
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+
+    ctx = ssl.create_default_context()
+    ctx.check_hostname = False
+    ctx.verify_mode = ssl.CERT_NONE
+
+    req = urllib.request.Request(url, data=payload, headers=headers, method="POST")
+    try:
+        with urllib.request.urlopen(req, context=ctx, timeout=timeout) as resp:
+            data = resp.read().decode("utf-8")
+            res = json.loads(data)
+            if res.get("rc") == 0:
+                return True, res.get("stdout", "").strip() or "OK"
+            elif "stdout" in res:
+                return False, (res.get("stdout") or res.get("stderr") or "failed").strip()
+            elif "error" in res:
+                return False, res["error"]
+            return True, json.dumps(res)
+    except urllib.error.HTTPError as he:
+        try:
+            err_body = he.read().decode("utf-8")
+            return False, f"HTTP {he.code}: {err_body}"
+        except Exception:
+            return False, f"HTTP {he.code}"
+    except Exception as e:
+        return False, str(e)
+
 
 def utcnow():
     return datetime.now(timezone.utc).isoformat()
@@ -416,6 +497,35 @@ def process_messages(raw_messages, agent, thread_id, thread_name, last_wm, follo
         if author == "assistant":
             markers = list(iter_result_markers(text))
             verbs = list(iter_verb_markers(text))
+            # Check for structured tool / command calls from the agent
+            tool_calls = parse_tool_calls(text)
+            if tool_calls and not dry_run:
+                for op, args in tool_calls:
+                    print(f"[{agent}] Executing tool '{op}' from message {mid[:8]} in thread {thread_name or thread_id[:8]}")
+                    t_ok, t_res = execute_agent_tool(agent, op, args)
+                    tool_record = {
+                        "ts": utcnow(),
+                        "type": "tool_exec",
+                        "agent": agent,
+                        "thread_id": thread_id,
+                        "op": op,
+                        "args": args,
+                        "success": t_ok,
+                        "result_snippet": str(t_res)[:300],
+                        "msg_id": mid,
+                    }
+                    append_jsonl(JOB_LOG, tool_record)
+
+                    # Post tool output back to the originating thread via fast headless gateway
+                    if thread_id and re.fullmatch(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", thread_id.lower()):
+                        try:
+                            import muse_hybrid
+                            status_label = "OUTPUT" if t_ok else "ERROR"
+                            resp_text = f"[{status_label} {op}]\n{t_res}"
+                            muse_hybrid.send_message(agent, resp_text, thread_id=thread_id, wait=0)
+                        except Exception as te:
+                            sys.stderr.write(f"warning: failed to post tool response back to thread: {te}\n")
+
             if markers or verbs:
                 for job_id, result_text in markers:
                     is_fail = is_fail_result(result_text)