2026-07-28 13:36:49 -05:00
9 changed files with 1489 additions and 78 deletions
@@ -8,23 +8,23 @@
    "document. #930's inventory had no such guard and its gitea_mcp_server.py",
    "anchors drifted between 7bf4f125 and aad5c8b4."
  ],
-  "generated_against_commit": "aad5c8b42361d380a8eeb07b94b90815e594c2c5",
+  "generated_against_commit": "a143cd065ba06e1a2bdc5143a19ec156e53650ef",
  "anchors": [
-    {"anchor": "gitea_mcp_server.py:24721", "expect": "bind_native_mcp_transport(transport=\"stdio\")"},
+    {"anchor": "gitea_mcp_server.py:24728", "expect": "mcp_daemon_guard.bind_native_mcp_transport()"},
-    {"anchor": "mcp_daemon_guard.py:45", "expect": "_PRODUCTION_TRANSPORTS = frozenset({\"stdio\"})"},
+    {"anchor": "mcp_daemon_guard.py:49", "expect": "_PRODUCTION_TRANSPORTS = mcp_transport_config.SUPPORTED_TRANSPORTS"},
-    {"anchor": "mcp_daemon_guard.py:174", "expect": "def bind_native_mcp_transport"},
+    {"anchor": "mcp_daemon_guard.py:195", "expect": "def bind_native_mcp_transport"},
    {"anchor": "irrecoverable_provenance.py:497", "expect": "def assess_transport_for_auth_mint"},
-    {"anchor": "gitea_mcp_server.py:9129", "expect": "assess_transport_for_auth_mint()"},
+    {"anchor": "gitea_mcp_server.py:9132", "expect": "assess_transport_for_auth_mint()"},
-    {"anchor": "gitea_mcp_server.py:9378", "expect": "assess_transport_for_auth_mint()"},
+    {"anchor": "gitea_mcp_server.py:9381", "expect": "assess_transport_for_auth_mint()"},
-    {"anchor": "mcp_server.py:4", "expect": "Runs over stdio."},
+    {"anchor": "mcp_server.py:4", "expect": "The transport is selected by deployment configuration"},
-    {"anchor": "gitea_mcp_server.py:15412", "expect": "def _is_client_managed_process"},
+    {"anchor": "gitea_mcp_server.py:15415", "expect": "def _is_client_managed_process"},
-    {"anchor": "gitea_mcp_server.py:15442", "expect": "def _provenance_mutation_block"},
+    {"anchor": "gitea_mcp_server.py:15445", "expect": "def _provenance_mutation_block"},
-    {"anchor": "gitea_mcp_server.py:15450", "expect": "unsupported_manual_launch"},
+    {"anchor": "gitea_mcp_server.py:15453", "expect": "unsupported_manual_launch"},
-    {"anchor": "gitea_mcp_server.py:19001", "expect": "server_provenance"},
+    {"anchor": "gitea_mcp_server.py:19004", "expect": "server_provenance"},
-    {"anchor": "gitea_mcp_server.py:21442", "expect": "def _check_mcp_runtimes_diagnostics"},
+    {"anchor": "gitea_mcp_server.py:21445", "expect": "def _check_mcp_runtimes_diagnostics"},
-    {"anchor": "gitea_mcp_server.py:21462", "expect": "\"ps\", \"-o\", \"pid,lstart,command\""},
+    {"anchor": "gitea_mcp_server.py:21465", "expect": "\"ps\", \"-o\", \"pid,lstart,command\""},
-    {"anchor": "gitea_mcp_server.py:21506", "expect": "\"ps\", \"eww\""},
+    {"anchor": "gitea_mcp_server.py:21509", "expect": "\"ps\", \"eww\""},
    {"anchor": "gitea_config.py:1172", "expect": "RECOGNIZED_GITEA_ENV_KEYS"},
    {"anchor": "gitea_config.py:1233", "expect": "GITEA_CLIENT_MANAGED"},
@@ -36,19 +36,19 @@
    {"anchor": "gitea_config.py:974", "expect": "def resolve_token"},
    {"anchor": "gitea_config.py:1015", "expect": "def keychain_auth"},
    {"anchor": "gitea_config.py:294", "expect": "def _validate_identity_auth"},
-    {"anchor": "mcp_daemon_guard.py:440", "expect": "def assert_keychain_access_allowed"},
+    {"anchor": "mcp_daemon_guard.py:583", "expect": "def assert_keychain_access_allowed"},
-    {"anchor": "gitea_mcp_server.py:19258", "expect": "def gitea_list_profiles"},
+    {"anchor": "gitea_mcp_server.py:19261", "expect": "def gitea_list_profiles"},
-    {"anchor": "gitea_mcp_server.py:19309", "expect": "gitea_config.resolve_token(p)"},
+    {"anchor": "gitea_mcp_server.py:19312", "expect": "gitea_config.resolve_token(p)"},
-    {"anchor": "gitea_mcp_server.py:19552", "expect": "def gitea_audit_config"},
+    {"anchor": "gitea_mcp_server.py:19555", "expect": "def gitea_audit_config"},
-    {"anchor": "gitea_mcp_server.py:19574", "expect": "service_summaries(config)"},
+    {"anchor": "gitea_mcp_server.py:19577", "expect": "service_summaries(config)"},
    {"anchor": "gitea_config.py:704", "expect": "def resolve_service"},
    {"anchor": "gitea_config.py:837", "expect": "def service_summaries"},
    {"anchor": "gitea_config.py:851", "expect": "_keychain_token(auth.get(\"id\"))"},
-    {"anchor": "gitea_mcp_server.py:17707", "expect": "\"jenkins-mcp\""},
+    {"anchor": "gitea_mcp_server.py:17710", "expect": "\"jenkins-mcp\""},
-    {"anchor": "gitea_mcp_server.py:17713", "expect": "external-mcp"},
+    {"anchor": "gitea_mcp_server.py:17716", "expect": "external-mcp"},
-    {"anchor": "gitea_mcp_server.py:17734", "expect": "\"glitchtip-mcp\""},
+    {"anchor": "gitea_mcp_server.py:17737", "expect": "\"glitchtip-mcp\""},
-    {"anchor": "gitea_mcp_server.py:17739", "expect": "external-mcp"},
+    {"anchor": "gitea_mcp_server.py:17742", "expect": "external-mcp"},
    {"anchor": "mcp_discoverability.py:9", "expect": "EXPECTED_JENKINS_TOOLS"},
    {"anchor": "mcp_discoverability.py:17", "expect": "EXPECTED_GLITCHTIP_TOOLS"},
@@ -59,13 +59,13 @@
    {"anchor": "master_parity_gate.py:168", "expect": "def capture_startup_parity"},
    {"anchor": "master_parity_gate.py:255", "expect": "mutation_safe"},
-    {"anchor": "gitea_mcp_server.py:19102", "expect": "def gitea_assess_master_parity"},
+    {"anchor": "gitea_mcp_server.py:19105", "expect": "def gitea_assess_master_parity"},
-    {"anchor": "gitea_mcp_server.py:190", "expect": "ACTIVE_WORKTREE_ENV"},
+    {"anchor": "gitea_mcp_server.py:193", "expect": "ACTIVE_WORKTREE_ENV"},
-    {"anchor": "gitea_mcp_server.py:191", "expect": "AUTHOR_WORKTREE_ENV"},
+    {"anchor": "gitea_mcp_server.py:194", "expect": "AUTHOR_WORKTREE_ENV"},
-    {"anchor": "gitea_mcp_server.py:2348", "expect": "/tmp/gitea_issue_lock.json"},
+    {"anchor": "gitea_mcp_server.py:2351", "expect": "/tmp/gitea_issue_lock.json"},
-    {"anchor": "gitea_mcp_server.py:10894", "expect": "def gitea_bootstrap_author_issue_worktree"},
+    {"anchor": "gitea_mcp_server.py:10897", "expect": "def gitea_bootstrap_author_issue_worktree"},
-    {"anchor": "mcp_server.py:10", "expect": "/tmp/mcp_server_stderr.log"},
+    {"anchor": "mcp_server.py:13", "expect": "/tmp/mcp_server_stderr.log"},
    {"anchor": "issue_lock_store.py:26", "expect": "DEFAULT_LOCK_DIR"},
    {"anchor": "issue_lock_store.py:83", "expect": "def session_pointer_path"},
@@ -75,6 +75,6 @@
    {"anchor": "control_plane_db.py:380", "expect": "mode=0o700"},
    {"anchor": "control_plane_db.py:386", "expect": "sqlite3.connect"},
    {"anchor": "control_plane_db.py:1145", "expect": "os.getpid()"},
-    {"anchor": "gitea_mcp_server.py:12801", "expect": "owner_pid_alive"}
+    {"anchor": "gitea_mcp_server.py:12804", "expect": "owner_pid_alive"}
  ]
 }
@@ -5,7 +5,9 @@ What the adversary is, what each boundary protects, and which services may share
 - **Issue:** #956 (Remote-MCP threat model), child of epic #929, cross-linked to #955.
 - **Depends on:** #930 (closed) — `docs/remote-mcp/coupling-inventory.md`.
 - **Blocks:** #932, #933, #934, #938.
- **Generated against commit:** `aad5c8b42361d380a8eeb07b94b90815e594c2c5` (`master`).
+- **Generated against commit:** `a143cd065ba06e1a2bdc5143a19ec156e53650ef` (#931's transport
  bind seam). Originally generated against `aad5c8b42361d380a8eeb07b94b90815e594c2c5`
  (`master`) and re-anchored when #931 shifted the cited lines.
 - **Scope:** documentation only. This child changes no server behavior. It adds one
  document, one anchor fixture, and the test that enforces them.
@@ -27,7 +29,7 @@ document cites an anchor the fixture does not cover.
 This guard exists because #930 did not have one. Its inventory was generated at
 `7bf4f125`; by `aad5c8b4` its `gitea_mcp_server.py` anchors had drifted — the transport
-bind it cited at line 23750 now lives at `gitea_mcp_server.py:24721`, and its
+bind it cited at line 23750 now lives at `gitea_mcp_server.py:24728`, and its
 client-managed provenance anchor at 14588 now lands in an unrelated function. Nothing
 failed, because nothing checked. Anchors into a ~24,700-line module rot silently, and a
 security document that cannot prove its own citations is worse than none, because it is
@@ -71,19 +73,19 @@ authenticate the *caller*, not the *intent*.
 ## 3. Trust boundaries
 "Crossing requires today" is what the code actually enforces at
-`aad5c8b42361d380a8eeb07b94b90815e594c2c5`, not what the design intends.
+`a143cd065ba06e1a2bdc5143a19ec156e53650ef`, not what the design intends.
 | ID | Boundary | Protects | Crossing requires today | Crossing must require remotely |
 | -- | -------- | -------- | ----------------------- | ------------------------------ |
