Add capability introspection REST endpoints

Adds operator-only endpoints for discovering organism capabilities:
- GET /api/v1/capabilities - list all listeners
- GET /api/v1/capabilities/{name} - detailed info with schema/example

These are REST-only for operators. Agents cannot access them -
they only know their declared peers (peer constraint isolation).

10 new tests for introspection functionality.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

tests/test_server.py

@@ -42,12 +42,15 @@ class MockListener:
     peers: List[str] = None
     payload_class: type = None
     schema: Any = None
+    root_tag: str = ""
 
     def __post_init__(self):
         if self.peers is None:
             self.peers = []
         if self.payload_class is None:
             self.payload_class = type("MockPayload", (), {"__module__": "test", "__name__": "MockPayload"})
+        if not self.root_tag:
+            self.root_tag = f"{self.name.lower()}.mockpayload"
 
 
 @dataclass
@@ -508,3 +511,108 @@ class TestWebSocket:
             data = websocket.receive_json()
             assert data["event"] == "subscribed"
             assert "filter" in data
+
+
+# ============================================================================
+# Test Capability Introspection
+# ============================================================================
+
+class TestCapabilityIntrospection:
+    """Test capability introspection endpoints."""
+
+    def test_list_capabilities(self, test_client):
+        """Test GET /capabilities lists all registered listeners."""
+        response = test_client.get("/api/v1/capabilities")
+        assert response.status_code == 200
+
+        data = response.json()
+        assert "capabilities" in data
+        assert "count" in data
+        assert data["count"] == 2  # greeter and shouter
+
+        names = [c["name"] for c in data["capabilities"]]
+        assert "greeter" in names
+        assert "shouter" in names
+
+    def test_list_capabilities_includes_details(self, test_client):
+        """Test capability listing includes required fields."""
+        response = test_client.get("/api/v1/capabilities")
+        data = response.json()
+
+        # Find greeter
+        greeter = next(c for c in data["capabilities"] if c["name"] == "greeter")
+
+        assert "description" in greeter
+        assert "isAgent" in greeter
+        assert greeter["isAgent"] is True
+        assert "peers" in greeter
+        assert "shouter" in greeter["peers"]
+        assert "rootTag" in greeter
+
+    def test_list_capabilities_sorted(self, test_client):
+        """Test capabilities are sorted by name."""
+        response = test_client.get("/api/v1/capabilities")
+        data = response.json()
+
+        names = [c["name"] for c in data["capabilities"]]
+        assert names == sorted(names)
+
+    def test_get_capability_detail(self, test_client):
+        """Test GET /capabilities/{name} returns detailed info."""
+        response = test_client.get("/api/v1/capabilities/greeter")
+        assert response.status_code == 200
+
+        data = response.json()
+        assert data["name"] == "greeter"
+        assert data["description"] == "Greeting agent"
+        assert data["isAgent"] is True
+        assert "shouter" in data["peers"]
+        assert "payloadClass" in data
+        assert "rootTag" in data
+
+    def test_get_capability_includes_example(self, test_client):
+        """Test capability detail includes example XML."""
+        response = test_client.get("/api/v1/capabilities/greeter")
+        data = response.json()
+
+        # Example XML may or may not be present depending on payload class
+        assert "exampleXml" in data
+
+    def test_get_capability_not_found(self, test_client):
+        """Test GET /capabilities/{name} returns 404 for unknown capability."""
+        response = test_client.get("/api/v1/capabilities/nonexistent")
+        assert response.status_code == 404
+
+    def test_get_capability_shouter(self, test_client):
+        """Test non-agent capability details."""
+        response = test_client.get("/api/v1/capabilities/shouter")
+        assert response.status_code == 200
+
+        data = response.json()
+        assert data["name"] == "shouter"
+        assert data["isAgent"] is False
+        assert data["peers"] == []
+
+
+class TestCapabilityIntrospectionState:
+    """Test capability introspection via ServerState."""
+
+    def test_get_capabilities_returns_list(self, server_state):
+        """Test get_capabilities returns CapabilityInfo list."""
+        capabilities = server_state.get_capabilities()
+        assert len(capabilities) == 2
+        assert all(hasattr(c, "name") for c in capabilities)
+        assert all(hasattr(c, "root_tag") for c in capabilities)
+
+    def test_get_capability_returns_detail(self, server_state):
+        """Test get_capability returns CapabilityDetail."""
+        detail = server_state.get_capability("greeter")
+        assert detail is not None
+        assert detail.name == "greeter"
+        assert detail.is_agent is True
+        assert "shouter" in detail.peers
+
+    def test_get_capability_not_found_returns_none(self, server_state):
+        """Test get_capability returns None for unknown."""
+        detail = server_state.get_capability("nonexistent")
+        assert detail is None

xml_pipeline/server/api.py

