turnstone/tests/test_workstream_kind.py

"""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.  ``skills``
    # joined in 1.6.0 (replaces legacy ``skill`` + ``list_skills``) — read
    # actions auto-approve on both kinds, write actions gate on
    # ``model.skills.write`` permission, and ``load`` errors on coord
    # sessions where it doesn't apply.  ``notify`` joined in 1.6.0 so
    # coords can post status updates at narrative beats without spawning
    # a child purely to ship a message.  ``read_resource``/``use_prompt``
    # joined in 1.8 (#725): coordinators get the full MCP surface —
    # tools, resources, prompts — persona-gated like every session.
    dual_kind = {"memory", "skills", "notify", "read_resource", "use_prompt"}

    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

    sess = ChatSession(
        client=MagicMock(),
        model="test-model",
        ui=_null_ui(),
        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.
    """
    sess = _make_coordinator(mcp_client=None)
    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 list is zeroed for coordinators.
    assert sess._task_tools == []


def _null_ui():
    class _NullUI:
        def __getattr__(self, _name):
            return lambda *a, **kw: None

    return _NullUI()


def _mcp_client_mock():
    from unittest.mock import MagicMock

    mcp_client = MagicMock()
    mcp_client.get_tools.return_value = [
        {"type": "function", "function": {"name": "mcp__foo__bar", "parameters": {}}}
    ]
    return mcp_client


def _make_coordinator(mcp_client, persona_snapshot=None):
    from unittest.mock import MagicMock

    from turnstone.core.session import ChatSession

    return ChatSession(
        client=MagicMock(),
        model="test-model",
        ui=_null_ui(),
        instructions=None,
        temperature=0.5,
        max_tokens=4096,
        tool_timeout=30,
        user_id="user-1",  # the constructor refuses anonymous coordinators
        kind="coordinator",
        mcp_client=mcp_client,
        persona_snapshot=persona_snapshot,
    )


def test_chatsession_coordinator_mcp_keyed_on_client_presence(tmp_db):
    """Coordinator MCP is keyed on CLIENT PRESENCE (#725).

    The console session factory — the only coordinator producer (the
    CLI is blocked by the anonymous-coordinator guard; nodes reject
    non-interactive kinds at create and at session-load) — passes the
    live console manager unconditionally; the persona MCP toggle
    governs the surface, like interactive.  Session-level contract: a
    coordinator constructed WITH an mcp_client treats MCP as enabled and
    runs the same listener/prime skeleton as interactive over the
    COORDINATOR_TOOLS base; one constructed WITHOUT keeps the fixed
    builtin surface.
    """
    from unittest.mock import patch

    # Cell (a): no client → today's fixed invariant, verbatim.
    sess = _make_coordinator(mcp_client=None)
    names = {t["function"]["name"] for t in sess._tools}
    assert "mcp__foo__bar" not in names
    assert "spawn_workstream" in names
    assert sess._task_tools == []

    # Cell (b): client present → merged surface + ALL THREE listeners +
    # the creator's pools primed (full parity with interactive).
    mcp_client = _mcp_client_mock()
    with patch("turnstone.core.session.try_prime_user_pools") as prime:
        sess = _make_coordinator(mcp_client=mcp_client)
    names = {t["function"]["name"] for t in sess._tools}
    assert "mcp__foo__bar" in names
    assert "spawn_workstream" in names  # base is additive, never replaced
    assert sess._task_tools == []  # coordinators have no task-agent lane
    mcp_client.add_listener.assert_called_once()
    assert mcp_client.add_listener.call_args.kwargs.get("user_id") == "user-1"
    mcp_client.add_resource_listener.assert_called_once()
    mcp_client.add_prompt_listener.assert_called_once()
    prime.assert_called_once()
    assert prime.call_args.args[1] == "user-1"


def test_chatsession_coordinator_persona_mcp_off_wins(tmp_db):
    """A coordinator persona with mcp=False gates the surface off even
    when the factory passed a live client — same precedence as
    interactive (the 1762 persona gate composes upstream of the kind
    branch), and ``_mcp_gated_off`` records that a real client was
    withheld so resume() refuses to adopt an MCP-on stamp."""
    from turnstone.core.personas import PersonaSnapshot

    mcp_client = _mcp_client_mock()
    snap = PersonaSnapshot(name="p", prompt="x", tools=None, mcp=False, memory=True)
    sess = _make_coordinator(mcp_client=mcp_client, persona_snapshot=snap)
    names = {t["function"]["name"] for t in sess._tools}
    assert "mcp__foo__bar" not in names
    mcp_client.add_listener.assert_not_called()
    mcp_client.add_resource_listener.assert_not_called()
    mcp_client.add_prompt_listener.assert_not_called()
    assert sess._mcp_gated_off is True


