AIGTWY-4878 [3/3] Test MCP registration and tool use in Claude and Codex

tests/e2e_cuj/MCP.md

@@ -0,0 +1,40 @@
+# MCP registration and listing CUJ
+
+The dedicated workspace must publish `mcp_servers.unity_catalog_location = ug_e2e.tools`
+for both agents. Fixtures are provisioned separately; this suite does not modify them.
+
+| Service | Read-only tool |
+| --- | --- |
+| `ug_e2e.tools.fixture_reader` | `read_fixture(run_id)` |
+| `ug_e2e.tools.fixture_metadata` | `describe_fixture(run_id)` |
+
+The two TUI journeys configure installed UG, open the real agent MCP inventory, and require
+connected servers with their tools loaded. Claude opens each server's details and tools
+view; Codex uses `/mcp verbose`. Both then call the two tools with a fresh run ID and return
+their receipts. Expected receipts are computed locally and withheld from the prompt.
+Terminal actions, rendered screens, and native transcripts are recorded by the harness.
+
+Both inventories must omit `ug_e2e.other_tools`, including the accessible
+`fixture_decoy` service and its `decoy_status` tool. Claude traverses the server menu
+until it wraps; Codex includes rendered scrollback from `/mcp verbose`.
+
+The command journey checks that `ug mcp list` shows both managed fixture rows for `claude, codex`
+with aggregate `connected` status and no `ug_e2e.other_tools` servers. This combines Claude's
+health check with Codex's `enabled` registration; Codex connectivity is verified by its TUI/tool
+journey, not the aggregate label. Local-server counts are not asserted.
+
+Run on a clean disposable POSIX host with `ug`, `claude`, `codex`, and `databricks` on `PATH`.
+Set `UG_CUJ_SP_CLIENT_ID` and `UG_CUJ_SP_CLIENT_SECRET`, install `pexpect==4.9.0` and
+`pyte==0.8.2`, and use a host without machine-wide agent settings. The test uses
+the dedicated workspace URL declared on `TestMcpRegistration`; no workspace URL environment
+variable is required.
+
+```bash
+uv run --with pexpect==4.9.0 --with pyte==0.8.2 pytest \
+  --confcutdir=tests/e2e_cuj tests/e2e_cuj \
+  -m 'managed and mcp_registration and workspace_isolated' -v
+```
+
+CI discovers these tests in the shared `dedicated-cuj` job alongside catalog discovery.
+Only live passes establish coverage. `ug status` is not part of this CUJ.
+Skill storage and provisioning are not part of this suite.

tests/e2e_cuj/pytest.ini

@@ -5,6 +5,7 @@ addopts = --strict-markers --tb=short
 markers =
     managed: requires an admin CodingAgentConfig published on the selected workspace
     catalog_discovery: managed schema-scoped model discovery and inference
+    mcp_registration: managed MCP registration, connection, and tool use
     workspace_isolated: reads preconfigured CUJ workspace state with isolated local sessions
     tui: real interactive terminal and completed task
     claude: only runs when Claude Code is explicitly selected

tests/e2e_cuj/test_mcp_registration.py

