Files
turnstone/tests/test_coordinator_tools.py
T
Patrick Buckley 9826ea15c5 feat(coordinator): phase 7 — governance + skill metadata + cross-cutt… (#383)
* feat(coordinator): phase 7 — governance + skill metadata + cross-cutting invariants

Combines three stacked sub-PRs into a single coordinator phase-7
shipment against the phase-7 plan doc.  The sub-PR structure (0 / A /
B) preserved on individual branches for reviewer drill-down; this
branch is the one reviewers should merge.

## Sub-PR 0 — service-auth boundary invariants

Shared helpers and contracts that lock the console ↔ node service-auth
boundary so later authz surfaces use them by construction.

- ``_effective_user_filter(request)`` in both ``turnstone.console.server``
  and ``turnstone.server`` with a shared ``DENY_EMPTY_SUB`` sentinel
  on ``turnstone.core.auth``.  Three-way return — admin/service
  bypass, scoped caller uid, or fail-closed sentinel on blank sub.
  Four callsite migrations (``_coordinator_rows``,
  ``coordinator_children``, ``coordinator_metrics``,
  ``cluster_ws_live_bulk``).

- ``StorageBackend`` class docstring codifies the tenancy contract
  (every list/count/aggregate method must accept ``user_id: str |
  None = None`` and push ``WHERE user_id = :user_id`` into SQL) and
  the ``_mapping`` row-access contract.  New
  ``turnstone.testing.row_contract`` ships ``assert_row_like()``.

- ``_verify_collector_service_scope`` probes an upstream node at boot
  with ``expected_node_id=_scope-probe_``; a 409 proves the scope
  gate was passed, a 403/401 sets ``collector_scope_error`` and
  causes ``cluster_snapshot`` / ``cluster_events_sse`` to return 503
  with a remediation hint.  Probe URL allowlist rejects non-http(s)
  schemes and 169.254.0.0/16 hosts.

- 4xx log-level floor on ``_NodeDashboardCache.get``,
  ``_fetch_live_block``, and ``_proxy_sse`` — dotted-hierarchy
  prefixes with bounded body previews.  ``_bounded_body_preview`` and
  ``_bounded_stream_preview`` strip control chars.

## Sub-PR A — coordinator governance core

Mid-session governance surface for coordinator workstreams.

- **Trusted-session mode.**  New ``coordinator.trust.send``
  permission (migration 042).  ``ChatSession.set_trust_send`` /
  ``revoke_tools`` methods with a ``_governance_lock``.  ``POST
  /v1/api/coordinator/{ws_id}/trust {send: bool}`` double-gated on
  ``admin.coordinator`` AND ``coordinator.trust.send`` with
  ``allow_service_bypass=False`` so service tokens can't escalate.
  ``_prepare_send_to_workstream`` auto-approves sends whose target is
  in the coordinator's own subtree; foreign ws_ids still require
  approval.  ``_is_own_subtree`` checks both ``parent_ws_id`` AND
  ``user_id`` to defend against cross-tenant row corruption.

- **Audit-layer credential redaction.**  ``record_audit`` walks
  ``detail`` (dicts, lists, tuples, sets, frozensets; keys too)
  and routes every string through ``redact_credentials`` + a C0
  control-char scrub.  New kw-only ``raw_detail=True`` opt-out.
  ``_has_any_string`` fast-path.  Audit action registry extended
  with the four new governance sub-prefixes.

- **Mid-session revocation + cascading stop.**  ``POST
  /v1/api/coordinator/{ws_id}/restrict {revoke: [...]}`` caps 256
  entries / 128 chars; ``_prepare_tool`` short-circuits with a
  tool-error.  ``POST /v1/api/coordinator/{ws_id}/stop_cascade``
  cancels the coord's in-flight generation then dispatches
  ``cancel_workstream`` for every direct child in parallel via
  ``asyncio.gather`` bounded by ``Semaphore(16)``.  Per-child
  outcomes split into ``cancelled`` / ``failed`` / ``skipped``
  (404 = already-gone rather than dispatch-broken).  Both endpoints
  apply ``allow_service_bypass=False`` on the admin gate.

- **Shared plumbing.**  ``_resolve_coord_session`` helper collapses
  the handler prelude three endpoints shared.  ``_emit_coord_audit``
  wraps ``record_audit`` in a dedicated ``ThreadPoolExecutor``
  (``app.state.audit_executor``) so audit bursts don't starve cancel
  dispatches.  ``_require_json_object`` guards body parsing so non-
  object JSON returns 400 instead of 500.

## Sub-PR B — skill metadata governance

- **Description validator (migration 043).**  ``prompt_templates``
  rows now require a non-empty ``description``.  Existing empty rows
  get backfilled with a ``"Skill: <name>"`` placeholder on upgrade.
  The installer (``admin_skill_discover``) and MCP prompt sync both
  synthesise a placeholder when the upstream description is blank
  so non-admin write paths satisfy the invariant.

- **Skill kind classifier (migration 044).**  New
  ``prompt_templates.kind`` column (``interactive`` / ``coordinator``
  / ``any``; defaults to ``any``).  New
  ``turnstone.core.skill_kind.SkillKind`` StrEnum is the single
  source of truth; Pydantic schemas type ``kind`` as ``SkillKind``
  (OpenAPI advertises the enum) and the handler validator catches
  the ValueError.  ``list_skills_filtered`` gains a
  ``kinds: list[str] | None = None`` SQL filter.
  ``CoordinatorClient.list_skills`` defaults to
  ``kinds=["coordinator", "any"]`` so interactive-only skills are
  hidden from the orchestrator.

- **``scan_status`` → ``risk_level`` rename (migration 045).**
  Lossless column rename to align with ``IntentVerdict.risk_level``
  terminology.  Swept storage (both backends + schema + protocol),
  handlers, API schemas, tool JSON, generated OpenAPI specs,
  TypeScript SDK types, frontend (``governance.js``), tests, and
  English prose in ``docs/judge.md`` + ``docs/tools.md``.  The
  user-facing on-load warning now reads ``has risk level:
  {risk_tier}``.  Tool JSON's ``risk_level`` enum corrected to the
  scanner's actual taxonomy (``safe / low / medium / high /
  critical``; was the never-shipped ``clean / flagged / unscanned /
  pending``).  Historical migration 021 left untouched.

