mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
fix: address Copilot PR review feedback
- 404 retry: use blocking lock acquire so retry waits for cache refresh to complete instead of skipping on contention - 404 retry: surface httpx.HTTPError as 502 instead of suppressing it and returning the original 404 - channel router: pass auto_approve_tools to create_workstream calls (was silently dropped for console-routed creates) - api-reference.md: document all /v1/api/route/* console routing proxy endpoints and console /metrics
This commit is contained in:
committed by
Patrick Buckley
parent
473298199d
commit
843fa04e65
@@ -1857,3 +1857,54 @@ turnstone_tokens_total{type="completion"} 12150
|
||||
turnstone_tool_calls_total{tool="bash"} 7
|
||||
turnstone_tool_calls_total{tool="read_file"} 3
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Console Routing Proxy Endpoints
|
||||
|
||||
These endpoints are served by the console (`turnstone-console`) and proxy
|
||||
requests to the correct server node via the hash ring bucket cache. In
|
||||
multi-node deployments, clients (SDK, channel gateway) talk to the console
|
||||
instead of individual server nodes.
|
||||
|
||||
### `POST /v1/api/route/workstreams/new`
|
||||
|
||||
Create a workstream via hash-ring routing. The console generates the `ws_id`,
|
||||
routes to the assigned node, and includes `node_url` in the response for
|
||||
direct SSE connections.
|
||||
|
||||
### `POST /v1/api/route/send`
|
||||
|
||||
Proxy a message to the workstream's assigned server node.
|
||||
|
||||
### `POST /v1/api/route/approve`
|
||||
|
||||
Proxy an approval response to the workstream's assigned server node.
|
||||
|
||||
### `POST /v1/api/route/cancel`
|
||||
|
||||
Cancel generation on a workstream.
|
||||
|
||||
### `POST /v1/api/route/command`
|
||||
|
||||
Send a slash command to a workstream.
|
||||
|
||||
### `POST /v1/api/route/plan`
|
||||
|
||||
Send plan review feedback to a workstream.
|
||||
|
||||
### `POST /v1/api/route/workstreams/close`
|
||||
|
||||
Close a workstream.
|
||||
|
||||
### `GET /v1/api/route?ws_id=X`
|
||||
|
||||
Look up which server node owns a workstream. Returns `{"node_url": "...", "node_id": "..."}`.
|
||||
Used by channel adapters to open direct SSE connections to the correct server node.
|
||||
|
||||
### `GET /metrics` (Console)
|
||||
|
||||
Prometheus metrics for the console routing layer. Includes:
|
||||
`turnstone_router_requests_total`, `turnstone_router_request_duration_seconds`,
|
||||
`turnstone_ring_membership_size`, `turnstone_ring_version`,
|
||||
`turnstone_ring_rebalance_total`, `turnstone_ring_migrations_total`.
|
||||
|
||||
@@ -153,10 +153,8 @@ class ChannelRouter:
|
||||
)
|
||||
|
||||
# 2. Create via SDK client with atomic resume.
|
||||
# Note: auto_approve_tools is not passed here because the server's
|
||||
# create endpoint does not accept it. Per-tool auto-approve is
|
||||
# handled channel-side in the adapter's _should_auto_approve().
|
||||
resume_ws = old_ws_id or ""
|
||||
_tools_csv = ",".join(self._auto_approve_tools) if self._auto_approve_tools else ""
|
||||
log.info(
|
||||
"channel_router.creating_workstream",
|
||||
channel_type=channel_type,
|
||||
@@ -171,6 +169,7 @@ class ChannelRouter:
|
||||
resume_ws=resume_ws,
|
||||
skill=self._skill,
|
||||
auto_approve=self._auto_approve,
|
||||
auto_approve_tools=_tools_csv,
|
||||
)
|
||||
ws_id = data.get("ws_id", "")
|
||||
else:
|
||||
@@ -181,6 +180,7 @@ class ChannelRouter:
|
||||
resume_ws=resume_ws,
|
||||
skill=self._skill,
|
||||
auto_approve=self._auto_approve,
|
||||
auto_approve_tools=_tools_csv,
|
||||
)
|
||||
ws_id = resp.ws_id
|
||||
data = {"ws_id": resp.ws_id, "name": resp.name}
|
||||
|
||||
@@ -12,7 +12,6 @@ from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import contextlib
|
||||
import functools
|
||||
import html
|
||||
import json
|
||||
@@ -745,16 +744,33 @@ async def route_proxy(request: Request) -> Response:
|
||||
# the last cache load), retry on the new node. If the route is the
|
||||
# same, return the 404 as-is — no loop, no scan.
|
||||
if resp.status_code == 404:
|
||||
router.refresh_cache()
|
||||
# Blocking refresh — wait for any in-progress refresh to finish
|
||||
# so the retry uses the latest data, not stale cache.
|
||||
router._refresh_lock.acquire()
|
||||
try:
|
||||
router._refresh_cache_locked()
|
||||
finally:
|
||||
router._refresh_lock.release()
|
||||
try:
|
||||
new_ref = router.route(ws_id)
|
||||
except (NoAvailableNodeError, ValueError):
|
||||
new_ref = ref
|
||||
if new_ref.node_id != ref.node_id:
|
||||
with contextlib.suppress(httpx.HTTPError):
|
||||
try:
|
||||
resp = await client.post(
|
||||
f"{new_ref.url}{upstream_path}", json=body, headers=headers
|
||||
)
|
||||
except httpx.HTTPError:
|
||||
return _record_route(
|
||||
request,
|
||||
method,
|
||||
502,
|
||||
t0,
|
||||
JSONResponse(
|
||||
{"error": f"retry node {new_ref.node_id} unreachable"},
|
||||
status_code=502,
|
||||
),
|
||||
)
|
||||
|
||||
return _record_route(
|
||||
request,
|
||||
|
||||
Reference in New Issue
Block a user