@@ -0,0 +1,231 @@
+"""Managed MCP registration, connection, and tool use in both real agent TUIs."""
+
+import hashlib
+import json
+import re
+import uuid
+
+import pytest
+
+from tests.integration.utils.evidence import (
+    agent_sessions,
+    assert_no_terminal_api_error,
+    assistant_answers,
+    is_child_session,
+)
+from tests.integration.utils.mcp import open_claude_mcp_inventory, open_codex_mcp_inventory
+from tests.integration.utils.terminal import TerminalProcess
+
+from .base import BaseCujTest
+from .helpers.constants import CLAUDE, CODEX, MANAGED_PATHS
+from .helpers.terminal import Terminal
+
+pytestmark = [pytest.mark.managed, pytest.mark.mcp_registration, pytest.mark.workspace_isolated]
+
+SERVICES = {
+    "ug_e2e.tools.fixture_metadata": "describe_fixture",
+    "ug_e2e.tools.fixture_reader": "read_fixture",
+}
+
+
+class McpFixtureTask:
+    def __init__(self):
+        self.run_id = uuid.uuid4().hex
+        self.expected = {
+            tool: hashlib.sha256(
+                b"unity-gateway-cuj-fixture\0"
+                + tool.encode("ascii")
+                + b"\0"
+                + self.run_id.encode("ascii")
+            ).hexdigest()
+            for tool in SERVICES.values()
+        }
+        calls = "; ".join(
+            f"call {tool} on {service.replace('.', '-')} with run_id={self.run_id}"
+            for service, tool in SERVICES.items()
+        )
+        self.prompt = (
+            f"Use these read-only MCP tools: {calls}. "
+            "Do not use shell commands, files, or subagents. Reply with only a JSON object "
+            "mapping each tool name to the text it returned, without code fences."
+        )
+
+    def completed(self, session, agent: str) -> bool:
+        for path, records in agent_sessions(session, agent).items():
+            if is_child_session(agent, path, records):
+                continue
+            for answer in assistant_answers(agent, records):
+                try:
+                    observed = json.loads(answer)
+                except json.JSONDecodeError:
+                    continue
+                if observed == self.expected:
+                    return True
+        return False
+
+
+def _wait_for_mcp_task(tui, task):
+    approved_calls = set()
+    expected_tools = {(service.replace(".", "-"), tool) for service, tool in SERVICES.items()}
+
+    def completed(screen):
+        assert_no_terminal_api_error(screen)
+        assert not any(
+            prompt in screen for prompt in ("Do you want to proceed?", "Do you want to allow")
+        ), "Unexpected permission request; inspect the actual command:\n" + screen
+
+        if (
+            "Allow for this session" in screen
+            and "Always allow" in screen
+            and "Cancel" in screen
+            and re.search(r"(?m)^[ \t]*[›❯>][ \t]*[1-4]\.", screen)
+        ):
+            questions = list(re.finditer(r"(?m)^[ \t]*(Allow\b[^\n?]+\?)[ \t]*$", screen))
+            assert questions, "Unexpected approval menu; inspect the actual command:\n" + screen
+            question = questions[-1].group(1)
+            approval = re.fullmatch(
+                r'Allow the (?P<server>\S+) MCP server to run tool "(?P<tool>[^"]+)"\?',
+                question,
+            )
+            assert tui.agent == CODEX and approval, (
+                "Unexpected Codex permission request; inspect the actual command:\n" + screen
+            )
+            tail = screen[questions[-1].end() :]
+            run_match = re.search(r"(?m)^[ \t]*run_id:[ \t]*(\S+)[ \t]*$", tail)
+            selected = re.search(
+                r"(?m)^[ \t]*[›❯>][ \t]*([1-4])\.[ \t]*(Allow(?: for this session)?|Always allow|Cancel)\b",
+                tail,
+            )
+            assert run_match and selected, (
+                "Unexpected Codex approval menu; inspect the actual command:\n" + screen
+            )
+            server = approval.group("server")
+            tool = approval.group("tool")
+            run_id = run_match.group(1)
+            assert (server, tool) in expected_tools and run_id == task.run_id, (
+                "Unexpected MCP permission request; inspect the actual command:\n" + screen
+            )
+            assert selected.group(1) == "1" and selected.group(2) == "Allow", (
+                "Codex MCP approval must remain a per-call Allow; inspect the actual menu:\n"
+                + screen
+            )
+            call = (server, tool, run_id)
+            if call not in approved_calls:
+                tui.send("\r", f"allow one MCP call: {server}.{tool}")
+                approved_calls.add(call)
+                return False
+        return task.completed(tui.session, tui.agent)
+
+    tui.wait_for(
+        completed,
+        "a completed assistant answer with both verified MCP receipts",
+        timeout=240,
+    )
+
+
+@pytest.fixture(autouse=True)
+def _revert_mcp_test_state(cuj):
+    yield
+    session, _, _ = cuj
+    state_dir = session.home / ".ucode"
+    if any(
+        (state_dir / name).is_file() for name in ("state.json", "managed-backups/manifest.json")
+    ):
+        with TerminalProcess(
+            session,
+            "ug",
+            [str(session.binary), "revert"],
+            "mcp-registration-cleanup-revert",
+        ) as terminal:
+            terminal.finish()
+    assert not any(path.exists() for path in MANAGED_PATHS), (
+        "MCP registration CUJ teardown left machine-wide agent settings"
+    )
+
+
+class TestMcpRegistration(BaseCujTest):
+    WORKSPACE_URL = "https://dbc-bbdd5508-648e.cloud.databricks.com"
+
+    def test_mcp_list_reports_managed_fixture_servers(self, cuj):
+        """Scenario: configure both agents and inspect unscoped `ug mcp list` output.
+
+        Expected: both managed fixture rows list Claude/Codex and aggregate `connected` status;
+        no `ug_e2e.other_tools` server appears. Codex connectivity is verified by its TUI journey.
+        """
+        session, workspace, _recorder = cuj
+        session.configure(
+            [
+                "configure",
+                "--agents",
+                f"{CLAUDE},{CODEX}",
+                "--workspace",
+                workspace.url,
+                "--skip-upgrade",
+            ]
+        )
+        listing = session.run("mcp", "list", timeout=120)
+        output = f"{listing.stdout}\n{listing.stderr}"
+        for service in SERVICES:
+            server_name = service.replace(".", "-")
+            rows = [
+                " ".join(line.split())
+                for line in output.splitlines()
+                if line.split()[:1] == [server_name]
+            ]
+            expected_row = f"{server_name} {service} (managed) {CLAUDE}, {CODEX} connected"
+            assert rows == [expected_row], output
+        assert "ug_e2e.other_tools" not in output, output
+        assert "ug_e2e-other_tools" not in output, output
+
+    @pytest.mark.claude
+    @pytest.mark.tui
+    def test_mcp_registration_claude_servers_connect_and_work(self, cuj):
+        """Scenario: configure UG, inspect Claude's MCP menu, then call both fixture tools.
+
+        Expected: both scoped servers are registered, connected, and expose their tools;
+        no ug_e2e.other_tools server or decoy tool appears in the complete inventory;
+        a parent assistant answer contains the correct receipts for a fresh run ID.
+        """
+        session, workspace, _recorder = cuj
+        session.configure(["configure", "--workspace", workspace.url, "--skip-upgrade"])
+        task = McpFixtureTask()
+        allowed_tools = ",".join(
+            f"mcp__{service.replace('.', '-')}__{tool}" for service, tool in SERVICES.items()
+        )
+        with Terminal(
+            session,
+            "mcp-registration-claude",
+            [CLAUDE, "--", "--allowedTools", allowed_tools],
+        ) as tui:
+            tui.boot()
+            inventory = open_claude_mcp_inventory(tui, SERVICES)
+            assert "ug_e2e-other_tools" not in inventory, inventory
+            assert "decoy_status" not in inventory, inventory
+            tui.submit(task.prompt)
+            _wait_for_mcp_task(tui, task)
+            tui.exit_normally()
+        assert task.completed(session, CLAUDE)
+        session.assert_not_routed()
+
+    @pytest.mark.codex
+    @pytest.mark.tui
+    def test_mcp_registration_codex_servers_connect_and_work(self, cuj):
+        """Scenario: configure UG, inspect Codex's MCP inventory, then call both fixture tools.
+
+        Expected: both scoped servers are registered, connected, and expose their tools;
+        no ug_e2e.other_tools server or decoy tool appears in the complete inventory;
+        a completed parent turn returns the correct receipts for a fresh run ID.
+        """
+        session, workspace, _recorder = cuj
+        session.configure(["configure", "--workspace", workspace.url, "--skip-upgrade"])
+        task = McpFixtureTask()
+        with Terminal(session, "mcp-registration-codex", [CODEX]) as tui:
+            tui.boot()
+            inventory = open_codex_mcp_inventory(tui, SERVICES)
+            assert "ug_e2e-other_tools" not in inventory, inventory
+            assert "decoy_status" not in inventory, inventory
+            tui.submit(task.prompt)
+            _wait_for_mcp_task(tui, task)
+            tui.exit_normally()
+        assert task.completed(session, CODEX)
+        session.assert_not_routed()

