mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-23 12:24:46 -06:00
Patrick Buckley
553d73109b
* feat(coordinator): phase 5 — harness-test polish + wait_for_workstream + judge fix
Closes the bug list surfaced by the 2026-04-17 coordinator harness test
plus the post-phase-4 wait_for_workstream ask, and folds in three
adjacent cleanups that landed in the same window. Tightens defense-in-
depth on the model-invoked mutating ops, fixes the LLM judge silent
no-op, kills the inspect-poll token burn, and rounds out a handful of
observability / docstring / spec gaps.
The session-factory pre-resolve at console/session_factory.py and
server.py was rewriting `judge.model` from an alias (e.g. `judge-mini`)
to the resolved underlying id (e.g. `gpt-5-mini`). IntentJudge then
checked `model_registry.has_alias(config.model)`, found nothing, and
fell back to the SESSION's provider/client with that bare model id —
silent `llm_fallback / "did not return a verdict"` whenever the
coordinator and judge alias resolved to different providers.
Pass the alias through unchanged; IntentJudge's existing alias-
resolution path picks up the matching client + provider. Validate
the alias exists so an obvious typo still surfaces, but don't replace
the model field.
Regression: `test_alias_uses_registry_provider_not_session_provider`
constructs an alias whose provider differs from the session's and
asserts the judge picks up the alias's provider/client/model;
`test_coordinator_tool_call_returns_llm_verdict_not_fallback` asserts
the verdict tier is `llm` (not `llm_fallback`) on the happy path.
New `cancel_workstream` tool (approval required, primary_key=ws_id) —
cancels in-flight generation, unblocks any pending approval / plan,
moves the child to idle, leaves the row in storage so a fresh
send_to_workstream lands cleanly. Re-uses the existing
`/v1/api/route/cancel` route + `route.cancel` audit namespace; no
new server endpoint.
`CoordinatorClient.cancel/close_workstream/delete/send` now enforce
a tenant guard inline (`_is_own_subtree`) — only the coordinator
itself or one of its own children is targetable. Foreign ids return
the same 404-shape inspect/wait_for_workstream use, so the model
can't distinguish foreign from missing (no existence oracle).
Defense-in-depth — the upstream node enforcement is the perimeter,
this is the second line.
`list_workstreams` advertised `state="deleted"` and an
`include_closed=true` that surfaced deleted rows. Hard-deletes
cascade the workstream + conversation rows out of storage, so
deleted is unreachable in normal operation. Doc-only fix; the
synthetic-test path that registers `state="deleted"` rows still
works (terminal-state filter still excludes them via
`_terminal_states = {"closed", "deleted"}` in list_children).
Documented that the 120s service-registry heartbeat window means a
node returned by list_nodes can drop out before a follow-up
`spawn_workstream(target_node=…)` lands — the spawn fails with "No
available node for routing" rather than falling back. Two-line
clarification on each tool. No code change (a code fallback is a
bigger discussion deferred to 1.6).
`close_workstream` accepts `reason`; the upstream server handler now
persists it to `workstream_config.close_reason` (capped at 512 BYTES,
sliced on UTF-8 not code points so a CJK / emoji-heavy payload can't
4× the documented budget). `CoordinatorClient.inspect()` reads it
and surfaces as `close_reason` in the result dict — only for
terminal-state children (closed/error/deleted) so the live-child hot
path doesn't pay a per-inspect DB round-trip.
Tests: server-side persistence covers success / no-reason /
length-cap / non-string / storage-failure / multi-byte-utf8 paths;
client-side surface covers terminal vs. live workstreams.
For idle children whose node-dashboard live counter is 0 (the live
block only surfaces in-flight token counters), fall back to
`SUM(prompt_tokens + completion_tokens)` from `usage_events` so the
inspect output reflects cumulative spend.
New `storage.sum_workstream_tokens(ws_id) -> int` on the protocol +
both backends. The fallback is folded INTO `_fetch_cluster_live` so
the merged live block (with persisted total applied) is what gets
cached — back-to-back inspects of an idle child amortize through
the existing 2s LRU cache instead of each firing a fresh aggregation.
`CoordinatorClient.list_skills()` now projects `allowed_tools` per
skill — capped at 20 with a `+N more` sentinel so a skill that
whitelists a wide MCP surface doesn't bloat the per-row payload.
Reads the existing `prompt_templates.allowed_tools` column; no
storage change. Coordinators no longer have to guess what tools a
skill brings.
`route_create` now sets `routing_strategy: "hash_ring" | "target_node"
| "resume"` on the spawn response so the coordinator's spawn
response (and the `spawn_workstream` tool output) carries why a
given node was chosen. 3 lines + 3 covering tests in
test_console_routing_proxy.py.
New coordinator tool `wait_for_workstream(ws_ids, timeout=60,
mode='any'|'all')` that absorbs the wait into a single tool call —
the model sees one call + one result regardless of how long the
children take. Kills the busy-poll inspect loop that burned 20+
turns on a 3-child fan-out.
Storage-poll loop with batched primitives —
`get_workstreams_batch` + `sum_workstream_tokens_batch` issue exactly
two storage calls per tick regardless of N. At the cap (32 ws_ids /
600s / 0.5s tick) that's ~2400 round-trips for a full wait, down
from ~38k under the naive per-id shape.
Validation single-source-of-truth: the client owns mode whitelist,
ws_ids dedup + cap, timeout coerce + clamp. The session preparer
is a thin pass-through that builds the header + dispatches; bad
input surfaces at exec time as a tool error via `result.get("error")`.
Tenant-isolation collapse: missing-row and cross-tenant cases both
return `state="denied"` so wait can't be used as an existence oracle
(matches the 404-mask contract `inspect` uses).
Prompt-side: tools_coordinator.md adds a `wait_for_workstream`
pattern + an explicit "PREFER wait_for_workstream OVER a loop of
inspect_workstream" line in the workflow-shape section.
Replaces the quote-bracketed substring LIKE/ILIKE pattern with proper
JSON-array containment. The previous shape effectively did
`LOWER(tags) LIKE '%"<lower-tag>"%'`, which broke for tag values
containing `"` (the JSON encoder escapes it to `\"` and the literal-
substring search misses), `\` (encoded as `\\`), or non-ASCII
characters that the encoder rendered as `\uXXXX`. Also exposed a
small spoofing surface — `tags=["foo\","bar"]` would have matched a
query for `bar`. Real-world tag values are alphanumeric+dash today
so it hadn't fired in production, but the fix is small.
- SQLite: `EXISTS (SELECT 1 FROM json_each(prompt_templates.tags)
WHERE lower(value) = lower(:tag))` (JSON1 extension; SQLite 3.38+).
- PostgreSQL: `EXISTS (SELECT 1 FROM jsonb_array_elements_text(
prompt_templates.tags::jsonb) AS jat(elem) WHERE lower(jat.elem) =
lower(:tag))`.
Three new tests prove the substring pattern was broken for
quoted / backslash / unicode tag values; the existing case-fold +
wildcard tests continue to pin the contract.
Phase 1 added the coordinator workstream API; phase 2 added only
`/open` to the OpenAPI catalog and missed every other coordinator
endpoint plus phase 3's `/children`, `/tasks`, and the
`/cluster/ws/{ws_id}/detail` aggregator. SDK consumers + operators
browsing `/docs` couldn't discover the surface. Doc-only addition:
12 endpoints + 9 new Pydantic models, all under the `Coordinator`
OpenAPI tag so /docs groups them together.
Sidebar re-fetches `GET /tasks` on every `task_list` `tool_result`
SSE event. A model that runs `add → list` (or any back-to-back
mutation pair) double-fetches the same envelope. Coalesced into
one fetch per 150ms window via a new `loadTasksDebounced` wrapper;
direct UI actions (refresh button, page load) keep calling
`loadTasks` directly so user clicks aren't delayed.
- `ruff check turnstone tests` — clean
- `mypy turnstone` — clean (157 source files)
- `pytest -m "not live"` — 4284 passed, 3 deselected (was 4226 on
main; +58 new tests across coordinator client, tools, judge,
storage, console routing proxy, server close-handler,
storage_skills_filtered, OpenAPI catalog, server close-reason
persistence)
- New tools added: 2 (cancel_workstream, wait_for_workstream) —
TOOLS count 28 → 30; coordinator subset 9 → 11; auto_approve adds
wait_for_workstream; primary_key adds cancel_workstream
- New OpenAPI endpoints: 12 (every phase-1/2/3 coordinator route +
the cluster-inspect aggregator)
- New storage protocol methods: 3 (sum_workstream_tokens,
sum_workstream_tokens_batch, get_workstreams_batch)
All phase 1 / 2 / 3 / 4 invariants preserved: COORDINATOR_TOOLS /
INTERACTIVE_TOOLS disjoint; coordinator sessions have no MCP surface;
list-style tools return {items, truncated}; route-proxy emits
route.<action> audit on 2xx; 404-mask on ownership failures; tenant
filters pushed into SQL; per-coordinator JWT carries scope context.
* fix(coordinator): address Copilot review on PR #378
Three valid Copilot findings on the wait_for_workstream surface:
1. ``wait_for_workstream.json`` description claimed the tool returns a
top-level mapping ``ws_id -> {state, tokens, updated}`` plus
elapsed/complete/mode at the same level, but the actual shape is
``{results: {ws_id: {...}}, elapsed, complete, mode}``. Description
now matches the implementation. Also adds ``deleted`` to the
advertised terminal-state list (it's in ``_WAIT_REAL_TERMINAL_STATES``;
the doc and runtime now agree).
2. ``CoordinatorClient.wait_for_workstream`` docstring listed
``idle / error / closed`` as the real terminal set but the constant
includes ``deleted``. Same fix — list ``deleted`` with a parenthetical
noting it's unreachable in normal operation (hard-delete cascades the
row).
3. Storage protocol docstring math: ``sum_workstream_tokens_batch``
claimed "from ~38k to ~1200" round-trips per wait at the cap, but
``wait_for_workstream`` issues TWO storage calls per tick
(``get_workstreams_batch`` + this one), so 1200 ticks × 2 = ~2400.
Updated to "~2400" with the math spelled out.
Also a clean rebase onto today's main (PR #377 — the rebalancer node_id
snapshot doc — landed since phase 5's last push). Single conflict in
``inspect_workstream.json`` resolved by keeping both notes (rebalancer
node_id binding semantics + the new ``close_reason`` surface from phase
5); ``spawn_workstream.json`` auto-merged.
The github-code-quality bot also flagged three items on
``_protocol.py`` asking to replace ``...`` with ``pass`` in Protocol
method bodies. Refuted: ``...`` is the canonical PEP 544 idiom for
Protocol method bodies and the rest of the file uses it consistently.
The bot's lint rule misfires for ``Protocol`` classes.
Verification:
- ``ruff check turnstone tests`` clean
- ``mypy turnstone`` clean (158 source files)
- ``pytest -m "not live"`` — 4308 passed, 3 deselected (no test count
change; pure doc/comment edits)
175 lines
6.3 KiB
Python
175 lines
6.3 KiB
Python
"""Tests for OpenAPI spec generation."""
|
|
|
|
import json
|
|
|
|
|
|
class TestServerSpec:
|
|
"""Validate the generated server OpenAPI spec."""
|
|
|
|
def test_valid_openapi_version(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
assert spec["openapi"] == "3.1.0"
|
|
|
|
def test_has_info(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
assert "title" in spec["info"]
|
|
assert "version" in spec["info"]
|
|
|
|
def test_has_all_api_endpoints(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
paths = set(spec["paths"].keys())
|
|
expected = {
|
|
"/v1/api/workstreams",
|
|
"/v1/api/dashboard",
|
|
"/v1/api/workstreams/saved",
|
|
"/v1/api/send",
|
|
"/v1/api/approve",
|
|
"/v1/api/plan",
|
|
"/v1/api/command",
|
|
"/v1/api/events",
|
|
"/v1/api/events/global",
|
|
"/v1/api/workstreams/new",
|
|
"/v1/api/workstreams/close",
|
|
"/v1/api/auth/login",
|
|
"/v1/api/auth/logout",
|
|
"/health",
|
|
}
|
|
assert expected.issubset(paths), f"Missing: {expected - paths}"
|
|
|
|
def test_schemas_not_empty(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
assert len(spec["components"]["schemas"]) > 0
|
|
|
|
def test_json_serializable(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
result = json.dumps(spec)
|
|
assert len(result) > 100
|
|
|
|
def test_send_endpoint_has_request_body(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
send = spec["paths"]["/v1/api/send"]["post"]
|
|
assert "requestBody" in send
|
|
assert "application/json" in send["requestBody"]["content"]
|
|
|
|
def test_health_endpoint_not_versioned(self):
|
|
from turnstone.api.server_spec import build_server_spec
|
|
|
|
spec = build_server_spec()
|
|
assert "/health" in spec["paths"]
|
|
assert "/v1/health" not in spec["paths"]
|
|
|
|
|
|
class TestConsoleSpec:
|
|
"""Validate the generated console OpenAPI spec."""
|
|
|
|
def test_valid_openapi_version(self):
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
assert spec["openapi"] == "3.1.0"
|
|
|
|
def test_has_cluster_endpoints(self):
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
paths = set(spec["paths"].keys())
|
|
expected = {
|
|
"/v1/api/cluster/overview",
|
|
"/v1/api/cluster/nodes",
|
|
"/v1/api/cluster/workstreams",
|
|
"/v1/api/cluster/node/{node_id}",
|
|
"/v1/api/cluster/workstreams/new",
|
|
"/v1/api/cluster/events",
|
|
}
|
|
assert expected.issubset(paths), f"Missing: {expected - paths}"
|
|
|
|
def test_json_serializable(self):
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
result = json.dumps(spec)
|
|
assert len(result) > 100
|
|
|
|
def test_nodes_endpoint_has_query_params(self):
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
nodes = spec["paths"]["/v1/api/cluster/nodes"]["get"]
|
|
assert "parameters" in nodes
|
|
param_names = [p["name"] for p in nodes["parameters"]]
|
|
assert "sort" in param_names
|
|
assert "limit" in param_names
|
|
|
|
def test_has_coordinator_endpoints(self):
|
|
"""Phase 1-3 coordinator routes must appear in the OpenAPI catalog —
|
|
the spec was missing every coordinator endpoint except ``/open``,
|
|
so SDK consumers and operators couldn't discover the surface
|
|
from /docs. Pin the full set so a future regression that drops
|
|
one fails loudly."""
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
paths = set(spec["paths"].keys())
|
|
expected = {
|
|
"/v1/api/coordinator/new",
|
|
"/v1/api/coordinator",
|
|
"/v1/api/coordinator/{ws_id}",
|
|
"/v1/api/coordinator/{ws_id}/open",
|
|
"/v1/api/coordinator/{ws_id}/send",
|
|
"/v1/api/coordinator/{ws_id}/approve",
|
|
"/v1/api/coordinator/{ws_id}/cancel",
|
|
"/v1/api/coordinator/{ws_id}/close",
|
|
"/v1/api/coordinator/{ws_id}/events",
|
|
"/v1/api/coordinator/{ws_id}/history",
|
|
"/v1/api/coordinator/{ws_id}/children",
|
|
"/v1/api/coordinator/{ws_id}/tasks",
|
|
"/v1/api/cluster/ws/{ws_id}/detail",
|
|
}
|
|
assert expected.issubset(paths), f"Missing: {expected - paths}"
|
|
|
|
def test_coordinator_create_has_request_body_and_201(self):
|
|
"""Coordinator create returns 201 (not 200) and accepts a body."""
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
op = spec["paths"]["/v1/api/coordinator/new"]["post"]
|
|
assert "requestBody" in op
|
|
assert "application/json" in op["requestBody"]["content"]
|
|
# Pin the 201 success code.
|
|
assert "201" in op["responses"]
|
|
|
|
def test_coordinator_history_has_limit_query_param(self):
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
op = spec["paths"]["/v1/api/coordinator/{ws_id}/history"]["get"]
|
|
param_names = [p["name"] for p in op.get("parameters", [])]
|
|
assert "ws_id" in param_names # auto-added from path
|
|
assert "limit" in param_names
|
|
|
|
def test_coordinator_endpoints_share_tag(self):
|
|
"""All coordinator endpoints (including the cluster-inspect one)
|
|
live under the same OpenAPI tag so /docs groups them together."""
|
|
from turnstone.api.console_spec import build_console_spec
|
|
|
|
spec = build_console_spec()
|
|
coord_paths = [p for p in spec["paths"] if "/coordinator" in p]
|
|
coord_paths.append("/v1/api/cluster/ws/{ws_id}/detail")
|
|
for path in coord_paths:
|
|
for op in spec["paths"][path].values():
|
|
assert "Coordinator" in op.get("tags", []), (
|
|
f"{path} missing Coordinator tag (tags={op.get('tags')})"
|
|
)
|