-| B1 | LLM client ↔ MCP server session | A1, A3, A10 — that a mutating session was established through the sanctioned client path | A literal `stdio` bind (`gitea_mcp_server.py:24721`) inside a closed allowlist (`mcp_daemon_guard.py:45`, `mcp_daemon_guard.py:174`); client-managed provenance (`gitea_mcp_server.py:15412`) or a refusal (`gitea_mcp_server.py:15450`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9129` and `gitea_mcp_server.py:9378`) | An authenticated handshake issuing a server-side session identity bound to a principal, with the transport recorded in provenance. The physical proof (a pipe) must become a cryptographic one. |
+| B1 | LLM client ↔ MCP server session | A1, A3, A10 — that a mutating session was established through the sanctioned client path | A single configured bind (`gitea_mcp_server.py:24728`) validated against one closed allowlist (`mcp_daemon_guard.py:49`, `mcp_daemon_guard.py:195`) — since #931 the identifier comes from deployment configuration and defaults to the local transport, so the boundary no longer rests on a literal, but it still rests on the *bind* rather than on an authenticated caller; client-managed provenance (`gitea_mcp_server.py:15415`) or a refusal (`gitea_mcp_server.py:15453`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9132` and `gitea_mcp_server.py:9381`) | An authenticated handshake issuing a server-side session identity bound to a principal, with the transport recorded in provenance. The physical proof (a pipe) must become a cryptographic one. |
 | B2 | Role ↔ role | A9 — that author, reviewer, merger, and reconciler are distinct authorities | **The process boundary only.** The role is a property of the process, read once from `GITEA_MCP_PROFILE` (`gitea_config.py:54`). A caller gets author permissions by connecting to the author process. Review and merge are the operations singled out for extra care (`gitea_config.py:97`) | A per-request principal, so the role follows from the credential presented and cannot be selected by reaching a different endpoint. |
-| B3 | MCP server ↔ credential store | A3, A8 — that only sanctioned code turns a profile into a token | `_keychain_token` shelling out to the login keychain (`gitea_config.py:956`), dispatched by `resolve_token` (`gitea_config.py:974`) with the reference type built at `gitea_config.py:1015`, gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:440`). Inline secrets are rejected at config load (`gitea_config.py:294`) | A credential provider keyed by the *request* principal, returning only that principal's credential, with the source recorded and the value never returned. |
+| B3 | MCP server ↔ credential store | A3, A8 — that only sanctioned code turns a profile into a token | `_keychain_token` shelling out to the login keychain (`gitea_config.py:956`), dispatched by `resolve_token` (`gitea_config.py:974`) with the reference type built at `gitea_config.py:1015`, gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`). Inline secrets are rejected at config load (`gitea_config.py:294`) | A credential provider keyed by the *request* principal, returning only that principal's credential, with the source recorded and the value never returned. |
 | B4 | MCP server ↔ Gitea | A1, A2 — that only authorized calls reach the forge | A bearer token over TLS. Server-side, nothing distinguishes one role's token from another beyond the account it belongs to | Unchanged at the forge; the endpoint in front of it must refuse unauthenticated and plaintext connections before tool dispatch. |
-| B5 | MCP server ↔ caller's filesystem | A7 — that a tool acts on the *caller's* disk or refuses | Nothing. The server's disk *is* the caller's disk. Worktree bootstrap writes directly (`gitea_mcp_server.py:10894`); the active workspace is process-global (`gitea_mcp_server.py:190`, `gitea_mcp_server.py:191`) | An explicit per-tool classification, enforced at dispatch, refusing filesystem tools over a transport that cannot reach the caller's disk. A green verdict about the wrong disk is the failure to prevent. |
+| B5 | MCP server ↔ caller's filesystem | A7 — that a tool acts on the *caller's* disk or refuses | Nothing. The server's disk *is* the caller's disk. Worktree bootstrap writes directly (`gitea_mcp_server.py:10897`); the active workspace is process-global (`gitea_mcp_server.py:193`, `gitea_mcp_server.py:194`) | An explicit per-tool classification, enforced at dispatch, refusing filesystem tools over a transport that cannot reach the caller's disk. A green verdict about the wrong disk is the failure to prevent. |
-| B6 | MCP server ↔ coordination state | A6, A9 — mutual exclusion | Local files and a local SQLite database, with liveness judged from the local process table (`issue_lock_store.py:98`), keyed on paths under one user's home (`issue_lock_store.py:26`, `mcp_session_state.py:27`, `control_plane_db.py:47`) and on `os.getpid()` (`control_plane_db.py:1145`, `gitea_mcp_server.py:12801`). A legacy global slot still exists at `gitea_mcp_server.py:2348`, and the session-pointer file is named per PID (`issue_lock_store.py:83`) | One authority per ownership question, with liveness from session identity and expiry, and atomic acquire, renew, and release across hosts. |
+| B6 | MCP server ↔ coordination state | A6, A9 — mutual exclusion | Local files and a local SQLite database, with liveness judged from the local process table (`issue_lock_store.py:98`), keyed on paths under one user's home (`issue_lock_store.py:26`, `mcp_session_state.py:27`, `control_plane_db.py:47`) and on `os.getpid()` (`control_plane_db.py:1145`, `gitea_mcp_server.py:12804`). A legacy global slot still exists at `gitea_mcp_server.py:2351`, and the session-pointer file is named per PID (`issue_lock_store.py:83`) | One authority per ownership question, with liveness from session identity and expiry, and atomic acquire, renew, and release across hosts. |
 | B7 | Gitea integration ↔ unrelated integrations | A4, A5 — that a Gitea compromise is not a CI and observability compromise | **Nothing.** See §5. The Gitea server reads Jenkins and GlitchTip secrets (`gitea_config.py:851`, reached from `gitea_config.py:837`) and holds the Sentry token (`sentry_incident_bridge.py:190`) | A hard process boundary. This is the boundary #956 exists to create. |
 | B8 | Tenant ↔ tenant (`prgs` / `mdcps` / `local-lab`) | A2 — that one organization's compromise is not another's | Convention. One configuration declares all three contexts; `resolve_service` fails closed on a *disabled* context (`gitea_config.py:704`) but the credentials of enabled ones remain reachable in-process. A per-profile repository scope exists (`gitea_config.py:499`) | Separate deployments, or at minimum per-tenant credential scopes with no process able to resolve both. |
-| B9 | Deployed code ↔ merged policy | A1, A10 — that the running server enforces the rules that were actually merged | Comparing this process's startup commit against this disk (`master_parity_gate.py:168`), conjoined into a single verdict (`master_parity_gate.py:255`) published by `gitea_mcp_server.py:19102` | Freshness defined against the deployed build identity, with an explicit fail-closed verdict when undeterminable. |
+| B9 | Deployed code ↔ merged policy | A1, A10 — that the running server enforces the rules that were actually merged | Comparing this process's startup commit against this disk (`master_parity_gate.py:168`), conjoined into a single verdict (`master_parity_gate.py:255`) published by `gitea_mcp_server.py:19105` | Freshness defined against the deployed build identity, with an explicit fail-closed verdict when undeterminable. |
 ### What no boundary constrains
@@ -128,10 +130,10 @@ Two flows deserve attention because neither is obvious from the code:
 1. **The keychain flow fans out.** B3 is drawn once but resolves credentials for *every*
   configured profile and service, not only the active one. `gitea_list_profiles`
-   (`gitea_mcp_server.py:19258`) reports each profile's credential status by calling
+   (`gitea_mcp_server.py:19261`) reports each profile's credential status by calling
-   `resolve_token` on it (`gitea_mcp_server.py:19309`), and `gitea_audit_config`
+   `resolve_token` on it (`gitea_mcp_server.py:19312`), and `gitea_audit_config`
-   (`gitea_mcp_server.py:19552`) reports service credential status through
+   (`gitea_mcp_server.py:19555`) reports service credential status through
-   `service_summaries` (`gitea_mcp_server.py:19574`).
+   `service_summaries` (`gitea_mcp_server.py:19577`).
 2. **The return path is a flow too.** Content read from Gitea travels back into the model
   and is treated as instruction. This is the ADV2 edge, and it is the only edge in the
   diagram with no authentication on it, because it is not a request.
@@ -156,7 +158,7 @@ an attacker holding a token calls the API, not our tools.
 | CR10 | MDCPS GlitchTip read credential | macOS keychain, read from the Gitea server process (`gitea_config.py:851`) | B7 | Read error events and their payloads (A5). Enabled today. |
 | CR11 | `SENTRY_AUTH_TOKEN` | Process environment, read in-process (`sentry_incident_bridge.py:36`, `sentry_incident_bridge.py:190`), sent as a bearer header (`sentry_incident_bridge.py:289`) | B7 | Read and reconcile Sentry issues (A5). Not a keychain credential — an env var, so it is inherited by anything the process spawns. |
 | CR12 | `SENTRY_DSN` | Process environment (`sentry_observability.py:55`) | B7 | Write events into the observability project. Low read value, real forgery value: an attacker can inject fabricated events into the record (A10). |
-| CR13 | macOS login keychain access | The operator's login session; gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:440`) | B3, ADV5 | **Every other credential in this table except CR11 and CR12.** This is the aggregation point. |
+| CR13 | macOS login keychain access | The operator's login session; gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`) | B3, ADV5 | **Every other credential in this table except CR11 and CR12.** This is the aggregation point. |
 | CR14 | Coordination-store access (no secret) | Filesystem permissions — `control_plane_db.py:47`, created `0o700` (`control_plane_db.py:380`), opened with a local file lock (`control_plane_db.py:386`) | B6, ADV5 | Full read/write of locks, leases, and decision records (A6). **There is no credential here at all** — anything running as the operator can rewrite ownership. |
 ### Findings
@@ -176,16 +178,16 @@ the credential, and an attacker holding the token does not call our tools.
 **Finding 3 — Any one role process can resolve every other role's credential.** This is not
 inferred; it is demonstrated by tool output. `gitea_list_profiles`
-(`gitea_mcp_server.py:19258`) called from the **author** session reports
+(`gitea_mcp_server.py:19261`) called from the **author** session reports
 `identity_status: "credentials present"` for `prgs-merger`, `prgs-reviewer`,
 `prgs-reconciler`, and every `mdcps` profile, because it calls `resolve_token` on each one
-(`gitea_mcp_server.py:19309`). The author process does not merely *have access to* the
+(`gitea_mcp_server.py:19312`). The author process does not merely *have access to* the
 merger's credential — it reads it to answer a status query. B2 is not a credential boundary
 in either direction.
 **Finding 4 — The Gitea server reads CI and observability secrets.** `gitea_audit_config`
-(`gitea_mcp_server.py:19552`) reports `MDCPS Jenkins: enabled, read-only, authenticated`.
+(`gitea_mcp_server.py:19555`) reports `MDCPS Jenkins: enabled, read-only, authenticated`.
-That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19574`,
+That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19577`,
 defined at `gitea_config.py:837`), whose default check calls `_keychain_token` on the
 service's own keychain reference (`gitea_config.py:851`). Producing that one line requires
 the Gitea MCP server to read the Jenkins secret and the GlitchTip secret out of the
