[eric] agents: a PTC script can fan independent tool calls out instead of paying latency per item (ENG-417)

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01U6zrBsUCNzpMBnov3rTVYV

backend/apps/agents/ptc_mcp_server.py

@@ -19,6 +19,13 @@ SCRIPT_TIMEOUT_S = 300.0
 MAX_TOOL_CALLS = 50
 MAX_STDOUT_BYTES = 50_000
 
+# How many of a batch's calls may be in flight at once. The sidecar already runs a thread per MCP
+# call, so this is not new concurrency; the cap is here so a 25-call fan-out cannot crowd out the
+# chat's other builtin tools while it runs. Module-global, so two scripts in one sidecar share it
+# rather than each helping themselves to a full width.
+SCRIPT_FANOUT_WIDTH = 8
+P_FANOUT_SLOTS = threading.Semaphore(SCRIPT_FANOUT_WIDTH)
+
 # Read/write-safe, ungated tools only. Everything else is invisible to scripts on purpose.
 SCRIPT_ALLOWED_TOOLS = ("WebSearch", "WebFetch", "MemoryRead", "MemoryWrite", "SettingsRead", "Skill")
 
@@ -30,9 +37,16 @@ TOOLS = [
             "the script print()s comes back to you, so bulky intermediate results never enter "
             "your context. Use this whenever a task needs 3+ tool calls whose raw outputs you "
             "would only aggregate anyway (fetch N pages and extract one fact each, search then "
-            "fetch the top hits, sweep memory). The script gets one function: "
-            "call_tool(name, args_dict) -> str, valid names: "
-            + ", ".join(SCRIPT_ALLOWED_TOOLS) + ". A failed tool raises PtcToolError (catchable). "
+            "fetch the top hits, sweep memory). Valid tool names: "
+            + ", ".join(SCRIPT_ALLOWED_TOOLS) + ". Two functions:\n"
+            "call_tools(calls) -> list of results IN THE SAME ORDER, run concurrently. Use this "
+            "whenever the calls do not depend on each other, which is most sweeps. Each call is "
+            "{'name': ..., 'args': {...}}; each result has .ok, .text and .error, and one failure "
+            "never kills the batch. Example: "
+            "for r in call_tools([{'name': 'WebFetch', 'args': {'url': u}} for u in urls]): "
+            "print(r.text if r.ok else 'skipped ' + r.error)\n"
+            "call_tool(name, args_dict) -> str, one call, for a CHAIN where each call needs the "
+            "previous result. A failed tool raises PtcToolError (catchable).\n"
             "Print ONLY your distilled findings. Budget: "
             f"{MAX_TOOL_CALLS} tool calls, {SCRIPT_TIMEOUT_S:.0f}s, printed output capped at 50KB."
         ),
