feat(mcp-proxy): add trust boundary redirect contract

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

@@ -1137,6 +1137,97 @@ def stop_managed_launch(*, project_dir: Path) -> dict[str, Any]:
 ProtectionMode = Literal["controlled MCP route", "advisory", "not protected"]
 McpRouteState = Literal["configured", "missing"]
 EvidenceObservationState = Literal["observed", "ready_no_records", "not_initialized"]
+HookCoverageState = Literal["not reported", "missing", "installed", "observed", "stale"]
+TRUST_BOUNDARY_HUMAN_LINE = "Protection scope: project MCP route, not host-wide"
+
+
+def build_trust_boundary(
+    *,
+    home: Path,
+    project_dir: Path,
+    profile: RuntimeProfileSpec | None = None,
+    approval_center: CenterStatus | None = None,
+    mcp_route_state: McpRouteState | None = None,
+    evidence_state: EvidenceObservationState | None = None,
+    protection_mode: ProtectionMode | None = None,
+    hook_coverage: HookCoverageState = "not reported",
+) -> dict[str, Any]:
+    """Return bounded local trust-boundary facts for status/doctor surfaces."""
+
+    config_exists = (proxy_dir(home) / "config.json").is_file()
+    resolved_route = mcp_route_state or classify_mcp_route_state(home)
+    resolved_evidence = evidence_state or classify_evidence_state(
+        home=home,
+        config_exists=config_exists,
+    )
+    center = approval_center or check_approval_center_status(home)
+    resolved_protection = protection_mode
+    if resolved_protection is None and profile is not None:
+        resolved_protection = derive_protection_mode(
+            mcp_route_state=resolved_route,
+            center=center,
+            profile_id=profile.profile_id,
+        )
+    approval_wait: ApprovalWaitState = "not applicable"
+    if profile is not None:
+        approval_wait = classify_approval_wait_mode(home, profile)
+    else:
+        approval_wait = classify_setup_approval_wait_mode(home)
+    return {
+        "scope": "project",
+        "host_wide_control_claim": False,
+        "proxy_route_state": resolved_route,
+        "approval_center_state": center.state,
+        "approval_wait_mode": approval_wait,
+        "local_proof_state": resolved_evidence,
+        "workspace_root_ref": bounded_path_ref(project_dir)["ref"] or "",
+        "hook_coverage": hook_coverage,
+        "protection_mode": resolved_protection or "not protected",
+        "boundary_note": (
+            "AgentVeil trust applies to routed MCP actions in this project. "
+            "Native host tools outside the MCP route are not host-wide contained."
+        ),
+    }
+
+
+def classify_setup_approval_wait_mode(home: Path) -> ApprovalWaitState:
+    """Return approval wait-mode state from project proxy config when known."""
+
+    config_path = proxy_dir(home) / "config.json"
+    if not config_path.is_file():
+        return "not configured"
+    try:
+        config_payload = json.loads(config_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError, UnicodeDecodeError):
+        return "not configured"
+    if not isinstance(config_payload, dict):
+        return "not configured"
+    approval = config_payload.get("approval")
+    if isinstance(approval, Mapping) and approval.get("wait_for_decision") is True:
+        return "enabled"
+    return "not configured"
+
+
+def derive_hook_coverage_from_connector_status(
+    status: Mapping[str, Any],
+) -> HookCoverageState:
+    """Map existing connector status hook fields to bounded hook coverage."""
+
+    if status.get("hook_evidence_observed") is True:
+        return "observed"
+    if status.get("mcp_route_observed") is True:
+        return "observed"
+    hook = str(status.get("hook") or "")
+    hook_state = str(status.get("hook_state") or "")
+    if hook in {"missing"} or hook_state in {"missing"}:
+        return "missing"
+    if hook in {"stale"} or hook_state in {"stale", "invalid-json"}:
+        return "stale"
+    if hook_state in {"fired", "protected"} or status.get("status") == "protected":
+        return "observed"
+    if hook in {"installed", "present", "fired"} or hook_state in {"installed", "fired"}:
+        return "installed"
+    return "missing"
 
 
 @dataclass(frozen=True)
