From 9de77c3ee3ce3765d7ea159b17ce4cf0cb5f8ec3 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 30 Mar 2026 16:52:30 -0700 Subject: [PATCH] feat: extend SDK clients for internal dogfooding Server SDK create_workstream: add initial_message, auto_approve_tools, user_id, ws_id params (all optional, omitted when empty). Console SDK: add auto_approve, auto_approve_tools, user_id to create_workstream. Add 8 route_* methods for the routing proxy path (/api/route/*): route_create_workstream, route_send, route_approve, route_plan_feedback, route_close, route_cancel, route_command, route_lookup. Sync mirrors for all. Prepares for channel gateway and scheduler to use SDK clients instead of raw httpx calls. --- tests/test_sdk_console.py | 256 ++++++++++++++++++++++++++++++++++++++ tests/test_sdk_server.py | 51 ++++++++ turnstone/sdk/console.py | 171 +++++++++++++++++++++++++ turnstone/sdk/server.py | 20 +++ 4 files changed, 498 insertions(+) diff --git a/tests/test_sdk_console.py b/tests/test_sdk_console.py index a5870ee9..aa35569d 100644 --- a/tests/test_sdk_console.py +++ b/tests/test_sdk_console.py @@ -379,3 +379,259 @@ async def test_list_schedule_runs(): assert len(resp.runs) == 1 assert resp.runs[0].run_id == "r1" assert resp.runs[0].status == "dispatched" + + +# --------------------------------------------------------------------------- +# create_workstream extended params +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_create_workstream_extended_params(): + """New optional params appear in JSON body only when non-empty.""" + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response( + {"status": "dispatched", "correlation_id": "abc", "target_node": "n1"} + ) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.create_workstream( + node_id="n1", + name="ext", + auto_approve=True, + auto_approve_tools="read_file", + user_id="u42", + ) + assert captured_body["node_id"] == "n1" + assert captured_body["name"] == "ext" + assert captured_body["auto_approve"] is True + assert captured_body["auto_approve_tools"] == "read_file" + assert captured_body["user_id"] == "u42" + + +@pytest.mark.anyio +async def test_create_workstream_omits_empty_new_params(): + """Default-valued new params should not appear in JSON body.""" + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response( + {"status": "dispatched", "correlation_id": "abc", "target_node": "n1"} + ) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.create_workstream(name="min") + assert captured_body == {"name": "min"} + assert "auto_approve" not in captured_body + assert "auto_approve_tools" not in captured_body + assert "user_id" not in captured_body + + +# --------------------------------------------------------------------------- +# Route methods +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_route_create_workstream(): + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response({"ws_id": "ws1", "node_url": "http://n1:8080", "node_id": "n1"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.route_create_workstream( + name="routed", + model="gpt-5", + auto_approve=True, + target_node="n1", + user_id="u1", + ) + assert resp["ws_id"] == "ws1" + assert resp["node_url"] == "http://n1:8080" + assert captured_body["name"] == "routed" + assert captured_body["model"] == "gpt-5" + assert captured_body["auto_approve"] is True + assert captured_body["target_node"] == "n1" + assert captured_body["user_id"] == "u1" + + +@pytest.mark.anyio +async def test_route_create_workstream_omits_defaults(): + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response({"ws_id": "ws1", "node_url": "http://n1:8080"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_create_workstream(name="bare") + assert captured_body == {"name": "bare"} + + +@pytest.mark.anyio +async def test_route_send(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["path"] = request.url.path + captured["body"] = json.loads(request.content) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.route_send("Hello", "ws1") + assert resp["status"] == "ok" + assert captured["path"] == "/v1/api/route/send" + assert captured["body"] == {"message": "Hello", "ws_id": "ws1"} + + +@pytest.mark.anyio +async def test_route_approve(): + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_approve(ws_id="ws1", approved=False, feedback="no", always=True) + assert captured_body["ws_id"] == "ws1" + assert captured_body["approved"] is False + assert captured_body["feedback"] == "no" + assert captured_body["always"] is True + + +@pytest.mark.anyio +async def test_route_approve_omits_defaults(): + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_approve(ws_id="ws1", approved=True) + assert captured_body == {"ws_id": "ws1", "approved": True} + assert "feedback" not in captured_body + assert "always" not in captured_body + + +@pytest.mark.anyio +async def test_route_plan_feedback(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["path"] = request.url.path + captured["body"] = json.loads(request.content) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_plan_feedback(ws_id="ws1", feedback="approved") + assert captured["path"] == "/v1/api/route/plan" + assert captured["body"] == {"ws_id": "ws1", "feedback": "approved"} + + +@pytest.mark.anyio +async def test_route_close(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["path"] = request.url.path + captured["body"] = json.loads(request.content) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.route_close("ws1") + assert resp["status"] == "ok" + assert captured["path"] == "/v1/api/route/workstreams/close" + assert captured["body"] == {"ws_id": "ws1"} + + +@pytest.mark.anyio +async def test_route_cancel(): + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_cancel("ws1", force=True) + assert captured_body == {"ws_id": "ws1", "force": True} + + +@pytest.mark.anyio +async def test_route_cancel_omits_force_when_false(): + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_cancel("ws1") + assert captured_body == {"ws_id": "ws1"} + assert "force" not in captured_body + + +@pytest.mark.anyio +async def test_route_command(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["path"] = request.url.path + captured["body"] = json.loads(request.content) + return _json_response({"status": "ok"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + await client.route_command(ws_id="ws1", command="/clear") + assert captured["path"] == "/v1/api/route/command" + assert captured["body"] == {"ws_id": "ws1", "command": "/clear"} + + +@pytest.mark.anyio +async def test_route_lookup(): + captured: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured["path"] = request.url.path + captured["url"] = str(request.url) + return _json_response({"node_url": "http://n1:8080", "node_id": "n1"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneConsole(httpx_client=hc) + resp = await client.route_lookup("ws1") + assert resp["node_url"] == "http://n1:8080" + assert resp["node_id"] == "n1" + assert captured["path"] == "/v1/api/route" + assert "ws_id=ws1" in captured["url"] diff --git a/tests/test_sdk_server.py b/tests/test_sdk_server.py index 699571d0..610ab5ce 100644 --- a/tests/test_sdk_server.py +++ b/tests/test_sdk_server.py @@ -279,3 +279,54 @@ async def test_request_body_correct(): client = AsyncTurnstoneServer(httpx_client=hc) await client.send("Hello world", "ws_123") assert captured_body == {"message": "Hello world", "ws_id": "ws_123"} + + +# --------------------------------------------------------------------------- +# create_workstream extended params +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_create_workstream_extended_params(): + """New optional params appear in JSON body only when non-empty.""" + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return httpx.Response(200, json={"ws_id": "ws_ext", "name": "ext"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + await client.create_workstream( + name="ext", + initial_message="hi", + auto_approve_tools="read_file,write_file", + user_id="u42", + ws_id="ws_custom", + ) + assert captured_body["name"] == "ext" + assert captured_body["initial_message"] == "hi" + assert captured_body["auto_approve_tools"] == "read_file,write_file" + assert captured_body["user_id"] == "u42" + assert captured_body["ws_id"] == "ws_custom" + + +@pytest.mark.anyio +async def test_create_workstream_omits_empty_params(): + """Empty-string params should NOT appear in the JSON body.""" + captured_body: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + captured_body.update(json.loads(request.content)) + return httpx.Response(200, json={"ws_id": "ws_min", "name": "min"}) + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc: + client = AsyncTurnstoneServer(httpx_client=hc) + await client.create_workstream(name="min") + assert captured_body == {"name": "min"} + assert "initial_message" not in captured_body + assert "auto_approve_tools" not in captured_body + assert "user_id" not in captured_body + assert "ws_id" not in captured_body diff --git a/turnstone/sdk/console.py b/turnstone/sdk/console.py index 27f4544a..cb509430 100644 --- a/turnstone/sdk/console.py +++ b/turnstone/sdk/console.py @@ -153,6 +153,9 @@ class AsyncTurnstoneConsole(_BaseClient): initial_message: str = "", skill: str = "", resume_ws: str = "", + auto_approve: bool = False, + auto_approve_tools: str = "", + user_id: str = "", ) -> ConsoleCreateWsResponse: body: dict[str, Any] = {} if node_id: @@ -167,6 +170,12 @@ class AsyncTurnstoneConsole(_BaseClient): body["skill"] = skill if resume_ws: body["resume_ws"] = resume_ws + if auto_approve: + body["auto_approve"] = True + if auto_approve_tools: + body["auto_approve_tools"] = auto_approve_tools + if user_id: + body["user_id"] = user_id return await self._request( "POST", "/v1/api/cluster/workstreams/new", @@ -174,6 +183,101 @@ class AsyncTurnstoneConsole(_BaseClient): response_model=ConsoleCreateWsResponse, ) + # -- routing proxy ------------------------------------------------------- + + async def route_create_workstream( + self, + *, + name: str = "", + model: str = "", + auto_approve: bool = False, + auto_approve_tools: str = "", + initial_message: str = "", + skill: str = "", + resume_ws: str = "", + target_node: str = "", + user_id: str = "", + ) -> dict[str, Any]: + """Create a workstream via the console's routing proxy. + + Posts to /v1/api/route/workstreams/new. Returns the full response + dict including node_url and node_id. + """ + body: dict[str, Any] = {} + if name: + body["name"] = name + if model: + body["model"] = model + if auto_approve: + body["auto_approve"] = True + if auto_approve_tools: + body["auto_approve_tools"] = auto_approve_tools + if initial_message: + body["initial_message"] = initial_message + if skill: + body["skill"] = skill + if resume_ws: + body["resume_ws"] = resume_ws + if target_node: + body["target_node"] = target_node + if user_id: + body["user_id"] = user_id + return await self._request("POST", "/v1/api/route/workstreams/new", json_body=body) + + async def route_send(self, message: str, ws_id: str) -> dict[str, Any]: + """Send a message via the routing proxy.""" + return await self._request( + "POST", "/v1/api/route/send", json_body={"message": message, "ws_id": ws_id} + ) + + async def route_approve( + self, + *, + ws_id: str, + approved: bool = True, + feedback: str = "", + always: bool = False, + ) -> dict[str, Any]: + """Approve or reject a pending tool call via the routing proxy.""" + body: dict[str, Any] = {"ws_id": ws_id, "approved": approved} + if feedback: + body["feedback"] = feedback + if always: + body["always"] = True + return await self._request("POST", "/v1/api/route/approve", json_body=body) + + async def route_plan_feedback(self, *, ws_id: str, feedback: str) -> dict[str, Any]: + """Send plan feedback via the routing proxy.""" + return await self._request( + "POST", "/v1/api/route/plan", json_body={"ws_id": ws_id, "feedback": feedback} + ) + + async def route_close(self, ws_id: str) -> dict[str, Any]: + """Close a workstream via the routing proxy.""" + return await self._request( + "POST", "/v1/api/route/workstreams/close", json_body={"ws_id": ws_id} + ) + + async def route_cancel(self, ws_id: str, *, force: bool = False) -> dict[str, Any]: + """Cancel the current turn via the routing proxy.""" + body: dict[str, Any] = {"ws_id": ws_id} + if force: + body["force"] = True + return await self._request("POST", "/v1/api/route/cancel", json_body=body) + + async def route_command(self, *, ws_id: str, command: str) -> dict[str, Any]: + """Send a slash command via the routing proxy.""" + return await self._request( + "POST", "/v1/api/route/command", json_body={"ws_id": ws_id, "command": command} + ) + + async def route_lookup(self, ws_id: str) -> dict[str, Any]: + """Look up which server node owns a workstream. + + Returns {"node_url": "...", "node_id": "..."}. + """ + return await self._request("GET", "/v1/api/route", params={"ws_id": ws_id}) + # -- streaming ----------------------------------------------------------- async def stream_cluster_events(self) -> AsyncIterator[ClusterEvent]: @@ -921,6 +1025,9 @@ class TurnstoneConsole: initial_message: str = "", skill: str = "", resume_ws: str = "", + auto_approve: bool = False, + auto_approve_tools: str = "", + user_id: str = "", ) -> ConsoleCreateWsResponse: return self._runner.run( self._async.create_workstream( @@ -930,9 +1037,73 @@ class TurnstoneConsole: initial_message=initial_message, skill=skill, resume_ws=resume_ws, + auto_approve=auto_approve, + auto_approve_tools=auto_approve_tools, + user_id=user_id, ) ) + # -- routing proxy ------------------------------------------------------- + + def route_create_workstream( + self, + *, + name: str = "", + model: str = "", + auto_approve: bool = False, + auto_approve_tools: str = "", + initial_message: str = "", + skill: str = "", + resume_ws: str = "", + target_node: str = "", + user_id: str = "", + ) -> dict[str, Any]: + return self._runner.run( + self._async.route_create_workstream( + name=name, + model=model, + auto_approve=auto_approve, + auto_approve_tools=auto_approve_tools, + initial_message=initial_message, + skill=skill, + resume_ws=resume_ws, + target_node=target_node, + user_id=user_id, + ) + ) + + def route_send(self, message: str, ws_id: str) -> dict[str, Any]: + return self._runner.run(self._async.route_send(message, ws_id)) + + def route_approve( + self, + *, + ws_id: str, + approved: bool = True, + feedback: str = "", + always: bool = False, + ) -> dict[str, Any]: + return self._runner.run( + self._async.route_approve( + ws_id=ws_id, approved=approved, feedback=feedback, always=always + ) + ) + + def route_plan_feedback(self, *, ws_id: str, feedback: str) -> dict[str, Any]: + return self._runner.run(self._async.route_plan_feedback(ws_id=ws_id, feedback=feedback)) + + def route_close(self, ws_id: str) -> dict[str, Any]: + return self._runner.run(self._async.route_close(ws_id)) + + def route_cancel(self, ws_id: str, *, force: bool = False) -> dict[str, Any]: + return self._runner.run(self._async.route_cancel(ws_id, force=force)) + + def route_command(self, *, ws_id: str, command: str) -> dict[str, Any]: + return self._runner.run(self._async.route_command(ws_id=ws_id, command=command)) + + def route_lookup(self, ws_id: str) -> dict[str, Any]: + return self._runner.run(self._async.route_lookup(ws_id)) + # -- streaming ----------------------------------------------------------- def stream_cluster_events(self) -> Iterator[ClusterEvent]: diff --git a/turnstone/sdk/server.py b/turnstone/sdk/server.py index 053c21cc..fe440710 100644 --- a/turnstone/sdk/server.py +++ b/turnstone/sdk/server.py @@ -92,6 +92,10 @@ class AsyncTurnstoneServer(_BaseClient): auto_approve: bool = False, resume_ws: str = "", skill: str = "", + initial_message: str = "", + auto_approve_tools: str = "", + user_id: str = "", + ws_id: str = "", ) -> CreateWorkstreamResponse: body: dict[str, Any] = {} if name: @@ -104,6 +108,14 @@ class AsyncTurnstoneServer(_BaseClient): body["resume_ws"] = resume_ws if skill: body["skill"] = skill + if initial_message: + body["initial_message"] = initial_message + if auto_approve_tools: + body["auto_approve_tools"] = auto_approve_tools + if user_id: + body["user_id"] = user_id + if ws_id: + body["ws_id"] = ws_id return await self._request( "POST", "/v1/api/workstreams/new", @@ -439,6 +451,10 @@ class TurnstoneServer: auto_approve: bool = False, resume_ws: str = "", skill: str = "", + initial_message: str = "", + auto_approve_tools: str = "", + user_id: str = "", + ws_id: str = "", ) -> CreateWorkstreamResponse: return self._runner.run( self._async.create_workstream( @@ -447,6 +463,10 @@ class TurnstoneServer: auto_approve=auto_approve, resume_ws=resume_ws, skill=skill, + initial_message=initial_message, + auto_approve_tools=auto_approve_tools, + user_id=user_id, + ws_id=ws_id, ) )