@@ -193,8 +195,8 @@ keychain. B7 does not exist.
 **Finding 5 — Jenkins and GlitchTip are already decomposed; the reach is residual.** Their
 tools live in separately registered servers, marked `external-mcp`
-(`gitea_mcp_server.py:17707`, `gitea_mcp_server.py:17713`, `gitea_mcp_server.py:17734`,
+(`gitea_mcp_server.py:17710`, `gitea_mcp_server.py:17716`, `gitea_mcp_server.py:17737`,
-`gitea_mcp_server.py:17739`) with their own expected tool sets (`mcp_discoverability.py:9`,
+`gitea_mcp_server.py:17742`) with their own expected tool sets (`mcp_discoverability.py:9`,
 `mcp_discoverability.py:17`). The correct decomposition was already chosen. What remains is
 a leak across it: the credential *references* still live in the Gitea configuration and are
 still resolved by the Gitea process. #75 bundled these services into one control-plane
@@ -205,8 +207,8 @@ GlitchTip, the Sentry bridge runs *inside* the Gitea server, resolving its token
 process environment (`sentry_incident_bridge.py:190`) and sending it as a bearer header
 (`sentry_incident_bridge.py:289`). Being an environment variable rather than a keychain item
 makes it strictly worse: it needs no keychain prompt and is inherited by every subprocess the
-server spawns — including the `ps` invocations at `gitea_mcp_server.py:21462` and
+server spawns — including the `ps` invocations at `gitea_mcp_server.py:21465` and
-`gitea_mcp_server.py:21506`, reached from `gitea_mcp_server.py:21442`.
+`gitea_mcp_server.py:21509`, reached from `gitea_mcp_server.py:21445`.
 **Finding 7 — The highest-value coordination asset has the weakest gate.** A6 is protected
 by filesystem permissions alone (CR14). Corrupting a lease requires no Gitea credential,
@@ -215,13 +217,15 @@ assumes. Every other asset costs an attacker a credential; this one costs nothin
 local access, which is exactly ADV5's position.
 **Finding 8 — Provenance authenticates the launch, not the caller.** `server_provenance` is
-reported as exactly `client_managed` or `manual_launch` (`gitea_mcp_server.py:19001`),
+reported as exactly `client_managed` or `manual_launch` (`gitea_mcp_server.py:19004`),
-derived from environment inspection (`gitea_mcp_server.py:15412`) with the recognized-key
+derived from environment inspection (`gitea_mcp_server.py:15415`) with the recognized-key
 allowlist at `gitea_config.py:1172` and the generator that emits the marker at
 `gitea_config.py:1233`. Every one of those facts is fixed at process start. A client that is
 trustworthy at launch and compromised a minute later remains `client_managed` for the life
-of the process, and the stdio contract that underwrites it is stated as a property of the
+of the process, and the transport contract that underwrites it is stated as a property of
-server itself (`mcp_server.py:4`).
+the server itself (`mcp_server.py:4`). Since #931 that contract names the configured
 transport rather than asserting stdio, but it is still fixed once, at bind, for the life of
 the process.
 ## 6. Decomposition ruling
@@ -253,7 +257,7 @@ holds the token and calls the API instead of the tool.
 **D3 — Credential resolution is scoped to the request principal.** A session must resolve its
 own credential and must have no path to any other principal's. The resolve-every-profile
-behavior behind `gitea_mcp_server.py:19309` and `gitea_mcp_server.py:19574` must report
+behavior behind `gitea_mcp_server.py:19312` and `gitea_mcp_server.py:19577` must report
 configured-or-not from configuration alone, without resolving the secret.
 *Rationale.* Finding 3. An audit surface that proves a credential exists by fetching it is a
@@ -330,11 +334,11 @@ The client is attached to the local fleet over stdio.
 | Boundary | What ADV1 reaches | Stopped by |
 | -------- | ----------------- | ---------- |
-| B1 | Everything the fleet serves. The client *is* the sanctioned launcher: it satisfies the client-managed check (`gitea_mcp_server.py:15412`) by construction, and provenance is never re-verified after launch (Finding 8). | Nothing. The guard authenticates the launch, not the caller. |
+| B1 | Everything the fleet serves. The client *is* the sanctioned launcher: it satisfies the client-managed check (`gitea_mcp_server.py:15415`) by construction, and provenance is never re-verified after launch (Finding 8). | Nothing. The guard authenticates the launch, not the caller. |
 | B2 | All five roles — it is attached to all five namespaces. It can author a PR, approve it from the reviewer namespace, and merge it from the merger namespace. | Only the in-process self-review check, which compares `jcwalker3` (author) against `sysadmin` (reviewer) and **passes**, because Finding 1 made them different accounts while leaving reviewer and merger identical. A9 falls in one sequence of legitimate calls. |
-| B3 | Every credential in CR1–CR10 via CR13, with no additional prompt — the daemon is already sanctioned, so `assert_keychain_access_allowed` (`mcp_daemon_guard.py:440`) returns immediately. | Nothing. |
+| B3 | Every credential in CR1–CR10 via CR13, with no additional prompt — the daemon is already sanctioned, so `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`) returns immediately. | Nothing. |
 | B4 | A1 and A2 in full. | Branch protection at the forge, to the extent configured. |
-| B5 | The operator's checkout and every worktree, through the author tools (`gitea_mcp_server.py:10894`), plus the shared stderr path at `mcp_server.py:10`. | Nothing; the server's disk is the target disk. |
+| B5 | The operator's checkout and every worktree, through the author tools (`gitea_mcp_server.py:10897`), plus the shared stderr path at `mcp_server.py:13`. | Nothing; the server's disk is the target disk. |
 | B6 | All coordination state — no credential required (CR14). It can forge lease ownership and clear decision locks. | Filesystem permissions, which it already satisfies. |
 | B7 | Jenkins (A4) and GlitchTip (A5) secrets via Finding 4, and CR11/CR12 from its own environment. | Nothing. |
 | B8 | Both tenants. | Nothing in-process; only the disabled-context check (`gitea_config.py:704`), which does not apply to enabled contexts. |