def test_chatsession_coordinator_drop_mcp_surface_resets_to_coordinator_tools(tmp_db):
    """The resume()-adopts-MCP-off-stamp path (_drop_mcp_surface) on a
    COORDINATOR resets to COORDINATOR_TOOLS — never the interactive lanes
    — and removes all three listeners under the tracked registration
    identity.  Direct cell for the docstring's both-kinds claim; every
    sibling coordinator MCP transition has one."""
    from unittest.mock import patch

    mcp_client = _mcp_client_mock()
    with patch("turnstone.core.session.try_prime_user_pools"):
        sess = _make_coordinator(mcp_client=mcp_client)
    assert "mcp__foo__bar" in {t["function"]["name"] for t in sess._tools}

    sess._drop_mcp_surface()

    names = {t["function"]["name"] for t in sess._tools}
    assert "mcp__foo__bar" not in names
    assert "spawn_workstream" in names
    assert sess._task_tools == []
    assert sess._mcp_client is None
    mcp_client.remove_listener.assert_called_once()
    assert mcp_client.remove_listener.call_args.kwargs.get("user_id") == "user-1"
    mcp_client.remove_resource_listener.assert_called_once()
    mcp_client.remove_prompt_listener.assert_called_once()


def test_chatsession_coordinator_drop_mcp_surface_after_rebind_uses_tracked_id(tmp_db):
    """After bind_acting_user re-scopes the listener registrations to a
    NEW operator, a subsequent surface drop must remove them under the
    tracked _mcp_listener_user_id (the rebound identity) — removal keyed
    on the owner id would leave the rebound registrations leaking."""
    from unittest.mock import patch

    mcp_client = _mcp_client_mock()
    with patch("turnstone.core.session.try_prime_user_pools"):
        sess = _make_coordinator(mcp_client=mcp_client)
        sess.bind_acting_user("user-2")

    assert sess._mcp_listener_user_id == "user-2"

    sess._drop_mcp_surface()

    assert mcp_client.remove_listener.call_args.kwargs.get("user_id") == "user-2"
    assert mcp_client.remove_resource_listener.call_args.kwargs.get("user_id") == "user-2"
    assert mcp_client.remove_prompt_listener.call_args.kwargs.get("user_id") == "user-2"


def test_coordinator_catalog_change_rebuilds_merged_tools(tmp_db):
    """A registered coordinator's tool listener rebuilds the merged set on
    catalog change — the surface tracks admin edits and reconnects instead
    of freezing at construction (the reversed early-return)."""
    mcp_client = _mcp_client_mock()
    sess = _make_coordinator(mcp_client=mcp_client)
    cb = mcp_client.add_listener.call_args.args[0]
    mcp_client.get_tools.return_value = [
        {"type": "function", "function": {"name": "mcp__foo__baz", "parameters": {}}}
    ]
    cb()
    names = {t["function"]["name"] for t in sess._tools}
    assert "mcp__foo__baz" in names
    assert "mcp__foo__bar" not in names  # fresh merge, not accretion
    assert "spawn_workstream" in names  # base survives every rebuild
    assert sess._task_tools == []


def test_coordinator_construction_race_converges_post_snapshot_change(tmp_db):
    """A catalog change landing AFTER the authoritative read must be
    converged by the end-of-construction recheck.  The seq bump rides
    get_tools' FIRST call so it lands strictly after the constructor's
    snapshot — a bump at add_listener time would fold into the snapshot
    and pass whether or not the recheck runs for coordinators."""
    from unittest.mock import MagicMock

    mcp_client = MagicMock()
    v1 = [{"type": "function", "function": {"name": "mcp__foo__v1", "parameters": {}}}]
    v2 = [{"type": "function", "function": {"name": "mcp__foo__v2", "parameters": {}}}]
    calls = {"n": 0}

    def _get_tools(user_id=None):
        calls["n"] += 1
        if calls["n"] == 1:
            # Simulate a notification landing between the authoritative
            # read and the end of tool setup: advance the bound session's
            # change counter (the listener callback's __self__).
            cb = mcp_client.add_listener.call_args.args[0]
            cb.__self__._mcp_tools_change_seq += 1
            return v1
        return v2

    mcp_client.get_tools.side_effect = _get_tools
    sess = _make_coordinator(mcp_client=mcp_client)
    names = {t["function"]["name"] for t in sess._tools}
    assert "mcp__foo__v2" in names, "post-snapshot catalog change was lost for a coordinator"