@@ -18,6 +18,9 @@ from fastapi import APIRouter, HTTPException, Query
 from xml_pipeline.server.models import (
     AgentInfo,
     AgentListResponse,
+    CapabilityDetail,
+    CapabilityInfo,
+    CapabilityListResponse,
     ErrorResponse,
     InjectRequest,
     InjectResponse,
@@ -50,6 +53,39 @@ def create_router(state: "ServerState") -> APIRouter:
         """Get sanitized organism configuration (no secrets)."""
         return state.get_organism_config()
 
+    # =========================================================================
+    # Capability Introspection Endpoints (for operators, not agents)
+    # =========================================================================
+
+    @router.get("/capabilities", response_model=CapabilityListResponse)
+    async def list_capabilities() -> CapabilityListResponse:
+        """
+        List all registered capabilities in the organism.
+
+        This endpoint is for operator introspection only.
+        Agents cannot access this - they only know their declared peers.
+        """
+        capabilities = state.get_capabilities()
+        return CapabilityListResponse(
+            capabilities=capabilities,
+            count=len(capabilities),
+        )
+
+    @router.get("/capabilities/{name}", response_model=CapabilityDetail)
+    async def get_capability(name: str) -> CapabilityDetail:
+        """
+        Get detailed capability info including schema and example.
+
+        This endpoint is for operator introspection only.
+        """
+        capability = state.get_capability(name)
+        if capability is None:
+            raise HTTPException(
+                status_code=404,
+                detail=f"Capability not found: {name}",
+            )
+        return capability
+
     # =========================================================================
     # Agent Endpoints
     # =========================================================================

xml_pipeline/server/models.py

@@ -259,3 +259,38 @@ class ErrorResponse(CamelModel):
 
     error: str
     detail: Optional[str] = None
+
+
+# =============================================================================
+# Capability/Introspection Models
+# =============================================================================
+
+
+class CapabilityInfo(CamelModel):
+    """Basic capability info for listing."""
+
+    name: str
+    description: str
+    is_agent: bool = Field(alias="isAgent")
+    peers: List[str] = Field(default_factory=list)
+    root_tag: str = Field(alias="rootTag")
+
+
+class CapabilityDetail(CamelModel):
+    """Detailed capability info including schema."""
+
+    name: str
+    description: str
+    is_agent: bool = Field(alias="isAgent")
+    peers: List[str] = Field(default_factory=list)
+    root_tag: str = Field(alias="rootTag")
+    payload_class: str = Field(alias="payloadClass")
+    schema_xsd: Optional[str] = Field(None, alias="schemaXsd")
+    example_xml: Optional[str] = Field(None, alias="exampleXml")
+
+
+class CapabilityListResponse(CamelModel):
+    """Response for GET /capabilities."""
+
+    capabilities: List[CapabilityInfo]
+    count: int

xml_pipeline/server/state.py

@@ -21,6 +21,8 @@ from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set
 from xml_pipeline.server.models import (
     AgentInfo,
     AgentState,
+    CapabilityDetail,
+    CapabilityInfo,
     MessageInfo,
     OrganismInfo,
     OrganismStatus,
@@ -415,6 +417,77 @@ class ServerState:
             return etree.tostring(listener.schema, encoding="unicode", pretty_print=True)
         return None
 
+    # =========================================================================
+    # Capability Introspection (for operators, not agents)
+    # =========================================================================
+
+    def get_capabilities(self) -> List[CapabilityInfo]:
+        """
+        List all registered capabilities.
+
+        This is for operator introspection via REST API.
+        Agents cannot access this - they only know their peers.
+        """
+        capabilities = []
+        for name, listener in self.pump.listeners.items():
+            capabilities.append(
+                CapabilityInfo(
+                    name=name,
+                    description=listener.description,
+                    is_agent=listener.is_agent,
+                    peers=list(listener.peers),
+                    root_tag=listener.root_tag,
+                )
+            )
+        # Sort by name for consistent ordering
+        capabilities.sort(key=lambda c: c.name)
+        return capabilities
+
+    def get_capability(self, name: str) -> Optional[CapabilityDetail]:
+        """
+        Get detailed capability info including schema.
+
+        This is for operator introspection via REST API.
+        """
+        listener = self.pump.listeners.get(name)
+        if listener is None:
+            return None
+
+        # Get XSD schema
+        schema_xsd = None
+        if listener.schema is not None:
+            from lxml import etree
+            try:
+                # Get the schema document from the XMLSchema object
+                # XMLSchema wraps an Element, we need to serialize it
+                schema_xsd = etree.tostring(
+                    listener.schema, encoding="unicode", pretty_print=True
+                )
+            except Exception:
+                pass
+
+        # Generate example XML
+        example_xml = None
+        if hasattr(listener.payload_class, '__dataclass_fields__'):
+            fields = listener.payload_class.__dataclass_fields__
+            class_name = listener.payload_class.__name__
+            lines = [f"<{class_name}>"]
+            for fname in fields:
+                lines.append(f"  <{fname}>...</{fname}>")
+            lines.append(f"</{class_name}>")
+            example_xml = "\n".join(lines)
+
+        return CapabilityDetail(
+            name=name,
+            description=listener.description,
+            is_agent=listener.is_agent,
+            peers=list(listener.peers),
+            root_tag=listener.root_tag,
+            payload_class=f"{listener.payload_class.__module__}.{listener.payload_class.__name__}",
+            schema_xsd=schema_xsd,
+            example_xml=example_xml,
+        )
+
     def get_threads(
         self,
         status: Optional[ThreadStatus] = None,