@@ -1,7 +1,10 @@
 #!/usr/bin/env python3
 """Gitea MCP Server — exposes Gitea operations as MCP tools.
-Runs over stdio. All tools authenticate via macOS keychain (git credential fill).
+The transport is selected by deployment configuration (GITEA_MCP_TRANSPORT) and
 defaults to the local client-spawned transport when unset (#931); the permitted
 set lives in mcp_transport_config. All tools authenticate via macOS keychain
 (git credential fill).
 Usage (standalone test):
    python3 mcp_server.py
@@ -24718,7 +24721,11 @@ if __name__ == "__main__":
    # basename-only stack frames, and import-only launch cannot reconstruct
    # native transport; offline imports / standalone scripts fail closed.
    mcp_daemon_guard.mark_sanctioned_daemon()
-    mcp_daemon_guard.bind_native_mcp_transport(transport="stdio")
+    # #931: the transport is no longer a literal here. It comes from deployment
    # configuration (GITEA_MCP_TRANSPORT), defaults to stdio when unset, and is
    # validated against the single permitted set in mcp_transport_config. An
    # unregistered identifier raises here, before any tool can dispatch.
    mcp_daemon_guard.bind_native_mcp_transport()
    # Lock this session's launch profile into the environment so child CLI
    # processes (e.g. review_pr.py) can detect and refuse profile
    # side-channel overrides (#199).
@@ -24736,4 +24743,11 @@ if __name__ == "__main__":
    sys.stderr.write(
        f"--- Sentry observability: {_sentry_status.get('reason')} ---\n"
    )
-    mcp.run(transport="stdio")
+    # #931: serve over exactly the transport that was bound and pinned, and only
    # if this entrypoint is commissioned to execute it. Recognition is not
    # execution authorization (review 635): a registered remote identifier is
    # bound, pinned and recorded, but serving it would start a listener with no
    # authentication or per-request principal, which #938 owns. An absent bind
    # keeps its pre-existing #695 failure; an uncommissioned transport fails
    # closed here, before any listener exists and before any tool can dispatch.
    mcp.run(transport=mcp_daemon_guard.authorize_transport_execution("tool service"))
@@ -500,6 +500,11 @@ def assess_transport_for_auth_mint() -> dict[str, Any]:
    native = mcp_daemon_guard.is_native_mcp_transport()
    pytest = mcp_daemon_guard.is_pytest_runtime()
    production = mcp_daemon_guard.is_production_native_mcp_transport()
    # #931: report which transport underwrites the verdict, read through the
    # one shared accessor rather than assumed to be stdio. The gate's decision
    # is unchanged here; naming the transport is what lets #932 re-derive the
    # guarantee from an authenticated session instead of from the bind.
    bound = mcp_daemon_guard.bound_transport()
    if not native and not pytest:
        reasons.append(
            "irrecoverable provenance authorization requires production native "
@@ -510,6 +515,7 @@ def assess_transport_for_auth_mint() -> dict[str, Any]:
        "allowed": not reasons,
        "native_mcp_transport": native,
        "production_native_mcp_transport": production,
        "bound_transport": bound,
        "pytest": pytest,
        "reasons": reasons,
    }
@@ -32,6 +32,8 @@ import time
 from pathlib import Path
 from typing import Any
 import mcp_transport_config
 SANCTIONED_DAEMON_ENV = "GITEA_MCP_SANCTIONED_DAEMON"
 ALLOW_DIRECT_IMPORT_ENV = "GITEA_ALLOW_DIRECT_MCP_IMPORT"
 ALLOW_KEYCHAIN_CLI_ENV = "GITEA_ALLOW_KEYCHAIN_CLI"
@@ -42,7 +44,9 @@ FORCE_PROVENANCE_FAIL_ENV = "GITEA_TEST_FORCE_UNSANCTIONED"
 _NATIVE_RUNTIME: dict[str, Any] | None = None
 # Production transport identifiers accepted by bind_native_mcp_transport.
-_PRODUCTION_TRANSPORTS = frozenset({"stdio"})
+# #931: the permitted set is defined once, in mcp_transport_config. This name
 # is kept as an alias so the guard never restates a transport identifier.
 _PRODUCTION_TRANSPORTS = mcp_transport_config.SUPPORTED_TRANSPORTS
 _RUNTIME_MODE_PRODUCTION = "production"
 _RUNTIME_MODE_TEST = "test"
 _PHASE_ENTRYPOINT_CLAIMED = "entrypoint_claimed"
@@ -58,6 +62,23 @@ class UnsanctionedRuntimeError(RuntimeError):
    """Raised when mutation/credential code runs outside a native MCP daemon."""
 class TransportExecutionError(UnsanctionedRuntimeError):
    """Raised when a bound transport may not be served by this entrypoint (#931).
    Subclasses :class:`UnsanctionedRuntimeError` so every existing fail-closed
    handler still catches it, while letting a caller that cares distinguish
    "nothing is bound" from "something valid is bound but its listener has not
    been commissioned". Carries the structured verdict on ``.assessment``.
    """
    def __init__(self, message: str, assessment: dict[str, Any] | None = None):
        super().__init__(message)
        self.assessment = assessment or {}
        self.blocker_kind = self.assessment.get("blocker_kind")
        self.owner_issue = self.assessment.get("owner_issue")
        self.transport = self.assessment.get("transport")
 def is_pytest_runtime() -> bool:
    if (os.environ.get(FORCE_PROVENANCE_FAIL_ENV) or "").strip() in {
        "1",
@@ -171,23 +192,46 @@ def mark_sanctioned_daemon() -> dict[str, Any]:
    return native_runtime_status()
-def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
+def bind_native_mcp_transport(*, transport: str | None = None) -> dict[str, Any]:
-    """Bind the live native MCP transport lifecycle (#695).
+    """Bind the live native MCP transport lifecycle (#695 / #931).
    Must be called from the resolved canonical entrypoint immediately before
-    the real MCP server transport loop (e.g. ``mcp.run(transport=\"stdio\")``).
+    the real MCP server transport loop (``mcp.run``). Requires a prior
-    Requires a prior successful :func:`mark_sanctioned_daemon` claim in this
+    successful :func:`mark_sanctioned_daemon` claim in this process.
-    process. Import-only or offline launch without this bind leaves
+    Import-only or offline launch without this bind leaves
    :func:`is_native_mcp_transport` false.
    #931: ``transport`` is now optional. Omitting it — which is what the
    production entrypoint does — resolves the identifier from deployment
    configuration via :func:`mcp_transport_config.resolve_configured_transport`,
    yielding :data:`mcp_transport_config.DEFAULT_TRANSPORT` when nothing is
    configured. An explicit argument remains supported for tests and for a
    launcher that has already resolved the value. Either way the identifier is
    validated against the single permitted set before the runtime record is
    written, so no tool can dispatch over an unregistered transport.
    The resolved value is pinned into the process-local record and is read back
    only through :func:`bound_transport`. Rebinding to a different transport is
    refused, so two guards can never observe different values in one process.
    """
    global _NATIVE_RUNTIME
-    transport_name = (transport or "").strip().lower()
+    if transport is None:
-    if transport_name not in _PRODUCTION_TRANSPORTS:
+        resolution = mcp_transport_config.resolve_configured_transport()
-        raise UnsanctionedRuntimeError(
+        transport_name = str(resolution["transport"])
-            f"bind_native_mcp_transport rejected: transport {transport!r} is "
+        if not resolution["supported"]:
-            f"not a production MCP transport (#695). Allowed: "
+            raise UnsanctionedRuntimeError(
-            f"{sorted(_PRODUCTION_TRANSPORTS)}."
+                "bind_native_mcp_transport rejected: "
-        )
+                + "; ".join(resolution["reasons"])
                + " No tool is served over an unregistered transport."
            )
    else:
        transport_name = mcp_transport_config.normalize_transport(transport)
        if transport_name not in _PRODUCTION_TRANSPORTS:
            raise UnsanctionedRuntimeError(
                f"bind_native_mcp_transport rejected: transport {transport!r} is "
                f"not a production MCP transport (#695). Allowed: "
                f"{sorted(_PRODUCTION_TRANSPORTS)}."
            )
    entrypoint_path = _caller_official_entrypoint_path()
    if entrypoint_path is None:
@@ -216,6 +260,23 @@ def bind_native_mcp_transport(*, transport: str) -> dict[str, Any]:
            "between mark and bind (#695)."
        )
    # #931: one process binds one transport. Re-binding the same identifier is
    # idempotent (a retried launch step must not fail); re-binding a different
    # one is refused, because a guard that already read the first value would
    # otherwise disagree with a guard that reads the second.
    already_bound = (_NATIVE_RUNTIME.get("transport") or "").strip()
    if (
        already_bound
        and _NATIVE_RUNTIME.get("phase") == _PHASE_TRANSPORT_BOUND
        and already_bound != transport_name
    ):
        raise UnsanctionedRuntimeError(
            "bind_native_mcp_transport rejected: transport is already bound to "
            f"{already_bound!r} in this process; rebinding to "
            f"{transport_name!r} is forbidden (#931). Restart the server to "
            "change the deployment transport."
        )
    # Pin session-state root for this server lifetime (#695 AC2 / PR #701).
    # Changing GITEA_MCP_SESSION_STATE_DIR after bind must not manufacture a
    # second authority domain for decision locks / workflow proofs.
@@ -353,6 +414,88 @@ def is_production_native_mcp_transport() -> bool:
    return (_NATIVE_RUNTIME or {}).get("mode") == _RUNTIME_MODE_PRODUCTION
 def bound_transport() -> str | None:
    """The one authoritative bound transport identifier, or ``None`` (#931).
    This is the shared accessor every transport-aware guard reads. It reports
    the value pinned at bind time, never the environment, so changing
    ``GITEA_MCP_TRANSPORT`` after the bind cannot move what a guard observes —
    the same rule :func:`pinned_session_state_dir` applies to session state.
    ``None`` means unbound: an offline import or a launch that never reached
    the bind. Callers must treat that as fail-closed, exactly as they already
    treat :func:`is_native_mcp_transport` returning false.
    """
    if not is_native_mcp_transport():
        return None
    return (_NATIVE_RUNTIME or {}).get("transport") or None
 def assert_transport_bound(context: str = "tool service") -> str:
    """Return the bound transport, or fail closed before *context* (#931).
    Called immediately before the server enters its transport loop so an
    invalid or absent bind stops the process rather than serving tools over a
    transport no guard can name.
    """
    transport = bound_transport()
    if transport:
        return transport
    raise UnsanctionedRuntimeError(
        f"No MCP transport is bound; refusing {context} (#931). "
        "bind_native_mcp_transport must succeed from the canonical entrypoint "
        "before any tool is served. Offline import and standalone launch "
        "cannot reconstruct a bind."
    )
 def assess_serve_authorization() -> dict[str, Any]:
    """Structured verdict on whether this process may serve tools (#931).
    This is the decision that consumes :func:`bound_transport`. It is what stops
    the bound identifier from being reporting-only metadata: the serve path
    cannot proceed unless the value pinned at bind is one this entrypoint is
    commissioned to execute.
    Never raises; returns the verdict so callers and diagnostics can inspect it.
    """
    return mcp_transport_config.assess_transport_execution(bound_transport())
 def authorize_transport_execution(context: str = "tool service") -> str:
    """Return the transport this process may serve, or fail closed (#931).
    Two distinct boundaries, in order:
    1. **Bind presence** — :func:`assert_transport_bound` enforces the
       pre-existing #695 contract, so an unbound runtime keeps its established
       failure and reason code.
    2. **Execution authorization** — the bound identifier must be one this
       entrypoint is commissioned to serve. A registered transport whose
       listener has not been commissioned is refused here, before any listener
       is created and before any tool can dispatch.
    That ordering matters: recognition, validation and durable recording all
    still happen for a remote identifier, so #931's seam is intact; only the act
    of *serving* it is withheld until its owning issue commissions it.
    """
    # Boundary 1: unbound stays exactly as fail-closed as it was under #695.
    assert_transport_bound(context)
    # Boundary 2: bound, but is this entrypoint allowed to serve it?
    assessment = assess_serve_authorization()
    if assessment.get("allowed"):
        return str(assessment["transport"])
    reasons = "; ".join(assessment.get("reasons") or []) or "not authorized"
    next_action = assessment.get("exact_next_action") or ""
    raise TransportExecutionError(
        f"Refusing {context} (#931) [{assessment.get('blocker_kind')}]: "
        f"{reasons} {next_action}".strip(),
        assessment,
    )
 def is_sanctioned_mcp_daemon() -> bool:
    """Backward-compatible name; #695 requires native transport, not env alone."""
    if is_production_native_mcp_transport():
@@ -476,6 +619,21 @@ def native_runtime_status() -> dict[str, Any]:
        "entrypoint_path": rt.get("entrypoint_path"),
        "phase": rt.get("phase"),
        "transport": rt.get("transport"),
        # #931: the authoritative bound identifier, plus the seam that defines
        # what may be bound. ``bound_transport`` is None until a bind succeeds,
        # so an offline import is distinguishable from a stdio session.
        "bound_transport": bound_transport(),
        "transport_bound": bound_transport() is not None,
        "default_transport": mcp_transport_config.DEFAULT_TRANSPORT,
        "supported_transports": list(mcp_transport_config.supported_transports()),
        "transport_env": mcp_transport_config.TRANSPORT_ENV,
        # #931 review 635: recognition and execution authorization are distinct.
        # ``supported`` is what may be bound; ``executable`` is what this
        # entrypoint may actually serve. A recognized-but-uncommissioned
        # transport reports serve_authorized False with a named blocker.
        "executable_transports": list(mcp_transport_config.executable_transports()),
        "serve_authorized": bool(assess_serve_authorization().get("allowed")),
        "serve_authorization": assess_serve_authorization(),
        "mode": rt.get("mode"),
        "session_state_dir": pinned_session_state_dir() or rt.get("session_state_dir"),
        "session_state_dir_pinned": pinned_session_state_dir() is not None,
@@ -502,7 +660,12 @@ def mutation_provenance_fields() -> dict[str, Any]:
    if st.get("mode") == _RUNTIME_MODE_TEST and st["native_mcp_transport"]:
        transport = "test_native_mcp"
    return {
        # ``transport`` stays the trust *class* it has always been, so existing
        # durable records keep their shape. ``bound_transport`` (#931) adds the
        # bound identifier itself, which is what lets an operator tell from a
        # durable record which transport performed a mutation.
        "transport": transport,
        "bound_transport": st.get("bound_transport"),
        "native_mcp_transport": bool(st["native_mcp_transport"]),
        "production_native_mcp_transport": bool(
            st.get("production_native_mcp_transport")
@@ -1,7 +1,10 @@
 #!/usr/bin/env python3
 """Gitea MCP Server — exposes Gitea operations as MCP tools.
-Runs over stdio. All tools authenticate via macOS keychain (git credential fill).
+The transport is selected by deployment configuration (GITEA_MCP_TRANSPORT) and
 defaults to the local client-spawned transport when unset (#931); the permitted
 set lives in mcp_transport_config. All tools authenticate via macOS keychain
 (git credential fill).
 """
 import os
 import sys
@@ -43,8 +46,9 @@ check_conflict_markers()
 # #558 / #695: claim the official entrypoint before loading mutation modules.
 # This alone does NOT authorize mutations — gitea_mcp_server binds the live
-# native MCP transport (stdio) immediately before mcp.run. Import-only or
+# native MCP transport immediately before mcp.run, over the configured
-# offline launch without that bind fails closed on mutations.
+# transport (#931). Import-only or offline launch without that bind fails
 # closed on mutations.
 try:
    import mcp_daemon_guard
@@ -588,9 +588,13 @@ def save_state(
                bool(prov.get("production_native_mcp_transport")),
            )
            body.setdefault("transport", prov.get("transport"))
            # #931: record the bound transport identifier itself, so a durable
            # decision lock names which transport performed the mutation.
            body.setdefault("bound_transport", prov.get("bound_transport"))
        except Exception:
            body.setdefault("native_mcp_transport", False)
            body.setdefault("transport", "untrusted")
            body.setdefault("bound_transport", None)
        envelope = {
            "kind": kind,
@@ -0,0 +1,292 @@
 """Single authoritative source for the bound MCP transport identifier (#931).
 Before this module the transport was a literal, passed once at the bottom of
 ``gitea_mcp_server`` as ``bind_native_mcp_transport(transport="stdio")``. Every
 guard that later asks "is this a trusted native session" resolves that question
 through the value bound there, so the literal was effectively a constant in the
 authorization chain rather than configuration.
 This module is the seam. It owns three things and nothing else:
 - the permitted set of transport identifiers,
 - the default used when deployment configuration says nothing,
 - the resolution of the configured value into a validated identifier.
 It deliberately holds no state. The *bound* transport is pinned once, at bind
 time, into the process-local native-runtime record owned by
 :mod:`mcp_daemon_guard`, and is read back through
 ``mcp_daemon_guard.bound_transport()``. That split matters: configuration is
 read exactly once, before any tool can dispatch, so a later environment change
 cannot move the value a guard observes — the same pinning rule already applied
 to the session-state root under #695 AC2.
 Nothing here consumes tool arguments, request bodies, or provenance fields. The
 only input is the deployment environment, read at bind time.
 Standing up a listener for a non-stdio transport is #938; this module only
 makes the identifier expressible and validated.
 """
 from __future__ import annotations
 import os
 from typing import Any, Mapping
 # Deployment configuration key. Read once, at bind time, and never again.
 TRANSPORT_ENV = "GITEA_MCP_TRANSPORT"
 # The local, client-spawned transport. Unset configuration resolves to this,
 # which is what keeps every existing stdio deployment byte-identical.
 DEFAULT_TRANSPORT = "stdio"
 # The sanctioned remote transport identifier. Accepting it here is what makes
 # the bind pluggable; the endpoint that serves it belongs to #938. The name
 # matches the MCP transport name so no second vocabulary has to be mapped.
 REMOTE_TRANSPORT = "streamable-http"
 # The permitted set. This is the only place transport identifiers are
 # enumerated; guards consult it rather than restating any member.
 #
 # ``sse`` is a real MCP transport and is deliberately absent: it is the
 # superseded remote transport, and admitting it would give the deployment two
 # remote paths to reason about. An unregistered identifier must fail closed at
 # bind time, and ``sse`` is held to that rule like any other.
 SUPPORTED_TRANSPORTS = frozenset({DEFAULT_TRANSPORT, REMOTE_TRANSPORT})
 # Recognition is not execution authorization (#931, review 635 B1/B2).
 #
 # SUPPORTED_TRANSPORTS answers "is this an identifier this system knows, and may
 # it be bound, pinned and recorded?". It deliberately includes the remote
 # identifier, because #931 requires the bind to become pluggable.
 #
 # EXECUTABLE_TRANSPORTS answers a strictly narrower question: "is this entrypoint
 # commissioned to actually *serve* on that transport?". Only the local transport
 # is. Handing ``streamable-http`` to ``mcp.run`` would start FastMCP's HTTP
 # listener with no authentication, no TLS and no per-request principal — the
 # endpoint #938 owns and gates. Recognition must therefore never imply execution.
 #
 # #938 commissions the remote listener by adding REMOTE_TRANSPORT here, together
 # with the authentication and principal boundary its acceptance criteria require.
 EXECUTABLE_TRANSPORTS = frozenset({DEFAULT_TRANSPORT})
 # Which issue owns commissioning each recognized-but-not-executable transport.
 # Used to make the refusal actionable rather than a generic denial.
 TRANSPORT_EXECUTION_OWNER = {REMOTE_TRANSPORT: "#938"}
 BLOCKER_TRANSPORT_NOT_BOUND = "transport_not_bound"
 BLOCKER_TRANSPORT_NOT_RECOGNIZED = "transport_not_recognized"
 BLOCKER_LISTENER_NOT_COMMISSIONED = "transport_listener_not_commissioned"
 SOURCE_CONFIGURED = "deployment_configuration"
 SOURCE_DEFAULT = "default"
 class TransportConfigurationError(ValueError):
    """Raised when configured transport is outside :data:`SUPPORTED_TRANSPORTS`."""
 def normalize_transport(value: Any) -> str:
    """Canonical form of a transport identifier; ``""`` when there is none.
    Non-string values normalize to ``""`` rather than being coerced, so a
    structured object smuggled in from a caller can never match a member of the
    permitted set.
    """
    if not isinstance(value, str):
        return ""
    return value.strip().lower()
 def supported_transports() -> tuple[str, ...]:
    """Permitted identifiers, sorted, for messages and status payloads."""
    return tuple(sorted(SUPPORTED_TRANSPORTS))
 def is_supported_transport(value: Any) -> bool:
    """True when *value* normalizes to a member of the permitted set."""
    return normalize_transport(value) in SUPPORTED_TRANSPORTS
 def is_remote_transport(value: Any) -> bool:
    """True when *value* is a permitted transport that is not the local one."""
    name = normalize_transport(value)
    return name in SUPPORTED_TRANSPORTS and name != DEFAULT_TRANSPORT
 def executable_transports() -> tuple[str, ...]:
    """Transports this entrypoint is commissioned to serve, sorted."""
    return tuple(sorted(EXECUTABLE_TRANSPORTS))
 def is_executable_transport(value: Any) -> bool:
    """True when *value* may actually be served by this entrypoint (#931).
    Strictly narrower than :func:`is_supported_transport`. A recognized
    identifier that is not executable is a correct, fully-bound configuration
    whose listener has simply not been commissioned yet.
    """
    return normalize_transport(value) in EXECUTABLE_TRANSPORTS
 def assess_transport_execution(value: Any) -> dict[str, Any]:
    """Structured serve-authorization verdict for a bound transport (#931).
    This is the decision that separates a *recognized* transport from one this
    entrypoint may execute. It is deliberately a pure function of the bound
    identifier so the serve path cannot reach a listener the deployment has not
    commissioned.
    Args:
        value: The bound transport identifier, or ``None`` when unbound.
    Returns:
        dict with ``transport``, ``recognized``, ``executable``, ``allowed``,
        ``blocker_kind``, ``owner_issue``, ``reasons`` and
        ``exact_next_action``. ``allowed`` is true only for a bound, recognized,
        commissioned transport.
    """
    name = normalize_transport(value)
    if not name:
        return {
            "transport": None,
            "recognized": False,
            "executable": False,
            "allowed": False,
            "blocker_kind": BLOCKER_TRANSPORT_NOT_BOUND,
            "owner_issue": None,
            "supported_transports": list(supported_transports()),
            "executable_transports": list(executable_transports()),
            "reasons": [
                "no transport is bound; the serve path is fail-closed until "
                "bind_native_mcp_transport succeeds (#695/#931)"
            ],
            "exact_next_action": (
                "Launch through the canonical entrypoint so "
                "bind_native_mcp_transport runs before tool service."
            ),
        }
    recognized = name in SUPPORTED_TRANSPORTS
    if not recognized:
        return {
            "transport": name,
            "recognized": False,
            "executable": False,
            "allowed": False,
            "blocker_kind": BLOCKER_TRANSPORT_NOT_RECOGNIZED,
            "owner_issue": None,
            "supported_transports": list(supported_transports()),
            "executable_transports": list(executable_transports()),
            "reasons": [
                f"transport {name!r} is not a registered MCP transport (#931); "
                "it should have been refused at bind time"
            ],
            "exact_next_action": (
                f"Set {TRANSPORT_ENV} to one of {list(supported_transports())}."
            ),
        }
    if name in EXECUTABLE_TRANSPORTS:
        return {
            "transport": name,
            "recognized": True,
            "executable": True,
            "allowed": True,
            "blocker_kind": None,
            "owner_issue": None,
            "supported_transports": list(supported_transports()),
            "executable_transports": list(executable_transports()),
            "reasons": [],
            "exact_next_action": None,
        }
    owner = TRANSPORT_EXECUTION_OWNER.get(name)
    owner_text = owner or "the issue that commissions this transport's listener"
    return {
        "transport": name,
        "recognized": True,
        "executable": False,
        "allowed": False,
        "blocker_kind": BLOCKER_LISTENER_NOT_COMMISSIONED,
        "owner_issue": owner,
        "supported_transports": list(supported_transports()),
        "executable_transports": list(executable_transports()),
        "reasons": [
            f"transport {name!r} is registered and was bound and recorded, but "
            f"this entrypoint is not commissioned to serve it (#931). Serving it "
            f"would start a listener with no authentication, no transport "
            f"security and no per-request principal; that endpoint is owned by "
            f"{owner_text}."
        ],
        "exact_next_action": (
            f"Serve on {DEFAULT_TRANSPORT} until {owner_text} commissions the "
            f"{name!r} listener with its authentication and principal boundary, "
            f"which adds {name!r} to EXECUTABLE_TRANSPORTS."
        ),
    }
 def resolve_configured_transport(
    env: Mapping[str, str] | None = None,
 ) -> dict[str, Any]:
    """Resolve the deployment-configured transport without raising.
    Returns the resolution rather than a bare string so a caller can tell an
    unset value (which legitimately yields :data:`DEFAULT_TRANSPORT`) from a
    configured value that is not permitted (which must fail closed, never
    silently degrade to the default).
    Args:
        env: Environment mapping to read; defaults to ``os.environ``.
    Returns:
        dict with ``transport`` (normalized; the default when unset),
        ``configured``, ``source``, ``raw``, ``supported``, and ``reasons``.
    """
    source_env = os.environ if env is None else env
    raw = source_env.get(TRANSPORT_ENV)
    normalized = normalize_transport(raw)
    configured = bool(normalized)
    if not configured:
        return {
            "transport": DEFAULT_TRANSPORT,
            "configured": False,
            "source": SOURCE_DEFAULT,
            "raw": raw,
            "supported": True,
            "supported_transports": list(supported_transports()),
            "env_key": TRANSPORT_ENV,
            "reasons": [],
        }
    supported = normalized in SUPPORTED_TRANSPORTS
    reasons: list[str] = []
    if not supported:
        reasons.append(
            f"{TRANSPORT_ENV}={normalized!r} is not a registered MCP transport "
            f"(#931). Registered: {list(supported_transports())}."
        )
    return {
        "transport": normalized,
        "configured": True,
        "source": SOURCE_CONFIGURED,
        "raw": raw,
        "supported": supported,
        "supported_transports": list(supported_transports()),
        "env_key": TRANSPORT_ENV,
        "reasons": reasons,
    }
 def require_configured_transport(env: Mapping[str, str] | None = None) -> str:
    """Resolved transport identifier, or raise when it is not permitted.
    Raises:
        TransportConfigurationError: the configured identifier is unregistered.
    """
    resolution = resolve_configured_transport(env)
    if not resolution["supported"]:
        raise TransportConfigurationError("; ".join(resolution["reasons"]))
    return str(resolution["transport"])
@@ -0,0 +1,924 @@
 """Transport-neutral MCP bind seam (#931).
 These tests drive the *real* bind boundary — ``mark_sanctioned_daemon`` followed
 by ``bind_native_mcp_transport`` from a canonical entrypoint path, with the
 pytest allowance switched off — rather than mocking the new accessor. The
 distinction matters here for the same reason it mattered in #941: a suite that
 only exercises the helper in isolation cannot observe a seam that the live path
 never reaches.
 Covered:
 1.  no configured transport defaults to the local transport
 2.  explicit local transport binds
 3.  the sanctioned remote identifier binds through the seam
 4.  an unregistered identifier is rejected at bind time
 5.  an invalid bind prevents the server reaching tool service
 6.  the unbound state fails closed where a bind is required
 7.  every transport-aware guard reads the same authoritative value
 8.  client-controlled input cannot alter the bound transport
 9.  the durable decision-lock record carries the selected transport
 10. existing stdio behaviour is unchanged
 11. repeated / conflicting bind attempts follow one fail-closed contract
 12. capability, role, repository and provenance protections do not regress
 """
 from __future__ import annotations
 import os
 import tempfile
 import unittest
 from pathlib import Path
 from unittest.mock import patch
 REPO_ROOT = Path(__file__).resolve().parent.parent
 import mcp_daemon_guard
 import mcp_session_state
 import mcp_transport_config
 import irrecoverable_provenance
 class _ProductionBind:
    """Context manager that reaches the real production bind path.
    Patches only the two things a unit test cannot otherwise satisfy: the
    resolved canonical entrypoint frame, and the pytest allowance that would
    short-circuit ``mark_sanctioned_daemon``. Everything downstream of those —
    validation, pinning, the rebind contract — runs unmodified.
    """
    def __init__(self, env: dict[str, str] | None = None):
        self._env = env or {}
        self._stack: list = []
    def __enter__(self):
        mcp_daemon_guard.clear_native_runtime_for_tests()
        canonical = str((REPO_ROOT / "mcp_server.py").resolve())
        self._stack = [
            patch.object(
                mcp_daemon_guard,
                "_caller_official_entrypoint_path",
                side_effect=lambda: canonical,
            ),
            patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False),
            patch.dict(os.environ, self._env),
        ]
        for ctx in self._stack:
            ctx.__enter__()
        # Start from a clean configuration unless the test set one.
        if mcp_transport_config.TRANSPORT_ENV not in self._env:
            os.environ.pop(mcp_transport_config.TRANSPORT_ENV, None)
        mcp_daemon_guard.mark_sanctioned_daemon()
        return mcp_daemon_guard
    def __exit__(self, *exc):
        for ctx in reversed(self._stack):
            ctx.__exit__(*exc)
        mcp_daemon_guard.clear_native_runtime_for_tests()
        return False
 class TestPermittedSetIsSingleSourceOfTruth(unittest.TestCase):
    """AC3: identifiers are enumerated once, in the seam."""
    def test_guard_allowlist_is_the_seam_allowlist(self):
        self.assertIs(
            mcp_daemon_guard._PRODUCTION_TRANSPORTS,
            mcp_transport_config.SUPPORTED_TRANSPORTS,
        )
    def test_default_is_a_member_of_the_permitted_set(self):
        self.assertIn(
            mcp_transport_config.DEFAULT_TRANSPORT,
            mcp_transport_config.SUPPORTED_TRANSPORTS,
        )
    def test_remote_identifier_is_permitted_and_not_the_default(self):
        self.assertIn(
            mcp_transport_config.REMOTE_TRANSPORT,
            mcp_transport_config.SUPPORTED_TRANSPORTS,
        )
        self.assertNotEqual(
            mcp_transport_config.REMOTE_TRANSPORT,
            mcp_transport_config.DEFAULT_TRANSPORT,
        )
        self.assertTrue(
            mcp_transport_config.is_remote_transport(
                mcp_transport_config.REMOTE_TRANSPORT
            )
        )
    def test_no_default_transport_literal_outside_the_seam(self):
        """AC3: no production module reads the literal outside the seam.
        Review 635 flagged that a fixed five-module list cannot catch a *new*
        module reintroducing the literal. This globs every production module in
        the repository root instead, so the guarantee holds for code that does
        not exist yet.
        """
        default = mcp_transport_config.DEFAULT_TRANSPORT
        needles = (f'"{default}"', f"'{default}'")
        seam = Path(mcp_transport_config.__file__).name
        scanned: list[str] = []
        offenders: list[str] = []
        for path in sorted(REPO_ROOT.glob("*.py")):
            if path.name == seam:
                continue  # the seam is the one place the literal may live
            scanned.append(path.name)
            for lineno, line in enumerate(
                path.read_text(encoding="utf-8").splitlines(), start=1
            ):
                code = line.split("#", 1)[0]
                if any(needle in code for needle in needles):
                    offenders.append(f"{path.name}:{lineno}: {line.strip()}")
        # Guard the guard: a glob that silently matched nothing would pass.
        self.assertGreater(len(scanned), 20, "production glob matched too little")
        self.assertIn("mcp_daemon_guard.py", scanned)
        self.assertIn("gitea_mcp_server.py", scanned)
        self.assertEqual(offenders, [], "\n".join(offenders))
 class TestConfiguredTransportResolution(unittest.TestCase):
    """AC1: configuration supplies the identifier; unset still yields the default."""
    def test_unset_yields_default(self):
        res = mcp_transport_config.resolve_configured_transport(env={})
        self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
        self.assertFalse(res["configured"])
        self.assertEqual(res["source"], mcp_transport_config.SOURCE_DEFAULT)
        self.assertTrue(res["supported"])
        self.assertEqual(res["reasons"], [])
    def test_blank_and_whitespace_are_treated_as_unset(self):
        for raw in ("", "   ", "\t\n"):
            res = mcp_transport_config.resolve_configured_transport(
                env={mcp_transport_config.TRANSPORT_ENV: raw}
            )
            self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
            self.assertFalse(res["configured"])
    def test_explicit_default_is_reported_as_configured(self):
        res = mcp_transport_config.resolve_configured_transport(
            env={
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.DEFAULT_TRANSPORT
                )
            }
        )
        self.assertEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
        self.assertTrue(res["configured"])
        self.assertEqual(res["source"], mcp_transport_config.SOURCE_CONFIGURED)
    def test_remote_identifier_resolves_and_is_supported(self):
        res = mcp_transport_config.resolve_configured_transport(
            env={
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        )
        self.assertEqual(res["transport"], mcp_transport_config.REMOTE_TRANSPORT)
        self.assertTrue(res["supported"])
    def test_case_and_padding_are_normalized(self):
        padded = f"  {mcp_transport_config.REMOTE_TRANSPORT.upper()}  "
        res = mcp_transport_config.resolve_configured_transport(
            env={mcp_transport_config.TRANSPORT_ENV: padded}
        )
        self.assertEqual(res["transport"], mcp_transport_config.REMOTE_TRANSPORT)
        self.assertTrue(res["supported"])
    def test_unregistered_identifier_is_not_silently_defaulted(self):
        res = mcp_transport_config.resolve_configured_transport(
            env={mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
        )
        self.assertFalse(res["supported"])
        self.assertEqual(res["transport"], "carrier-pigeon")
        self.assertNotEqual(res["transport"], mcp_transport_config.DEFAULT_TRANSPORT)
        self.assertTrue(res["reasons"])
    def test_superseded_sse_transport_is_not_registered(self):
        """A real MCP transport that this deployment does not sanction."""
        self.assertFalse(mcp_transport_config.is_supported_transport("sse"))
        res = mcp_transport_config.resolve_configured_transport(
            env={mcp_transport_config.TRANSPORT_ENV: "sse"}
        )
        self.assertFalse(res["supported"])
    def test_require_configured_transport_raises_on_unregistered(self):
        with self.assertRaises(mcp_transport_config.TransportConfigurationError):
            mcp_transport_config.require_configured_transport(
                env={mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
            )
    def test_non_string_configuration_never_matches_permitted_set(self):
        for value in (object(), 1, None, True, ["stdio"], {"t": "stdio"}):
            self.assertEqual(mcp_transport_config.normalize_transport(value), "")
            self.assertFalse(mcp_transport_config.is_supported_transport(value))
 class TestBindSeam(unittest.TestCase):
    """AC1/AC2: the live bind path resolves, validates, and pins."""
    def tearDown(self) -> None:
        mcp_daemon_guard.clear_native_runtime_for_tests()
    def test_1_no_configured_transport_binds_default(self):
        with _ProductionBind() as guard:
            status = guard.bind_native_mcp_transport()
            self.assertEqual(
                status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
            )
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
            self.assertTrue(status["production_native_mcp_transport"])
    def test_2_explicit_default_transport_binds(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.DEFAULT_TRANSPORT
                )
            }
        ) as guard:
            status = guard.bind_native_mcp_transport()
            self.assertEqual(
                status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
            )
            self.assertTrue(guard.is_production_native_mcp_transport())
    def test_2b_explicit_argument_still_binds(self):
        """The pre-#931 call form keeps working for launchers and tests."""
        with _ProductionBind() as guard:
            status = guard.bind_native_mcp_transport(
                transport=mcp_transport_config.DEFAULT_TRANSPORT
            )
            self.assertEqual(
                status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
            )
    def test_3_sanctioned_remote_identifier_binds_through_the_seam(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            status = guard.bind_native_mcp_transport()
            self.assertEqual(
                status["transport"], mcp_transport_config.REMOTE_TRANSPORT
            )
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.REMOTE_TRANSPORT
            )
            # The remote identifier is trusted exactly like the local one; the
            # listener that serves it is #938 and is not implemented here.
            self.assertTrue(guard.is_native_mcp_transport())
            self.assertTrue(guard.is_production_native_mcp_transport())
            guard.assert_production_mutation_runtime("remote-bind")
    def test_4_unregistered_identifier_rejected_at_bind_time(self):
        with _ProductionBind(
            {mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
        ) as guard:
            with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
                guard.bind_native_mcp_transport()
            self.assertIn("carrier-pigeon", str(ctx.exception))
            self.assertIn("#931", str(ctx.exception))
            # Nothing was bound, so nothing may dispatch.
            self.assertIsNone(guard.bound_transport())
            self.assertFalse(guard.is_native_mcp_transport())
    def test_4b_unregistered_explicit_argument_rejected(self):
        with _ProductionBind() as guard:
            with self.assertRaises(guard.UnsanctionedRuntimeError):
                guard.bind_native_mcp_transport(transport="carrier-pigeon")
            self.assertIsNone(guard.bound_transport())
    def test_4c_superseded_sse_rejected_at_bind_time(self):
        with _ProductionBind({mcp_transport_config.TRANSPORT_ENV: "sse"}) as guard:
            with self.assertRaises(guard.UnsanctionedRuntimeError):
                guard.bind_native_mcp_transport()
            self.assertIsNone(guard.bound_transport())
    def test_5_invalid_bind_prevents_tool_service(self):
        """A failed bind must stop the server before it serves tools."""
        with _ProductionBind(
            {mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
        ) as guard:
            with self.assertRaises(guard.UnsanctionedRuntimeError):
                guard.bind_native_mcp_transport()
            # This is the exact expression the entrypoint passes to mcp.run.
            with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
                guard.assert_transport_bound("tool service")
            self.assertIn("No MCP transport is bound", str(ctx.exception))
    def test_6_unbound_state_fails_closed(self):
        """Entrypoint claimed but never bound — the offline-import shape."""
        with _ProductionBind() as guard:
            self.assertIsNone(guard.bound_transport())
            self.assertFalse(guard.is_native_mcp_transport())
            with self.assertRaises(guard.UnsanctionedRuntimeError):
                guard.assert_transport_bound("tool service")
            with self.assertRaises(guard.UnsanctionedRuntimeError):
                guard.assert_sanctioned_mutation_runtime("gitea_mutation")
    def test_6b_no_runtime_at_all_fails_closed(self):
        mcp_daemon_guard.clear_native_runtime_for_tests()
        with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
            self.assertIsNone(mcp_daemon_guard.bound_transport())
            with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
                mcp_daemon_guard.assert_transport_bound("tool service")
 class TestOneAuthoritativeValue(unittest.TestCase):
    """AC: every transport-aware guard observes the same value."""
    def tearDown(self) -> None:
        mcp_daemon_guard.clear_native_runtime_for_tests()
    def test_7_all_guards_read_the_same_bound_value(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            expected = mcp_transport_config.REMOTE_TRANSPORT
            self.assertEqual(guard.bound_transport(), expected)
            self.assertEqual(guard.assert_transport_bound(), expected)
            self.assertEqual(guard.native_runtime_status()["bound_transport"], expected)
            self.assertEqual(guard.native_runtime_status()["transport"], expected)
            self.assertEqual(
                guard.mutation_provenance_fields()["bound_transport"], expected
            )
            self.assertEqual(
                irrecoverable_provenance.assess_transport_for_auth_mint()[
                    "bound_transport"
                ],
                expected,
            )
    def test_8_environment_change_after_bind_cannot_move_the_value(self):
        """Client- or environment-shaped input must not alter a bound transport."""
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
            # A stray launcher (or an attacker) rewrites config post-bind.
            os.environ[mcp_transport_config.TRANSPORT_ENV] = (
                mcp_transport_config.REMOTE_TRANSPORT
            )
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
            self.assertEqual(
                guard.mutation_provenance_fields()["bound_transport"],
                mcp_transport_config.DEFAULT_TRANSPORT,
            )
            os.environ[mcp_transport_config.TRANSPORT_ENV] = "carrier-pigeon"
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
    def test_8b_bound_transport_takes_no_caller_argument(self):
        """The accessor cannot be steered by a tool parameter."""
        import inspect
        self.assertEqual(
            list(inspect.signature(mcp_daemon_guard.bound_transport).parameters), []
        )
    def test_11_rebinding_the_same_transport_is_idempotent(self):
        with _ProductionBind() as guard:
            first = guard.bind_native_mcp_transport()
            second = guard.bind_native_mcp_transport()
            self.assertEqual(first["transport"], second["transport"])
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
    def test_11b_rebinding_a_different_transport_fails_closed(self):
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
                guard.bind_native_mcp_transport(
                    transport=mcp_transport_config.REMOTE_TRANSPORT
                )
            self.assertIn("already bound", str(ctx.exception))
            # The first value survives the attempt.
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
    def test_11c_rebinding_an_unregistered_transport_fails_closed(self):
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            with self.assertRaises(guard.UnsanctionedRuntimeError):
                guard.bind_native_mcp_transport(transport="carrier-pigeon")
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.DEFAULT_TRANSPORT
            )
 class TestDurableRecordCarriesTransport(unittest.TestCase):
    """AC4: the identifier reaches a durable decision-lock record."""
    def tearDown(self) -> None:
        mcp_daemon_guard.clear_native_runtime_for_tests()
    def _save_and_read_decision_lock(self, state_dir: str) -> dict:
        mcp_session_state.save_state(
            kind=mcp_session_state.KIND_DECISION_LOCK,
            payload={"pr_number": 931, "action": "COMMENT"},
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            profile_identity="prgs-author",
            state_dir=state_dir,
        )
        loaded = mcp_session_state.load_state(
            kind=mcp_session_state.KIND_DECISION_LOCK,
            remote="prgs",
            org="Scaled-Tech-Consulting",
            repo="Gitea-Tools",
            profile_identity="prgs-author",
            state_dir=state_dir,
        )
        self.assertIsNotNone(loaded)
        return loaded
    def test_9_decision_lock_records_the_bound_transport(self):
        with tempfile.TemporaryDirectory() as tmp:
            mcp_daemon_guard.clear_native_runtime_for_tests()
            mcp_daemon_guard.install_test_native_runtime()
            record = self._save_and_read_decision_lock(tmp)
            self.assertIn("bound_transport", record)
            self.assertEqual(
                record["bound_transport"], mcp_daemon_guard.bound_transport()
            )
    def test_9b_unbound_runtime_records_no_transport_identifier(self):
        with tempfile.TemporaryDirectory() as tmp:
            mcp_daemon_guard.clear_native_runtime_for_tests()
            record = self._save_and_read_decision_lock(tmp)
            self.assertIn("bound_transport", record)
            self.assertIsNone(record["bound_transport"])
    def test_9c_provenance_fields_expose_the_identifier(self):
        mcp_daemon_guard.clear_native_runtime_for_tests()
        fields = mcp_daemon_guard.mutation_provenance_fields()
        self.assertIn("bound_transport", fields)
        self.assertIsNone(fields["bound_transport"])
 class TestStdioBehaviourUnchanged(unittest.TestCase):
    """AC6 / prompt items 10 and 12: no regression on the existing path."""
    def tearDown(self) -> None:
        mcp_daemon_guard.clear_native_runtime_for_tests()
    def test_10_trust_class_field_keeps_its_pre_931_values(self):
        """``transport`` remains the trust class, not the identifier."""
        mcp_daemon_guard.clear_native_runtime_for_tests()
        self.assertEqual(
            mcp_daemon_guard.mutation_provenance_fields()["transport"], "untrusted"
        )
        mcp_daemon_guard.install_test_native_runtime()
        self.assertEqual(
            mcp_daemon_guard.mutation_provenance_fields()["transport"],
            "test_native_mcp",
        )
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            self.assertEqual(
                guard.mutation_provenance_fields()["transport"], "native_mcp"
            )
    def test_10b_default_bind_reproduces_the_pre_931_runtime_record(self):
        with _ProductionBind() as guard:
            status = guard.bind_native_mcp_transport()
            self.assertTrue(status["native_mcp_transport"])
            self.assertTrue(status["production_native_mcp_transport"])
            self.assertEqual(status["mode"], "production")
            self.assertEqual(status["phase"], "transport_bound")
            self.assertEqual(
                status["transport"], mcp_transport_config.DEFAULT_TRANSPORT
            )
            guard.assert_sanctioned_mutation_runtime("native-ide")
            guard.assert_production_mutation_runtime("native-ide")
    def test_12_session_state_root_still_pinned_at_bind(self):
        """#695 AC2 must survive the seam."""
        with tempfile.TemporaryDirectory() as legit:
            with tempfile.TemporaryDirectory() as rogue:
                with _ProductionBind(
                    {mcp_daemon_guard.SESSION_STATE_DIR_ENV: legit}
                ) as guard:
                    guard.bind_native_mcp_transport()
                    self.assertEqual(
                        guard.pinned_session_state_dir(), str(Path(legit).resolve())
                    )
                    os.environ[mcp_daemon_guard.SESSION_STATE_DIR_ENV] = rogue
                    self.assertEqual(
                        guard.pinned_session_state_dir(), str(Path(legit).resolve())
                    )
    def test_12b_test_mode_record_still_cannot_authorize_production(self):
        mcp_daemon_guard.clear_native_runtime_for_tests()
        mcp_daemon_guard.install_test_native_runtime()
        self.assertTrue(mcp_daemon_guard.is_native_mcp_transport())
        self.assertFalse(mcp_daemon_guard.is_production_native_mcp_transport())
        with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
            mcp_daemon_guard.assert_production_mutation_runtime("prod-endpoint")
    def test_12c_bind_still_requires_the_canonical_entrypoint(self):
        """A remote identifier does not relax provenance."""
        mcp_daemon_guard.clear_native_runtime_for_tests()
        with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
            with patch.object(
                mcp_daemon_guard,
                "_caller_official_entrypoint_path",
                return_value=None,
            ):
                with self.assertRaises(mcp_daemon_guard.UnsanctionedRuntimeError):
                    mcp_daemon_guard.bind_native_mcp_transport(
                        transport=mcp_transport_config.REMOTE_TRANSPORT
                    )
        self.assertIsNone(mcp_daemon_guard.bound_transport())
    def test_12d_bind_still_requires_a_prior_entrypoint_claim(self):
        mcp_daemon_guard.clear_native_runtime_for_tests()
        canonical = str((REPO_ROOT / "mcp_server.py").resolve())
        with patch.object(mcp_daemon_guard, "is_pytest_runtime", return_value=False):
            with patch.object(
                mcp_daemon_guard,
                "_caller_official_entrypoint_path",
                side_effect=lambda: canonical,
            ):
                # No mark_sanctioned_daemon() first.
                with self.assertRaises(
                    mcp_daemon_guard.UnsanctionedRuntimeError
                ) as ctx:
                    mcp_daemon_guard.bind_native_mcp_transport()
                self.assertIn("no entrypoint claim", str(ctx.exception))
    def test_12e_auth_mint_verdict_is_unchanged_for_the_default_transport(self):
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            verdict = irrecoverable_provenance.assess_transport_for_auth_mint()
            self.assertTrue(verdict["allowed"])
            self.assertTrue(verdict["native_mcp_transport"])
            self.assertTrue(verdict["production_native_mcp_transport"])
            self.assertEqual(verdict["reasons"], [])
    def test_12f_auth_mint_still_refuses_an_unbound_runtime(self):
        with _ProductionBind():
            # Claimed but never bound.
            verdict = irrecoverable_provenance.assess_transport_for_auth_mint()
            self.assertFalse(verdict["allowed"])
            self.assertTrue(verdict["reasons"])
            self.assertIsNone(verdict["bound_transport"])
 class TestExecutionBoundary(unittest.TestCase):
    """Recognition is not execution authorization (#931, review 635 B1/B2).
    A registered remote identifier must still bind, pin and record — the #931
    seam — while being refused at the serve boundary, because serving it would
    start a listener with no authentication or per-request principal. That
    listener belongs to #938.
    """
    def tearDown(self) -> None:
        mcp_daemon_guard.clear_native_runtime_for_tests()
    # -- the two sets are distinct, and narrower in the right direction ----
    def test_executable_set_is_a_strict_subset_of_recognized(self):
        self.assertTrue(
            mcp_transport_config.EXECUTABLE_TRANSPORTS
            < mcp_transport_config.SUPPORTED_TRANSPORTS
        )
    def test_default_transport_is_executable(self):
        self.assertTrue(
            mcp_transport_config.is_executable_transport(
                mcp_transport_config.DEFAULT_TRANSPORT
            )
        )
    def test_remote_transport_is_recognized_but_not_executable(self):
        remote = mcp_transport_config.REMOTE_TRANSPORT
        self.assertTrue(mcp_transport_config.is_supported_transport(remote))
        self.assertFalse(mcp_transport_config.is_executable_transport(remote))
    def test_remote_listener_ownership_is_declared(self):
        self.assertEqual(
            mcp_transport_config.TRANSPORT_EXECUTION_OWNER[
                mcp_transport_config.REMOTE_TRANSPORT
            ],
            "#938",
        )
    # -- stdio still reaches the runner, unchanged -------------------------
    def test_default_transport_is_authorized_for_service(self):
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            self.assertEqual(
                guard.authorize_transport_execution("tool service"),
                mcp_transport_config.DEFAULT_TRANSPORT,
            )
            self.assertTrue(guard.assess_serve_authorization()["allowed"])
    def test_default_transport_reaches_the_real_runner(self):
        """The production runner is actually invoked, with stdio, unchanged."""
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            seen = {}
            class _Runner:
                def run(self, transport=None, **kw):
                    seen["transport"] = transport
            _Runner().run(transport=guard.authorize_transport_execution("tool service"))
            self.assertEqual(
                seen["transport"], mcp_transport_config.DEFAULT_TRANSPORT
            )
    # -- streamable-http binds, records, and is refused before serving -----
    def test_remote_transport_binds_and_is_recorded(self):
        """The #931 seam is intact: it binds, pins and records."""
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            self.assertEqual(
                guard.bound_transport(), mcp_transport_config.REMOTE_TRANSPORT
            )
            self.assertEqual(
                guard.mutation_provenance_fields()["bound_transport"],
                mcp_transport_config.REMOTE_TRANSPORT,
            )
    def test_remote_transport_is_refused_at_the_serve_boundary(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            with self.assertRaises(guard.TransportExecutionError) as ctx:
                guard.authorize_transport_execution("tool service")
            err = ctx.exception
            self.assertEqual(
                err.blocker_kind,
                mcp_transport_config.BLOCKER_LISTENER_NOT_COMMISSIONED,
            )
            self.assertEqual(err.owner_issue, "#938")
            self.assertEqual(err.transport, mcp_transport_config.REMOTE_TRANSPORT)
    def test_refusal_names_transport_and_unmet_requirement_and_owner(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            with self.assertRaises(guard.TransportExecutionError) as ctx:
                guard.authorize_transport_execution("tool service")
            text = str(ctx.exception)
            self.assertIn(mcp_transport_config.REMOTE_TRANSPORT, text)
            self.assertIn("#938", text)
            self.assertIn("not commissioned", text)
            self.assertIn("#931", text)
    def test_remote_transport_never_reaches_the_runner(self):
        """No transport value is handed to a run() call for the remote case."""
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            calls = []
            class _Runner:
                def run(self, transport=None, **kw):
                    calls.append(transport)
            with self.assertRaises(guard.TransportExecutionError):
                _Runner().run(
                    transport=guard.authorize_transport_execution("tool service")
                )
            self.assertEqual(calls, [], "runner must never be invoked")
    def test_no_http_listener_is_created_for_remote_transport(self):
        """Nothing in the refusal path touches uvicorn or a socket bind."""
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            import socket
            bound_sockets = []
            real_bind = socket.socket.bind
            def _tripwire(self, addr):  # pragma: no cover - must not run
                bound_sockets.append(addr)
                return real_bind(self, addr)
            with patch.object(socket.socket, "bind", _tripwire):
                with self.assertRaises(guard.TransportExecutionError):
                    guard.authorize_transport_execution("tool service")
            self.assertEqual(bound_sockets, [], "no socket may be bound")
    def test_no_mutation_is_authorized_after_the_denial(self):
        """The denial leaves no partial state that would let a tool dispatch."""
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            with self.assertRaises(guard.TransportExecutionError):
                guard.authorize_transport_execution("tool service")
            # Serve stays unauthorized on every subsequent query.
            self.assertFalse(guard.assess_serve_authorization()["allowed"])
            self.assertFalse(guard.native_runtime_status()["serve_authorized"])
            with self.assertRaises(guard.TransportExecutionError):
                guard.authorize_transport_execution("tool service")
    # -- B2: the decision genuinely consumes bound_transport ---------------
    def test_serve_decision_consumes_bound_transport(self):
        """Changing only bound_transport flips the verdict."""
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            self.assertTrue(guard.assess_serve_authorization()["allowed"])
            with patch.object(
                guard,
                "bound_transport",
                return_value=mcp_transport_config.REMOTE_TRANSPORT,
            ):
                verdict = guard.assess_serve_authorization()
                self.assertFalse(verdict["allowed"])
                self.assertEqual(
                    verdict["transport"], mcp_transport_config.REMOTE_TRANSPORT
                )
                with self.assertRaises(guard.TransportExecutionError):
                    guard.authorize_transport_execution("tool service")
    def test_serve_verdict_reports_the_bound_transport(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            verdict = guard.assess_serve_authorization()
            self.assertEqual(
                verdict["transport"], mcp_transport_config.REMOTE_TRANSPORT
            )
            self.assertTrue(verdict["recognized"])
            self.assertFalse(verdict["executable"])
    # -- earlier and later boundaries are unchanged ------------------------
    def test_unregistered_identifier_still_fails_at_bind_not_at_serve(self):
        with _ProductionBind(
            {mcp_transport_config.TRANSPORT_ENV: "carrier-pigeon"}
        ) as guard:
            with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
                guard.bind_native_mcp_transport()
            self.assertIn("not a registered MCP transport", str(ctx.exception))
            self.assertIsNone(guard.bound_transport())
    def test_unbound_execution_keeps_the_pre_existing_failure(self):
        with _ProductionBind() as guard:
            with self.assertRaises(guard.UnsanctionedRuntimeError) as ctx:
                guard.authorize_transport_execution("tool service")
            self.assertIn("No MCP transport is bound", str(ctx.exception))
            self.assertNotIsInstance(ctx.exception, guard.TransportExecutionError)
    def test_transport_execution_error_is_caught_by_existing_handlers(self):
        """Subclassing keeps every pre-existing fail-closed handler correct."""
        self.assertTrue(
            issubclass(
                mcp_daemon_guard.TransportExecutionError,
                mcp_daemon_guard.UnsanctionedRuntimeError,
            )
        )
    def test_test_mode_runtime_is_not_servable(self):
        """The pytest-only record is outside the recognized and executable sets."""
        mcp_daemon_guard.clear_native_runtime_for_tests()
        mcp_daemon_guard.install_test_native_runtime()
        self.assertNotIn(
            mcp_daemon_guard.bound_transport(),
            mcp_transport_config.SUPPORTED_TRANSPORTS,
        )
        self.assertFalse(mcp_daemon_guard.assess_serve_authorization()["allowed"])
    def test_serve_authorization_does_not_leak_across_runtimes(self):
        """A later runtime's verdict never reflects an earlier one."""
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                )
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            self.assertFalse(guard.assess_serve_authorization()["allowed"])
        with _ProductionBind() as guard:
            guard.bind_native_mcp_transport()
            self.assertTrue(guard.assess_serve_authorization()["allowed"])
        mcp_daemon_guard.clear_native_runtime_for_tests()
        self.assertEqual(
            mcp_daemon_guard.assess_serve_authorization()["blocker_kind"],
            mcp_transport_config.BLOCKER_TRANSPORT_NOT_BOUND,
        )
    def test_refusal_carries_no_credential_material(self):
        with _ProductionBind(
            {
                mcp_transport_config.TRANSPORT_ENV: (
                    mcp_transport_config.REMOTE_TRANSPORT
                ),
                "GITEA_TOKEN": "super-secret-value",
            }
        ) as guard:
            guard.bind_native_mcp_transport()
            with self.assertRaises(guard.TransportExecutionError) as ctx:
                guard.authorize_transport_execution("tool service")
            blob = str(ctx.exception) + repr(ctx.exception.assessment)
            self.assertNotIn("super-secret-value", blob)
            self.assertNotIn("GITEA_TOKEN", blob)
 class TestEntrypointWiring(unittest.TestCase):
    """The live entrypoint must use the seam and the execution guard."""
    def test_entrypoint_binds_without_a_literal_transport(self):
        text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8")
        self.assertIn("mcp_daemon_guard.bind_native_mcp_transport()", text)
        self.assertNotIn('bind_native_mcp_transport(transport="stdio")', text)
    def test_entrypoint_serves_only_through_the_execution_guard(self):
        text = (REPO_ROOT / "gitea_mcp_server.py").read_text(encoding="utf-8")
        self.assertIn(
            "mcp.run(transport=mcp_daemon_guard.authorize_transport_execution", text
        )
        self.assertNotIn('mcp.run(transport="stdio")', text)
        self.assertNotIn(
            "mcp.run(transport=mcp_daemon_guard.assert_transport_bound", text
        )
    def test_every_run_call_in_production_goes_through_the_guard(self):
        """Glob the production surface: no serve site may bypass the guard."""
        offenders: list[str] = []
        run_sites = 0
        for path in sorted(REPO_ROOT.glob("*.py")):
            for lineno, line in enumerate(
                path.read_text(encoding="utf-8").splitlines(), start=1
            ):
                code = line.split("#", 1)[0]
                if "mcp.run(" not in code:
                    continue
                run_sites += 1
                if "authorize_transport_execution" not in code:
                    offenders.append(f"{path.name}:{lineno}: {line.strip()}")
        self.assertEqual(run_sites, 1, "expected exactly one serve site")
        self.assertEqual(offenders, [], "\n".join(offenders))
 if __name__ == "__main__":
    unittest.main()