@@ -1151,6 +1242,7 @@ class LaunchStatusView:
     proof_hint: str
     next_step: str
     diagnostics: tuple[str, ...]
+    trust_boundary: dict[str, Any]
 
     def to_dict(self) -> dict[str, Any]:
         payload = self.status.to_dict()
@@ -1165,6 +1257,7 @@ def to_dict(self) -> dict[str, Any]:
                 "proof_hint": self.proof_hint,
                 "next_step": self.next_step,
                 "diagnostics": list(self.diagnostics),
+                "trust_boundary": self.trust_boundary,
             }
         )
         return payload
@@ -1358,6 +1451,15 @@ def build_launch_status_view(
         child_running=status.child_running,
         evidence_state=evidence_state,
     )
+    trust_boundary = build_trust_boundary(
+        home=home,
+        project_dir=project_dir,
+        profile=profile,
+        approval_center=status.approval_center,
+        mcp_route_state=mcp_route_state,
+        evidence_state=evidence_state,
+        protection_mode=protection_mode,
+    )
     return LaunchStatusView(
         status=status,
         profile=profile,
@@ -1367,6 +1469,7 @@ def build_launch_status_view(
         proof_hint=proof_hint,
         next_step=next_step,
         diagnostics=diagnostics,
+        trust_boundary=trust_boundary,
     )
 
 
@@ -1377,6 +1480,7 @@ def format_launch_status_human(view: LaunchStatusView) -> list[str]:
         "AgentVeil managed runtime status",
         f"Profile:         {view.profile.display_name} ({status.profile_id}) — {activity}",
         f"Protection:      {view.protection_mode}",
+        TRUST_BOUNDARY_HUMAN_LINE,
         f"Controls:        {view.profile.control_surface}",
         f"Limitation:      {view.profile.known_limitations}",
         f"Approval center: {status.approval_center.state}",
@@ -1483,6 +1587,7 @@ class LaunchDoctorReport:
     blocking: tuple[str, ...]
     next_step: str
     diagnostics: tuple[str, ...]
+    trust_boundary: dict[str, Any]
 
     def to_dict(self) -> dict[str, Any]:
         return {
@@ -1502,6 +1607,7 @@ def to_dict(self) -> dict[str, Any]:
             "blocking": list(self.blocking),
             "next_step": self.next_step,
             "diagnostics": list(self.diagnostics),
+            "trust_boundary": self.trust_boundary,
         }
 
 
@@ -1686,6 +1792,20 @@ def build_launch_doctor_report(
         blocking=blocking,
         ready=ready,
     )
+    protection_mode = derive_protection_mode(
+        mcp_route_state=mcp_route_state,
+        center=approval_center,
+        profile_id=profile.profile_id,
+    )
+    trust_boundary = build_trust_boundary(
+        home=home,
+        project_dir=project_dir,
+        profile=profile,
+        approval_center=approval_center,
+        mcp_route_state=mcp_route_state,
+        evidence_state=evidence_state,
+        protection_mode=protection_mode,
+    )
     return LaunchDoctorReport(
         profile_id=profile.profile_id,
         profile_label=profile.display_name,
@@ -1701,6 +1821,7 @@ def build_launch_doctor_report(
         blocking=blocking,
         next_step=next_step,
         diagnostics=diagnostics,
+        trust_boundary=trust_boundary,
     )
 
 
@@ -1713,6 +1834,7 @@ def format_launch_doctor_human(report: LaunchDoctorReport) -> list[str]:
         f"Approval Center:  {report.approval_center}",
         f"Approval wait:    {report.approval_wait}",
         f"Local proof:      {report.local_proof}",
+        TRUST_BOUNDARY_HUMAN_LINE,
     ]
     if report.provider_key != "not applicable":
         lines.append(f"Provider key:     {report.provider_key}")
@@ -1747,6 +1869,9 @@ def format_launch_doctor_human(report: LaunchDoctorReport) -> list[str]:
     "build_launch_doctor_diagnostics",
     "build_launch_doctor_next_step",
     "build_launch_doctor_report",
+    "build_trust_boundary",
+    "classify_setup_approval_wait_mode",
+    "derive_hook_coverage_from_connector_status",
     "build_launch_diagnostics",
     "build_launch_next_step",
     "build_launch_proof_hint",

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

@@ -1560,6 +1560,34 @@ def run_setup_wizard_cli(
     return 0 if result.ok or result.setup_status == "incomplete" else 2
 
 
+def _attach_connector_trust_boundary(
+    status: dict[str, Any],
+    *,
+    home: Path,
+    project_dir: Path,
+) -> dict[str, Any]:
+    from agentveil_mcp_proxy.agent_launcher import (
+        CenterStatus,
+        build_trust_boundary,
+        derive_hook_coverage_from_connector_status,
+    )
+    from agentveil_mcp_proxy.approval.server import inspect_managed_approval_center
+
+    enriched = dict(status)
+    center = inspect_managed_approval_center(home)
+    enriched["trust_boundary"] = build_trust_boundary(
+        home=home,
+        project_dir=project_dir,
+        approval_center=CenterStatus(
+            state=center.state,
+            pid=getattr(center, "pid", None),
+            port=getattr(center, "port", None),
+        ),
+        hook_coverage=derive_hook_coverage_from_connector_status(enriched),
+    )
+    return enriched
+
+
 def print_setup_status_cli(
     *,
     home: Path | None = None,
@@ -1584,6 +1612,14 @@ def print_setup_status_cli(
         proxy_config_path=config_path,
     )
     payload = setup_status_to_dict(status)
+    project_dir = home.parent if home.name == ".avp" else home
+    from agentveil_mcp_proxy.agent_launcher import build_trust_boundary
+
+    payload["trust_boundary"] = build_trust_boundary(
+        home=home,
+        project_dir=project_dir,
+        hook_coverage="not reported",
+    )
     assert_setup_output_is_privacy_safe(payload)
     if output_json:
         _print_json(payload, sink)
@@ -4497,6 +4533,7 @@ def run_setup_cursor_status_cli(*, workspace: Path | None, output_json: bool) ->
     status = cursor_setup.connector_status(target, home=home)
     if status["approval_center"] != "running" and status["status"] != "unsafe":
         status["status"] = "advisory"
+    status = _attach_connector_trust_boundary(status, home=home, project_dir=target)
     if output_json:
         _print_operator_json(status)
     else:
@@ -5440,6 +5477,7 @@ def run_setup_codex_status_cli(
         center_state=center.state,
         proxy_command=proxy_command or _resolve_setup_proxy_command(),
     )
+    status = _attach_connector_trust_boundary(status, home=home, project_dir=target)
     if output_json:
         _print_operator_json(status)
     else:
@@ -5729,6 +5767,7 @@ def run_setup_gemini_status_cli(
         center_state=center.state,
         proxy_command=proxy_command or _resolve_setup_proxy_command(),
     )
+    status = _attach_connector_trust_boundary(status, home=home, project_dir=target)
     if output_json:
         _print_operator_json(status)
     else:
@@ -5823,6 +5862,7 @@ def run_setup_connector_status_cli(*, project_dir: Path | None, output_json: boo
     # status accordingly so we never say protected with a dead approval path.
     if center.state != "running" and status["status"] != "unsafe":
         status["status"] = "unsafe" if status["mcp_route"] == "missing" else "advisory"
+    status = _attach_connector_trust_boundary(status, home=home, project_dir=target)
     if output_json:
         _print_operator_json(status)
     else: