Add shared runtime tenant policy

docs/phase-foundation-roadmap.md

@@ -103,7 +103,7 @@ quotas, worker isolation, distributed state, hybrid routing, and E2E proof.
 - [x] Define the decision gates for tenant isolation, authorization, quotas,
       session affinity, distributed state, cost controls, audit, and failure
       domains.
-- [ ] Add shared-runtime threat model and tenant model.
+- [x] Add shared-runtime threat model and tenant model.
 - [ ] Add remote broker API contract for authenticated discovery, describe,
       call, status, cancellation, streaming, and audit.
 - [ ] Add session affinity and state-placement rules.

docs/shared-runtime-guardrails.md

@@ -53,6 +53,47 @@ Contract statement: session affinity must be designed before shared runtime.
 No gate can be satisfied by prose alone. Each gate needs a testable contract,
 a default-deny behavior, and a migration path from the local runtime.
 
+## Threat Model
+
+The P3 threat model treats tenant, workspace, user, upstream, token, log, runtime-state, and audit isolation as separate failure boundaries.
+A shared runtime must prove each boundary before hosted execution can be
+supported.
+
+The current contract remains:
+
+```yaml
+hosted_execution_supported: false
+default_execution_boundary: local_edge
+```
+
+Threats that must fail closed:
+
+- cross-tenant tool discovery
+- cross-workspace tool calls
+- user impersonation
+- upstream state reuse across tenants
+- token reuse across tenants, workspaces, users, or upstreams
+- log records that mix tenant or user scope
+- runtime-state conflict without a lock and recovery journal
+- audit events without tenant, workspace, user, action, result, and denial reason
+
+The local edge broker remains the default.
+Contract statement: unknown upstream classes default to local-only.
+Browser, file-access, OAuth, local-secret, and stateful upstreams are local-only
+until a later task proves a narrower safe class.
+
+## Tenant Model
+
+Every shared-runtime decision requires a tenant context with `tenant_id`,
+`workspace_id`, and `user_id`. The IDs are routing and audit scopes, not local
+paths, account names, email addresses, token values, or private inventory.
+
+The placement rule is deliberately narrow: stateless allowlisted upstreams are shared-worker eligible only when they require no local state. Every other upstream remains local edge.
+
+The code contract lives in `mcp_broker.shared_runtime_policy`. It defines the
+required isolation domains, validates tenant context, and returns placement
+decisions without starting hosted workers or changing runtime state.
+
 ## Mandatory Non-Goals
 
 Phase 3 does not add hosted execution. It does not add remote tool calls, remote

src/mcp_broker/shared_runtime_policy.py

@@ -0,0 +1,102 @@
+"""Shared-runtime tenant and isolation policy contracts."""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+
+SHARED_RUNTIME_POLICY_SCHEMA_VERSION = 1
+LOCAL_EDGE_BOUNDARY = "local_edge"
+SHARED_WORKER_BOUNDARY = "shared_worker"
+REQUIRED_ISOLATION_DOMAINS = (
+    "tenant",
+    "workspace",
+    "user",
+    "upstream",
+    "token",
+    "log",
+    "runtime_state",
+    "audit",
+)
+LOCAL_ONLY_UPSTREAM_CLASSES = frozenset(
+    {
+        "browser",
+        "file_access",
+        "local_secret",
+        "oauth",
+        "stateful",
+    }
+)
+
+
+class SharedRuntimePolicyError(ValueError):
+    """Raised when shared-runtime policy input is unsafe."""
+
+
+def build_shared_runtime_policy() -> dict[str, Any]:
+    return {
+        "schema_version": SHARED_RUNTIME_POLICY_SCHEMA_VERSION,
+        "hosted_execution_supported": False,
+        "default_execution_boundary": LOCAL_EDGE_BOUNDARY,
+        "isolation_domains": list(REQUIRED_ISOLATION_DOMAINS),
+        "tenant_context_required": ["tenant_id", "workspace_id", "user_id"],
+        "upstream_defaults": {
+            "unknown": LOCAL_EDGE_BOUNDARY,
+            "stateful": LOCAL_EDGE_BOUNDARY,
+            "stateless": LOCAL_EDGE_BOUNDARY,
+        },
+    }
+
+
+def validate_shared_runtime_policy(policy: Mapping[str, Any]) -> Mapping[str, Any]:
+    if policy.get("schema_version") != SHARED_RUNTIME_POLICY_SCHEMA_VERSION:
+        raise SharedRuntimePolicyError("shared runtime policy schema_version is invalid")
+    if policy.get("hosted_execution_supported") is not False:
+        raise SharedRuntimePolicyError("hosted execution must remain unsupported")
+    if policy.get("default_execution_boundary") != LOCAL_EDGE_BOUNDARY:
+        raise SharedRuntimePolicyError("default execution boundary must be local_edge")
+    domains = policy.get("isolation_domains")
+    if tuple(domains or ()) != REQUIRED_ISOLATION_DOMAINS:
+        raise SharedRuntimePolicyError("shared runtime isolation domains are incomplete")
+    return policy
+
+
+def validate_tenant_context(context: Mapping[str, Any]) -> dict[str, str]:
+    return {
+        "tenant_id": _required_identifier(context, "tenant_id"),
+        "workspace_id": _required_identifier(context, "workspace_id"),
+        "user_id": _required_identifier(context, "user_id"),
+    }
+
+
+def decide_upstream_placement(
+    *,
+    upstream_class: str,
+    allowlisted: bool,
+    requires_local_state: bool,
+) -> dict[str, Any]:
+    normalized_class = upstream_class.strip().lower()
+    if (
+        normalized_class == "stateless"
+        and allowlisted
+        and not requires_local_state
+    ):
+        return {
+            "execution_boundary": SHARED_WORKER_BOUNDARY,
+            "shared_worker_eligible": True,
+            "reason": "allowlisted_stateless_upstream",
+        }
+    return {
+        "execution_boundary": LOCAL_EDGE_BOUNDARY,
+        "shared_worker_eligible": False,
+        "reason": "local_state_or_unapproved_upstream_class",
+    }
+
+
+def _required_identifier(context: Mapping[str, Any], field: str) -> str:
+    value = context.get(field)
+    if not isinstance(value, str) or not value.strip():
+        raise SharedRuntimePolicyError(f"{field} is required")
+    if "/" in value or "\\" in value:
+        raise SharedRuntimePolicyError(f"{field} must not contain path separators")
+    return value

