mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
refactor: channel router + scheduler use SDK clients
ChannelRouter: replace raw httpx with AsyncTurnstoneServer (single-node) and AsyncTurnstoneConsole route methods (multi-node). Remove _post() helper, _route_path(), and manual JSON construction. Scheduler: replace raw httpx.Client with TurnstoneServer (sync). Lazy per-node client cache with token rotation and stale client pruning. Clean remaining Redis/MQ references from tests, docs, and config: - test_tls_admin: redis.internal -> app.internal - test_config: [redis] test data -> [database] - docs/channels.md, console.md: rewrite for HTTP architecture - docs/api-reference.md, openshell.md: remove stale diagram/Redis refs - turnstone.example.toml: remove [redis] section - .pre-commit-config.yaml: remove types-redis dependency - QUICKSTART.md: remove bridge/Redis from deployment descriptions
This commit is contained in:
Patrick Buckley
committed by
Patrick Buckley
Patrick Buckley
parent
9de77c3ee3
commit
a7d9461735
@@ -10,7 +10,7 @@ repos:
|
||||
rev: v1.19.1
|
||||
hooks:
|
||||
- id: mypy
|
||||
additional_dependencies: [types-redis>=4.6, redis>=7.2]
|
||||
additional_dependencies: []
|
||||
args: [--config-file=pyproject.toml]
|
||||
pass_filenames: false
|
||||
entry: mypy turnstone/
|
||||
|
||||
+2
-2
@@ -45,9 +45,9 @@ That's it — no flags, no arguments. The wizard prompts for everything.
|
||||
The wizard supports two deployment modes:
|
||||
|
||||
- **Single-node production** (`docker compose --profile production up`) —
|
||||
1 server + bridge + console + PostgreSQL + Redis. Good for most use cases.
|
||||
1 server + console + PostgreSQL. Good for most use cases.
|
||||
- **Multi-node cluster** (`docker compose --profile cluster up`) —
|
||||
10-node server/bridge fleet + PostgreSQL + Redis. For high-throughput or
|
||||
10-node server fleet + console + PostgreSQL. For high-throughput or
|
||||
HA deployments.
|
||||
|
||||
## Example Session
|
||||
|
||||
@@ -2,8 +2,6 @@
|
||||
|
||||
## Overview
|
||||
|
||||
> See also: [MQ Protocol diagram](diagrams/png/06-mq-protocol.png) | [Message Routing diagram](diagrams/png/07-message-routing.png) | [Redis Key Schema diagram](diagrams/png/08-redis-key-schema.png)
|
||||
|
||||
`turnstone-server` exposes a browser-based chat UI backed by a
|
||||
**Starlette** ASGI application served by **uvicorn**. The server uses
|
||||
**Server-Sent Events (SSE)** via `sse-starlette` for real-time streaming
|
||||
@@ -523,7 +521,7 @@ inactivity.
|
||||
|
||||
Each SSE connection to a workstream receives its own delivery queue. Events
|
||||
produced by the worker thread are fanned out to all registered listener queues,
|
||||
so multiple consumers (browser, bridge, console proxy, SDK) can connect
|
||||
so multiple consumers (browser, console proxy, SDK) can connect
|
||||
simultaneously and each receives every event. On reconnect the client receives
|
||||
a full history replay, so no catch-up mechanism is needed.
|
||||
|
||||
|
||||
+18
-25
@@ -1,9 +1,10 @@
|
||||
# Channel Integrations
|
||||
|
||||
The `turnstone-channel` gateway connects external messaging platforms to
|
||||
turnstone workstreams via Redis MQ. Each platform adapter translates
|
||||
turnstone workstreams via direct HTTP to the server (single-node) or the
|
||||
console routing proxy (multi-node). Each platform adapter translates
|
||||
platform-native events (messages, button clicks, slash commands) into
|
||||
turnstone MQ messages, and renders workstream output back into the
|
||||
turnstone API calls, and renders workstream output back into the
|
||||
platform's UI.
|
||||
|
||||
Discord ships as the first adapter. The adapter protocol is designed for
|
||||
@@ -20,10 +21,9 @@ Discord Gateway
|
||||
turnstone-channel (Discord adapter)
|
||||
|
|
||||
v
|
||||
Redis MQ
|
||||
|
|
||||
v
|
||||
turnstone-bridge ──> turnstone-server
|
||||
turnstone-server (direct HTTP)
|
||||
or
|
||||
turnstone-console (routing proxy, multi-node)
|
||||
```
|
||||
|
||||
Key components:
|
||||
@@ -34,10 +34,7 @@ Key components:
|
||||
`send_approval_request()`, `send_plan_review()`, and `create_thread()`.
|
||||
- **ChannelRouter** (`turnstone/channels/_routing.py`) — maps
|
||||
channel/thread IDs to turnstone workstream IDs. Handles workstream
|
||||
creation via MQ, stale route detection, and user identity resolution.
|
||||
- **AsyncRedisBroker** (`turnstone/mq/async_broker.py`) — async Redis
|
||||
client compatible with discord.py's event loop. Used by the router for
|
||||
pub/sub and queue operations.
|
||||
creation via HTTP, stale route detection, and user identity resolution.
|
||||
- **channel_users table** — maps `(channel_type, channel_user_id)` to a
|
||||
turnstone `user_id`. Messages from unlinked users are silently dropped.
|
||||
- **channel_routes table** — persistent channel-to-workstream mappings.
|
||||
@@ -84,8 +81,7 @@ TURNSTONE_DISCORD_GUILD=123456789 # optional, restrict to one guild
|
||||
turnstone-channel \
|
||||
--discord-token "your-bot-token" \
|
||||
--discord-guild 123456789 \
|
||||
--redis-host localhost \
|
||||
--redis-port 6379
|
||||
--server-url http://localhost:8080
|
||||
```
|
||||
|
||||
**Docker Compose** (production profile):
|
||||
@@ -138,7 +134,7 @@ An admin can also force-link or unlink users via the console admin panel
|
||||
thread auto-creates a new workstream and atomically resumes the
|
||||
previous workstream via the `resume_ws` field on
|
||||
`CreateWorkstreamMessage`. The server resumes the workstream during
|
||||
creation (same HTTP request), and the bridge emits a
|
||||
creation (same HTTP request), and the server emits a
|
||||
`WorkstreamResumedEvent` back to the channel. The thread receives a
|
||||
*"Resumed: {name} ({count} messages restored)"* confirmation.
|
||||
|
||||
@@ -160,8 +156,7 @@ an orange embed with:
|
||||
- Tool name and argument preview
|
||||
- **Approve** (green), **Reject** (red), **Always Approve** (gray) buttons
|
||||
- Only linked users can interact with approval buttons
|
||||
- The approval decision is forwarded through MQ to the bridge, which
|
||||
relays it to the server
|
||||
- The approval decision is forwarded to the server via HTTP
|
||||
|
||||
Buttons use static `custom_id` values so they survive bot restarts.
|
||||
Correlation data (`ws_id`, `correlation_id`) is stored in the embed footer.
|
||||
@@ -181,7 +176,7 @@ Plan review requests are displayed as a blue embed with:
|
||||
- **Approve Plan** (green) button — approves the plan with empty feedback
|
||||
- **Request Changes** (gray) button — opens a modal for feedback text
|
||||
(up to 2000 characters)
|
||||
- Feedback is forwarded through MQ as a `PlanFeedbackMessage`
|
||||
- Feedback is forwarded to the server via HTTP
|
||||
|
||||
---
|
||||
|
||||
@@ -192,10 +187,8 @@ Plan review requests are displayed as a blue embed with:
|
||||
| `--discord-token` | `TURNSTONE_DISCORD_TOKEN` | — | Bot token (required to enable Discord) |
|
||||
| `--discord-guild` | — | `0` (all guilds) | Restrict to a single Discord guild |
|
||||
| `--discord-channels` | — | empty (all) | Comma-separated channel IDs to allow |
|
||||
| `--redis-host` | `REDIS_HOST` | `localhost` | Redis host |
|
||||
| `--redis-port` | — | `6379` | Redis port |
|
||||
| `--redis-password` | `REDIS_PASSWORD` | — | Redis password |
|
||||
| `--redis-db` | — | `0` | Redis DB number |
|
||||
| `--server-url` | `TURNSTONE_SERVER_URL` | `http://localhost:8080` | Server URL (single-node) |
|
||||
| `--console-url` | `TURNSTONE_CONSOLE_URL` | — | Console URL (multi-node routing proxy) |
|
||||
| `--model` | — | server default | Default model for new workstreams |
|
||||
| `--auto-approve` | — | `false` | Auto-approve ALL tool calls (skips approval buttons entirely) |
|
||||
| `--http-host` | — | `127.0.0.1` | HTTP server bind address for notify endpoint |
|
||||
@@ -232,13 +225,13 @@ See [Security: Database Schema](security.md#database-schema) for the
|
||||
3. **Eviction** — the server evicts an idle workstream for capacity. The
|
||||
route is preserved and the thread stays open.
|
||||
4. **Reactivation** — the next message in the thread detects the stale
|
||||
route (no MQ owner) and creates a new workstream with the old `ws_id`
|
||||
as `resume_ws` on the `CreateWorkstreamMessage`. The server resumes
|
||||
route and creates a new workstream with the old `ws_id`
|
||||
as `resume_ws` on the creation request. The server resumes
|
||||
the workstream during creation (no separate command or reverse lookup
|
||||
needed). The bridge emits a `WorkstreamResumedEvent` to the channel, and
|
||||
needed). The channel receives a `WorkstreamResumedEvent`, and
|
||||
the thread displays *"Resumed: {name} ({count} messages restored)"*.
|
||||
If the old workstream was pruned, a fresh one starts with no error.
|
||||
5. **Close** — `/close` command closes the workstream via MQ, deletes the
|
||||
5. **Close** — `/close` command closes the workstream via HTTP, deletes the
|
||||
route, unsubscribes from events, and archives the Discord thread.
|
||||
|
||||
---
|
||||
@@ -264,7 +257,7 @@ Two modes:
|
||||
|
||||
### Delivery Flow
|
||||
|
||||
Notifications bypass MQ for lower latency. The server calls the channel
|
||||
Notifications use direct HTTP for low latency. The server calls the channel
|
||||
gateway directly over HTTP:
|
||||
|
||||
1. The LLM calls the `notify` tool with a message and target
|
||||
|
||||
+23
-54
@@ -1,16 +1,16 @@
|
||||
# Cluster Dashboard (turnstone-console)
|
||||
|
||||
`turnstone-console` is a cluster management service that provides cluster-wide visibility and control across all turnstone nodes. It connects to the shared Redis broker, discovers nodes via heartbeat keys, polls each node's HTTP API for workstream data, and subscribes to a cluster event channel for real-time state changes.
|
||||
`turnstone-console` is a cluster management service that provides cluster-wide visibility and control across all turnstone nodes. It discovers nodes via the `services` database table, polls each node's HTTP API for workstream data, and receives real-time state changes via HTTP polling.
|
||||
|
||||
The console also supports **workstream creation** (dispatched via MQ to target nodes) and a **reverse proxy** that serves each node's server UI through the console port — so users only need network access to the console, not to individual server nodes.
|
||||
The console also supports **workstream creation** (dispatched via HTTP proxy to target nodes) and a **reverse proxy** that serves each node's server UI through the console port — so users only need network access to the console, not to individual server nodes.
|
||||
|
||||
## Architecture
|
||||
|
||||
> See also: [Console Data Flow diagram](diagrams/png/11-console-data-flow.png)
|
||||
|
||||
```
|
||||
┌── Redis ←── turnstone-bridge ←── turnstone-server
|
||||
│ (MQ) (per node) (per node)
|
||||
┌── services table ── turnstone-server
|
||||
│ (node registry) (per node)
|
||||
turnstone-console ──────┤
|
||||
(one instance) │
|
||||
└── turnstone-server (direct HTTP proxy)
|
||||
@@ -21,45 +21,29 @@ turnstone-console ──────┤
|
||||
|
||||
Data flows in two directions:
|
||||
|
||||
- **Inbound (monitoring):** Bridges publish state changes to `{prefix}:events:cluster` on Redis pub/sub. The console subscribes for real-time updates and periodically polls each node's `GET /v1/api/dashboard` for full workstream snapshots.
|
||||
- **Outbound (control):** The console pushes `CreateWorkstreamMessage` to Redis inbound queues targeting specific nodes. Bridges pick up these messages and create workstreams on their local servers.
|
||||
- **Inbound (monitoring):** The console discovers nodes via the `services` database table (nodes register on startup and send periodic heartbeats). It periodically polls each node's `GET /v1/api/dashboard` for full workstream snapshots and `GET /health` for node health.
|
||||
- **Outbound (control):** The console proxies workstream creation requests to target nodes via HTTP.
|
||||
- **Proxy (pass-through):** The console reverse-proxies each node's server UI at `/node/{node_id}/`, forwarding HTTP and SSE traffic so the browser never contacts server nodes directly.
|
||||
|
||||
### Data Sources
|
||||
|
||||
| Source | Method | Direction | Data |
|
||||
|--------|--------|-----------|------|
|
||||
| Redis heartbeats | `SCAN turnstone:node:*` | Read | Node discovery (node_id, server_url, started) |
|
||||
| Redis pub/sub | `SUBSCRIBE turnstone:events:cluster` | Read | State changes, creates, closes, renames |
|
||||
| `services` table | Database query | Read | Node discovery (node_id, server_url, started) |
|
||||
| Node HTTP API | `GET {server_url}/v1/api/dashboard` | Read | Full workstream list with tokens, context, activity |
|
||||
| Node HTTP API | `GET {server_url}/health` | Read | Node health status |
|
||||
| Redis inbound queue | `RPUSH turnstone:inbound:{node_id}` | Write | Workstream creation commands |
|
||||
| Node HTTP API | `POST {server_url}/v1/api/workstreams/new` | Write | Workstream creation |
|
||||
| Node HTTP API | `GET/POST {server_url}/*` | Proxy | Server UI, API requests, SSE streams |
|
||||
|
||||
### Redis Key: Cluster Event Channel
|
||||
|
||||
Bridges publish to `{prefix}:events:cluster` whenever a workstream state change, creation, closure, or rename occurs. Events include `node_id` so the console can attribute them to the correct node.
|
||||
|
||||
Event types on the cluster channel:
|
||||
|
||||
| Event | Fields | Trigger |
|
||||
|-------|--------|---------|
|
||||
| `cluster_state` | ws_id, state, node_id, tokens, context_ratio, activity | Workstream state transition |
|
||||
| `ws_created` | ws_id, name, node_id | New workstream created |
|
||||
| `ws_closed` | ws_id | Workstream closed |
|
||||
| `ws_rename` | ws_id, name | Workstream renamed |
|
||||
|
||||
---
|
||||
|
||||
## ClusterCollector
|
||||
|
||||
The collector (`turnstone/console/collector.py`) maintains an in-memory snapshot of all nodes and workstreams. Three daemon threads handle data acquisition:
|
||||
The collector (`turnstone/console/collector.py`) maintains an in-memory snapshot of all nodes and workstreams. Two daemon threads handle data acquisition:
|
||||
|
||||
1. **Event subscriber** — subscribes to `{prefix}:events:cluster` via `RedisBroker.subscribe_cluster()`. Applies state changes, creates, closes, and renames to the in-memory model immediately.
|
||||
1. **Node discovery** — queries the `services` database table every 15 seconds. Adds newly discovered nodes, removes expired ones (stale heartbeats), emits `node_joined` / `node_lost` events to SSE listeners.
|
||||
|
||||
2. **Node discovery** — scans heartbeat keys every 15 seconds via `broker.list_nodes()`. Adds newly discovered nodes, removes expired ones, emits `node_joined` / `node_lost` events to SSE listeners.
|
||||
|
||||
3. **Poll loop** — fetches `GET /v1/api/dashboard` and `GET /health` from each known node every 10 seconds. Uses `ThreadPoolExecutor(max_workers=50)` for parallelism. Each poll replaces the node's workstream list with the authoritative server data.
|
||||
2. **Poll loop** — fetches `GET /v1/api/dashboard` and `GET /health` from each known node every 10 seconds. Uses `ThreadPoolExecutor(max_workers=50)` for parallelism. Each poll replaces the node's workstream list with the authoritative server data.
|
||||
|
||||
A `get_snapshot()` method builds the full cluster state under a single lock acquisition — overview aggregates and per-node workstream lists in one atomic read. This is served both as a REST endpoint and as the initial SSE event on client connect.
|
||||
|
||||
@@ -183,7 +167,7 @@ Full cluster state in a single response — all nodes with their workstreams plu
|
||||
|
||||
### `POST /v1/api/cluster/workstreams/new`
|
||||
|
||||
Create a new workstream on a target node. Dispatches a `CreateWorkstreamMessage` through the Redis MQ pipeline — the bridge on the target node picks it up and creates the workstream on the server. Requires `write` scope.
|
||||
Create a new workstream on a target node. The console proxies the creation request to the target node's HTTP API. Requires `write` scope.
|
||||
|
||||
Request:
|
||||
|
||||
@@ -197,9 +181,9 @@ Request:
|
||||
|
||||
All fields are optional:
|
||||
- `node_id` — targeting mode:
|
||||
- **omitted or `"auto"`** — console picks the reachable node with the most available capacity (max_ws - ws_total) and pushes to its directed queue.
|
||||
- **`"pool"`** — pushes to the shared inbound queue; the next available bridge picks it up (true general-pool dispatch).
|
||||
- **specific node ID** — pushes to that node's directed queue.
|
||||
- **omitted or `"auto"`** — console picks the reachable node with the most available capacity (max_ws - ws_total) and proxies the request to it.
|
||||
- **`"pool"`** — console picks a reachable node with available capacity using round-robin selection.
|
||||
- **specific node ID** — proxies the request to that node directly.
|
||||
- `name` — workstream display name. Auto-generated if omitted.
|
||||
- `model` — model alias from the target node's registry. Uses the node's default model if omitted.
|
||||
|
||||
@@ -213,7 +197,7 @@ Response:
|
||||
}
|
||||
```
|
||||
|
||||
Creation is asynchronous — the response confirms the MQ message was dispatched. A `ws_created` event on the cluster SSE stream confirms the workstream was actually created.
|
||||
The response confirms the workstream creation request was proxied to the target node. A `ws_created` event on the cluster SSE stream confirms the workstream was actually created.
|
||||
|
||||
### `GET /v1/api/cluster/events`
|
||||
|
||||
@@ -395,7 +379,7 @@ Breadcrumb: `Cluster > Running` or `Cluster > db-west-04`. Server-side paginated
|
||||
|
||||
Triggered by the "+ new" header button. A modal dialog with:
|
||||
|
||||
- **Node selector** — dropdown with three targeting modes: "Auto (best available)" picks the node with the most headroom, "General pool (any node)" pushes to the shared queue for any bridge to pick up, or a specific node from the list (showing capacity).
|
||||
- **Node selector** — dropdown with three targeting modes: "Auto (best available)" picks the node with the most headroom, "General pool (any node)" picks a node with available capacity using round-robin, or a specific node from the list (showing capacity).
|
||||
- **Profile** — optional dropdown listing enabled skills. Applies the skill's model, auto-approve policy, token budget, and other behavioral settings at creation time.
|
||||
- **Name** — optional text input. Auto-generated if left empty.
|
||||
- **Model** — optional text input for a model alias from the target node's registry.
|
||||
@@ -482,17 +466,17 @@ to create the initial admin user and receive a JWT in one step. See
|
||||
|
||||
## Scheduled Tasks
|
||||
|
||||
The console includes a background **TaskScheduler** daemon that creates workstreams on a timed basis via the MQ broker. It supports cron-based recurring schedules and one-shot `at` schedules.
|
||||
The console includes a background **TaskScheduler** daemon that creates workstreams on a timed basis via HTTP proxy to target nodes. It supports cron-based recurring schedules and one-shot `at` schedules.
|
||||
|
||||
### Architecture
|
||||
|
||||
The scheduler runs as a daemon thread inside the console process. Every `check_interval` seconds (default 15) it:
|
||||
|
||||
1. Acquires a distributed lock via Redis `SET NX EX` (prevents duplicate dispatch in multi-console deployments)
|
||||
1. Acquires a distributed lock via the `system_settings` table (prevents duplicate dispatch in multi-console deployments)
|
||||
2. Queries the storage backend for tasks whose `next_run <= now` and `enabled = true`
|
||||
3. Dispatches each due task as one or more `CreateWorkstreamMessage` via MQ
|
||||
3. Dispatches each due task as one or more workstream creation requests via HTTP proxy
|
||||
4. Updates `last_run` and computes the next `next_run` (or disables one-shot `at` tasks)
|
||||
5. Releases the lock via Lua script (safe conditional delete)
|
||||
5. Releases the lock
|
||||
|
||||
Run history is automatically pruned (runs older than 90 days) approximately once per hour.
|
||||
|
||||
@@ -508,7 +492,7 @@ Run history is automatically pruned (runs older than 90 days) approximately once
|
||||
| Mode | Behavior |
|
||||
|------|----------|
|
||||
| `auto` | Picks the reachable node with the most available capacity |
|
||||
| `pool` | Pushes to the shared inbound queue (any bridge picks it up) |
|
||||
| `pool` | Picks a reachable node with available capacity using round-robin |
|
||||
| `all` | Fan-out to all reachable nodes (capped at `max_fan_out`, default 20) |
|
||||
| `<node_id>` | Targets a specific node by ID |
|
||||
|
||||
@@ -645,10 +629,6 @@ CLI flags for `turnstone-console`:
|
||||
|------|---------|-------------|
|
||||
| `--host` | `0.0.0.0` | Bind host |
|
||||
| `--port` | `8090` | HTTP port |
|
||||
| `--redis-host` | `localhost` | Redis host |
|
||||
| `--redis-port` | `6379` | Redis port |
|
||||
| `--redis-password` | `$REDIS_PASSWORD` | Redis password |
|
||||
| `--redis-db` | `0` | Redis DB |
|
||||
| `--poll-interval` | `10` | Node polling interval (seconds) |
|
||||
| `--auth-token` | `$TURNSTONE_AUTH_TOKEN` | Bearer token for server node communication and proxy |
|
||||
| `--log-level` | `INFO` | Log level |
|
||||
@@ -661,11 +641,6 @@ host = "0.0.0.0"
|
||||
port = 8090
|
||||
url = "http://localhost:8090" # used by CLI /cluster commands
|
||||
poll_interval = 10
|
||||
|
||||
[redis]
|
||||
host = "localhost"
|
||||
port = 6379
|
||||
password = "my-redis-password"
|
||||
```
|
||||
|
||||
---
|
||||
@@ -673,17 +648,11 @@ password = "my-redis-password"
|
||||
## Deployment
|
||||
|
||||
```bash
|
||||
# Start Redis
|
||||
redis-server
|
||||
|
||||
# Start turnstone servers (one per node)
|
||||
turnstone-server --port 8080
|
||||
|
||||
# Start bridges (one per server)
|
||||
turnstone-bridge --server-url http://localhost:8080 --node-id node-a
|
||||
|
||||
# Start cluster console (one instance)
|
||||
turnstone-console --redis-host localhost --port 8090 --auth-token "$TURNSTONE_AUTH_TOKEN"
|
||||
turnstone-console --port 8090 --auth-token "$TURNSTONE_AUTH_TOKEN"
|
||||
```
|
||||
|
||||
Open `http://localhost:8090` for the cluster dashboard. Create workstreams via the "+ new" button. Click any workstream to open the proxied server UI — no direct access to server ports required.
|
||||
|
||||
@@ -98,7 +98,6 @@ cannot bypass the proxy.
|
||||
| `skills_registry` | `skills.sh` | Skill discovery |
|
||||
| `github_api` | `api.github.com` (read-only L7), `raw.githubusercontent.com` | Skill fetch, GitHub API |
|
||||
| `mcp_registry` | `registry.modelcontextprotocol.io` (read-only L7) | MCP server discovery |
|
||||
| `redis` | `127.0.0.1:6379` | Message queue |
|
||||
| `web_fetch_common` | readthedocs, python docs, GitHub Pages, PyPI, npm, Stack Overflow, Wikipedia | Curated web_fetch domains |
|
||||
| `bash_network_tools` | Same as `web_fetch_common` | curl/wget from bash tool |
|
||||
| `package_registries` | `pypi.org`, `files.pythonhosted.org` | pip/uv package installs |
|
||||
@@ -283,6 +282,5 @@ For production deployments:
|
||||
- [ ] Review and trim `web_fetch_common` domains to your actual needs
|
||||
- [ ] Remove `package_registries` policy if pip/uv installs are not needed
|
||||
- [ ] Add your OIDC provider endpoint if using SSO
|
||||
- [ ] Set Redis `allowed_ips` to your actual Redis host if not localhost
|
||||
- [ ] Consider removing `bash_network_tools` entirely if bash should not have
|
||||
network access
|
||||
|
||||
+12
-17
@@ -339,10 +339,9 @@ secret. If no secret is configured, an ephemeral key is generated at
|
||||
startup and a warning is logged — JWTs will not survive restarts or work
|
||||
across nodes.
|
||||
|
||||
The bridge and console **require** `TURNSTONE_JWT_SECRET` when no
|
||||
`--auth-token` is provided. They exit with an error if the secret is
|
||||
missing, since ephemeral secrets would silently break inter-service
|
||||
communication.
|
||||
The console **requires** `TURNSTONE_JWT_SECRET` when no `--auth-token`
|
||||
is provided. It exits with an error if the secret is missing, since
|
||||
ephemeral secrets would silently break inter-service communication.
|
||||
|
||||
---
|
||||
|
||||
@@ -484,30 +483,26 @@ static token is used as a final fallback.
|
||||
|
||||
### Service-to-service authentication
|
||||
|
||||
The bridge and console collector use `ServiceTokenManager` for
|
||||
auto-rotating JWTs when communicating with server nodes:
|
||||
The console collector uses `ServiceTokenManager` for auto-rotating
|
||||
JWTs when communicating with server nodes:
|
||||
|
||||
| Service | Identity | Scope | Audience | Purpose |
|
||||
|---------|----------|-------|----------|---------|
|
||||
| Bridge | `bridge` | `approve` | `turnstone-server` | Tool approval proxy, message relay |
|
||||
| Console collector | `console-collector` | `read` | `turnstone-server` | Node health polling |
|
||||
| Console proxy (fallback) | `console-proxy` | `approve` | `turnstone-server` | Proxied API calls when no user context |
|
||||
| Channel notify | `system` | `write` | `turnstone-channel` | Notification delivery to channel gateway |
|
||||
|
||||
Service tokens use 1-hour expiry with automatic refresh via
|
||||
`ServiceTokenManager`. The bridge injects auth headers per-request via
|
||||
httpx event hooks to ensure rotated tokens are picked up on SSE
|
||||
reconnects.
|
||||
`ServiceTokenManager`.
|
||||
|
||||
### User identity in MQ-dispatched workstreams
|
||||
|
||||
When the console creates a workstream via MQ (the normal path), the
|
||||
authenticated user's `user_id` is embedded in the
|
||||
`CreateWorkstreamMessage`. The bridge forwards this `user_id` in the
|
||||
HTTP payload when calling the server's `POST /v1/api/workstreams/new`.
|
||||
The server accepts a `user_id` from the request body **only when the
|
||||
caller is a trusted service** — identified by `token_source` matching
|
||||
`bridge`, `console-proxy`, or `console`. Regular API callers cannot
|
||||
When the console creates a workstream (the normal path), the
|
||||
authenticated user's `user_id` is forwarded in the HTTP payload when
|
||||
calling the server's `POST /v1/api/workstreams/new`. The server
|
||||
accepts a `user_id` from the request body **only when the caller is a
|
||||
trusted service** — identified by `token_source` matching
|
||||
`console-proxy` or `console`. Regular API callers cannot
|
||||
override `user_id`; the server always uses their JWT identity.
|
||||
|
||||
Note that the channel gateway uses a distinct JWT audience
|
||||
|
||||
+1
-3
@@ -27,7 +27,7 @@ Settings resolution differs between entry points:
|
||||
|
||||
| Entry point | Chain |
|
||||
|-------------|-------|
|
||||
| **Server** (`turnstone-server`, `turnstone-bridge`) | CLI flag > ConfigStore > registry default |
|
||||
| **Server** (`turnstone-server`) | CLI flag > ConfigStore > registry default |
|
||||
| **CLI** (`turnstone`) | CLI flag > config.toml > argparse default |
|
||||
|
||||
The server's `apply_config()` ignores config.toml sections that overlap with
|
||||
@@ -46,9 +46,7 @@ connection, Redis, auth secrets, server bind address). These stay in
|
||||
|----------|---------|-------|
|
||||
| API credentials | `[api]` | config.toml / env |
|
||||
| Database | `[database]` | config.toml / env |
|
||||
| Redis | `[redis]` | config.toml / env |
|
||||
| Auth | `[auth]` | config.toml / env |
|
||||
| Bridge identity | `[bridge]` | config.toml / env |
|
||||
| Console bind | `[console]` | config.toml / env |
|
||||
|
||||
**ConfigStore settings** (48 settings) are loaded from the database after
|
||||
|
||||
+1
-1
@@ -690,7 +690,7 @@ that external tools are read-only. However, global overrides such as
|
||||
`--skip-permissions` will auto-approve all tools, including MCP tools. The
|
||||
interactive "Always" button adds specific tool types to the per-tool auto-approve
|
||||
set. The web UI and server use `approval_label` for MCP tools, giving
|
||||
per-prompt/per-resource granularity. The CLI and bridge use `func_name`, which
|
||||
per-prompt/per-resource granularity. The CLI uses `func_name`, which
|
||||
gives per-tool-type granularity (e.g., all `use_prompt` calls).
|
||||
|
||||
### Sub-agent availability
|
||||
|
||||
+176
-59
@@ -2,13 +2,12 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from turnstone.channels._routing import ChannelRouter
|
||||
from turnstone.sdk._types import TurnstoneAPIError
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -31,16 +30,13 @@ def router(mock_storage: MagicMock) -> ChannelRouter:
|
||||
)
|
||||
|
||||
|
||||
def _ok_response(json_data: object = None) -> httpx.Response:
|
||||
"""Build a mock 200 response with optional JSON body."""
|
||||
import json
|
||||
|
||||
content = json.dumps(json_data or {"status": "ok"}).encode()
|
||||
return httpx.Response(
|
||||
200,
|
||||
content=content,
|
||||
headers={"content-type": "application/json"},
|
||||
request=httpx.Request("POST", "http://test"),
|
||||
@pytest.fixture
|
||||
def console_router(mock_storage: MagicMock) -> ChannelRouter:
|
||||
return ChannelRouter(
|
||||
server_url="http://localhost:8080/v1",
|
||||
storage=mock_storage,
|
||||
console_url="http://localhost:8081/v1",
|
||||
api_token="tok-test",
|
||||
)
|
||||
|
||||
|
||||
@@ -61,53 +57,82 @@ class TestResolveUser:
|
||||
|
||||
class TestSendMessage:
|
||||
@pytest.mark.anyio
|
||||
async def test_posts_to_server(
|
||||
async def test_calls_server_send(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mock_post = AsyncMock(return_value=_ok_response())
|
||||
monkeypatch.setattr(router, "_post", mock_post)
|
||||
assert router._server is not None
|
||||
mock_send = AsyncMock()
|
||||
monkeypatch.setattr(router._server, "send", mock_send)
|
||||
await router.send_message("ws-1", "hello world")
|
||||
mock_post.assert_awaited_once_with("/api/send", {"ws_id": "ws-1", "message": "hello world"})
|
||||
mock_send.assert_awaited_once_with("hello world", "ws-1")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_calls_console_route_send(
|
||||
self, console_router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_send = AsyncMock()
|
||||
monkeypatch.setattr(console_router._console, "route_send", mock_send)
|
||||
await console_router.send_message("ws-1", "hello world")
|
||||
mock_send.assert_awaited_once_with("hello world", "ws-1")
|
||||
|
||||
|
||||
class TestSendApproval:
|
||||
@pytest.mark.anyio
|
||||
async def test_posts_to_server(
|
||||
async def test_calls_server_approve(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mock_post = AsyncMock(return_value=_ok_response())
|
||||
monkeypatch.setattr(router, "_post", mock_post)
|
||||
assert router._server is not None
|
||||
mock_approve = AsyncMock()
|
||||
monkeypatch.setattr(router._server, "approve", mock_approve)
|
||||
await router.send_approval("ws-1", "corr-abc", approved=True, feedback="ok")
|
||||
mock_post.assert_awaited_once_with(
|
||||
"/api/approve",
|
||||
{"ws_id": "ws-1", "approved": True, "always": False, "feedback": "ok"},
|
||||
mock_approve.assert_awaited_once_with(
|
||||
ws_id="ws-1", approved=True, feedback="ok", always=False
|
||||
)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_omits_empty_feedback(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mock_post = AsyncMock(return_value=_ok_response())
|
||||
monkeypatch.setattr(router, "_post", mock_post)
|
||||
assert router._server is not None
|
||||
mock_approve = AsyncMock()
|
||||
monkeypatch.setattr(router._server, "approve", mock_approve)
|
||||
await router.send_approval("ws-1", "corr-abc", approved=False)
|
||||
mock_post.assert_awaited_once_with(
|
||||
"/api/approve",
|
||||
{"ws_id": "ws-1", "approved": False, "always": False},
|
||||
mock_approve.assert_awaited_once_with(
|
||||
ws_id="ws-1", approved=False, feedback=None, always=False
|
||||
)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_calls_console_route_approve(
|
||||
self, console_router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_approve = AsyncMock()
|
||||
monkeypatch.setattr(console_router._console, "route_approve", mock_approve)
|
||||
await console_router.send_approval("ws-1", "corr-abc", approved=True, always=True)
|
||||
mock_approve.assert_awaited_once_with(ws_id="ws-1", approved=True, feedback="", always=True)
|
||||
|
||||
|
||||
class TestSendPlanFeedback:
|
||||
@pytest.mark.anyio
|
||||
async def test_posts_to_server(
|
||||
async def test_calls_server_plan_feedback(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mock_post = AsyncMock(return_value=_ok_response())
|
||||
monkeypatch.setattr(router, "_post", mock_post)
|
||||
assert router._server is not None
|
||||
mock_plan = AsyncMock()
|
||||
monkeypatch.setattr(router._server, "plan_feedback", mock_plan)
|
||||
await router.send_plan_feedback("ws-2", "corr-xyz", "looks good")
|
||||
mock_post.assert_awaited_once_with(
|
||||
"/api/plan",
|
||||
{"ws_id": "ws-2", "feedback": "looks good"},
|
||||
)
|
||||
mock_plan.assert_awaited_once_with(ws_id="ws-2", feedback="looks good")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_calls_console_route_plan_feedback(
|
||||
self, console_router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_plan = AsyncMock()
|
||||
monkeypatch.setattr(console_router._console, "route_plan_feedback", mock_plan)
|
||||
await console_router.send_plan_feedback("ws-2", "corr-xyz", "looks good")
|
||||
mock_plan.assert_awaited_once_with(ws_id="ws-2", feedback="looks good")
|
||||
|
||||
|
||||
class TestDeleteRoute:
|
||||
@@ -121,20 +146,42 @@ class TestDeleteRoute:
|
||||
|
||||
class TestGetOrCreateWorkstream:
|
||||
@pytest.mark.anyio
|
||||
async def test_creates_new_workstream(
|
||||
async def test_creates_new_workstream_via_server(
|
||||
self,
|
||||
router: ChannelRouter,
|
||||
mock_storage: MagicMock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
mock_post = AsyncMock(
|
||||
return_value=_ok_response({"ws_id": "ws-new", "name": "test", "resumed": False}),
|
||||
)
|
||||
monkeypatch.setattr(router, "_post", mock_post)
|
||||
assert router._server is not None
|
||||
mock_create = AsyncMock()
|
||||
mock_create.return_value = MagicMock(ws_id="ws-new", name="test")
|
||||
monkeypatch.setattr(router._server, "create_workstream", mock_create)
|
||||
ws_id, is_new = await router.get_or_create_workstream("discord", "ch-1", name="test")
|
||||
assert ws_id == "ws-new"
|
||||
assert is_new is True
|
||||
mock_storage.create_channel_route.assert_called_once_with("discord", "ch-1", "ws-new")
|
||||
mock_create.assert_awaited_once()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_creates_new_workstream_via_console(
|
||||
self,
|
||||
console_router: ChannelRouter,
|
||||
mock_storage: MagicMock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_create = AsyncMock(
|
||||
return_value={"ws_id": "ws-new", "name": "test", "node_url": "http://node1:8080/v1"}
|
||||
)
|
||||
monkeypatch.setattr(console_router._console, "route_create_workstream", mock_create)
|
||||
ws_id, is_new = await console_router.get_or_create_workstream(
|
||||
"discord", "ch-1", name="test"
|
||||
)
|
||||
assert ws_id == "ws-new"
|
||||
assert is_new is True
|
||||
mock_storage.create_channel_route.assert_called_once_with("discord", "ch-1", "ws-new")
|
||||
# Node URL should be cached.
|
||||
assert console_router._node_urls["ws-new"] == "http://node1:8080/v1"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_returns_existing_alive_workstream(
|
||||
@@ -167,46 +214,116 @@ class TestGetOrCreateWorkstream:
|
||||
}
|
||||
# Alive check returns False — ws is not alive.
|
||||
monkeypatch.setattr(router, "_is_ws_alive", AsyncMock(return_value=False))
|
||||
# POST to create returns a new ws_id.
|
||||
create_resp = _ok_response({"ws_id": "ws-resumed", "name": "test", "resumed": True})
|
||||
captured: list[dict[str, Any]] = []
|
||||
# Server create returns a resumed workstream.
|
||||
assert router._server is not None
|
||||
mock_create = AsyncMock()
|
||||
mock_create.return_value = MagicMock(ws_id="ws-resumed", name="test")
|
||||
monkeypatch.setattr(router._server, "create_workstream", mock_create)
|
||||
|
||||
async def _fake_post(path: str, body: dict[str, Any]) -> httpx.Response:
|
||||
captured.append({"path": path, "body": body})
|
||||
return create_resp
|
||||
|
||||
monkeypatch.setattr(router, "_post", _fake_post)
|
||||
ws_id, is_new = await router.get_or_create_workstream("discord", "ch-1", name="test")
|
||||
assert ws_id == "ws-resumed"
|
||||
assert is_new is True
|
||||
# Should have deleted the stale route and created a new one.
|
||||
mock_storage.delete_channel_route.assert_called_once_with("discord", "ch-1")
|
||||
mock_storage.create_channel_route.assert_called_once_with("discord", "ch-1", "ws-resumed")
|
||||
# The create body should include resume_ws pointing at the old ws.
|
||||
create_call = captured[0]
|
||||
assert create_call["body"]["resume_ws"] == "ws-stale"
|
||||
# The create call should include resume_ws pointing at the old ws.
|
||||
mock_create.assert_awaited_once()
|
||||
call_kwargs = mock_create.call_args[1]
|
||||
assert call_kwargs["resume_ws"] == "ws-stale"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_sends_initial_message_for_new_workstream(
|
||||
self,
|
||||
router: ChannelRouter,
|
||||
mock_storage: MagicMock,
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
assert router._server is not None
|
||||
mock_create = AsyncMock()
|
||||
mock_create.return_value = MagicMock(ws_id="ws-new", name="test")
|
||||
monkeypatch.setattr(router._server, "create_workstream", mock_create)
|
||||
mock_send = AsyncMock()
|
||||
monkeypatch.setattr(router._server, "send", mock_send)
|
||||
|
||||
await router.get_or_create_workstream("discord", "ch-1", name="test", initial_message="hi")
|
||||
mock_send.assert_awaited_once_with("hi", "ws-new")
|
||||
|
||||
|
||||
class TestCloseWorkstream:
|
||||
@pytest.mark.anyio
|
||||
async def test_posts_to_server(
|
||||
async def test_calls_server_close(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
mock_post = AsyncMock(return_value=_ok_response())
|
||||
monkeypatch.setattr(router, "_post", mock_post)
|
||||
assert router._server is not None
|
||||
mock_close = AsyncMock()
|
||||
monkeypatch.setattr(router._server, "close_workstream", mock_close)
|
||||
await router.close_workstream("ws-1")
|
||||
mock_post.assert_awaited_once_with(
|
||||
"/api/workstreams/close",
|
||||
{"ws_id": "ws-1"},
|
||||
)
|
||||
mock_close.assert_awaited_once_with("ws-1")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_catches_api_error(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert router._server is not None
|
||||
mock_close = AsyncMock(side_effect=TurnstoneAPIError(404, "not found"))
|
||||
monkeypatch.setattr(router._server, "close_workstream", mock_close)
|
||||
# Should not raise.
|
||||
await router.close_workstream("ws-1")
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_calls_console_route_close(
|
||||
self, console_router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_close = AsyncMock()
|
||||
monkeypatch.setattr(console_router._console, "route_close", mock_close)
|
||||
await console_router.close_workstream("ws-1")
|
||||
mock_close.assert_awaited_once_with("ws-1")
|
||||
|
||||
|
||||
class TestAclose:
|
||||
@pytest.mark.anyio
|
||||
async def test_closes_client(
|
||||
async def test_closes_server_client(
|
||||
self, router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert router._server is not None
|
||||
mock_close = AsyncMock()
|
||||
monkeypatch.setattr(router._client, "aclose", mock_close)
|
||||
monkeypatch.setattr(router._server, "aclose", mock_close)
|
||||
await router.aclose()
|
||||
mock_close.assert_awaited_once()
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_closes_console_client(
|
||||
self, console_router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_close = AsyncMock()
|
||||
monkeypatch.setattr(console_router._console, "aclose", mock_close)
|
||||
await console_router.aclose()
|
||||
mock_close.assert_awaited_once()
|
||||
|
||||
|
||||
class TestGetNodeUrl:
|
||||
@pytest.mark.anyio
|
||||
async def test_returns_cached_url(self, router: ChannelRouter) -> None:
|
||||
router._node_urls["ws-1"] = "http://node1:8080/v1"
|
||||
url = await router.get_node_url("ws-1")
|
||||
assert url == "http://node1:8080/v1"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_falls_back_to_server_url(self, router: ChannelRouter) -> None:
|
||||
url = await router.get_node_url("ws-unknown")
|
||||
assert url == "http://localhost:8080/v1"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_queries_console_route_lookup(
|
||||
self, console_router: ChannelRouter, monkeypatch: pytest.MonkeyPatch
|
||||
) -> None:
|
||||
assert console_router._console is not None
|
||||
mock_lookup = AsyncMock(return_value={"node_url": "http://node2:8080/v1", "node_id": "n2"})
|
||||
monkeypatch.setattr(console_router._console, "route_lookup", mock_lookup)
|
||||
url = await console_router.get_node_url("ws-1")
|
||||
assert url == "http://node2:8080/v1"
|
||||
mock_lookup.assert_awaited_once_with("ws-1")
|
||||
# Should be cached now.
|
||||
assert console_router._node_urls["ws-1"] == "http://node2:8080/v1"
|
||||
|
||||
@@ -21,20 +21,20 @@ def test_load_config_missing_file(tmp_path):
|
||||
def test_load_config_valid_toml(tmp_path):
|
||||
_reset_cache()
|
||||
cfg = tmp_path / "config.toml"
|
||||
cfg.write_text('[redis]\nhost = "10.0.0.1"\nport = 6380\npassword = "secret"\n')
|
||||
cfg.write_text('[database]\nhost = "10.0.0.1"\nport = 5432\nname = "turnstone"\n')
|
||||
set_config_path(str(cfg))
|
||||
result = load_config()
|
||||
assert result["redis"]["host"] == "10.0.0.1"
|
||||
assert result["redis"]["port"] == 6380
|
||||
assert result["redis"]["password"] == "secret"
|
||||
assert result["database"]["host"] == "10.0.0.1"
|
||||
assert result["database"]["port"] == 5432
|
||||
assert result["database"]["name"] == "turnstone"
|
||||
|
||||
|
||||
def test_load_config_section(tmp_path):
|
||||
_reset_cache()
|
||||
cfg = tmp_path / "config.toml"
|
||||
cfg.write_text('[api]\nbase_url = "http://x:8000/v1"\n[redis]\nhost = "y"\n')
|
||||
cfg.write_text('[api]\nbase_url = "http://x:8000/v1"\n[database]\nhost = "y"\n')
|
||||
set_config_path(str(cfg))
|
||||
assert load_config("redis") == {"host": "y"}
|
||||
assert load_config("database") == {"host": "y"}
|
||||
assert load_config("api") == {"base_url": "http://x:8000/v1"}
|
||||
assert load_config("nonexistent") == {}
|
||||
|
||||
|
||||
@@ -1723,7 +1723,7 @@ class TestCreateWorkstreamUserIdTrust:
|
||||
"""Replicate the trust check from server.py:create_workstream."""
|
||||
auth = auth_result
|
||||
uid: str = getattr(auth, "user_id", "") or ""
|
||||
trusted_sources = {"bridge", "console"}
|
||||
trusted_sources = {"console"}
|
||||
if (
|
||||
body.get("user_id")
|
||||
and isinstance(body["user_id"], str)
|
||||
@@ -1733,13 +1733,13 @@ class TestCreateWorkstreamUserIdTrust:
|
||||
uid = body["user_id"]
|
||||
return uid
|
||||
|
||||
def test_bridge_can_forward_user_id(self):
|
||||
def test_console_can_forward_user_id(self):
|
||||
from turnstone.core.auth import AuthResult
|
||||
|
||||
auth = AuthResult(
|
||||
user_id="bridge",
|
||||
user_id="console",
|
||||
scopes=frozenset({"approve"}),
|
||||
token_source="bridge",
|
||||
token_source="console",
|
||||
)
|
||||
uid = self._extract_uid({"user_id": "real-user-abc"}, auth)
|
||||
assert uid == "real-user-abc"
|
||||
|
||||
+65
-49
@@ -8,6 +8,7 @@ from unittest.mock import MagicMock, patch
|
||||
import pytest
|
||||
|
||||
from turnstone.console.scheduler import TaskScheduler
|
||||
from turnstone.sdk._types import TurnstoneAPIError
|
||||
|
||||
|
||||
def _wire_lock_storage(storage: MagicMock, initial: dict[str, str] | None = None) -> None:
|
||||
@@ -78,6 +79,13 @@ def _make_node(node_id="node-001", reachable=True, ws_total=2, max_ws=10):
|
||||
}
|
||||
|
||||
|
||||
def _mock_create_response(ws_id: str = "ws_abc123") -> MagicMock:
|
||||
"""Build a mock CreateWorkstreamResponse with the given ws_id."""
|
||||
resp = MagicMock()
|
||||
resp.ws_id = ws_id
|
||||
return resp
|
||||
|
||||
|
||||
class TestSchedulerTick:
|
||||
"""Tests for _tick() lock acquisition and dispatch logic."""
|
||||
|
||||
@@ -135,18 +143,18 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
mock_post.assert_called_once()
|
||||
url = mock_post.call_args[0][0]
|
||||
assert "http://node-001:8080/v1/api/workstreams/new" in url
|
||||
mock_create.assert_called_once()
|
||||
storage.record_task_run.assert_called_once()
|
||||
run_kwargs = storage.record_task_run.call_args[1]
|
||||
assert run_kwargs["node_id"] == "node-001"
|
||||
assert run_kwargs["status"] == "dispatched"
|
||||
assert run_kwargs["ws_id"] == "ws_abc123"
|
||||
|
||||
def test_dispatch_pool_mode(self, mocks):
|
||||
collector, storage = mocks
|
||||
@@ -159,12 +167,13 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
mock_post.assert_called_once()
|
||||
mock_create.assert_called_once()
|
||||
storage.record_task_run.assert_called_once()
|
||||
|
||||
def test_dispatch_all_mode(self, mocks):
|
||||
@@ -181,12 +190,13 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
assert mock_post.call_count == 2
|
||||
assert mock_create.call_count == 2
|
||||
assert storage.record_task_run.call_count == 2
|
||||
|
||||
def test_dispatch_specific_node(self, mocks):
|
||||
@@ -199,14 +209,15 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
mock_post.assert_called_once()
|
||||
url = mock_post.call_args[0][0]
|
||||
assert "node-001" in url
|
||||
mock_create.assert_called_once()
|
||||
run_kwargs = storage.record_task_run.call_args[1]
|
||||
assert run_kwargs["node_id"] == "node-001"
|
||||
|
||||
def test_at_task_disables_after_dispatch(self, mocks):
|
||||
collector, storage = mocks
|
||||
@@ -219,9 +230,10 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
):
|
||||
scheduler._tick()
|
||||
|
||||
# At-task should be disabled after dispatch
|
||||
@@ -243,9 +255,10 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
):
|
||||
scheduler._tick()
|
||||
|
||||
update_calls = storage.update_scheduled_task.call_args_list
|
||||
@@ -301,12 +314,13 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage, max_fan_out=3)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
assert mock_post.call_count == 3
|
||||
assert mock_create.call_count == 3
|
||||
assert storage.record_task_run.call_count == 3
|
||||
|
||||
def test_specific_node_target(self, mocks):
|
||||
@@ -320,17 +334,18 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
mock_post.assert_called_once()
|
||||
url = mock_post.call_args[0][0]
|
||||
assert "node-custom-123" in url
|
||||
mock_create.assert_called_once()
|
||||
run_kwargs = storage.record_task_run.call_args[1]
|
||||
assert run_kwargs["node_id"] == "node-custom-123"
|
||||
|
||||
def test_user_id_in_dispatched_body(self, mocks):
|
||||
"""Dispatched HTTP body should include created_by as user_id."""
|
||||
def test_user_id_in_dispatched_call(self, mocks):
|
||||
"""Dispatched SDK call should include created_by as user_id."""
|
||||
collector, storage = mocks
|
||||
|
||||
task = _make_task(target_mode="auto", created_by="u_scheduler_admin")
|
||||
@@ -341,18 +356,17 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.return_value = MagicMock(status_code=200)
|
||||
mock_post.return_value.raise_for_status = MagicMock()
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
return_value=_mock_create_response(),
|
||||
) as mock_create:
|
||||
scheduler._tick()
|
||||
|
||||
body = mock_post.call_args[1]["json"]
|
||||
assert body["user_id"] == "u_scheduler_admin"
|
||||
|
||||
def test_http_failure_records_failure(self, mocks):
|
||||
"""HTTP errors during dispatch should record a failure."""
|
||||
import httpx
|
||||
_, kwargs = mock_create.call_args
|
||||
assert kwargs["user_id"] == "u_scheduler_admin"
|
||||
|
||||
def test_sdk_failure_records_failure(self, mocks):
|
||||
"""SDK errors during dispatch should record a failure."""
|
||||
collector, storage = mocks
|
||||
|
||||
task = _make_task(target_mode="auto")
|
||||
@@ -363,8 +377,10 @@ class TestSchedulerTick:
|
||||
}
|
||||
|
||||
scheduler = TaskScheduler(collector, storage)
|
||||
with patch.object(scheduler._http_client, "post") as mock_post:
|
||||
mock_post.side_effect = httpx.ConnectError("connection refused")
|
||||
with patch(
|
||||
"turnstone.console.scheduler.TurnstoneServer.create_workstream",
|
||||
side_effect=TurnstoneAPIError(502, "Bad Gateway"),
|
||||
):
|
||||
scheduler._tick()
|
||||
|
||||
storage.record_task_run.assert_called_once()
|
||||
|
||||
@@ -52,11 +52,11 @@ class TestServiceRegistry:
|
||||
|
||||
def test_list_filters_by_type(self, storage):
|
||||
storage.register_service("channel", "ch-1", "http://localhost:8091")
|
||||
storage.register_service("bridge", "br-1", "http://localhost:8080")
|
||||
storage.register_service("worker", "wk-1", "http://localhost:8080")
|
||||
channels = storage.list_services("channel", max_age_seconds=120)
|
||||
bridges = storage.list_services("bridge", max_age_seconds=120)
|
||||
workers = storage.list_services("worker", max_age_seconds=120)
|
||||
assert len(channels) == 1
|
||||
assert len(bridges) == 1
|
||||
assert len(workers) == 1
|
||||
|
||||
def test_deregister(self, storage):
|
||||
storage.register_service("channel", "ch-1", "http://localhost:8091")
|
||||
|
||||
@@ -134,13 +134,13 @@ def test_cli_bootstrap(tmp_path):
|
||||
from turnstone.admin import _cmd_tls_bootstrap
|
||||
|
||||
out = tmp_path / "certs"
|
||||
args = argparse.Namespace(out=str(out), issue=["redis.internal", "pg.internal"])
|
||||
args = argparse.Namespace(out=str(out), issue=["app.internal", "pg.internal"])
|
||||
_cmd_tls_bootstrap(args)
|
||||
|
||||
assert (out / "ca.pem").exists()
|
||||
assert b"BEGIN CERTIFICATE" in (out / "ca.pem").read_bytes()
|
||||
# Check certs were issued
|
||||
assert (out / "certs" / "redis.internal").exists()
|
||||
assert (out / "certs" / "app.internal").exists()
|
||||
assert (out / "certs" / "pg.internal").exists()
|
||||
|
||||
|
||||
|
||||
@@ -53,17 +53,6 @@
|
||||
# sslcert = "" # path to client cert (mTLS)
|
||||
# sslkey = "" # path to client key (mTLS)
|
||||
|
||||
# --- Redis (bridge, console, channel) ---
|
||||
|
||||
[redis]
|
||||
# url = "" # redis://host:6379/0 or rediss://host:6380/0
|
||||
# env: TURNSTONE_REDIS_URL
|
||||
# TLS params (passed through to Redis connection):
|
||||
# tls = false # enable TLS (also auto-enabled by rediss:// scheme)
|
||||
# tls_ca = "" # path to CA cert
|
||||
# tls_cert = "" # path to client cert (mTLS)
|
||||
# tls_key = "" # path to client key (mTLS)
|
||||
|
||||
# --- Auth (node, console) ---
|
||||
|
||||
[auth]
|
||||
|
||||
@@ -1,17 +1,19 @@
|
||||
"""Channel router -- maps external channels/threads to turnstone workstreams.
|
||||
|
||||
:class:`ChannelRouter` uses direct HTTP calls to the turnstone server API
|
||||
and the storage backend for persistent channel-to-workstream mappings.
|
||||
:class:`ChannelRouter` uses the turnstone SDK clients to communicate with
|
||||
the server (single-node) or console (multi-node) API, and the storage
|
||||
backend for persistent channel-to-workstream mappings.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import httpx
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from turnstone.core.log import get_logger
|
||||
from turnstone.sdk._types import TurnstoneAPIError
|
||||
from turnstone.sdk.console import AsyncTurnstoneConsole
|
||||
from turnstone.sdk.server import AsyncTurnstoneServer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from turnstone.core.storage import StorageBackend
|
||||
@@ -22,7 +24,7 @@ _WS_CREATE_TIMEOUT = 30.0 # seconds
|
||||
|
||||
|
||||
class ChannelRouter:
|
||||
"""Manage channel-to-workstream routing via the server REST API.
|
||||
"""Manage channel-to-workstream routing via SDK clients.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@@ -58,42 +60,35 @@ class ChannelRouter:
|
||||
# Populated when console_url is set and the create response
|
||||
# includes node_url.
|
||||
self._node_urls: dict[str, str] = {}
|
||||
headers: dict[str, str] = {}
|
||||
if api_token:
|
||||
headers["Authorization"] = f"Bearer {api_token}"
|
||||
# When a console_url is configured, control-plane POSTs go to the
|
||||
# console's routing proxy; otherwise they go directly to the server.
|
||||
base = self._console_url if self._console_url else self._server_url
|
||||
self._client = httpx.AsyncClient(
|
||||
base_url=base,
|
||||
headers=headers,
|
||||
timeout=_WS_CREATE_TIMEOUT,
|
||||
)
|
||||
|
||||
# SDK clients: use console for multi-node, server for single-node.
|
||||
self._console: AsyncTurnstoneConsole | None = None
|
||||
self._server: AsyncTurnstoneServer | None = None
|
||||
if self._console_url:
|
||||
self._console = AsyncTurnstoneConsole(
|
||||
base_url=self._console_url,
|
||||
token=api_token,
|
||||
timeout=_WS_CREATE_TIMEOUT,
|
||||
)
|
||||
else:
|
||||
self._server = AsyncTurnstoneServer(
|
||||
base_url=self._server_url,
|
||||
token=api_token,
|
||||
timeout=_WS_CREATE_TIMEOUT,
|
||||
)
|
||||
|
||||
# -- lifecycle -----------------------------------------------------------
|
||||
|
||||
async def aclose(self) -> None:
|
||||
"""Close the underlying HTTP client."""
|
||||
await self._client.aclose()
|
||||
"""Close the underlying SDK clients."""
|
||||
if self._server:
|
||||
await self._server.aclose()
|
||||
if self._console:
|
||||
await self._console.aclose()
|
||||
log.info("channel_router.closed")
|
||||
|
||||
# -- internal helpers ----------------------------------------------------
|
||||
|
||||
def _route_path(self, path: str) -> str:
|
||||
"""Map a server API path to the console routing proxy path when needed.
|
||||
|
||||
E.g. ``/api/send`` → ``/api/route/send`` when console routing is active.
|
||||
"""
|
||||
if self._console_url and path.startswith("/api/"):
|
||||
return path.replace("/api/", "/api/route/", 1)
|
||||
return path
|
||||
|
||||
async def _post(self, path: str, body: dict[str, Any]) -> httpx.Response:
|
||||
"""POST JSON to the server (or console proxy) and return the response."""
|
||||
resp = await self._client.post(self._route_path(path), json=body)
|
||||
resp.raise_for_status()
|
||||
return resp
|
||||
|
||||
async def _is_ws_alive(self, ws_id: str) -> bool:
|
||||
"""Check whether *ws_id* is a known workstream.
|
||||
|
||||
@@ -157,18 +152,11 @@ class ChannelRouter:
|
||||
channel_id=channel_id,
|
||||
)
|
||||
|
||||
# 2. Create via HTTP API with atomic resume.
|
||||
# 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 ""
|
||||
body: dict[str, Any] = {
|
||||
"name": name,
|
||||
"model": model,
|
||||
"resume_ws": resume_ws,
|
||||
"skill": self._skill,
|
||||
"auto_approve": self._auto_approve,
|
||||
}
|
||||
log.info(
|
||||
"channel_router.creating_workstream",
|
||||
channel_type=channel_type,
|
||||
@@ -176,9 +164,26 @@ class ChannelRouter:
|
||||
resume_ws=resume_ws or None,
|
||||
)
|
||||
|
||||
resp = await self._post("/api/workstreams/new", body)
|
||||
data = resp.json()
|
||||
ws_id: str = data.get("ws_id", "")
|
||||
if self._console:
|
||||
data = await self._console.route_create_workstream(
|
||||
name=name,
|
||||
model=model,
|
||||
resume_ws=resume_ws,
|
||||
skill=self._skill,
|
||||
auto_approve=self._auto_approve,
|
||||
)
|
||||
ws_id = data.get("ws_id", "")
|
||||
else:
|
||||
assert self._server is not None
|
||||
resp = await self._server.create_workstream(
|
||||
name=name,
|
||||
model=model,
|
||||
resume_ws=resume_ws,
|
||||
skill=self._skill,
|
||||
auto_approve=self._auto_approve,
|
||||
)
|
||||
ws_id = resp.ws_id
|
||||
data = {"ws_id": resp.ws_id, "name": resp.name}
|
||||
|
||||
if not ws_id:
|
||||
msg_err = "workstream creation returned empty ws_id"
|
||||
@@ -192,7 +197,11 @@ class ChannelRouter:
|
||||
|
||||
# 3. Send the initial message if this is a brand-new workstream.
|
||||
if initial_message and not resume_ws:
|
||||
await self._post("/api/send", {"ws_id": ws_id, "message": initial_message})
|
||||
if self._console:
|
||||
await self._console.route_send(initial_message, ws_id)
|
||||
else:
|
||||
assert self._server is not None
|
||||
await self._server.send(initial_message, ws_id)
|
||||
|
||||
# 4. Persist the route.
|
||||
await asyncio.to_thread(
|
||||
@@ -218,14 +227,13 @@ class ChannelRouter:
|
||||
url = self._node_urls.get(ws_id)
|
||||
if url:
|
||||
return url
|
||||
if self._console_url:
|
||||
if self._console:
|
||||
try:
|
||||
resp = await self._client.get("/api/route", params={"ws_id": ws_id})
|
||||
if resp.status_code == 200:
|
||||
node_url = resp.json().get("node_url", "")
|
||||
if node_url:
|
||||
self._node_urls[ws_id] = node_url.rstrip("/")
|
||||
return self._node_urls[ws_id]
|
||||
data = await self._console.route_lookup(ws_id)
|
||||
node_url = data.get("node_url", "")
|
||||
if node_url:
|
||||
self._node_urls[ws_id] = node_url.rstrip("/")
|
||||
return self._node_urls[ws_id]
|
||||
except Exception:
|
||||
pass
|
||||
return self._server_url
|
||||
@@ -248,7 +256,11 @@ class ChannelRouter:
|
||||
|
||||
async def send_message(self, ws_id: str, message: str) -> None:
|
||||
"""Send a user message to a workstream via the server API."""
|
||||
await self._post("/api/send", {"ws_id": ws_id, "message": message})
|
||||
if self._console:
|
||||
await self._console.route_send(message, ws_id)
|
||||
else:
|
||||
assert self._server is not None
|
||||
await self._server.send(message, ws_id)
|
||||
log.debug("channel_router.send_message", ws_id=ws_id)
|
||||
|
||||
async def send_approval(
|
||||
@@ -260,14 +272,15 @@ class ChannelRouter:
|
||||
always: bool = False,
|
||||
) -> None:
|
||||
"""Approve or deny a pending tool call via the server API."""
|
||||
body: dict[str, Any] = {
|
||||
"ws_id": ws_id,
|
||||
"approved": approved,
|
||||
"always": always,
|
||||
}
|
||||
if feedback:
|
||||
body["feedback"] = feedback
|
||||
await self._post("/api/approve", body)
|
||||
if self._console:
|
||||
await self._console.route_approve(
|
||||
ws_id=ws_id, approved=approved, feedback=feedback, always=always
|
||||
)
|
||||
else:
|
||||
assert self._server is not None
|
||||
await self._server.approve(
|
||||
ws_id=ws_id, approved=approved, feedback=feedback or None, always=always
|
||||
)
|
||||
log.debug(
|
||||
"channel_router.send_approval",
|
||||
ws_id=ws_id,
|
||||
@@ -277,7 +290,11 @@ class ChannelRouter:
|
||||
|
||||
async def send_plan_feedback(self, ws_id: str, correlation_id: str, feedback: str) -> None:
|
||||
"""Respond to a plan review via the server API."""
|
||||
await self._post("/api/plan", {"ws_id": ws_id, "feedback": feedback})
|
||||
if self._console:
|
||||
await self._console.route_plan_feedback(ws_id=ws_id, feedback=feedback)
|
||||
else:
|
||||
assert self._server is not None
|
||||
await self._server.plan_feedback(ws_id=ws_id, feedback=feedback)
|
||||
log.debug(
|
||||
"channel_router.send_plan_feedback",
|
||||
ws_id=ws_id,
|
||||
@@ -302,11 +319,15 @@ class ChannelRouter:
|
||||
"""Close a workstream via the server API."""
|
||||
self._node_urls.pop(ws_id, None)
|
||||
try:
|
||||
await self._post("/api/workstreams/close", {"ws_id": ws_id})
|
||||
if self._console:
|
||||
await self._console.route_close(ws_id)
|
||||
else:
|
||||
assert self._server is not None
|
||||
await self._server.close_workstream(ws_id)
|
||||
log.info("channel_router.close_workstream", ws_id=ws_id)
|
||||
except httpx.HTTPStatusError as exc:
|
||||
except TurnstoneAPIError as exc:
|
||||
log.warning(
|
||||
"channel_router.close_workstream_failed",
|
||||
ws_id=ws_id,
|
||||
status=exc.response.status_code,
|
||||
status=exc.status_code,
|
||||
)
|
||||
|
||||
@@ -572,19 +572,24 @@ def _weight_based_assignments(nodes: list[RingNode]) -> list[tuple[int, str]]:
|
||||
algorithm won't try to "correct" on the next run.
|
||||
"""
|
||||
total_weight = sum(n.weight for n in nodes)
|
||||
assignments: list[tuple[int, str]] = []
|
||||
# Sort nodes for determinism
|
||||
# Compute per-node counts using the same int() + remainder distribution
|
||||
# as rebalance_once step 7, so seeding is a guaranteed noop on first rebalance.
|
||||
sorted_nodes = sorted(nodes, key=lambda n: n.node_id)
|
||||
counts: dict[str, int] = {}
|
||||
assigned = 0
|
||||
for n in sorted_nodes:
|
||||
c = int((n.weight / total_weight) * RING_SIZE)
|
||||
counts[n.node_id] = c
|
||||
assigned += c
|
||||
# Distribute remainder to heaviest nodes (same as rebalance_once step 7)
|
||||
remainder_pool = sorted(counts, key=lambda nid: counts[nid], reverse=True)
|
||||
for i in range(RING_SIZE - assigned):
|
||||
counts[remainder_pool[i % len(remainder_pool)]] += 1
|
||||
|
||||
assignments: list[tuple[int, str]] = []
|
||||
bucket = 0
|
||||
for i, node in enumerate(sorted_nodes):
|
||||
if i == len(sorted_nodes) - 1:
|
||||
# Last node gets the remainder (avoids rounding gaps)
|
||||
count = RING_SIZE - bucket
|
||||
else:
|
||||
count = round((node.weight / total_weight) * RING_SIZE)
|
||||
for _ in range(count):
|
||||
if bucket >= RING_SIZE:
|
||||
break
|
||||
for node in sorted_nodes:
|
||||
for _ in range(counts[node.node_id]):
|
||||
assignments.append((bucket, node.node_id))
|
||||
bucket += 1
|
||||
return assignments
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
"""Background task scheduler for timed workstream dispatch.
|
||||
|
||||
Runs as a daemon thread inside the console process. Checks for due tasks
|
||||
every ``check_interval`` seconds and dispatches them via HTTP POST to
|
||||
server nodes' ``/v1/api/workstreams/new`` endpoint.
|
||||
every ``check_interval`` seconds and dispatches them to server nodes via
|
||||
the :class:`~turnstone.sdk.server.TurnstoneServer` SDK client.
|
||||
|
||||
Uses a ``system_settings`` row for distributed locking in multi-console
|
||||
deployments.
|
||||
@@ -16,9 +16,10 @@ import uuid
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import httpx
|
||||
import structlog
|
||||
|
||||
from turnstone.sdk.server import TurnstoneServer
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from turnstone.console.collector import ClusterCollector
|
||||
from turnstone.core.auth import ServiceTokenManager
|
||||
@@ -67,7 +68,8 @@ class TaskScheduler:
|
||||
self._lock_owner = uuid.uuid4().hex
|
||||
self._api_token = api_token
|
||||
self._token_manager = token_manager
|
||||
self._http_client = httpx.Client(timeout=30)
|
||||
self._sdk_clients: dict[str, TurnstoneServer] = {}
|
||||
self._last_token: str = ""
|
||||
|
||||
def start(self) -> None:
|
||||
"""Start the scheduler daemon thread."""
|
||||
@@ -81,7 +83,9 @@ class TaskScheduler:
|
||||
self._stop_event.set()
|
||||
if self._thread is not None:
|
||||
self._thread.join(timeout=5)
|
||||
self._http_client.close()
|
||||
for client in self._sdk_clients.values():
|
||||
client.close()
|
||||
self._sdk_clients.clear()
|
||||
log.info("scheduler.stopped")
|
||||
|
||||
def _loop(self) -> None:
|
||||
@@ -183,6 +187,14 @@ class TaskScheduler:
|
||||
log.info("scheduler.pruned_audit", count=audit_pruned)
|
||||
except Exception:
|
||||
log.warning("scheduler.prune_audit_error", exc_info=True)
|
||||
# Prune SDK clients for nodes no longer in the cluster
|
||||
if self._sdk_clients and self._collector:
|
||||
live_urls = {n.get("server_url", "") for n in self._collector.get_all_nodes()}
|
||||
stale = [u for u in self._sdk_clients if u not in live_urls]
|
||||
for url in stale:
|
||||
self._sdk_clients.pop(url).close()
|
||||
if stale:
|
||||
log.info("scheduler.pruned_sdk_clients", count=len(stale))
|
||||
finally:
|
||||
self._release_lock()
|
||||
|
||||
@@ -252,17 +264,29 @@ class TaskScheduler:
|
||||
raw = task.get("auto_approve_tools", "")
|
||||
return [t.strip() for t in raw.split(",") if t.strip()]
|
||||
|
||||
def _auth_headers(self) -> dict[str, str]:
|
||||
"""Build auth headers for HTTP dispatch.
|
||||
def _get_sdk_client(self, node_url: str) -> TurnstoneServer:
|
||||
"""Return a cached :class:`TurnstoneServer` for *node_url*.
|
||||
|
||||
Prefers a :class:`ServiceTokenManager` (auto-rotating JWT) over a
|
||||
static API token. Returns an empty dict when neither is configured.
|
||||
When a :class:`ServiceTokenManager` is configured, the client is
|
||||
re-created whenever the token rotates so that fresh JWTs are used.
|
||||
"""
|
||||
token = self._api_token
|
||||
if self._token_manager is not None:
|
||||
return dict(self._token_manager.bearer_header)
|
||||
if self._api_token:
|
||||
return {"Authorization": f"Bearer {self._api_token}"}
|
||||
return {}
|
||||
token = self._token_manager.token
|
||||
|
||||
if token != self._last_token:
|
||||
# Token rotated — close all stale clients.
|
||||
for client in self._sdk_clients.values():
|
||||
client.close()
|
||||
self._sdk_clients.clear()
|
||||
self._last_token = token
|
||||
|
||||
if node_url not in self._sdk_clients:
|
||||
self._sdk_clients[node_url] = TurnstoneServer(
|
||||
base_url=node_url,
|
||||
token=token,
|
||||
)
|
||||
return self._sdk_clients[node_url]
|
||||
|
||||
def _get_node_url(self, node_id: str) -> str:
|
||||
"""Resolve a node_id to its server URL via the collector."""
|
||||
@@ -273,39 +297,35 @@ class TaskScheduler:
|
||||
return ""
|
||||
|
||||
def _dispatch_to_node(self, task: dict[str, Any], node_id: str, now: str) -> None:
|
||||
"""POST to a specific node's /v1/api/workstreams/new endpoint."""
|
||||
"""Dispatch a workstream to a specific node via the SDK client."""
|
||||
server_url = self._get_node_url(node_id)
|
||||
if not server_url:
|
||||
self._record_failure(task, now, f"No URL for node {node_id}")
|
||||
return
|
||||
|
||||
correlation_id = uuid.uuid4().hex
|
||||
body: dict[str, Any] = {
|
||||
"name": task["name"],
|
||||
"model": task.get("model", ""),
|
||||
"initial_message": task["initial_message"],
|
||||
"auto_approve": bool(task.get("auto_approve", 0)),
|
||||
"auto_approve_tools": ",".join(self._parse_tools(task)),
|
||||
"user_id": task.get("created_by", ""),
|
||||
"skill": task.get("skill", ""),
|
||||
}
|
||||
try:
|
||||
resp = self._http_client.post(
|
||||
f"{server_url.rstrip('/')}/v1/api/workstreams/new",
|
||||
json=body,
|
||||
headers=self._auth_headers(),
|
||||
client = self._get_sdk_client(server_url)
|
||||
resp = client.create_workstream(
|
||||
name=task["name"],
|
||||
model=task.get("model", ""),
|
||||
initial_message=task["initial_message"],
|
||||
auto_approve=bool(task.get("auto_approve", 0)),
|
||||
auto_approve_tools=",".join(self._parse_tools(task)),
|
||||
user_id=task.get("created_by", ""),
|
||||
skill=task.get("skill", ""),
|
||||
)
|
||||
resp.raise_for_status()
|
||||
ws_id = resp.ws_id
|
||||
except Exception:
|
||||
self._record_failure(task, now, f"HTTP dispatch to {node_id} failed")
|
||||
log.warning("scheduler.http_dispatch_failed", node_id=node_id, exc_info=True)
|
||||
self._record_failure(task, now, f"SDK dispatch to {node_id} failed")
|
||||
log.warning("scheduler.sdk_dispatch_failed", node_id=node_id, exc_info=True)
|
||||
return
|
||||
|
||||
self._storage.record_task_run(
|
||||
run_id=uuid.uuid4().hex,
|
||||
task_id=task["task_id"],
|
||||
node_id=node_id,
|
||||
ws_id="",
|
||||
ws_id=ws_id,
|
||||
correlation_id=correlation_id,
|
||||
started=now,
|
||||
status="dispatched",
|
||||
|
||||
@@ -5736,6 +5736,7 @@ def create_app(
|
||||
Route("/api/route/approve", route_proxy, methods=["POST"]),
|
||||
Route("/api/route/cancel", route_proxy, methods=["POST"]),
|
||||
Route("/api/route/command", route_proxy, methods=["POST"]),
|
||||
Route("/api/route/plan", route_proxy, methods=["POST"]),
|
||||
Route("/api/route/workstreams/close", route_proxy, methods=["POST"]),
|
||||
Route("/api/route", route_lookup, methods=["GET"]),
|
||||
Route("/api/models", list_available_models),
|
||||
|
||||
Reference in New Issue
Block a user