Add Python and TypeScript client SDKs for server and console APIs (#19)

* Add Python and TypeScript client SDKs for server and console APIs

Python SDK (turnstone/sdk/) with sync + async clients for both server
and console APIs. Returns Pydantic models directly, streams SSE events
as typed dataclasses. 27 event types with registry-based deserialization.
High-level send_and_wait() for request-response patterns.

TypeScript SDK (sdk/typescript/) with zero browser dependencies. Uses
fetch + ReadableStream for SSE parsing. Discriminated union event types
with type guards. Same API surface as Python SDK.

63 Python tests, 21 TypeScript tests (vitest). Comprehensive docs at
docs/sdk.md with SDK architecture diagram.

* Address PR #19 review feedback + fix lint

- Fix consume_task leak in send_and_wait when send() raises (try/finally)
- Fix TS sendAndWait: open SSE before send, plumb AbortSignal for timeout
- Add signal param to TS streamSSE for cancellation support
- Fix SSE parser: join multi-line data: fields with \n per spec, handle CRLF
- Fix generate-types.py sys.path (parents[3] not parents[2])
- Document token ignored when httpx_client provided
- Document TS timeout units as milliseconds
- Fix stale docstring in test_sdk_sse.py
- Fix import sorting (ruff I001)
This commit is contained in:
Patrick Buckley
2026-03-03 21:22:24 -08:00
committed by GitHub
parent 62a4ceac96
commit 5ee539c983
41 changed files with 7658 additions and 3 deletions
+28
View File
@@ -26,6 +26,34 @@ All API endpoints use the `/v1/` prefix. Non-API endpoints (`/`, `/health`, `/me
- **OpenAPI spec**: `GET /openapi.json` — machine-readable OpenAPI 3.1 schema
- **Swagger UI**: `GET /docs` — interactive API explorer (loads from CDN)
### Client SDKs
Typed client libraries for programmatic access to both the server and console APIs.
**Python** (included in the `turnstone` package):
```python
from turnstone.sdk import TurnstoneServer
with TurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
ws = client.create_workstream(name="demo")
result = client.send_and_wait("Hello!", ws.ws_id)
print(result.content)
```
Async variant: `AsyncTurnstoneServer` / `AsyncTurnstoneConsole`.
**TypeScript** (`sdk/typescript/`):
```typescript
import { TurnstoneServer } from "@turnstone/sdk";
const client = new TurnstoneServer({ baseUrl: "http://localhost:8080", token: "tok_xxx" });
const ws = await client.createWorkstream({ name: "demo" });
const result = await client.sendAndWait("Hello!", ws.ws_id);
console.log(result.content);
```
---
## Endpoints
+54 -1
View File
@@ -50,11 +50,26 @@ turnstone/
safety.py Command safety validation (blocked patterns, sanitization)
sandbox.py Math code sandboxing (AST validation, subprocess execution)
web.py Web utilities (HTML stripping, SSRF prevention)
api/
schemas.py Shared Pydantic v2 models (auth, errors, WorkstreamState)
server_schemas.py Server endpoint request/response models
console_schemas.py Console endpoint request/response models
openapi.py OpenAPI 3.1 spec builder
server_spec.py Server endpoint catalog → build_server_spec()
console_spec.py Console endpoint catalog → build_console_spec()
docs.py /openapi.json + /docs (Swagger UI) handler factories
sdk/
server.py AsyncTurnstoneServer + TurnstoneServer (HTTP client)
console.py AsyncTurnstoneConsole + TurnstoneConsole (HTTP client)
events.py 27 SSE event dataclasses with type registry
_base.py Shared httpx async client, auth, error handling
_sync.py Background event loop for sync wrappers
_types.py TurnResult + TurnstoneAPIError
mq/
protocol.py Inbound/outbound message dataclasses (JSON serialization)
broker.py Abstract MessageBroker protocol + RedisBroker
bridge.py Bridge service (queue ↔ turnstone-server HTTP API)
client.py TurnstoneClient library + TurnResult for external systems
client.py TurnstoneClient library + TurnResult for MQ-based access
console/
collector.py ClusterCollector — aggregates state from all nodes via Redis + HTTP
server.py Cluster dashboard HTTP server + SSE + CLI entry point
@@ -1053,3 +1068,41 @@ preserves:
After compaction, `_read_files` is cleared to force re-reads before edits,
since file contents are no longer in the message history.
---
## Client SDK
> See also: [SDK Architecture diagram](diagrams/png/13-sdk-architecture.png) | [SDK Documentation](sdk.md)
The `turnstone/sdk/` package provides typed HTTP clients for programmatic access
to both the server and console APIs. It wraps REST endpoints with methods that
return Pydantic models, and SSE endpoints with async/sync iterators that yield
typed event dataclasses.
**Two client pairs** (sync + async):
- `TurnstoneServer` / `AsyncTurnstoneServer` — server API (workstreams, chat, streaming, sessions)
- `TurnstoneConsole` / `AsyncTurnstoneConsole` — console API (cluster overview, nodes, workstreams)
**Design**: async-first with thin sync wrappers. `_BaseClient` provides httpx
setup, auth headers, `_request()` (REST) and `_stream_sse()` (SSE). Sync
clients delegate through `_SyncRunner` which maintains a persistent background
event loop on a daemon thread.
**Event types**: 27 standalone dataclasses in `events.py` with a type-registry
pattern matching `OutboundEvent.from_json()` from `mq/protocol.py`. Events are
decoupled from the MQ package so SDK consumers don't need the `redis` dependency.
**TypeScript SDK**: `sdk/typescript/` — separate npm package with the same API
surface. Zero browser dependencies, SSE via `fetch` + `ReadableStream` parsing.
```python
# Python quick start
from turnstone.sdk import TurnstoneServer
with TurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
ws = client.create_workstream(name="demo")
result = client.send_and_wait("Hello!", ws.ws_id)
print(result.content)
```
+37
View File
@@ -11,6 +11,8 @@ skinparam component {
BackgroundColor<<console>> #B2EBF2
BackgroundColor<<ui>> #F0F4C3
BackgroundColor<<artifact>> #ECEFF1
BackgroundColor<<sdk>> #FFCDD2
BackgroundColor<<api>> #D1C4E9
}
' Entry points
@@ -73,6 +75,23 @@ package "turnstone/ui/" <<Rectangle>> {
component [spinner.py\nTerminal spinner] as spinner <<ui>>
}
' API schemas
package "turnstone/api/" <<Rectangle>> {
component [schemas.py\nShared Pydantic models] as apischemas <<api>>
component [server_spec.py\nServer OpenAPI spec] as serverspec <<api>>
component [console_spec.py\nConsole OpenAPI spec] as consolespec <<api>>
component [openapi.py\nSpec builder] as openapi <<api>>
component [docs.py\nSwagger UI handler] as apidocs <<api>>
}
' SDK
package "turnstone/sdk/" <<Rectangle>> {
component [server.py\nTurnstoneServer (sync+async)] as sdkserver <<sdk>>
component [console.py\nTurnstoneConsole (sync+async)] as sdkconsole <<sdk>>
component [events.py\n27 SSE event types] as sdkevents <<sdk>>
component [_base.py\nhttpx client base] as sdkbase <<sdk>>
}
' Tool schemas
package "turnstone/tools/" <<Rectangle>> {
component [*.json\n14 tool schemas] as schemas <<artifact>>
@@ -152,4 +171,22 @@ consoleserver --> config
consoleserver --> auth
collector --> broker
' API dependencies
serverspec --> openapi
consolespec --> openapi
serverspec --> apischemas
consolespec --> apischemas
server --> apidocs
server --> serverspec
consoleserver --> apidocs
consoleserver --> consolespec
' SDK dependencies
sdkserver --> sdkbase
sdkconsole --> sdkbase
sdkserver --> sdkevents
sdkconsole --> sdkevents
sdkserver --> apischemas : returns models
sdkconsole --> apischemas : returns models
@enduml
+155
View File
@@ -0,0 +1,155 @@
@startuml
!theme plain
title Turnstone — Client SDK Architecture
skinparam class {
BackgroundColor<<async>> #C8E6C9
BackgroundColor<<sync>> #B8D4E3
BackgroundColor<<event>> #FFE0B2
BackgroundColor<<type>> #F0F4C3
BackgroundColor<<ts>> #E1BEE7
}
skinparam packageBorderColor #888888
skinparam ArrowColor #555555
' Python SDK
package "turnstone/sdk/ (Python)" {
abstract class _BaseClient <<async>> {
- _client: httpx.AsyncClient
- _owns_client: bool
+ _request(method, path, ...) → T
+ _stream_sse(path, ...) → AsyncIterator
+ aclose()
}
class AsyncTurnstoneServer <<async>> {
+ list_workstreams()
+ dashboard()
+ create_workstream()
+ close_workstream()
+ send(message, ws_id)
+ approve()
+ plan_feedback()
+ command()
+ stream_events(ws_id)
+ stream_global_events()
+ send_and_wait()
+ list_sessions()
+ login() / logout()
+ health()
}
class AsyncTurnstoneConsole <<async>> {
+ overview()
+ nodes()
+ workstreams()
+ node_detail()
+ create_workstream()
+ stream_cluster_events()
+ login() / logout()
+ health()
}
class TurnstoneServer <<sync>> {
- _async: AsyncTurnstoneServer
- _runner: _SyncRunner
.. delegates all methods ..
+ __enter__ / __exit__
}
class TurnstoneConsole <<sync>> {
- _async: AsyncTurnstoneConsole
- _runner: _SyncRunner
.. delegates all methods ..
+ __enter__ / __exit__
}
class _SyncRunner <<sync>> {
- _loop: EventLoop
- _thread: Thread
+ run(coro) → T
+ run_iter(async_gen) → Iterator
+ close()
}
class TurnResult <<type>> {
+ ws_id: str
+ content_parts: list[str]
+ reasoning_parts: list[str]
+ tool_results: list
+ errors: list[str]
+ timed_out: bool
--
+ content: str
+ reasoning: str
+ ok: bool
}
class ServerEvent <<event>> {
+ type: str
+ ws_id: str
+ from_dict() → ServerEvent
}
class ClusterEvent <<event>> {
+ type: str
+ from_dict() → ClusterEvent
}
_BaseClient <|-- AsyncTurnstoneServer
_BaseClient <|-- AsyncTurnstoneConsole
TurnstoneServer --> AsyncTurnstoneServer : wraps
TurnstoneServer --> _SyncRunner : uses
TurnstoneConsole --> AsyncTurnstoneConsole : wraps
TurnstoneConsole --> _SyncRunner : uses
AsyncTurnstoneServer ..> TurnResult : returns
AsyncTurnstoneServer ..> ServerEvent : yields
AsyncTurnstoneConsole ..> ClusterEvent : yields
}
' TypeScript SDK
package "sdk/typescript/ (TypeScript)" {
class "BaseClient" as TSBase <<ts>> {
# baseUrl: string
# token: string
# fetchFn: fetch
# request<T>()
# streamSSE<T>()
}
class "TurnstoneServer" as TSServer <<ts>> {
+ listWorkstreams()
+ send()
+ streamEvents()
+ sendAndWait()
...
}
class "TurnstoneConsole" as TSConsole <<ts>> {
+ overview()
+ nodes()
+ clusterEvents()
...
}
TSBase <|-- TSServer
TSBase <|-- TSConsole
}
' External connections
class "turnstone-server :8080" as Server <<artifact>>
class "turnstone-console :8081" as Console <<artifact>>
AsyncTurnstoneServer --> Server : httpx REST + SSE
AsyncTurnstoneConsole --> Console : httpx REST + SSE
TSServer --> Server : fetch REST + SSE
TSConsole --> Console : fetch REST + SSE
note right of AsyncTurnstoneServer
Returns Pydantic models from
turnstone.api.server_schemas
(no type duplication)
end note
@enduml
+2 -2
View File
@@ -1,3 +1,3 @@
version https://git-lfs.github.com/spec/v1
oid sha256:29534422fc31eee613f70a479aa14de5278b98c49bb75fce7a63b72e248f1149
size 323269
oid sha256:ab184b57aff615d64082434faf09ec444bd2f26643269c37d3b51fa6868b45da
size 326359
@@ -0,0 +1,3 @@
version https://git-lfs.github.com/spec/v1
oid sha256:c9823a41e09611c5c0530d9fc12ad4139cfcc3ae238dc665b2888ec94d7d6781
size 195708
+258
View File
@@ -0,0 +1,258 @@
# Turnstone Client SDK
> See also: [API Reference](api-reference.md) | [Architecture](architecture.md) | [SDK Class Diagram](diagrams/png/13-sdk-architecture.png)
Typed HTTP client libraries for programmatic access to the turnstone server and console APIs. Available in Python (sync + async) and TypeScript.
---
## Python SDK
The Python SDK is included in the `turnstone` package — no extra install required. It wraps the REST and SSE endpoints with typed methods that return Pydantic models directly.
### Quick Start
```python
from turnstone.sdk import TurnstoneServer
# Synchronous client
with TurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
# Create a workstream
ws = client.create_workstream(name="Analysis")
# Send a message and wait for the full response
result = client.send_and_wait("Summarize this codebase.", ws.ws_id)
print(result.content)
# Stream events in real time
for event in client.stream_events(ws.ws_id):
if event.type == "content":
print(event.text, end="", flush=True)
# Close when done
client.close_workstream(ws.ws_id)
```
### Async Client
```python
import asyncio
from turnstone.sdk import AsyncTurnstoneServer
async def main():
async with AsyncTurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
ws = await client.create_workstream(name="demo")
async for event in client.stream_events(ws.ws_id):
if event.type == "content":
print(event.text, end="", flush=True)
asyncio.run(main())
```
### Server Client API
Both `TurnstoneServer` (sync) and `AsyncTurnstoneServer` (async) expose:
| Category | Method | Returns |
|----------|--------|---------|
| **Workstreams** | `list_workstreams()` | `ListWorkstreamsResponse` |
| | `dashboard()` | `DashboardResponse` |
| | `create_workstream(*, name, model, auto_approve)` | `CreateWorkstreamResponse` |
| | `close_workstream(ws_id)` | `StatusResponse` |
| **Chat** | `send(message, ws_id)` | `SendResponse` |
| | `approve(*, ws_id, approved, feedback, always)` | `StatusResponse` |
| | `plan_feedback(*, ws_id, feedback)` | `StatusResponse` |
| | `command(*, ws_id, command)` | `StatusResponse` |
| **Streaming** | `stream_events(ws_id)` | `Iterator[ServerEvent]` |
| | `stream_global_events()` | `Iterator[ServerEvent]` |
| **High-level** | `send_and_wait(message, ws_id, *, timeout, on_event)` | `TurnResult` |
| **Sessions** | `list_sessions()` | `ListSessionsResponse` |
| **Auth** | `login(token)` | `AuthLoginResponse` |
| | `logout()` | `StatusResponse` |
| **Health** | `health()` | `HealthResponse` |
### Console Client API
Both `TurnstoneConsole` (sync) and `AsyncTurnstoneConsole` (async) expose:
| Category | Method | Returns |
|----------|--------|---------|
| **Cluster** | `overview()` | `ClusterOverviewResponse` |
| | `nodes(*, sort, limit, offset)` | `ClusterNodesResponse` |
| | `workstreams(*, state, node, search, sort, page, per_page)` | `ClusterWorkstreamsResponse` |
| | `node_detail(node_id)` | `NodeDetailResponse` |
| | `create_workstream(*, node_id, name, model)` | `ConsoleCreateWsResponse` |
| **Streaming** | `stream_cluster_events()` | `Iterator[ClusterEvent]` |
| **Auth** | `login(token)` / `logout()` | `AuthLoginResponse` / `StatusResponse` |
| **Health** | `health()` | `ConsoleHealthResponse` |
### Event Types
SSE events are deserialized into typed dataclasses. Use `event.type` to discriminate.
**Per-workstream events** (from `stream_events(ws_id)`):
| Type | Class | Key Fields |
|------|-------|------------|
| `connected` | `ConnectedEvent` | `model`, `model_alias`, `skip_permissions` |
| `history` | `HistoryEvent` | `messages` |
| `content` | `ContentEvent` | `text` |
| `reasoning` | `ReasoningEvent` | `text` |
| `tool_info` | `ToolInfoEvent` | `items` |
| `approve_request` | `ApproveRequestEvent` | `items` |
| `tool_result` | `ToolResultEvent` | `call_id`, `name`, `output` |
| `tool_output_chunk` | `ToolOutputChunkEvent` | `call_id`, `chunk` |
| `status` | `StatusEvent` | `prompt_tokens`, `total_tokens`, `pct`, `effort` |
| `plan_review` | `PlanReviewEvent` | `content` |
| `error` | `ErrorEvent` | `message` |
| `info` | `InfoEvent` | `message` |
| `stream_end` | `StreamEndEvent` | — |
**Global events** (from `stream_global_events()`):
| Type | Class | Key Fields |
|------|-------|------------|
| `ws_state` | `WsStateEvent` | `ws_id`, `state`, `tokens`, `activity` |
| `ws_activity` | `WsActivityEvent` | `ws_id`, `activity`, `activity_state` |
| `ws_rename` | `WsRenameEvent` | `ws_id`, `name` |
| `ws_closed` | `WsClosedEvent` | `ws_id` |
**Cluster events** (from `stream_cluster_events()`):
| Type | Class | Key Fields |
|------|-------|------------|
| `node_joined` | `NodeJoinedEvent` | `node_id` |
| `node_lost` | `NodeLostEvent` | `node_id` |
| `cluster_state` | `ClusterStateEvent` | `ws_id`, `node_id`, `state`, `tokens` |
| `ws_created` | `ClusterWsCreatedEvent` | `ws_id`, `node_id`, `name` |
### TurnResult
The `send_and_wait()` method returns a `TurnResult` that aggregates the full response:
```python
result = client.send_and_wait("Hello", ws_id, timeout=60)
result.content # Full text response
result.reasoning # Chain-of-thought (if shown)
result.tool_results # List of (tool_name, output) tuples
result.errors # Any error messages
result.ok # True if no errors and not timed out
result.timed_out # True if timeout expired
```
### Error Handling
Non-2xx responses raise `TurnstoneAPIError`:
```python
from turnstone.sdk import TurnstoneServer, TurnstoneAPIError
try:
client.send("hi", "bad_ws_id")
except TurnstoneAPIError as e:
print(e.status_code) # 404
print(e.message) # "Unknown workstream"
```
---
## TypeScript SDK
Located at `sdk/typescript/`. Zero runtime dependencies for browsers; uses native `fetch` and `ReadableStream` for SSE parsing.
### Quick Start
```typescript
import { TurnstoneServer } from "@turnstone/sdk";
const client = new TurnstoneServer({
baseUrl: "http://localhost:8080",
token: "tok_xxx",
});
// Create workstream and send message
const ws = await client.createWorkstream({ name: "demo" });
const result = await client.sendAndWait("Hello!", ws.ws_id);
console.log(result.content);
// Stream events
for await (const event of client.streamEvents(ws.ws_id)) {
if (event.type === "content") {
process.stdout.write(event.text);
}
}
```
### Console Client
```typescript
import { TurnstoneConsole } from "@turnstone/sdk";
const console = new TurnstoneConsole({
baseUrl: "http://localhost:8081",
token: "tok_xxx",
});
const overview = await console.overview();
console.log(`Nodes: ${overview.nodes}, Workstreams: ${overview.workstreams}`);
// Stream cluster events
for await (const event of console.clusterEvents()) {
console.log(event.type, event);
}
```
### Type Safety
All event types are modeled as a discriminated union:
```typescript
import { isContentEvent, isErrorEvent } from "@turnstone/sdk";
import type { ServerEvent } from "@turnstone/sdk";
function handleEvent(event: ServerEvent) {
if (isContentEvent(event)) {
// event is narrowed to ContentEvent
console.log(event.text);
} else if (isErrorEvent(event)) {
console.error(event.message);
}
}
```
### Custom Fetch
The client accepts a custom `fetch` implementation for testing or Node.js environments:
```typescript
const client = new TurnstoneServer({
baseUrl: "http://localhost:8080",
fetch: myCustomFetch,
});
```
---
## Architecture
```
turnstone/sdk/ Python SDK (sub-package)
_base.py Shared httpx async client, auth, error handling
_sync.py Background event loop for sync wrappers
_types.py TurnResult + TurnstoneAPIError
events.py 27 SSE event dataclasses with type registry
server.py AsyncTurnstoneServer + TurnstoneServer
console.py AsyncTurnstoneConsole + TurnstoneConsole
sdk/typescript/ TypeScript SDK (npm package)
src/base.ts fetch wrapper, auth, SSE streaming
src/server.ts TurnstoneServer class
src/console.ts TurnstoneConsole class
src/events.ts Discriminated union events + type guards
src/sse.ts ReadableStream SSE parser
src/types.ts Request/response interfaces
```
The Python SDK reuses Pydantic models from `turnstone/api/` directly — no schema duplication. The TypeScript SDK has hand-written interfaces matching those models.
Both SDKs follow the same design: typed methods for REST endpoints, async iterators for SSE streams, and a high-level `send_and_wait` method for simple request-response patterns.
+1
View File
@@ -66,6 +66,7 @@ include = [
"turnstone/console/static/*.js",
"turnstone/shared_static/*.css",
"turnstone/shared_static/*.js",
"turnstone/sdk/py.typed",
]
[tool.pytest.ini_options]
+3
View File
@@ -0,0 +1,3 @@
node_modules/
dist/
*.tsbuildinfo
+869
View File
@@ -0,0 +1,869 @@
{
"openapi": "3.1.0",
"info": {
"title": "turnstone Console API",
"version": "0.3.0",
"description": "Cluster-wide visibility and control across all turnstone nodes."
},
"paths": {
"/v1/api/cluster/overview": {
"get": {
"summary": "Cluster state summary",
"operationId": "v1_api_cluster_overview_get",
"tags": [
"Cluster"
],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ClusterOverviewResponse"
}
}
}
}
}
}
},
"/v1/api/cluster/nodes": {
"get": {
"summary": "Paginated node list",
"operationId": "v1_api_cluster_nodes_get",
"tags": [
"Cluster"
],
"parameters": [
{
"name": "sort",
"in": "query",
"required": false,
"schema": {
"type": "string",
"default": "activity",
"enum": [
"activity",
"tokens",
"name"
]
},
"description": "Sort field"
},
{
"name": "limit",
"in": "query",
"required": false,
"schema": {
"type": "integer",
"default": 100
},
"description": "Page size"
},
{
"name": "offset",
"in": "query",
"required": false,
"schema": {
"type": "integer",
"default": 0
},
"description": "Pagination offset"
}
],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ClusterNodesResponse"
}
}
}
}
}
}
},
"/v1/api/cluster/workstreams": {
"get": {
"summary": "Filtered workstream list",
"operationId": "v1_api_cluster_workstreams_get",
"tags": [
"Cluster"
],
"parameters": [
{
"name": "state",
"in": "query",
"required": false,
"schema": {
"type": "string",
"enum": [
"running",
"thinking",
"attention",
"idle",
"error"
]
},
"description": "Filter by state"
},
{
"name": "node",
"in": "query",
"required": false,
"schema": {
"type": "string"
},
"description": "Filter by node_id"
},
{
"name": "search",
"in": "query",
"required": false,
"schema": {
"type": "string"
},
"description": "Search in name/title/node"
},
{
"name": "sort",
"in": "query",
"required": false,
"schema": {
"type": "string",
"default": "state",
"enum": [
"state",
"tokens",
"name"
]
},
"description": "Sort field"
},
{
"name": "page",
"in": "query",
"required": false,
"schema": {
"type": "integer",
"default": 1
},
"description": "Page number"
},
{
"name": "per_page",
"in": "query",
"required": false,
"schema": {
"type": "integer",
"default": 50
},
"description": "Items per page (max 200)"
}
],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ClusterWorkstreamsResponse"
}
}
}
}
}
}
},
"/v1/api/cluster/node/{node_id}": {
"get": {
"summary": "Single node detail",
"operationId": "v1_api_cluster_node_{node_id}_get",
"tags": [
"Cluster"
],
"parameters": [
{
"name": "node_id",
"in": "path",
"required": true,
"schema": {
"type": "string"
}
}
],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/NodeDetailResponse"
}
}
}
},
"404": {
"description": "Error 404",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
}
}
}
},
"/v1/api/cluster/workstreams/new": {
"post": {
"summary": "Create workstream via MQ dispatch",
"operationId": "v1_api_cluster_workstreams_new_post",
"tags": [
"Cluster"
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ConsoleCreateWsRequest"
}
}
}
},
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ConsoleCreateWsResponse"
}
}
}
},
"400": {
"description": "Error 400",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"404": {
"description": "Error 404",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
},
"503": {
"description": "Error 503",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
}
}
}
},
"/v1/api/cluster/events": {
"get": {
"summary": "Cluster SSE event stream",
"operationId": "v1_api_cluster_events_get",
"tags": [
"Streaming"
],
"description": "Server-Sent Events stream for real-time cluster updates. Returns text/event-stream with node_joined, node_lost, cluster_state, ws_created, ws_closed, ws_rename events.",
"responses": {
"200": {
"description": "Success"
}
}
}
},
"/v1/api/auth/login": {
"post": {
"summary": "Authenticate with a token",
"operationId": "v1_api_auth_login_post",
"tags": [
"Auth"
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/AuthLoginRequest"
}
}
}
},
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/AuthLoginResponse"
}
}
}
},
"401": {
"description": "Error 401",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ErrorResponse"
}
}
}
}
}
}
},
"/v1/api/auth/logout": {
"post": {
"summary": "Clear auth cookie",
"operationId": "v1_api_auth_logout_post",
"tags": [
"Auth"
],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/StatusResponse"
}
}
}
}
}
}
},
"/health": {
"get": {
"summary": "Console health check",
"operationId": "health_get",
"tags": [
"Observability"
],
"responses": {
"200": {
"description": "Success",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ConsoleHealthResponse"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"ErrorResponse": {
"description": "Standard error response body.",
"properties": {
"error": {
"description": "Error message",
"title": "Error",
"type": "string"
}
},
"required": [
"error"
],
"title": "ErrorResponse",
"type": "object"
},
"StatusResponse": {
"description": "Generic success response.",
"properties": {
"status": {
"default": "ok",
"examples": [
"ok"
],
"title": "Status",
"type": "string"
}
},
"title": "StatusResponse",
"type": "object"
},
"AuthLoginRequest": {
"description": "POST /v1/api/auth/login request body.",
"properties": {
"token": {
"description": "Bearer token to authenticate",
"title": "Token",
"type": "string"
}
},
"required": [
"token"
],
"title": "AuthLoginRequest",
"type": "object"
},
"AuthLoginResponse": {
"description": "POST /v1/api/auth/login success response.",
"properties": {
"status": {
"default": "ok",
"title": "Status",
"type": "string"
},
"role": {
"description": "Assigned role",
"examples": [
"full",
"read"
],
"title": "Role",
"type": "string"
}
},
"required": [
"role"
],
"title": "AuthLoginResponse",
"type": "object"
},
"ClusterOverviewResponse": {
"properties": {
"nodes": {
"default": 0,
"title": "Nodes",
"type": "integer"
},
"workstreams": {
"default": 0,
"title": "Workstreams",
"type": "integer"
},
"states": {
"$ref": "#/components/schemas/StateCounts",
"default": {
"running": 0,
"thinking": 0,
"attention": 0,
"idle": 0,
"error": 0
}
},
"aggregate": {
"$ref": "#/components/schemas/ClusterAggregate",
"default": {
"total_tokens": 0,
"total_tool_calls": 0
}
},
"version_drift": {
"default": false,
"title": "Version Drift",
"type": "boolean"
},
"versions": {
"default": [],
"items": {
"type": "string"
},
"title": "Versions",
"type": "array"
}
},
"title": "ClusterOverviewResponse",
"type": "object"
},
"ClusterAggregate": {
"properties": {
"total_tokens": {
"default": 0,
"title": "Total Tokens",
"type": "integer"
},
"total_tool_calls": {
"default": 0,
"title": "Total Tool Calls",
"type": "integer"
}
},
"title": "ClusterAggregate",
"type": "object"
},
"StateCounts": {
"properties": {
"running": {
"default": 0,
"title": "Running",
"type": "integer"
},
"thinking": {
"default": 0,
"title": "Thinking",
"type": "integer"
},
"attention": {
"default": 0,
"title": "Attention",
"type": "integer"
},
"idle": {
"default": 0,
"title": "Idle",
"type": "integer"
},
"error": {
"default": 0,
"title": "Error",
"type": "integer"
}
},
"title": "StateCounts",
"type": "object"
},
"ClusterNodesResponse": {
"properties": {
"nodes": {
"items": {
"$ref": "#/components/schemas/ClusterNodeInfo"
},
"title": "Nodes",
"type": "array"
},
"total": {
"default": 0,
"title": "Total",
"type": "integer"
}
},
"required": [
"nodes"
],
"title": "ClusterNodesResponse",
"type": "object"
},
"ClusterNodeInfo": {
"properties": {
"node_id": {
"title": "Node Id",
"type": "string"
},
"server_url": {
"default": "",
"title": "Server Url",
"type": "string"
},
"ws_total": {
"default": 0,
"title": "Ws Total",
"type": "integer"
},
"ws_running": {
"default": 0,
"title": "Ws Running",
"type": "integer"
},
"ws_thinking": {
"default": 0,
"title": "Ws Thinking",
"type": "integer"
},
"ws_attention": {
"default": 0,
"title": "Ws Attention",
"type": "integer"
},
"ws_idle": {
"default": 0,
"title": "Ws Idle",
"type": "integer"
},
"ws_error": {
"default": 0,
"title": "Ws Error",
"type": "integer"
},
"total_tokens": {
"default": 0,
"title": "Total Tokens",
"type": "integer"
},
"started": {
"default": 0.0,
"title": "Started",
"type": "number"
},
"reachable": {
"default": true,
"title": "Reachable",
"type": "boolean"
},
"health": {
"additionalProperties": {
"type": "string"
},
"title": "Health",
"type": "object"
},
"version": {
"default": "",
"title": "Version",
"type": "string"
}
},
"required": [
"node_id"
],
"title": "ClusterNodeInfo",
"type": "object"
},
"ClusterWorkstreamsResponse": {
"properties": {
"workstreams": {
"items": {
"$ref": "#/components/schemas/ClusterWorkstreamInfo"
},
"title": "Workstreams",
"type": "array"
},
"total": {
"default": 0,
"title": "Total",
"type": "integer"
},
"page": {
"default": 1,
"title": "Page",
"type": "integer"
},
"per_page": {
"default": 50,
"title": "Per Page",
"type": "integer"
},
"pages": {
"default": 1,
"title": "Pages",
"type": "integer"
}
},
"required": [
"workstreams"
],
"title": "ClusterWorkstreamsResponse",
"type": "object"
},
"ClusterWorkstreamInfo": {
"properties": {
"id": {
"title": "Id",
"type": "string"
},
"name": {
"default": "",
"title": "Name",
"type": "string"
},
"state": {
"default": "",
"title": "State",
"type": "string"
},
"node": {
"default": "",
"title": "Node",
"type": "string"
},
"title": {
"default": "",
"title": "Title",
"type": "string"
},
"tokens": {
"default": 0,
"title": "Tokens",
"type": "integer"
},
"context_ratio": {
"default": 0.0,
"title": "Context Ratio",
"type": "number"
},
"activity": {
"default": "",
"title": "Activity",
"type": "string"
},
"activity_state": {
"default": "",
"title": "Activity State",
"type": "string"
},
"tool_calls": {
"default": 0,
"title": "Tool Calls",
"type": "integer"
}
},
"required": [
"id"
],
"title": "ClusterWorkstreamInfo",
"type": "object"
},
"NodeDetailResponse": {
"properties": {
"node_id": {
"title": "Node Id",
"type": "string"
},
"server_url": {
"default": "",
"title": "Server Url",
"type": "string"
},
"health": {
"additionalProperties": {
"type": "string"
},
"title": "Health",
"type": "object"
},
"workstreams": {
"default": [],
"items": {
"$ref": "#/components/schemas/ClusterWorkstreamInfo"
},
"title": "Workstreams",
"type": "array"
},
"aggregate": {
"additionalProperties": {
"type": "integer"
},
"title": "Aggregate",
"type": "object"
},
"reachable": {
"default": true,
"title": "Reachable",
"type": "boolean"
}
},
"required": [
"node_id"
],
"title": "NodeDetailResponse",
"type": "object"
},
"ConsoleCreateWsRequest": {
"properties": {
"node_id": {
"default": "",
"description": "Target node: specific ID, 'auto', 'pool', or empty for auto",
"title": "Node Id",
"type": "string"
},
"name": {
"default": "",
"description": "Workstream name (auto-generated if empty)",
"title": "Name",
"type": "string"
},
"model": {
"default": "",
"description": "Model alias from node registry",
"title": "Model",
"type": "string"
}
},
"title": "ConsoleCreateWsRequest",
"type": "object"
},
"ConsoleCreateWsResponse": {
"properties": {
"status": {
"default": "ok",
"title": "Status",
"type": "string"
},
"correlation_id": {
"default": "",
"title": "Correlation Id",
"type": "string"
},
"target_node": {
"default": "",
"title": "Target Node",
"type": "string"
}
},
"title": "ConsoleCreateWsResponse",
"type": "object"
},
"ConsoleHealthResponse": {
"properties": {
"status": {
"default": "ok",
"examples": [
"ok"
],
"title": "Status",
"type": "string"
},
"service": {
"default": "turnstone-console",
"title": "Service",
"type": "string"
},
"nodes": {
"default": 0,
"title": "Nodes",
"type": "integer"
},
"workstreams": {
"default": 0,
"title": "Workstreams",
"type": "integer"
},
"version_drift": {
"default": false,
"title": "Version Drift",
"type": "boolean"
},
"versions": {
"default": [],
"items": {
"type": "string"
},
"title": "Versions",
"type": "array"
}
},
"title": "ConsoleHealthResponse",
"type": "object"
}
}
}
}
File diff suppressed because it is too large Load Diff
+1346
View File
File diff suppressed because it is too large Load Diff
+38
View File
@@ -0,0 +1,38 @@
{
"name": "@turnstone/sdk",
"version": "0.3.0",
"description": "TypeScript client SDK for the turnstone AI orchestration platform",
"type": "module",
"main": "./dist/index.js",
"types": "./dist/index.d.ts",
"exports": {
".": {
"import": "./dist/index.js",
"types": "./dist/index.d.ts"
}
},
"scripts": {
"build": "tsc",
"test": "vitest run",
"test:watch": "vitest",
"typecheck": "tsc --noEmit",
"generate-types": "python scripts/generate-types.py"
},
"files": [
"dist",
"src"
],
"keywords": [
"turnstone",
"ai",
"llm",
"agent",
"sdk",
"client"
],
"license": "BUSL-1.1",
"devDependencies": {
"typescript": "^5.4",
"vitest": "^2.0"
}
}
+40
View File
@@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""Export OpenAPI specs to JSON files for TypeScript type reference.
Usage:
python scripts/generate-types.py
Writes:
openapi-server.json — Server API OpenAPI 3.1 spec
openapi-console.json — Console API OpenAPI 3.1 spec
"""
import json
import sys
from pathlib import Path
# Ensure the turnstone package is importable (repo root is 3 levels up)
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
from turnstone.api.console_spec import build_console_spec
from turnstone.api.server_spec import build_server_spec
output_dir = Path(__file__).resolve().parent.parent
def main() -> None:
server_spec = build_server_spec()
console_spec = build_console_spec()
server_path = output_dir / "openapi-server.json"
console_path = output_dir / "openapi-console.json"
server_path.write_text(json.dumps(server_spec, indent=2) + "\n")
console_path.write_text(json.dumps(console_spec, indent=2) + "\n")
print(f"Wrote {server_path} ({len(server_spec['paths'])} paths)")
print(f"Wrote {console_path} ({len(console_spec['paths'])} paths)")
if __name__ == "__main__":
main()
+107
View File
@@ -0,0 +1,107 @@
import { TurnstoneAPIError } from "./errors.js";
import { parseSSEStream } from "./sse.js";
export interface ClientOptions {
/** Server base URL (e.g. "http://localhost:8080"). */
baseUrl: string;
/** Bearer token for authentication. */
token?: string;
/** Custom fetch implementation (defaults to globalThis.fetch). */
fetch?: typeof globalThis.fetch;
}
export interface RequestOptions {
json?: object;
params?: Record<string, string | number>;
}
export class BaseClient {
protected readonly baseUrl: string;
protected readonly token: string;
protected readonly fetchFn: typeof globalThis.fetch;
constructor(options: ClientOptions) {
this.baseUrl = options.baseUrl.replace(/\/$/, "");
this.token = options.token ?? "";
this.fetchFn = options.fetch ?? globalThis.fetch.bind(globalThis);
}
protected async request<T>(
method: string,
path: string,
options?: RequestOptions,
): Promise<T> {
const headers: Record<string, string> = {
"Content-Type": "application/json",
};
if (this.token) {
headers["Authorization"] = `Bearer ${this.token}`;
}
let url = `${this.baseUrl}${path}`;
if (options?.params) {
const searchParams = new URLSearchParams();
for (const [key, value] of Object.entries(options.params)) {
if (value !== undefined && value !== "") {
searchParams.set(key, String(value));
}
}
const qs = searchParams.toString();
if (qs) url += `?${qs}`;
}
const resp = await this.fetchFn(url, {
method,
headers,
body: options?.json ? JSON.stringify(options.json) : undefined,
});
if (!resp.ok) {
let msg = "";
try {
const body = (await resp.json()) as Record<string, unknown>;
msg = (body.error as string) ?? (body.detail as string) ?? "";
} catch {
msg = await resp.text().catch(() => "");
}
throw new TurnstoneAPIError(resp.status, msg || `HTTP ${resp.status}`);
}
return (await resp.json()) as T;
}
protected async *streamSSE<T = Record<string, unknown>>(
path: string,
params?: Record<string, string | number>,
signal?: AbortSignal,
): AsyncIterableIterator<T> {
const headers: Record<string, string> = {
Accept: "text/event-stream",
};
if (this.token) {
headers["Authorization"] = `Bearer ${this.token}`;
}
let url = `${this.baseUrl}${path}`;
if (params) {
const searchParams = new URLSearchParams();
for (const [key, value] of Object.entries(params)) {
if (value !== undefined && value !== "") {
searchParams.set(key, String(value));
}
}
const qs = searchParams.toString();
if (qs) url += `?${qs}`;
}
const resp = await this.fetchFn(url, { method: "GET", headers, signal });
if (!resp.ok) {
throw new TurnstoneAPIError(
resp.status,
`SSE connection failed: HTTP ${resp.status}`,
);
}
yield* parseSSEStream<T>(resp);
}
}
+88
View File
@@ -0,0 +1,88 @@
import { BaseClient, type ClientOptions } from "./base.js";
import type { ClusterEvent } from "./events.js";
import type {
AuthLoginResponse,
ClusterNodesResponse,
ClusterOverviewResponse,
ClusterWorkstreamsResponse,
ConsoleCreateWsRequest,
ConsoleCreateWsResponse,
ConsoleHealthResponse,
NodeDetailResponse,
NodesOptions,
StatusResponse,
WorkstreamsOptions,
} from "./types.js";
/** Async client for the turnstone console API. */
export class TurnstoneConsole extends BaseClient {
constructor(options: ClientOptions) {
super(options);
}
// -- Cluster overview -----------------------------------------------------
async overview(): Promise<ClusterOverviewResponse> {
return this.request("GET", "/v1/api/cluster/overview");
}
async nodes(opts?: NodesOptions): Promise<ClusterNodesResponse> {
return this.request("GET", "/v1/api/cluster/nodes", {
params: {
sort: opts?.sort ?? "activity",
limit: opts?.limit ?? 100,
offset: opts?.offset ?? 0,
},
});
}
async workstreams(
opts?: WorkstreamsOptions,
): Promise<ClusterWorkstreamsResponse> {
const params: Record<string, string | number> = {
sort: opts?.sort ?? "state",
page: opts?.page ?? 1,
per_page: opts?.per_page ?? 50,
};
if (opts?.state) params.state = opts.state;
if (opts?.node) params.node = opts.node;
if (opts?.search) params.search = opts.search;
return this.request("GET", "/v1/api/cluster/workstreams", { params });
}
async nodeDetail(nodeId: string): Promise<NodeDetailResponse> {
return this.request("GET", `/v1/api/cluster/node/${nodeId}`);
}
async createWorkstream(
opts?: ConsoleCreateWsRequest,
): Promise<ConsoleCreateWsResponse> {
return this.request("POST", "/v1/api/cluster/workstreams/new", {
json: opts,
});
}
// -- Streaming ------------------------------------------------------------
async *clusterEvents(): AsyncIterableIterator<ClusterEvent> {
yield* this.streamSSE<ClusterEvent>("/v1/api/cluster/events");
}
// -- Auth -----------------------------------------------------------------
async login(token: string): Promise<AuthLoginResponse> {
return this.request("POST", "/v1/api/auth/login", {
json: { token },
});
}
async logout(): Promise<StatusResponse> {
return this.request("POST", "/v1/api/auth/logout");
}
// -- Health ---------------------------------------------------------------
async health(): Promise<ConsoleHealthResponse> {
return this.request("GET", "/health");
}
}
+10
View File
@@ -0,0 +1,10 @@
/** Raised when a turnstone server returns a non-2xx response. */
export class TurnstoneAPIError extends Error {
constructor(
public readonly statusCode: number,
public readonly errorMessage: string,
) {
super(`HTTP ${statusCode}: ${errorMessage}`);
this.name = "TurnstoneAPIError";
}
}
+239
View File
@@ -0,0 +1,239 @@
// ---------------------------------------------------------------------------
// Server SSE events
// ---------------------------------------------------------------------------
export interface ConnectedEvent {
type: "connected";
model: string;
model_alias: string;
skip_permissions: boolean;
}
export interface HistoryEvent {
type: "history";
messages: Array<Record<string, unknown>>;
}
export interface ThinkingStartEvent {
type: "thinking_start";
}
export interface ThinkingStopEvent {
type: "thinking_stop";
}
export interface ContentEvent {
type: "content";
text: string;
}
export interface ReasoningEvent {
type: "reasoning";
text: string;
}
export interface StreamEndEvent {
type: "stream_end";
}
export interface ToolInfoEvent {
type: "tool_info";
items: Array<Record<string, unknown>>;
}
export interface ApproveRequestEvent {
type: "approve_request";
items: Array<Record<string, unknown>>;
}
export interface ToolResultEvent {
type: "tool_result";
call_id: string;
name: string;
output: string;
}
export interface ToolOutputChunkEvent {
type: "tool_output_chunk";
call_id: string;
chunk: string;
}
export interface StatusEvent {
type: "status";
prompt_tokens: number;
completion_tokens: number;
total_tokens: number;
context_window: number;
pct: number;
effort: string;
}
export interface PlanReviewEvent {
type: "plan_review";
content: string;
}
export interface InfoEvent {
type: "info";
message: string;
}
export interface ErrorEvent {
type: "error";
message: string;
}
export interface BusyErrorEvent {
type: "busy_error";
message: string;
}
export interface ClearUiEvent {
type: "clear_ui";
}
// Global events
export interface WsStateEvent {
type: "ws_state";
ws_id: string;
state: string;
tokens: number;
context_ratio: number;
activity: string;
activity_state: string;
}
export interface WsActivityEvent {
type: "ws_activity";
ws_id: string;
activity: string;
activity_state: string;
}
export interface WsRenameEvent {
type: "ws_rename";
ws_id: string;
name: string;
}
export interface WsClosedEvent {
type: "ws_closed";
ws_id: string;
name?: string;
}
/** Discriminated union of all server SSE event types. */
export type ServerEvent =
| ConnectedEvent
| HistoryEvent
| ThinkingStartEvent
| ThinkingStopEvent
| ContentEvent
| ReasoningEvent
| StreamEndEvent
| ToolInfoEvent
| ApproveRequestEvent
| ToolResultEvent
| ToolOutputChunkEvent
| StatusEvent
| PlanReviewEvent
| InfoEvent
| ErrorEvent
| BusyErrorEvent
| ClearUiEvent
| WsStateEvent
| WsActivityEvent
| WsRenameEvent
| WsClosedEvent;
// ---------------------------------------------------------------------------
// Console cluster SSE events
// ---------------------------------------------------------------------------
export interface NodeJoinedEvent {
type: "node_joined";
node_id: string;
}
export interface NodeLostEvent {
type: "node_lost";
node_id: string;
}
export interface ClusterStateEvent {
type: "cluster_state";
ws_id: string;
node_id: string;
state: string;
tokens: number;
context_ratio: number;
activity: string;
activity_state: string;
}
export interface ClusterWsCreatedEvent {
type: "ws_created";
ws_id: string;
node_id: string;
name: string;
}
export interface ClusterWsClosedEvent {
type: "ws_closed";
ws_id: string;
}
export interface ClusterWsRenameEvent {
type: "ws_rename";
ws_id: string;
name: string;
}
/** Discriminated union of all console cluster SSE event types. */
export type ClusterEvent =
| NodeJoinedEvent
| NodeLostEvent
| ClusterStateEvent
| ClusterWsCreatedEvent
| ClusterWsClosedEvent
| ClusterWsRenameEvent;
// ---------------------------------------------------------------------------
// Type guards
// ---------------------------------------------------------------------------
export function isContentEvent(e: ServerEvent): e is ContentEvent {
return e.type === "content";
}
export function isReasoningEvent(e: ServerEvent): e is ReasoningEvent {
return e.type === "reasoning";
}
export function isErrorEvent(e: ServerEvent): e is ErrorEvent {
return e.type === "error";
}
export function isStreamEndEvent(e: ServerEvent): e is StreamEndEvent {
return e.type === "stream_end";
}
export function isToolResultEvent(e: ServerEvent): e is ToolResultEvent {
return e.type === "tool_result";
}
export function isWsStateEvent(e: ServerEvent): e is WsStateEvent {
return e.type === "ws_state";
}
export function isApproveRequestEvent(
e: ServerEvent,
): e is ApproveRequestEvent {
return e.type === "approve_request";
}
export function isPlanReviewEvent(e: ServerEvent): e is PlanReviewEvent {
return e.type === "plan_review";
}
+111
View File
@@ -0,0 +1,111 @@
/**
* @turnstone/sdk — TypeScript client SDK for the turnstone AI orchestration platform.
*
* @example
* ```ts
* import { TurnstoneServer } from "@turnstone/sdk";
*
* const client = new TurnstoneServer({
* baseUrl: "http://localhost:8080",
* token: "tok_xxx",
* });
*
* const ws = await client.createWorkstream({ name: "demo" });
* const result = await client.sendAndWait("Hello!", ws.ws_id);
* console.log(result.content);
* ```
*/
// Clients
export { TurnstoneServer } from "./server.js";
export { TurnstoneConsole } from "./console.js";
export type { ClientOptions } from "./base.js";
// Errors
export { TurnstoneAPIError } from "./errors.js";
// Event types and guards
export type {
ServerEvent,
ClusterEvent,
ConnectedEvent,
HistoryEvent,
ThinkingStartEvent,
ThinkingStopEvent,
ContentEvent,
ReasoningEvent,
StreamEndEvent,
ToolInfoEvent,
ApproveRequestEvent,
ToolResultEvent,
ToolOutputChunkEvent,
StatusEvent,
PlanReviewEvent,
InfoEvent,
ErrorEvent,
BusyErrorEvent,
ClearUiEvent,
WsStateEvent,
WsActivityEvent,
WsRenameEvent,
WsClosedEvent,
NodeJoinedEvent,
NodeLostEvent,
ClusterStateEvent,
ClusterWsCreatedEvent,
ClusterWsClosedEvent,
ClusterWsRenameEvent,
} from "./events.js";
export {
isContentEvent,
isReasoningEvent,
isErrorEvent,
isStreamEndEvent,
isToolResultEvent,
isWsStateEvent,
isApproveRequestEvent,
isPlanReviewEvent,
} from "./events.js";
// Request/response types
export type {
SendRequest,
SendResponse,
ApproveRequest,
PlanFeedbackRequest,
CommandRequest,
CreateWorkstreamRequest,
CreateWorkstreamResponse,
CloseWorkstreamRequest,
WorkstreamInfo,
ListWorkstreamsResponse,
DashboardWorkstream,
DashboardAggregate,
DashboardResponse,
SessionInfo,
ListSessionsResponse,
BackendStatus,
WorkstreamCounts,
HealthResponse,
AuthLoginRequest,
AuthLoginResponse,
StatusResponse,
ErrorResponse,
ClusterOverviewResponse,
ClusterNodeInfo,
ClusterNodesResponse,
ClusterWorkstreamInfo,
ClusterWorkstreamsResponse,
NodeDetailResponse,
ConsoleCreateWsRequest,
ConsoleCreateWsResponse,
ConsoleHealthResponse,
TurnResult,
SendAndWaitOptions,
NodesOptions,
WorkstreamsOptions,
} from "./types.js";
// SSE parser (for advanced usage)
export { parseSSEStream } from "./sse.js";
+202
View File
@@ -0,0 +1,202 @@
import { BaseClient, type ClientOptions } from "./base.js";
import type { ServerEvent } from "./events.js";
import type {
AuthLoginResponse,
CreateWorkstreamRequest,
CreateWorkstreamResponse,
DashboardResponse,
HealthResponse,
ListSessionsResponse,
ListWorkstreamsResponse,
SendAndWaitOptions,
SendResponse,
StatusResponse,
TurnResult,
} from "./types.js";
/** Async client for the turnstone server API. */
export class TurnstoneServer extends BaseClient {
constructor(options: ClientOptions) {
super(options);
}
// -- Workstream management ------------------------------------------------
async listWorkstreams(): Promise<ListWorkstreamsResponse> {
return this.request("GET", "/v1/api/workstreams");
}
async dashboard(): Promise<DashboardResponse> {
return this.request("GET", "/v1/api/dashboard");
}
async createWorkstream(
opts?: CreateWorkstreamRequest,
): Promise<CreateWorkstreamResponse> {
return this.request("POST", "/v1/api/workstreams/new", { json: opts });
}
async closeWorkstream(wsId: string): Promise<StatusResponse> {
return this.request("POST", "/v1/api/workstreams/close", {
json: { ws_id: wsId },
});
}
// -- Chat interaction -----------------------------------------------------
async send(message: string, wsId: string): Promise<SendResponse> {
return this.request("POST", "/v1/api/send", {
json: { message, ws_id: wsId },
});
}
async approve(opts: {
wsId: string;
approved?: boolean;
feedback?: string | null;
always?: boolean;
}): Promise<StatusResponse> {
return this.request("POST", "/v1/api/approve", {
json: {
ws_id: opts.wsId,
approved: opts.approved ?? true,
feedback: opts.feedback,
always: opts.always,
},
});
}
async planFeedback(opts: {
wsId: string;
feedback?: string;
}): Promise<StatusResponse> {
return this.request("POST", "/v1/api/plan", {
json: { ws_id: opts.wsId, feedback: opts.feedback ?? "" },
});
}
async command(opts: {
wsId: string;
command: string;
}): Promise<StatusResponse> {
return this.request("POST", "/v1/api/command", {
json: { ws_id: opts.wsId, command: opts.command },
});
}
// -- Streaming ------------------------------------------------------------
async *streamEvents(wsId: string): AsyncIterableIterator<ServerEvent> {
yield* this.streamSSE<ServerEvent>("/v1/api/events", { ws_id: wsId });
}
async *streamGlobalEvents(): AsyncIterableIterator<ServerEvent> {
yield* this.streamSSE<ServerEvent>("/v1/api/events/global");
}
// -- High-level convenience -----------------------------------------------
async sendAndWait(
message: string,
wsId: string,
opts?: SendAndWaitOptions,
): Promise<TurnResult> {
const result: TurnResult = {
wsId,
contentParts: [],
reasoningParts: [],
toolResults: [],
errors: [],
timedOut: false,
get content() {
return this.contentParts.join("");
},
get reasoning() {
return this.reasoningParts.join("");
},
get ok() {
return !this.timedOut && this.errors.length === 0;
},
};
// Open SSE stream BEFORE sending to avoid missing early events
const timeoutMs = opts?.timeout ?? 600_000;
const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), timeoutMs);
try {
// Start consuming the per-workstream SSE stream first
const events = this.streamSSE<ServerEvent>(
"/v1/api/events",
{ ws_id: wsId },
controller.signal,
);
const sendResp = await this.send(message, wsId);
if (sendResp.status === "busy") {
result.errors.push("Workstream is busy");
return result;
}
for await (const event of events) {
opts?.onEvent?.(event);
switch (event.type) {
case "content":
result.contentParts.push(event.text);
break;
case "reasoning":
result.reasoningParts.push(event.text);
break;
case "tool_result":
result.toolResults.push({
name: event.name,
output: event.output,
});
break;
case "error":
result.errors.push(event.message);
break;
case "ws_state":
if (event.state === "idle") return result;
break;
}
}
} catch (err) {
if (err instanceof DOMException && err.name === "AbortError") {
result.timedOut = true;
} else {
throw err;
}
} finally {
clearTimeout(timer);
controller.abort();
}
return result;
}
// -- Sessions -------------------------------------------------------------
async listSessions(): Promise<ListSessionsResponse> {
return this.request("GET", "/v1/api/sessions");
}
// -- Auth -----------------------------------------------------------------
async login(token: string): Promise<AuthLoginResponse> {
return this.request("POST", "/v1/api/auth/login", {
json: { token },
});
}
async logout(): Promise<StatusResponse> {
return this.request("POST", "/v1/api/auth/logout");
}
// -- Health ---------------------------------------------------------------
async health(): Promise<HealthResponse> {
return this.request("GET", "/health");
}
}
+66
View File
@@ -0,0 +1,66 @@
/**
* SSE stream parser for fetch ReadableStream.
*
* Parses standard Server-Sent Events from a `Response.body` stream.
* Works in browsers and Node.js 18+ natively (no dependencies).
*/
/**
* Parse an SSE stream and yield JSON-parsed data payloads.
*
* Handles the standard SSE format including multi-line `data:` fields
* (joined with `\n` per the SSE spec) and CRLF line endings.
*/
export async function* parseSSEStream<T = Record<string, unknown>>(
response: Response,
): AsyncIterableIterator<T> {
const body = response.body;
if (!body) return;
const reader = body.getReader();
const decoder = new TextDecoder();
let buffer = "";
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
buffer += decoder.decode(value, { stream: true });
// Normalize CRLF to LF
buffer = buffer.replace(/\r\n/g, "\n");
// Process complete SSE frames (separated by double newlines)
const frames = buffer.split("\n\n");
// Keep the last (possibly incomplete) frame in the buffer
buffer = frames.pop() ?? "";
for (const frame of frames) {
if (!frame.trim()) continue;
// Extract data lines from the frame, joining with \n per SSE spec
const dataLines: string[] = [];
for (const line of frame.split("\n")) {
if (line.startsWith("data: ")) {
dataLines.push(line.slice(6));
} else if (line.startsWith("data:")) {
dataLines.push(line.slice(5));
}
}
if (dataLines.length === 0) continue;
const data = dataLines.join("\n");
if (!data.trim()) continue;
try {
yield JSON.parse(data) as T;
} catch {
// Skip malformed JSON
}
}
}
} finally {
reader.releaseLock();
}
}
+287
View File
@@ -0,0 +1,287 @@
// ---------------------------------------------------------------------------
// Shared types
// ---------------------------------------------------------------------------
export interface ErrorResponse {
error: string;
}
export interface StatusResponse {
status: string;
}
export interface AuthLoginRequest {
token: string;
}
export interface AuthLoginResponse {
status: string;
role: string;
}
// ---------------------------------------------------------------------------
// Server API — Workstream management
// ---------------------------------------------------------------------------
export interface SendRequest {
message: string;
ws_id: string;
}
export interface SendResponse {
status: string;
}
export interface ApproveRequest {
approved: boolean;
feedback?: string | null;
always?: boolean;
ws_id: string;
}
export interface PlanFeedbackRequest {
feedback: string;
ws_id: string;
}
export interface CommandRequest {
command: string;
ws_id: string;
}
export interface CreateWorkstreamRequest {
name?: string;
model?: string;
auto_approve?: boolean;
}
export interface CreateWorkstreamResponse {
ws_id: string;
name: string;
}
export interface CloseWorkstreamRequest {
ws_id: string;
}
export interface WorkstreamInfo {
id: string;
name: string;
state: string;
session_id?: string | null;
}
export interface ListWorkstreamsResponse {
workstreams: WorkstreamInfo[];
}
export interface DashboardWorkstream {
id: string;
name: string;
state: string;
session_id?: string | null;
title?: string;
tokens?: number;
context_ratio?: number;
activity?: string;
activity_state?: string;
tool_calls?: number;
node?: string;
model?: string;
model_alias?: string;
}
export interface DashboardAggregate {
total_tokens: number;
total_tool_calls: number;
active_count: number;
total_count: number;
uptime_seconds?: number;
node?: string;
}
export interface DashboardResponse {
workstreams: DashboardWorkstream[];
aggregate: DashboardAggregate;
}
// ---------------------------------------------------------------------------
// Server API — Sessions
// ---------------------------------------------------------------------------
export interface SessionInfo {
session_id: string;
alias?: string | null;
title?: string | null;
created: string;
updated: string;
message_count: number;
}
export interface ListSessionsResponse {
sessions: SessionInfo[];
}
// ---------------------------------------------------------------------------
// Server API — Health
// ---------------------------------------------------------------------------
export interface BackendStatus {
status: string;
circuit_state: string;
}
export interface WorkstreamCounts {
total: number;
idle?: number;
thinking?: number;
running?: number;
attention?: number;
error?: number;
}
export interface HealthResponse {
status: string;
version?: string;
uptime_seconds?: number;
model?: string;
workstreams?: WorkstreamCounts;
backend?: BackendStatus | null;
}
// ---------------------------------------------------------------------------
// Console API
// ---------------------------------------------------------------------------
export interface StateCounts {
running?: number;
thinking?: number;
attention?: number;
idle?: number;
error?: number;
}
export interface ClusterAggregate {
total_tokens: number;
total_tool_calls: number;
}
export interface ClusterOverviewResponse {
nodes: number;
workstreams: number;
states: StateCounts;
aggregate: ClusterAggregate;
version_drift: boolean;
versions: string[];
}
export interface ClusterNodeInfo {
node_id: string;
server_url: string;
ws_total: number;
ws_running: number;
ws_thinking: number;
ws_attention: number;
ws_idle: number;
ws_error: number;
total_tokens: number;
started: number;
reachable: boolean;
health: Record<string, string>;
version: string;
}
export interface ClusterNodesResponse {
nodes: ClusterNodeInfo[];
total: number;
}
export interface ClusterWorkstreamInfo {
id: string;
name: string;
state: string;
node: string;
title?: string;
tokens?: number;
context_ratio?: number;
activity?: string;
activity_state?: string;
tool_calls?: number;
}
export interface ClusterWorkstreamsResponse {
workstreams: ClusterWorkstreamInfo[];
total: number;
page: number;
per_page: number;
pages: number;
}
export interface NodeDetailResponse {
node_id: string;
server_url: string;
health: Record<string, string>;
workstreams: ClusterWorkstreamInfo[];
aggregate: ClusterAggregate;
}
export interface ConsoleCreateWsRequest {
node_id?: string;
name?: string;
model?: string;
}
export interface ConsoleCreateWsResponse {
status: string;
correlation_id: string;
target_node: string;
}
export interface ConsoleHealthResponse {
status: string;
service: string;
nodes: number;
workstreams: number;
version_drift: boolean;
versions: string[];
}
// ---------------------------------------------------------------------------
// SDK-specific types
// ---------------------------------------------------------------------------
export interface TurnResult {
wsId: string;
contentParts: string[];
reasoningParts: string[];
toolResults: Array<{ name: string; output: string }>;
errors: string[];
timedOut: boolean;
content: string;
reasoning: string;
ok: boolean;
}
export interface SendAndWaitOptions {
/** Timeout in milliseconds (default: 600000 = 10 minutes). */
timeout?: number;
onEvent?: (event: import("./events.js").ServerEvent) => void;
}
export interface NodesOptions {
sort?: string;
limit?: number;
offset?: number;
}
export interface WorkstreamsOptions {
state?: string;
node?: string;
search?: string;
sort?: string;
page?: number;
per_page?: number;
}
// Re-export event types for convenience
export type { ServerEvent, ClusterEvent } from "./events.js";
+82
View File
@@ -0,0 +1,82 @@
import { describe, expect, it, vi } from "vitest";
import { TurnstoneConsole } from "../src/console.js";
function mockFetch(response: object): typeof globalThis.fetch {
return vi.fn().mockResolvedValue(
new Response(JSON.stringify(response), {
status: 200,
headers: { "content-type": "application/json" },
}),
);
}
describe("TurnstoneConsole", () => {
it("overview returns parsed response", async () => {
const fetchFn = mockFetch({
nodes: 2,
workstreams: 5,
states: { idle: 5 },
aggregate: { total_tokens: 1000, total_tool_calls: 0 },
version_drift: false,
versions: ["0.3.0"],
});
const client = new TurnstoneConsole({
baseUrl: "http://test",
fetch: fetchFn,
});
const resp = await client.overview();
expect(resp.nodes).toBe(2);
expect(resp.workstreams).toBe(5);
});
it("nodes passes query parameters", async () => {
const fetchFn = mockFetch({ nodes: [], total: 0 });
const client = new TurnstoneConsole({
baseUrl: "http://test",
fetch: fetchFn,
});
await client.nodes({ sort: "tokens", limit: 50, offset: 10 });
const [url] = (fetchFn as ReturnType<typeof vi.fn>).mock.calls[0];
expect(url).toContain("sort=tokens");
expect(url).toContain("limit=50");
expect(url).toContain("offset=10");
});
it("workstreams passes filter parameters", async () => {
const fetchFn = mockFetch({
workstreams: [],
total: 0,
page: 1,
per_page: 50,
pages: 0,
});
const client = new TurnstoneConsole({
baseUrl: "http://test",
fetch: fetchFn,
});
await client.workstreams({ state: "running", page: 2 });
const [url] = (fetchFn as ReturnType<typeof vi.fn>).mock.calls[0];
expect(url).toContain("state=running");
expect(url).toContain("page=2");
});
it("health returns parsed response", async () => {
const fetchFn = mockFetch({
status: "ok",
service: "turnstone-console",
nodes: 2,
workstreams: 5,
version_drift: false,
versions: ["0.3.0"],
});
const client = new TurnstoneConsole({
baseUrl: "http://test",
fetch: fetchFn,
});
const resp = await client.health();
expect(resp.status).toBe("ok");
expect(resp.nodes).toBe(2);
});
});
+69
View File
@@ -0,0 +1,69 @@
import { describe, expect, it } from "vitest";
import {
isContentEvent,
isErrorEvent,
isStreamEndEvent,
isToolResultEvent,
isWsStateEvent,
isApproveRequestEvent,
isPlanReviewEvent,
isReasoningEvent,
} from "../src/events.js";
import type { ServerEvent } from "../src/events.js";
describe("event type guards", () => {
it("isContentEvent", () => {
const e: ServerEvent = { type: "content", text: "hello" };
expect(isContentEvent(e)).toBe(true);
expect(isErrorEvent(e)).toBe(false);
});
it("isReasoningEvent", () => {
const e: ServerEvent = { type: "reasoning", text: "step 1" };
expect(isReasoningEvent(e)).toBe(true);
expect(isContentEvent(e)).toBe(false);
});
it("isErrorEvent", () => {
const e: ServerEvent = { type: "error", message: "bad" };
expect(isErrorEvent(e)).toBe(true);
});
it("isStreamEndEvent", () => {
const e: ServerEvent = { type: "stream_end" };
expect(isStreamEndEvent(e)).toBe(true);
});
it("isToolResultEvent", () => {
const e: ServerEvent = {
type: "tool_result",
call_id: "c1",
name: "search",
output: "found",
};
expect(isToolResultEvent(e)).toBe(true);
});
it("isWsStateEvent", () => {
const e: ServerEvent = {
type: "ws_state",
ws_id: "ws1",
state: "idle",
tokens: 0,
context_ratio: 0,
activity: "",
activity_state: "",
};
expect(isWsStateEvent(e)).toBe(true);
});
it("isApproveRequestEvent", () => {
const e: ServerEvent = { type: "approve_request", items: [] };
expect(isApproveRequestEvent(e)).toBe(true);
});
it("isPlanReviewEvent", () => {
const e: ServerEvent = { type: "plan_review", content: "## Plan" };
expect(isPlanReviewEvent(e)).toBe(true);
});
});
+114
View File
@@ -0,0 +1,114 @@
import { describe, expect, it, vi } from "vitest";
import { TurnstoneServer } from "../src/server.js";
import { TurnstoneAPIError } from "../src/errors.js";
function mockFetch(response: object, status = 200): typeof globalThis.fetch {
return vi.fn().mockResolvedValue(
new Response(JSON.stringify(response), {
status,
headers: { "content-type": "application/json" },
}),
);
}
function mockFetchError(
error: object,
status: number,
): typeof globalThis.fetch {
return vi.fn().mockResolvedValue(
new Response(JSON.stringify(error), {
status,
headers: { "content-type": "application/json" },
}),
);
}
describe("TurnstoneServer", () => {
it("listWorkstreams returns parsed response", async () => {
const fetchFn = mockFetch({
workstreams: [{ id: "ws1", name: "test", state: "idle" }],
});
const client = new TurnstoneServer({
baseUrl: "http://test",
fetch: fetchFn,
});
const resp = await client.listWorkstreams();
expect(resp.workstreams).toHaveLength(1);
expect(resp.workstreams[0].id).toBe("ws1");
expect(fetchFn).toHaveBeenCalledWith(
"http://test/v1/api/workstreams",
expect.objectContaining({ method: "GET" }),
);
});
it("createWorkstream sends correct body", async () => {
const fetchFn = mockFetch({ ws_id: "ws_new", name: "Analysis" });
const client = new TurnstoneServer({
baseUrl: "http://test",
fetch: fetchFn,
});
const resp = await client.createWorkstream({ name: "Analysis" });
expect(resp.ws_id).toBe("ws_new");
const [, init] = (fetchFn as ReturnType<typeof vi.fn>).mock.calls[0];
expect(JSON.parse(init.body)).toEqual({ name: "Analysis" });
});
it("send posts correct payload", async () => {
const fetchFn = mockFetch({ status: "ok" });
const client = new TurnstoneServer({
baseUrl: "http://test",
fetch: fetchFn,
});
await client.send("Hello", "ws1");
const [url, init] = (fetchFn as ReturnType<typeof vi.fn>).mock.calls[0];
expect(url).toBe("http://test/v1/api/send");
expect(JSON.parse(init.body)).toEqual({ message: "Hello", ws_id: "ws1" });
});
it("injects auth header when token provided", async () => {
const fetchFn = mockFetch({ workstreams: [] });
const client = new TurnstoneServer({
baseUrl: "http://test",
token: "tok_abc",
fetch: fetchFn,
});
await client.listWorkstreams();
const [, init] = (fetchFn as ReturnType<typeof vi.fn>).mock.calls[0];
expect(init.headers.Authorization).toBe("Bearer tok_abc");
});
it("throws TurnstoneAPIError on 404", async () => {
const fetchFn = mockFetchError({ error: "Not found" }, 404);
const client = new TurnstoneServer({
baseUrl: "http://test",
fetch: fetchFn,
});
await expect(client.send("hi", "bad_ws")).rejects.toThrow(
TurnstoneAPIError,
);
try {
await client.send("hi", "bad_ws");
} catch (e) {
expect(e).toBeInstanceOf(TurnstoneAPIError);
expect((e as TurnstoneAPIError).statusCode).toBe(404);
}
});
it("health returns parsed response", async () => {
const fetchFn = mockFetch({
status: "ok",
version: "0.3.0",
uptime_seconds: 120,
});
const client = new TurnstoneServer({
baseUrl: "http://test",
fetch: fetchFn,
});
const resp = await client.health();
expect(resp.status).toBe("ok");
expect(resp.version).toBe("0.3.0");
});
});
+68
View File
@@ -0,0 +1,68 @@
import { describe, expect, it } from "vitest";
import { parseSSEStream } from "../src/sse.js";
function makeSSEResponse(...events: string[]): Response {
const body = events.map((e) => `data: ${e}\n\n`).join("");
const encoder = new TextEncoder();
const stream = new ReadableStream({
start(controller) {
controller.enqueue(encoder.encode(body));
controller.close();
},
});
return new Response(stream, {
headers: { "content-type": "text/event-stream" },
});
}
describe("parseSSEStream", () => {
it("yields parsed JSON from SSE data lines", async () => {
const resp = makeSSEResponse(
'{"type": "content", "text": "hello"}',
'{"type": "stream_end"}',
);
const events: unknown[] = [];
for await (const event of parseSSEStream(resp)) {
events.push(event);
}
expect(events).toHaveLength(2);
expect(events[0]).toEqual({ type: "content", text: "hello" });
expect(events[1]).toEqual({ type: "stream_end" });
});
it("skips malformed JSON", async () => {
const resp = makeSSEResponse(
"not-json",
'{"type": "info", "message": "ok"}',
);
const events: unknown[] = [];
for await (const event of parseSSEStream(resp)) {
events.push(event);
}
expect(events).toHaveLength(1);
expect(events[0]).toEqual({ type: "info", message: "ok" });
});
it("handles multiple events in sequence", async () => {
const resp = makeSSEResponse(
'{"type": "connected", "model": "gpt-5"}',
'{"type": "content", "text": "a"}',
'{"type": "content", "text": "b"}',
'{"type": "status", "total_tokens": 10}',
'{"type": "stream_end"}',
);
const events: unknown[] = [];
for await (const event of parseSSEStream(resp)) {
events.push(event);
}
expect(events).toHaveLength(5);
const types = events.map((e) => (e as Record<string, unknown>).type);
expect(types).toEqual([
"connected",
"content",
"content",
"status",
"stream_end",
]);
});
});
+19
View File
@@ -0,0 +1,19 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "ESNext",
"moduleResolution": "bundler",
"declaration": true,
"declarationMap": true,
"sourceMap": true,
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"lib": ["ES2022", "DOM"]
},
"include": ["src/**/*.ts"],
"exclude": ["node_modules", "dist", "tests"]
}
+7
View File
@@ -0,0 +1,7 @@
import { defineConfig } from "vitest/config";
export default defineConfig({
test: {
include: ["tests/**/*.test.ts"],
},
});
+242
View File
@@ -0,0 +1,242 @@
"""Tests for turnstone.sdk.console — console client with mocked HTTP transport."""
from __future__ import annotations
import httpx
import pytest
from turnstone.sdk._types import TurnstoneAPIError
from turnstone.sdk.console import AsyncTurnstoneConsole
def _json_response(data: dict, status: int = 200) -> httpx.Response:
return httpx.Response(status, json=data)
def _mock_transport(
responses: dict[str, httpx.Response] | None = None,
) -> httpx.MockTransport:
table = responses or {}
def handler(request: httpx.Request) -> httpx.Response:
key = f"{request.method} {request.url.path}"
if key in table:
return table[key]
return httpx.Response(404, json={"error": "not found"})
return httpx.MockTransport(handler)
# ---------------------------------------------------------------------------
# Cluster overview
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_overview():
transport = _mock_transport(
{
"GET /v1/api/cluster/overview": _json_response(
{
"nodes": 2,
"workstreams": 5,
"states": {"running": 1, "idle": 4},
"aggregate": {"total_tokens": 1000, "total_tool_calls": 20},
"version_drift": False,
"versions": ["0.3.0"],
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.overview()
assert resp.nodes == 2
assert resp.workstreams == 5
@pytest.mark.anyio
async def test_nodes():
transport = _mock_transport(
{
"GET /v1/api/cluster/nodes": _json_response(
{
"nodes": [
{
"node_id": "n1",
"server_url": "http://localhost:8080",
"ws_total": 3,
"ws_running": 1,
"ws_thinking": 0,
"ws_attention": 0,
"ws_idle": 2,
"ws_error": 0,
"total_tokens": 500,
"started": 1700000000.0,
"reachable": True,
"health": {"status": "ok"},
"version": "0.3.0",
}
],
"total": 1,
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.nodes(sort="tokens", limit=50)
assert resp.total == 1
assert resp.nodes[0].node_id == "n1"
@pytest.mark.anyio
async def test_workstreams():
transport = _mock_transport(
{
"GET /v1/api/cluster/workstreams": _json_response(
{
"workstreams": [
{
"id": "ws1",
"name": "test",
"state": "running",
"node": "n1",
}
],
"total": 1,
"page": 1,
"per_page": 50,
"pages": 1,
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.workstreams(state="running", page=1)
assert resp.total == 1
@pytest.mark.anyio
async def test_node_detail():
transport = _mock_transport(
{
"GET /v1/api/cluster/node/n1": _json_response(
{
"node_id": "n1",
"server_url": "http://localhost:8080",
"health": {"status": "ok"},
"workstreams": [],
"aggregate": {"total_tokens": 0, "total_tool_calls": 0},
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.node_detail("n1")
assert resp.node_id == "n1"
@pytest.mark.anyio
async def test_create_workstream():
transport = _mock_transport(
{
"POST /v1/api/cluster/workstreams/new": _json_response(
{"status": "dispatched", "correlation_id": "abc123", "target_node": "n1"}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.create_workstream(node_id="n1", name="test")
assert resp.correlation_id == "abc123"
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_login():
transport = _mock_transport(
{"POST /v1/api/auth/login": _json_response({"status": "ok", "role": "read"})}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.login("tok_test")
assert resp.role == "read"
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_health():
transport = _mock_transport(
{
"GET /health": _json_response(
{
"status": "ok",
"service": "turnstone-console",
"nodes": 2,
"workstreams": 5,
"version_drift": False,
"versions": ["0.3.0"],
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
resp = await client.health()
assert resp.status == "ok"
assert resp.nodes == 2
# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_node_not_found():
transport = _mock_transport(
{"GET /v1/api/cluster/node/bad": httpx.Response(404, json={"error": "Node not found"})}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
with pytest.raises(TurnstoneAPIError) as exc_info:
await client.node_detail("bad")
assert exc_info.value.status_code == 404
@pytest.mark.anyio
async def test_query_params_passed():
"""Verify query params are sent correctly for paginated endpoints."""
captured_url: list[str] = []
def handler(request: httpx.Request) -> httpx.Response:
captured_url.append(str(request.url))
return httpx.Response(
200,
json={
"workstreams": [],
"total": 0,
"page": 2,
"per_page": 25,
"pages": 0,
},
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneConsole(httpx_client=hc)
await client.workstreams(state="running", page=2, per_page=25)
assert "state=running" in captured_url[0]
assert "page=2" in captured_url[0]
assert "per_page=25" in captured_url[0]
+287
View File
@@ -0,0 +1,287 @@
"""Tests for turnstone.sdk.events — SSE event deserialization."""
from turnstone.sdk.events import (
ApproveRequestEvent,
BusyErrorEvent,
ClearUiEvent,
ClusterEvent,
ClusterStateEvent,
ClusterWsClosedEvent,
ClusterWsCreatedEvent,
ClusterWsRenameEvent,
ConnectedEvent,
ContentEvent,
ErrorEvent,
HistoryEvent,
InfoEvent,
NodeJoinedEvent,
NodeLostEvent,
PlanReviewEvent,
ReasoningEvent,
ServerEvent,
StatusEvent,
StreamEndEvent,
ThinkingStartEvent,
ThinkingStopEvent,
ToolInfoEvent,
ToolOutputChunkEvent,
ToolResultEvent,
WsActivityEvent,
WsClosedEvent,
WsRenameEvent,
WsStateEvent,
)
# ---------------------------------------------------------------------------
# Per-workstream events
# ---------------------------------------------------------------------------
def test_connected_event():
e = ServerEvent.from_dict(
{"type": "connected", "model": "gpt-5", "model_alias": "fast", "skip_permissions": True}
)
assert isinstance(e, ConnectedEvent)
assert e.model == "gpt-5"
assert e.model_alias == "fast"
assert e.skip_permissions is True
def test_history_event():
msgs = [{"role": "user", "content": "hi"}]
e = ServerEvent.from_dict({"type": "history", "messages": msgs})
assert isinstance(e, HistoryEvent)
assert e.messages == msgs
def test_thinking_start_stop():
e1 = ServerEvent.from_dict({"type": "thinking_start"})
e2 = ServerEvent.from_dict({"type": "thinking_stop"})
assert isinstance(e1, ThinkingStartEvent)
assert isinstance(e2, ThinkingStopEvent)
def test_content_event():
e = ServerEvent.from_dict({"type": "content", "text": "hello"})
assert isinstance(e, ContentEvent)
assert e.text == "hello"
def test_reasoning_event():
e = ServerEvent.from_dict({"type": "reasoning", "text": "step 1"})
assert isinstance(e, ReasoningEvent)
assert e.text == "step 1"
def test_stream_end_event():
e = ServerEvent.from_dict({"type": "stream_end"})
assert isinstance(e, StreamEndEvent)
def test_tool_info_event():
items = [{"name": "search", "call_id": "c1"}]
e = ServerEvent.from_dict({"type": "tool_info", "items": items})
assert isinstance(e, ToolInfoEvent)
assert e.items == items
def test_approve_request_event():
items = [{"name": "bash", "call_id": "c2", "arguments": "ls"}]
e = ServerEvent.from_dict({"type": "approve_request", "items": items})
assert isinstance(e, ApproveRequestEvent)
assert len(e.items) == 1
def test_tool_result_event():
e = ServerEvent.from_dict(
{"type": "tool_result", "call_id": "c1", "name": "search", "output": "found it"}
)
assert isinstance(e, ToolResultEvent)
assert e.call_id == "c1"
assert e.name == "search"
assert e.output == "found it"
def test_tool_output_chunk_event():
e = ServerEvent.from_dict({"type": "tool_output_chunk", "call_id": "c1", "chunk": "line1\n"})
assert isinstance(e, ToolOutputChunkEvent)
assert e.chunk == "line1\n"
def test_status_event():
e = ServerEvent.from_dict(
{
"type": "status",
"prompt_tokens": 100,
"completion_tokens": 50,
"total_tokens": 150,
"context_window": 128000,
"pct": 0.12,
"effort": "medium",
}
)
assert isinstance(e, StatusEvent)
assert e.prompt_tokens == 100
assert e.total_tokens == 150
assert e.pct == 0.12
assert e.effort == "medium"
def test_plan_review_event():
e = ServerEvent.from_dict({"type": "plan_review", "content": "## Plan\n1. Do X"})
assert isinstance(e, PlanReviewEvent)
assert "Plan" in e.content
def test_info_event():
e = ServerEvent.from_dict({"type": "info", "message": "[compacted]"})
assert isinstance(e, InfoEvent)
assert e.message == "[compacted]"
def test_error_event():
e = ServerEvent.from_dict({"type": "error", "message": "Something broke"})
assert isinstance(e, ErrorEvent)
assert e.message == "Something broke"
def test_busy_error_event():
e = ServerEvent.from_dict({"type": "busy_error", "message": "Already processing a request."})
assert isinstance(e, BusyErrorEvent)
assert "Already" in e.message
def test_clear_ui_event():
e = ServerEvent.from_dict({"type": "clear_ui"})
assert isinstance(e, ClearUiEvent)
# ---------------------------------------------------------------------------
# Global events
# ---------------------------------------------------------------------------
def test_ws_state_event():
e = ServerEvent.from_dict(
{
"type": "ws_state",
"ws_id": "ws1",
"state": "thinking",
"tokens": 500,
"context_ratio": 0.3,
"activity": "Writing code",
"activity_state": "thinking",
}
)
assert isinstance(e, WsStateEvent)
assert e.ws_id == "ws1"
assert e.state == "thinking"
assert e.tokens == 500
def test_ws_activity_event():
e = ServerEvent.from_dict(
{"type": "ws_activity", "ws_id": "ws1", "activity": "reading", "activity_state": "tool"}
)
assert isinstance(e, WsActivityEvent)
assert e.activity == "reading"
def test_ws_rename_event():
e = ServerEvent.from_dict({"type": "ws_rename", "ws_id": "ws1", "name": "My Chat"})
assert isinstance(e, WsRenameEvent)
assert e.name == "My Chat"
def test_ws_closed_event():
e = ServerEvent.from_dict({"type": "ws_closed", "ws_id": "ws1", "name": "old"})
assert isinstance(e, WsClosedEvent)
assert e.name == "old"
# ---------------------------------------------------------------------------
# Cluster events
# ---------------------------------------------------------------------------
def test_node_joined_event():
e = ClusterEvent.from_dict({"type": "node_joined", "node_id": "host1_abc"})
assert isinstance(e, NodeJoinedEvent)
assert e.node_id == "host1_abc"
def test_node_lost_event():
e = ClusterEvent.from_dict({"type": "node_lost", "node_id": "host2_def"})
assert isinstance(e, NodeLostEvent)
assert e.node_id == "host2_def"
def test_cluster_state_event():
e = ClusterEvent.from_dict(
{
"type": "cluster_state",
"ws_id": "ws1",
"node_id": "n1",
"state": "running",
"tokens": 1000,
"context_ratio": 0.5,
"activity": "executing tool",
"activity_state": "tool",
}
)
assert isinstance(e, ClusterStateEvent)
assert e.node_id == "n1"
assert e.state == "running"
assert e.tokens == 1000
def test_cluster_ws_created_event():
e = ClusterEvent.from_dict(
{"type": "ws_created", "ws_id": "ws2", "node_id": "n1", "name": "New WS"}
)
assert isinstance(e, ClusterWsCreatedEvent)
assert e.ws_id == "ws2"
assert e.name == "New WS"
def test_cluster_ws_closed_event():
e = ClusterEvent.from_dict({"type": "ws_closed", "ws_id": "ws2"})
assert isinstance(e, ClusterWsClosedEvent)
assert e.ws_id == "ws2"
def test_cluster_ws_rename_event():
e = ClusterEvent.from_dict({"type": "ws_rename", "ws_id": "ws2", "name": "Renamed"})
assert isinstance(e, ClusterWsRenameEvent)
assert e.name == "Renamed"
# ---------------------------------------------------------------------------
# Edge cases
# ---------------------------------------------------------------------------
def test_unknown_server_event_falls_back():
e = ServerEvent.from_dict({"type": "future_event", "ws_id": "ws1"})
assert type(e) is ServerEvent
assert e.type == "future_event"
assert e.ws_id == "ws1"
def test_unknown_cluster_event_falls_back():
e = ClusterEvent.from_dict({"type": "future_cluster_event"})
assert type(e) is ClusterEvent
assert e.type == "future_cluster_event"
def test_extra_fields_ignored():
e = ServerEvent.from_dict({"type": "content", "text": "hi", "extra_field": 999})
assert isinstance(e, ContentEvent)
assert e.text == "hi"
def test_missing_type_defaults_to_base():
e = ServerEvent.from_dict({"ws_id": "ws1"})
assert type(e) is ServerEvent
assert e.ws_id == "ws1"
assert e.type == ""
+281
View File
@@ -0,0 +1,281 @@
"""Tests for turnstone.sdk.server — server client with mocked HTTP transport."""
from __future__ import annotations
import json
import httpx
import pytest
from turnstone.sdk._types import TurnstoneAPIError
from turnstone.sdk.server import AsyncTurnstoneServer
def _mock_transport(
responses: dict[str, httpx.Response] | None = None,
) -> httpx.MockTransport:
"""Create a mock transport that routes by method+path."""
table = responses or {}
def handler(request: httpx.Request) -> httpx.Response:
key = f"{request.method} {request.url.path}"
if key in table:
return table[key]
return httpx.Response(404, json={"error": "not found"})
return httpx.MockTransport(handler)
def _json_response(data: dict, status: int = 200) -> httpx.Response:
return httpx.Response(status, json=data)
# ---------------------------------------------------------------------------
# Workstream management
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_list_workstreams():
transport = _mock_transport(
{
"GET /v1/api/workstreams": _json_response(
{"workstreams": [{"id": "ws1", "name": "test", "state": "idle"}]}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.list_workstreams()
assert len(resp.workstreams) == 1
assert resp.workstreams[0].id == "ws1"
@pytest.mark.anyio
async def test_dashboard():
transport = _mock_transport(
{
"GET /v1/api/dashboard": _json_response(
{
"workstreams": [
{
"id": "ws1",
"name": "demo",
"state": "idle",
"tokens": 100,
"context_ratio": 0.1,
}
],
"aggregate": {
"total_tokens": 100,
"total_tool_calls": 5,
"active_count": 1,
"total_count": 1,
},
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.dashboard()
assert resp.aggregate.total_tokens == 100
assert len(resp.workstreams) == 1
@pytest.mark.anyio
async def test_create_workstream():
transport = _mock_transport(
{"POST /v1/api/workstreams/new": _json_response({"ws_id": "ws_new", "name": "Analysis"})}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.create_workstream(name="Analysis")
assert resp.ws_id == "ws_new"
assert resp.name == "Analysis"
@pytest.mark.anyio
async def test_close_workstream():
transport = _mock_transport(
{"POST /v1/api/workstreams/close": _json_response({"status": "ok"})}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.close_workstream("ws1")
assert resp.status == "ok"
# ---------------------------------------------------------------------------
# Chat interaction
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_send():
transport = _mock_transport({"POST /v1/api/send": _json_response({"status": "ok"})})
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.send("Hello", "ws1")
assert resp.status == "ok"
@pytest.mark.anyio
async def test_approve():
transport = _mock_transport({"POST /v1/api/approve": _json_response({"status": "ok"})})
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.approve(ws_id="ws1", approved=True, feedback="looks good")
assert resp.status == "ok"
@pytest.mark.anyio
async def test_plan_feedback():
transport = _mock_transport({"POST /v1/api/plan": _json_response({"status": "ok"})})
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.plan_feedback(ws_id="ws1", feedback="approved")
assert resp.status == "ok"
@pytest.mark.anyio
async def test_command():
transport = _mock_transport({"POST /v1/api/command": _json_response({"status": "ok"})})
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.command(ws_id="ws1", command="/clear")
assert resp.status == "ok"
# ---------------------------------------------------------------------------
# Sessions
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_list_sessions():
transport = _mock_transport(
{
"GET /v1/api/sessions": _json_response(
{
"sessions": [
{
"session_id": "s1",
"title": "test",
"created": "2024-01-01",
"updated": "2024-01-02",
"message_count": 5,
}
]
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.list_sessions()
assert len(resp.sessions) == 1
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_login():
transport = _mock_transport(
{"POST /v1/api/auth/login": _json_response({"status": "ok", "role": "full"})}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.login("test_token")
assert resp.role == "full"
@pytest.mark.anyio
async def test_logout():
transport = _mock_transport({"POST /v1/api/auth/logout": _json_response({"status": "ok"})})
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.logout()
assert resp.status == "ok"
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_health():
transport = _mock_transport(
{
"GET /health": _json_response(
{
"status": "ok",
"version": "0.3.0",
"uptime_seconds": 120.0,
"model": "gpt-5",
"workstreams": {"total": 1, "idle": 1},
}
)
}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
resp = await client.health()
assert resp.status == "ok"
assert resp.version == "0.3.0"
# ---------------------------------------------------------------------------
# Error handling
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_api_error_raised():
transport = _mock_transport(
{"POST /v1/api/send": httpx.Response(404, json={"error": "Unknown workstream"})}
)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
with pytest.raises(TurnstoneAPIError) as exc_info:
await client.send("hi", "bad_ws")
assert exc_info.value.status_code == 404
assert "Unknown workstream" in exc_info.value.message
@pytest.mark.anyio
async def test_auth_header_injected():
"""Verify the Authorization header is set when a token is provided."""
captured_headers: dict[str, str] = {}
def handler(request: httpx.Request) -> httpx.Response:
captured_headers.update(dict(request.headers))
return httpx.Response(200, json={"workstreams": []})
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
# Manually set auth header since we're injecting the client
hc.headers["Authorization"] = "Bearer tok_test"
client = AsyncTurnstoneServer(httpx_client=hc)
await client.list_workstreams()
assert captured_headers.get("authorization") == "Bearer tok_test"
@pytest.mark.anyio
async def test_request_body_correct():
"""Verify POST requests send the correct JSON body."""
captured_body: dict = {}
def handler(request: httpx.Request) -> httpx.Response:
captured_body.update(json.loads(request.content))
return httpx.Response(200, json={"status": "ok"})
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = AsyncTurnstoneServer(httpx_client=hc)
await client.send("Hello world", "ws_123")
assert captured_body == {"message": "Hello world", "ws_id": "ws_123"}
+110
View File
@@ -0,0 +1,110 @@
"""Tests for _BaseClient._stream_sse — SSE stream parsing."""
from __future__ import annotations
import httpx
import pytest
from turnstone.sdk._base import _BaseClient
def _sse_response(*events: str) -> httpx.Response:
"""Build a mock SSE response from data strings."""
body = ""
for event in events:
body += f"data: {event}\n\n"
return httpx.Response(
200,
content=body.encode(),
headers={"content-type": "text/event-stream"},
)
@pytest.mark.anyio
async def test_stream_sse_yields_json():
"""SSE stream with valid JSON payloads yields parsed dicts."""
def handler(request: httpx.Request) -> httpx.Response:
return _sse_response(
'{"type": "content", "text": "hello"}',
'{"type": "stream_end"}',
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = _BaseClient(httpx_client=hc)
events = []
async for data in client._stream_sse("/v1/api/events", params={"ws_id": "ws1"}):
events.append(data)
assert len(events) == 2
assert events[0]["type"] == "content"
assert events[1]["type"] == "stream_end"
@pytest.mark.anyio
async def test_stream_sse_skips_malformed_json():
"""Malformed JSON in SSE data is silently skipped."""
def handler(request: httpx.Request) -> httpx.Response:
return _sse_response(
"not-json",
'{"type": "content", "text": "ok"}',
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = _BaseClient(httpx_client=hc)
events = []
async for data in client._stream_sse("/test"):
events.append(data)
assert len(events) == 1
assert events[0]["type"] == "content"
@pytest.mark.anyio
async def test_stream_sse_skips_empty_data():
"""SSE frames with empty data field are skipped."""
def handler(request: httpx.Request) -> httpx.Response:
# Build response with an empty data line
body = 'data: \n\ndata: {"type": "info", "message": "ok"}\n\n'
return httpx.Response(
200,
content=body.encode(),
headers={"content-type": "text/event-stream"},
)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = _BaseClient(httpx_client=hc)
events = []
async for data in client._stream_sse("/test"):
events.append(data)
# Empty " " data is not valid JSON, so skipped; only "info" event remains
assert len(events) == 1
assert events[0]["type"] == "info"
@pytest.mark.anyio
async def test_stream_sse_multiple_events():
"""Multiple SSE events are yielded in order."""
event_data = [
'{"type": "connected", "model": "gpt-5"}',
'{"type": "content", "text": "word1 "}',
'{"type": "content", "text": "word2"}',
'{"type": "status", "total_tokens": 10}',
'{"type": "stream_end"}',
]
def handler(request: httpx.Request) -> httpx.Response:
return _sse_response(*event_data)
transport = httpx.MockTransport(handler)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as hc:
client = _BaseClient(httpx_client=hc)
events = []
async for data in client._stream_sse("/test"):
events.append(data)
assert len(events) == 5
types = [e["type"] for e in events]
assert types == ["connected", "content", "content", "status", "stream_end"]
+135
View File
@@ -0,0 +1,135 @@
"""Tests for synchronous SDK wrappers (TurnstoneServer, TurnstoneConsole)."""
from __future__ import annotations
import httpx
from turnstone.sdk._sync import _SyncRunner
from turnstone.sdk.console import AsyncTurnstoneConsole, TurnstoneConsole
from turnstone.sdk.server import AsyncTurnstoneServer, TurnstoneServer
def _json_response(data: dict, status: int = 200) -> httpx.Response:
return httpx.Response(status, json=data)
# ---------------------------------------------------------------------------
# _SyncRunner
# ---------------------------------------------------------------------------
def test_sync_runner_basic():
"""_SyncRunner can execute a simple async coroutine."""
runner = _SyncRunner()
try:
import asyncio
async def _add(a: int, b: int) -> int:
await asyncio.sleep(0)
return a + b
result = runner.run(_add(1, 2))
assert result == 3
finally:
runner.close()
def test_sync_runner_iter():
"""_SyncRunner.run_iter iterates over an async generator."""
runner = _SyncRunner()
try:
async def _gen():
for i in range(3):
yield i
items = list(runner.run_iter(_gen()))
assert items == [0, 1, 2]
finally:
runner.close()
# ---------------------------------------------------------------------------
# TurnstoneServer (sync)
# ---------------------------------------------------------------------------
def test_sync_server_list_workstreams():
"""Sync server client delegates to async and returns correct model."""
def handler(request: httpx.Request) -> httpx.Response:
return _json_response({"workstreams": [{"id": "ws1", "name": "test", "state": "idle"}]})
# We need to create the async client with a mock transport,
# then wrap it in the sync client
transport = httpx.MockTransport(handler)
hc = httpx.AsyncClient(transport=transport, base_url="http://test")
async_client = AsyncTurnstoneServer(httpx_client=hc)
server = TurnstoneServer.__new__(TurnstoneServer)
server._runner = _SyncRunner()
server._async = async_client
try:
resp = server.list_workstreams()
assert len(resp.workstreams) == 1
assert resp.workstreams[0].id == "ws1"
finally:
server.close()
def test_sync_server_context_manager():
"""TurnstoneServer can be used as a context manager."""
def handler(request: httpx.Request) -> httpx.Response:
return _json_response(
{"status": "ok", "version": "0.3.0", "uptime_seconds": 1.0, "model": "gpt-5"}
)
transport = httpx.MockTransport(handler)
hc = httpx.AsyncClient(transport=transport, base_url="http://test")
async_client = AsyncTurnstoneServer(httpx_client=hc)
server = TurnstoneServer.__new__(TurnstoneServer)
server._runner = _SyncRunner()
server._async = async_client
with server as s:
resp = s.health()
assert resp.status == "ok"
# ---------------------------------------------------------------------------
# TurnstoneConsole (sync)
# ---------------------------------------------------------------------------
def test_sync_console_overview():
"""Sync console client delegates to async and returns correct model."""
def handler(request: httpx.Request) -> httpx.Response:
return _json_response(
{
"nodes": 1,
"workstreams": 3,
"states": {"idle": 3},
"aggregate": {"total_tokens": 100, "total_tool_calls": 0},
"version_drift": False,
"versions": ["0.3.0"],
}
)
transport = httpx.MockTransport(handler)
hc = httpx.AsyncClient(transport=transport, base_url="http://test")
async_client = AsyncTurnstoneConsole(httpx_client=hc)
console = TurnstoneConsole.__new__(TurnstoneConsole)
console._runner = _SyncRunner()
console._async = async_client
try:
resp = console.overview()
assert resp.nodes == 1
assert resp.workstreams == 3
finally:
console.close()
+90
View File
@@ -0,0 +1,90 @@
"""Turnstone Client SDK — typed HTTP clients for server and console APIs.
Quick start::
from turnstone.sdk import TurnstoneServer
with TurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
ws = client.create_workstream(name="demo")
result = client.send_and_wait("Hello!", ws.ws_id)
print(result.content)
"""
from __future__ import annotations
from turnstone.sdk._types import TurnResult, TurnstoneAPIError
from turnstone.sdk.console import AsyncTurnstoneConsole, TurnstoneConsole
from turnstone.sdk.events import (
ApproveRequestEvent,
BusyErrorEvent,
ClearUiEvent,
ClusterEvent,
ClusterStateEvent,
ClusterWsClosedEvent,
ClusterWsCreatedEvent,
ClusterWsRenameEvent,
ConnectedEvent,
ContentEvent,
ErrorEvent,
HistoryEvent,
InfoEvent,
NodeJoinedEvent,
NodeLostEvent,
PlanReviewEvent,
ReasoningEvent,
ServerEvent,
StatusEvent,
StreamEndEvent,
ThinkingStartEvent,
ThinkingStopEvent,
ToolInfoEvent,
ToolOutputChunkEvent,
ToolResultEvent,
WsActivityEvent,
WsClosedEvent,
WsRenameEvent,
WsStateEvent,
)
from turnstone.sdk.server import AsyncTurnstoneServer, TurnstoneServer
__all__ = [
# Clients
"AsyncTurnstoneServer",
"TurnstoneServer",
"AsyncTurnstoneConsole",
"TurnstoneConsole",
# Result types
"TurnResult",
"TurnstoneAPIError",
# Server events
"ServerEvent",
"ConnectedEvent",
"HistoryEvent",
"ThinkingStartEvent",
"ThinkingStopEvent",
"ReasoningEvent",
"ContentEvent",
"StreamEndEvent",
"ToolInfoEvent",
"ApproveRequestEvent",
"ToolResultEvent",
"ToolOutputChunkEvent",
"StatusEvent",
"PlanReviewEvent",
"InfoEvent",
"ErrorEvent",
"BusyErrorEvent",
"ClearUiEvent",
"WsStateEvent",
"WsActivityEvent",
"WsRenameEvent",
"WsClosedEvent",
# Cluster events
"ClusterEvent",
"NodeJoinedEvent",
"NodeLostEvent",
"ClusterStateEvent",
"ClusterWsCreatedEvent",
"ClusterWsClosedEvent",
"ClusterWsRenameEvent",
]
+125
View File
@@ -0,0 +1,125 @@
"""Shared HTTP client base for the turnstone SDK."""
from __future__ import annotations
import contextlib
import json
from typing import TYPE_CHECKING, Any, Self, TypeVar, overload
import httpx
from turnstone.sdk._types import TurnstoneAPIError
if TYPE_CHECKING:
from collections.abc import AsyncIterator
T = TypeVar("T")
class _BaseClient:
"""Async HTTP client base shared by server and console clients."""
def __init__(
self,
base_url: str = "http://localhost:8080",
token: str = "",
timeout: float = 30.0,
httpx_client: httpx.AsyncClient | None = None,
) -> None:
"""Initialise the client.
When *httpx_client* is provided it is used directly and *base_url*,
*token*, and *timeout* are ignored — configure headers and base URL
on the injected client instead.
"""
headers: dict[str, str] = {}
if token:
headers["Authorization"] = f"Bearer {token}"
if httpx_client is not None:
self._client = httpx_client
self._owns_client = False
else:
self._client = httpx.AsyncClient(
base_url=base_url, timeout=timeout, headers=headers, follow_redirects=True
)
self._owns_client = True
# -- request helpers -----------------------------------------------------
@overload
async def _request(
self,
method: str,
path: str,
*,
json_body: dict[str, Any] | None = ...,
params: dict[str, Any] | None = ...,
response_model: type[T],
) -> T: ...
@overload
async def _request(
self,
method: str,
path: str,
*,
json_body: dict[str, Any] | None = ...,
params: dict[str, Any] | None = ...,
response_model: None = ...,
) -> dict[str, Any]: ...
async def _request(
self,
method: str,
path: str,
*,
json_body: dict[str, Any] | None = None,
params: dict[str, Any] | None = None,
response_model: type[Any] | None = None,
) -> Any:
"""Execute an HTTP request and return parsed response data.
Raises :class:`TurnstoneAPIError` on non-2xx responses.
"""
resp = await self._client.request(method, path, json=json_body, params=params)
if resp.status_code >= 400:
# Try to extract error message from JSON body
msg = ""
with contextlib.suppress(Exception):
body = resp.json()
msg = body.get("error", body.get("detail", ""))
if not msg:
msg = resp.text[:200]
raise TurnstoneAPIError(resp.status_code, msg)
data: dict[str, Any] = resp.json()
if response_model is not None:
return response_model.model_validate(data)
return data
async def _stream_sse(
self,
path: str,
*,
params: dict[str, Any] | None = None,
) -> AsyncIterator[dict[str, Any]]:
"""Open an SSE stream and yield parsed JSON dicts."""
from httpx_sse import aconnect_sse
async with aconnect_sse(self._client, "GET", path, params=params) as source:
async for sse in source.aiter_sse():
if sse.data:
with contextlib.suppress(json.JSONDecodeError):
yield json.loads(sse.data)
# -- lifecycle -----------------------------------------------------------
async def aclose(self) -> None:
"""Close the underlying HTTP client (if owned)."""
if self._owns_client:
await self._client.aclose()
async def __aenter__(self) -> Self:
return self
async def __aexit__(self, *exc: object) -> None:
await self.aclose()
+65
View File
@@ -0,0 +1,65 @@
"""Synchronous execution helper for the turnstone SDK.
Maintains a background event loop on a daemon thread so that async
client methods can be called from synchronous code without the
overhead of ``asyncio.run()`` per call.
"""
from __future__ import annotations
import asyncio
import threading
from typing import TYPE_CHECKING, Any, TypeVar
if TYPE_CHECKING:
import concurrent.futures
from collections.abc import AsyncIterator, Coroutine, Iterator
T = TypeVar("T")
class _SyncRunner:
"""Run async coroutines synchronously via a persistent background loop."""
def __init__(self) -> None:
self._loop: asyncio.AbstractEventLoop | None = None
self._thread: threading.Thread | None = None
self._lock = threading.Lock()
def _ensure_loop(self) -> asyncio.AbstractEventLoop:
with self._lock:
if self._loop is None or self._loop.is_closed():
self._loop = asyncio.new_event_loop()
self._thread = threading.Thread(target=self._loop.run_forever, daemon=True)
self._thread.start()
return self._loop
def run(self, coro: Coroutine[Any, Any, T]) -> T:
"""Submit *coro* to the background loop and block for the result."""
loop = self._ensure_loop()
future = asyncio.run_coroutine_threadsafe(coro, loop)
return future.result()
def run_iter(self, async_gen: AsyncIterator[T]) -> Iterator[T]:
"""Synchronously iterate over an async generator."""
loop = self._ensure_loop()
try:
while True:
awaitable = async_gen.__anext__()
future: concurrent.futures.Future[T] = asyncio.run_coroutine_threadsafe(
awaitable, # type: ignore[arg-type]
loop,
)
yield future.result()
except StopAsyncIteration:
return
def close(self) -> None:
"""Shut down the background event loop."""
if self._loop is not None and not self._loop.is_closed():
self._loop.call_soon_threadsafe(self._loop.stop)
if self._thread is not None:
self._thread.join(timeout=5)
self._loop.close()
self._loop = None
self._thread = None
+42
View File
@@ -0,0 +1,42 @@
"""Shared result types and exceptions for the turnstone SDK."""
from __future__ import annotations
from dataclasses import dataclass, field
@dataclass
class TurnResult:
"""Aggregated result of a send_and_wait call.
Mirrors the shape of :class:`turnstone.mq.client.TurnResult` but
operates over HTTP/SSE instead of Redis pub/sub.
"""
ws_id: str = ""
content_parts: list[str] = field(default_factory=list)
reasoning_parts: list[str] = field(default_factory=list)
tool_results: list[tuple[str, str]] = field(default_factory=list)
errors: list[str] = field(default_factory=list)
timed_out: bool = False
@property
def content(self) -> str:
return "".join(self.content_parts)
@property
def reasoning(self) -> str:
return "".join(self.reasoning_parts)
@property
def ok(self) -> bool:
return not self.timed_out and not self.errors
class TurnstoneAPIError(Exception):
"""Raised when a server returns a non-2xx response."""
def __init__(self, status_code: int, message: str) -> None:
self.status_code = status_code
self.message = message
super().__init__(f"HTTP {status_code}: {message}")
+235
View File
@@ -0,0 +1,235 @@
"""Typed HTTP clients for the turnstone console API.
Usage::
from turnstone.sdk import TurnstoneConsole
with TurnstoneConsole("http://localhost:8081", token="tok_xxx") as client:
overview = client.overview()
print(f"Nodes: {overview.nodes}, Workstreams: {overview.workstreams}")
"""
from __future__ import annotations
from typing import TYPE_CHECKING, Any
from turnstone.api.console_schemas import (
ClusterNodesResponse,
ClusterOverviewResponse,
ClusterWorkstreamsResponse,
ConsoleCreateWsResponse,
ConsoleHealthResponse,
NodeDetailResponse,
)
from turnstone.api.schemas import AuthLoginResponse, StatusResponse
from turnstone.sdk._base import _BaseClient
from turnstone.sdk._sync import _SyncRunner
from turnstone.sdk.events import ClusterEvent
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Iterator
import httpx
class AsyncTurnstoneConsole(_BaseClient):
"""Async client for the turnstone console API."""
def __init__(
self,
base_url: str = "http://localhost:8081",
token: str = "",
timeout: float = 30.0,
httpx_client: httpx.AsyncClient | None = None,
) -> None:
super().__init__(base_url=base_url, token=token, timeout=timeout, httpx_client=httpx_client)
# -- cluster overview ----------------------------------------------------
async def overview(self) -> ClusterOverviewResponse:
return await self._request(
"GET", "/v1/api/cluster/overview", response_model=ClusterOverviewResponse
)
async def nodes(
self,
*,
sort: str = "activity",
limit: int = 100,
offset: int = 0,
) -> ClusterNodesResponse:
params: dict[str, Any] = {"sort": sort, "limit": limit, "offset": offset}
return await self._request(
"GET", "/v1/api/cluster/nodes", params=params, response_model=ClusterNodesResponse
)
async def workstreams(
self,
*,
state: str = "",
node: str = "",
search: str = "",
sort: str = "state",
page: int = 1,
per_page: int = 50,
) -> ClusterWorkstreamsResponse:
params: dict[str, Any] = {"sort": sort, "page": page, "per_page": per_page}
if state:
params["state"] = state
if node:
params["node"] = node
if search:
params["search"] = search
return await self._request(
"GET",
"/v1/api/cluster/workstreams",
params=params,
response_model=ClusterWorkstreamsResponse,
)
async def node_detail(self, node_id: str) -> NodeDetailResponse:
return await self._request(
"GET", f"/v1/api/cluster/node/{node_id}", response_model=NodeDetailResponse
)
async def create_workstream(
self,
*,
node_id: str = "",
name: str = "",
model: str = "",
) -> ConsoleCreateWsResponse:
body: dict[str, Any] = {}
if node_id:
body["node_id"] = node_id
if name:
body["name"] = name
if model:
body["model"] = model
return await self._request(
"POST",
"/v1/api/cluster/workstreams/new",
json_body=body,
response_model=ConsoleCreateWsResponse,
)
# -- streaming -----------------------------------------------------------
async def stream_cluster_events(self) -> AsyncIterator[ClusterEvent]:
"""Iterate over cluster SSE events."""
async for data in self._stream_sse("/v1/api/cluster/events"):
yield ClusterEvent.from_dict(data)
# -- auth ----------------------------------------------------------------
async def login(self, token: str) -> AuthLoginResponse:
return await self._request(
"POST",
"/v1/api/auth/login",
json_body={"token": token},
response_model=AuthLoginResponse,
)
async def logout(self) -> StatusResponse:
return await self._request("POST", "/v1/api/auth/logout", response_model=StatusResponse)
# -- health --------------------------------------------------------------
async def health(self) -> ConsoleHealthResponse:
return await self._request("GET", "/health", response_model=ConsoleHealthResponse)
class TurnstoneConsole:
"""Synchronous client for the turnstone console API.
Wraps :class:`AsyncTurnstoneConsole` via a background event loop.
Usage::
with TurnstoneConsole("http://localhost:8081", token="tok_xxx") as client:
overview = client.overview()
print(f"Nodes: {overview.nodes}")
"""
def __init__(
self,
base_url: str = "http://localhost:8081",
token: str = "",
timeout: float = 30.0,
) -> None:
self._runner = _SyncRunner()
self._async = AsyncTurnstoneConsole(base_url=base_url, token=token, timeout=timeout)
# -- cluster overview ----------------------------------------------------
def overview(self) -> ClusterOverviewResponse:
return self._runner.run(self._async.overview())
def nodes(
self,
*,
sort: str = "activity",
limit: int = 100,
offset: int = 0,
) -> ClusterNodesResponse:
return self._runner.run(self._async.nodes(sort=sort, limit=limit, offset=offset))
def workstreams(
self,
*,
state: str = "",
node: str = "",
search: str = "",
sort: str = "state",
page: int = 1,
per_page: int = 50,
) -> ClusterWorkstreamsResponse:
return self._runner.run(
self._async.workstreams(
state=state, node=node, search=search, sort=sort, page=page, per_page=per_page
)
)
def node_detail(self, node_id: str) -> NodeDetailResponse:
return self._runner.run(self._async.node_detail(node_id))
def create_workstream(
self,
*,
node_id: str = "",
name: str = "",
model: str = "",
) -> ConsoleCreateWsResponse:
return self._runner.run(
self._async.create_workstream(node_id=node_id, name=name, model=model)
)
# -- streaming -----------------------------------------------------------
def stream_cluster_events(self) -> Iterator[ClusterEvent]:
return self._runner.run_iter(self._async.stream_cluster_events())
# -- auth ----------------------------------------------------------------
def login(self, token: str) -> AuthLoginResponse:
return self._runner.run(self._async.login(token))
def logout(self) -> StatusResponse:
return self._runner.run(self._async.logout())
# -- health --------------------------------------------------------------
def health(self) -> ConsoleHealthResponse:
return self._runner.run(self._async.health())
# -- lifecycle -----------------------------------------------------------
def close(self) -> None:
self._runner.run(self._async.aclose())
self._runner.close()
def __enter__(self) -> TurnstoneConsole:
return self
def __exit__(self, *exc: object) -> None:
self.close()
+300
View File
@@ -0,0 +1,300 @@
"""Standalone SSE event dataclasses for the turnstone SDK.
These types match the JSON payloads emitted by the server and console
SSE endpoints. They are intentionally decoupled from the MQ protocol
events in ``turnstone.mq.protocol`` so that SDK consumers do not need
the ``redis`` optional dependency.
"""
from __future__ import annotations
from dataclasses import dataclass, field, fields
from typing import Any
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _fields_of(cls: type[Any]) -> frozenset[str]:
return frozenset(f.name for f in fields(cls))
# ---------------------------------------------------------------------------
# Server per-workstream events (/v1/api/events?ws_id=X)
# ---------------------------------------------------------------------------
@dataclass
class ServerEvent:
"""Base class for all server SSE events."""
type: str = ""
ws_id: str = ""
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ServerEvent:
"""Deserialize an SSE JSON payload into a typed event."""
etype = data.get("type", "")
klass = _SERVER_REGISTRY.get(etype, ServerEvent)
valid = _fields_of(klass)
return klass(**{k: v for k, v in data.items() if k in valid})
@dataclass
class ConnectedEvent(ServerEvent):
type: str = "connected"
model: str = ""
model_alias: str = ""
skip_permissions: bool = False
@dataclass
class HistoryEvent(ServerEvent):
type: str = "history"
messages: list[dict[str, Any]] = field(default_factory=list)
@dataclass
class ThinkingStartEvent(ServerEvent):
type: str = "thinking_start"
@dataclass
class ThinkingStopEvent(ServerEvent):
type: str = "thinking_stop"
@dataclass
class ReasoningEvent(ServerEvent):
type: str = "reasoning"
text: str = ""
@dataclass
class ContentEvent(ServerEvent):
type: str = "content"
text: str = ""
@dataclass
class StreamEndEvent(ServerEvent):
type: str = "stream_end"
@dataclass
class ToolInfoEvent(ServerEvent):
type: str = "tool_info"
items: list[dict[str, Any]] = field(default_factory=list)
@dataclass
class ApproveRequestEvent(ServerEvent):
type: str = "approve_request"
items: list[dict[str, Any]] = field(default_factory=list)
@dataclass
class ToolResultEvent(ServerEvent):
type: str = "tool_result"
call_id: str = ""
name: str = ""
output: str = ""
@dataclass
class ToolOutputChunkEvent(ServerEvent):
type: str = "tool_output_chunk"
call_id: str = ""
chunk: str = ""
@dataclass
class StatusEvent(ServerEvent):
type: str = "status"
prompt_tokens: int = 0
completion_tokens: int = 0
total_tokens: int = 0
context_window: int = 0
pct: float = 0.0
effort: str = ""
@dataclass
class PlanReviewEvent(ServerEvent):
type: str = "plan_review"
content: str = ""
@dataclass
class InfoEvent(ServerEvent):
type: str = "info"
message: str = ""
@dataclass
class ErrorEvent(ServerEvent):
type: str = "error"
message: str = ""
@dataclass
class BusyErrorEvent(ServerEvent):
type: str = "busy_error"
message: str = ""
@dataclass
class ClearUiEvent(ServerEvent):
type: str = "clear_ui"
# ---------------------------------------------------------------------------
# Server global events (/v1/api/events/global)
# ---------------------------------------------------------------------------
@dataclass
class WsStateEvent(ServerEvent):
type: str = "ws_state"
state: str = ""
tokens: int = 0
context_ratio: float = 0.0
activity: str = ""
activity_state: str = ""
@dataclass
class WsActivityEvent(ServerEvent):
type: str = "ws_activity"
activity: str = ""
activity_state: str = ""
@dataclass
class WsRenameEvent(ServerEvent):
type: str = "ws_rename"
name: str = ""
@dataclass
class WsClosedEvent(ServerEvent):
type: str = "ws_closed"
name: str = ""
# ---------------------------------------------------------------------------
# Console cluster events (/v1/api/cluster/events)
# ---------------------------------------------------------------------------
@dataclass
class ClusterEvent:
"""Base class for console cluster SSE events."""
type: str = ""
@classmethod
def from_dict(cls, data: dict[str, Any]) -> ClusterEvent:
etype = data.get("type", "")
klass = _CLUSTER_REGISTRY.get(etype, ClusterEvent)
valid = _fields_of(klass)
return klass(**{k: v for k, v in data.items() if k in valid})
@dataclass
class NodeJoinedEvent(ClusterEvent):
type: str = "node_joined"
node_id: str = ""
@dataclass
class NodeLostEvent(ClusterEvent):
type: str = "node_lost"
node_id: str = ""
@dataclass
class ClusterStateEvent(ClusterEvent):
type: str = "cluster_state"
ws_id: str = ""
node_id: str = ""
state: str = ""
tokens: int = 0
context_ratio: float = 0.0
activity: str = ""
activity_state: str = ""
@dataclass
class ClusterWsCreatedEvent(ClusterEvent):
type: str = "ws_created"
ws_id: str = ""
node_id: str = ""
name: str = ""
@dataclass
class ClusterWsClosedEvent(ClusterEvent):
type: str = "ws_closed"
ws_id: str = ""
@dataclass
class ClusterWsRenameEvent(ClusterEvent):
type: str = "ws_rename"
ws_id: str = ""
name: str = ""
# ---------------------------------------------------------------------------
# Type registries (built after all classes are defined)
# ---------------------------------------------------------------------------
def _type_default(cls: type[Any]) -> str:
"""Return the default value of the ``type`` field for a dataclass."""
for f in fields(cls):
if f.name == "type":
return f.default # type: ignore[return-value]
return ""
_SERVER_REGISTRY: dict[str, type[ServerEvent]] = {
_type_default(cls): cls
for cls in [
ConnectedEvent,
HistoryEvent,
ThinkingStartEvent,
ThinkingStopEvent,
ReasoningEvent,
ContentEvent,
StreamEndEvent,
ToolInfoEvent,
ApproveRequestEvent,
ToolResultEvent,
ToolOutputChunkEvent,
StatusEvent,
PlanReviewEvent,
InfoEvent,
ErrorEvent,
BusyErrorEvent,
ClearUiEvent,
WsStateEvent,
WsActivityEvent,
WsRenameEvent,
WsClosedEvent,
]
}
_CLUSTER_REGISTRY: dict[str, type[ClusterEvent]] = {
_type_default(cls): cls
for cls in [
NodeJoinedEvent,
NodeLostEvent,
ClusterStateEvent,
ClusterWsCreatedEvent,
ClusterWsClosedEvent,
ClusterWsRenameEvent,
]
}
View File
+350
View File
@@ -0,0 +1,350 @@
"""Typed HTTP clients for the turnstone server API.
Usage::
from turnstone.sdk import TurnstoneServer
with TurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
ws = client.create_workstream(name="Analysis")
result = client.send_and_wait("Hello", ws.ws_id)
print(result.content)
"""
from __future__ import annotations
import asyncio
import contextlib
from typing import TYPE_CHECKING, Any
from turnstone.api.schemas import AuthLoginResponse, StatusResponse
from turnstone.api.server_schemas import (
CreateWorkstreamResponse,
DashboardResponse,
HealthResponse,
ListSessionsResponse,
ListWorkstreamsResponse,
SendResponse,
)
from turnstone.sdk._base import _BaseClient
from turnstone.sdk._sync import _SyncRunner
from turnstone.sdk._types import TurnResult
from turnstone.sdk.events import (
ContentEvent,
ErrorEvent,
ReasoningEvent,
ServerEvent,
ToolResultEvent,
WsStateEvent,
)
if TYPE_CHECKING:
from collections.abc import AsyncIterator, Callable, Iterator
import httpx
class AsyncTurnstoneServer(_BaseClient):
"""Async client for the turnstone server API."""
def __init__(
self,
base_url: str = "http://localhost:8080",
token: str = "",
timeout: float = 30.0,
httpx_client: httpx.AsyncClient | None = None,
) -> None:
super().__init__(base_url=base_url, token=token, timeout=timeout, httpx_client=httpx_client)
# -- workstream management -----------------------------------------------
async def list_workstreams(self) -> ListWorkstreamsResponse:
return await self._request(
"GET", "/v1/api/workstreams", response_model=ListWorkstreamsResponse
)
async def dashboard(self) -> DashboardResponse:
return await self._request("GET", "/v1/api/dashboard", response_model=DashboardResponse)
async def create_workstream(
self,
*,
name: str = "",
model: str = "",
auto_approve: bool = False,
) -> CreateWorkstreamResponse:
body: dict[str, Any] = {}
if name:
body["name"] = name
if model:
body["model"] = model
if auto_approve:
body["auto_approve"] = True
return await self._request(
"POST",
"/v1/api/workstreams/new",
json_body=body,
response_model=CreateWorkstreamResponse,
)
async def close_workstream(self, ws_id: str) -> StatusResponse:
return await self._request(
"POST",
"/v1/api/workstreams/close",
json_body={"ws_id": ws_id},
response_model=StatusResponse,
)
# -- chat interaction ----------------------------------------------------
async def send(self, message: str, ws_id: str) -> SendResponse:
return await self._request(
"POST",
"/v1/api/send",
json_body={"message": message, "ws_id": ws_id},
response_model=SendResponse,
)
async def approve(
self,
*,
ws_id: str,
approved: bool = True,
feedback: str | None = None,
always: bool = False,
) -> StatusResponse:
body: dict[str, Any] = {"ws_id": ws_id, "approved": approved}
if feedback is not None:
body["feedback"] = feedback
if always:
body["always"] = True
return await self._request(
"POST", "/v1/api/approve", json_body=body, response_model=StatusResponse
)
async def plan_feedback(self, *, ws_id: str, feedback: str = "") -> StatusResponse:
return await self._request(
"POST",
"/v1/api/plan",
json_body={"ws_id": ws_id, "feedback": feedback},
response_model=StatusResponse,
)
async def command(self, *, ws_id: str, command: str) -> StatusResponse:
return await self._request(
"POST",
"/v1/api/command",
json_body={"ws_id": ws_id, "command": command},
response_model=StatusResponse,
)
# -- streaming -----------------------------------------------------------
async def stream_events(self, ws_id: str) -> AsyncIterator[ServerEvent]:
"""Iterate over per-workstream SSE events."""
async for data in self._stream_sse("/v1/api/events", params={"ws_id": ws_id}):
yield ServerEvent.from_dict(data)
async def stream_global_events(self) -> AsyncIterator[ServerEvent]:
"""Iterate over global SSE events."""
async for data in self._stream_sse("/v1/api/events/global"):
yield ServerEvent.from_dict(data)
# -- high-level convenience ----------------------------------------------
async def send_and_wait(
self,
message: str,
ws_id: str,
*,
timeout: float = 600,
on_event: Callable[[ServerEvent], None] | None = None,
) -> TurnResult:
"""Send a message and wait for the turn to complete via SSE.
Opens the per-workstream SSE stream *before* sending the message
to avoid missing early events, then accumulates content / reasoning /
tool results / errors until a ``ws_state`` event with
``state="idle"`` arrives, or the timeout expires.
"""
result = TurnResult(ws_id=ws_id)
async def _consume() -> None:
async for data in self._stream_sse("/v1/api/events", params={"ws_id": ws_id}):
event = ServerEvent.from_dict(data)
if on_event:
on_event(event)
if isinstance(event, ContentEvent):
result.content_parts.append(event.text)
elif isinstance(event, ReasoningEvent):
result.reasoning_parts.append(event.text)
elif isinstance(event, ToolResultEvent):
result.tool_results.append((event.name, event.output))
elif isinstance(event, ErrorEvent):
result.errors.append(event.message)
elif isinstance(event, WsStateEvent) and event.state == "idle":
return
# Start SSE consumer BEFORE sending to avoid missing early events
consume_task = asyncio.create_task(_consume())
await asyncio.sleep(0) # yield to let SSE connection establish
try:
send_resp = await self.send(message, ws_id)
if send_resp.status == "busy":
result.errors.append("Workstream is busy")
return result
await asyncio.wait_for(consume_task, timeout=timeout)
except TimeoutError:
result.timed_out = True
finally:
# Always clean up the SSE consumer to prevent connection leaks
if not consume_task.done():
consume_task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await consume_task
return result
# -- sessions ------------------------------------------------------------
async def list_sessions(self) -> ListSessionsResponse:
return await self._request("GET", "/v1/api/sessions", response_model=ListSessionsResponse)
# -- auth ----------------------------------------------------------------
async def login(self, token: str) -> AuthLoginResponse:
return await self._request(
"POST",
"/v1/api/auth/login",
json_body={"token": token},
response_model=AuthLoginResponse,
)
async def logout(self) -> StatusResponse:
return await self._request("POST", "/v1/api/auth/logout", response_model=StatusResponse)
# -- health --------------------------------------------------------------
async def health(self) -> HealthResponse:
return await self._request("GET", "/health", response_model=HealthResponse)
class TurnstoneServer:
"""Synchronous client for the turnstone server API.
Wraps :class:`AsyncTurnstoneServer` via a background event loop.
Usage::
with TurnstoneServer("http://localhost:8080", token="tok_xxx") as client:
ws = client.create_workstream(name="Analysis")
result = client.send_and_wait("Hello", ws.ws_id)
print(result.content)
"""
def __init__(
self,
base_url: str = "http://localhost:8080",
token: str = "",
timeout: float = 30.0,
) -> None:
self._runner = _SyncRunner()
self._async = AsyncTurnstoneServer(base_url=base_url, token=token, timeout=timeout)
# -- workstream management -----------------------------------------------
def list_workstreams(self) -> ListWorkstreamsResponse:
return self._runner.run(self._async.list_workstreams())
def dashboard(self) -> DashboardResponse:
return self._runner.run(self._async.dashboard())
def create_workstream(
self,
*,
name: str = "",
model: str = "",
auto_approve: bool = False,
) -> CreateWorkstreamResponse:
return self._runner.run(
self._async.create_workstream(name=name, model=model, auto_approve=auto_approve)
)
def close_workstream(self, ws_id: str) -> StatusResponse:
return self._runner.run(self._async.close_workstream(ws_id))
# -- chat interaction ----------------------------------------------------
def send(self, message: str, ws_id: str) -> SendResponse:
return self._runner.run(self._async.send(message, ws_id))
def approve(
self,
*,
ws_id: str,
approved: bool = True,
feedback: str | None = None,
always: bool = False,
) -> StatusResponse:
return self._runner.run(
self._async.approve(ws_id=ws_id, approved=approved, feedback=feedback, always=always)
)
def plan_feedback(self, *, ws_id: str, feedback: str = "") -> StatusResponse:
return self._runner.run(self._async.plan_feedback(ws_id=ws_id, feedback=feedback))
def command(self, *, ws_id: str, command: str) -> StatusResponse:
return self._runner.run(self._async.command(ws_id=ws_id, command=command))
# -- streaming -----------------------------------------------------------
def stream_events(self, ws_id: str) -> Iterator[ServerEvent]:
return self._runner.run_iter(self._async.stream_events(ws_id))
def stream_global_events(self) -> Iterator[ServerEvent]:
return self._runner.run_iter(self._async.stream_global_events())
# -- high-level convenience ----------------------------------------------
def send_and_wait(
self,
message: str,
ws_id: str,
*,
timeout: float = 600,
on_event: Callable[[ServerEvent], None] | None = None,
) -> TurnResult:
return self._runner.run(
self._async.send_and_wait(message, ws_id, timeout=timeout, on_event=on_event)
)
# -- sessions ------------------------------------------------------------
def list_sessions(self) -> ListSessionsResponse:
return self._runner.run(self._async.list_sessions())
# -- auth ----------------------------------------------------------------
def login(self, token: str) -> AuthLoginResponse:
return self._runner.run(self._async.login(token))
def logout(self) -> StatusResponse:
return self._runner.run(self._async.logout())
# -- health --------------------------------------------------------------
def health(self) -> HealthResponse:
return self._runner.run(self._async.health())
# -- lifecycle -----------------------------------------------------------
def close(self) -> None:
self._runner.run(self._async.aclose())
self._runner.close()
def __enter__(self) -> TurnstoneServer:
return self
def __exit__(self, *exc: object) -> None:
self.close()