## Migrations

042 (``coordinator.trust.send`` perm — PR A)
043 (description backfill — PR B)
044 (``kind`` column add — PR B)
045 (``scan_status`` → ``risk_level`` rename — PR B)

All four use position-anchored permission strings / host-side
parse-filter-rejoin on downgrade where SQL ``REPLACE`` could
corrupt prefix-overlapping values.

## Verification

- ``ruff check turnstone tests`` clean.
- ``mypy turnstone`` clean on 165 source files.
- ``pytest -m "not live"``: 4431 passed (+85 over the phase-6
  baseline).  Includes +32 tests in ``tests/test_service_auth_boundary.py``
  and +38 in ``tests/test_coordinator_governance.py``; shared fixtures
  extracted to ``tests/_coord_test_helpers.py``.
- Generated OpenAPI JSON (``sdk/typescript/openapi-{console,server}.json``)
  regenerated via ``sdk/typescript/scripts/generate-types.py``; zero
  ``scan_status`` occurrences remaining outside the historical
  migration 021 and the rename migration 045.

## Security reviews

Both reviews flagged by the phase-7 plan (items 1 + 5, plus 0a's
refuse-to-serve gate) ran through the multi-stage ``/review``
pipeline twice per sub-PR; all confirmed findings landed in-branch.

* fixup(phase-7): CI lint + PR #383 review fixups

Addresses the lint CI failure (ruff format) plus 12 findings from the
two automated PR reviewers.

Copilot:
- ``_sqlite.list_installed_skill_urls`` / ``_postgresql.list_installed_skill_urls``
  used positional row indexing (``r[0]``/``r[1]``/``r[2]``) while this
  same PR's ``StorageBackend`` class docstring forbids it.  Switched
  both to ``r._mapping["..."]`` access.
- ``list_skills.json`` previously advertised ``risk_level=""`` as a
  filter for unscanned skills, but the implementation treats empty
  strings as "no filter".  Clarified the tool description to say
  omit the filter entirely to include unscanned rows, and added an
  explicit ``enum`` on the parameter restricting it to the scanner
  tiers.  ``_prepare_list_skills`` keeps the ``strip() or None``
  normalisation — unscanned filtering now has an unambiguous contract.
- ``test_storage_skills_filtered.test_risk_level_filter`` used the
  legacy ``clean`` / ``flagged`` values from the pre-rename column.
  Rewritten with the scanner's actual taxonomy (``safe`` / ``high``).

github-code-quality (CodeQL):
- ``test_deny_sentinel_is_singleton`` previously asserted
  ``cs.DENY_EMPTY_SUB is cs.DENY_EMPTY_SUB`` — an identical-expression
  comparison.  Rewritten as two separate ``from ... import ... as`` aliases
  (``FIRST_READ`` / ``SECOND_READ``) so the identity check is between
  distinct bindings.
- ``test_restrict_empty_revoke_is_noop_but_audits`` unpacked ``state``
  without using it.  Renamed to ``_state``.
- Mixed import styles in ``test_service_auth_boundary.py`` — the
  file previously used both ``import turnstone.console.server as cs``
  and ``from turnstone.console.server import ...`` for the same
  module (same story for ``turnstone.core.auth`` and
  ``turnstone.server``).  Consolidated to the ``from X import Y`` style
  used elsewhere in the file; the ``_fetch_live_block`` test now
  patches via pytest's ``monkeypatch`` fixture instead of a manual
  rebind through a module alias.

CI:
- ``ruff format`` reformatted one line in
  ``tests/test_coordinator_endpoints.py``.

Verification: ruff check + mypy clean (166 files); 4459 non-live
pytest pass.

* fix(tests): swap asyncio marker for anyio in service-auth boundary tests

PR #383 CI caught that the 13 ``@pytest.mark.asyncio`` decorators I
added in ``test_service_auth_boundary.py`` are an off-convention
choice — the rest of the repo uses ``@pytest.mark.anyio`` (148 sites
vs my 13).  The CI environment pulls in ``anyio`` but not
``pytest-asyncio``, so every async test in this one file was failing
with "async def functions are not natively supported".  It passed
locally by accident — my dev venv happens to have pytest-asyncio
installed ambiently.

Swapped all 13 marker sites to ``@pytest.mark.anyio``.  No functional
change; the tests run under the same default asyncio backend anyio
provides.

