mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
9a30530d41
* 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.
410 lines
15 KiB
Python
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()
|