feat(mcp-proxy): consume paid install safety and runtime advisory

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/paid_activation.py

@@ -204,6 +204,7 @@ def _envelope(
     activation: Mapping[str, Any],
     provider: PaidProviderSnapshot,
     install_state: Mapping[str, Any] | None = None,
+    install_safety_advisory: str | None = None,
 ) -> dict[str, Any]:
     payload = {
         "ok": True,
@@ -223,7 +224,11 @@ def _envelope(
             "public_fallback_available": install_state.get("public_fallback_available"),
             "error_code": install_state.get("error_code"),
             "last_installed_at": install_state.get("last_installed_at"),
+            "install_safety_state": install_state.get("install_safety_state"),
+            "install_safety_reason": install_state.get("install_safety_reason"),
         }
+    if install_safety_advisory:
+        payload["install_safety_advisory"] = install_safety_advisory
     return payload
 
 
@@ -233,15 +238,17 @@ def format_paid_human_output(payload: Mapping[str, Any]) -> str:
     install = payload.get("install") or {}
     if activation.get("status") == STATUS_ACTIVE and install:
         fallback = "available" if activation.get("public_fallback_available") else "unavailable"
-        return "\n".join(
-            [
-                f"Status: {activation['status']}",
-                f"Provider: {install.get('provider_id') or provider.get('provider_id') or 'private_v1'}",
-                f"Installed package: {install.get('package_name')}",
-                f"Installed version: {install.get('package_version')}",
-                f"Public fallback: {fallback}",
-            ]
-        )
+        lines = [
+            f"Status: {activation['status']}",
+            f"Provider: {install.get('provider_id') or provider.get('provider_id') or 'private_v1'}",
+            f"Installed package: {install.get('package_name')}",
+            f"Installed version: {install.get('package_version')}",
+            f"Public fallback: {fallback}",
+        ]
+        advisory = payload.get("install_safety_advisory")
+        if advisory:
+            lines.insert(0, str(advisory))
+        return "\n".join(lines)
 
     lines = [
         f"Paid provider: {'present' if payload['paid_provider_present'] else 'absent'}",
@@ -317,6 +324,7 @@ def build_paid_activate_payload(*, license_key: str, home: Path | None) -> dict[
             activation=activation,
             provider=result.provider,
             install_state=result.install_state,
+            install_safety_advisory=result.install_safety_advisory,
         )
 
     provider = activate_with_paid_provider(license_key=license_key.strip())