Files
turnstone/tests/test_workstream_kind.py
Patrick Buckley 9a30530d41 feat(coord): surface child errors, isolate tool exceptions, add memory tool (#443)
* feat(coord): surface child errors, isolate tool exceptions, add memory tool

Closes four coordinator gaps identified during operator triage:

1. Child workstream errors now surface in inspect/wait. Worker-thread
   exception text is sanitized (URL userinfo masked, sk-/Bearer/ghp_/
   github_pat_/AKIA tokens redacted, capped at 1024 chars) and persisted
   to workstream_config.last_error before _emit_state("error") fires, so
   coord polling never sees state=error with a missing cause. The row
   is cleared on recovery transitions (idle/running) so a once-leaked
   exception body doesn't outlive the failure. inspect_workstream and
   wait_for_workstream return last_error for state=error rows; the
   wait surface prefers it over the assistant-tail walk.

2. Tool exceptions now return as tool_results with sibling-aware
   guidance. ChatSession._safe_prepare_tool wraps every per-call
   _prepare_tool invocation; a buggy preparer becomes an error item
   for that call only — sibling parallel tool_calls keep going,
   never orphaning the assistant message's tool_calls block.
   run_one's runtime exception path includes the exception class
   and a short note that other tool calls in the batch completed
   independently so the model can recover.

3. Memory tool exposed to coordinator with a coord-only scope.
   memory.json gains coordinator: true + interactive: true + per-kind
   kind_variants. Coord sessions see scope enum ["coordinator"] and
   an orchestration-flavored description; IC sessions see ["global",
   "workstream", "user"] and the existing flavor. Coord-scope rows
   are private to the coordinator session (children cannot read or
   write them), closing the cross-session prompt-injection lane that
   an adversarially-steered child would otherwise have. Coord
   visibility is also restricted to coord-scope only — coords no
   longer see global / workstream / user memories that belong to the
   user's interactive sessions.

4. Per-call exception isolation in tool batches. _safe_prepare_tool
   was previously the implicit shield; now it's an explicit method
   with documented invariants. KeyboardInterrupt / GenerationCancelled
   re-raise so the cooperative cancel path still works.

Other notable changes:
- LAST_ERROR_CONFIG_KEY + persist_last_error / clear_last_error /
  load_last_error / sanitize_error_text moved to turnstone.core.memory
  (the storage facade hub) — readers in coordinator_client.py import
  the constant.
- Memory scope tuples extracted to module constants
  _VALID_MEMORY_SCOPES and _IMPLICIT_SCOPE_WALK; seven inline
  duplicates collapsed.
- tools.py grows _apply_kind_variant for the per-kind tool surface;
  tools without kind_variants pass through unchanged (no spurious
  deep-copies).
- Session adds _coordinator_scope_id, _default_memory_scope,
  _implicit_scope_walk, and _record_fatal_error chokepoints so the
  worker-thread fatal path is one site rather than three.
- Removed duplicate on_error / on_state_change emits from
  session_routes.py and coordinator_adapter.py — session.send()'s
  _record_fatal_error owns the sequence now.

Tests: 4742 pass (no live), +30 net since the baseline. Ruff + mypy
clean on every modified production file.

* fix(coord): redact secrets in tool error paths via output_guard

Copilot review flagged two paths where ``str(exc)`` flowed back into
the model-facing tool_result without going through the credential-
redaction the new fatal-error path applies:

  - ``ChatSession._safe_prepare_tool``: a preparer-side exception
    becomes an error item whose ``error`` field embedded the raw
    exception text.
  - ``ChatSession._execute_tools.run_one``: a runtime tool exception
    became an ``Error executing X: <e>`` tool_result, again with
    the raw exception text.

Both now route through ``sanitize_error_text`` (sanitised log line +
sanitised tool_result), and ``sanitize_error_text`` itself was
refactored to delegate to ``output_guard.redact_credentials`` instead
of carrying its own parallel regex catalog — the audit log + post-tool
guard already use that pattern set, so the credential definition
stays in one place.

Also extended ``_RE_CONNECTION_STRING`` in ``output_guard`` to cover
``http(s)://user:pass@host`` so a misconfigured ``OPENAI_BASE_URL``
that lands in an httpx ``ConnectError.__str__`` is redacted by every
caller of ``redact_credentials`` (audit details, close-reason
persistence, last_error, the two tool error paths).  The
host (useful for triage) survives; only the password is replaced
with the standard ``[REDACTED:password]`` marker.

Tests: full suite (4745 pass), ruff + mypy clean.  Two new tests pin
the redaction behaviour in both tool error paths so a future refactor
can't drift back to leaking ``str(exc)`` verbatim.
2026-04-28 12:19:34 -07:00

410 lines
15 KiB
Python

"""Tests for Phase A schema additions: ``kind`` + ``parent_ws_id`` on workstreams.
Covers:
- ``register_workstream`` persists the two new columns.
- ``get_workstream`` returns the full row including the new fields.
- ``list_workstreams`` filters on ``kind`` and ``parent_ws_id`` correctly.
- ``parent_ws_id`` empty-string normalization at the storage edge.
- Defaults remain ``"interactive"`` / ``NULL`` when not specified.
- ``Workstream`` dataclass exposes ``kind`` / ``parent_ws_id`` / ``user_id``
with safe defaults.
"""
from __future__ import annotations
from turnstone.core.workstream import Workstream
# ``storage`` comes from tests/conftest.py — backend-parametrized fixture that
# respects the ``--storage-backend`` flag so the same assertions run against
# both SQLite (default) and PostgreSQL (CI), closing the q-3 drift risk that
# sqlite↔postgres register/list/normalize semantics could diverge silently.
# ---------------------------------------------------------------------------
# register_workstream / get_workstream
# ---------------------------------------------------------------------------
def test_register_defaults_to_interactive_no_parent(storage):
storage.register_workstream("ws-a")
row = storage.get_workstream("ws-a")
assert row is not None
assert row["kind"] == "interactive"
assert row["parent_ws_id"] is None
def test_register_coordinator_kind_and_parent(storage):
storage.register_workstream("ws-coord", node_id="console", user_id="user-1", kind="coordinator")
storage.register_workstream(
"ws-child",
node_id="node-a",
user_id="user-1",
kind="interactive",
parent_ws_id="ws-coord",
)
coord = storage.get_workstream("ws-coord")
child = storage.get_workstream("ws-child")
assert coord is not None and child is not None
assert coord["kind"] == "coordinator"
assert coord["parent_ws_id"] is None
assert coord["user_id"] == "user-1"
assert child["kind"] == "interactive"
assert child["parent_ws_id"] == "ws-coord"
assert child["user_id"] == "user-1"
def test_register_normalizes_empty_parent_to_null(storage):
"""Empty-string parent_ws_id must be persisted as NULL so
``WHERE parent_ws_id IS NULL`` filters stay correct."""
storage.register_workstream("ws-a", parent_ws_id="")
row = storage.get_workstream("ws-a")
assert row is not None
assert row["parent_ws_id"] is None
def test_register_rejects_unknown_kind(storage):
"""Storage edge validates kind via WorkstreamKind(kind).value —
SDK / restore / direct callers can't silently corrupt the NOT NULL column
with typos or unknown values the way pre-PR #1 they could."""
import pytest as _pytest
with _pytest.raises(ValueError):
storage.register_workstream("ws-bogus", kind="interative") # typo
# Row was never inserted — no side effects on failure.
assert storage.get_workstream("ws-bogus") is None
def test_delete_workstream_nulls_child_parent_ws_id(storage):
"""Deleting a coordinator must null-out its children's parent_ws_id
so list_workstreams(parent_ws_id=<deleted>) doesn't keep returning
ghost-parented rows."""
storage.register_workstream("coord", kind="coordinator", user_id="user-1")
storage.register_workstream(
"child-a", kind="interactive", parent_ws_id="coord", user_id="user-1"
)
storage.register_workstream(
"child-b", kind="interactive", parent_ws_id="coord", user_id="user-1"
)
assert storage.delete_workstream("coord") is True
# Children still exist but with NULL parent_ws_id.
for cid in ("child-a", "child-b"):
row = storage.get_workstream(cid)
assert row is not None, f"{cid} should survive parent deletion"
assert row["parent_ws_id"] is None, f"{cid} still points at ghost coord"
# No rows match the deleted coord's parent filter.
assert storage.list_workstreams(parent_ws_id="coord") == []
def test_list_workstreams_filter_by_user_id(storage):
"""The user_id kwarg pushes tenant scoping into SQL so callers
can't forget to filter client-side."""
storage.register_workstream("ws-a", user_id="user-1")
storage.register_workstream("ws-b", user_id="user-1")
storage.register_workstream("ws-c", user_id="user-2")
storage.register_workstream("ws-ownerless") # no user_id
mine = storage.list_workstreams(user_id="user-1")
theirs = storage.list_workstreams(user_id="user-2")
ownerless = storage.list_workstreams(user_id="")
unfiltered = storage.list_workstreams()
assert {r[0] for r in mine} == {"ws-a", "ws-b"}
assert {r[0] for r in theirs} == {"ws-c"}
# Empty string is a real filter value (matches rows with stored "" owner).
# Rows with NULL owner are distinct and not matched.
assert "ws-ownerless" not in {r[0] for r in ownerless}
# No filter → all rows.
assert {r[0] for r in unfiltered} == {"ws-a", "ws-b", "ws-c", "ws-ownerless"}
def test_get_workstream_missing_returns_none(storage):
assert storage.get_workstream("nonexistent") is None
def test_get_workstream_includes_all_fields(storage):
storage.register_workstream(
"ws-full",
node_id="n1",
user_id="u1",
alias="alias-1",
title="Title 1",
name="name-1",
state="idle",
skill_id="skill-x",
skill_version=3,
kind="interactive",
parent_ws_id="parent-x",
)
row = storage.get_workstream("ws-full")
assert row is not None
for expected in (
"ws_id",
"node_id",
"user_id",
"alias",
"title",
"name",
"state",
"skill_id",
"skill_version",
"kind",
"parent_ws_id",
"created",
"updated",
):
assert expected in row
assert row["skill_version"] == 3
assert row["parent_ws_id"] == "parent-x"
# ---------------------------------------------------------------------------
# list_workstreams filter params
# ---------------------------------------------------------------------------
def test_list_workstreams_no_filters_unchanged(storage):
storage.register_workstream("ws-a")
storage.register_workstream("ws-b")
rows = storage.list_workstreams()
assert len(rows) == 2
def test_list_workstreams_filter_by_kind(storage):
storage.register_workstream("ws-int-1")
storage.register_workstream("ws-int-2")
storage.register_workstream("ws-coord", kind="coordinator")
interactive = storage.list_workstreams(kind="interactive")
coord = storage.list_workstreams(kind="coordinator")
assert {r[0] for r in interactive} == {"ws-int-1", "ws-int-2"}
assert {r[0] for r in coord} == {"ws-coord"}
def test_list_workstreams_filter_by_parent(storage):
storage.register_workstream("ws-coord", kind="coordinator")
storage.register_workstream("child-1", parent_ws_id="ws-coord")
storage.register_workstream("child-2", parent_ws_id="ws-coord")
storage.register_workstream("other-1") # no parent
children = storage.list_workstreams(parent_ws_id="ws-coord")
assert {r[0] for r in children} == {"child-1", "child-2"}
def test_list_workstreams_combined_filters(storage):
storage.register_workstream("ws-coord", kind="coordinator")
storage.register_workstream("child-1", parent_ws_id="ws-coord")
storage.register_workstream("child-coord", parent_ws_id="ws-coord", kind="coordinator")
# Children of ws-coord that are themselves interactive.
rows = storage.list_workstreams(parent_ws_id="ws-coord", kind="interactive")
assert {r[0] for r in rows} == {"child-1"}
def test_list_workstreams_node_id_filter_still_works(storage):
"""The existing ``node_id`` filter keeps working after the signature change."""
storage.register_workstream("ws-a", node_id="node-1")
storage.register_workstream("ws-b", node_id="node-2")
rows = storage.list_workstreams(node_id="node-1")
assert {r[0] for r in rows} == {"ws-a"}
def test_list_workstreams_returns_kind_and_parent_columns(storage):
storage.register_workstream("ws-coord", kind="coordinator")
storage.register_workstream("child-1", parent_ws_id="ws-coord")
rows = storage.list_workstreams()
by_id = {r[0]: r for r in rows}
# Columns: ws_id, node_id, name, state, created, updated, kind, parent_ws_id
coord_row = by_id["ws-coord"]
child_row = by_id["child-1"]
assert coord_row[6] == "coordinator"
assert coord_row[7] is None
assert child_row[6] == "interactive"
assert child_row[7] == "ws-coord"
# ---------------------------------------------------------------------------
# Workstream dataclass field additions
# ---------------------------------------------------------------------------
def test_workstream_dataclass_defaults():
ws = Workstream()
assert ws.user_id == ""
assert ws.kind == "interactive"
assert ws.parent_ws_id is None
def test_workstream_dataclass_accepts_coordinator_kind():
ws = Workstream(kind="coordinator", user_id="user-1")
assert ws.kind == "coordinator"
assert ws.user_id == "user-1"
assert ws.parent_ws_id is None
def test_workstream_dataclass_accepts_parent():
ws = Workstream(parent_ws_id="parent-x")
assert ws.parent_ws_id == "parent-x"
# ---------------------------------------------------------------------------
# Tool-namespace isolation between kinds
# ---------------------------------------------------------------------------
def test_interactive_and_coordinator_tool_sets_overlap_only_on_dual_kind():
"""Interactive ∩ coordinator must be exactly the explicitly dual-kind tools.
Regression guard for the latent threshold bug where coordinator-only
tools counted against the interactive session's tool-search
threshold, and a future reader might naively expose ``TOOLS`` (the
union) to an interactive session.
A small, explicit overlap is allowed: tools tagged with BOTH
``"coordinator": true`` and ``"interactive": true`` (e.g. ``memory``)
intentionally appear in both sets. The whitelist below is the
canonical list of dual-kind tools — any drift here is a real
review-worthy change, not just a count tweak.
"""
from turnstone.core.tools import COORDINATOR_TOOLS, INTERACTIVE_TOOLS, TOOLS
interactive_names = {t["function"]["name"] for t in INTERACTIVE_TOOLS}
coord_names = {t["function"]["name"] for t in COORDINATOR_TOOLS}
# Explicit dual-kind tools — deliberately in both sets.
dual_kind = {"memory"}
overlap = interactive_names & coord_names
assert overlap == dual_kind, (
f"interactive ∩ coordinator should be exactly {dual_kind}, got {overlap}. "
f"Update dual_kind if a new tool legitimately joins both sets."
)
# Coordinator set is non-empty (spawn/inspect/send/close/delete/list).
assert coord_names, "expected at least one coordinator tool"
# Union covers every loaded tool (no tool is in neither set).
all_names = {t["function"]["name"] for t in TOOLS}
assert interactive_names | coord_names == all_names
def test_chatsession_interactive_kind_excludes_coordinator_tools(tmp_db):
"""An interactive ``ChatSession`` does not surface coordinator tools."""
from unittest.mock import MagicMock
from turnstone.core.session import ChatSession
class _NullUI:
def __getattr__(self, _name):
return lambda *a, **kw: None
sess = ChatSession(
client=MagicMock(),
model="test-model",
ui=_NullUI(),
instructions=None,
temperature=0.5,
max_tokens=4096,
tool_timeout=30,
)
names = {t["function"]["name"] for t in sess._tools}
# None of the coordinator-only names should be in the interactive
# session's tool set.
for coord_name in (
"spawn_workstream",
"inspect_workstream",
"send_to_workstream",
"close_workstream",
"cancel_workstream",
"delete_workstream",
"list_workstreams",
"list_nodes",
"list_skills",
"tasks",
"wait_for_workstream",
):
assert coord_name not in names, f"{coord_name} leaked into interactive session tools"
def test_chatsession_coordinator_kind_excludes_interactive_tools(tmp_db):
"""A coordinator ``ChatSession`` sees only coordinator-kind tools.
``memory`` IS in the coord set (it's marked dual-kind in
``memory.json`` so coordinators can persist orchestration context
via the ``coordinator`` scope), but the IC-only tools (bash,
edit_file, ...) stay out — those operate on the local node and
have no meaningful semantics from the console.
"""
from unittest.mock import MagicMock
from turnstone.core.session import ChatSession
class _NullUI:
def __getattr__(self, _name):
return lambda *a, **kw: None
sess = ChatSession(
client=MagicMock(),
model="test-model",
ui=_NullUI(),
instructions=None,
temperature=0.5,
max_tokens=4096,
tool_timeout=30,
kind="coordinator",
)
names = {t["function"]["name"] for t in sess._tools}
# Coordinator tools present, IC-only tools absent.
assert "spawn_workstream" in names
assert "bash" not in names
assert "edit_file" not in names
# Memory is intentionally exposed — see docstring.
assert "memory" in names
# Sub-agent tool lists are zeroed for coordinators.
assert sess._task_tools == []
assert sess._agent_tools == []
def test_chatsession_coordinator_kind_does_not_merge_mcp_tools(tmp_db):
"""Coordinator ChatSession ignores any attached MCP client tool surface.
Coordinators are meta-orchestrators that spawn child workstreams;
MCP tools live on the children. Giving the coordinator direct MCP
access defeats the child-spawning pattern.
"""
from unittest.mock import MagicMock
from turnstone.core.session import ChatSession
class _NullUI:
def __getattr__(self, _name):
return lambda *a, **kw: None
mcp_client = MagicMock()
mcp_client.get_tools.return_value = [
{"type": "function", "function": {"name": "mcp__foo__bar", "parameters": {}}}
]
sess = ChatSession(
client=MagicMock(),
model="test-model",
ui=_NullUI(),
instructions=None,
temperature=0.5,
max_tokens=4096,
tool_timeout=30,
kind="coordinator",
mcp_client=mcp_client,
)
names = {t["function"]["name"] for t in sess._tools}
# No MCP tools in the coordinator surface.
assert "mcp__foo__bar" not in names
# And no MCP listeners were registered (defence-in-depth: MCP tool
# refreshes can't mutate the coordinator's fixed tool set).
mcp_client.add_listener.assert_not_called()
mcp_client.add_resource_listener.assert_not_called()
mcp_client.add_prompt_listener.assert_not_called()