Verification: ruff + mypy clean (166 files); 4459 non-live pytest
pass.
2026-04-18 10:20:19 -07:00

991 lines
36 KiB
Python

"""Tests for the coordinator prepare/exec dispatch on ChatSession.
We construct a ChatSession with ``kind="coordinator"`` and a mocked
``CoordinatorClient``, then drive ``_prepare_tool`` directly with tool
call dicts matching the shape the provider layer produces. This is a
unit-level test of the dispatch plumbing — end-to-end flows land in
Phase D's test_coordinator_end_to_end.
"""
from __future__ import annotations
import json
from typing import Any
from unittest.mock import ANY, MagicMock
import pytest
from turnstone.core.session import ChatSession
from turnstone.prompts import ClientType
class _StubUI:
"""Minimal SessionUI that records signals without doing anything with them."""
def __init__(self) -> None:
self._user_id = "user-1"
self.infos: list[str] = []
self.errors: list[str] = []
self.tool_results: list[tuple[str, str, str, bool]] = []
def on_info(self, msg: str) -> None:
self.infos.append(msg)
def on_error(self, msg: str) -> None:
self.errors.append(msg)
def on_tool_result(self, call_id: str, name: str, output: str, is_error: bool = False) -> None:
self.tool_results.append((call_id, name, output, is_error))
# Other SessionUI methods — only stubs, not exercised here.
def on_turn_start(self) -> None:
pass
def on_turn_end(self) -> None:
pass
def on_stream_start(self) -> None:
pass
def on_stream_end(self) -> None:
pass
def on_message_delta(self, delta: str) -> None:
pass
def on_reasoning_delta(self, delta: str) -> None:
pass
def on_tool_call(self, call_id: str, name: str, header: str, preview: str) -> None:
pass
def on_completion(self, content: str) -> None:
pass
def on_attention(self, header: str, preview: str = "") -> None:
pass
def wait_for_approval(
self,
call_id: str,
name: str,
header: str,
preview: str,
*,
label: str = "",
) -> tuple[bool, str | None]:
return True, None
@pytest.fixture
def coord_session(monkeypatch):
"""Build a coordinator ChatSession with a mocked CoordinatorClient.
Patches heavyweight init steps (_load_skills, _init_system_messages,
_save_config) to keep the test fast + isolated from the storage
registry.
"""
monkeypatch.setattr(ChatSession, "_load_skills", lambda self: None)
monkeypatch.setattr(ChatSession, "_init_system_messages", lambda self: None)
monkeypatch.setattr(ChatSession, "_save_config", lambda self: None)
ui = _StubUI()
coord_client = MagicMock()
sess = ChatSession(
client=MagicMock(),
model="gpt-test",
ui=ui, # type: ignore[arg-type]
instructions=None,
temperature=0.0,
max_tokens=1024,
tool_timeout=30,
context_window=16384,
ws_id="coord-1",
user_id="user-1",
client_type=ClientType.WEB,
kind="coordinator",
coord_client=coord_client,
)
return sess, coord_client, ui
# ---------------------------------------------------------------------------
# Tool set shape
# ---------------------------------------------------------------------------
def test_coordinator_session_uses_coordinator_tools(coord_session):
sess, _coord, _ui = coord_session
names = {t["function"]["name"] for t in sess._tools}
assert names == {
"spawn_workstream",
"inspect_workstream",
"send_to_workstream",
"close_workstream",
"cancel_workstream",
"delete_workstream",
"list_workstreams",
"list_nodes",
"list_skills",
"task_list",
"wait_for_workstream",
}
# Sub-agent tool sets are zeroed on coordinator sessions.
assert sess._task_tools == []
assert sess._agent_tools == []
# ---------------------------------------------------------------------------
# Helper: build a ChatCompletion-style tool_call dict
# ---------------------------------------------------------------------------
def _tc(name: str, args: dict[str, Any], call_id: str = "call-1") -> dict[str, Any]:
return {
"id": call_id,
"type": "function",
"function": {"name": name, "arguments": json.dumps(args)},
}
# ---------------------------------------------------------------------------
# spawn_workstream
# ---------------------------------------------------------------------------
def test_spawn_prepare_allows_empty_initial_message(coord_session):
"""Empty initial_message creates an idle child — matches tool JSON advertisement."""
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("spawn_workstream", {"initial_message": ""}))
assert "error" not in item
assert item["needs_approval"] is True
assert "idle workstream" in item["header"]
assert item["initial_message"] == ""
def test_spawn_prepare_needs_approval(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc("spawn_workstream", {"initial_message": "do a thing", "skill": "s"})
)
assert item["needs_approval"] is True
assert item["execute"].__func__ is ChatSession._exec_spawn_workstream
assert item["skill"] == "s"
def test_spawn_exec_calls_client_and_returns_summary(coord_session):
sess, coord, _ui = coord_session
coord.spawn.return_value = {
"ws_id": "child-7",
"name": "c",
"node_id": "node-1",
"status": 200,
}
item = sess._prepare_tool(_tc("spawn_workstream", {"initial_message": "hi"}))
call_id, output = sess._exec_spawn_workstream(item)
coord.spawn.assert_called_once()
_, kwargs = coord.spawn.call_args
assert kwargs["parent_ws_id"] == "coord-1"
assert kwargs["user_id"] == "user-1"
assert kwargs["initial_message"] == "hi"
assert call_id == "call-1"
assert "child-7" in output
def test_spawn_exec_surfaces_client_error(coord_session):
sess, coord, ui = coord_session
coord.spawn.return_value = {"error": "upstream unreachable", "status": 502}
item = sess._prepare_tool(_tc("spawn_workstream", {"initial_message": "hi"}))
_call_id, output = sess._exec_spawn_workstream(item)
assert "upstream unreachable" in output
# UI got an error result
assert ui.tool_results[-1][3] is True # is_error
# ---------------------------------------------------------------------------
# inspect_workstream
# ---------------------------------------------------------------------------
def test_inspect_prepare_is_auto_approved(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("inspect_workstream", {"ws_id": "child-x", "message_limit": 5}))
assert item["needs_approval"] is False
assert item["execute"].__func__ is ChatSession._exec_inspect_workstream
assert item["message_limit"] == 5
def test_inspect_prepare_requires_ws_id(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("inspect_workstream", {}))
assert "error" in item
def test_inspect_prepare_clamps_message_limit(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("inspect_workstream", {"ws_id": "x", "message_limit": 10000}))
assert item["message_limit"] == 200 # clamped
def test_inspect_exec_dispatches_to_client(coord_session):
sess, coord, _ui = coord_session
coord.inspect.return_value = {
"ws_id": "child-x",
"state": "idle",
"messages": [],
"verdicts": [],
}
item = sess._prepare_tool(_tc("inspect_workstream", {"ws_id": "child-x"}))
_call_id, output = sess._exec_inspect_workstream(item)
coord.inspect.assert_called_once_with(
"child-x", message_limit=20, include_provider_content=False
)
assert "child-x" in output
# ---------------------------------------------------------------------------
# send_to_workstream
# ---------------------------------------------------------------------------
def test_send_prepare_needs_approval(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("send_to_workstream", {"ws_id": "x", "message": "hello"}))
assert item["needs_approval"] is True
def test_send_prepare_rejects_empty_message(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("send_to_workstream", {"ws_id": "x", "message": ""}))
assert "error" in item
def test_send_exec_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.send.return_value = {"status": 200}
item = sess._prepare_tool(_tc("send_to_workstream", {"ws_id": "x", "message": "hi"}))
_call_id, output = sess._exec_send_to_workstream(item)
coord.send.assert_called_once_with("x", "hi")
assert "x" in output
# ---------------------------------------------------------------------------
# close_workstream
# ---------------------------------------------------------------------------
def test_close_prepare_needs_approval(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("close_workstream", {"ws_id": "x"}))
assert item["needs_approval"] is True
def test_close_exec_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.close_workstream.return_value = {"status": 200}
item = sess._prepare_tool(_tc("close_workstream", {"ws_id": "x"}))
_call_id, output = sess._exec_close_workstream(item)
# Default (no reason) — kwargs carry empty reason through the call.
coord.close_workstream.assert_called_once_with("x", reason="")
parsed = json.loads(output)
assert parsed["closed"] is True
assert "reason" not in parsed # omitted when empty
def test_close_exec_forwards_reason(coord_session):
"""reason is wired through both CoordinatorClient.close_workstream
and the tool-result payload so the coordinator's message stream
records why the close happened."""
sess, coord, _ui = coord_session
coord.close_workstream.return_value = {"status": 200}
item = sess._prepare_tool(_tc("close_workstream", {"ws_id": "x", "reason": "task done"}))
_call_id, output = sess._exec_close_workstream(item)
coord.close_workstream.assert_called_once_with("x", reason="task done")
parsed = json.loads(output)
assert parsed["reason"] == "task done"
# ---------------------------------------------------------------------------
# delete_workstream
# ---------------------------------------------------------------------------
# ---------------------------------------------------------------------------
# cancel_workstream
# ---------------------------------------------------------------------------
def test_cancel_prepare_needs_approval(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("cancel_workstream", {"ws_id": "x"}))
assert item["needs_approval"] is True
assert "cancel_workstream" in item["header"]
def test_cancel_prepare_requires_ws_id(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("cancel_workstream", {}))
assert "error" in item
def test_cancel_exec_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.cancel.return_value = {"status": 200}
item = sess._prepare_tool(_tc("cancel_workstream", {"ws_id": "x"}))
_call_id, output = sess._exec_cancel_workstream(item)
coord.cancel.assert_called_once_with("x")
parsed = json.loads(output)
assert parsed["cancelled"] is True
assert parsed["ws_id"] == "x"
def test_cancel_exec_surfaces_client_error(coord_session):
sess, coord, ui = coord_session
coord.cancel.return_value = {"error": "ws not found", "status": 404}
item = sess._prepare_tool(_tc("cancel_workstream", {"ws_id": "x"}))
_call_id, output = sess._exec_cancel_workstream(item)
assert "ws not found" in output
assert ui.tool_results[-1][3] is True # is_error
# ---------------------------------------------------------------------------
# wait_for_workstream
# ---------------------------------------------------------------------------
def test_wait_prepare_is_auto_approved(coord_session):
"""Prepare is a thin pass-through — auto-approved, no validation;
the client owns ws_ids dedup / cap / timeout clamp / mode whitelist."""
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc(
"wait_for_workstream",
{"ws_ids": ["a", "b"], "timeout": 5, "mode": "all"},
)
)
assert item["needs_approval"] is False
# Raw args pass through verbatim — the client validates / dedups.
assert item["ws_ids"] == ["a", "b"]
assert item["mode"] == "all"
assert item["timeout"] == 5
def test_wait_exec_surfaces_client_validation_error(coord_session):
"""Bad input is rejected by the client and surfaced as a tool error
via the result.get('error') branch in exec — single source of truth
for validation."""
sess, coord, ui = coord_session
coord.wait_for_workstream.return_value = {
"error": "ws_ids must contain at least one valid id",
"results": {},
"complete": False,
"elapsed": 0.0,
"mode": "any",
}
item = sess._prepare_tool(_tc("wait_for_workstream", {"ws_ids": []}))
_call_id, output = sess._exec_wait_for_workstream(item)
assert "must contain at least one" in output
assert ui.tool_results[-1][3] is True # is_error
def test_wait_exec_dispatches_raw_args_to_client(coord_session):
sess, coord, _ui = coord_session
coord.wait_for_workstream.return_value = {
"results": {"a": {"state": "idle", "tokens": 0}},
"complete": True,
"elapsed": 0.5,
"mode": "any",
}
item = sess._prepare_tool(_tc("wait_for_workstream", {"ws_ids": ["a"], "timeout": 30}))
_call_id, output = sess._exec_wait_for_workstream(item)
# Args forwarded raw (timeout int, default mode="any") — client
# handles the float coerce + clamp. ``since`` + ``progress_callback``
# are optional observability kwargs added for the wait dashboard /
# diff-hint items (#14, #18); match them via ANY so this assertion
# stays focused on the raw dispatch.
coord.wait_for_workstream.assert_called_once_with(
["a"], timeout=30, mode="any", since=None, progress_callback=ANY
)
parsed = json.loads(output)
assert parsed["complete"] is True
assert parsed["mode"] == "any"
def test_wait_exec_default_timeout_when_omitted(coord_session):
"""timeout=None (omitted) becomes 60.0 in exec so the client receives
a numeric value — explicit ``timeout=0`` is preserved (one-shot
poll) by passing the raw arg straight through."""
sess, coord, _ui = coord_session
coord.wait_for_workstream.return_value = {
"results": {"a": {"state": "idle", "tokens": 0}},
"complete": True,
"elapsed": 0.0,
"mode": "any",
}
item = sess._prepare_tool(_tc("wait_for_workstream", {"ws_ids": ["a"]}))
sess._exec_wait_for_workstream(item)
coord.wait_for_workstream.assert_called_once_with(
["a"], timeout=60.0, mode="any", since=None, progress_callback=ANY
)
def test_wait_exec_preserves_explicit_zero_timeout(coord_session):
"""Explicit ``timeout=0`` reaches the client untouched."""
sess, coord, _ui = coord_session
coord.wait_for_workstream.return_value = {
"results": {"a": {"state": "idle", "tokens": 0}},
"complete": True,
"elapsed": 0.0,
"mode": "any",
}
item = sess._prepare_tool(_tc("wait_for_workstream", {"ws_ids": ["a"], "timeout": 0}))
sess._exec_wait_for_workstream(item)
coord.wait_for_workstream.assert_called_once_with(
["a"], timeout=0, mode="any", since=None, progress_callback=ANY
)
def test_delete_prepare_needs_approval(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("delete_workstream", {"ws_id": "x"}))
assert item["needs_approval"] is True
assert "irreversible" in item["header"].lower()
def test_delete_exec_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.delete.return_value = {"status": 200}
item = sess._prepare_tool(_tc("delete_workstream", {"ws_id": "x"}))
_call_id, output = sess._exec_delete_workstream(item)
coord.delete.assert_called_once_with("x")
parsed = json.loads(output)
assert parsed["deleted"] is True
# ---------------------------------------------------------------------------
# list_workstreams
# ---------------------------------------------------------------------------
def test_list_prepare_is_auto_approved(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("list_workstreams", {}))
assert item["needs_approval"] is False
def test_list_prepare_defaults_parent_to_self_ws(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("list_workstreams", {}))
assert item["parent_ws_id"] == "coord-1"
def test_list_prepare_accepts_explicit_parent(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc("list_workstreams", {"parent_ws_id": "other-coord", "state": "idle"})
)
assert item["parent_ws_id"] == "other-coord"
assert item["state"] == "idle"
def test_list_exec_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.list_children.return_value = {
"children": [
{"ws_id": "a", "state": "idle"},
{"ws_id": "b", "state": "running"},
],
"truncated": False,
}
item = sess._prepare_tool(_tc("list_workstreams", {}))
_call_id, output = sess._exec_list_workstreams(item)
coord.list_children.assert_called_once()
parsed = json.loads(output)
assert parsed["parent_ws_id"] == "coord-1"
assert len(parsed["children"]) == 2
assert parsed["truncated"] is False
def test_list_exec_surfaces_truncated_sentinel(coord_session):
sess, coord, _ui = coord_session
coord.list_children.return_value = {
"children": [{"ws_id": "a", "state": "idle"}],
"truncated": True,
}
item = sess._prepare_tool(_tc("list_workstreams", {}))
_call_id, output = sess._exec_list_workstreams(item)
parsed = json.loads(output)
assert parsed["truncated"] is True
# ---------------------------------------------------------------------------
# Defensive guard: missing coord_client
# ---------------------------------------------------------------------------
def test_prepare_fails_cleanly_when_coord_client_missing(monkeypatch):
"""If somehow a coordinator-kind session is built without a coord_client,
prepare methods return an error item rather than NPE."""
monkeypatch.setattr(ChatSession, "_load_skills", lambda self: None)
monkeypatch.setattr(ChatSession, "_init_system_messages", lambda self: None)
monkeypatch.setattr(ChatSession, "_save_config", lambda self: None)
ui = _StubUI()
sess = ChatSession(
client=MagicMock(),
model="m",
ui=ui, # type: ignore[arg-type]
instructions=None,
temperature=0.0,
max_tokens=1024,
tool_timeout=30,
context_window=16384,
ws_id="coord-1",
kind="coordinator",
coord_client=None,
)
for tool, args in (
("spawn_workstream", {"initial_message": "hi"}),
("inspect_workstream", {"ws_id": "x"}),
("send_to_workstream", {"ws_id": "x", "message": "m"}),
("close_workstream", {"ws_id": "x"}),
("delete_workstream", {"ws_id": "x"}),
("list_workstreams", {}),
("list_nodes", {}),
("list_skills", {}),
("task_list", {"action": "list"}),
):
item = sess._prepare_tool(_tc(tool, args))
assert "error" in item, f"{tool} did not error on missing coord_client"
# ---------------------------------------------------------------------------
# list_nodes
# ---------------------------------------------------------------------------
def test_list_nodes_prepare_is_auto_approved(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("list_nodes", {}))
assert item["needs_approval"] is False
assert item["filters"] == {}
assert item["limit"] == 100
def test_list_nodes_prepare_accepts_filters(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc("list_nodes", {"filters": {"arch": "x86_64", "capability": "gpu"}})
)
assert item["filters"] == {"arch": "x86_64", "capability": "gpu"}
def test_list_nodes_prepare_drops_invalid_filter_types(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc(
"list_nodes",
{"filters": {"arch": "x86_64", "bad": {"nested": "dict"}, "": "empty-key"}},
)
)
# Nested dict values + empty keys are filtered out; string + primitive kept.
assert item["filters"] == {"arch": "x86_64"}
def test_list_nodes_prepare_clamps_limit(coord_session):
sess, _coord, _ui = coord_session
over = sess._prepare_tool(_tc("list_nodes", {"limit": 9999}))
assert over["limit"] == 500
# limit == 0 falls back to the default (100), not 1 — consistent with
# the other coordinator list tools' ``int(args.get("limit") or 100)``.
zero = sess._prepare_tool(_tc("list_nodes", {"limit": 0}))
assert zero["limit"] == 100
neg = sess._prepare_tool(_tc("list_nodes", {"limit": -5}))
assert neg["limit"] == 1 # negative values clamp to 1
def test_list_nodes_exec_dispatches_to_client(coord_session):
sess, coord, ui = coord_session
coord.list_nodes.return_value = {
"nodes": [{"node_id": "n1", "metadata": {"arch": {"value": "x86_64", "source": "auto"}}}],
"truncated": False,
}
item = sess._prepare_tool(_tc("list_nodes", {"filters": {"arch": "x86_64"}}))
call_id, output = sess._exec_list_nodes(item)
assert call_id == "call-1"
parsed = json.loads(output)
assert parsed["nodes"][0]["node_id"] == "n1"
assert parsed["truncated"] is False
coord.list_nodes.assert_called_once_with(
filters={"arch": "x86_64"},
limit=100,
include_network_detail=False,
include_inactive=False,
)
def test_list_nodes_exec_surfaces_truncated_sentinel(coord_session):
sess, coord, ui = coord_session
coord.list_nodes.return_value = {"nodes": [], "truncated": True}
item = sess._prepare_tool(_tc("list_nodes", {}))
_, _ = sess._exec_list_nodes(item)
# Summary reported to UI carries the "truncated" hint.
assert any("truncated" in r[2] for r in ui.tool_results)
# ---------------------------------------------------------------------------
# list_skills
# ---------------------------------------------------------------------------
def test_list_skills_prepare_is_auto_approved(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("list_skills", {}))
assert item["needs_approval"] is False
assert item["category"] is None
assert item["tag"] is None
assert item["risk_level"] is None
assert item["enabled_only"] is False
assert item["limit"] == 100
def test_list_skills_prepare_accepts_filters(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc(
"list_skills",
{"category": "ops", "tag": "gpu", "risk_level": "clean", "enabled_only": True},
)
)
assert item["category"] == "ops"
assert item["tag"] == "gpu"
assert item["risk_level"] == "clean"
assert item["enabled_only"] is True
def test_list_skills_prepare_tolerates_non_string_filters(coord_session):
"""A malformed model call with non-string filter values must NOT
raise AttributeError during ``.strip()`` — the prepare path should
coerce non-strings to ``None`` and proceed."""
sess, _coord, _ui = coord_session
item = sess._prepare_tool(
_tc(
"list_skills",
{"category": 42, "tag": ["not", "a", "string"], "risk_level": {"bad": 1}},
)
)
assert "error" not in item
assert item["category"] is None
assert item["tag"] is None
assert item["risk_level"] is None
def test_list_skills_prepare_parses_enabled_only_string_forms(coord_session):
"""``bool("false")`` is True (non-empty string). The prepare path
must interpret common string forms the way the model would expect."""
sess, _coord, _ui = coord_session
for raw, expected in (
("true", True),
("True", True),
("1", True),
("false", False),
("False", False),
("0", False),
("", False),
(True, True),
(False, False),
):
item = sess._prepare_tool(_tc("list_skills", {"enabled_only": raw}))
assert item.get("enabled_only") is expected, (
f"enabled_only={raw!r}{item.get('enabled_only')!r}, expected {expected!r}"
)
def test_list_skills_exec_dispatches_to_client(coord_session):
sess, coord, ui = coord_session
coord.list_skills.return_value = {
"skills": [{"name": "alpha", "tags": ["gpu"]}],
"truncated": False,
}
item = sess._prepare_tool(_tc("list_skills", {"category": "ops", "tag": "gpu"}))
call_id, output = sess._exec_list_skills(item)
assert call_id == "call-1"
parsed = json.loads(output)
assert parsed["skills"][0]["name"] == "alpha"
coord.list_skills.assert_called_once_with(
category="ops",
tag="gpu",
risk_level=None,
enabled_only=False,
limit=100,
)
def test_list_skills_exec_surfaces_truncated_sentinel(coord_session):
sess, coord, ui = coord_session
coord.list_skills.return_value = {"skills": [], "truncated": True}
item = sess._prepare_tool(_tc("list_skills", {}))
_, _ = sess._exec_list_skills(item)
assert any("truncated" in r[2] for r in ui.tool_results)
# ---------------------------------------------------------------------------
# task_list
# ---------------------------------------------------------------------------
def test_task_list_list_is_auto_approved(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "list"}))
assert item["needs_approval"] is False
assert item["action"] == "list"
def test_task_list_bare_string_fallback_uses_action_primary_key(coord_session):
"""A model that emits an unquoted ``list`` as the arguments blob
lands on the ``primary_key=action`` fallback and recovers. Before
the fix primary_key was ``title`` so the fallback produced
``{"title": "list"}`` and hit the required-action rejection."""
sess, _coord, _ui = coord_session
call = {
"id": "c1",
"type": "function",
"function": {"name": "task_list", "arguments": "list"},
}
item = sess._prepare_tool(call)
assert "error" not in item
assert item["action"] == "list"
def test_task_list_mutating_actions_need_approval(coord_session):
sess, _coord, _ui = coord_session
add_item = sess._prepare_tool(_tc("task_list", {"action": "add", "title": "plan"}))
assert add_item["needs_approval"] is True
update_item = sess._prepare_tool(
_tc("task_list", {"action": "update", "task_id": "tsk_1", "status": "done"})
)
assert update_item["needs_approval"] is True
remove_item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": "tsk_1"}))
assert remove_item["needs_approval"] is True
reorder_item = sess._prepare_tool(
_tc("task_list", {"action": "reorder", "task_ids": ["tsk_1"]})
)
assert reorder_item["needs_approval"] is True
def test_task_list_unknown_action_errors(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "wat"}))
assert "error" in item
def test_task_list_non_string_action_errors_cleanly(coord_session):
"""A malformed ``action=42`` must NOT raise AttributeError during
``.strip().lower()`` — coerce to the empty string and fall through
to the enum-check error."""
sess, _coord, _ui = coord_session
for bad_action in (42, None, ["list"], {"a": 1}, True):
item = sess._prepare_tool(_tc("task_list", {"action": bad_action}))
assert "error" in item, f"action={bad_action!r} did not produce a clean error"
def test_task_list_add_rejects_non_string_title_and_status(coord_session):
"""Add branch: ``title=42`` / ``status=0`` must NOT raise
AttributeError during ``.strip()``; produce a clean error item."""
sess, _coord, _ui = coord_session
for bad in ({"action": "add", "title": 42}, {"action": "add", "title": "ok", "status": 0}):
item = sess._prepare_tool(_tc("task_list", bad))
assert "error" in item, f"args={bad!r} did not produce a clean error"
def test_task_list_remove_non_string_task_id_errors_cleanly(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": 42}))
assert "error" in item
def test_task_list_add_requires_title(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "add", "title": ""}))
assert "error" in item
def test_task_list_update_requires_task_id(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "update", "status": "done"}))
assert "error" in item
def test_task_list_update_rejects_non_string_field_values(coord_session):
"""Preview must not diverge from execute: reject non-string field
values at prepare time rather than silently coercing to None."""
sess, _coord, _ui = coord_session
for field in ("title", "status", "child_ws_id"):
item = sess._prepare_tool(
_tc("task_list", {"action": "update", "task_id": "t1", field: 42})
)
assert "error" in item, f"update with non-string {field} should error"
def test_task_list_update_requires_at_least_one_field(coord_session):
"""update with only task_id is a no-op — reject to save an approval prompt."""
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "update", "task_id": "t1"}))
assert "error" in item
def test_task_list_remove_requires_task_id(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "remove"}))
assert "error" in item
def test_task_list_reorder_requires_list_of_strings(coord_session):
sess, _coord, _ui = coord_session
item = sess._prepare_tool(_tc("task_list", {"action": "reorder", "task_ids": [1, 2]}))
assert "error" in item
def test_task_list_exec_list_returns_tasks(coord_session):
sess, coord, _ui = coord_session
coord.task_list_get.return_value = {
"version": 1,
"tasks": [{"id": "tsk_1", "title": "do", "status": "pending"}],
}
item = sess._prepare_tool(_tc("task_list", {"action": "list"}))
_, output = sess._exec_task_list(item)
parsed = json.loads(output)
assert parsed["tasks"][0]["id"] == "tsk_1"
assert parsed["truncated"] is False
def test_task_list_exec_list_page_caps_at_200(coord_session):
sess, coord, _ui = coord_session
coord.task_list_get.return_value = {
"version": 1,
"tasks": [{"id": f"tsk_{i}", "title": "x", "status": "pending"} for i in range(250)],
}
item = sess._prepare_tool(_tc("task_list", {"action": "list"}))
_, output = sess._exec_task_list(item)
parsed = json.loads(output)
assert len(parsed["tasks"]) == 200
assert parsed["truncated"] is True
def test_task_list_exec_add_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.task_list_add.return_value = {"id": "tsk_new", "title": "plan"}
item = sess._prepare_tool(
_tc("task_list", {"action": "add", "title": "plan", "status": "pending"})
)
_, _ = sess._exec_task_list(item)
coord.task_list_add.assert_called_once_with(
sess._ws_id, title="plan", status="pending", child_ws_id=""
)
def test_task_list_exec_reorder_surfaces_permutation_error(coord_session):
sess, coord, _ui = coord_session
coord.task_list_reorder.return_value = {"error": "task_ids must be a permutation..."}
item = sess._prepare_tool(_tc("task_list", {"action": "reorder", "task_ids": ["wrong"]}))
_, output = sess._exec_task_list(item)
parsed = json.loads(output)
assert "error" in parsed
def test_task_list_exec_remove_passes_client_dict_through(coord_session):
"""The client returns a dict; exec must pass it through without
synthesising a generic 'not found' message that would mask corrupt-
envelope errors from the LLM."""
sess, coord, _ui = coord_session
coord.task_list_remove.return_value = {
"error": "task_list envelope is corrupt on disk; refusing to overwrite."
}
item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": "x"}))
_, output = sess._exec_task_list(item)
parsed = json.loads(output)
assert "corrupt" in parsed["error"]
def test_task_list_exec_remove_success_dispatches(coord_session):
sess, coord, _ui = coord_session
coord.task_list_remove.return_value = {"ok": True, "task_id": "tsk_1"}
item = sess._prepare_tool(_tc("task_list", {"action": "remove", "task_id": "tsk_1"}))
_, output = sess._exec_task_list(item)
parsed = json.loads(output)
assert parsed.get("ok") is True
# ---------------------------------------------------------------------------
# Smoke-test regressions — empty-arg tool calls, metadata stripping,
# provider-content trimming
# ---------------------------------------------------------------------------
def test_prepare_tool_empty_arguments_string_parses_as_object(coord_session):
"""Some providers emit an empty string when a tool is invoked with
no arguments (all params optional). The empty string must be
treated as ``{}`` rather than dropped into the malformed-JSON
error branch — otherwise zero-arg coordinator tool calls fail."""
sess, coord, _ui = coord_session
coord.list_nodes.return_value = {"nodes": [], "truncated": False}
tc = {
"id": "call-empty",
"type": "function",
"function": {"name": "list_nodes", "arguments": ""},
}
item = sess._prepare_tool(tc)
# No error field, prepared for list_nodes exec.
assert "error" not in item
assert item["func_name"] == "list_nodes"
def test_list_nodes_strips_interfaces_by_default(coord_session):
"""Default ``list_nodes`` output omits the auto-populated
``interfaces`` key — it leaks internal RFC 1918 addresses and the
model never uses it for routing decisions."""
sess, coord, _ui = coord_session
item = sess._prepare_tool(_tc("list_nodes", {}))
coord.list_nodes.assert_not_called() # prepare doesn't fire the client yet
assert item["include_network_detail"] is False
sess._exec_list_nodes(item)
coord.list_nodes.assert_called_once()
kwargs = coord.list_nodes.call_args.kwargs
assert kwargs.get("include_network_detail") is False
def test_list_nodes_include_network_detail_opt_in(coord_session):
"""Opt-in flag flips include_network_detail=True through to the client."""
sess, coord, _ui = coord_session
item = sess._prepare_tool(_tc("list_nodes", {"include_network_detail": True}))
assert item["include_network_detail"] is True
sess._exec_list_nodes(item)
kwargs = coord.list_nodes.call_args.kwargs
assert kwargs.get("include_network_detail") is True
def test_inspect_workstream_default_trims_provider_content(coord_session):
"""Default ``inspect_workstream`` threads
``include_provider_content=False`` through to the client so the
``_provider_content`` / ``provider_blocks`` duplicates don't bloat
the response."""
sess, coord, _ui = coord_session
item = sess._prepare_tool(_tc("inspect_workstream", {"ws_id": "abc123"}))
assert item["include_provider_content"] is False
sess._exec_inspect_workstream(item)
kwargs = coord.inspect.call_args.kwargs
assert kwargs.get("include_provider_content") is False
def test_inspect_workstream_include_provider_content_opt_in(coord_session):
sess, coord, _ui = coord_session
item = sess._prepare_tool(
_tc(
"inspect_workstream",
{"ws_id": "abc123", "include_provider_content": True},
)
)
assert item["include_provider_content"] is True
sess._exec_inspect_workstream(item)
kwargs = coord.inspect.call_args.kwargs
assert kwargs.get("include_provider_content") is True