From bf064ac896ae05c202fbdcac6330c3c04126c3ae Mon Sep 17 00:00:00 2001 From: Elias Posen Date: Mon, 27 Jul 2026 13:28:28 -0400 Subject: [PATCH 1/2] reduce verbose info logs --- CHANGELOG.md | 2 ++ crates/pctx_code_mode/src/code_mode.rs | 21 ++++++++----- crates/pctx_registry/src/registry.rs | 41 ++++++++++++++++---------- 3 files changed, 41 insertions(+), 23 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f0b61c6..6fccf0c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Moved full MCP tool call / callback inputs and outputs, and `execute_typescript`/`execute_bash` code and output, from INFO to DEBUG logs to declutter traces. Added DEBUG logs describing tool result shape (structured content, JSON parse success/fallback). + ### Fixed - Declared `rmcp` minimum raised from 1.2.0 to 1.8.0, the version the code diff --git a/crates/pctx_code_mode/src/code_mode.rs b/crates/pctx_code_mode/src/code_mode.rs index 931f422..b5b097b 100644 --- a/crates/pctx_code_mode/src/code_mode.rs +++ b/crates/pctx_code_mode/src/code_mode.rs @@ -479,7 +479,7 @@ impl CodeMode { } /// Execute bash commands directly in the virtual filesystem - #[instrument(skip(self), ret(Display), err)] + #[instrument(skip(self), err)] pub async fn execute_bash(&self, command: &str) -> Result { debug!(command = %command, "Executing bash command"); @@ -544,15 +544,18 @@ export default result;"#, warn!("Bash execution failed with exit code {exit_code}: {stderr}"); } - Ok(ExecuteBashOutput { + let output = ExecuteBashOutput { exit_code, stdout, stderr, - }) + }; + debug!("Bash execution result:\n{output}"); + + Ok(output) } /// Execute TypeScript code with access to registered tools and virtual filesystem - #[instrument(skip(self, registry), ret(Display), err)] + #[instrument(skip(self, registry, code), err)] pub async fn execute_typescript( &self, code: &str, @@ -664,7 +667,7 @@ export default result;"#, } }; - debug!(to_execute = %to_execute, "Executing TypeScript in sandbox"); + debug!("Executing TypeScript in sandbox:\n{to_execute}"); let execution_res = pctx_executor::execute( &to_execute, @@ -678,13 +681,17 @@ export default result;"#, warn!("TypeScript execution failed: {:?}", execution_res.stderr); } - Ok(ExecuteTypescriptOutput { + let output = ExecuteTypescriptOutput { success: execution_res.success, stdout: execution_res.stdout, stderr: execution_res.stderr, output: execution_res.output, registry: execution_res.registry, trace: execution_res.trace, - }) + }; + + debug!("TypeScript execution result:\n{output}"); + + Ok(output) } } diff --git a/crates/pctx_registry/src/registry.rs b/crates/pctx_registry/src/registry.rs index 1e7919b..ec52907 100644 --- a/crates/pctx_registry/src/registry.rs +++ b/crates/pctx_registry/src/registry.rs @@ -15,7 +15,7 @@ use std::{ sync::{Arc, RwLock}, time::SystemTime, }; -use tracing::{debug, info, instrument, warn}; +use tracing::{debug, instrument, warn}; pub type CallbackFn = Arc< dyn Fn( @@ -237,13 +237,7 @@ impl PctxRegistry { /// /// This function will return an error if an action by the provided id doesn't exist /// or if the action itself fails - #[instrument( - name = "invoke_registry_action", - skip_all, - fields(id=id, args = json!(args).to_string()), - ret(Display), - err - )] + #[instrument(name = "invoke_registry_action", skip_all, fields(id = id), err)] pub async fn invoke( &self, id: &str, @@ -333,25 +327,40 @@ impl PctxRegistry { } // Prefer structuredContent if available, otherwise use content array - let has_structured = tool_result.structured_content.is_some(); let val = if let Some(structured) = tool_result.structured_content { + debug!(tool = %mcp_id.id(), "tool result: using structured content"); structured } else if let Some(RawContent::Text(text_content)) = tool_result.content.first().map(|a| &**a) { // Try to parse as JSON, fallback to string value - serde_json::from_str(&text_content.text) - .or_else(|_| Ok(serde_json::Value::String(text_content.text.clone()))) - .map_err(|e: serde_json::Error| { - RegistryError::ToolCall(format!("Failed to parse content: {e}")) - })? + match serde_json::from_str(&text_content.text) { + Ok(json) => { + debug!( + tool = %mcp_id.id(), + "tool result: parsed text content as JSON" + ); + json + } + Err(e) => { + debug!( + tool = %mcp_id.id(), + error = %e, + "tool result: text content is not JSON, using raw string" + ); + serde_json::Value::String(text_content.text.clone()) + } + } } else { // Return the whole content array as JSON + debug!( + tool = %mcp_id.id(), + content_len = tool_result.content.len(), + "tool result: no structured or text content, using raw content array" + ); json!(tool_result.content) }; - info!(structured_content = has_structured, result =? &val, "Tool result"); - Ok(val) })(); From c8ea3a91bc08f581d5b4cd632c81a5069731fde0 Mon Sep 17 00:00:00 2001 From: Elias Posen Date: Mon, 27 Jul 2026 14:47:14 -0400 Subject: [PATCH 2/2] fix concurrency --- pctx-py/CHANGELOG.md | 8 + pctx-py/src/pctx_client/_websocket_client.py | 36 +++- pctx-py/tests/test_integration.py | 150 +++++++++++++ pctx-py/uv.lock | 212 +++++++++---------- plans/tool-call-concurrency-controls.md | 66 ++++++ 5 files changed, 362 insertions(+), 110 deletions(-) create mode 100644 plans/tool-call-concurrency-controls.md diff --git a/pctx-py/CHANGELOG.md b/pctx-py/CHANGELOG.md index 3281618..aa2ae38 100644 --- a/pctx-py/CHANGELOG.md +++ b/pctx-py/CHANGELOG.md @@ -17,6 +17,14 @@ For changes to the underlying Rust crates and CLI, see the ### Fixed +- Tool calls now run concurrently. Each request is handled in its own task + rather than awaited inside the WebSocket read loop, so code fanning out with + `Promise.all` takes the time of its slowest call instead of the sum of all + of them. +- Sync tools run on a worker thread instead of blocking the event loop, so one + slow sync tool no longer stalls the calls beside it. Their bodies now execute + off the main thread, so anything they share must be thread-safe. + ## [v0.4.4] - 2026-07-22 ### Fixed diff --git a/pctx-py/src/pctx_client/_websocket_client.py b/pctx-py/src/pctx_client/_websocket_client.py index 29c350e..6926fc1 100644 --- a/pctx-py/src/pctx_client/_websocket_client.py +++ b/pctx-py/src/pctx_client/_websocket_client.py @@ -68,6 +68,10 @@ def __init__( self._headers = headers or {} self._pending_executions: dict[str | int, asyncio.Future] = {} self._request_counter = 0 + self._message_handler_task: asyncio.Task | None = None + # In-flight tool executions. Held strongly because asyncio only keeps + # weak references to tasks, and cancelled as a group on disconnect. + self._tool_tasks: set[asyncio.Task] = set() async def _connect(self, code_mode_session: str): """ @@ -100,6 +104,10 @@ async def _disconnect(self): if self._message_handler_task: self._message_handler_task.cancel() + for task in self._tool_tasks: + task.cancel() + self._tool_tasks.clear() + if self.ws: await self.ws.close() self.ws = None @@ -192,8 +200,14 @@ async def _handle_messages(self): message: WebSocketMessage = adapter.validate_json(message_data) if isinstance(message, ExecuteToolRequest): - res = await self._handle_execute_tool(message) - await self._send(res) + # Run the tool in its own task so this loop stays free to + # read the next message. Awaiting it here would serialize + # every tool call the server dispatches, so code that fans + # out with `Promise.all` would take the sum of its calls + # rather than the slowest one. + task = asyncio.create_task(self._execute_tool(message)) + self._tool_tasks.add(task) + task.add_done_callback(self._tool_tasks.discard) elif isinstance(message, ExecuteCodeResponse): future = self._pending_executions.get(message.id) if future is not None: @@ -216,6 +230,16 @@ async def _handle_messages(self): except Exception as e: print(f"Message handler error: {e}") + async def _execute_tool(self, req: ExecuteToolRequest): + """Run one tool request and send its response back to the server.""" + try: + res = await self._handle_execute_tool(req) + await self._send(res) + except asyncio.CancelledError: + raise + except Exception as e: + print(f"Error executing tool: {e}") + async def _handle_execute_tool( self, req: ExecuteToolRequest ) -> ExecuteToolResponse | JsonRpcError: @@ -240,10 +264,14 @@ async def _handle_execute_tool( args = req.params.args or {} try: if isinstance(tool, Tool): + # Sync tools go to a worker thread. Calling one inline would + # block the event loop for its whole duration, which stalls + # every other in-flight tool call behind it -- and with it the + # loop reading further requests off the WebSocket. if tool.input_schema is None: - output = tool.invoke() + output = await asyncio.to_thread(tool.invoke) else: - output = tool.invoke(**args) + output = await asyncio.to_thread(tool.invoke, **args) else: if tool.input_schema is None: output = await tool.ainvoke() diff --git a/pctx-py/tests/test_integration.py b/pctx-py/tests/test_integration.py index afa21ff..efbe796 100644 --- a/pctx-py/tests/test_integration.py +++ b/pctx-py/tests/test_integration.py @@ -1,5 +1,8 @@ """Integration tests for pctx code mode against a running server""" +import asyncio +import threading +import time from datetime import datetime import pytest @@ -796,3 +799,150 @@ def format_result(value: int, label: str) -> str: "Please ensure the pctx server is running.\n" "Start the server with: pctx server start" ) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_concurrent_async_tool_calls_run_in_parallel(): + """Tools fanned out with `Promise.all` must execute concurrently. + + The client used to await each tool request inside its WebSocket read loop, + so a batch of N calls took the sum of their durations instead of the + slowest one. Assert on observed overlap rather than wall time alone, so + the test fails on serialization rather than on a slow machine. + """ + sleep_secs = 0.5 + calls = 4 + + in_flight = 0 + max_in_flight = 0 + + @tool + async def slow_echo(value: int) -> int: + """Sleep briefly, then echo the value back""" + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + await asyncio.sleep(sleep_secs) + return value + finally: + in_flight -= 1 + + try: + async with Pctx(tools=[slow_echo], execute_timeout=60) as pctx: + code = """ + async function run() { + const values = await Promise.all([ + Tools.slowEcho({ value: 1 }), + Tools.slowEcho({ value: 2 }), + Tools.slowEcho({ value: 3 }), + Tools.slowEcho({ value: 4 }), + ]); + return { values }; + } + """ + + start = time.perf_counter() + output = await pctx.execute_typescript(code) + elapsed = time.perf_counter() - start + + assert output.success, f"Execution should succeed, got: {output.stderr}" + assert output.output is not None, "Execution should return output" + assert output.output.get("values") == [1, 2, 3, 4], ( + f"Expected all four calls to return, got: {output.output}" + ) + + assert max_in_flight == calls, ( + f"All {calls} tool calls should be in flight at once, " + f"peaked at {max_in_flight} -- the client is serializing them" + ) + + # Serialized dispatch takes calls * sleep_secs; concurrent dispatch + # takes ~sleep_secs. Half way between the two is a wide enough + # margin to absorb session setup and round-trip overhead. + serial_secs = calls * sleep_secs + assert elapsed < serial_secs / 2, ( + f"Concurrent calls took {elapsed:.2f}s; serialized dispatch " + f"would take ~{serial_secs:.2f}s" + ) + except ConnectionError: + pytest.fail( + "Failed to connect to pctx server at http://localhost:8080.\n" + "Please ensure the pctx server is running.\n" + "Start the server with: pctx server start" + ) + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_concurrent_sync_tool_calls_run_in_parallel(): + """Sync tools fanned out with `Promise.all` must also run concurrently. + + A sync tool body blocks whatever thread it runs on, so calling it inline + on the event loop would stall every other in-flight call behind it. They + run on worker threads instead -- hence the blocking `time.sleep` here, and + the lock around the counters, which the tool bodies touch off-thread. + """ + sleep_secs = 0.5 + calls = 4 + + lock = threading.Lock() + in_flight = 0 + max_in_flight = 0 + + @tool + def slow_echo_sync(value: int) -> int: + """Block briefly, then echo the value back""" + nonlocal in_flight, max_in_flight + with lock: + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + try: + time.sleep(sleep_secs) + return value + finally: + with lock: + in_flight -= 1 + + try: + async with Pctx(tools=[slow_echo_sync], execute_timeout=60) as pctx: + code = """ + async function run() { + const values = await Promise.all([ + Tools.slowEchoSync({ value: 1 }), + Tools.slowEchoSync({ value: 2 }), + Tools.slowEchoSync({ value: 3 }), + Tools.slowEchoSync({ value: 4 }), + ]); + return { values }; + } + """ + + start = time.perf_counter() + output = await pctx.execute_typescript(code) + elapsed = time.perf_counter() - start + + assert output.success, f"Execution should succeed, got: {output.stderr}" + assert output.output is not None, "Execution should return output" + assert output.output.get("values") == [1, 2, 3, 4], ( + f"Expected all four calls to return, got: {output.output}" + ) + + assert max_in_flight == calls, ( + f"All {calls} tool calls should be in flight at once, " + f"peaked at {max_in_flight} -- sync tools are blocking the " + f"event loop instead of running on worker threads" + ) + + serial_secs = calls * sleep_secs + assert elapsed < serial_secs / 2, ( + f"Concurrent calls took {elapsed:.2f}s; serialized dispatch " + f"would take ~{serial_secs:.2f}s" + ) + except ConnectionError: + pytest.fail( + "Failed to connect to pctx server at http://localhost:8080.\n" + "Please ensure the pctx server is running.\n" + "Start the server with: pctx server start" + ) diff --git a/pctx-py/uv.lock b/pctx-py/uv.lock index 2751b3e..4592113 100644 --- a/pctx-py/uv.lock +++ b/pctx-py/uv.lock @@ -1785,17 +1785,17 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "colorama", marker = "sys_platform == 'win32'" }, - { name = "decorator" }, - { name = "exceptiongroup" }, - { name = "jedi" }, - { name = "matplotlib-inline" }, - { name = "pexpect", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, - { name = "prompt-toolkit" }, - { name = "pygments" }, - { name = "stack-data" }, - { name = "traitlets" }, - { name = "typing-extensions" }, + { name = "colorama", marker = "python_full_version < '3.11' and sys_platform == 'win32'" }, + { name = "decorator", marker = "python_full_version < '3.11'" }, + { name = "exceptiongroup", marker = "python_full_version < '3.11'" }, + { name = "jedi", marker = "python_full_version < '3.11'" }, + { name = "matplotlib-inline", marker = "python_full_version < '3.11'" }, + { name = "pexpect", marker = "python_full_version < '3.11' and sys_platform != 'emscripten' and sys_platform != 'win32'" }, + { name = "prompt-toolkit", marker = "python_full_version < '3.11'" }, + { name = "pygments", marker = "python_full_version < '3.11'" }, + { name = "stack-data", marker = "python_full_version < '3.11'" }, + { name = "traitlets", marker = "python_full_version < '3.11'" }, + { name = "typing-extensions", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e5/61/1810830e8b93c72dcd3c0f150c80a00c3deb229562d9423807ec92c3a539/ipython-8.38.0.tar.gz", hash = "sha256:9cfea8c903ce0867cc2f23199ed8545eb741f3a69420bfcf3743ad1cec856d39", size = 5513996, upload-time = "2026-01-05T10:59:06.901Z" } wheels = [ @@ -1815,17 +1815,17 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "colorama", marker = "sys_platform == 'win32'" }, - { name = "decorator" }, - { name = "ipython-pygments-lexers" }, - { name = "jedi" }, - { name = "matplotlib-inline" }, - { name = "pexpect", marker = "sys_platform != 'emscripten' and sys_platform != 'win32'" }, - { name = "prompt-toolkit" }, - { name = "pygments" }, - { name = "stack-data" }, - { name = "traitlets" }, - { name = "typing-extensions", marker = "python_full_version < '3.12'" }, + { name = "colorama", marker = "python_full_version >= '3.11' and sys_platform == 'win32'" }, + { name = "decorator", marker = "python_full_version >= '3.11'" }, + { name = "ipython-pygments-lexers", marker = "python_full_version >= '3.11'" }, + { name = "jedi", marker = "python_full_version >= '3.11'" }, + { name = "matplotlib-inline", marker = "python_full_version >= '3.11'" }, + { name = "pexpect", marker = "python_full_version >= '3.11' and sys_platform != 'emscripten' and sys_platform != 'win32'" }, + { name = "prompt-toolkit", marker = "python_full_version >= '3.11'" }, + { name = "pygments", marker = "python_full_version >= '3.11'" }, + { name = "stack-data", marker = "python_full_version >= '3.11'" }, + { name = "traitlets", marker = "python_full_version >= '3.11'" }, + { name = "typing-extensions", marker = "python_full_version == '3.11.*'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a6/60/2111715ea11f39b1535bed6024b7dec7918b71e5e5d30855a5b503056b50/ipython-9.10.0.tar.gz", hash = "sha256:cd9e656be97618a0676d058134cd44e6dc7012c0e5cb36a9ce96a8c904adaf77", size = 4426526, upload-time = "2026-02-02T10:00:33.594Z" } wheels = [ @@ -1837,7 +1837,7 @@ name = "ipython-pygments-lexers" version = "1.1.1" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "pygments" }, + { name = "pygments", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/ef/4c/5dd1d8af08107f88c7f741ead7a40854b8ac24ddf9ae850afbcf698aa552/ipython_pygments_lexers-1.1.1.tar.gz", hash = "sha256:09c0138009e56b6854f9535736f4171d855c8c08a563a0dcd8022f78355c7e81", size = 8393, upload-time = "2025-01-17T11:24:34.505Z" } wheels = [ @@ -2371,7 +2371,7 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "mdurl" }, + { name = "mdurl", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/38/71/3b932df36c1a044d397a1f92d1cf91ee0a503d91e470cbd670aa66b07ed0/markdown-it-py-3.0.0.tar.gz", hash = "sha256:e3f60a94fa066dc52ec76661e37c851cb232d92f9886b15cb560aaada2df8feb", size = 74596, upload-time = "2023-06-03T06:41:14.443Z" } wheels = [ @@ -2391,7 +2391,7 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "mdurl" }, + { name = "mdurl", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/5b/f5/4ec618ed16cc4f8fb3b701563655a69816155e79e24a17b651541804721d/markdown_it_py-4.0.0.tar.gz", hash = "sha256:cb0a2b4aa34f932c007117b194e945bd74e0ec24133ceb5bac59009cda1cb9f3", size = 73070, upload-time = "2025-08-11T12:57:52.854Z" } wheels = [ @@ -2837,12 +2837,12 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" } }, - { name = "jinja2" }, - { name = "markdown-it-py", version = "3.0.0", source = { registry = "https://pypi.org/simple" } }, - { name = "mdit-py-plugins" }, - { name = "pyyaml" }, - { name = "sphinx", version = "8.1.3", source = { registry = "https://pypi.org/simple" } }, + { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "jinja2", marker = "python_full_version < '3.11'" }, + { name = "markdown-it-py", version = "3.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "mdit-py-plugins", marker = "python_full_version < '3.11'" }, + { name = "pyyaml", marker = "python_full_version < '3.11'" }, + { name = "sphinx", version = "8.1.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/66/a5/9626ba4f73555b3735ad86247a8077d4603aa8628537687c839ab08bfe44/myst_parser-4.0.1.tar.gz", hash = "sha256:5cfea715e4f3574138aecbf7d54132296bfd72bb614d31168f48c477a830a7c4", size = 93985, upload-time = "2025-02-12T10:53:03.833Z" } wheels = [ @@ -2862,12 +2862,12 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" } }, - { name = "jinja2" }, - { name = "markdown-it-py", version = "4.0.0", source = { registry = "https://pypi.org/simple" } }, - { name = "mdit-py-plugins" }, - { name = "pyyaml" }, - { name = "sphinx", version = "9.0.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, + { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "jinja2", marker = "python_full_version >= '3.11'" }, + { name = "markdown-it-py", version = "4.0.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "mdit-py-plugins", marker = "python_full_version >= '3.11'" }, + { name = "pyyaml", marker = "python_full_version >= '3.11'" }, + { name = "sphinx", version = "9.0.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" }, { name = "sphinx", version = "9.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/33/fa/7b45eef11b7971f0beb29d27b7bfe0d747d063aa29e170d9edd004733c8a/myst_parser-5.0.0.tar.gz", hash = "sha256:f6f231452c56e8baa662cc352c548158f6a16fcbd6e3800fc594978002b94f3a", size = 98535, upload-time = "2026-01-15T09:08:18.036Z" } @@ -3459,7 +3459,7 @@ wheels = [ [[package]] name = "pctx-client" -version = "0.4.3" +version = "0.4.4" source = { editable = "." } dependencies = [ { name = "docstring-parser" }, @@ -3580,7 +3580,7 @@ name = "pexpect" version = "4.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "ptyprocess" }, + { name = "ptyprocess", marker = "python_full_version < '3.13' or sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/92/cc564bf6381ff43ce1f4d06852fc19a2f11d180f23dc32d9588bee2f149d/pexpect-4.9.0.tar.gz", hash = "sha256:ee7d41123f3c9911050ea2c2dac107568dc43b2d3b0c7557a33212c398ead30f", size = 166450, upload-time = "2023-11-25T09:07:26.339Z" } wheels = [ @@ -5056,7 +5056,7 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" } }, + { name = "numpy", version = "2.2.6", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/0f/37/6964b830433e654ec7485e45a00fc9a27cf868d622838f6b6d9c5ec0d532/scipy-1.15.3.tar.gz", hash = "sha256:eae3cf522bc7df64b42cad3925c876e1b0b6c35c1337c93e12c0f366f55b0eaf", size = 59419214, upload-time = "2025-05-08T16:13:05.955Z" } wheels = [ @@ -5120,7 +5120,7 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "numpy", version = "2.4.2", source = { registry = "https://pypi.org/simple" } }, + { name = "numpy", version = "2.4.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/56/3e/9cca699f3486ce6bc12ff46dc2031f1ec8eb9ccc9a320fdaf925f1417426/scipy-1.17.0.tar.gz", hash = "sha256:2591060c8e648d8b96439e111ac41fd8342fdeff1876be2e19dea3fe8930454e", size = 30396830, upload-time = "2026-01-10T21:34:23.009Z" } wheels = [ @@ -5191,8 +5191,8 @@ name = "secretstorage" version = "3.5.0" source = { registry = "https://pypi.org/simple" } dependencies = [ - { name = "cryptography" }, - { name = "jeepney" }, + { name = "cryptography", marker = "python_full_version < '3.13' or sys_platform != 'win32'" }, + { name = "jeepney", marker = "python_full_version < '3.13' or sys_platform != 'win32'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/1c/03/e834bcd866f2f8a49a85eaff47340affa3bfa391ee9912a952a1faa68c7b/secretstorage-3.5.0.tar.gz", hash = "sha256:f04b8e4689cbce351744d5537bf6b1329c6fc68f91fa666f60a380edddcd11be", size = 19884, upload-time = "2025-11-23T19:02:53.191Z" } wheels = [ @@ -5243,23 +5243,23 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "alabaster" }, - { name = "babel" }, - { name = "colorama", marker = "sys_platform == 'win32'" }, - { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" } }, - { name = "imagesize" }, - { name = "jinja2" }, - { name = "packaging" }, - { name = "pygments" }, - { name = "requests" }, - { name = "snowballstemmer" }, - { name = "sphinxcontrib-applehelp" }, - { name = "sphinxcontrib-devhelp" }, - { name = "sphinxcontrib-htmlhelp" }, - { name = "sphinxcontrib-jsmath" }, - { name = "sphinxcontrib-qthelp" }, - { name = "sphinxcontrib-serializinghtml" }, - { name = "tomli" }, + { name = "alabaster", marker = "python_full_version < '3.11'" }, + { name = "babel", marker = "python_full_version < '3.11'" }, + { name = "colorama", marker = "python_full_version < '3.11' and sys_platform == 'win32'" }, + { name = "docutils", version = "0.21.2", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "imagesize", marker = "python_full_version < '3.11'" }, + { name = "jinja2", marker = "python_full_version < '3.11'" }, + { name = "packaging", marker = "python_full_version < '3.11'" }, + { name = "pygments", marker = "python_full_version < '3.11'" }, + { name = "requests", marker = "python_full_version < '3.11'" }, + { name = "snowballstemmer", marker = "python_full_version < '3.11'" }, + { name = "sphinxcontrib-applehelp", marker = "python_full_version < '3.11'" }, + { name = "sphinxcontrib-devhelp", marker = "python_full_version < '3.11'" }, + { name = "sphinxcontrib-htmlhelp", marker = "python_full_version < '3.11'" }, + { name = "sphinxcontrib-jsmath", marker = "python_full_version < '3.11'" }, + { name = "sphinxcontrib-qthelp", marker = "python_full_version < '3.11'" }, + { name = "sphinxcontrib-serializinghtml", marker = "python_full_version < '3.11'" }, + { name = "tomli", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/6f/6d/be0b61178fe2cdcb67e2a92fc9ebb488e3c51c4f74a36a7824c0adf23425/sphinx-8.1.3.tar.gz", hash = "sha256:43c1911eecb0d3e161ad78611bc905d1ad0e523e4ddc202a58a821773dc4c927", size = 8184611, upload-time = "2024-10-13T20:27:13.93Z" } wheels = [ @@ -5274,23 +5274,23 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "alabaster" }, - { name = "babel" }, - { name = "colorama", marker = "sys_platform == 'win32'" }, - { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" } }, - { name = "imagesize" }, - { name = "jinja2" }, - { name = "packaging" }, - { name = "pygments" }, - { name = "requests" }, - { name = "roman-numerals" }, - { name = "snowballstemmer" }, - { name = "sphinxcontrib-applehelp" }, - { name = "sphinxcontrib-devhelp" }, - { name = "sphinxcontrib-htmlhelp" }, - { name = "sphinxcontrib-jsmath" }, - { name = "sphinxcontrib-qthelp" }, - { name = "sphinxcontrib-serializinghtml" }, + { name = "alabaster", marker = "python_full_version == '3.11.*'" }, + { name = "babel", marker = "python_full_version == '3.11.*'" }, + { name = "colorama", marker = "python_full_version == '3.11.*' and sys_platform == 'win32'" }, + { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" }, + { name = "imagesize", marker = "python_full_version == '3.11.*'" }, + { name = "jinja2", marker = "python_full_version == '3.11.*'" }, + { name = "packaging", marker = "python_full_version == '3.11.*'" }, + { name = "pygments", marker = "python_full_version == '3.11.*'" }, + { name = "requests", marker = "python_full_version == '3.11.*'" }, + { name = "roman-numerals", marker = "python_full_version == '3.11.*'" }, + { name = "snowballstemmer", marker = "python_full_version == '3.11.*'" }, + { name = "sphinxcontrib-applehelp", marker = "python_full_version == '3.11.*'" }, + { name = "sphinxcontrib-devhelp", marker = "python_full_version == '3.11.*'" }, + { name = "sphinxcontrib-htmlhelp", marker = "python_full_version == '3.11.*'" }, + { name = "sphinxcontrib-jsmath", marker = "python_full_version == '3.11.*'" }, + { name = "sphinxcontrib-qthelp", marker = "python_full_version == '3.11.*'" }, + { name = "sphinxcontrib-serializinghtml", marker = "python_full_version == '3.11.*'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/42/50/a8c6ccc36d5eacdfd7913ddccd15a9cee03ecafc5ee2bc40e1f168d85022/sphinx-9.0.4.tar.gz", hash = "sha256:594ef59d042972abbc581d8baa577404abe4e6c3b04ef61bd7fc2acbd51f3fa3", size = 8710502, upload-time = "2025-12-04T07:45:27.343Z" } wheels = [ @@ -5309,23 +5309,23 @@ resolution-markers = [ "python_full_version == '3.12.*'", ] dependencies = [ - { name = "alabaster" }, - { name = "babel" }, - { name = "colorama", marker = "sys_platform == 'win32'" }, - { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" } }, - { name = "imagesize" }, - { name = "jinja2" }, - { name = "packaging" }, - { name = "pygments" }, - { name = "requests" }, - { name = "roman-numerals" }, - { name = "snowballstemmer" }, - { name = "sphinxcontrib-applehelp" }, - { name = "sphinxcontrib-devhelp" }, - { name = "sphinxcontrib-htmlhelp" }, - { name = "sphinxcontrib-jsmath" }, - { name = "sphinxcontrib-qthelp" }, - { name = "sphinxcontrib-serializinghtml" }, + { name = "alabaster", marker = "python_full_version >= '3.12'" }, + { name = "babel", marker = "python_full_version >= '3.12'" }, + { name = "colorama", marker = "python_full_version >= '3.12' and sys_platform == 'win32'" }, + { name = "docutils", version = "0.22.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, + { name = "imagesize", marker = "python_full_version >= '3.12'" }, + { name = "jinja2", marker = "python_full_version >= '3.12'" }, + { name = "packaging", marker = "python_full_version >= '3.12'" }, + { name = "pygments", marker = "python_full_version >= '3.12'" }, + { name = "requests", marker = "python_full_version >= '3.12'" }, + { name = "roman-numerals", marker = "python_full_version >= '3.12'" }, + { name = "snowballstemmer", marker = "python_full_version >= '3.12'" }, + { name = "sphinxcontrib-applehelp", marker = "python_full_version >= '3.12'" }, + { name = "sphinxcontrib-devhelp", marker = "python_full_version >= '3.12'" }, + { name = "sphinxcontrib-htmlhelp", marker = "python_full_version >= '3.12'" }, + { name = "sphinxcontrib-jsmath", marker = "python_full_version >= '3.12'" }, + { name = "sphinxcontrib-qthelp", marker = "python_full_version >= '3.12'" }, + { name = "sphinxcontrib-serializinghtml", marker = "python_full_version >= '3.12'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/cd/bd/f08eb0f4eed5c83f1ba2a3bd18f7745a2b1525fad70660a1c00224ec468a/sphinx-9.1.0.tar.gz", hash = "sha256:7741722357dd75f8190766926071fed3bdc211c74dd2d7d4df5404da95930ddb", size = 8718324, upload-time = "2025-12-31T15:09:27.646Z" } wheels = [ @@ -5340,12 +5340,12 @@ resolution-markers = [ "python_full_version < '3.11'", ] dependencies = [ - { name = "colorama" }, - { name = "sphinx", version = "8.1.3", source = { registry = "https://pypi.org/simple" } }, - { name = "starlette" }, - { name = "uvicorn" }, - { name = "watchfiles" }, - { name = "websockets" }, + { name = "colorama", marker = "python_full_version < '3.11'" }, + { name = "sphinx", version = "8.1.3", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "starlette", marker = "python_full_version < '3.11'" }, + { name = "uvicorn", marker = "python_full_version < '3.11'" }, + { name = "watchfiles", marker = "python_full_version < '3.11'" }, + { name = "websockets", marker = "python_full_version < '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/a5/2c/155e1de2c1ba96a72e5dba152c509a8b41e047ee5c2def9e9f0d812f8be7/sphinx_autobuild-2024.10.3.tar.gz", hash = "sha256:248150f8f333e825107b6d4b86113ab28fa51750e5f9ae63b59dc339be951fb1", size = 14023, upload-time = "2024-10-02T23:15:30.172Z" } wheels = [ @@ -5365,13 +5365,13 @@ resolution-markers = [ "python_full_version == '3.11.*'", ] dependencies = [ - { name = "colorama" }, - { name = "sphinx", version = "9.0.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.12'" }, + { name = "colorama", marker = "python_full_version >= '3.11'" }, + { name = "sphinx", version = "9.0.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version == '3.11.*'" }, { name = "sphinx", version = "9.1.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.12'" }, - { name = "starlette" }, - { name = "uvicorn" }, - { name = "watchfiles" }, - { name = "websockets" }, + { name = "starlette", marker = "python_full_version >= '3.11'" }, + { name = "uvicorn", marker = "python_full_version >= '3.11'" }, + { name = "watchfiles", marker = "python_full_version >= '3.11'" }, + { name = "websockets", marker = "python_full_version >= '3.11'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e0/3c/a59a3a453d4133777f7ed2e83c80b7dc817d43c74b74298ca0af869662ad/sphinx_autobuild-2025.8.25.tar.gz", hash = "sha256:9cf5aab32853c8c31af572e4fecdc09c997e2b8be5a07daf2a389e270e85b213", size = 15200, upload-time = "2025-08-25T18:44:55.436Z" } wheels = [ diff --git a/plans/tool-call-concurrency-controls.md b/plans/tool-call-concurrency-controls.md new file mode 100644 index 0000000..ed97f04 --- /dev/null +++ b/plans/tool-call-concurrency-controls.md @@ -0,0 +1,66 @@ +# Tool call concurrency controls + +## Problem +Tool calls used to be dispatched **one at a time**: the Python client awaited each tool inside its WebSocket read loop, so it never read the next request until the current tool finished. Code fanning out with `Promise.all` took the *sum* of its calls instead of the slowest one. Everything above the client was already concurrent — the Deno op is `#[op2(async)]` ([invoke_ops.rs:13](../crates/pctx_code_execution_runtime/src/invoke_ops.rs#L13)), the registry holds no lock across its await ([registry.rs:241](../crates/pctx_registry/src/registry.rs#L241)), and the server gives each call its own request id and response channel ([ws_manager.rs:146](../crates/pctx_session_server/src/state/ws_manager.rs#L146)). The requests left the server together and queued at the client. + +## Current solution (shipped) +[`_websocket_client.py`](../pctx-py/src/pctx_client/_websocket_client.py): +1. Each `ExecuteToolRequest` runs in its own `asyncio.Task`, so the read loop stays free. Tasks are held in a set (asyncio keeps only weak references) and cancelled as a group on disconnect. +2. Sync tools run via `asyncio.to_thread` — calling one inline blocked the event loop for its whole duration, stalling every other in-flight call behind it. + +Measured, 4 × 2s tool under `Promise.all`: **8.12s → 2.11s**, all four starting at t=0.14. Covered by `test_concurrent_tool_calls_run_in_parallel` and `test_concurrent_sync_tool_calls_run_in_parallel` ([test_integration.py](../pctx-py/tests/test_integration.py)), which assert on observed overlap rather than wall time so they fail on serialization, not on a slow machine. + +**Dispatch is now unbounded for async tools** — N concurrent calls in JS means N concurrent tasks in Python. What follows is about bounding that. + +--- + +## The constraint that shapes every option + +The server's per-tool timeout starts at **dispatch**, not at execution: [`execute_callback`](../crates/pctx_session_server/src/state/ws_manager.rs#L146) sends the request, *then* starts `tokio::time::timeout`. Any client-side queue silently spends the tool's timeout budget waiting. + +Measured — 40 sync calls × 2s work, `tool_timeout_secs=5`: + +``` +starts: 14 @ 0.19s | 14 @ 2.2s | 12 @ 4.2s +result: ok=28, failed=12 + 'Tool `tools__slow_sync` timed out after 5s' +``` + +12 calls died on a 5s deadline while doing 2s of work, purely from queue wait. **A concurrency cap converts a throughput problem into hard failures.** Any cap has to be introduced with this in mind. + +## Where we are today + +Measured on a 10-core machine, 40-way fan-out: + +| | peak in-flight | wall | +|---|---|---| +| async tools | 40 (unbounded) | 0.66s | +| sync tools | **14** | 1.68s | + +The 14 is `min(32, cpu_count + 4)` — asyncio's default `ThreadPoolExecutor`, inherited via `to_thread`. So there is **already a cap on the sync path**, it is just accidental and machine-dependent: 14 here, 6 on a 2-core CI box, moving the timeout knee with the hardware. + +--- + +## Options + +| Option | How | Pros | Cons | +|---|---|---|---| +| **A. Explicit sync executor** | Dedicated `ThreadPoolExecutor` with a configured size instead of asyncio's default | Replaces an invisible machine-dependent cap with a stated one; no behaviour change where `cpu+4 == size` | Doesn't bound async tools | +| **B. Opt-in `max_concurrent_tools`** | Semaphore around tool dispatch, default `None` (unbounded) | Available for rate-limited upstreams; preserves current behaviour by default | Queue wait burns the server timeout (above); one slow tool blocks unrelated fast ones | +| **C. Server sends its deadline** | Include the deadline in `ExecuteToolRequest` | Client can fail fast with a real reason, or drop work it can't service in time; makes caps *safe* rather than merely possible | Protocol + server change; both sides must ship | +| **D. Per-tool limits in the tool body** | Caller puts a semaphore inside their own tool function | Correctly scoped per upstream; no client change | Not discoverable; every caller reimplements it | + +### Why not a global semaphore on its own +A single cap means a slow Excel call blocks an unrelated fast Graph call. A client-level cap is a **resource safety valve** (sockets, memory, threads), not a throttle. Rate limiting is per-upstream and belongs where the upstream is known — **D**, or **B** scoped per tool rather than globally. + +--- + +## Recommendation +Ship the current fix as-is; it is a strict improvement and adds no cap where none existed. + +Then, cheapest-first: +1. **A** — make the sync cap explicit. It's the one cap that already exists and already bites, silently and differently per machine. +2. **B**, defaulting to unbounded. Given the timeout coupling, unbounded is the correct default: everything starts immediately and gets its full budget. Offer the knob to callers protecting an upstream, and document that `tool_timeout_secs` must cover *queue wait plus call duration*, not just call duration. +3. **C** if caps become load-bearing. Without the deadline, a queued client is blind to a clock already running against it, and overload surfaces as mystery timeouts. + +Treat **D** as the documented answer for per-upstream rate limiting regardless of whether B lands.