tests/integration/utils/mcp.py

@@ -0,0 +1,102 @@
+"""Inspect MCP inventories through the real agent terminals."""
+
+import re
+
+import pyte
+
+
+def open_claude_mcp_inventory(tui, services: dict[str, str]) -> str:
+    tui.submit("/mcp")
+    tui.wait_for(
+        lambda screen: "Manage MCP servers" in screen and tui.selected_line(),
+        "Claude's MCP menu",
+        timeout=120,
+    )
+    tui.wait_for(
+        lambda screen: "connecting" not in screen.casefold(),
+        "Claude's MCP startup checks to finish before opening server details",
+        timeout=120,
+    )
+    first = tui.selected_line()
+    screens = []
+    for _ in range(100):
+        screens.append(tui.visible)
+        current = tui.selected_line()
+        tui.send("\x1b[B", "inspect the next MCP inventory row")
+        tui.wait_for(
+            lambda screen, current=current: tui.selected_line() != current,
+            "the next MCP inventory row",
+        )
+        if tui.selected_line() == first:
+            break
+    else:
+        raise AssertionError("Claude's MCP inventory did not wrap within 100 rows")
+    inventory = "\n".join(screens)
+    tui.actions.append({"reason": "claude-mcp-inventory", "screen": inventory})
+    for service, tool in services.items():
+        name = service.replace(".", "-")
+        tui.choose("Manage MCP servers", name)
+        tui.wait_for(
+            lambda screen, name=name: (
+                name in screen.casefold()
+                and re.search(r"(?m)^\s*Status:\s*(?:[✓✔]\s*)?connected\s*$", screen)
+                and "View tools" in screen
+                and "tools fetch failed" not in screen
+            ),
+            f"{name} connected with tools loaded",
+            timeout=120,
+        )
+        tui.actions.append({"reason": f"mcp-connected-{name}", "screen": tui.visible})
+        tui.choose("MCP Server", "View tools")
+        tui.wait_for(
+            lambda screen, tool=tool: re.search(rf"\b{re.escape(tool)}\b", screen),
+            f"{tool} in {name}'s tools menu",
+        )
+        tui.actions.append({"reason": f"mcp-tools-{name}", "screen": tui.visible})
+        tui.send("\x1b", "return from the MCP tools view")
+        tui.wait_for(
+            lambda screen: "MCP Server" in screen or "Manage MCP servers" in screen,
+            "the MCP server detail or inventory",
+        )
+        if "Manage MCP servers" not in tui.visible:
+            tui.send("\x1b", "return to the MCP inventory")
+            tui.wait_for(
+                lambda screen: "Manage MCP servers" in screen,
+                "Claude's MCP inventory",
+            )
+    tui.send("\x1b", "close Claude's MCP menu")
+    tui.wait_for(
+        lambda screen: (
+            "Manage MCP servers" not in screen and re.search(r"(?m)^\s*[❯›>]\s*(?!\d+[.)])", screen)
+        ),
+        "Claude's prompt after closing the MCP menu",
+    )
+    return inventory
+
+
+def open_codex_mcp_inventory(tui, services: dict[str, str]) -> str:
+    output_start = len(tui.output)
+    tui.submit("/mcp verbose")
+    for service, tool in services.items():
+        name = re.escape(service.replace(".", "-"))
+        tui.wait_for(
+            lambda screen, name=name, tool=tool: re.search(
+                rf"(?m)^\s*• {name}: connected \(1 tool\)\s*\n"
+                rf"\s*• Auth: [^\n]*\n"
+                rf"\s*• Tools: {re.escape(tool)}\s*$",
+                screen,
+            ),
+            f"{service} connected with {tool} loaded in Codex's MCP inventory",
+            timeout=120,
+        )
+    screen = pyte.HistoryScreen(140, 60, history=1000)
+    pyte.Stream(screen).feed("".join(tui.output[output_start:]))
+    assert len(screen.history.top) < screen.history.size, "Codex's MCP inventory was truncated"
+    history = [
+        "".join(line[column].data for column in range(screen.columns))
+        for line in screen.history.top
+    ]
+    inventory = "\n".join([*history, *screen.display])
+    assert "MCP Tools" in inventory, inventory
+    tui.actions.append({"reason": "codex-mcp-inventory", "screen": inventory})
+    return inventory