tests/journey/test_shared_runtime_guardrails_contract.py

@@ -19,6 +19,8 @@ def test_shared_runtime_guardrails_document_phase_3_gates() -> None:
         "## Current Boundary",
         "## Preconditions",
         "## Decision Gates",
+        "## Threat Model",
+        "## Tenant Model",
         "## Mandatory Non-Goals",
     ]
     required_terms = [
@@ -33,6 +35,11 @@ def test_shared_runtime_guardrails_document_phase_3_gates() -> None:
         "cost controls",
         "audit",
         "failure domains",
+        "tenant, workspace, user, upstream, token, log, runtime-state, and audit",
+        "unknown upstream classes default to local-only",
+        "stateless allowlisted upstreams are shared-worker eligible only when they require no local state",
+        "hosted_execution_supported: false",
+        "default_execution_boundary: local_edge",
         "local edge broker remains the default",
         "no remote listener",
         "no shared upstream execution",

tests/unit/test_shared_runtime_policy_contract.py

@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+import pytest
+
+
+pytestmark = pytest.mark.unit
+
+
+def test_shared_runtime_policy_defines_required_isolation_domains() -> None:
+    from mcp_broker.shared_runtime_policy import (
+        REQUIRED_ISOLATION_DOMAINS,
+        build_shared_runtime_policy,
+        validate_shared_runtime_policy,
+    )
+
+    policy = build_shared_runtime_policy()
+
+    assert policy["schema_version"] == 1
+    assert policy["hosted_execution_supported"] is False
+    assert policy["default_execution_boundary"] == "local_edge"
+    assert tuple(policy["isolation_domains"]) == REQUIRED_ISOLATION_DOMAINS
+    assert validate_shared_runtime_policy(policy) == policy
+
+
+@pytest.mark.parametrize(
+    "upstream_class",
+    [
+        "unknown",
+        "stateful",
+        "browser",
+        "file_access",
+        "oauth",
+        "local_secret",
+    ],
+)
+def test_unknown_and_stateful_upstreams_default_to_local_only(
+    upstream_class: str,
+) -> None:
+    from mcp_broker.shared_runtime_policy import decide_upstream_placement
+
+    decision = decide_upstream_placement(
+        upstream_class=upstream_class,
+        allowlisted=False,
+        requires_local_state=True,
+    )
+
+    assert decision == {
+        "execution_boundary": "local_edge",
+        "shared_worker_eligible": False,
+        "reason": "local_state_or_unapproved_upstream_class",
+    }
+
+
+def test_stateless_allowlisted_upstream_is_only_shared_worker_eligible_without_local_state() -> None:
+    from mcp_broker.shared_runtime_policy import decide_upstream_placement
+
+    eligible = decide_upstream_placement(
+        upstream_class="stateless",
+        allowlisted=True,
+        requires_local_state=False,
+    )
+    denied = decide_upstream_placement(
+        upstream_class="stateless",
+        allowlisted=True,
+        requires_local_state=True,
+    )
+
+    assert eligible == {
+        "execution_boundary": "shared_worker",
+        "shared_worker_eligible": True,
+        "reason": "allowlisted_stateless_upstream",
+    }
+    assert denied["execution_boundary"] == "local_edge"
+    assert denied["shared_worker_eligible"] is False
+
+
+def test_tenant_context_requires_tenant_workspace_and_user_ids() -> None:
+    from mcp_broker.shared_runtime_policy import (
+        SharedRuntimePolicyError,
+        validate_tenant_context,
+    )
+
+    valid = validate_tenant_context(
+        {
+            "tenant_id": "tenant-acme",
+            "workspace_id": "workspace-platform",
+            "user_id": "user-123",
+        }
+    )
+
+    assert valid == {
+        "tenant_id": "tenant-acme",
+        "workspace_id": "workspace-platform",
+        "user_id": "user-123",
+    }
+    with pytest.raises(SharedRuntimePolicyError, match="tenant_id"):
+        validate_tenant_context({"workspace_id": "workspace-platform", "user_id": "user-123"})
+    with pytest.raises(SharedRuntimePolicyError, match="workspace_id"):
+        validate_tenant_context({"tenant_id": "tenant-acme", "user_id": "user-123"})
+    with pytest.raises(SharedRuntimePolicyError, match="user_id"):
+        validate_tenant_context(
+            {"tenant_id": "tenant-acme", "workspace_id": "workspace-platform"}
+        )
+
+
+def test_policy_validation_rejects_missing_isolation_domains() -> None:
+    from mcp_broker.shared_runtime_policy import (
+        SharedRuntimePolicyError,
+        build_shared_runtime_policy,
+        validate_shared_runtime_policy,
+    )
+
+    policy = build_shared_runtime_policy()
+    policy["isolation_domains"] = ["tenant"]
+
+    with pytest.raises(SharedRuntimePolicyError, match="isolation domains"):
+        validate_shared_runtime_policy(policy)