Nexum-Expanded/runtime/tests/test_protocols.py

from __future__ import annotations

import json
import subprocess
from pathlib import Path
import tomllib

import pytest

from nexum_core.api import NEXUM_CORE_VERSION
from nexum_runtime.capabilities import capability_catalog
from nexum_runtime.executor import execute_tool_call
from nexum_runtime.protocols import a2a_request, agent_card, mcp_request
from nexum_runtime.tooling.contracts import ToolCall


def test_protocol_version_matches_release_package() -> None:
    pyproject = Path(__file__).resolve().parents[1] / "pyproject.toml"
    project = tomllib.loads(pyproject.read_text(encoding="utf-8"))["project"]

    assert NEXUM_CORE_VERSION == project["version"]


def test_capability_catalog_exposes_general_capabilities() -> None:
    catalog = capability_catalog()
    assert catalog["count"] >= 26
    identifiers = [row["capability_id"] for row in catalog["capabilities"]]
    assert len(identifiers) == len(set(identifiers))
    assert "self-correction" in identifiers
    assert "self-improvement" in identifiers
    assert "portable-connectors" in identifiers
    assert "language-packs" in identifiers
    rows = {
        row["capability_id"]: row for row in catalog["capabilities"]
    }
    assert "without a configured topology limit" in rows["self-improvement"][
        "description"
    ]
    assert "model" in rows["parallel-dependencies"]["surfaces"]


def test_mcp_initialization_listing_and_real_tool_call(tmp_path: Path) -> None:
    initialized = mcp_request(
        {"jsonrpc": "2.0", "id": 1, "method": "initialize"},
        workspace=str(tmp_path),
        session_id="mcp-session",
        tools_enabled=True,
    )
    assert initialized["result"]["serverInfo"]["name"] == "Nexum NNF X"
    assert initialized["result"]["serverInfo"]["version"] == NEXUM_CORE_VERSION
    listed = mcp_request(
        {"jsonrpc": "2.0", "id": 2, "method": "tools/list"},
        workspace=str(tmp_path),
        session_id="mcp-session",
        tools_enabled=True,
    )
    names = {row["name"] for row in listed["result"]["tools"]}
    assert {
        "Read",
        "GitStatus",
        "TaskStart",
        "BrowserNavigate",
        "ReproductionRun",
        "TriageCreate",
        "DisclosureCreate",
        "PatchApply",
    } <= names
    assert "AgentDelegate" not in names

    (tmp_path / "evidence.txt").write_text("real-mcp-result", encoding="utf-8")
    called = mcp_request(
        {
            "jsonrpc": "2.0",
            "id": 3,
            "method": "tools/call",
            "params": {
                "name": "Read",
                "arguments": {"path": "evidence.txt"},
                "call_id": "mcp-read",
            },
        },
        workspace=str(tmp_path),
        session_id="mcp-session",
        tools_enabled=True,
    )
    structured = called["result"]["structuredContent"]
    assert structured["ok"] is True
    assert structured["output"] == "real-mcp-result"

    with pytest.raises(ValueError, match="caller-owned"):
        mcp_request(
            {
                "jsonrpc": "2.0",
                "id": 4,
                "method": "tools/call",
                "params": {
                    "name": "AgentDelegate",
                    "arguments": {
                        "agent": "reviewer",
                        "objective": "Review the change",
                    },
                },
            },
            workspace=str(tmp_path),
            session_id="mcp-session",
            tools_enabled=True,
        )


def test_repository_connector_runs_through_common_executor(tmp_path: Path) -> None:
    subprocess.run(["git", "init", "-q"], cwd=tmp_path, check=True)
    (tmp_path / "tracked.txt").write_text("value", encoding="utf-8")
    result = execute_tool_call(
        ToolCall(
            name="GitStatus",
            args={"repository": "."},
            raw="",
            call_id="git-status",
        ),
        cwd=str(tmp_path),
        session_id="repository-session",
    )
    assert result.ok is True
    assert "tracked.txt" in result.output
    assert result.artifact_id.startswith("art_")


def test_agent_card_and_delegated_request_use_shared_harness_result() -> None:
    card = agent_card("https://nexum.example")
    assert card["url"] == "https://nexum.example/a2a"
    assert card["version"] == NEXUM_CORE_VERSION
    assert card["capabilities"]["streaming"] is False
    assert len(card["skills"]) > 20

    received: list[dict[str, object]] = []

    def run_agent(payload: dict[str, object]) -> dict[str, object]:
        received.append(payload)
        return {
            "message": "completed",
            "final_text": "delegated result",
            "metadata": {"session_id": payload["session_id"]},
        }

    response = a2a_request(
        {
            "jsonrpc": "2.0",
            "id": "request-1",
            "method": "message/send",
            "params": {
                "contextId": "delegated-session",
                "message": {
                    "parts": [{"kind": "text", "text": "inspect the workspace"}]
                },
            },
        },
        run_agent=run_agent,
    )
    assert received == [
        {"prompt": "inspect the workspace", "session_id": "delegated-session"}
    ]
    artifact = response["result"]["artifacts"][0]
    assert artifact["parts"][0]["text"] == "delegated result"


def test_delegated_request_exposes_pending_model_selected_calls() -> None:
    def run_agent(_payload: dict[str, object]) -> dict[str, object]:
        return {
            "message": "external_tool_handoff",
            "tool_calls": [
                {
                    "id": "call_external",
                    "type": "function",
                    "function": {
                        "name": "ExternalLookup",
                        "arguments": '{"query":"status"}',
                    },
                }
            ],
        }

    payload = {
        "jsonrpc": "2.0",
        "id": "request-2",
        "method": "message/send",
        "params": {
            "contextId": "delegated-session",
            "message": {"parts": [{"kind": "text", "text": "look it up"}]},
        },
    }
    response = a2a_request(payload, run_agent=run_agent)
    assert response["result"]["status"]["state"] == "input-required"
    pending = response["result"]["metadata"]["pending_tool_calls"]
    assert pending[0]["id"] == "call_external"

    payload["method"] = "message/stream"
    with pytest.raises(ValueError, match="unsupported"):
        a2a_request(payload, run_agent=run_agent)


def test_dynamic_tool_is_rediscoverable_after_registration(tmp_path: Path) -> None:
    created = execute_tool_call(
        ToolCall(
            name="CreateTool",
            args={
                "name": "workspace_summary",
                "command": "pwd",
                "description": "Show the active workspace",
            },
            raw="",
            call_id="create-dynamic",
        ),
        cwd=str(tmp_path),
        session_id="dynamic-session",
    )
    assert created.ok is True
    catalog = execute_tool_call(
        ToolCall(
            name="ToolCatalog",
            args={"query": "workspace_summary"},
            raw="",
            call_id="catalog-dynamic",
        ),
        cwd=str(tmp_path),
        session_id="dynamic-session",
    )
    rows = json.loads(catalog.output)["tools"]
    assert rows[0]["name"] == "workspace_summary"
    assert rows[0]["invocation"] == "RunDynamicTool"