mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
8d1190d17a
- docs/tools.md: sync the tool-JSON metadata-keys table to _META_KEYS — it had drifted to 3 of 8 keys (coordinator, interactive, kind_variants were already missing; cwd_note/workspace_note are new). - tests: cover the third note-rebuild trigger (_drop_mcp_surface) with a count==1 assertion on both lanes, and pin the deliberately uniform workspace_note wording across the fs tools so a one-file reword cannot drift the copies apart.
232 lines
10 KiB
Python
232 lines
10 KiB
Python
"""Working-directory/workspace notes rendered into fs-tool descriptions.
|
|
|
|
Covers the pure renderer (``apply_cwd_context``), the metadata invariants the
|
|
session wiring relies on, and the ChatSession build sites (construction, MCP
|
|
rebuild) including the guarded ``os.getcwd()`` read and the task-agent lane.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from turnstone.core.session import ChatSession
|
|
from turnstone.core.tools import (
|
|
_META,
|
|
COORDINATOR_TOOLS,
|
|
INTERACTIVE_TOOLS,
|
|
TASK_AGENT_TOOLS,
|
|
TOOLS,
|
|
apply_cwd_context,
|
|
)
|
|
|
|
_FS_TOOLS = ("bash", "read_file", "write_file", "edit_file", "search", "diff_file")
|
|
|
|
|
|
def _desc(tools: list[dict], name: str) -> str:
|
|
for t in tools:
|
|
if t["function"]["name"] == name:
|
|
return t["function"]["description"]
|
|
raise AssertionError(f"tool {name!r} not in list")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# apply_cwd_context (pure renderer)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestApplyCwdContext:
|
|
def test_notes_rendered_on_fs_tools(self):
|
|
out = apply_cwd_context(INTERACTIVE_TOOLS, "/data", "/workspace")
|
|
assert "Commands run in /data" in _desc(out, "bash")
|
|
assert "cd does not persist" in _desc(out, "bash")
|
|
assert "The user's workspace directory is /workspace." in _desc(out, "bash")
|
|
for name in ("read_file", "write_file", "edit_file", "search", "diff_file"):
|
|
assert "Relative paths resolve against /data." in _desc(out, name)
|
|
assert "The user's workspace directory is /workspace." in _desc(out, name)
|
|
|
|
def test_noteless_tools_pass_through_by_reference(self):
|
|
out = apply_cwd_context(INTERACTIVE_TOOLS, "/data", "/workspace")
|
|
by_name = {t["function"]["name"]: t for t in out}
|
|
base_by_name = {t["function"]["name"]: t for t in INTERACTIVE_TOOLS}
|
|
assert by_name["web_search"] is base_by_name["web_search"]
|
|
# Noted tools are fresh copies.
|
|
assert by_name["bash"] is not base_by_name["bash"]
|
|
|
|
def test_module_constants_never_mutated(self):
|
|
# The fs tool dicts are SHARED across TOOLS/INTERACTIVE_TOOLS/
|
|
# TASK_AGENT_TOOLS and aliased through merge_mcp_tools output — an
|
|
# in-place append would corrupt every list at once.
|
|
before = {name: _desc(TOOLS, name) for name in _FS_TOOLS}
|
|
apply_cwd_context(INTERACTIVE_TOOLS, "/data", "/workspace")
|
|
apply_cwd_context(TASK_AGENT_TOOLS, "/data", "/workspace")
|
|
for name in _FS_TOOLS:
|
|
assert _desc(TOOLS, name) == before[name]
|
|
assert "/data" not in _desc(INTERACTIVE_TOOLS, name)
|
|
|
|
def test_empty_working_dir_drops_cwd_note_only(self):
|
|
out = apply_cwd_context(INTERACTIVE_TOOLS, "", "/workspace")
|
|
assert "Commands run in" not in _desc(out, "bash")
|
|
assert "Relative paths resolve" not in _desc(out, "read_file")
|
|
assert "The user's workspace directory is /workspace." in _desc(out, "bash")
|
|
|
|
def test_empty_workspace_drops_workspace_note_only(self):
|
|
out = apply_cwd_context(INTERACTIVE_TOOLS, "/data", "")
|
|
assert "Commands run in /data" in _desc(out, "bash")
|
|
assert "workspace directory" not in _desc(out, "bash")
|
|
|
|
def test_both_empty_is_pass_through(self):
|
|
out = apply_cwd_context(INTERACTIVE_TOOLS, "", "")
|
|
assert out == INTERACTIVE_TOOLS
|
|
assert out is not INTERACTIVE_TOOLS # still a fresh list
|
|
|
|
def test_mcp_style_tool_untouched(self):
|
|
mcp_tool = {
|
|
"type": "function",
|
|
"function": {"name": "mcp__srv__thing", "description": "Does a thing."},
|
|
}
|
|
out = apply_cwd_context([mcp_tool], "/data", "/workspace")
|
|
assert out[0] is mcp_tool
|
|
assert out[0]["function"]["description"] == "Does a thing."
|
|
|
|
def test_paths_with_braces_are_literal(self):
|
|
# str.replace substitution — a path containing brace characters must
|
|
# land verbatim (str.format would raise or mangle here).
|
|
out = apply_cwd_context(INTERACTIVE_TOOLS, "/data/{odd}", "")
|
|
assert "Commands run in /data/{odd}" in _desc(out, "bash")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Metadata invariants the session wiring relies on
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestNoteMetadataInvariants:
|
|
def test_all_fs_tools_declare_both_notes(self):
|
|
for name in _FS_TOOLS:
|
|
assert _META[name].get("cwd_note"), name
|
|
assert _META[name].get("workspace_note"), name
|
|
|
|
def test_no_coordinator_tool_declares_notes(self):
|
|
# The coordinator build sites skip _apply_cwd_notes on the strength
|
|
# of this invariant.
|
|
for t in COORDINATOR_TOOLS:
|
|
name = t["function"]["name"]
|
|
meta = _META.get(name) or {}
|
|
assert not meta.get("cwd_note"), name
|
|
assert not meta.get("workspace_note"), name
|
|
|
|
def test_notes_stripped_from_wire_schema(self):
|
|
# _META_KEYS extraction: the raw JSON keys must not leak into the
|
|
# OpenAI function dict sent to providers.
|
|
for t in TOOLS:
|
|
assert "cwd_note" not in t["function"]
|
|
assert "workspace_note" not in t["function"]
|
|
|
|
def test_workspace_note_wording_uniform(self):
|
|
# The workspace fact is one node-level value, so its sentence is
|
|
# deliberately identical across the fs tools (unlike cwd_note, whose
|
|
# prose is per-tool). Guards a one-file reword from drifting the
|
|
# copies apart, independent of the exact wording.
|
|
notes = {_META[name]["workspace_note"] for name in _FS_TOOLS}
|
|
assert len(notes) == 1, notes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ChatSession build sites
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_session(**kwargs):
|
|
defaults = dict(
|
|
client=MagicMock(),
|
|
model="test-model",
|
|
ui=MagicMock(),
|
|
instructions=None,
|
|
temperature=0.5,
|
|
max_tokens=1024,
|
|
tool_timeout=10,
|
|
)
|
|
defaults.update(kwargs)
|
|
return ChatSession(**defaults)
|
|
|
|
|
|
class TestSessionCwdNotes:
|
|
def test_fresh_session_carries_cwd_note(self, tmp_db):
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=None):
|
|
session = _make_session()
|
|
assert f"Commands run in {os.getcwd()}" in _desc(session._tools, "bash")
|
|
assert f"Relative paths resolve against {os.getcwd()}." in _desc(
|
|
session._tools, "read_file"
|
|
)
|
|
|
|
def test_task_lane_carries_cwd_note(self, tmp_db):
|
|
# Sub-agents use self._task_tools, a separate list from self._tools —
|
|
# they run in this same process, so the same cwd applies.
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=None):
|
|
session = _make_session()
|
|
assert f"Commands run in {os.getcwd()}" in _desc(session._task_tools, "bash")
|
|
|
|
def test_getcwd_failure_degrades_to_noteless(self, tmp_db):
|
|
# A deleted cwd (eval workdir teardown) must not break session
|
|
# construction or an MCP background-thread rebuild.
|
|
with (
|
|
patch("turnstone.core.session.get_workspace_dir", return_value=None),
|
|
patch("os.getcwd", side_effect=OSError("cwd deleted")),
|
|
):
|
|
session = _make_session()
|
|
assert "Commands run in" not in _desc(session._tools, "bash")
|
|
assert session._tools # built fine, just note-less
|
|
|
|
def test_workspace_rendered_when_dir_exists(self, tmp_db, tmp_path):
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=str(tmp_path)):
|
|
session = _make_session()
|
|
assert f"The user's workspace directory is {tmp_path}." in _desc(session._tools, "bash")
|
|
|
|
def test_workspace_skipped_when_dir_missing(self, tmp_db, tmp_path):
|
|
missing = tmp_path / "does-not-exist"
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=str(missing)):
|
|
session = _make_session()
|
|
assert "workspace directory" not in _desc(session._tools, "bash")
|
|
|
|
def test_workspace_skipped_when_equal_to_cwd(self, tmp_db):
|
|
# e.g. an operator who set working_dir: /workspace on the container —
|
|
# one fact, not two copies of the same path.
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=os.getcwd()):
|
|
session = _make_session()
|
|
desc = _desc(session._tools, "bash")
|
|
assert f"Commands run in {os.getcwd()}" in desc
|
|
assert "workspace directory" not in desc
|
|
|
|
def test_constants_pristine_after_session_build(self, tmp_db):
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=None):
|
|
_make_session()
|
|
assert os.getcwd() not in _desc(INTERACTIVE_TOOLS, "bash")
|
|
assert os.getcwd() not in _desc(TOOLS, "bash")
|
|
|
|
def test_mcp_rebuild_keeps_single_note(self, tmp_db):
|
|
# The MCP list_changed rebuild re-derives from pristine bases — the
|
|
# note must survive exactly once (double-append is the failure the
|
|
# assignment-time design must never regress into).
|
|
mock_mcp = MagicMock()
|
|
mock_mcp.get_tools.return_value = []
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=None):
|
|
session = _make_session(mcp_client=mock_mcp)
|
|
session._on_mcp_tools_changed()
|
|
session._on_mcp_tools_changed()
|
|
assert _desc(session._tools, "bash").count(f"Commands run in {os.getcwd()}") == 1
|
|
assert _desc(session._task_tools, "bash").count(f"Commands run in {os.getcwd()}") == 1
|
|
|
|
def test_mcp_drop_surface_keeps_single_note(self, tmp_db):
|
|
# The MCP-disconnect rebuild (_drop_mcp_surface, reached via resume()
|
|
# adopting an MCP-off persona) is the third rebuild trigger — the note
|
|
# must survive it, exactly once, on both lanes.
|
|
mock_mcp = MagicMock()
|
|
mock_mcp.get_tools.return_value = []
|
|
with patch("turnstone.core.session.get_workspace_dir", return_value=None):
|
|
session = _make_session(mcp_client=mock_mcp)
|
|
session._drop_mcp_surface()
|
|
assert session._mcp_client is None
|
|
assert _desc(session._tools, "bash").count(f"Commands run in {os.getcwd()}") == 1
|
|
assert _desc(session._task_tools, "bash").count(f"Commands run in {os.getcwd()}") == 1
|