Update for pdf_ingest and pdf_create

engine/src/stirling/agents/__init__.py

@@ -1,5 +1,8 @@
 """Agent modules for Stirling AI reasoning flows."""
 
+from collections.abc import Iterable
+
+from ._registry import AgentDescriptor, RegisterableAgent
 from .execution import ExecutionPlanningAgent
 from .orchestrator import OrchestratorAgent
 from .pdf_create import PdfCreateAgent
@@ -8,7 +11,17 @@ from .pdf_questions import PdfQuestionAgent
 from .pdf_review import PdfReviewAgent
 from .user_spec import UserSpecAgent
 
+
+def build_descriptors(agents: Iterable[RegisterableAgent]) -> list[AgentDescriptor]:
+    """The canonical descriptor list driving both orchestrator routing and the MCP
+    manifest. Pass the live agent singletons. Adding an agent means implementing
+    ``describe`` and including its instance in the caller's list.
+    """
+    return [agent.describe() for agent in agents]
+
+
 __all__ = [
+    "AgentDescriptor",
     "ExecutionPlanningAgent",
     "OrchestratorAgent",
     "PdfCreateAgent",
@@ -17,5 +30,7 @@ __all__ = [
     "PdfEditPlanSelection",
     "PdfQuestionAgent",
     "PdfReviewAgent",
+    "RegisterableAgent",
     "UserSpecAgent",
+    "build_descriptors",
 ]

engine/src/stirling/agents/_registry.py

@@ -0,0 +1,91 @@
+"""Single source of truth for how each agent is exposed.
+
+An agent declares one :class:`AgentDescriptor` via :meth:`RegisterableAgent.describe`.
+Two projections are derived from the collected descriptors, so neither has to be
+hand-maintained:
+
+* the **orchestrator** builds its delegate ``ToolOutput`` union and ``resume``
+  dispatch from descriptors whose ``orchestrator`` route is set;
+* the **MCP capabilities manifest** is built from descriptors' ``mcp`` rows.
+
+Adding an agent therefore means implementing ``describe`` and adding the instance
+to ``build_descriptors`` — the orchestrator and the manifest both update for free.
+
+Note: this ``AgentDescriptor`` registry (how an agent is *published*) is unrelated
+to the runtime "capability" toolsets like ``ContradictionCapability`` /
+``RagCapability`` (tools *injected into* an agent run).
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from typing import Literal
+
+from pydantic import BaseModel
+from pydantic_ai.output import ToolOutput
+from pydantic_ai.tools import RunContext
+
+from stirling.contracts import OrchestratorRequest, OrchestratorResponse, SupportedCapability
+from stirling.services import AppRuntime
+
+OrchestrateFn = Callable[[OrchestratorRequest], Awaitable[OrchestratorResponse]]
+
+
+@dataclass(frozen=True)
+class OrchestratorDeps:
+    runtime: AppRuntime
+    request: OrchestratorRequest
+
+
+@dataclass(frozen=True)
+class OrchestratorRoute:
+    """How an agent is exposed to the top-level orchestrator LLM and resume path.
+
+    ``capability`` keys the resume dispatch: the orchestrator re-enters this
+    delegate when a ``resume_with`` of the same value arrives.
+    """
+
+    capability: SupportedCapability
+    tool_name: str
+    tool_description: str
+    orchestrate: OrchestrateFn
+
+    async def _invoke(self, ctx: RunContext[OrchestratorDeps]) -> OrchestratorResponse:
+        return await self.orchestrate(ctx.deps.request)
+
+    def tool_output(self) -> ToolOutput[OrchestratorResponse]:
+        return ToolOutput(self._invoke, name=self.tool_name, description=self.tool_description)
+
+
+@dataclass(frozen=True)
+class McpCapability:
+    """One row in the MCP capabilities manifest the Java MCP server publishes."""
+
+    id: str
+    description: str
+    input_model: type[BaseModel]
+    mode: Literal["sync", "async"]
+    required_scope: str
+    route: str
+
+
+@dataclass(frozen=True)
+class AgentDescriptor:
+    """How one agent is published. ``orchestrator`` set => routable by the
+    top-level orchestrator; ``mcp`` non-empty => exposed in the MCP manifest.
+    The two are independent: an agent may be one, the other, or both."""
+
+    orchestrator: OrchestratorRoute | None = None
+    mcp: tuple[McpCapability, ...] = ()
+
+
+class RegisterableAgent(ABC):
+    """Base for any agent that publishes itself to the orchestrator and/or MCP.
+
+    Enforces a uniform ``describe`` entry point that startup wiring collects via
+    ``build_descriptors``."""
+
+    @abstractmethod
+    def describe(self) -> AgentDescriptor: ...

engine/src/stirling/agents/execution.py

@@ -1,13 +1,32 @@
 from __future__ import annotations
 
+from stirling.agents._registry import AgentDescriptor, McpCapability, RegisterableAgent
 from stirling.contracts import AgentExecutionRequest, CannotContinueExecutionAction, NextExecutionAction
 from stirling.services import AppRuntime
 
 
-class ExecutionPlanningAgent:
+class ExecutionPlanningAgent(RegisterableAgent):
     def __init__(self, runtime: AppRuntime) -> None:
         self.runtime = runtime
 
+    def describe(self) -> AgentDescriptor:
+        # MCP-only: an internal sub-agent the orchestrator never delegates to.
+        return AgentDescriptor(
+            mcp=(
+                McpCapability(
+                    id="agent-next-action",
+                    description=(
+                        "Decide the next execution step for an in-progress agent workflow. Returns a"
+                        " ToolCall, Completed, or CannotContinue action."
+                    ),
+                    input_model=AgentExecutionRequest,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/agents/next-action",
+                ),
+            ),
+        )
+
     async def next_action(self, request: AgentExecutionRequest) -> NextExecutionAction:
         return CannotContinueExecutionAction(
             reason=f"Execution planning is not implemented yet for step {request.current_step_index}."

engine/src/stirling/agents/ledger/agent.py

@@ -29,6 +29,7 @@ from pydantic import BaseModel, Field
 from pydantic_ai import Agent
 from pydantic_ai.exceptions import AgentRunError
 
+from stirling.agents._registry import AgentDescriptor, McpCapability, RegisterableAgent
 from stirling.contracts.ledger import (
     Discrepancy,
     DiscrepancyKind,
@@ -113,7 +114,7 @@ class StatementsResult(BaseModel):
 # ---------------------------------------------------------------------------
 
 
-class MathAuditorAgent:
+class MathAuditorAgent(RegisterableAgent):
     """
     Encapsulates the Ledger Auditor pipeline.
 
@@ -121,6 +122,37 @@ class MathAuditorAgent:
     pre-built Model objects and ModelSettings.
     """
 
+    def describe(self) -> AgentDescriptor:
+        # MCP-only: the orchestrator reaches the math auditor indirectly (an agent
+        # emits a plan with the MATH_AUDITOR_AGENT tool, which Java runs via the
+        # examine/deliberate routes), so there is no orchestrator delegate here.
+        return AgentDescriptor(
+            mcp=(
+                McpCapability(
+                    id="math-audit-examine",
+                    description=(
+                        "Examine a folio manifest of financial / numeric documents and surface the"
+                        " evidence that needs to be checked for arithmetic consistency."
+                    ),
+                    input_model=FolioManifest,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/ai/math-auditor-agent/examine",
+                ),
+                McpCapability(
+                    id="math-audit-deliberate",
+                    description=(
+                        "Render a deliberated verdict on a single piece of evidence the examine step"
+                        " surfaced (does the arithmetic check out, with what caveats)."
+                    ),
+                    input_model=Evidence,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/ai/math-auditor-agent/deliberate",
+                ),
+            ),
+        )
+
     def __init__(self, runtime: AppRuntime) -> None:
         fast_model = runtime.fast_model
         model_settings = runtime.fast_model_settings

engine/src/stirling/agents/orchestrator.py

@@ -1,91 +1,45 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass
-from typing import assert_never
 
 from pydantic_ai import Agent
 from pydantic_ai.output import ToolOutput
 from pydantic_ai.tools import RunContext
 
-from stirling.agents.pdf_create import PdfCreateAgent
-from stirling.agents.pdf_edit import PdfEditAgent
-from stirling.agents.pdf_questions import PdfQuestionAgent
-from stirling.agents.pdf_review import PdfReviewAgent
-from stirling.agents.user_spec import UserSpecAgent
+from stirling.agents._registry import AgentDescriptor, OrchestratorDeps, OrchestratorRoute
 from stirling.contracts import (
-    AgentDraftWorkflowResponse,
     ConvertMarkdownResponse,
     ExtractedTextArtifact,
     OrchestratorRequest,
     OrchestratorResponse,
-    PdfEditResponse,
-    PdfQuestionOrchestrateResponse,
-    PdfReviewOrchestrateResponse,
     SupportedCapability,
     UnsupportedCapabilityResponse,
     format_conversation_history,
     format_file_names,
 )
-from stirling.contracts.pdf_create import PdfCreateOrchestrateResponse
 from stirling.services import AppRuntime
 
 logger = logging.getLogger(__name__)
 
 
-@dataclass(frozen=True)
-class OrchestratorDeps:
-    runtime: AppRuntime
-    request: OrchestratorRequest
-
-
 class OrchestratorAgent:
-    def __init__(self, runtime: AppRuntime) -> None:
+    def __init__(self, runtime: AppRuntime, descriptors: list[AgentDescriptor]) -> None:
         self.runtime = runtime
+        routes = [d.orchestrator for d in descriptors if d.orchestrator is not None]
+        self._delegates_by_capability: dict[SupportedCapability, OrchestratorRoute] = {
+            route.capability: route for route in routes
+        }
         self.agent = Agent(
             model=runtime.fast_model,
             output_type=[
-                ToolOutput(
-                    self.delegate_pdf_edit,
-                    name="delegate_pdf_edit",
-                    description="Delegate requests for PDF modifications and return the PDF edit result.",
-                ),
-                ToolOutput(
-                    self.delegate_pdf_question,
-                    name="delegate_pdf_question",
-                    description="Delegate questions about PDF contents and return the PDF question result.",
-                ),
-                ToolOutput(
-                    self.delegate_user_spec,
-                    name="delegate_user_spec",
-                    description="Delegate requests to create or revise a user agent spec and return the draft result.",
-                ),
-                ToolOutput(
-                    self.delegate_pdf_review,
-                    name="delegate_pdf_review",
-                    description=(
-                        "Delegate requests to review a PDF and leave review comments, notes, or"
-                        " sticky-note annotations on the document itself. Use this when the user"
-                        " wants the PDF returned with comments attached (e.g. 'review this',"
-                        " 'add review comments', 'flag unclear sentences', 'annotate with"
-                        " feedback')."
-                    ),
-                ),
+                *(route.tool_output() for route in routes),
+                # PDF ingest is a canned one-shot (no agent, never resumed), so like
+                # unsupported_capability it lives inline rather than in the registry.
                 ToolOutput(
                     self.delegate_pdf_ingest,
                     name="delegate_pdf_ingest",
                     description=(
-                        "Delegate requests to convert a PDF to Markdown or extract its content as readable text."
-                    ),
-                ),
-                ToolOutput(
-                    self.delegate_pdf_create,
-                    name="delegate_pdf_create",
-                    description=(
-                        "Delegate requests to create a new PDF document from scratch based on a"
-                        " description. Use this when the user wants to generate a new document"
-                        " (e.g. 'create an invoice', 'write a report', 'make a contract',"
-                        " 'draft a letter'). No input file is required."
+                        "Delegate any request to convert a PDF to Markdown or extract its content as readable text."
                     ),
                 ),
                 ToolOutput(
@@ -98,16 +52,7 @@ class OrchestratorAgent:
             system_prompt=(
                 "You are the top-level orchestrator. "
                 "Choose exactly one output function that best handles the request. "
-                "Use delegate_pdf_edit for any requested modification of one or more PDFs. "
-                "Use delegate_pdf_question for questions about the contents of the attached PDFs. "
-                "Use delegate_user_spec for requests to create or define an agent spec. "
-                "Use delegate_pdf_review when the user wants the PDF returned with review"
-                " comments attached — anything like 'review this', 'annotate with comments',"
-                " 'leave feedback on the PDF'. "
-                "Use delegate_pdf_create when the user wants to generate a new document from"
-                " scratch with no input file — invoices, reports, letters, contracts, etc. "
-                "Use delegate_pdf_ingest for any request to convert a PDF to Markdown "
-                "or extract its content as readable text. "
+                "Consult each delegate tool's description and pick the single best fit. "
                 "Use unsupported_capability when the user asks about the assistant itself "
                 "or when none of the other outputs fit; supply a helpful message."
             ),
@@ -132,70 +77,23 @@ class OrchestratorAgent:
         return result.output
 
     async def _resume(self, request: OrchestratorRequest, capability: SupportedCapability) -> OrchestratorResponse:
-        """Fast-path to get back to the correct endpoint without having to call AI.
+        """Fast-path back to the right delegate without consulting the LLM.
 
         Also the entry point for the *multi-turn* flow where a delegate emits a plan with
         ``resume_with`` set — Java runs the plan, captures any tool reports as artifacts, and
-        re-enters via this method so the delegate can digest the reports.
+        re-enters here so the delegate can digest the reports.
         """
-        match capability:
-            case SupportedCapability.PDF_QUESTION:
-                return await self._run_pdf_question(request)
-            case SupportedCapability.PDF_REVIEW:
-                return await self._run_pdf_review(request)
-            case SupportedCapability.PDF_EDIT:
-                return await self._run_pdf_edit(request)
-            case SupportedCapability.AGENT_DRAFT:
-                return await self._run_agent_draft(request)
-            case SupportedCapability.PDF_CREATE:
-                return await self._run_pdf_create(request)
-            case (
-                SupportedCapability.ORCHESTRATE
-                | SupportedCapability.AGENT_REVISE
-                | SupportedCapability.AGENT_NEXT_ACTION
-                | SupportedCapability.MATH_AUDITOR_AGENT
-            ):
-                raise ValueError(f"Cannot resume orchestrator with capability: {capability}")
-            case _ as unreachable:
-                assert_never(unreachable)
-
-    async def delegate_pdf_edit(self, ctx: RunContext[OrchestratorDeps]) -> PdfEditResponse:
-        return await self._run_pdf_edit(ctx.deps.request)
-
-    async def _run_pdf_edit(self, request: OrchestratorRequest) -> PdfEditResponse:
-        return await PdfEditAgent(self.runtime).orchestrate(request)
-
-    async def delegate_pdf_question(self, ctx: RunContext[OrchestratorDeps]) -> PdfQuestionOrchestrateResponse:
-        return await self._run_pdf_question(ctx.deps.request)
-
-    async def _run_pdf_question(self, request: OrchestratorRequest) -> PdfQuestionOrchestrateResponse:
-        return await PdfQuestionAgent(self.runtime).orchestrate(request)
-
-    async def delegate_user_spec(self, ctx: RunContext[OrchestratorDeps]) -> AgentDraftWorkflowResponse:
-        return await self._run_agent_draft(ctx.deps.request)
-
-    async def _run_agent_draft(self, request: OrchestratorRequest) -> AgentDraftWorkflowResponse:
-        return await UserSpecAgent(self.runtime).orchestrate(request)
+        route = self._delegates_by_capability.get(capability)
+        if route is None:
+            raise ValueError(f"Cannot resume orchestrator with capability: {capability}")
+        return await route.orchestrate(request)
 
     async def delegate_pdf_ingest(self, ctx: RunContext[OrchestratorDeps]) -> ConvertMarkdownResponse:
-        request = ctx.deps.request
         return ConvertMarkdownResponse(
             reason="PDF to Markdown requested — Java converts deterministically.",
-            files_to_ingest=request.files,
+            files_to_ingest=ctx.deps.request.files,
         )
 
-    async def delegate_pdf_review(self, ctx: RunContext[OrchestratorDeps]) -> PdfReviewOrchestrateResponse:
-        return await self._run_pdf_review(ctx.deps.request)
-
-    async def _run_pdf_review(self, request: OrchestratorRequest) -> PdfReviewOrchestrateResponse:
-        return await PdfReviewAgent(self.runtime).orchestrate(request)
-
-    async def delegate_pdf_create(self, ctx: RunContext[OrchestratorDeps]) -> PdfCreateOrchestrateResponse:
-        return await self._run_pdf_create(ctx.deps.request)
-
-    async def _run_pdf_create(self, request: OrchestratorRequest) -> PdfCreateOrchestrateResponse:
-        return await PdfCreateAgent(self.runtime).orchestrate(request)
-
     async def unsupported_capability(
         self,
         ctx: RunContext[OrchestratorDeps],

engine/src/stirling/agents/pdf_comment/agent.py

@@ -20,6 +20,7 @@ import logging
 from pydantic import Field
 from pydantic_ai import Agent
 
+from stirling.agents._registry import AgentDescriptor, McpCapability, RegisterableAgent
 from stirling.agents.pdf_comment.prompts import COMMENT_AGENT_SYSTEM_PROMPT
 from stirling.contracts.pdf_comments import (
     MAX_COMMENT_TEXT_LENGTH,
@@ -66,7 +67,7 @@ class LlmCommentOutput(ApiModel):
     rationale: str = Field(max_length=1_000)
 
 
-class PdfCommentAgent:
+class PdfCommentAgent(RegisterableAgent):
     """Encapsulates the single-shot PDF comment generation pipeline.
 
     Instantiated once at app startup with an :class:`AppRuntime`, which
@@ -82,6 +83,22 @@ class PdfCommentAgent:
             model_settings=runtime.fast_model_settings,
         )
 
+    def describe(self) -> AgentDescriptor:
+        # MCP-only: invoked as a tool inside the PDF review plan, never a
+        # top-level orchestrator delegate.
+        return AgentDescriptor(
+            mcp=(
+                McpCapability(
+                    id="pdf-comment-generate",
+                    description="Generate inline review comments for a PDF document.",
+                    input_model=PdfCommentRequest,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/pdf-comment/generate",
+                ),
+            ),
+        )
+
     async def generate(self, request: PdfCommentRequest) -> PdfCommentResponse:
         """Run the agent against a ``PdfCommentRequest`` and return comments.
 

engine/src/stirling/agents/pdf_create/agent.py

@@ -28,10 +28,12 @@ from jinja2 import Environment, FileSystemLoader
 from pydantic_ai import Agent
 from pydantic_ai.output import NativeOutput
 
+from stirling.agents._registry import AgentDescriptor, OrchestratorRoute, RegisterableAgent
 from stirling.contracts import (
     EditCannotDoResponse,
     EditPlanResponse,
     OrchestratorRequest,
+    SupportedCapability,
     ToolOperationStep,
     format_conversation_history,
 )
@@ -317,7 +319,7 @@ def _safe_filename(title: str) -> str:
 # ── Agent ─────────────────────────────────────────────────────────────────────────────────────────
 
 
-class PdfCreateAgent:
+class PdfCreateAgent(RegisterableAgent):
     def __init__(self, runtime: AppRuntime) -> None:
         self.runtime = runtime
         self._jinja_env = _build_jinja_env()
@@ -343,6 +345,21 @@ class PdfCreateAgent:
             model_settings={**runtime.smart_model_settings, "temperature": 0.3},
         )
 
+    def describe(self) -> AgentDescriptor:
+        return AgentDescriptor(
+            orchestrator=OrchestratorRoute(
+                capability=SupportedCapability.PDF_CREATE,
+                tool_name="delegate_pdf_create",
+                tool_description=(
+                    "Delegate requests to create a new PDF document from scratch based on a"
+                    " description. Use this when the user wants to generate a new document"
+                    " (e.g. 'create an invoice', 'write a report', 'make a contract',"
+                    " 'draft a letter'). No input file is required."
+                ),
+                orchestrate=self.orchestrate,
+            ),
+        )
+
     async def orchestrate(self, request: OrchestratorRequest) -> PdfCreateOrchestrateResponse:
         history = format_conversation_history(request.conversation_history)
 

engine/src/stirling/agents/pdf_edit.py

@@ -9,6 +9,7 @@ from pydantic_ai import Agent
 from pydantic_ai.output import NativeOutput
 
 from stirling.agents._page_text import format_page_text, get_extracted_text_artifact, has_page_text
+from stirling.agents._registry import AgentDescriptor, McpCapability, OrchestratorRoute, RegisterableAgent
 from stirling.contracts import (
     EditCannotDoResponse,
     EditClarificationRequest,
@@ -171,11 +172,35 @@ class PdfEditParameterSelector:
         )
 
 
-class PdfEditAgent:
+class PdfEditAgent(RegisterableAgent):
     def __init__(self, runtime: AppRuntime) -> None:
         self.runtime = runtime
         self.parameter_selector = PdfEditParameterSelector(runtime)
 
+    def describe(self) -> AgentDescriptor:
+        return AgentDescriptor(
+            orchestrator=OrchestratorRoute(
+                capability=SupportedCapability.PDF_EDIT,
+                tool_name="delegate_pdf_edit",
+                tool_description="Delegate any requested modification of one or more PDFs; returns the edit result.",
+                orchestrate=self.orchestrate,
+            ),
+            mcp=(
+                McpCapability(
+                    id="pdf-edit-plan",
+                    description=(
+                        "Produce an edit plan (a structured sequence of PDF operations) from a"
+                        " natural-language edit request. The plan is executed by Java through the job"
+                        " pipeline; this capability does not modify files itself."
+                    ),
+                    input_model=PdfEditRequest,
+                    mode="async",
+                    required_scope="mcp.tools.write",
+                    route="/api/v1/pdf-edit",
+                ),
+            ),
+        )
+
     async def orchestrate(self, request: OrchestratorRequest) -> PdfEditResponse:
         """Entry point for the orchestrator delegate — adapts the orchestrator's
         request shape into a :class:`PdfEditRequest` and runs the standard

engine/src/stirling/agents/pdf_questions.py

@@ -5,6 +5,7 @@ import logging
 from pydantic_ai import Agent
 from pydantic_ai.output import NativeOutput
 
+from stirling.agents._registry import AgentDescriptor, McpCapability, OrchestratorRoute, RegisterableAgent
 from stirling.agents.contradiction import ContradictionCapability, ContradictionDetector
 from stirling.agents.math_presentation import MathIntentClassifier, extract_math_verdict
 from stirling.agents.shared import ChunkedReasoner, WholeDocReaderCapability
@@ -85,7 +86,7 @@ _MATH_SYNTH_SYSTEM_PROMPT = (
 )
 
 
-class PdfQuestionAgent:
+class PdfQuestionAgent(RegisterableAgent):
     def __init__(self, runtime: AppRuntime) -> None:
         self.runtime = runtime
         self._math_synth_agent: Agent[None, str] = Agent(
@@ -103,6 +104,26 @@ class PdfQuestionAgent:
         # (mirrors the chunked-reasoner pattern).
         self._contradiction_detector = ContradictionDetector(runtime)
 
+    def describe(self) -> AgentDescriptor:
+        return AgentDescriptor(
+            orchestrator=OrchestratorRoute(
+                capability=SupportedCapability.PDF_QUESTION,
+                tool_name="delegate_pdf_question",
+                tool_description="Delegate questions about the contents of the attached PDFs; returns the answer.",
+                orchestrate=self.orchestrate,
+            ),
+            mcp=(
+                McpCapability(
+                    id="pdf-question-answer",
+                    description="Answer a natural-language question about a PDF document.",
+                    input_model=PdfQuestionRequest,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/pdf-question",
+                ),
+            ),
+        )
+
     async def handle(self, request: PdfQuestionRequest) -> PdfQuestionResponse:
         logger.info(
             "[pdf-question] handle: files=%s question=%r",

engine/src/stirling/agents/pdf_review.py

@@ -31,6 +31,7 @@ from typing import Literal
 from pydantic import Field
 from pydantic_ai import Agent
 
+from stirling.agents._registry import AgentDescriptor, OrchestratorRoute, RegisterableAgent
 from stirling.agents.contradiction import ContradictionDetector, ContradictionIntentClassifier
 from stirling.agents.contradiction.detector import _escape_for_tag
 from stirling.agents.contradiction.prompts import REVIEW_LOCALISER_PROMPT
@@ -104,7 +105,7 @@ class _LocalisedContradictionReport(ApiModel):
     comments: list[_PairedLocalisedContradiction] = Field(default_factory=list)
 
 
-class PdfReviewAgent:
+class PdfReviewAgent(RegisterableAgent):
     def __init__(self, runtime: AppRuntime) -> None:
         self.runtime = runtime
         self._localiser_agent: Agent[None, _LocalisedVerdict] = Agent(
@@ -128,6 +129,21 @@ class PdfReviewAgent:
         # request's stages.
         self._contradiction_detector = ContradictionDetector(runtime)
 
+    def describe(self) -> AgentDescriptor:
+        return AgentDescriptor(
+            orchestrator=OrchestratorRoute(
+                capability=SupportedCapability.PDF_REVIEW,
+                tool_name="delegate_pdf_review",
+                tool_description=(
+                    "Delegate requests to review a PDF and leave review comments, notes, or"
+                    " sticky-note annotations on the document itself. Use this when the user"
+                    " wants the PDF returned with comments attached (e.g. 'review this',"
+                    " 'add review comments', 'flag unclear sentences', 'annotate with feedback')."
+                ),
+                orchestrate=self.orchestrate,
+            ),
+        )
+
     async def orchestrate(self, request: OrchestratorRequest) -> PdfReviewOrchestrateResponse:
         """Entry point for the orchestrator delegate.
 

engine/src/stirling/agents/user_spec.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 from pydantic_ai import Agent
 from pydantic_ai.output import NativeOutput
 
+from stirling.agents._registry import AgentDescriptor, McpCapability, OrchestratorRoute, RegisterableAgent
 from stirling.agents.pdf_edit import PdfEditAgent
 from stirling.contracts import (
     AgentDraft,
@@ -18,6 +19,7 @@ from stirling.contracts import (
     OrchestratorRequest,
     PdfEditRequest,
     PdfEditTerminalResponse,
+    SupportedCapability,
     format_conversation_history,
 )
 from stirling.models import ApiModel
@@ -30,7 +32,7 @@ class UserSpecMetadata(ApiModel):
     objective: str
 
 
-class UserSpecAgent:
+class UserSpecAgent(RegisterableAgent):
     def __init__(self, runtime: AppRuntime) -> None:
         self.runtime = runtime
         self.pdf_edit_agent = PdfEditAgent(runtime)
@@ -45,6 +47,39 @@ class UserSpecAgent:
             model_settings=runtime.smart_model_settings,
         )
 
+    def describe(self) -> AgentDescriptor:
+        return AgentDescriptor(
+            orchestrator=OrchestratorRoute(
+                capability=SupportedCapability.AGENT_DRAFT,
+                tool_name="delegate_user_spec",
+                tool_description="Delegate requests to create or define an agent spec; returns the draft result.",
+                orchestrate=self.orchestrate,
+            ),
+            mcp=(
+                McpCapability(
+                    id="agent-draft",
+                    description=(
+                        "Draft a structured agent specification from a free-text description"
+                        " of the task the user wants automated."
+                    ),
+                    input_model=AgentDraftRequest,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/ai/agents/draft",
+                ),
+                McpCapability(
+                    id="agent-revise",
+                    description=(
+                        "Revise an existing draft agent specification based on user feedback or constraint changes."
+                    ),
+                    input_model=AgentRevisionRequest,
+                    mode="sync",
+                    required_scope="mcp.tools.read",
+                    route="/api/v1/ai/agents/revise",
+                ),
+            ),
+        )
+
     async def orchestrate(self, request: OrchestratorRequest) -> AgentDraftWorkflowResponse:
         """Entry point for the orchestrator delegate — adapts the orchestrator's
         request shape into an :class:`AgentDraftRequest` and runs the standard

engine/src/stirling/api/agent_capabilities.py

@@ -1,160 +1,46 @@
-"""
-Curated registry of agent capabilities the MCP server (Java side) is allowed to publish.
+"""Serialize the MCP capabilities manifest the Java MCP server pulls at boot.
 
-Internal sub-agents (currently only ``ExecutionPlanningAgent`` - it lives behind the orchestrator
-and has no end-user-facing API surface) are intentionally absent. The handoff spec calls for
-"user-facing" capabilities only; revisit this list when adding a new agent and ask whether MCP
-clients should be able to invoke it directly.
+The manifest is *derived* from the agent registry: every agent declares its
+exposed capabilities in ``describe()`` (see ``stirling.agents._registry``), and
+this module flattens the ``mcp`` rows of the startup descriptor list into the
+wire shape Java consumes. There is no separately maintained capability list to
+keep in sync — adding an MCP capability means adding an ``McpCapability`` to the
+owning agent's descriptor.
 
-The Java side pulls ``/api/v1/agents/capabilities`` once at boot and again every few minutes; the
-manifest is the authoritative source for the ``stirling_ai`` MCP tool's operation enum.
+Curation note: exposure is opt-in. An agent is published to MCP only if its
+descriptor carries one or more ``McpCapability`` rows; registering an agent with
+the orchestrator does not auto-expose it over the (OAuth-scoped) MCP surface.
+
+The Java side pulls ``/api/v1/agents/capabilities`` once at boot and again every
+few minutes; the manifest is the authoritative source for the ``stirling_ai`` MCP
+tool's operation enum.
 """
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from collections.abc import Iterable
 from typing import Any
 
-from pydantic import BaseModel
-
-from stirling.contracts import (
-    AgentDraftRequest,
-    AgentExecutionRequest,
-    AgentRevisionRequest,
-    Evidence,
-    FolioManifest,
-    PdfCommentRequest,
-    PdfEditRequest,
-    PdfQuestionRequest,
-)
+from stirling.agents import AgentDescriptor
 
 
-@dataclass(frozen=True)
-class AgentCapability:
-    """One row in the curated manifest.
+def manifest_payload(descriptors: Iterable[AgentDescriptor]) -> dict[str, Any]:
+    """Flatten the ``mcp`` rows of the descriptor list to the wire shape.
 
-    Attributes:
-        id: stable capability identifier (used as the operation enum value in
-            ``stirling_ai``). Avoid renaming - clients persist these.
-        description: one-line human-friendly summary shown inside MCP tool descriptions.
-        input_model: Pydantic class whose JSON Schema becomes the capability's
-            ``input_schema``. Auto-derived; do not hand-write schemas.
-        mode: ``"sync"`` if the capability returns content inline, ``"async"`` if it returns a
-            plan that Java executes via the job pipeline.
-        required_scope: coarse OAuth scope. ``mcp.tools.read`` for pure-read capabilities
-            (Q&A, audits) and ``mcp.tools.write`` for anything that yields a plan / file.
-        route: HTTP path Java POSTs to when invoking this capability. When a capability does
-            not have a stable per-agent route yet, use the generic invoke fallback at
-            ``/api/v1/agents/invoke/{id}``.
-    """
-
-    id: str
-    description: str
-    input_model: type[BaseModel]
-    mode: str
-    required_scope: str
-    route: str
-
-
-EXPOSED_CAPABILITIES: list[AgentCapability] = [
-    AgentCapability(
-        id="pdf-question-answer",
-        description="Answer a natural-language question about a PDF document.",
-        input_model=PdfQuestionRequest,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/pdf-question",
-    ),
-    AgentCapability(
-        id="pdf-edit-plan",
-        description=(
-            "Produce an edit plan (a structured sequence of PDF operations) from a"
-            " natural-language edit request. The plan is executed by Java through the job"
-            " pipeline; this capability does not modify files itself."
-        ),
-        input_model=PdfEditRequest,
-        mode="async",
-        required_scope="mcp.tools.write",
-        route="/api/v1/pdf-edit",
-    ),
-    AgentCapability(
-        id="agent-draft",
-        description=(
-            "Draft a structured agent specification from a free-text description of the task the user wants automated."
-        ),
-        input_model=AgentDraftRequest,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/ai/agents/draft",
-    ),
-    AgentCapability(
-        id="agent-revise",
-        description=("Revise an existing draft agent specification based on user feedback or constraint changes."),
-        input_model=AgentRevisionRequest,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/ai/agents/revise",
-    ),
-    AgentCapability(
-        id="math-audit-examine",
-        description=(
-            "Examine a folio manifest of financial / numeric documents and surface the"
-            " evidence that needs to be checked for arithmetic consistency."
-        ),
-        input_model=FolioManifest,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/ai/math-auditor-agent/examine",
-    ),
-    AgentCapability(
-        id="math-audit-deliberate",
-        description=(
-            "Render a deliberated verdict on a single piece of evidence the examine step"
-            " surfaced (does the arithmetic check out, with what caveats)."
-        ),
-        input_model=Evidence,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/ai/math-auditor-agent/deliberate",
-    ),
-    AgentCapability(
-        id="pdf-comment-generate",
-        description="Generate inline review comments for a PDF document.",
-        input_model=PdfCommentRequest,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/pdf-comment/generate",
-    ),
-    AgentCapability(
-        id="agent-next-action",
-        description=(
-            "Decide the next execution step for an in-progress agent workflow. Returns a"
-            " ToolCall, Completed, or CannotContinue action."
-        ),
-        input_model=AgentExecutionRequest,
-        mode="sync",
-        required_scope="mcp.tools.read",
-        route="/api/v1/agents/next-action",
-    ),
-]
-
-
-def manifest_payload() -> dict[str, Any]:
-    """Serialize the curated registry to the wire shape consumed by Java.
-
-    Schema is derived from ``input_model.model_json_schema()`` so we never hand-write JSON
-    Schema - the Pydantic model is the single source of truth.
+    Schema is derived from ``input_model.model_json_schema()`` so we never
+    hand-write JSON Schema - the Pydantic model is the single source of truth.
     """
     items: list[dict[str, Any]] = []
-    for cap in EXPOSED_CAPABILITIES:
-        items.append(
-            {
-                "id": cap.id,
-                "description": cap.description,
-                "input_schema": cap.input_model.model_json_schema(),
-                "mode": cap.mode,
-                "required_scope": cap.required_scope,
-                "route": cap.route,
-            }
-        )
+    for descriptor in descriptors:
+        for cap in descriptor.mcp:
+            items.append(
+                {
+                    "id": cap.id,
+                    "description": cap.description,
+                    "input_schema": cap.input_model.model_json_schema(),
+                    "mode": cap.mode,
+                    "required_scope": cap.required_scope,
+                    "route": cap.route,
+                }
+            )
     return {"version": 1, "capabilities": items}

engine/src/stirling/api/app.py

@@ -12,9 +12,12 @@ from pydantic_ai.models.instrumented import InstrumentationSettings
 from stirling.agents import (
     ExecutionPlanningAgent,
     OrchestratorAgent,
+    PdfCreateAgent,
     PdfEditAgent,
     PdfQuestionAgent,
+    PdfReviewAgent,
     UserSpecAgent,
+    build_descriptors,
 )
 from stirling.agents.ledger import MathAuditorAgent
 from stirling.agents.pdf_comment import PdfCommentAgent
@@ -88,13 +91,28 @@ async def lifespan(fast_api: FastAPI):
     runtime = build_runtime(settings)
     fast_api.state.settings = settings
     fast_api.state.runtime = runtime
-    fast_api.state.orchestrator_agent = OrchestratorAgent(runtime)
     fast_api.state.pdf_edit_agent = PdfEditAgent(runtime)
     fast_api.state.pdf_question_agent = PdfQuestionAgent(runtime)
+    fast_api.state.pdf_review_agent = PdfReviewAgent(runtime)
+    fast_api.state.pdf_create_agent = PdfCreateAgent(runtime)
     fast_api.state.user_spec_agent = UserSpecAgent(runtime)
     fast_api.state.execution_planning_agent = ExecutionPlanningAgent(runtime)
     fast_api.state.math_auditor_agent = MathAuditorAgent(runtime)
     fast_api.state.pdf_comment_agent = PdfCommentAgent(runtime)
+    # One descriptor list drives both orchestrator routing and the MCP manifest.
+    fast_api.state.agent_descriptors = build_descriptors(
+        [
+            fast_api.state.pdf_edit_agent,
+            fast_api.state.pdf_question_agent,
+            fast_api.state.user_spec_agent,
+            fast_api.state.pdf_review_agent,
+            fast_api.state.pdf_create_agent,
+            fast_api.state.pdf_comment_agent,
+            fast_api.state.math_auditor_agent,
+            fast_api.state.execution_planning_agent,
+        ]
+    )
+    fast_api.state.orchestrator_agent = OrchestratorAgent(runtime, fast_api.state.agent_descriptors)
     tracer_provider = setup_posthog_tracking(settings)
     if tracer_provider:
         Agent.instrument_all(InstrumentationSettings(tracer_provider=tracer_provider))

engine/src/stirling/api/routes/agent_capabilities.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from typing import Any
 
-from fastapi import APIRouter
+from fastapi import APIRouter, Request
 
 from stirling.api.agent_capabilities import manifest_payload
 
@@ -12,11 +12,11 @@ router = APIRouter(prefix="/api/v1/agents", tags=["agents"])
 
 
 @router.get("/capabilities")
-def get_capabilities() -> dict[str, Any]:
-    """Return the curated agent capabilities manifest.
+def get_capabilities(request: Request) -> dict[str, Any]:
+    """Return the agent capabilities manifest, derived from the startup registry.
 
     Gated by ``EngineSharedSecretMiddleware`` when the ``STIRLING_ENGINE_SHARED_SECRET`` env var
     is configured. In dev/local mode (no secret set), the endpoint is open - the engine binds to
     localhost only by default, so this is acceptable while iterating.
     """
-    return manifest_payload()
+    return manifest_payload(request.app.state.agent_descriptors)

engine/tests/agents/test_orchestrator_pdf_comment.py

@@ -1,11 +1,10 @@
 """
-Orchestrator ``delegate_pdf_review`` contract test.
+PDF-review delegate contract test.
 
-The real orchestrator delegates PDF-review requests via a pydantic-ai tool
-output. Exercising the full ``agent.run(...)`` call would hit the LLM and
-requires building a real ``RunContext`` — so instead this test invokes
-``delegate_pdf_review`` directly with a minimal ``deps`` stand-in. That's
-enough to verify the wire contract the orchestrator produces:
+The orchestrator routes PDF-review requests to ``PdfReviewAgent.orchestrate``
+(the orchestrator merely selects the delegate; the review logic lives on the
+agent). Exercising the agent directly avoids the LLM routing call and verifies
+the wire contract the delegate produces for a plain prose-review request:
 
 * it returns an ``EditPlanResponse``;
 * with exactly one step;
@@ -16,13 +15,11 @@ enough to verify the wire contract the orchestrator produces:
 
 from __future__ import annotations
 
-from dataclasses import dataclass
-from types import SimpleNamespace
 from unittest.mock import AsyncMock, patch
 
 import pytest
 
-from stirling.agents import OrchestratorAgent
+from stirling.agents import PdfReviewAgent
 from stirling.contracts import AiFile, OrchestratorRequest
 from stirling.contracts.pdf_edit import EditPlanResponse
 from stirling.models import FileId
@@ -30,27 +27,28 @@ from stirling.models.agent_tool_models import AgentToolId, PdfCommentAgentParams
 from stirling.services.runtime import AppRuntime
 
 
-@dataclass(frozen=True)
-class _FakeDeps:
-    request: OrchestratorRequest
-
-
 @pytest.mark.anyio
-async def test_delegate_pdf_review_wires_prompt_to_tool_step(runtime: AppRuntime) -> None:
-    orchestrator = OrchestratorAgent(runtime)
+async def test_pdf_review_wires_prompt_to_tool_step(runtime: AppRuntime) -> None:
+    review_agent = PdfReviewAgent(runtime)
     request = OrchestratorRequest(
         user_message="please add review comments flagging ambiguous dates",
         files=[AiFile(id=FileId("contract-id"), name="contract.pdf")],
     )
-    ctx = SimpleNamespace(deps=_FakeDeps(request=request))
 
-    # PdfReviewAgent now classifies math intent locally via a tiny LLM. Stub it
-    # to false so this test stays focused on the prose-review wire contract.
-    with patch(
-        "stirling.agents.pdf_review.MathIntentClassifier.classify",
-        new=AsyncMock(return_value=False),
+    # PdfReviewAgent classifies math and contradiction intent locally via tiny
+    # LLMs. Stub both to false so this test stays focused on the prose-review
+    # wire contract.
+    with (
+        patch(
+            "stirling.agents.pdf_review.MathIntentClassifier.classify",
+            new=AsyncMock(return_value=False),
+        ),
+        patch(
+            "stirling.agents.pdf_review.ContradictionIntentClassifier.classify",
+            new=AsyncMock(return_value=False),
+        ),
     ):
-        response = await orchestrator.delegate_pdf_review(ctx)  # type: ignore[arg-type]
+        response = await review_agent.orchestrate(request)
 
     assert isinstance(response, EditPlanResponse)
     assert len(response.steps) == 1

engine/tests/agents/test_orchestrator_routing.py

@@ -0,0 +1,144 @@
+"""Behavioural lock: a scripted top-level tool call reaches the right delegate.
+
+No real LLM — a :class:`FunctionModel` scripts the exact tool call the orchestrator
+would have received, and we assert which delegate handled it. Built on the real
+descriptor list (via ``build_descriptors``) with each agent's ``orchestrate``
+swapped for a recording spy, so the test stays honest to whatever agents are
+actually registered.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import replace
+
+import pytest
+from pydantic_ai.messages import ModelMessage, ModelResponse, ToolCallPart
+from pydantic_ai.models.function import AgentInfo, FunctionModel
+
+from stirling.agents import OrchestratorAgent, build_descriptors
+from stirling.agents._registry import AgentDescriptor, OrchestratorRoute, RegisterableAgent
+from stirling.contracts import (
+    ConvertMarkdownResponse,
+    EditCannotDoResponse,
+    EditPlanResponse,
+    OrchestratorRequest,
+    OrchestratorResponse,
+    PdfQuestionNotFoundResponse,
+    SupportedCapability,
+)
+from stirling.services.runtime import AppRuntime
+
+_REACHED: list[SupportedCapability] = []
+
+
+class _SpyAgent(RegisterableAgent):
+    """Stand-in agent that reproduces a real delegate's tool surface but records
+    the reach and returns a fixed sentinel instead of doing work."""
+
+    def __init__(
+        self,
+        capability: SupportedCapability,
+        tool_name: str,
+        response: OrchestratorResponse,
+    ) -> None:
+        self._capability = capability
+        self._tool_name = tool_name
+        self._response = response
+
+    def describe(self) -> AgentDescriptor:
+        return AgentDescriptor(
+            orchestrator=OrchestratorRoute(
+                capability=self._capability,
+                tool_name=self._tool_name,
+                tool_description=f"spy for {self._tool_name}",
+                orchestrate=self._orchestrate,
+            ),
+        )
+
+    async def _orchestrate(self, _request: OrchestratorRequest) -> OrchestratorResponse:
+        _REACHED.append(self._capability)
+        return self._response
+
+
+def _spies() -> list[RegisterableAgent]:
+    return [
+        _SpyAgent(SupportedCapability.PDF_EDIT, "delegate_pdf_edit", EditCannotDoResponse(reason="spy")),
+        _SpyAgent(
+            SupportedCapability.PDF_QUESTION,
+            "delegate_pdf_question",
+            PdfQuestionNotFoundResponse(reason="spy"),
+        ),
+        _SpyAgent(SupportedCapability.PDF_REVIEW, "delegate_pdf_review", EditPlanResponse(summary="", steps=[])),
+        _SpyAgent(SupportedCapability.PDF_CREATE, "delegate_pdf_create", EditPlanResponse(summary="", steps=[])),
+    ]
+
+
+def _script(tool_name: str) -> Callable[[list[ModelMessage], AgentInfo], ModelResponse]:
+    def call(_messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
+        return ModelResponse(parts=[ToolCallPart(tool_name=tool_name, args={})])
+
+    return call
+
+
+async def _route(runtime: AppRuntime, tool_name: str) -> OrchestratorResponse:
+    _REACHED.clear()
+    scripted = replace(runtime, fast_model=FunctionModel(_script(tool_name)))
+    orchestrator = OrchestratorAgent(scripted, build_descriptors(_spies()))
+    return await orchestrator.handle(OrchestratorRequest(user_message="x"))
+
+
+@pytest.mark.anyio
+async def test_delegate_pdf_edit_reaches_edit_delegate(runtime: AppRuntime) -> None:
+    response = await _route(runtime, "delegate_pdf_edit")
+    assert _REACHED == [SupportedCapability.PDF_EDIT]
+    assert isinstance(response, EditCannotDoResponse)
+
+
+@pytest.mark.anyio
+async def test_delegate_pdf_question_reaches_question_delegate(runtime: AppRuntime) -> None:
+    response = await _route(runtime, "delegate_pdf_question")
+    assert _REACHED == [SupportedCapability.PDF_QUESTION]
+    assert isinstance(response, PdfQuestionNotFoundResponse)
+
+
+@pytest.mark.anyio
+async def test_delegate_pdf_review_reaches_review_delegate(runtime: AppRuntime) -> None:
+    response = await _route(runtime, "delegate_pdf_review")
+    assert _REACHED == [SupportedCapability.PDF_REVIEW]
+    assert isinstance(response, EditPlanResponse)
+
+
+@pytest.mark.anyio
+async def test_delegate_pdf_create_reaches_create_delegate(runtime: AppRuntime) -> None:
+    response = await _route(runtime, "delegate_pdf_create")
+    assert _REACHED == [SupportedCapability.PDF_CREATE]
+    assert isinstance(response, EditPlanResponse)
+
+
+@pytest.mark.anyio
+async def test_delegate_pdf_ingest_returns_convert_markdown(runtime: AppRuntime) -> None:
+    # pdf_ingest is a canned inline orchestrator tool (no agent, never resumed) —
+    # no reach recorded, just a deterministic convert response.
+    response = await _route(runtime, "delegate_pdf_ingest")
+    assert _REACHED == []
+    assert isinstance(response, ConvertMarkdownResponse)
+
+
+@pytest.mark.anyio
+async def test_resume_dispatches_to_matching_delegate(runtime: AppRuntime) -> None:
+    _REACHED.clear()
+    orchestrator = OrchestratorAgent(runtime, build_descriptors(_spies()))
+    await orchestrator.handle(OrchestratorRequest(user_message="x", resume_with=SupportedCapability.PDF_REVIEW))
+    assert _REACHED == [SupportedCapability.PDF_REVIEW]
+
+
+@pytest.mark.anyio
+async def test_resume_with_unroutable_capability_raises(runtime: AppRuntime) -> None:
+    # MATH_AUDITOR_AGENT is MCP-only — not an orchestrator delegate, so it has no
+    # resumable route and resuming into it must raise.
+    orchestrator = OrchestratorAgent(runtime, build_descriptors(_spies()))
+    with pytest.raises(ValueError, match="Cannot resume"):
+        await orchestrator.handle(
+            OrchestratorRequest(user_message="x", resume_with=SupportedCapability.MATH_AUDITOR_AGENT)
+        )

engine/tests/test_agent_capabilities.py

@@ -0,0 +1,154 @@
+"""Lock the MCP capabilities manifest wire shape.
+
+Exercises the real ``GET /api/v1/agents/capabilities`` endpoint (so it covers the
+actual startup wiring of ``app.state.agent_descriptors``) and pins every
+capability's id, metadata, and Pydantic-derived input schema. The manifest is
+built by flattening each agent's ``describe()`` rows; this suite is the guard
+that the derived manifest never drifts from the published contract.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from dataclasses import dataclass
+from typing import Any
+
+import pytest
+from conftest import build_app_settings
+from fastapi.testclient import TestClient
+from pydantic import BaseModel
+
+from stirling.api import app
+from stirling.config import load_settings
+from stirling.contracts import (
+    AgentDraftRequest,
+    AgentExecutionRequest,
+    AgentRevisionRequest,
+    Evidence,
+    FolioManifest,
+    PdfCommentRequest,
+    PdfEditRequest,
+    PdfQuestionRequest,
+)
+
+
+@pytest.fixture
+def manifest() -> Iterator[dict[str, Any]]:
+    # Force test settings for the lifespan (other suites pop this override, so we
+    # can't rely on a module-level set), then enter the client as a context manager
+    # so the lifespan runs and populates ``app.state.agent_descriptors`` — the
+    # manifest is built from that real startup registration, not a duplicated list.
+    app.dependency_overrides[load_settings] = build_app_settings
+    try:
+        with TestClient(app) as client:
+            response = client.get("/api/v1/agents/capabilities")
+    finally:
+        app.dependency_overrides.pop(load_settings, None)
+    assert response.status_code == 200
+    yield response.json()
+
+
+@dataclass(frozen=True)
+class _Expected:
+    description: str
+    mode: str
+    required_scope: str
+    route: str
+    input_model: type[BaseModel]
+
+
+# Expected per-capability metadata, keyed by id. Order-independent on purpose —
+# Java consumes the manifest as a keyed operation registry, not a sequence.
+_EXPECTED: dict[str, _Expected] = {
+    "pdf-question-answer": _Expected(
+        description="Answer a natural-language question about a PDF document.",
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/pdf-question",
+        input_model=PdfQuestionRequest,
+    ),
+    "pdf-edit-plan": _Expected(
+        description=(
+            "Produce an edit plan (a structured sequence of PDF operations) from a"
+            " natural-language edit request. The plan is executed by Java through the job"
+            " pipeline; this capability does not modify files itself."
+        ),
+        mode="async",
+        required_scope="mcp.tools.write",
+        route="/api/v1/pdf-edit",
+        input_model=PdfEditRequest,
+    ),
+    "agent-draft": _Expected(
+        description=(
+            "Draft a structured agent specification from a free-text description of the task the user wants automated."
+        ),
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/ai/agents/draft",
+        input_model=AgentDraftRequest,
+    ),
+    "agent-revise": _Expected(
+        description="Revise an existing draft agent specification based on user feedback or constraint changes.",
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/ai/agents/revise",
+        input_model=AgentRevisionRequest,
+    ),
+    "math-audit-examine": _Expected(
+        description=(
+            "Examine a folio manifest of financial / numeric documents and surface the"
+            " evidence that needs to be checked for arithmetic consistency."
+        ),
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/ai/math-auditor-agent/examine",
+        input_model=FolioManifest,
+    ),
+    "math-audit-deliberate": _Expected(
+        description=(
+            "Render a deliberated verdict on a single piece of evidence the examine step"
+            " surfaced (does the arithmetic check out, with what caveats)."
+        ),
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/ai/math-auditor-agent/deliberate",
+        input_model=Evidence,
+    ),
+    "pdf-comment-generate": _Expected(
+        description="Generate inline review comments for a PDF document.",
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/pdf-comment/generate",
+        input_model=PdfCommentRequest,
+    ),
+    "agent-next-action": _Expected(
+        description=(
+            "Decide the next execution step for an in-progress agent workflow. Returns a"
+            " ToolCall, Completed, or CannotContinue action."
+        ),
+        mode="sync",
+        required_scope="mcp.tools.read",
+        route="/api/v1/agents/next-action",
+        input_model=AgentExecutionRequest,
+    ),
+}
+
+
+def test_manifest_version(manifest: dict[str, Any]) -> None:
+    assert manifest["version"] == 1
+
+
+def test_manifest_exposes_exactly_the_expected_capabilities(manifest: dict[str, Any]) -> None:
+    ids = {c["id"] for c in manifest["capabilities"]}
+    assert ids == set(_EXPECTED)
+
+
+def test_manifest_capability_metadata_and_schema(manifest: dict[str, Any]) -> None:
+    by_id = {c["id"]: c for c in manifest["capabilities"]}
+    for cap_id, expected in _EXPECTED.items():
+        entry = by_id[cap_id]
+        assert entry["description"] == expected.description, cap_id
+        assert entry["mode"] == expected.mode, cap_id
+        assert entry["required_scope"] == expected.required_scope, cap_id
+        assert entry["route"] == expected.route, cap_id
+        assert entry["input_schema"] == expected.input_model.model_json_schema(), cap_id