feat(mcp-proxy): send install clone advisory context

Add bounded install_clone_context validation for Runtime Gate package install and clone advisory metadata. Wire MCP package mutation tools to send only bounded refs and cover direct SDK privacy rejection plus signed advisory receipt verification.

Implemented with assistance from Codex.

Author: oleg-bk

agentveil/agent.py

@@ -35,6 +35,7 @@
     AVPValidationError,
     AVPServerError,
 )
+from agentveil.runtime_install_clone import validate_install_clone_context
 
 log = logging.getLogger("agentveil")
 
@@ -1335,13 +1336,18 @@ def runtime_evaluate(
         payload_hash: Optional[str] = None,
         risk_class: Optional[str] = None,
         policy_context_hash: Optional[str] = None,
+        install_clone_context: Optional[dict] = None,
     ) -> dict:
         """
         Evaluate whether this agent may perform one action right now.
 
         The agent DID is always derived from this SDK identity; callers cannot
         override it. The returned decision is one of ALLOW, BLOCK, or
         WAITING_FOR_HUMAN_APPROVAL.
+
+        Optional ``install_clone_context`` is bounded advisory metadata for
+        package install/clone tools. It must not contain raw paths, URLs,
+        prompts, secrets, or package-manager command payloads.
         """
         body_data = {
             "agent_did": self._did,
@@ -1360,6 +1366,10 @@ def runtime_evaluate(
             body_data["risk_class"] = risk_class
         if policy_context_hash is not None:
             body_data["policy_context_hash"] = policy_context_hash
+        if install_clone_context is not None:
+            body_data["install_clone_context"] = validate_install_clone_context(
+                install_clone_context
+            )
         return self._post_json("/v1/runtime/evaluate", body_data)
 
     def get_runtime_decision(self, audit_id: str) -> dict:

agentveil/runtime_install_clone.py

@@ -0,0 +1,200 @@
+"""Bounded install/clone advisory context for Runtime Gate requests.
+
+Public clients must not forward raw paths, URLs, prompts, package-manager
+commands, or secrets in ``install_clone_context``. This module validates a
+caller-supplied dict against the private-compatible bounded vocabulary before
+any HTTP body is built.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from typing import Any, Mapping
+
+from agentveil.exceptions import AVPValidationError
+
+_SOURCE_REF_PATTERN = re.compile(r"^src_[a-z0-9_-]{1,58}$")
+_NAMESPACE_REF_PATTERN = re.compile(r"^ns_[a-z0-9_-]{1,58}$")
+_PACKAGE_REF_PATTERN = re.compile(r"^pkg_[a-z0-9_-]{1,58}$")
+_HASH_PATTERN = re.compile(r"^sha256:[0-9a-f]{64}$")
+
+_SOURCE_REF_KINDS = frozenset({
+    "user_pinned",
+    "workspace_registry",
+    "internal_registry",
+    "approved_namespace",
+    "model_suggested",
+    "metadata_suggested",
+    "tool_output_suggested",
+    "unknown",
+})
+_INTENT_SOURCES = frozenset({
+    "user_direct",
+    "user_pinned",
+    "workspace_policy",
+    "agent_plan",
+    "metadata",
+    "tool_output",
+    "remote_context",
+    "unknown",
+})
+_TARGET_SOURCES = frozenset({
+    "user_pinned",
+    "workspace_registry",
+    "internal_registry",
+    "tool_output",
+    "metadata",
+    "model_suggested",
+    "remote_context",
+    "unknown",
+})
+_TOOL_SOURCES = frozenset({
+    "approved_registry",
+    "workspace_registry",
+    "local_manifest",
+    "mcp_server_schema",
+    "model_suggested",
+    "unknown",
+})
+_METADATA_INFLUENCES = frozenset({
+    "none",
+    "low",
+    "medium",
+    "high",
+    "unknown",
+})
+
+_REQUIRED_KEYS = frozenset({
+    "operation",
+    "source_ref",
+    "source_ref_kind",
+    "user_pinned_source",
+    "intent_source",
+    "target_source",
+    "tool_source",
+    "metadata_influence",
+})
+_OPTIONAL_KEYS = frozenset({
+    "package_namespace",
+    "requested_package",
+    "expected_package",
+    "expected_hash",
+    "resource_hash",
+    "payload_hash",
+})
+_ALLOWED_KEYS = _REQUIRED_KEYS | _OPTIONAL_KEYS
+
+_FORBIDDEN_SUBSTRINGS = (
+    "://",
+    "/Users/",
+    "/private/",
+    "/var/folders/",
+    "/home/",
+    "\\Users\\",
+    "X-Amz-",
+    "sk_live_",
+    "sk_test_",
+    "ghp_",
+    "-----BEGIN",
+)
+
+
+def validate_install_clone_context(context: Mapping[str, Any]) -> dict[str, Any]:
+    """Return a bounded copy of ``install_clone_context`` or raise.
+
+    Rejects unknown keys, invalid vocabulary, and values that look like raw
+    paths, URLs, or secret markers. Returns a fresh bounded copy.
+    """
+
+    if not isinstance(context, Mapping):
+        raise AVPValidationError("install_clone_context must be an object")
+
+    keys = set(context)
+    extra = keys - _ALLOWED_KEYS
+    if extra:
+        raise AVPValidationError(
+            f"install_clone_context contains forbidden keys: {sorted(extra)}"
+        )
+    missing = _REQUIRED_KEYS - keys
+    if missing:
+        raise AVPValidationError(
+            f"install_clone_context missing required keys: {sorted(missing)}"
+        )
+
+    operation = context["operation"]
+    if operation not in {"install", "clone"}:
+        raise AVPValidationError("install_clone_context.operation invalid")
+
+    source_ref = _require_str(context["source_ref"], "source_ref")
+    if not _SOURCE_REF_PATTERN.fullmatch(source_ref):
+        raise AVPValidationError("install_clone_context.source_ref invalid")
+
+    source_ref_kind = _require_str(context["source_ref_kind"], "source_ref_kind")
+    if source_ref_kind not in _SOURCE_REF_KINDS:
+        raise AVPValidationError("install_clone_context.source_ref_kind invalid")
+
+    if not isinstance(context["user_pinned_source"], bool):
+        raise AVPValidationError("install_clone_context.user_pinned_source must be bool")
+
+    intent_source = _require_str(context["intent_source"], "intent_source")
+    if intent_source not in _INTENT_SOURCES:
+        raise AVPValidationError("install_clone_context.intent_source invalid")
+
+    target_source = _require_str(context["target_source"], "target_source")
+    if target_source not in _TARGET_SOURCES:
+        raise AVPValidationError("install_clone_context.target_source invalid")
+
+    tool_source = _require_str(context["tool_source"], "tool_source")
+    if tool_source not in _TOOL_SOURCES:
+        raise AVPValidationError("install_clone_context.tool_source invalid")
+
+    metadata_influence = _require_str(context["metadata_influence"], "metadata_influence")
+    if metadata_influence not in _METADATA_INFLUENCES:
+        raise AVPValidationError("install_clone_context.metadata_influence invalid")
+
+    bounded: dict[str, Any] = {
+        "operation": operation,
+        "source_ref": source_ref,
+        "source_ref_kind": source_ref_kind,
+        "user_pinned_source": context["user_pinned_source"],
+        "intent_source": intent_source,
+        "target_source": target_source,
+        "tool_source": tool_source,
+        "metadata_influence": metadata_influence,
+    }
+
+    for key in ("package_namespace", "requested_package", "expected_package"):
+        if key not in context or context[key] is None:
+            continue
+        value = _require_str(context[key], key)
+        pattern = _NAMESPACE_REF_PATTERN if key == "package_namespace" else _PACKAGE_REF_PATTERN
+        if not pattern.fullmatch(value):
+            raise AVPValidationError(f"install_clone_context.{key} invalid")
+        bounded[key] = value
+
+    for key in ("expected_hash", "resource_hash", "payload_hash"):
+        if key not in context or context[key] is None:
+            continue
+        value = _require_str(context[key], key)
+        if not _HASH_PATTERN.fullmatch(value):
+            raise AVPValidationError(f"install_clone_context.{key} invalid")
+        bounded[key] = value
+
+    serialized = json.dumps(bounded, sort_keys=True)
+    lowered = serialized.lower()
+    for marker in _FORBIDDEN_SUBSTRINGS:
+        if marker.lower() in lowered:
+            raise AVPValidationError(
+                "install_clone_context contains forbidden raw path/url/secret marker"
+            )
+    return bounded
+
+
+def _require_str(value: Any, field: str) -> str:
+    if not isinstance(value, str):
+        raise AVPValidationError(f"install_clone_context.{field} must be a string")
+    text = value.strip()
+    if not text or len(text) > 71:
+        raise AVPValidationError(f"install_clone_context.{field} invalid")
+    return text

packages/agentveil-mcp-proxy/agentveil_mcp_proxy/classification.py

@@ -171,6 +171,17 @@
     "pip_run_script": RiskClass.DESTRUCTIVE,
 }
 
+# Package install/clone-relevant tools that may attach bounded Runtime Gate
+# advisory context. Read-only package status tools are intentionally excluded.
+_PACKAGE_INSTALL_CLONE_CONTEXT_TOOLS = frozenset({
+    "pip_install",
+    "pip_uninstall",
+    "pip_update",
+    "pip_run_script",
+})
+INSTALL_CLONE_SOURCE_REF = "src_package_route_builtin"
+INSTALL_CLONE_PACKAGE_REF = "pkg_package_route_builtin"
+
 # Fetch/network MCP tools (e.g. the official MCP "fetch" server's `fetch` tool)
 # take a URL argument. The tool name `fetch` matches the generic _READ prefix,
 # so a benign public fetch already infers READ. The risk that this prefix misses
@@ -277,7 +288,7 @@ class ClassifiedToolCall:
     def backend_metadata(self) -> dict[str, Any]:
         """Return privacy-filtered metadata intended for later backend calls."""
 
-        return {
+        metadata = {
             "action": self.action,
             "action_hash": self.action_hash if self.action == self.action_hash else None,
             "resource": self.resource,
@@ -291,6 +302,10 @@ def backend_metadata(self) -> dict[str, Any]:
                 else self.policy_evaluation.would_decision.value
             ),
         }
+        install_clone_context = build_install_clone_context(self.tool)
+        if install_clone_context is not None:
+            metadata["install_clone_context"] = install_clone_context
+        return metadata
 
     def local_evidence_metadata(self) -> dict[str, Any]:
         """Return local-only metadata for future evidence slices."""
@@ -399,6 +414,29 @@ def extract_resource(arguments: Mapping[str, Any]) -> str | None:
     return None
 
 
+def build_install_clone_context(tool: str) -> dict[str, Any] | None:
+    """Return bounded install/clone advisory context for package mutation tools.
+
+    Returns ``None`` for non-package tools. Includes only stable bounded refs,
+    without raw package names, paths, URLs, prompts, source, or secrets.
+    """
+
+    if tool not in _PACKAGE_INSTALL_CLONE_CONTEXT_TOOLS:
+        return None
+    return {
+        "operation": "install",
+        "source_ref": INSTALL_CLONE_SOURCE_REF,
+        "source_ref_kind": "workspace_registry",
+        "user_pinned_source": False,
+        "intent_source": "user_direct",
+        "target_source": "workspace_registry",
+        "tool_source": "approved_registry",
+        "metadata_influence": "none",
+        "requested_package": INSTALL_CLONE_PACKAGE_REF,
+        "expected_package": INSTALL_CLONE_PACKAGE_REF,
+    }
+
+
 def infer_action_family(tool: str) -> str:
     # claim-check: allow "privacy-safe" describes a coarse label helper, not full data safety.
     """Return a coarse, privacy-safe action family label for one MCP tool name."""
@@ -515,8 +553,11 @@ def _normalize_json(value: Any) -> Any:
 __all__ = [
     "ClassifiedToolCall",
     "HASH_PREFIX",
+    "INSTALL_CLONE_PACKAGE_REF",
+    "INSTALL_CLONE_SOURCE_REF",
     "REDACTED",
     "ToolCallClassifier",
+    "build_install_clone_context",
     "extract_resource",
     "infer_action_family",
     "infer_risk_class",

packages/agentveil-mcp-proxy/agentveil_mcp_proxy/runtime_gate.py

@@ -20,7 +20,9 @@
 from agentveil.agent import AVPAgent
 from agentveil.data_integrity import DataIntegrityError, verify_eddsa_jcs_2022
 from agentveil.delegation import DelegationInvalid, verify_delegation
+from agentveil.exceptions import AVPValidationError
 from agentveil.proof import ProofVerificationError, verify_signed_jcs
+from agentveil.runtime_install_clone import validate_install_clone_context
 from agentveil_mcp_proxy.circuit_breaker import (
     CircuitBreaker,
     CircuitBreakerOpenError,
@@ -88,6 +90,7 @@ class _RuntimeGateRequest:
     payload_hash: str
     risk_class: str
     policy_context_hash: str
+    install_clone_context: Mapping[str, Any] | None = None
 
 
 class RuntimeGateClient:
@@ -191,15 +194,20 @@ def evaluate(self, classification: ClassifiedToolCall) -> RuntimeGateDecision:
             raise RuntimeGateUnavailableError("runtime gate circuit breaker open") from exc
         try:
             try:
-                response = self.agent.runtime_evaluate(
-                    action=request.action,
-                    resource=request.resource,
-                    environment=request.environment,
-                    delegation_receipt=self.control_grant,
-                    payload_hash=request.payload_hash,
-                    risk_class=request.risk_class,
-                    policy_context_hash=request.policy_context_hash,
-                )
+                evaluate_kwargs: dict[str, Any] = {
+                    "action": request.action,
+                    "resource": request.resource,
+                    "environment": request.environment,
+                    "delegation_receipt": self.control_grant,
+                    "payload_hash": request.payload_hash,
+                    "risk_class": request.risk_class,
+                    "policy_context_hash": request.policy_context_hash,
+                }
+                if request.install_clone_context is not None:
+                    evaluate_kwargs["install_clone_context"] = dict(
+                        request.install_clone_context
+                    )
+                response = self.agent.runtime_evaluate(**evaluate_kwargs)
             except Exception as exc:
                 raise RuntimeGateUnavailableError("runtime gate request failed") from exc
             if not isinstance(response, Mapping):
@@ -243,6 +251,14 @@ def seen_receipt_cache_size(self) -> int:
 
     def _build_request(self, classification: ClassifiedToolCall) -> _RuntimeGateRequest:
         metadata = classification.backend_metadata()
+        install_clone_context = metadata.get("install_clone_context")
+        if install_clone_context is not None:
+            if not isinstance(install_clone_context, Mapping):
+                raise RuntimeGateUnavailableError("install_clone_context invalid")
+            try:
+                install_clone_context = validate_install_clone_context(install_clone_context)
+            except AVPValidationError as exc:
+                raise RuntimeGateUnavailableError("install_clone_context invalid") from exc
         return _RuntimeGateRequest(
             action=_runtime_field(metadata.get("action")),
             resource=_runtime_field(metadata.get("resource")),
@@ -253,6 +269,7 @@ def _build_request(self, classification: ClassifiedToolCall) -> _RuntimeGateRequ
                 metadata.get("policy_context_hash"),
                 "policy_context_hash",
             ),
+            install_clone_context=install_clone_context,
         )
 
     def _decision_receipt_jcs(self, response: Mapping[str, Any]) -> str: