feat(registry): Cedar policy modules + skills support (#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/prompt_builder.py

@@ -76,6 +76,10 @@ def build_system_prompt(
     if channel_addendum:
         system_prompt += channel_addendum
 
+    # Registry (#246) skill fragments — appended after channel guidance so
+    # skill instructions also benefit from recency weighting.
+    system_prompt += _registry_skill_addendum(config)
+
     return system_prompt
 
 
@@ -109,9 +113,38 @@ def build_repoless_system_prompt(
     if channel_addendum:
         system_prompt += channel_addendum
 
+    system_prompt += _registry_skill_addendum(config)
+
     return system_prompt
 
 
+def _registry_skill_addendum(config: TaskConfig) -> str:
+    """Return the appended prompt text for registry-resolved skills (#246), or "".
+
+    Each resolved ``skill`` asset carries a ``content`` prompt fragment (resolved
+    from the descriptor; see REGISTRY.md §8) plus ``tool_hints`` in its
+    descriptor. We render one section per skill so the agent picks up the
+    guidance; ``tool_hints`` are advisory text (a skill cannot invoke tools — it
+    only references tools an MCP server separately provides).
+    """
+    skills = (config.resolved_assets or {}).get("skills") or []
+    sections: list[str] = []
+    for skill in skills:
+        content = skill.get("content")
+        if not isinstance(content, str) or not content:
+            continue
+        name = f"{skill.get('namespace')}/{skill.get('name')}"
+        descriptor = skill.get("descriptor") or {}
+        hints = descriptor.get("tool_hints")
+        hint_line = ""
+        if isinstance(hints, list) and hints:
+            hint_line = f"\n\n_Suggested tools: {', '.join(str(h) for h in hints)}._"
+        sections.append(f"\n\n## Skill: {name}\n\n{content}{hint_line}")
+    if sections:
+        log("TASK", f"registry: applied {len(sections)} skill fragment(s) to the system prompt")
+    return "".join(sections)
+
+
 def _render_memory_context(hydrated_context: HydratedContext | None) -> str:
     """Render the memory-context block shared by repo-bound and repo-less prompts."""
     if not (hydrated_context and hydrated_context.memory_context):

agent/src/registry_loader.py

@@ -0,0 +1,178 @@
+"""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.
+
+Each asset kind lands where it is consumed:
+- ``mcp_server`` → merged into ``<repo_dir>/.mcp.json`` **here**, alongside
+  whatever ``channel_mcp.py`` wrote, so the Claude Agent SDK
+  (``setting_sources=["project"]``) picks it up at session start.
+- ``cedar_policy_module`` → applied ORCHESTRATOR-side (the resolved text is
+  merged into the ``cedar_policies`` payload, byte-identical to inline policies);
+  the loader below is log-only.
+- ``skill`` → applied at system-prompt assembly
+  (``prompt_builder._registry_skill_addendum``); the loader below is log-only.
+
+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]]) -> None:
+    """Log resolved cedar_policy_module assets (#246).
+
+    Cedar text is applied ORCHESTRATOR-side: the resolved module ``content`` is
+    merged into the payload's ``cedar_policies`` list before it reaches the
+    agent, so registry-sourced Cedar is byte-identical to inline blueprint
+    policies and flows through the same ``PolicyEngine(extra_policies=...)``
+    path (REGISTRY.md §8). The agent therefore does NOT re-apply Cedar here —
+    doing so would double-load it. This only surfaces what resolved so it shows
+    in the task log alongside the MCP/skill loaders.
+    """
+    if resolved_cedar_modules:
+        names = ", ".join(
+            f"{a.get('namespace')}/{a.get('name')}@{a.get('version')}"
+            for a in resolved_cedar_modules
+        )
+        log(
+            "TASK",
+            f"registry: {len(resolved_cedar_modules)} cedar_policy_module asset(s) "
+            f"applied via cedar_policies payload: {names}",
+        )
+
+
+def apply_skills(repo_dir: str, resolved_skills: list[dict[str, Any]]) -> None:
+    """Log resolved skill assets (#246).
+
+    Skill prompt fragments are applied when the system prompt is assembled
+    (``prompt_builder._registry_skill_addendum``), not here — a skill is prompt
+    text, so it must land in the prompt, and the prompt is built earlier in the
+    pipeline than this loader runs. This only surfaces what resolved so it shows
+    in the task log alongside the MCP/cedar loaders.
+    """
+    if resolved_skills:
+        names = ", ".join(
+            f"{a.get('namespace')}/{a.get('name')}@{a.get('version')}" for a in resolved_skills
+        )
+        log("TASK", f"registry: {len(resolved_skills)} skill asset(s) applied to prompt: {names}")
+
+
+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. MCP servers are merged into ``.mcp.json`` here; Cedar
+    modules are applied orchestrator-side (merged into ``cedar_policies``) and
+    skills are applied at system-prompt assembly — the cedar/skill calls below
+    are log-only for those, since their content lands elsewhere in the pipeline.
+    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;