portfolio-optimiser-claude/tests/test_sdk_tool_namespace_loadbearing.py

"""B4: does the Claude Agent SDK hand us server identity for free in tool names?

The D7 mirroring queue (``docs/2026-08-18-funn-koeer-og-gater.md § D7-speilingskøen``)
asks one question of this side: when ``create_sdk_mcp_server`` emits a tool, does
that tool's name carry the server it belongs to? The MAF sibling gets tool names
WITHOUT a server prefix (their S3.4/F10), so a recorder there must carry server
identity itself. Whether we pay the same price was never measured here.

MEASURED 2026-08-25, offline, against claude-agent-sdk 0.2.139 (the version
``tests/test_sdk_version_guard.py`` pins as read): **NO — not free.** The Python
package emits the bare tool name; ``mcp__`` appears in ZERO of its 24 source files
(positive control: ``create_sdk_mcp_server`` IS findable by the same query).
Server identity lives on ``McpSdkServerConfig["name"]`` and ``Server.name``,
disjoint from every name the tool list carries.

HONEST LIMIT — what this does NOT say. The namespaced form ``mcp__<server>__<tool>``
does exist: it is built inside the CLI bundled with the SDK (178 literal
occurrences of ``mcp__`` in ``_bundled/claude``; the construction reads
``` `mcp__${...}__${...}` ```). That was READ from the artifact, not observed in a
run of ours — seeing it emitted would take a live ``query()``, which the D6 cost
rule and this suite's offline invariant both forbid. So the finding is scoped to
the seam we can actually build on: the in-process construction side. A recorder
hung there sees the bare name and must be told which server it came from —
the same shape the sibling pays for, arrived at after 2026-08-09 and therefore
NOT readable as independent convergence.

Offline-safe: constructing an SDK MCP server touches no network and needs no API
key. Nothing here starts the bundled CLI.
"""

from __future__ import annotations

import asyncio
from typing import Any

from claude_agent_sdk import McpSdkServerConfig, create_sdk_mcp_server, tool

SERVER_NAME = "tool_call_recorder"
BARE_TOOL_NAMES = ("record_call", "flush_calls")

# What a "free" server identity would have to look like for B4 to be answered
# YES. Kept as data so the assertions below read as the question, not as a
# restatement of the answer.
NAMESPACED_PREFIX = f"mcp__{SERVER_NAME}__"


@tool("record_call", "Record one tool call", {"name": str})
async def _record_call(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": args["name"]}]}


@tool("flush_calls", "Flush recorded calls", {})
async def _flush_calls(args: dict[str, Any]) -> dict[str, Any]:
    return {"content": [{"type": "text", "text": "flushed"}]}


def _server(*, tools: list[Any] | None = None) -> McpSdkServerConfig:
    return create_sdk_mcp_server(
        name=SERVER_NAME,
        version="1.0.0",
        tools=[_record_call, _flush_calls] if tools is None else tools,
    )


def _emitted_tool_names(config: McpSdkServerConfig) -> list[str]:
    """Read the tool names the server ACTUALLY emits.

    Not a model of the emission — the registered ``tools/list`` handler is
    invoked and its result read, because a check that models a tool instead of
    reading it is green-but-dead (økt 28). The coupling that buys this is the
    MCP lowlevel ``Server.request_handlers`` dispatch table; should its shape
    change, the lookup below raises rather than quietly returning nothing, so
    the failure mode is RED, not a false negative.
    """
    from mcp.types import ListToolsRequest

    handler = config["instance"].request_handlers[ListToolsRequest]

    async def _call() -> list[str]:
        result = await handler(ListToolsRequest(method="tools/list"))
        return [t.name for t in result.root.tools]

    return asyncio.run(_call())


class TestEmittedToolNamesCarryNoServerIdentity:
    def test_the_emission_is_readable_and_the_population_is_two(self) -> None:
        # POSITIVE CONTROL, and it runs FIRST: every negative below is worth
        # exactly as much as this query's ability to find anything at all. An
        # empty tool list would make "no name carries the prefix" vacuously
        # true — so the denominator is asserted, not assumed.
        names = _emitted_tool_names(_server())
        assert len(names) == 2, f"emission unreadable or empty — measured {names!r}"
        assert all(isinstance(n, str) and n for n in names)

    def test_no_emitted_name_carries_the_server_prefix(self) -> None:
        # THE B4 ANSWER, pinned. Red the day the SDK starts namespacing at
        # construction time — which is the day this repo can stop carrying
        # server identity by hand and the D7 queue note becomes wrong.
        names = _emitted_tool_names(_server())
        assert names == list(BARE_TOOL_NAMES)
        assert not any(n.startswith("mcp__") for n in names), (
            f"the SDK now namespaces at construction time — B4 flipped to YES: {names!r}"
        )
        assert not any(SERVER_NAME in n for n in names)

    def test_the_name_is_passed_through_untouched(self) -> None:
        # POSITIVE CONTROL FOR THE NEGATIVE. The assertion above only means
        # something if a prefix WOULD have shown up had one been there. Feed
        # the server a tool already wearing the namespaced form: it comes back
        # verbatim. So the emission is pass-through, the query can see a
        # prefix, and its absence above is a measurement — not a blind spot.
        @tool(f"{NAMESPACED_PREFIX}record_call", "Pre-namespaced", {"name": str})
        async def _prefixed(args: dict[str, Any]) -> dict[str, Any]:
            return {"content": [{"type": "text", "text": args["name"]}]}

        names = _emitted_tool_names(_server(tools=[_prefixed]))
        assert names == [f"{NAMESPACED_PREFIX}record_call"]

    def test_server_identity_exists_but_lives_off_the_tool_name(self) -> None:
        # The disjointness is the whole finding: identity is AVAILABLE, just
        # not on the tool. A recorder must join the two itself — this pins
        # both halves so "free" cannot be assumed from either one alone.
        config = _server()
        assert config["type"] == "sdk"
        assert config["name"] == SERVER_NAME
        assert config["instance"].name == SERVER_NAME
        assert SERVER_NAME not in "".join(_emitted_tool_names(config))