@@ -41,7 +55,7 @@ TOOLS = [
             "properties": {
                 "script": {
                     "type": "string",
-                    "description": "Python source. Example: results = call_tool('WebSearch', {'query': 'x'}) then loop call_tool('WebFetch', {'url': u}) and print the aggregate.",
+                    "description": "Python source. Example: hits = call_tool('WebSearch', {'query': 'x'}) (a chain step), then call_tools([{'name': 'WebFetch', 'args': {'url': u}} for u in urls]) to fetch them all at once, and print the aggregate.",
                 },
             },
             "required": ["script"],
@@ -94,6 +108,28 @@ def p_dispatch(name: str, args: dict) -> dict:
     return {"text": p_result_text(result), "is_error": bool(result.get("isError"))}
 
 
+def p_dispatch_batch(calls: list, deadline: float) -> list:
+    """Fan a batch out and gather in CALL ORDER. One call's failure is that call's result, never the
+    batch's, which is the whole reason a script would use this instead of a loop."""
+    p_out: list = [None] * len(calls)
+
+    def p_one(i: int, call: dict) -> None:
+        with P_FANOUT_SLOTS:
+            p_out[i] = p_dispatch(str(call.get("name", "")), dict(call.get("args") or {}))
+
+    p_threads = [threading.Thread(target=p_one, args=(i, c), daemon=True) for i, c in enumerate(calls)]
+    for t in p_threads:
+        t.start()
+    for t in p_threads:
+        # The script budget is the only ceiling; joining past it would let a batch outlive the
+        # deadline the reaper is already enforcing on the child.
+        t.join(timeout=max(0.0, deadline - time.monotonic()))
+    return [
+        r if r is not None else {"text": "call did not finish inside the script budget", "is_error": True}
+        for r in p_out
+    ]
+
+
 def p_elide(text: str, cap: int = MAX_STDOUT_BYTES) -> str:
     raw = text.encode("utf-8", errors="replace")
     if len(raw) <= cap:
@@ -175,6 +211,18 @@ def handle_tool_call(tool_name: str, arguments: dict) -> dict:
                 if not out.strip():
                     return p_mcp_text(f"script printed nothing; print your findings next time.{footer}", is_error=True)
                 return p_mcp_text(out + footer)
+            batch = msg.get("calls")
+            if isinstance(batch, list):
+                p_budget = f"tool call budget ({MAX_TOOL_CALLS}) exhausted; print what you have"
+                # Per item, not per batch: the calls that fit still run, and the rest say why not.
+                p_room = max(0, MAX_TOOL_CALLS - calls_used)
+                p_run, p_over = batch[:p_room], batch[p_room:]
+                calls_used += len(batch)
+                p_results = p_dispatch_batch(p_run, deadline) if p_run else []
+                p_results += [{"text": p_budget, "is_error": True} for _ in p_over]
+                proc.stdin.write(json.dumps({"seq": msg.get("seq"), "results": p_results}) + "\n")
+                proc.stdin.flush()
+                continue
             call = msg.get("call")
             if not isinstance(call, dict):
                 continue

backend/apps/agents/ptc_script_runner.py

@@ -8,8 +8,15 @@ Protocol, all newline-delimited JSON over the real stdin/stdout:
   parent -> child   {"script": "<python source>"}
   child  -> parent  {"call": {"name": str, "args": {}}, "seq": int}
   parent -> child   {"seq": int, "text": str, "is_error": bool}
+  child  -> parent  {"calls": [{"name": str, "args": {}}, ...], "seq": int}
+  parent -> child   {"seq": int, "results": [{"text": str, "is_error": bool}, ...]}
   child  -> parent  {"done": true, "stdout": str, "calls": int, "error": str|null}
 
+A batch is ONE message and the child stays strictly request/response: the parent owns the fan-out,
+so it also owns the width cap and the call budget, and results come back in call order by
+construction. Threading the child instead would need a reply demultiplexer inside model-written
+scope, for the same wall-clock and a race surface we would have to defend forever.
+
 The script's own print() goes to an in-memory buffer (the real stdout is the RPC channel),
 capped so a runaway loop can't balloon the process. Runs with a scrubbed env and no auth
 token: every tool call is brokered by the parent sidecar, which owns the allowlist."""
@@ -30,6 +37,20 @@ class PtcToolError(Exception):
     pass
 
 
+class ToolResult:
+    """One call's outcome inside a batch. A failure is DATA here, not an exception, because one bad
+    URL in twenty must not throw away the other nineteen."""
+
+    def __init__(self, name, text, error):
+        self.name = name
+        self.text = text
+        self.error = error
+        self.ok = error is None
+
+    def __repr__(self):
+        return f"ToolResult(name={self.name!r}, ok={self.ok}, error={self.error!r})"
+
+
 def p_send(obj):
     p_rpc_out.write(json.dumps(obj) + "\n")
     p_rpc_out.flush()
@@ -59,6 +80,43 @@ def call_tool(name, args=None):
     return str(reply.get("text", ""))
 
 
+MAX_BATCH = 25
+
+
+def call_tools(calls):
+    """Run independent tool calls CONCURRENTLY and return their results in the SAME ORDER.
+
+    Each call is {"name": str, "args": dict}. Returns a list of ToolResult (.ok, .text, .error).
+    Use this when the calls do not depend on each other; keep call_tool for a chain."""
+    global p_seq, p_calls
+    if not isinstance(calls, (list, tuple)):
+        raise PtcToolError("call_tools needs a list of {'name': ..., 'args': {...}} calls")
+    p_batch = []
+    for c in calls:
+        if not isinstance(c, dict) or not isinstance(c.get("name"), str) or not c.get("name"):
+            raise PtcToolError("each call needs a non-empty 'name' and an optional 'args' dict")
+        p_batch.append({"name": c["name"], "args": dict(c.get("args") or {})})
+    if not p_batch:
+        return []
+    if len(p_batch) > MAX_BATCH:
+        raise PtcToolError(f"call_tools takes at most {MAX_BATCH} calls at once; send them in chunks")
+    p_seq += 1
+    p_calls += len(p_batch)
+    p_send({"calls": p_batch, "seq": p_seq})
+    reply = p_recv()
+    p_results = reply.get("results")
+    if not isinstance(p_results, list) or len(p_results) != len(p_batch):
+        raise PtcToolError("sidecar returned a malformed batch reply")
+    return [
+        ToolResult(
+            p_batch[i]["name"],
+            "" if r.get("is_error") else str(r.get("text", "")),
+            str(r.get("text", "tool failed")) if r.get("is_error") else None,
+        )
+        for i, r in enumerate(p_results)
+    ]
+
+
 class P_CappedBuffer(io.StringIO):
     def write(self, s):
         if self.tell() < P_STDOUT_CAP_BYTES:
@@ -72,7 +130,13 @@ def main():
     buf = P_CappedBuffer()
     sys.stdout = buf
     error = None
-    scope = {"call_tool": call_tool, "PtcToolError": PtcToolError, "__name__": "__ptc_script__"}
+    scope = {
+        "call_tool": call_tool,
+        "call_tools": call_tools,
+        "ToolResult": ToolResult,
+        "PtcToolError": PtcToolError,
+        "__name__": "__ptc_script__",
+    }
     try:
         exec(compile(script, "<tool_script>", "exec"), scope)
     except BaseException as e:

backend/tests/test_ptc_call_tools.py

@@ -0,0 +1,165 @@
+"""Pins the RunToolScript fan-out (ENG-417): independent tool calls run concurrently, results come
+back in call order, and one failure is that call's result rather than the batch's.
+
+The wall-clock assertions inject ~1s of latency on purpose. A fan-out benchmark against a local
+no-op tool measures thread overhead and "disproves" the win: 20 localhost fetches ran 36ms
+sequential vs 75ms parallel, which is why every timing case here has a real sleep in it."""
+
+import threading
+import time
+
+from backend.apps.agents import ptc_mcp_server as ptc
+
+P_LATENCY_S = 1.0
+
+
+class P_SlowCore:
+    """Every tool sleeps, the way a real fetch does. Records peak concurrency, because "did it
+    actually run in parallel" is the claim, not "did it finish sooner"."""
+
+    def __init__(self, latency=P_LATENCY_S, fail_urls=()):
+        self.P_ROUTE = {name: self for name in ptc.SCRIPT_ALLOWED_TOOLS}
+        self.latency = latency
+        self.fail_urls = set(fail_urls)
+        self.live = 0
+        self.peak = 0
+        self.lock = threading.Lock()
+
+    def p_call(self, mod, name, args):
+        with self.lock:
+            self.live += 1
+            self.peak = max(self.peak, self.live)
+        try:
+            time.sleep(self.latency)
+            url = str(args.get("url", ""))
+            if url in self.fail_urls:
+                return {"content": [{"type": "text", "text": f"404 for {url}"}], "isError": True}
+            return {"content": [{"type": "text", "text": f"BODY:{url or name}"}]}
+        finally:
+            with self.lock:
+                self.live -= 1
+
+
+def p_run(script):
+    return ptc.handle_tool_call("RunToolScript", {"script": script})
+
+
+def p_text(result):
+    return result["content"][0]["text"]
+
+
+def teardown_function(fn):
+    ptc.set_core_for_tests(None)
+
+
+P_URLS = [f"http://x/{i}" for i in range(8)]
+P_FANOUT = (
+    f"urls = {P_URLS!r}\n"
+    "for r in call_tools([{'name': 'WebFetch', 'args': {'url': u}} for u in urls]):\n"
+    "    print(r.text if r.ok else 'FAILED ' + r.error)\n"
+)
+P_SEQUENTIAL = (
+    f"urls = {P_URLS!r}\n"
+    "for u in urls:\n"
+    "    try:\n"
+    "        print(call_tool('WebFetch', {'url': u}))\n"
+    "    except PtcToolError as e:\n"
+    "        print('FAILED', e)\n"
+)
+
+
+def test_a_batch_beats_the_loop_by_about_the_fanout_width():
+    """The acceptance case: N independent calls at ~1s each finish in ~N/width, not N."""
+    ptc.set_core_for_tests(P_SlowCore())
+    t0 = time.monotonic()
+    fan = p_run(P_FANOUT)
+    p_fan_s = time.monotonic() - t0
+
+    ptc.set_core_for_tests(P_SlowCore())
+    t0 = time.monotonic()
+    seq = p_run(P_SEQUENTIAL)
+    p_seq_s = time.monotonic() - t0
+
+    assert p_seq_s > len(P_URLS) * P_LATENCY_S * 0.9, "the sequential arm did not actually serialize"
+    assert p_fan_s < p_seq_s / 3, f"fan-out {p_fan_s:.1f}s vs sequential {p_seq_s:.1f}s"
+    # Identical output bytes: the win may not cost a single character of the answer.
+    assert p_text(fan).split("[script ran")[0] == p_text(seq).split("[script ran")[0]
+
+
+def test_it_really_runs_concurrently_not_just_faster():
+    """Wall clock alone would also pass if a cache made the second arm free."""
+    core = P_SlowCore()
+    ptc.set_core_for_tests(core)
+    p_run(P_FANOUT)
+    assert core.peak > 1, "nothing overlapped"
+    assert core.peak <= ptc.SCRIPT_FANOUT_WIDTH, f"fan-out width breached: {core.peak}"
+
+
+def test_results_come_back_in_call_order_however_they_finish():
+    """Staggered latency, so the completion order is provably not the call order."""
+
+    class P_Staggered(P_SlowCore):
+        def p_call(self, mod, name, args):
+            i = int(str(args.get("url", "http://x/0")).rsplit("/", 1)[1])
+            time.sleep(0.05 * (8 - i))
+            return {"content": [{"type": "text", "text": f"BODY:{args.get('url')}"}]}
+
+    ptc.set_core_for_tests(P_Staggered())
+    out = p_text(p_run(P_FANOUT))
+    assert [ln for ln in out.splitlines() if ln.startswith("BODY:")] == [f"BODY:{u}" for u in P_URLS]
+
+
+def test_one_failure_does_not_poison_the_batch():
+    ptc.set_core_for_tests(P_SlowCore(latency=0.05, fail_urls={P_URLS[3]}))
+    out = p_text(p_run(P_FANOUT))
+    assert out.count("BODY:") == len(P_URLS) - 1
+    assert "FAILED 404 for http://x/3" in out
+
+
+def test_the_budget_is_spent_per_item_so_the_calls_that_fit_still_run():
+    """A batch over the cap used to be all-or-nothing thinking; the ones that fit run, the rest
+    say why not, and the total is still capped."""
+    ptc.set_core_for_tests(P_SlowCore(latency=0.01))
+    n = ptc.MAX_TOOL_CALLS + 4
+    out = p_text(p_run(
+        f"urls = [f'http://x/{{i}}' for i in range({n})]\n"
+        "rs = []\n"
+        "for i in range(0, len(urls), 20):\n"
+        "    rs += call_tools([{'name': 'WebFetch', 'args': {'url': u}} for u in urls[i:i+20]])\n"
+        "print('ok', sum(1 for r in rs if r.ok), 'blocked', sum(1 for r in rs if not r.ok))\n"
+    ))
+    assert f"ok {ptc.MAX_TOOL_CALLS} blocked 4" in out
+
+
+def test_an_empty_batch_is_free_and_a_huge_one_is_refused_before_the_wire():
+    ptc.set_core_for_tests(P_SlowCore(latency=0.01))
+    assert "EMPTY []" in p_text(p_run("print('EMPTY', call_tools([]))"))
+    r = p_run("call_tools([{'name': 'WebFetch', 'args': {}} for _ in range(200)])")
+    assert r.get("isError") and "at most" in p_text(r)
+
+
+def test_a_malformed_call_is_rejected_and_names_what_is_wrong():
+    ptc.set_core_for_tests(P_SlowCore(latency=0.01))
+    r = p_run("call_tools(['WebFetch'])")
+    assert r.get("isError") and "non-empty 'name'" in p_text(r)
+
+
+def test_the_allowlist_still_holds_inside_a_batch():
+    """The batch path is a second door to p_dispatch; a gate on one door only is this repo's
+    recurring defect."""
+    ptc.set_core_for_tests(P_SlowCore(latency=0.01))
+    out = p_text(p_run(
+        "r = call_tools([{'name': 'Bash', 'args': {'command': 'echo hi'}}])[0]\n"
+        "print('OK' if r.ok else 'REFUSED ' + r.error)\n"
+    ))
+    assert "REFUSED" in out and "not callable from scripts" in out
+
+
+def test_the_fan_out_does_not_extend_the_script_deadline():
+    """RunToolScript is exempt from the 25s wedge watchdog, so its own 300s budget is the only
+    ceiling; a batch may not become an unbounded wait."""
+    src = open("backend/apps/agents/ptc_mcp_server.py").read()
+    i = src.index("def p_dispatch_batch")
+    body = src[i:i + 1200]
+    assert "t.join(timeout=" in body and "deadline - time.monotonic()" in body
+    assert "SCRIPT_TIMEOUT_S" not in body, "the batch must ride the caller's deadline, not a fresh one"