From feb353d0c32068af3b21d65b06d304f2b01b67c8 Mon Sep 17 00:00:00 2001 From: ciregenz Date: Thu, 27 Aug 2026 23:15:55 -0700 Subject: [PATCH] [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) Claude-Session: https://claude.ai/code/session_01U6zrBsUCNzpMBnov3rTVYV --- backend/apps/agents/ptc_mcp_server.py | 56 +++++++- backend/apps/agents/ptc_script_runner.py | 66 ++++++++- backend/tests/test_ptc_call_tools.py | 165 +++++++++++++++++++++++ 3 files changed, 282 insertions(+), 5 deletions(-) create mode 100644 backend/tests/test_ptc_call_tools.py diff --git a/backend/apps/agents/ptc_mcp_server.py b/backend/apps/agents/ptc_mcp_server.py index d4192ba6..5e1acd8c 100644 --- a/backend/apps/agents/ptc_mcp_server.py +++ b/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 diff --git a/backend/apps/agents/ptc_script_runner.py b/backend/apps/agents/ptc_script_runner.py index 3aa2cb7e..602eb2a0 100644 --- a/backend/apps/agents/ptc_script_runner.py +++ b/backend/apps/agents/ptc_script_runner.py @@ -8,8 +8,15 @@ Protocol, all newline-delimited JSON over the real stdin/stdout: parent -> child {"script": ""} 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, "", "exec"), scope) except BaseException as e: diff --git a/backend/tests/test_ptc_call_tools.py b/backend/tests/test_ptc_call_tools.py new file mode 100644 index 00000000..5ce6afc5 --- /dev/null +++ b/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"