def test_coordinator_bind_acting_user_rescopes_listeners_and_primes(tmp_db):
    """Coordinators are multi-sender: a fresh turn from operator B must
    re-scope the MCP listeners to B and prime B's pools, so B dispatches
    against B's catalog — never A's (#725, security-critical cell)."""
    from unittest.mock import patch

    mcp_client = _mcp_client_mock()
    with patch("turnstone.core.session.try_prime_user_pools") as prime:
        sess = _make_coordinator(mcp_client=mcp_client)
        sess.bind_acting_user("user-2")
    assert mcp_client.remove_listener.call_args.kwargs.get("user_id") == "user-1"
    assert mcp_client.add_listener.call_args.kwargs.get("user_id") == "user-2"
    assert mcp_client.remove_resource_listener.call_args.kwargs.get("user_id") == "user-1"
    assert mcp_client.add_resource_listener.call_args.kwargs.get("user_id") == "user-2"
    assert mcp_client.remove_prompt_listener.call_args.kwargs.get("user_id") == "user-1"
    assert mcp_client.add_prompt_listener.call_args.kwargs.get("user_id") == "user-2"
    assert prime.call_args.args[1] == "user-2"
    assert sess._mcp_effective_user_id == "user-2"


def test_coordinator_close_removes_listeners_owner_only(tmp_db):
    """Exit-path cell (i): closing an owner-only coordinator removes all
    three listeners under the owner identity."""
    mcp_client = _mcp_client_mock()
    sess = _make_coordinator(mcp_client=mcp_client)
    sess.close()
    assert mcp_client.remove_listener.call_args.kwargs.get("user_id") == "user-1"
    assert mcp_client.remove_resource_listener.call_args.kwargs.get("user_id") == "user-1"
    assert mcp_client.remove_prompt_listener.call_args.kwargs.get("user_id") == "user-1"


def test_coordinator_close_after_rebind_removes_under_new_identity(tmp_db):
    """Exit-path cell (ii), the wrong-field regression pin: after operator
    B rebinds, eviction/close must remove listeners under B's uid (the
    tracked ``_mcp_listener_user_id``), not the owner's ``_mcp_user_id`` —
    owner-only closes cannot distinguish the two fields (both equal the
    owner at construction)."""
    from unittest.mock import patch

    mcp_client = _mcp_client_mock()
    with patch("turnstone.core.session.try_prime_user_pools"):
        sess = _make_coordinator(mcp_client=mcp_client)
        sess.bind_acting_user("user-2")
    sess.close()
    assert mcp_client.remove_listener.call_args.kwargs.get("user_id") == "user-2"
    assert mcp_client.remove_resource_listener.call_args.kwargs.get("user_id") == "user-2"
    assert mcp_client.remove_prompt_listener.call_args.kwargs.get("user_id") == "user-2"


def test_coordinator_read_resource_use_prompt_on_wire_and_approval(tmp_db):
    """Dual-kind read_resource/use_prompt (#725) — the WIRE matrix via
    _get_active_tools(): present with a client exposing nonzero
    resource/prompt counts; STRIPPED without a client; STRIPPED with a
    client whose counts are zero (the _without_tool gate keys on client
    presence AND per-user catalog count).  Base-list membership is
    asserted separately — both tools sit in COORDINATOR_TOOLS
    unconditionally.  Approval: needs_approval preserved
    (auto_approve:false)."""
    mcp_client = _mcp_client_mock()
    # Explicit integers: a bare MagicMock return is truthy, which would
    # let the present-cell pass without the count gate ever working.
    mcp_client.resource_count_for_user.return_value = 2
    mcp_client.prompt_count_for_user.return_value = 1
    sess = _make_coordinator(mcp_client=mcp_client)
    assert {"read_resource", "use_prompt"} <= {t["function"]["name"] for t in sess._tools}
    wire = {t["function"]["name"] for t in sess._get_active_tools()}
    assert {"read_resource", "use_prompt"} <= wire
    item = sess._prepare_read_resource("c1", {"uri": "res://x"})
    assert item.get("needs_approval") is True

    bare = _make_coordinator(mcp_client=None)
    assert {"read_resource", "use_prompt"} <= {t["function"]["name"] for t in bare._tools}
    bare_wire = {t["function"]["name"] for t in bare._get_active_tools()}
    assert not ({"read_resource", "use_prompt"} & bare_wire), (
        "client-less coordinator must not advertise MCP catalog tools on the wire"
    )

    zero = _mcp_client_mock()
    zero.resource_count_for_user.return_value = 0
    zero.prompt_count_for_user.return_value = 0
    sess_zero = _make_coordinator(mcp_client=zero)
    zero_wire = {t["function"]["name"] for t in sess_zero._get_active_tools()}
    assert not ({"read_resource", "use_prompt"} & zero_wire), (
        "zero-count catalogs must strip read_resource/use_prompt from the wire"
    )