feat(registry): orchestrator + agent + MCP server E2E (#246)

agent/src/models.py

@@ -221,6 +221,12 @@ class TaskConfig(BaseModel):
     trace: bool = False
     # Enriched mid-flight by pipeline.py:
     cedar_policies: list[str] = []
+    # Registry (#246): the resolved asset bundle threaded from the orchestrator
+    # payload — ``{mcp_servers, cedar_policy_modules, skills}`` each a list of
+    # resolved-asset dicts. Applied by registry_loader.py at task start (MVP:
+    # mcp_servers merged into .mcp.json; cedar/skills staged for PR 3). Empty
+    # dict when the blueprint pinned no assets.
+    resolved_assets: dict[str, list[dict]] = {}
     # Cedar HITL (§7.3, §10.2). Per-task approval defaults threaded
     # from the orchestrator payload; consumed by PolicyEngine at
     # construction so the engine seeds ApprovalAllowlist and adopts

agent/src/pipeline.py

@@ -40,6 +40,7 @@
 )
 from progress_writer import _ProgressWriter
 from prompt_builder import build_system_prompt, discover_project_config
+from registry_loader import apply_resolved_assets
 from shell import log, log_error_cw
 from telemetry import (
     _TrajectoryWriter,
@@ -616,6 +617,7 @@ def run_task(
     branch_name: str = "",
     pr_number: str = "",
     cedar_policies: list[str] | None = None,
+    resolved_assets: dict[str, list[dict]] | None = None,
     approval_timeout_s: int | None = None,
     initial_approvals: list[str] | None = None,
     initial_approval_gate_count: int = 0,
@@ -670,6 +672,12 @@ def run_task(
     if cedar_policies:
         config.cedar_policies = cedar_policies
 
+    # Registry (#246): thread the resolved asset bundle onto config so the
+    # loader (below, after channel MCP wiring) can apply it. Same post-build
+    # injection pattern as cedar_policies.
+    if resolved_assets:
+        config.resolved_assets = resolved_assets
+
     # Export session-tag values so tenant-data boto3 clients (DDB/S3) assume
     # the per-task SessionRole with {user_id, repo, task_id} tags. No-op when
     # AGENT_SESSION_ROLE_ARN is unset (local/dev/tests).
@@ -865,6 +873,12 @@ def _on_trace_truncated(max_bytes: int, first_dropped: int) -> None:
                 resolve_jira_oauth_token(config.channel_metadata)
             configure_channel_mcp(setup.repo_dir, config.channel_source)
 
+            # Registry (#246): apply resolved assets AFTER the channel MCP so a
+            # registry-published MCP server merges alongside (not over) the
+            # channel entry in .mcp.json, and both are present before
+            # discover_project_config scans. No-op when no assets were pinned.
+            apply_resolved_assets(setup.repo_dir, config.resolved_assets)
+
             # 👀 on the Linear issue — acknowledges the task is picked up.
             # No-op for non-Linear tasks. Best-effort; failures are logged
             # but do not block the pipeline. Capture the reaction id so we

agent/src/registry_loader.py

@@ -0,0 +1,163 @@
+"""Apply registry-resolved assets (#246) to the agent's runtime environment.
+
+The orchestrator resolves a blueprint's ``registry://`` pins at the create-task
+boundary and threads a ``resolved_assets`` bundle into the invocation payload
+(see docs/design/REGISTRY.md §7/§8). This module is the agent-side consumer:
+it takes that bundle and applies each asset kind with the right mechanism.
+
+MVP (this PR) wires ``mcp_server`` end-to-end — the resolved server config is
+merged into ``<repo_dir>/.mcp.json`` alongside whatever ``channel_mcp.py``
+wrote, so the Claude Agent SDK (``setting_sources=["project"]``) picks it up at
+session start. ``cedar_policy_module`` and ``skill`` are stubs here; PR 3 fills
+them in (Cedar text appended to the PolicyEngine set; skill prompt fragments
+into the SDK setting sources).
+
+Design note — parity with channel_mcp.py: both write ``.mcp.json`` by
+read-merge-write on the ``mcpServers`` map, never clobbering other servers.
+Registry assets and the channel MCP can therefore coexist in one file.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+from shell import log
+
+#: Key under which the resolved MCP server config lives in the descriptor
+#: (REGISTRY.md §3.3). The value is a ready-to-write ``mcpServers`` entry.
+_MCP_SERVER_CONFIG_KEY = "server_config"
+
+
+def _read_existing_mcp_config(path: str) -> dict[str, Any]:
+    """Return the parsed .mcp.json at ``path``, or an empty dict if absent/invalid.
+
+    Mirrors ``channel_mcp._read_existing_mcp_config`` — malformed JSON is logged
+    and treated as absent rather than crashing the agent on a broken file.
+    """
+    if not os.path.isfile(path):
+        return {}
+    try:
+        with open(path, encoding="utf-8") as f:
+            parsed = json.load(f)
+        if isinstance(parsed, dict):
+            return parsed
+        log("WARN", f"Ignoring non-object .mcp.json at {path} (got {type(parsed).__name__})")
+    except (OSError, json.JSONDecodeError) as e:
+        log("WARN", f"Failed to read existing .mcp.json at {path}: {type(e).__name__}: {e}")
+    return {}
+
+
+def _mcp_server_key(asset: dict[str, Any]) -> str:
+    """The ``mcpServers`` key an asset registers under.
+
+    Prefer the descriptor's ``tool_prefix`` stripped of the ``mcp__`` scaffolding
+    (so tools surface consistently), falling back to ``namespace-name`` which is
+    always present and unique per asset.
+    """
+    namespace = asset.get("namespace", "")
+    name = asset.get("name", "")
+    return f"{namespace}-{name}".strip("-") or name or "registry-mcp"
+
+
+def apply_mcp_assets(repo_dir: str, resolved_mcp_servers: list[dict[str, Any]]) -> int:
+    """Merge resolved MCP server assets into ``<repo_dir>/.mcp.json``.
+
+    Read-merge-write on the ``mcpServers`` map, preserving any channel MCP or
+    repo-committed entries. Each asset's descriptor carries a ``server_config``
+    (the ``mcpServers`` entry value); assets missing it are skipped with a warn
+    rather than writing a malformed entry.
+
+    Args:
+      repo_dir: the cloned-repo working directory the SDK uses as ``cwd``.
+      resolved_mcp_servers: the ``mcp_servers`` list from the resolved bundle;
+        each item is a resolved-asset dict (kind/namespace/name/version/descriptor).
+
+    Returns:
+      The number of MCP entries written (0 when nothing to apply, or on a
+      write failure — logged).
+    """
+    if not resolved_mcp_servers:
+        return 0
+    if not repo_dir or not os.path.isdir(repo_dir):
+        log("WARN", f"apply_mcp_assets: repo_dir missing or not a directory: {repo_dir!r}")
+        return 0
+
+    mcp_path = os.path.join(repo_dir, ".mcp.json")
+    config = _read_existing_mcp_config(mcp_path)
+    servers = config.get("mcpServers")
+    if not isinstance(servers, dict):
+        servers = {}
+
+    written = 0
+    for asset in resolved_mcp_servers:
+        descriptor = asset.get("descriptor") or {}
+        server_config = descriptor.get(_MCP_SERVER_CONFIG_KEY)
+        if not isinstance(server_config, dict):
+            log(
+                "WARN",
+                f"apply_mcp_assets: asset {asset.get('namespace')}/{asset.get('name')} "
+                f"has no '{_MCP_SERVER_CONFIG_KEY}' in its descriptor; skipping",
+            )
+            continue
+        servers[_mcp_server_key(asset)] = server_config
+        written += 1
+
+    if written == 0:
+        return 0
+
+    config["mcpServers"] = servers
+    try:
+        with open(mcp_path, "w", encoding="utf-8") as f:
+            json.dump(config, f, indent=2)
+            f.write("\n")
+    except OSError as e:
+        log("ERROR", f"apply_mcp_assets: failed to write {mcp_path}: {e}")
+        return 0
+
+    log("TASK", f"registry: merged {written} MCP server asset(s) into {mcp_path}")
+    return written
+
+
+def apply_cedar_modules(resolved_cedar_modules: list[dict[str, Any]]) -> list[str]:
+    """Return Cedar policy text for resolved cedar_policy_module assets.
+
+    STUB (PR 3): the resolver already returns these assets, but wiring the text
+    into the PolicyEngine's policy set is deferred. Returns an empty list today
+    so callers can unconditionally extend their policy list.
+    """
+    if resolved_cedar_modules:
+        log(
+            "TASK",
+            f"registry: {len(resolved_cedar_modules)} cedar_policy_module asset(s) "
+            "resolved but not yet applied (PR 3)",
+        )
+    return []
+
+
+def apply_skills(repo_dir: str, resolved_skills: list[dict[str, Any]]) -> int:
+    """Apply resolved skill assets to the SDK setting sources.
+
+    STUB (PR 3): returns 0 today. The resolver returns these assets; loading the
+    prompt fragments into the SDK's setting_sources is deferred to PR 3.
+    """
+    if resolved_skills:
+        log(
+            "TASK",
+            f"registry: {len(resolved_skills)} skill asset(s) resolved but not yet applied (PR 3)",
+        )
+    return 0
+
+
+def apply_resolved_assets(repo_dir: str, resolved_assets: dict[str, list[dict]]) -> None:
+    """Apply an entire resolved-asset bundle to the runtime.
+
+    Dispatches each kind to its loader. MVP applies MCP servers; cedar/skills are
+    logged-only stubs (PR 3). Safe to call with an empty bundle (no-op).
+    """
+    if not resolved_assets:
+        return
+    apply_mcp_assets(repo_dir, resolved_assets.get("mcp_servers") or [])
+    apply_cedar_modules(resolved_assets.get("cedar_policy_modules") or [])
+    apply_skills(repo_dir, resolved_assets.get("skills") or [])

agent/src/server.py

@@ -391,6 +391,7 @@ def _run_task_background(
     branch_name: str = "",
     pr_number: str = "",
     cedar_policies: list[str] | None = None,
+    resolved_assets: dict[str, list[dict]] | None = None,
     approval_timeout_s: int | None = None,
     initial_approvals: list[str] | None = None,
     initial_approval_gate_count: int = 0,
@@ -477,6 +478,7 @@ def _run_task_background(
             branch_name=branch_name,
             pr_number=pr_number,
             cedar_policies=cedar_policies,
+            resolved_assets=resolved_assets,
             approval_timeout_s=approval_timeout_s,
             initial_approvals=initial_approvals,
             initial_approval_gate_count=initial_approval_gate_count,
@@ -531,6 +533,10 @@ def _extract_invocation_params(inp: dict, request: Request) -> dict:
     branch_name = inp.get("branch_name", "")
     pr_number = str(inp.get("pr_number", ""))
     cedar_policies = inp.get("cedar_policies") or []
+    # Registry (#246): the resolved asset bundle the orchestrator stamped after
+    # resolving the blueprint's pins. Absent when no assets were pinned; the
+    # loader treats an empty dict as a no-op.
+    resolved_assets = inp.get("resolved_assets") or {}
     # Cedar HITL (§7.3) — per-task approval defaults + seeded allowlist.
     # Both are forwarded verbatim to the pipeline; the engine
     # validates shape at construction time and raises on bad input.
@@ -637,6 +643,7 @@ def _extract_invocation_params(inp: dict, request: Request) -> dict:
         "branch_name": branch_name,
         "pr_number": pr_number,
         "cedar_policies": cedar_policies,
+        "resolved_assets": resolved_assets,
         "approval_timeout_s": approval_timeout_s,
         "initial_approvals": initial_approvals,
         "initial_approval_gate_count": initial_approval_gate_count,

agent/src/workflow/__init__.py

@@ -42,7 +42,7 @@
     gate_status,
     run_workflow,
 )
-from .validator import assert_valid, validate_workflow
+from .validator import assert_valid, is_registry_ref, validate_workflow
 
 __all__ = [
     "STEP_HANDLERS",
@@ -62,6 +62,7 @@
     "WorkflowValidationError",
     "assert_valid",
     "gate_status",
+    "is_registry_ref",
     "load_workflow",
     "load_workflow_file",
     "policy_principal_for",

agent/src/workflow/validator.py

@@ -46,8 +46,23 @@
 
 # Built-in (Phase 1-3) policy modules / MCP servers. Registry refs (registry://)
 # are accepted syntactically now and resolved against #246 in Phase 4 (rule 8).
+#
+# _REGISTRY_REF is the #246 asset grammar (docs/design/REGISTRY.md §6, ADR-018):
+#   registry://<kind>/<namespace>/<name>@<constraint>
+# The kind segment is snake_case (mcp_server, cedar_policy_module) and the
+# @<constraint> pin is MANDATORY (fail-closed, ADR-018 sub-decision 6) — exact,
+# caret, or tilde semver only; floating (*, latest, >=) is rejected. This regex
+# MUST stay byte-for-byte identical to the TypeScript ``parseRef``
+# (cdk/src/handlers/shared/registry-resolver.ts); the ``contracts/registry-resolution/``
+# corpus is the agreement both sides reproduce.
 _BUILTIN_REF = re.compile(r"^builtin/[a-z][a-z0-9_]*$")
-_REGISTRY_REF = re.compile(r"^registry://[a-z][a-z0-9-]*/[a-z0-9][a-z0-9./-]*$")
+_REGISTRY_REF = re.compile(
+    r"^registry://"
+    r"[a-z][a-z0-9_]*/"  # kind (snake_case)
+    r"[a-z][a-z0-9-]*/"  # namespace
+    r"[a-z0-9][a-z0-9._-]*"  # name
+    r"@[\^~]?\d+\.\d+\.\d+(?:-[0-9A-Za-z.-]+)?$"  # @constraint (exact/caret/tilde)
+)
 
 # Mutating built-in tools — forbidden under the read-only tier (rule 6) and when
 # read_only:true (rule 4, shape half is in the schema).
@@ -113,6 +128,16 @@ def _ref_resolves(ref: str) -> bool:
     return bool(_BUILTIN_REF.match(ref) or _REGISTRY_REF.match(ref))
 
 
+def is_registry_ref(ref: str) -> bool:
+    """True iff ``ref`` matches the #246 registry URI grammar (REGISTRY.md §6).
+
+    Public entry point for the ``contracts/registry-resolution/`` parity corpus:
+    this MUST agree byte-for-byte with the TypeScript ``parseRef``. Does not
+    match ``builtin/*`` — grammar only, not resolvability.
+    """
+    return bool(_REGISTRY_REF.match(ref))
+
+
 # --- the rules ---------------------------------------------------------------
 # Each returns a list of human-readable messages (empty == passed). Rule numbers
 # match WORKFLOWS.md §"Validation rules". Rules 3/4/7 also have a schema half;

agent/tests/test_registry_grammar_corpus.py

@@ -0,0 +1,65 @@
+"""Registry URI grammar parity — agent side (#246).
+
+Loads every ``contracts/registry-resolution/grammar-*.json`` fixture and asserts
+that ``workflow.is_registry_ref(ref)`` agrees with the fixture's ``expected.valid``.
+This is the golden contract that the TypeScript ``parseRef``
+(cdk/src/handlers/shared/registry-resolver.ts) must also reproduce — mirroring
+the cross-engine pattern in ``test_cedar_parity.py`` and the workflow-validation
+corpus.
+
+If the regex and a fixture disagree, CI fails before the change ships: either
+the grammar regressed, or the fixture must be updated as a recorded decision.
+See ``contracts/registry-resolution/README.md`` and REGISTRY.md §6.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+from workflow import is_registry_ref
+
+_FIXTURE_DIR = (
+    Path(os.path.dirname(__file__)) / ".." / ".." / "contracts" / "registry-resolution"
+).resolve()
+
+
+def _load_grammar_fixtures() -> list[dict]:
+    assert _FIXTURE_DIR.is_dir(), (
+        f"expected fixture dir at {_FIXTURE_DIR}; see contracts/registry-resolution/README.md"
+    )
+    fixtures = []
+    for path in sorted(_FIXTURE_DIR.glob("grammar-*.json")):
+        fixture = json.loads(path.read_text(encoding="utf-8"))
+        for required in ("name", "ref", "expected"):
+            if required not in fixture:
+                raise AssertionError(f"{path.name}: missing required field {required!r}")
+        if "valid" not in fixture["expected"]:
+            raise AssertionError(f"{path.name}: expected missing 'valid'")
+        fixtures.append(fixture)
+    assert fixtures, f"no grammar-*.json fixtures under {_FIXTURE_DIR}"
+    return fixtures
+
+
+_FIXTURES = _load_grammar_fixtures()
+
+
+@pytest.mark.parametrize("fixture", _FIXTURES, ids=[f["name"] for f in _FIXTURES])
+def test_grammar_matches_fixture(fixture: dict) -> None:
+    observed = is_registry_ref(fixture["ref"])
+    expected = fixture["expected"]["valid"]
+    assert observed == expected, (
+        f"fixture {fixture['name']!r}: grammar drift — is_registry_ref({fixture['ref']!r}) "
+        f"returned {observed}, fixture expects {expected}"
+    )
+
+
+def test_corpus_covers_valid_and_invalid() -> None:
+    """The corpus must exercise both accept and reject paths."""
+    valids = [f for f in _FIXTURES if f["expected"]["valid"]]
+    invalids = [f for f in _FIXTURES if not f["expected"]["valid"]]
+    assert valids, "grammar corpus must include at least one valid ref"
+    assert invalids, "grammar corpus must include at least one invalid ref"