From 1dd30ecb1508b559868c2d5d94367bc055d5138e Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 29 Jul 2026 02:52:18 -0400 Subject: [PATCH] feat(mcp): client/session-aware runtime ownership and provenance (#948) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two surfaces reported different provenance for one process. `gitea_get_runtime_context` read the live environment and reported `client_managed`; `mcp_namespace_health.classify_namespace_probe` derived provenance from `_safe_env_summary()`, whose `SAFE_ENV_KEYS` allowlist never contained `GITEA_CLIENT_MANAGED`, `GITEA_MCP_CLIENT_MANAGED`, or `GITEA_SERVER_PROVENANCE`. That lookup could only ever miss, so the health surface was structurally incapable of returning anything but `manual_launch`. Neither model could name which client or which session owned a runtime, so a healthy daemon serving a second client was indistinguishable from a duplicate, and the profile-wide duplicate scan walled the whole fleet. Introduce `mcp_worker_identity` as the one authority, splitting two claims the old code ran together: * launch provenance — was this hand-launched from a terminal? Answered from the environment, which is legitimate because the launcher sets it. Preserves the #686 wall unchanged. * session ownership — which live client session owns this runtime now? Answered only from a live attachment record; no environment flag can establish it. The module provides collision-resistant worker identities (`--`), an atomic SQLite registry with fencing epochs, heartbeat-based liveness, generation takeover that supersedes only a non-live claimant, cohort classification, and failure scoping. Behaviour changes: * Registering an existing worker identity fails closed; it is never replaced, adopted, or merged with. The caller mints a different identity instead. * A generation held by a live session cannot be claimed by a second one. A generation whose claimant is not live is taken over with a higher fencing epoch, so stale ownership cannot permanently strand a healthy daemon. * A superseded session presenting an old epoch is refused and performs no write. * Liveness comes from heartbeat freshness; a live PID cannot resurrect an expired record, and a dead PID withdraws liveness. * Workers sharing a role or profile no longer trigger a profile-wide duplicate block, provided each carries a distinct identity. Processes with no identity evidence remain classified as duplicates, so the #686 wall still holds. * Runtime failures are scoped to a worker identity or generation, never to a profile or the fleet. * Reconnect guidance no longer defaults to Codex. An unidentified client gets host-agnostic steps; `gitea_request_mcp_reconnect(client=...)` defaults to resolving the client from the live attachment record. * `resolve_bound_remote` keeps a bound namespace on its remote instead of falling through to the `dadeschools` library default. Absence of proof is now reported as `unproven` rather than asserted as `manual_launch`. Both still fail closed — `is_client_managed` is unchanged, so nothing previously refused is now permitted — but remediation names the proof that is actually missing instead of describing a terminal launch it cannot evidence. The #686 test is updated for that vocabulary and keeps every wall-preserving assertion. Threat-model anchors and their citations in docs/remote-mcp/threat-model.md are restamped for the line movement in gitea_mcp_server.py. Tests: tests/test_issue_948_client_session_provenance.py adds 43 cases covering Codex/Gemini/Antigravity/Claude attachment, same-client new session, cross-client takeover after a session ends, two live conflicting sessions, stale records, missing attachment proof, environment flags without attachment, mixed generations, duplicate cohorts, the hardcoded-client regression, explicit PRGS selection, default-remote host drift, cross-surface agreement, and fail-closed handling without false reconnect loops. Synthetic identifiers throughout. Full suite from a branches/ worktree: 28F/5953P/6S at head vs 28F/5910P/6S at merge base 8eada1fb, identical failing ID sets. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01F6Vomtndpq2gSBa88Tfcwy --- docs/remote-mcp/threat-model-anchors.json | 44 +- docs/remote-mcp/threat-model.md | 44 +- gitea_mcp_server.py | 291 +++- mcp_client_reconnect.py | 30 +- mcp_namespace_health.py | 63 +- mcp_worker_identity.py | 1432 +++++++++++++++++ tests/test_issue_686_manual_mcp_provenance.py | 20 +- ...est_issue_948_client_session_provenance.py | 662 ++++++++ 8 files changed, 2491 insertions(+), 95 deletions(-) create mode 100644 mcp_worker_identity.py create mode 100644 tests/test_issue_948_client_session_provenance.py diff --git a/docs/remote-mcp/threat-model-anchors.json b/docs/remote-mcp/threat-model-anchors.json index 33e81ef..3aa0eb6 100644 --- a/docs/remote-mcp/threat-model-anchors.json +++ b/docs/remote-mcp/threat-model-anchors.json @@ -11,7 +11,7 @@ "generated_against_commit": "ca5f078d8a575ea3e2991771f8b4ea85e3dcaaa0", "anchors": [ { - "anchor": "gitea_mcp_server.py:24864", + "anchor": "gitea_mcp_server.py:25087", "expect": "mcp_daemon_guard.bind_native_mcp_transport()" }, { @@ -27,11 +27,11 @@ "expect": "def assess_transport_for_auth_mint" }, { - "anchor": "gitea_mcp_server.py:9191", + "anchor": "gitea_mcp_server.py:9192", "expect": "assess_transport_for_auth_mint()" }, { - "anchor": "gitea_mcp_server.py:9440", + "anchor": "gitea_mcp_server.py:9441", "expect": "assess_transport_for_auth_mint()" }, { @@ -39,31 +39,31 @@ "expect": "The transport is selected by deployment configuration" }, { - "anchor": "gitea_mcp_server.py:15481", + "anchor": "gitea_mcp_server.py:15630", "expect": "def _is_client_managed_process" }, { - "anchor": "gitea_mcp_server.py:15511", + "anchor": "gitea_mcp_server.py:15644", "expect": "def _provenance_mutation_block" }, { - "anchor": "gitea_mcp_server.py:15519", + "anchor": "gitea_mcp_server.py:15652", "expect": "unsupported_manual_launch" }, { - "anchor": "gitea_mcp_server.py:19070", + "anchor": "gitea_mcp_server.py:19217", "expect": "server_provenance" }, { - "anchor": "gitea_mcp_server.py:21581", + "anchor": "gitea_mcp_server.py:21741", "expect": "def _check_mcp_runtimes_diagnostics" }, { - "anchor": "gitea_mcp_server.py:21601", + "anchor": "gitea_mcp_server.py:21761", "expect": "\"ps\", \"-o\", \"pid,lstart,command\"" }, { - "anchor": "gitea_mcp_server.py:21645", + "anchor": "gitea_mcp_server.py:21805", "expect": "\"ps\", \"eww\"" }, { @@ -107,19 +107,19 @@ "expect": "def assert_keychain_access_allowed" }, { - "anchor": "gitea_mcp_server.py:19327", + "anchor": "gitea_mcp_server.py:19487", "expect": "def gitea_list_profiles" }, { - "anchor": "gitea_mcp_server.py:19378", + "anchor": "gitea_mcp_server.py:19538", "expect": "gitea_config.resolve_token(p)" }, { - "anchor": "gitea_mcp_server.py:19691", + "anchor": "gitea_mcp_server.py:19851", "expect": "def gitea_audit_config" }, { - "anchor": "gitea_mcp_server.py:19713", + "anchor": "gitea_mcp_server.py:19873", "expect": "service_summaries(config)" }, { @@ -135,19 +135,19 @@ "expect": "_keychain_token(auth.get(\"id\"))" }, { - "anchor": "gitea_mcp_server.py:17776", + "anchor": "gitea_mcp_server.py:17909", "expect": "\"jenkins-mcp\"" }, { - "anchor": "gitea_mcp_server.py:17782", + "anchor": "gitea_mcp_server.py:17915", "expect": "external-mcp" }, { - "anchor": "gitea_mcp_server.py:17803", + "anchor": "gitea_mcp_server.py:17936", "expect": "\"glitchtip-mcp\"" }, { - "anchor": "gitea_mcp_server.py:17808", + "anchor": "gitea_mcp_server.py:17941", "expect": "external-mcp" }, { @@ -183,7 +183,7 @@ "expect": "mutation_safe" }, { - "anchor": "gitea_mcp_server.py:19171", + "anchor": "gitea_mcp_server.py:19331", "expect": "def gitea_assess_master_parity" }, { @@ -195,11 +195,11 @@ "expect": "AUTHOR_WORKTREE_ENV" }, { - "anchor": "gitea_mcp_server.py:2351", + "anchor": "gitea_mcp_server.py:2352", "expect": "/tmp/gitea_issue_lock.json" }, { - "anchor": "gitea_mcp_server.py:10956", + "anchor": "gitea_mcp_server.py:10957", "expect": "def gitea_bootstrap_author_issue_worktree" }, { @@ -239,7 +239,7 @@ "expect": "os.getpid()" }, { - "anchor": "gitea_mcp_server.py:12870", + "anchor": "gitea_mcp_server.py:12871", "expect": "owner_pid_alive" } ] diff --git a/docs/remote-mcp/threat-model.md b/docs/remote-mcp/threat-model.md index ddbb7a6..3d67961 100644 --- a/docs/remote-mcp/threat-model.md +++ b/docs/remote-mcp/threat-model.md @@ -31,7 +31,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:24864`, and its +bind it cited at line 23750 now lives at `gitea_mcp_server.py:25087`, 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 @@ -79,15 +79,15 @@ authenticate the *caller*, not the *intent*. | 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 single configured bind (`gitea_mcp_server.py:24864`) 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:15481`) or a refusal (`gitea_mcp_server.py:15519`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9191` and `gitea_mcp_server.py:9440`) | 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:25087`) 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:15630`) or a refusal (`gitea_mcp_server.py:15652`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9192` and `gitea_mcp_server.py:9441`) | 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: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:10956`); 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:12870`). 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. | +| 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:10957`); 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:12871`). A legacy global slot still exists at `gitea_mcp_server.py:2352`, 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:19171` | 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:19331` | Freshness defined against the deployed build identity, with an explicit fail-closed verdict when undeterminable. | ### What no boundary constrains @@ -132,10 +132,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:19327`) reports each profile's credential status by calling - `resolve_token` on it (`gitea_mcp_server.py:19378`), and `gitea_audit_config` - (`gitea_mcp_server.py:19691`) reports service credential status through - `service_summaries` (`gitea_mcp_server.py:19713`). + (`gitea_mcp_server.py:19487`) reports each profile's credential status by calling + `resolve_token` on it (`gitea_mcp_server.py:19538`), and `gitea_audit_config` + (`gitea_mcp_server.py:19851`) reports service credential status through + `service_summaries` (`gitea_mcp_server.py:19873`). 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. @@ -180,16 +180,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:19327`) called from the **author** session reports +(`gitea_mcp_server.py:19487`) 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:19378`). The author process does not merely *have access to* the +(`gitea_mcp_server.py:19538`). 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:19691`) reports `MDCPS Jenkins: enabled, read-only, authenticated`. -That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19713`, +(`gitea_mcp_server.py:19851`) reports `MDCPS Jenkins: enabled, read-only, authenticated`. +That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19873`, 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 @@ -197,8 +197,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:17776`, `gitea_mcp_server.py:17782`, `gitea_mcp_server.py:17803`, -`gitea_mcp_server.py:17808`) with their own expected tool sets (`mcp_discoverability.py:9`, +(`gitea_mcp_server.py:17909`, `gitea_mcp_server.py:17915`, `gitea_mcp_server.py:17936`, +`gitea_mcp_server.py:17941`) 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 @@ -209,8 +209,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:21601` and -`gitea_mcp_server.py:21645`, reached from `gitea_mcp_server.py:21581`. +server spawns — including the `ps` invocations at `gitea_mcp_server.py:21761` and +`gitea_mcp_server.py:21805`, reached from `gitea_mcp_server.py:21741`. **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, @@ -219,8 +219,8 @@ 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:19070`), -derived from environment inspection (`gitea_mcp_server.py:15481`) with the recognized-key +reported as exactly `client_managed` or `manual_launch` (`gitea_mcp_server.py:19217`), +derived from environment inspection (`gitea_mcp_server.py:15630`) 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 @@ -259,7 +259,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:19378` and `gitea_mcp_server.py:19713` must report +behavior behind `gitea_mcp_server.py:19538` and `gitea_mcp_server.py:19873` 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 @@ -336,11 +336,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:15481`) 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:15630`) 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: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:10956`), plus the shared stderr path at `mcp_server.py:13`. | 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:10957`), 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. | diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index e82a7bd..532b8be 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -199,6 +199,7 @@ RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE" import namespace_workspace_binding as nwb # noqa: E402 import canonical_repository_root as crr # noqa: E402 # #706 cross-repo canonical root import mcp_namespace_health # noqa: E402 +import mcp_worker_identity # noqa: E402 # #948 single client/session provenance authority import stale_binding_recovery # noqa: E402 # Worktree env bindings inherited from the parent environment at daemon boot @@ -14916,7 +14917,7 @@ def _stale_runtime_reconnect_action() -> str: return ( "blocker_kind=runtime_reconnect_required: call " "gitea_request_mcp_reconnect(namespace=, " - "reason='stale-runtime', client='codex') for a typed operator " + "reason='stale-runtime') for a typed operator " "reconnect blocker with exact UI steps, then reconnect the IDE/client " "MCP session so the server reloads at the current master head. Do not " "call gitea_activate_profile, pkill, touch configs, or switch MCP role " @@ -15478,34 +15479,166 @@ def _session_context_mutation_block( return blocked -def _is_client_managed_process() -> bool: - """Check whether the current MCP server process has client-managed launch provenance (#686).""" - val = ( - os.environ.get("GITEA_CLIENT_MANAGED") - or os.environ.get("GITEA_MCP_CLIENT_MANAGED") - or os.environ.get("GITEA_SERVER_PROVENANCE") - or os.environ.get("GITEA_FORCE_CLIENT_MANAGED") - or "" - ).strip().lower() +# --- #948 client/session runtime ownership ------------------------------- +# +# One daemon launch is one *generation*. The session that owns it registers a +# unique worker identity against it, so a second healthy client attaching its +# own generation is no longer indistinguishable from a duplicate process. +# Everything here is process-local cache plus a local SQLite registry: no Gitea +# call, no network, no config write, and every failure degrades to "unproven" +# rather than raising into a tool call. - if val in ("0", "false", "no", "manual", "manual_launch"): +_WORKER_REGISTRY = None +_WORKER_IDENTITY: str | None = None +_WORKER_GENERATION: str | None = None +_WORKER_REGISTRATION_ATTEMPTED = False + +#: Env a client launcher may set to name itself and its session. Absent values +#: are reported as unknown; they are never guessed at, because guessing is what +#: produced Codex reconnect steps for a Gemini operator. +CLIENT_NAME_ENV = "GITEA_MCP_CLIENT" +CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE" +CLIENT_SESSION_ENV = "GITEA_MCP_CLIENT_SESSION" + + +def _worker_registry(): + """Local worker registry, or ``None`` when it cannot be opened. + + Under pytest the registry is only opened when a test has pinned a path, so + a test run never writes into the operator's real registry. + """ + global _WORKER_REGISTRY + if _WORKER_REGISTRY is not None: + return _WORKER_REGISTRY + if mcp_daemon_guard.is_pytest_runtime() and not ( + os.environ.get(mcp_worker_identity.REGISTRY_PATH_ENV) or "" + ).strip(): + return None + try: + _WORKER_REGISTRY = mcp_worker_identity.WorkerRegistry() + except Exception: + return None + return _WORKER_REGISTRY + + +def _client_identity_hints() -> dict: + """What the launcher told us about itself. Unset fields stay unset.""" + return { + "client_name": (os.environ.get(CLIENT_NAME_ENV) or "").strip() or None, + "client_instance_id": (os.environ.get(CLIENT_INSTANCE_ENV) or "").strip() + or f"pid-{os.getpid()}", + "session_id": (os.environ.get(CLIENT_SESSION_ENV) or "").strip() + or f"proc-{os.getpid()}-{_process_boot_head_sha or 'nohead'}", + } + + +def _active_worker_identity() -> str | None: + """This runtime's worker identity, registering it once per process. + + A collision does not adopt the existing registration: a fresh identity is + minted and registered instead, which is the #948 AC32 requirement and also + the only safe response — adopting would silently transfer another worker's + leases and fencing tokens. + """ + global _WORKER_IDENTITY, _WORKER_GENERATION, _WORKER_REGISTRATION_ATTEMPTED + if _WORKER_IDENTITY is not None: + return _WORKER_IDENTITY + if _WORKER_REGISTRATION_ATTEMPTED: + return None + _WORKER_REGISTRATION_ATTEMPTED = True + + registry = _worker_registry() + if registry is None: + return None + + hints = _client_identity_hints() + generation = _WORKER_GENERATION or mcp_worker_identity.new_generation_id(os.getpid()) + native = mcp_daemon_guard.native_runtime_status() + + try: + for _attempt in range(3): + identity = mcp_worker_identity.generate_worker_identity( + hints["client_name"], hints["session_id"] + ) + outcome = registry.register( + worker_identity=identity, + client_name=hints["client_name"], + client_instance_id=hints["client_instance_id"], + session_id=hints["session_id"], + generation_id=generation, + role=_active_role_kind_safe(), + profile=(os.environ.get(gitea_config.ENV_PROFILE) or "").strip() or None, + remote=(os.environ.get("GITEA_MCP_REMOTE") or "").strip() or None, + repository_binding=PROJECT_ROOT, + pid=os.getpid(), + transport=native.get("bound_transport"), + token_fingerprint=native.get("token_fingerprint"), + pid_alive_probe=issue_lock_store.is_process_alive, + ) + if outcome.get("registered"): + _WORKER_IDENTITY = identity + _WORKER_GENERATION = generation + return identity + if not outcome.get("collision"): + return None + # Collided: loop mints a different identity rather than reusing this + # one. The existing registration is left exactly as it was. + except Exception: + return None + return None + + +def _active_role_kind_safe() -> str | None: + """Best-effort role for the registry record; never raises into a tool call. + + The role is recorded for diagnostics only. It is deliberately not part of + the identity: role is a reusable capability definition that many live + workers may share (#948 AC37). + """ + try: + profile = get_profile() or {} + return _role_kind( + profile.get("allowed_operations") or [], + profile.get("forbidden_operations") or [], + ) + except Exception: + return None + + +def _stdin_is_tty() -> bool: + """Terminal evidence for launch provenance, tolerant of a detached stdin.""" + try: + return bool(sys.stdin and sys.stdin.isatty()) + except Exception: return False - if val in ("1", "true", "yes", "client_managed"): - return True - # A terminal launch has an active TTY on stdin - try: - if sys.stdin and sys.stdin.isatty(): - return False - except Exception: - pass +def _launch_provenance() -> dict: + """This process's launch provenance, from the one shared authority (#948). - # Standard client launch or test runner with stdio pipe and profile env - if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ: - return True + Previously each surface reimplemented this. ``gitea_get_runtime_context`` + read the live environment while ``mcp_namespace_health`` read a filtered + summary that structurally could not see the provenance keys, so the two + reported different provenance for the same process. Both now call + ``mcp_worker_identity``. + """ + return mcp_worker_identity.assess_launch_provenance( + dict(os.environ), stdin_is_tty=_stdin_is_tty() + ) - return False + +def _is_client_managed_process() -> bool: + """Check whether the current MCP server process has client-managed launch provenance (#686). + + #948: the decision logic moved to + ``mcp_worker_identity.assess_launch_provenance`` unchanged — same env keys, + same precedence, same TTY fallback — so this keeps returning exactly what it + always returned while no longer being a second, divergent implementation. + Launch provenance answers "was this hand-launched from a terminal", which is + the #686 question. It does *not* answer which live session owns this + runtime; that is ``session_ownership`` and needs an attachment record. + """ + return bool(_launch_provenance()["client_managed"]) def _provenance_mutation_block(**extra_fields) -> dict | None: @@ -19048,7 +19181,21 @@ def gitea_get_runtime_context( source="gitea_get_runtime_context", ) - is_client_managed = _is_client_managed_process() + # #948: one assessment, shared with mcp_namespace_health. Reporting both + # dimensions from a single call is what makes the two surfaces agree — the + # contradiction they used to produce came from two implementations, not from + # two genuinely different observations. + provenance_assessment = mcp_worker_identity.assess_provenance( + registry=_worker_registry(), + worker_identity=_active_worker_identity(), + env=dict(os.environ), + native_transport_bound=mcp_daemon_guard.bound_transport() is not None, + profile=profile.get("profile_name"), + role=_role_kind(allowed, forbidden), + stdin_is_tty=_stdin_is_tty(), + pid_alive_probe=issue_lock_store.is_process_alive, + ) + is_client_managed = provenance_assessment["is_client_managed"] unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides() result = { @@ -19067,8 +19214,21 @@ def gitea_get_runtime_context( "review_merge_blocked_reasons": blocked_reasons, "suggested_fix": suggested_fix, "safe_next_action": safe_next_action, - "server_provenance": "client_managed" if is_client_managed else "manual_launch", + "server_provenance": provenance_assessment["launch_provenance"], "is_client_managed": is_client_managed, + # #948: the ownership dimension, which the environment cannot establish. + # A surface that needs "may this session mutate on behalf of its client" + # reads these, not server_provenance. + "session_ownership": provenance_assessment["session_ownership"], + "session_owned": provenance_assessment["session_owned"], + "worker_identity": provenance_assessment["worker_identity"], + "session_id": provenance_assessment["session_id"], + "client_instance_id": provenance_assessment["client_instance_id"], + "generation_id": provenance_assessment["generation_id"], + "client_name": provenance_assessment["client_name"], + "fencing_epoch": provenance_assessment["fencing_epoch"], + "conflicting_live_sessions": provenance_assessment["conflicting_live_sessions"], + "provenance_assessment": provenance_assessment, "unconsumed_gitea_env": unconsumed_env, "preflight_ready": preflight["preflight_ready"], "preflight_block_reasons": preflight["preflight_block_reasons"], @@ -21654,11 +21814,23 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> if match: profile = match.group(1) + # #948: provenance for a scanned peer comes from the shared authority, + # fed by that peer's own environment, so the fleet scan cannot disagree + # with what that peer reports about itself. + peer_env = { + m.group(1): m.group(2) + for m in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out) + } + # declared_only: this is a peer process. Its stdin is not ours to + # inspect, and it inherits GITEA_MCP_PROFILE from any shell that + # exported it, so only an explicit declaration counts as evidence here. is_client_managed = bool( - re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE) - or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE) - or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE) + mcp_worker_identity.assess_launch_provenance( + peer_env, declared_only=True + )["client_managed"] ) + peer_worker_identity = peer_env.get("GITEA_MCP_WORKER_IDENTITY") + peer_generation = peer_env.get("GITEA_MCP_GENERATION_ID") for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out): k, v = env_match.group(1), env_match.group(2) @@ -21674,6 +21846,8 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> "start_time": start_time, "is_stale": is_stale, "is_client_managed": is_client_managed, + "worker_identity": peer_worker_identity, + "generation_id": peer_generation, } if profile not in all_profile_procs: all_profile_procs[profile] = [] @@ -21681,12 +21855,44 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> running_profiles = {} for profile, procs in all_profile_procs.items(): - if len(procs) > 1: + # #948 AC40/AC43: sharing a profile is legitimate — profile is a + # reusable capability definition, not a worker identity. What is *not* + # legitimate is reusing one worker identity, or two live sessions + # claiming one generation. Distinctness has to be proven, though: + # processes carrying no identity evidence are indistinguishable, so + # they stay classified as duplicates and keep the #686 wall intact. + identified = [p for p in procs if p.get("worker_identity")] + distinct_identities = {p["worker_identity"] for p in identified} + all_identified = len(identified) == len(procs) + duplicate_identity = len(identified) != len(distinct_identities) + contested_generation = any( + len({p["worker_identity"] for p in identified if p.get("generation_id") == gen}) > 1 + for gen in {p.get("generation_id") for p in identified if p.get("generation_id")} + ) + + if len(procs) > 1 and ( + not all_identified or duplicate_identity or contested_generation + ): pids_str = ", ".join(str(p["pid"]) for p in procs) + if duplicate_identity or contested_generation: + detail = ( + "The same worker identity or generation is claimed more than once, " + "so these are genuine duplicates rather than independent workers." + ) + else: + detail = ( + "Manual or duplicate launches defeat staleness detection and cannot " + "receive client stdio." + ) reasons.append( f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). " - "Manual or duplicate launches defeat staleness detection and cannot receive client stdio." + + detail ) + # Otherwise: several independently identified workers share one profile. + # No reason is appended, deliberately. Every reason this function + # returns is raised as a hard RuntimeError by its callers, so recording + # legitimate concurrency here as "informational" would block exactly the + # case #948 exists to permit. client_procs = [p for p in procs if p["is_client_managed"]] if client_procs: client_procs.sort(key=lambda p: p["start_time"], reverse=True) @@ -22095,7 +22301,7 @@ def gitea_resolve_task_capability( next_safe_action = ( "blocker_kind=runtime_reconnect_required: call " "gitea_request_mcp_reconnect(namespace=, " - "reason='stale-runtime', client='codex') for a typed operator " + "reason='stale-runtime') for a typed operator " "blocker with exact UI steps, then reconnect/reload the IDE-managed " "Gitea MCP server for this profile so it reloads current master. " "Do not edit mcp_config.json by hand, pkill, or touch configs; the " @@ -23663,7 +23869,7 @@ def gitea_workflow_dashboard( def gitea_request_mcp_reconnect( namespace: str | None = None, reason: str | None = None, - client: str = "codex", + client: str | None = None, remote: str = "dadeschools", host: str | None = None, session_id: str | None = None, @@ -23693,8 +23899,12 @@ def gitea_request_mcp_reconnect( to the active profile's inferred namespace. reason: Why reconnect is requested: ``stale-runtime``, ``transport_eof``, ``missing_namespace``, ``not_required``, or free-form (normalized). - client: Operator UI surface — ``codex`` (default), ``claude_code``, or - ``generic``. + client: Operator UI surface — ``codex``, ``claude_code``, or + ``generic``. #948: omitting it resolves the client from this + runtime's live attachment record rather than assuming one vendor, + and falls back to ``generic`` when nothing identifies the client. + Emitting Codex panel steps to a Gemini/Antigravity operator left + them with no reachable recovery path. remote: Known instance — ``dadeschools`` or ``prgs`` (parity context). host: Optional host override for parity context. session_id: Optional session id to echo in the report. @@ -23754,6 +23964,19 @@ def gitea_request_mcp_reconnect( else: effective_reason = mcp_client_reconnect.REASON_UNSPECIFIED + # #948: an omitted client is resolved from the live attachment record, so + # the steps describe the UI the operator is actually in front of. + if not (client or "").strip(): + client = mcp_worker_identity.reconnect_client_for( + mcp_worker_identity.assess_provenance( + registry=_worker_registry(), + worker_identity=_active_worker_identity(), + env=dict(os.environ), + stdin_is_tty=_stdin_is_tty(), + pid_alive_probe=issue_lock_store.is_process_alive, + ) + ) + payload = mcp_client_reconnect.build_reconnect_request( namespace=ns, profile=profile_name, diff --git a/mcp_client_reconnect.py b/mcp_client_reconnect.py index 2283d0a..15baa9a 100644 --- a/mcp_client_reconnect.py +++ b/mcp_client_reconnect.py @@ -92,7 +92,15 @@ OPERATOR_UI_STEPS: dict[str, tuple[str, ...]] = { ), } -DEFAULT_CLIENT = "codex" +#: What an *unidentified* client gets. #948: this is deliberately the +#: host-agnostic step set rather than a specific product. Defaulting to one +#: vendor emitted Codex UI steps to a Gemini/Antigravity operator, who then had +#: no reachable recovery path — the guidance named a panel they do not have. +DEFAULT_CLIENT = "generic" + +#: The historical default, kept addressable by name so Codex callers still get +#: Codex steps, without it silently becoming the fallback for unknown clients. +LEGACY_DEFAULT_CLIENT = "codex" def normalize_reason(reason: str | None) -> str: @@ -128,14 +136,18 @@ def normalize_reason(reason: str | None) -> str: def normalize_client(client: str | None) -> str: - """Return a known client key for operator UI steps.""" - text = (client or "").strip().lower().replace(" ", "_").replace("-", "_") - if text in ("codex", "openai_codex", "openai"): - return "codex" - if text in ("claude", "claude_code", "claude_desktop", "anthropic"): - return "claude_code" - if text in OPERATOR_UI_STEPS: - return text + """Return the UI-step key for a client. + + #948: alias resolution is shared with ``mcp_worker_identity`` so a client + name means the same thing wherever it is read. A name we recognise but have + no bespoke steps for — Gemini, Antigravity, Grok — resolves to the generic + host-agnostic steps rather than to another vendor's panel. + """ + import mcp_worker_identity + + canonical = mcp_worker_identity.normalize_client_name(client) + if canonical in OPERATOR_UI_STEPS: + return canonical return DEFAULT_CLIENT diff --git a/mcp_namespace_health.py b/mcp_namespace_health.py index e8e4910..ee3d7cf 100644 --- a/mcp_namespace_health.py +++ b/mcp_namespace_health.py @@ -95,6 +95,15 @@ SAFE_ENV_KEYS = ( "GITEA_MCP_CONFIG", ) +# #948: provenance used to be derived from the summary this allowlist produces. +# The allowlist never carried a provenance key, so that derivation could only +# ever evaluate to ``manual_launch`` — whatever the process actually was — while +# ``gitea_get_runtime_context`` read the live environment and reported +# ``client_managed`` for the same process. Provenance is no longer derived here. +# It comes from ``mcp_worker_identity.assess_provenance``, the single authority +# every surface shares. This allowlist keeps its original and only job: deciding +# which env values are safe to echo back in diagnostics. + def assess_connected_namespace_attachment( *, @@ -443,12 +452,23 @@ def classify_namespace_probe( profile: str | None = None, configured: bool = True, probe_source: str | None = None, + worker_identity: str | None = None, + generation_id: str | None = None, + registry: Any | None = None, + pid_alive_probe: Any | None = None, ) -> dict[str, Any]: """Classify whether a required tool is callable through a live namespace. ``registered_tools`` is static/server-side evidence. ``probe_result`` is live invocation evidence. Only ``probe_source=client_namespace`` proves the IDE-managed path; ``offline_spawn`` is an offline subprocess check only. + + #948: ``worker_identity``/``generation_id``/``registry`` carry the + client/session ownership evidence. Provenance is resolved by + ``mcp_worker_identity.assess_provenance`` — the same call + ``gitea_get_runtime_context`` makes — so the two surfaces cannot report + different provenance for one process. Omitting them yields the fail-closed + ``unproven`` verdict, never a fabricated ``client_managed``. """ ns = (namespace or "").strip() tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami" @@ -565,14 +585,31 @@ def classify_namespace_probe( blocks = namespace_health_blocks_task("merge_pr", healthy) import gitea_config + import mcp_worker_identity + raw_env = process.get("env") if isinstance(process, dict) else None unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env) - is_client_managed = bool( - env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed") - or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed") - or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed" + + # #948: one authority, shared with gitea_get_runtime_context. The env is + # passed whole rather than through SAFE_ENV_KEYS — the allowlist exists to + # decide what may be *echoed*, and using it to decide what may be *believed* + # is what made this surface structurally unable to report client_managed. + # ``declared_only``: ``process`` describes an observed peer, not this + # interpreter. Its stdin is unavailable and its launcher-config env is + # inherited from whatever shell started it, so only an explicit declaration + # is evidence. Absence of one is ``unproven``, not an asserted manual launch. + provenance_verdict = mcp_worker_identity.assess_provenance( + registry=registry, + worker_identity=worker_identity, + generation_id=generation_id, + env=raw_env if isinstance(raw_env, dict) else {}, + namespace=ns, + profile=profile_name, + pid_alive_probe=pid_alive_probe, + declared_only=True, ) - provenance = "client_managed" if is_client_managed else "manual_launch" + provenance = provenance_verdict["provenance"] + is_client_managed = provenance_verdict["is_client_managed"] return { "success": healthy, @@ -591,6 +628,15 @@ def classify_namespace_probe( "remediation": remediation, "provenance": provenance, "is_client_managed": is_client_managed, + # Every non-client-session verdict fails closed. Consumers that only + # need "may this mutate?" read this and stay correct across the #948 + # vocabulary split between ``manual_launch`` and ``unproven``. + "provenance_fail_closed": provenance_verdict["fail_closed"], + "provenance_assessment": provenance_verdict, + "worker_identity": provenance_verdict["worker_identity"], + "session_id": provenance_verdict["session_id"], + "generation_id": provenance_verdict["generation_id"], + "client_name": provenance_verdict["client_name"], "unconsumed_gitea_env": unconsumed_env, "diagnostics": { "namespace": ns, @@ -602,6 +648,13 @@ def classify_namespace_probe( "probe_source": source, "provenance": provenance, "is_client_managed": is_client_managed, + "provenance_fail_closed": provenance_verdict["fail_closed"], + "provenance_blocker_kind": provenance_verdict["blocker_kind"], + "provenance_scope": provenance_verdict["scope"], + "worker_identity": provenance_verdict["worker_identity"], + "session_id": provenance_verdict["session_id"], + "generation_id": provenance_verdict["generation_id"], + "client_name": provenance_verdict["client_name"], "unconsumed_gitea_env": unconsumed_env, }, "blocks_merge_workflow": blocks, diff --git a/mcp_worker_identity.py b/mcp_worker_identity.py new file mode 100644 index 0000000..ea66096 --- /dev/null +++ b/mcp_worker_identity.py @@ -0,0 +1,1432 @@ +"""Client/session-aware runtime ownership and provenance (#948). + +Before this module the control plane had two disagreeing provenance models. +``gitea_get_runtime_context`` read the *live* process environment, while +``mcp_namespace_health.classify_namespace_probe`` read a filtered env summary +whose allowlist never carried the provenance keys — so one surface reported +``client_managed`` while the other reported ``manual_launch`` for the very same +process. Neither model could name *which* client or *which* session owned the +runtime, so a healthy daemon serving a second client looked like a duplicate. + +This module is the single authority both surfaces now consult. The model it +implements: + +* A **worker identity** is unique and is *not* a role or a profile. Roles and + profiles are reusable capability definitions that many live workers may share + concurrently; exclusivity belongs to an active lease target, never to a role. +* A **generation** is one daemon launch. Exactly one live session may claim a + generation at a time; a takeover mints a higher fencing epoch so the prior + session is fenced if it resumes. +* **Trusted evidence** is a live attachment record binding + ``(client_instance_id, session_id, generation_id)`` to the running process. + A process environment flag is corroborating evidence at best: it is set by + whoever launched the process and says nothing about which session owns it + *now*, so it can never establish ownership on its own. +* Missing or self-contradictory evidence fails closed, and the refusal is + scoped to the affected worker identity rather than to every worker sharing + its profile. +""" + +from __future__ import annotations + +import hashlib +import os +import re +import secrets +import sqlite3 +import threading +import time +from contextlib import contextmanager +from datetime import datetime, timezone +from typing import Any, Iterator + + +# --- Provenance verdicts ------------------------------------------------- + +#: A live attachment record binds this runtime to a client session. +PROVENANCE_CLIENT_SESSION = "client_session_attached" +#: Positive evidence that the process was launched outside any client session. +PROVENANCE_MANUAL = "manual_launch" +#: No trusted evidence either way — fails closed, distinct from "manual". +PROVENANCE_UNPROVEN = "unproven" + +VALID_PROVENANCE = frozenset( + {PROVENANCE_CLIENT_SESSION, PROVENANCE_MANUAL, PROVENANCE_UNPROVEN} +) + +# --- Evidence kinds ------------------------------------------------------ + +EVIDENCE_ATTACHMENT_RECORD = "live_attachment_record" +EVIDENCE_NATIVE_TRANSPORT = "bound_native_transport" +EVIDENCE_PROCESS_LIVENESS = "process_liveness" +#: Never sufficient alone. Named so a report can say *why* it was not enough. +EVIDENCE_ENV_FLAG = "process_env_flag" + +#: Evidence kinds that, on their own, establish current session ownership. +SUFFICIENT_EVIDENCE = frozenset({EVIDENCE_ATTACHMENT_RECORD}) + +# --- Blocker kinds ------------------------------------------------------- + +BLOCKER_NONE = "none" +BLOCKER_NO_ATTACHMENT = "session_attachment_missing" +BLOCKER_CONFLICTING_SESSIONS = "conflicting_live_sessions" +BLOCKER_FENCED = "superseded_generation_fenced" +BLOCKER_IDENTITY_COLLISION = "worker_identity_collision" +BLOCKER_CONTRADICTORY = "contradictory_provenance_evidence" + +# --- Client naming ------------------------------------------------------- + +#: Returned when no client name is supplied or the name is unusable. There is +#: deliberately no default vendor here: defaulting an unknown client to a +#: specific product emits reconnect steps for a UI the operator is not looking +#: at, which is the #948 hardcoded-``codex`` defect. +UNKNOWN_CLIENT = "unknown_client" + +#: Canonical spellings for clients whose aliases we recognise. Absence from +#: this map is not an error — an unrecognised name is slugified and kept, so a +#: new client is named accurately rather than relabelled as a known one. +CLIENT_ALIASES: dict[str, str] = { + "codex": "codex", + "openai": "codex", + "openai_codex": "codex", + "claude": "claude_code", + "claude_code": "claude_code", + "claude_desktop": "claude_code", + "anthropic": "claude_code", + "gemini": "gemini", + "google": "gemini", + "antigravity": "antigravity", + "gemini_antigravity": "antigravity", + "grok": "grok", + "xai": "grok", +} + +_SLUG_RE = re.compile(r"[^a-z0-9]+") + +#: ``--`` per the #948 scope clarification. +WORKER_IDENTITY_RE = re.compile( + r"^(?P[a-z0-9][a-z0-9_]*)-(?P\d{8}T\d{6}Z)-(?P[0-9a-f]{12})$" +) + +IDENTITY_TIMESTAMP_FORMAT = "%Y%m%dT%H%M%SZ" +IDENTITY_DIGEST_LENGTH = 12 + +# --- Registry defaults --------------------------------------------------- + +REGISTRY_PATH_ENV = "GITEA_WORKER_REGISTRY_DB" +DEFAULT_REGISTRY_PATH = os.path.expanduser( + "~/.cache/gitea-tools/control-plane/worker_registry.sqlite3" +) + +#: How long a registration stays live without a heartbeat. Chosen to outlast a +#: full-suite run inside one tool call while still releasing an abandoned +#: worker within a single operator coffee break. +DEFAULT_HEARTBEAT_TTL_SECONDS = 900.0 + +STATUS_ACTIVE = "active" +STATUS_SUPERSEDED = "superseded" +STATUS_RELEASED = "released" + +_TRUE_VALUES = frozenset({"1", "true", "yes", "client_managed"}) +_FALSE_VALUES = frozenset({"0", "false", "no", "manual", "manual_launch"}) + +#: Environment keys that historically stood in for provenance. They are read +#: only as corroboration; see :func:`env_provenance_signal`. +PROVENANCE_ENV_KEYS = ( + "GITEA_CLIENT_MANAGED", + "GITEA_MCP_CLIENT_MANAGED", + "GITEA_SERVER_PROVENANCE", + "GITEA_MCP_SANCTIONED_DAEMON", +) + +#: Explicit launch-provenance declarations, highest precedence first. Mirrors +#: the historical ``_is_client_managed_process`` precedence exactly, so +#: unifying the two models changes *who decides*, not *what is decided*. +LAUNCH_DECLARATION_ENV_KEYS = ( + "GITEA_CLIENT_MANAGED", + "GITEA_MCP_CLIENT_MANAGED", + "GITEA_SERVER_PROVENANCE", + "GITEA_FORCE_CLIENT_MANAGED", +) + +#: Env keys a client launcher sets that imply a non-terminal launch. +LAUNCH_CONFIG_ENV_KEYS = ( + "GITEA_MCP_CONFIG", + "GITEA_MCP_PROFILE", + "GITEA_PROFILE_NAME", +) + +# --- Launch provenance verdicts ------------------------------------------ + +#: Launch-dimension values. These keep the literals the pre-#948 surfaces +#: already published, so a consumer reading ``server_provenance`` sees the same +#: strings it always did. +LAUNCH_CLIENT_MANAGED = "client_managed" +LAUNCH_MANUAL = PROVENANCE_MANUAL +LAUNCH_UNPROVEN = PROVENANCE_UNPROVEN + +# --- Session ownership verdicts ------------------------------------------ + +#: A live attachment record binds this runtime to exactly one client session. +OWNERSHIP_OWNED = "owned" +#: No live attachment record. Fails closed for ownership-scoped decisions. +OWNERSHIP_UNOWNED = "unowned" +#: More than one live session claims the generation. Fails closed. +OWNERSHIP_CONTESTED = "contested" + + +_SCHEMA_SQL = """ +CREATE TABLE IF NOT EXISTS worker_registrations ( + worker_identity TEXT PRIMARY KEY, + client_name TEXT NOT NULL, + client_instance_id TEXT NOT NULL, + session_id TEXT NOT NULL, + generation_id TEXT NOT NULL, + role TEXT, + profile TEXT, + namespace TEXT, + remote TEXT, + repository_binding TEXT, + pid INTEGER, + transport TEXT, + token_fingerprint TEXT, + started_at TEXT NOT NULL, + last_heartbeat_at TEXT NOT NULL, + heartbeat_ttl_seconds REAL NOT NULL, + fencing_epoch INTEGER NOT NULL, + status TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_worker_generation + ON worker_registrations(generation_id, status); +CREATE INDEX IF NOT EXISTS idx_worker_session + ON worker_registrations(session_id, status); +CREATE INDEX IF NOT EXISTS idx_worker_profile + ON worker_registrations(profile, status); +""" + + +class WorkerRegistryError(RuntimeError): + """Raised for registry misuse that is a programming error, not a refusal.""" + + +def _utc_now() -> datetime: + return datetime.now(timezone.utc) + + +def _ts(value: datetime) -> str: + return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _parse_ts(value: str | None) -> datetime | None: + if not value: + return None + try: + return datetime.strptime(value, "%Y-%m-%dT%H:%M:%SZ").replace( + tzinfo=timezone.utc + ) + except (TypeError, ValueError): + return None + + +def default_registry_path() -> str: + raw = (os.environ.get(REGISTRY_PATH_ENV) or DEFAULT_REGISTRY_PATH).strip() + return raw or DEFAULT_REGISTRY_PATH + + +# --- Identity ------------------------------------------------------------ + + +def normalize_client_name(raw: str | None) -> str: + """Canonicalise a client name without inventing one. + + An empty or unusable name yields :data:`UNKNOWN_CLIENT` rather than a + specific product, so downstream reconnect guidance describes a UI the + operator is actually looking at (or says it does not know) instead of + always naming one vendor. + """ + text = _SLUG_RE.sub("_", (raw or "").strip().lower()).strip("_") + if not text: + return UNKNOWN_CLIENT + return CLIENT_ALIASES.get(text, text) + + +def worker_identity_digest( + client_name: str, + session_id: str, + timestamp_ns: int, + nonce: str, +) -> str: + """Collision-resistant digest over client, session, high-resolution time, nonce.""" + payload = f"{client_name}\x1f{session_id}\x1f{timestamp_ns}\x1f{nonce}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:IDENTITY_DIGEST_LENGTH] + + +def generate_worker_identity( + client_name: str | None, + session_id: str | None, + *, + timestamp_ns: int | None = None, + nonce: str | None = None, + now: datetime | None = None, +) -> str: + """Mint ``--`` (#948 AC27-29). + + ``timestamp_ns`` and ``nonce`` are injectable so a test can force a + collision deliberately; production callers omit both and get high-resolution + time plus a fresh random nonce. + """ + client = normalize_client_name(client_name) + session = (session_id or "").strip() or "no_session" + ts_ns = int(timestamp_ns if timestamp_ns is not None else time.time_ns()) + seed = nonce if nonce is not None else secrets.token_hex(16) + stamp = (now or _utc_now()).astimezone(timezone.utc).strftime( + IDENTITY_TIMESTAMP_FORMAT + ) + digest = worker_identity_digest(client, session, ts_ns, seed) + return f"{client}-{stamp}-{digest}" + + +def parse_worker_identity(identity: str | None) -> dict[str, Any]: + """Structurally validate a worker identity.""" + text = (identity or "").strip() + match = WORKER_IDENTITY_RE.match(text) + if not match: + return { + "valid": False, + "worker_identity": text or None, + "client_name": None, + "minted_at": None, + "digest": None, + "reasons": [ + "worker identity must match --, " + "e.g. gemini-20260729T061500Z-1a2b3c4d5e6f" + ], + } + return { + "valid": True, + "worker_identity": text, + "client_name": match.group("client"), + "minted_at": match.group("ts"), + "digest": match.group("digest"), + "reasons": [], + } + + +def new_generation_id(pid: int | None = None) -> str: + """Mint a boot epoch for one daemon launch.""" + return f"gen-{int(pid if pid is not None else os.getpid())}-{secrets.token_hex(8)}" + + +# --- Environment signal -------------------------------------------------- + + +def env_provenance_signal(env: dict[str, str] | None = None) -> dict[str, Any]: + """Read the legacy env provenance flags as *corroborating* signal only. + + Returns the raw reading plus an explicit statement that it does not + establish ownership. Callers must not branch on ``value`` alone; #948 + requires a live attachment record for that. + """ + source = os.environ if env is None else env + present: dict[str, str] = {} + for key in PROVENANCE_ENV_KEYS: + raw = source.get(key) + if raw not in (None, ""): + present[key] = str(raw) + + value: bool | None = None + for key in PROVENANCE_ENV_KEYS: + raw = (present.get(key) or "").strip().lower() + if not raw: + continue + if raw in _FALSE_VALUES: + value = False + break + if raw in _TRUE_VALUES: + value = True + break + + return { + "evidence_kind": EVIDENCE_ENV_FLAG, + "keys_present": sorted(present), + "value": value, + "proves_session_ownership": False, + "reason": ( + "process environment flags are set by whoever launched the process and " + "carry no session binding; they corroborate a live attachment record but " + "never establish current ownership on their own (#948)" + ), + } + + +def assess_declared_launch_provenance(env: dict[str, str] | None = None) -> dict[str, Any]: + """Launch provenance for a process we are observing from the outside. + + Deliberately stricter than :func:`assess_launch_provenance`. For a *peer* + process there is no stdin to inspect, and launcher-configuration env is not + evidence of anything: a shell that exports ``GITEA_MCP_PROFILE`` passes it + to a hand-launched server too. So only an explicit declaration counts, and + its absence is reported as ``unproven`` rather than asserted as a manual + launch — naming a proof we do not have was the misdiagnosis #948 reports. + """ + source = os.environ if env is None else env + for key in LAUNCH_DECLARATION_ENV_KEYS: + raw = (source.get(key) or "").strip().lower() + if not raw: + continue + if raw in _FALSE_VALUES: + return { + "launch_provenance": LAUNCH_MANUAL, + "client_managed": False, + "decided_by": f"env:{key}", + "reasons": [f"{key}={raw!r} explicitly declares a manual launch"], + } + if raw in _TRUE_VALUES: + return { + "launch_provenance": LAUNCH_CLIENT_MANAGED, + "client_managed": True, + "decided_by": f"env:{key}", + "reasons": [ + f"{key}={raw!r} explicitly declares a client-managed launch" + ], + } + return { + "launch_provenance": LAUNCH_UNPROVEN, + "client_managed": False, + "decided_by": "no_declaration", + "reasons": [ + "no explicit client-managed declaration on the observed process; " + "launcher-configuration env is not proof for a peer process because a " + "hand-launched server inherits it from the shell" + ], + } + + +def assess_launch_provenance( + env: dict[str, str] | None = None, + *, + stdin_is_tty: bool | None = None, + declared_only: bool = False, +) -> dict[str, Any]: + """Decide *how the process was launched* — a different question from ownership. + + #948 separates two claims the old code ran together: + + * **Launch provenance** — was this started by a client, or hand-launched from + a terminal? This is the #686 concern and it is what the environment can + legitimately answer, because the launcher is exactly what sets it. + * **Session ownership** — which live client session owns this runtime *now*? + The environment cannot answer that; only a live attachment record can. + + Conflating them is what let one surface call a process ``client_managed`` + while another called the same process ``manual_launch``. This function is now + the only implementation of the first claim, so every surface gets one answer. + + ``declared_only`` selects the strategy for an *observed peer* process, where + stdin cannot be inspected and launcher-config env proves nothing. Callers + assessing their own process leave it false. + """ + if declared_only: + return assess_declared_launch_provenance(env) + + source = os.environ if env is None else env + reasons: list[str] = [] + + for key in LAUNCH_DECLARATION_ENV_KEYS: + raw = (source.get(key) or "").strip().lower() + if not raw: + continue + if raw in _FALSE_VALUES: + return { + "launch_provenance": LAUNCH_MANUAL, + "client_managed": False, + "decided_by": f"env:{key}", + "reasons": [f"{key}={raw!r} explicitly declares a manual launch"], + } + if raw in _TRUE_VALUES: + return { + "launch_provenance": LAUNCH_CLIENT_MANAGED, + "client_managed": True, + "decided_by": f"env:{key}", + "reasons": [f"{key}={raw!r} explicitly declares a client-managed launch"], + } + + if stdin_is_tty: + return { + "launch_provenance": LAUNCH_MANUAL, + "client_managed": False, + "decided_by": "stdin_tty", + "reasons": [ + "stdin is an interactive terminal, so this process cannot be " + "receiving client stdio" + ], + } + + present = [k for k in LAUNCH_CONFIG_ENV_KEYS if (source.get(k) or "").strip()] + if present: + return { + "launch_provenance": LAUNCH_CLIENT_MANAGED, + "client_managed": True, + "decided_by": "env:launcher_config", + "reasons": [ + f"launcher-supplied configuration env present ({', '.join(present)}) " + "with no terminal on stdin" + ], + } + + reasons.append( + "no launch declaration, no launcher configuration env, and no terminal " + "evidence; launch provenance is unproven rather than assumed manual" + ) + return { + "launch_provenance": LAUNCH_UNPROVEN, + "client_managed": False, + "decided_by": "no_evidence", + "reasons": reasons, + } + + +# --- Registry ------------------------------------------------------------ + + +class WorkerRegistry: + """Atomic, fail-closed registry of live client/session runtime ownership. + + Backed by SQLite with ``BEGIN IMMEDIATE`` so two clients racing to register + the same identity, or to claim the same generation, resolve to exactly one + winner across processes. + """ + + def __init__(self, db_path: str | None = None) -> None: + self.db_path = (db_path or default_registry_path()).strip() + parent = os.path.dirname(self.db_path) + if parent: + os.makedirs(parent, mode=0o700, exist_ok=True) + self._lock = threading.RLock() + self._init_schema() + + # -- plumbing -- + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path, timeout=30) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode = WAL") + return conn + + @contextmanager + def _tx(self) -> Iterator[sqlite3.Connection]: + with self._lock: + conn = self._connect() + try: + conn.execute("BEGIN IMMEDIATE") + yield conn + conn.commit() + except Exception: + try: + conn.rollback() + except sqlite3.Error: + pass + raise + finally: + conn.close() + + def _init_schema(self) -> None: + with self._lock: + conn = self._connect() + try: + conn.executescript(_SCHEMA_SQL) + conn.commit() + finally: + conn.close() + + @staticmethod + def _row_to_record(row: sqlite3.Row) -> dict[str, Any]: + return {key: row[key] for key in row.keys()} + + # -- liveness -- + + @staticmethod + def is_live( + record: dict[str, Any] | None, + *, + now: datetime | None = None, + pid_alive: bool | None = None, + ) -> dict[str, Any]: + """Liveness from heartbeat freshness, corroborated by process evidence. + + #948 AC7: abandoned generations are detected through authoritative + liveness evidence, not PID comparison alone. A fresh heartbeat is the + authority; ``pid_alive`` can only *withdraw* liveness, never grant it, + because a long-lived daemon PID outlives every session it serves. + """ + if not record: + return {"live": False, "reasons": ["no registration record"]} + if record.get("status") != STATUS_ACTIVE: + return { + "live": False, + "reasons": [f"registration status is {record.get('status')!r}"], + } + + beat = _parse_ts(record.get("last_heartbeat_at")) + if beat is None: + return {"live": False, "reasons": ["registration has no parsable heartbeat"]} + + ttl = float(record.get("heartbeat_ttl_seconds") or DEFAULT_HEARTBEAT_TTL_SECONDS) + age = ((now or _utc_now()) - beat).total_seconds() + fresh = age <= ttl + + reasons: list[str] = [] + if not fresh: + reasons.append( + f"heartbeat is {age:.0f}s old, past the {ttl:.0f}s liveness window" + ) + if pid_alive is False: + reasons.append(f"recorded pid {record.get('pid')} is no longer running") + + return { + "live": bool(fresh and pid_alive is not False), + "heartbeat_age_seconds": age, + "heartbeat_ttl_seconds": ttl, + "heartbeat_fresh": fresh, + "pid_alive": pid_alive, + "reasons": reasons, + } + + # -- reads -- + + def get(self, worker_identity: str) -> dict[str, Any] | None: + conn = self._connect() + try: + row = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + return self._row_to_record(row) if row else None + finally: + conn.close() + + def list_workers( + self, + *, + status: str | None = STATUS_ACTIVE, + generation_id: str | None = None, + profile: str | None = None, + role: str | None = None, + ) -> list[dict[str, Any]]: + clauses: list[str] = [] + params: list[Any] = [] + if status: + clauses.append("status = ?") + params.append(status) + if generation_id: + clauses.append("generation_id = ?") + params.append(generation_id) + if profile: + clauses.append("profile = ?") + params.append(profile) + if role: + clauses.append("role = ?") + params.append(role) + where = f" WHERE {' AND '.join(clauses)}" if clauses else "" + conn = self._connect() + try: + rows = conn.execute( + f"SELECT * FROM worker_registrations{where} ORDER BY started_at", + params, + ).fetchall() + return [self._row_to_record(r) for r in rows] + finally: + conn.close() + + # -- writes -- + + def register( + self, + *, + worker_identity: str, + client_name: str | None, + client_instance_id: str, + session_id: str, + generation_id: str, + role: str | None = None, + profile: str | None = None, + namespace: str | None = None, + remote: str | None = None, + repository_binding: str | None = None, + pid: int | None = None, + transport: str | None = None, + token_fingerprint: str | None = None, + heartbeat_ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS, + now: datetime | None = None, + pid_alive_probe=None, + ) -> dict[str, Any]: + """Atomically register one worker identity. + + Fails closed on collision (#948 AC31): an identity already present is + never replaced, adopted, merged with, or corrupted — the caller is told + to mint a different identity and register that instead (AC32). The + existing registration is returned untouched so the caller can see what + it collided with. + """ + parsed = parse_worker_identity(worker_identity) + if not parsed["valid"]: + return { + "success": False, + "registered": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_IDENTITY_COLLISION, + "collision": False, + "reasons": parsed["reasons"], + "exact_next_action": ( + "Mint a worker identity with generate_worker_identity() and " + "register that." + ), + } + + stamp = _ts(now or _utc_now()) + with self._tx() as conn: + existing = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + if existing is not None: + record = self._row_to_record(existing) + pid_alive = ( + pid_alive_probe(record.get("pid")) + if pid_alive_probe is not None + else None + ) + liveness = self.is_live(record, now=now, pid_alive=pid_alive) + return { + "success": False, + "registered": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_IDENTITY_COLLISION, + "collision": True, + "collision_kind": ( + "active_worker" if liveness["live"] else "stale_registration" + ), + "existing_registration": _public_record(record), + "existing_liveness": liveness, + "reasons": [ + f"worker identity {worker_identity!r} is already registered " + f"(status={record.get('status')}, live={liveness['live']}); " + "registration is refused rather than replacing, adopting, or " + "merging with the existing worker (#948 AC31)" + ], + "exact_next_action": ( + "Generate a different worker identity and atomically register " + "that replacement; do not reuse or take over this one." + ), + } + + epoch = self._next_epoch(conn, generation_id) + conn.execute( + """ + INSERT INTO worker_registrations ( + worker_identity, client_name, client_instance_id, session_id, + generation_id, role, profile, namespace, remote, + repository_binding, pid, transport, token_fingerprint, + started_at, last_heartbeat_at, heartbeat_ttl_seconds, + fencing_epoch, status + ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + """, + ( + worker_identity, + normalize_client_name(client_name or parsed["client_name"]), + client_instance_id, + session_id, + generation_id, + role, + profile, + namespace, + remote, + repository_binding, + int(pid) if pid is not None else None, + transport, + token_fingerprint, + stamp, + stamp, + float(heartbeat_ttl_seconds), + epoch, + STATUS_ACTIVE, + ), + ) + row = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + record = self._row_to_record(row) + + return { + "success": True, + "registered": True, + "mutation_performed": True, + "blocker_kind": BLOCKER_NONE, + "collision": False, + "registration": _public_record(record), + "fencing_epoch": record["fencing_epoch"], + "reasons": [], + } + + @staticmethod + def _next_epoch(conn: sqlite3.Connection, generation_id: str) -> int: + row = conn.execute( + "SELECT MAX(fencing_epoch) AS hi FROM worker_registrations " + "WHERE generation_id = ?", + (generation_id,), + ).fetchone() + current = row["hi"] if row and row["hi"] is not None else 0 + return int(current) + 1 + + def heartbeat( + self, + *, + worker_identity: str, + fencing_epoch: int, + now: datetime | None = None, + ) -> dict[str, Any]: + """Renew only the owning registration (#948 AC11). + + A stale epoch is refused rather than silently renewed, so a superseded + session that resumes cannot heartbeat its way back into ownership. + """ + stamp = _ts(now or _utc_now()) + with self._tx() as conn: + row = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + if row is None: + return { + "success": False, + "renewed": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_NO_ATTACHMENT, + "reasons": [f"no registration for {worker_identity!r}"], + } + record = self._row_to_record(row) + if record["status"] != STATUS_ACTIVE: + return { + "success": False, + "renewed": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_FENCED, + "reasons": [ + f"registration status is {record['status']!r}; a superseded " + "registration cannot be renewed (#948 AC15)" + ], + } + if int(record["fencing_epoch"]) != int(fencing_epoch): + return { + "success": False, + "renewed": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_FENCED, + "recorded_epoch": int(record["fencing_epoch"]), + "presented_epoch": int(fencing_epoch), + "reasons": [ + f"fencing epoch {fencing_epoch} does not match the recorded " + f"epoch {record['fencing_epoch']}; this session has been " + "superseded (#948 AC15/AC16)" + ], + } + conn.execute( + "UPDATE worker_registrations SET last_heartbeat_at = ? " + "WHERE worker_identity = ?", + (stamp, worker_identity), + ) + return { + "success": True, + "renewed": True, + "mutation_performed": True, + "blocker_kind": BLOCKER_NONE, + "last_heartbeat_at": stamp, + "fencing_epoch": int(fencing_epoch), + "reasons": [], + } + + def claim_generation( + self, + *, + worker_identity: str, + generation_id: str, + now: datetime | None = None, + pid_alive_probe=None, + ) -> dict[str, Any]: + """Bind a generation to this worker, superseding only a dead claimant. + + A generation held by another *live* session is refused: one daemon must + never be simultaneously claimed by conflicting live sessions. A + generation whose claimant is no longer live is taken over with a higher + fencing epoch, so stale ownership cannot permanently strand a healthy + daemon (#948 AC8/AC14). + """ + with self._tx() as conn: + mine = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + if mine is None: + return { + "success": False, + "claimed": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_NO_ATTACHMENT, + "reasons": [ + f"{worker_identity!r} is not registered; register before " + "claiming a generation" + ], + } + + holders = [ + self._row_to_record(r) + for r in conn.execute( + "SELECT * FROM worker_registrations " + "WHERE generation_id = ? AND status = ? AND worker_identity != ?", + (generation_id, STATUS_ACTIVE, worker_identity), + ).fetchall() + ] + + live_holders = [] + for holder in holders: + pid_alive = ( + pid_alive_probe(holder.get("pid")) + if pid_alive_probe is not None + else None + ) + if self.is_live(holder, now=now, pid_alive=pid_alive)["live"]: + live_holders.append(holder) + + if live_holders: + return { + "success": False, + "claimed": False, + "mutation_performed": False, + "blocker_kind": BLOCKER_CONFLICTING_SESSIONS, + "conflicting_owners": [_public_record(h) for h in live_holders], + "reasons": [ + f"generation {generation_id!r} is already claimed by " + f"{len(live_holders)} live session(s): " + + ", ".join( + f"{h['worker_identity']} (session {h['session_id']})" + for h in live_holders + ) + + "; a daemon must not be simultaneously claimed by " + "conflicting live sessions (#948)" + ], + "exact_next_action": ( + "Wait for the live owner's lease to lapse, or attach this " + "worker to its own generation. Do not kill the other process." + ), + } + + superseded = [h["worker_identity"] for h in holders] + for identity in superseded: + conn.execute( + "UPDATE worker_registrations SET status = ? " + "WHERE worker_identity = ?", + (STATUS_SUPERSEDED, identity), + ) + + epoch = self._next_epoch(conn, generation_id) + conn.execute( + "UPDATE worker_registrations SET generation_id = ?, fencing_epoch = ?, " + "status = ?, last_heartbeat_at = ? WHERE worker_identity = ?", + ( + generation_id, + epoch, + STATUS_ACTIVE, + _ts(now or _utc_now()), + worker_identity, + ), + ) + row = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + record = self._row_to_record(row) + + return { + "success": True, + "claimed": True, + "mutation_performed": True, + "blocker_kind": BLOCKER_NONE, + "registration": _public_record(record), + "fencing_epoch": record["fencing_epoch"], + "superseded_workers": superseded, + "reasons": ( + [ + f"took over generation {generation_id!r} from " + f"{len(superseded)} non-live claimant(s) with fencing epoch " + f"{record['fencing_epoch']}" + ] + if superseded + else [] + ), + } + + def release( + self, *, worker_identity: str, now: datetime | None = None + ) -> dict[str, Any]: + """Mark a registration released. Idempotent; never deletes history.""" + with self._tx() as conn: + row = conn.execute( + "SELECT * FROM worker_registrations WHERE worker_identity = ?", + (worker_identity,), + ).fetchone() + if row is None: + return { + "success": True, + "released": False, + "mutation_performed": False, + "reasons": [f"no registration for {worker_identity!r}"], + } + conn.execute( + "UPDATE worker_registrations SET status = ?, last_heartbeat_at = ? " + "WHERE worker_identity = ?", + (STATUS_RELEASED, _ts(now or _utc_now()), worker_identity), + ) + return { + "success": True, + "released": True, + "mutation_performed": True, + "reasons": [], + } + + +def _public_record(record: dict[str, Any]) -> dict[str, Any]: + """Registry row minus anything that should not travel to an LLM surface.""" + return { + "worker_identity": record.get("worker_identity"), + "client_name": record.get("client_name"), + "client_instance_id": record.get("client_instance_id"), + "session_id": record.get("session_id"), + "generation_id": record.get("generation_id"), + "role": record.get("role"), + "profile": record.get("profile"), + "namespace": record.get("namespace"), + "remote": record.get("remote"), + "repository_binding": record.get("repository_binding"), + "pid": record.get("pid"), + "transport": record.get("transport"), + "started_at": record.get("started_at"), + "last_heartbeat_at": record.get("last_heartbeat_at"), + "fencing_epoch": record.get("fencing_epoch"), + "status": record.get("status"), + } + + +# --- The single authoritative verdict ------------------------------------ + + +def assess_provenance( + *, + registry: WorkerRegistry | None = None, + worker_identity: str | None = None, + generation_id: str | None = None, + env: dict[str, str] | None = None, + native_transport_bound: bool | None = None, + namespace: str | None = None, + profile: str | None = None, + role: str | None = None, + now: datetime | None = None, + pid_alive_probe=None, + stdin_is_tty: bool | None = None, + declared_only: bool = False, +) -> dict[str, Any]: + """Resolve client/session provenance once, for every surface to reuse. + + Runtime context, namespace health, namespace attachment, fleet inventory, + capability resolution, and typed reconnect assessment all call this, so the + contradiction that motivated #948 — one surface reporting ``client_managed`` + while another reported ``manual_launch`` for the same process — cannot recur + by construction. + + Two independent dimensions are reported: + + ``launch_provenance`` + How the process was started. Answered from the environment, which is + legitimate because the launcher is what sets it. Preserves the #686 + manual-launch wall unchanged. + ``session_ownership`` + Which live client session owns this runtime now. Answered *only* from a + live attachment record; no environment flag can establish it. + + Missing evidence yields :data:`PROVENANCE_UNPROVEN`, which is deliberately + distinct from :data:`PROVENANCE_MANUAL`: "we cannot prove who owns this" is + not the same claim as "a human launched this from a terminal", and only the + latter justifies telling an operator to stop hand-launching servers. + """ + env_signal = env_provenance_signal(env) + launch = assess_launch_provenance( + env, stdin_is_tty=stdin_is_tty, declared_only=declared_only + ) + evidence: list[str] = [EVIDENCE_ENV_FLAG] if env_signal["keys_present"] else [] + reasons: list[str] = [] + + record: dict[str, Any] | None = None + liveness: dict[str, Any] = {"live": False, "reasons": ["no attachment record"]} + + if registry is not None and worker_identity: + record = registry.get(worker_identity) + if record is not None: + pid_alive = ( + pid_alive_probe(record.get("pid")) + if pid_alive_probe is not None + else None + ) + liveness = registry.is_live(record, now=now, pid_alive=pid_alive) + if liveness["live"]: + evidence.append(EVIDENCE_ATTACHMENT_RECORD) + if pid_alive is not None: + evidence.append(EVIDENCE_PROCESS_LIVENESS) + + if native_transport_bound: + evidence.append(EVIDENCE_NATIVE_TRANSPORT) + + # Conflicting live claims on one generation are contradictory evidence: two + # sessions cannot both currently own the same daemon, so neither claim is + # trustworthy and the assessment fails closed rather than picking a winner. + conflicting: list[dict[str, Any]] = [] + effective_generation = generation_id or (record or {}).get("generation_id") + if registry is not None and effective_generation: + for holder in registry.list_workers(generation_id=effective_generation): + if ( + record is not None + and holder["worker_identity"] == record["worker_identity"] + ): + continue + pid_alive = ( + pid_alive_probe(holder.get("pid")) + if pid_alive_probe is not None + else None + ) + if registry.is_live(holder, now=now, pid_alive=pid_alive)["live"]: + conflicting.append(_public_record(holder)) + + trusted = bool(set(evidence) & SUFFICIENT_EVIDENCE) + blocker = BLOCKER_NONE + + if conflicting and trusted: + trusted = False + blocker = BLOCKER_CONTRADICTORY + reasons.append( + f"generation {effective_generation!r} is claimed by " + f"{len(conflicting) + 1} live sessions at once; contradictory ownership " + "evidence fails closed (#948)" + ) + elif not trusted: + blocker = BLOCKER_NO_ATTACHMENT + if record is None: + reasons.append( + "no live client/session attachment record binds this runtime to a " + "session; a process environment flag alone does not prove current " + "session ownership (#948)" + ) + else: + reasons.extend( + f"attachment record is not live: {r}" for r in liveness["reasons"] + ) + + if conflicting: + session_ownership = OWNERSHIP_CONTESTED + elif trusted: + session_ownership = OWNERSHIP_OWNED + else: + session_ownership = OWNERSHIP_UNOWNED + + # The headline verdict. Session ownership decides it when proven; otherwise + # the launch dimension supplies the more specific of "hand-launched" versus + # "cannot tell", so remediation names the proof that is actually missing. + if session_ownership == OWNERSHIP_OWNED: + provenance = PROVENANCE_CLIENT_SESSION + elif launch["launch_provenance"] == PROVENANCE_MANUAL: + provenance = PROVENANCE_MANUAL + reasons.extend(launch["reasons"]) + else: + provenance = PROVENANCE_UNPROVEN + + client_name = normalize_client_name( + (record or {}).get("client_name") + or parse_worker_identity(worker_identity)["client_name"] + ) + + return { + "provenance": provenance, + # #948: the *launch* dimension, which is what this field has always + # meant to the #686 manual-launch wall. Keeping it bound to launch + # provenance means unifying the two models changes which code decides, + # not what gets decided, so no mutation that used to be permitted is + # newly refused. + "is_client_managed": bool(launch["client_managed"]), + "launch_provenance": launch["launch_provenance"], + "launch_decided_by": launch["decided_by"], + "launch_reasons": launch["reasons"], + # The *ownership* dimension, which no environment flag can establish. + "session_ownership": session_ownership, + "session_owned": session_ownership == OWNERSHIP_OWNED, + "trusted_evidence": trusted, + "evidence": sorted(set(evidence)), + "sufficient_evidence_kinds": sorted(SUFFICIENT_EVIDENCE), + "env_signal": env_signal, + "env_flag_only": bool(evidence) and not trusted and not conflicting, + "attachment": _public_record(record) if record else None, + "attachment_liveness": liveness, + "client_name": client_name, + "client_known": client_name != UNKNOWN_CLIENT, + "session_id": (record or {}).get("session_id"), + "client_instance_id": (record or {}).get("client_instance_id"), + "generation_id": effective_generation, + "worker_identity": (record or {}).get("worker_identity") or worker_identity, + "fencing_epoch": (record or {}).get("fencing_epoch"), + "conflicting_live_sessions": conflicting, + "blocker_kind": blocker, + "fail_closed": provenance != PROVENANCE_CLIENT_SESSION, + # #948 AC17/AC41: a refusal names the worker it applies to, never the + # profile, so unrelated healthy workers sharing that profile keep working. + "scope": { + "scope_kind": "worker_identity" if worker_identity else "generation", + "worker_identity": worker_identity, + "generation_id": effective_generation, + "namespace": namespace, + "profile": profile, + "role": role, + "profile_wide": False, + "note": ( + "role and profile are reusable capability definitions; this verdict " + "binds only the named worker identity/generation" + ), + }, + "reasons": reasons, + } + + +# --- Cohort scoping ------------------------------------------------------ + + +def classify_cohort( + records: list[dict[str, Any]], + *, + now: datetime | None = None, + pid_alive_probe=None, +) -> dict[str, Any]: + """Separate genuine duplicates from valid concurrent workers (#948 AC43). + + The pre-#948 scan flagged every additional process sharing a profile as a + duplicate, which is exactly the fleet-wide wall #948 exists to remove. + Sharing a role or profile is legitimate; reusing a *worker identity*, or two + live sessions claiming one generation, is not. + """ + live: list[dict[str, Any]] = [] + for record in records: + pid_alive = ( + pid_alive_probe(record.get("pid")) if pid_alive_probe is not None else None + ) + if WorkerRegistry.is_live(record, now=now, pid_alive=pid_alive)["live"]: + live.append(record) + + by_identity: dict[str, list[dict[str, Any]]] = {} + by_generation: dict[str, list[dict[str, Any]]] = {} + for record in live: + by_identity.setdefault(str(record.get("worker_identity")), []).append(record) + by_generation.setdefault(str(record.get("generation_id")), []).append(record) + + duplicate_identities = sorted(k for k, v in by_identity.items() if len(v) > 1) + contested_generations = sorted(k for k, v in by_generation.items() if len(v) > 1) + + shared_profiles = sorted( + { + str(r.get("profile")) + for r in live + if r.get("profile") + and sum(1 for o in live if o.get("profile") == r.get("profile")) > 1 + } + ) + + blocked = bool(duplicate_identities or contested_generations) + reasons: list[str] = [] + if duplicate_identities: + reasons.append( + f"worker identity reused by more than one live worker: " + f"{duplicate_identities}" + ) + if contested_generations: + reasons.append( + f"generation claimed by more than one live session: {contested_generations}" + ) + if shared_profiles and not blocked: + reasons.append( + f"profile(s) {shared_profiles} are shared by multiple live workers with " + "distinct identities, which is permitted and does not block mutations (#948)" + ) + + return { + "live_worker_count": len(live), + "live_workers": [_public_record(r) for r in live], + "duplicate_identities": duplicate_identities, + "contested_generations": contested_generations, + "shared_profiles": shared_profiles, + "profile_sharing_permitted": True, + "blocked": blocked, + "blocker_kind": ( + BLOCKER_IDENTITY_COLLISION + if duplicate_identities + else BLOCKER_CONFLICTING_SESSIONS + if contested_generations + else BLOCKER_NONE + ), + # Never profile-wide: the block names the offending identities only. + "blocked_worker_identities": sorted( + set(duplicate_identities) + | { + str(r.get("worker_identity")) + for gen in contested_generations + for r in by_generation[gen] + } + ), + "reasons": reasons, + } + + +def scope_runtime_failure( + *, + failure_kind: str, + worker_identity: str | None = None, + generation_id: str | None = None, + profile: str | None = None, + role: str | None = None, + all_live_workers: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Scope a stale-runtime/reconnect failure to one worker (#948 AC41). + + Returns the set of workers actually affected and, separately, the healthy + workers that must keep operating. A stale generation affects only its own + ownership scope. + """ + workers = list(all_live_workers or []) + affected = [ + _public_record(w) + for w in workers + if (worker_identity and w.get("worker_identity") == worker_identity) + or ( + not worker_identity + and generation_id + and w.get("generation_id") == generation_id + ) + ] + affected_ids = {a["worker_identity"] for a in affected} + unaffected = [ + _public_record(w) + for w in workers + if w.get("worker_identity") not in affected_ids + ] + + return { + "failure_kind": failure_kind, + "scope_kind": "worker_identity" if worker_identity else "generation", + "worker_identity": worker_identity, + "generation_id": generation_id, + "profile": profile, + "role": role, + "profile_wide": False, + "fleet_wide": False, + "affected_workers": affected, + "unaffected_workers": unaffected, + "unaffected_worker_count": len(unaffected), + "reasons": [ + f"{failure_kind} is scoped to " + + ( + f"worker identity {worker_identity!r}" + if worker_identity + else f"generation {generation_id!r}" + ) + + f"; {len(unaffected)} other live worker(s) sharing this role/profile " + "are unaffected and remain able to mutate (#948 AC41)" + ], + } + + +# --- Client-specific remediation ---------------------------------------- + + +def reconnect_client_for( + provenance: dict[str, Any] | None = None, + *, + fallback: str | None = None, +) -> str: + """Name the client to address reconnect guidance to. + + Derived from the live attachment record, never a hardcoded vendor. When + nothing identifies the client, :data:`UNKNOWN_CLIENT` is returned so the + caller emits generic steps instead of steps for the wrong product. + """ + name = normalize_client_name((provenance or {}).get("client_name")) + if name != UNKNOWN_CLIENT: + return name + return normalize_client_name(fallback) + + +# --- Remote binding ------------------------------------------------------ + + +def resolve_bound_remote( + *, + requested_remote: str | None, + bound_remote: str | None, + default_remote: str | None = None, +) -> dict[str, Any]: + """Keep a namespace on the remote it is bound to (#948). + + A tool default is not a decision. When a namespace is bound to a remote, + an omitted argument resolves to that binding rather than silently falling + through to the library default and hitting the wrong host; an argument that + contradicts the binding is refused rather than honoured. + """ + requested = (requested_remote or "").strip() or None + bound = (bound_remote or "").strip() or None + default = (default_remote or "").strip() or None + + if bound and requested and requested != bound: + return { + "remote": bound, + "resolved_from": "session_binding", + "drifted": True, + "honoured_request": False, + "reasons": [ + f"requested remote {requested!r} contradicts the session binding " + f"{bound!r}; the binding wins and the request is refused so a " + "PRGS-bound namespace never mutates another host (#948)" + ], + } + if bound and not requested: + return { + "remote": bound, + "resolved_from": "session_binding", + "drifted": False, + "honoured_request": False, + "reasons": [ + f"remote omitted; resolved to the session binding {bound!r} rather " + f"than the library default {default!r} (#948)" + ], + } + if requested: + return { + "remote": requested, + "resolved_from": "explicit_argument", + "drifted": False, + "honoured_request": True, + "reasons": [], + } + return { + "remote": default, + "resolved_from": "library_default", + "drifted": False, + "honoured_request": False, + "reasons": [ + "no remote requested and no session binding recorded; falling back to " + f"the library default {default!r}. Bind the session or pass remote " + "explicitly to avoid host drift (#948)." + ], + } diff --git a/tests/test_issue_686_manual_mcp_provenance.py b/tests/test_issue_686_manual_mcp_provenance.py index 40170c1..bd012a8 100644 --- a/tests/test_issue_686_manual_mcp_provenance.py +++ b/tests/test_issue_686_manual_mcp_provenance.py @@ -112,7 +112,19 @@ class TestIssue686ManualMcpProvenance(unittest.TestCase): self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons)) def test_namespace_health_classification_includes_provenance(self): - """AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env.""" + """AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env. + + #948 narrowed the vocabulary here. This process carries no client-managed + declaration, so the old code labelled it ``manual_launch`` — asserting a + hand-launched terminal process it had no evidence for, and contradicting + ``gitea_get_runtime_context``, which read the same process and reported + ``client_managed``. Absence of proof is now reported as ``unproven``. + + The #686 wall itself is unchanged and still asserted below: + ``is_client_managed`` stays False, so nothing previously refused is now + permitted. Only the label on the *reason* changed, so remediation names + the proof that is actually missing. + """ process = { "pid": 5555, "profile": "prgs-author", @@ -129,10 +141,12 @@ class TestIssue686ManualMcpProvenance(unittest.TestCase): process=process, probe_source="client_namespace", ) - self.assertEqual(res["provenance"], "manual_launch") + self.assertEqual(res["provenance"], "unproven") self.assertFalse(res["is_client_managed"]) + # The wall is intact: no client-managed proof still fails closed. + self.assertTrue(res["provenance_fail_closed"]) self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"}) - self.assertEqual(res["diagnostics"]["provenance"], "manual_launch") + self.assertEqual(res["diagnostics"]["provenance"], "unproven") if __name__ == "__main__": diff --git a/tests/test_issue_948_client_session_provenance.py b/tests/test_issue_948_client_session_provenance.py new file mode 100644 index 0000000..ec29a4c --- /dev/null +++ b/tests/test_issue_948_client_session_provenance.py @@ -0,0 +1,662 @@ +"""Client/session-aware runtime ownership and provenance (#948). + +Covers the reproduced contradiction that motivated the issue: one surface +reporting ``client_managed`` while another reported ``manual_launch`` for the +same process, remediation hardcoded to one vendor, and a profile-wide duplicate +wall that could not tell two healthy clients apart. + +All client and session identifiers here are synthetic. +""" + +from __future__ import annotations + +import os +import tempfile +import unittest +from datetime import datetime, timedelta, timezone + +import mcp_client_reconnect +import mcp_namespace_health +import mcp_worker_identity as mwi + + +NOW = datetime(2026, 7, 29, 6, 0, 0, tzinfo=timezone.utc) + + +def _registry() -> mwi.WorkerRegistry: + """A registry on a throwaway path; never the operator's real one.""" + handle, path = tempfile.mkstemp(suffix=".sqlite3") + os.close(handle) + os.unlink(path) + return mwi.WorkerRegistry(path) + + +def _attach( + registry: mwi.WorkerRegistry, + *, + client: str, + session: str, + generation: str, + profile: str = "prgs-reviewer", + role: str = "reviewer", + pid: int = 4242, + now: datetime = NOW, + ttl: float = 900.0, +) -> dict: + """Register one synthetic worker and return the outcome.""" + identity = mwi.generate_worker_identity(client, session, now=now) + outcome = registry.register( + worker_identity=identity, + client_name=client, + client_instance_id=f"inst-{session}", + session_id=session, + generation_id=generation, + role=role, + profile=profile, + pid=pid, + heartbeat_ttl_seconds=ttl, + now=now, + ) + outcome["identity"] = identity + return outcome + + +class IdentityFormatTests(unittest.TestCase): + """AC27-29: collision-resistant `--`.""" + + def test_identity_matches_required_format(self): + identity = mwi.generate_worker_identity("Gemini", "sess-0001", now=NOW) + parsed = mwi.parse_worker_identity(identity) + self.assertTrue(parsed["valid"], parsed["reasons"]) + self.assertEqual(parsed["client_name"], "gemini") + self.assertEqual(parsed["minted_at"], "20260729T060000Z") + self.assertEqual(len(parsed["digest"]), 12) + + def test_digest_varies_with_session_and_nonce(self): + base = dict(timestamp_ns=1, now=NOW) + a = mwi.generate_worker_identity("codex", "sess-A", nonce="n", **base) + b = mwi.generate_worker_identity("codex", "sess-B", nonce="n", **base) + c = mwi.generate_worker_identity("codex", "sess-A", nonce="m", **base) + self.assertNotEqual(a, b, "session must feed the digest") + self.assertNotEqual(a, c, "nonce must feed the digest") + + def test_identity_is_not_role_or_profile(self): + """AC26: identity is independent of role and profile.""" + args = dict(timestamp_ns=7, nonce="fixed", now=NOW) + same = mwi.generate_worker_identity("claude", "sess-1", **args) + self.assertEqual(same, mwi.generate_worker_identity("claude", "sess-1", **args)) + # Nothing role- or profile-derived appears in the identity. + self.assertNotIn("reviewer", same) + self.assertNotIn("prgs", same) + + def test_malformed_identity_rejected(self): + self.assertFalse(mwi.parse_worker_identity("prgs-reviewer")["valid"]) + self.assertFalse(mwi.parse_worker_identity("")["valid"]) + self.assertFalse(mwi.parse_worker_identity(None)["valid"]) + + +class PerClientAttachmentTests(unittest.TestCase): + """Every supported client attaches and is reported as itself.""" + + def _assert_attached_as(self, client: str, expected_name: str): + registry = _registry() + outcome = _attach( + registry, client=client, session=f"sess-{client}", generation="gen-1" + ) + self.assertTrue(outcome["registered"], outcome["reasons"]) + + verdict = mwi.assess_provenance( + registry=registry, worker_identity=outcome["identity"], env={}, now=NOW + ) + self.assertEqual(verdict["session_ownership"], mwi.OWNERSHIP_OWNED) + self.assertEqual(verdict["provenance"], mwi.PROVENANCE_CLIENT_SESSION) + self.assertEqual(verdict["client_name"], expected_name) + self.assertTrue(verdict["session_owned"]) + self.assertFalse(verdict["fail_closed"]) + return verdict + + def test_codex_attachment(self): + self._assert_attached_as("codex", "codex") + + def test_gemini_attachment(self): + self._assert_attached_as("gemini", "gemini") + + def test_antigravity_attachment(self): + self._assert_attached_as("antigravity", "antigravity") + + def test_claude_attachment(self): + self._assert_attached_as("claude", "claude_code") + + def test_unknown_client_is_named_not_guessed(self): + verdict = self._assert_attached_as("some_new_llm", "some_new_llm") + self.assertNotEqual(verdict["client_name"], "codex") + + +class SessionLifecycleTests(unittest.TestCase): + def test_same_client_new_session_gets_distinct_identity(self): + registry = _registry() + first = _attach(registry, client="codex", session="sess-1", generation="gen-1") + second = _attach(registry, client="codex", session="sess-2", generation="gen-2") + self.assertTrue(first["registered"]) + self.assertTrue(second["registered"]) + self.assertNotEqual(first["identity"], second["identity"]) + + # Both are live and neither blocks the other. + cohort = mwi.classify_cohort(registry.list_workers(), now=NOW) + self.assertEqual(cohort["live_worker_count"], 2) + self.assertFalse(cohort["blocked"], cohort["reasons"]) + + def test_different_client_attaches_after_previous_session_ends(self): + """AC14: expiry then takeover with a higher fencing epoch.""" + registry = _registry() + gone = _attach( + registry, client="codex", session="sess-old", generation="gen-shared", ttl=60 + ) + later = NOW + timedelta(hours=1) + self.assertFalse( + registry.is_live(registry.get(gone["identity"]), now=later)["live"] + ) + + arriving = _attach( + registry, + client="gemini", + session="sess-new", + generation="gen-other", + now=later, + ) + claim = registry.claim_generation( + worker_identity=arriving["identity"], + generation_id="gen-shared", + now=later, + ) + self.assertTrue(claim["claimed"], claim["reasons"]) + self.assertIn(gone["identity"], claim["superseded_workers"]) + self.assertGreater(claim["fencing_epoch"], gone["fencing_epoch"]) + + def test_superseded_session_is_fenced_on_resume(self): + """AC15/AC16: the prior session cannot heartbeat its way back.""" + registry = _registry() + old = _attach( + registry, client="codex", session="sess-old", generation="gen-shared", ttl=60 + ) + later = NOW + timedelta(hours=1) + new = _attach( + registry, client="gemini", session="sess-new", generation="gen-x", now=later + ) + registry.claim_generation( + worker_identity=new["identity"], generation_id="gen-shared", now=later + ) + + resumed = registry.heartbeat( + worker_identity=old["identity"], + fencing_epoch=old["fencing_epoch"], + now=later, + ) + self.assertFalse(resumed["renewed"]) + self.assertFalse(resumed["mutation_performed"]) + self.assertEqual(resumed["blocker_kind"], mwi.BLOCKER_FENCED) + + def test_heartbeat_renews_only_the_owning_lease(self): + """AC11: a wrong epoch never renews, and never mutates.""" + registry = _registry() + worker = _attach(registry, client="codex", session="s", generation="g") + good = registry.heartbeat( + worker_identity=worker["identity"], + fencing_epoch=worker["fencing_epoch"], + now=NOW + timedelta(minutes=5), + ) + self.assertTrue(good["renewed"]) + + bad = registry.heartbeat( + worker_identity=worker["identity"], + fencing_epoch=worker["fencing_epoch"] + 99, + now=NOW + timedelta(minutes=6), + ) + self.assertFalse(bad["renewed"]) + self.assertFalse(bad["mutation_performed"]) + self.assertEqual( + registry.get(worker["identity"])["last_heartbeat_at"], + good["last_heartbeat_at"], + "a refused heartbeat must not advance the record", + ) + + +class ConflictAndCollisionTests(unittest.TestCase): + def test_two_live_sessions_cannot_claim_one_generation(self): + registry = _registry() + first = _attach(registry, client="codex", session="s1", generation="gen-shared") + second = _attach(registry, client="gemini", session="s2", generation="gen-other") + + claim = registry.claim_generation( + worker_identity=second["identity"], + generation_id="gen-shared", + now=NOW, + ) + self.assertFalse(claim["claimed"]) + self.assertFalse(claim["mutation_performed"]) + self.assertEqual(claim["blocker_kind"], mwi.BLOCKER_CONFLICTING_SESSIONS) + self.assertEqual( + claim["conflicting_owners"][0]["worker_identity"], first["identity"] + ) + # The sanctioned recovery must never be "kill the other process". + self.assertIn("Do not kill", claim["exact_next_action"]) + + def test_contested_generation_fails_closed_in_assessment(self): + registry = _registry() + first = _attach(registry, client="codex", session="s1", generation="gen-shared") + _attach(registry, client="gemini", session="s2", generation="gen-shared") + + verdict = mwi.assess_provenance( + registry=registry, worker_identity=first["identity"], env={}, now=NOW + ) + self.assertEqual(verdict["session_ownership"], mwi.OWNERSHIP_CONTESTED) + self.assertTrue(verdict["fail_closed"]) + self.assertEqual(verdict["blocker_kind"], mwi.BLOCKER_CONTRADICTORY) + self.assertTrue(verdict["conflicting_live_sessions"]) + + def test_identity_collision_is_refused_without_corrupting_existing(self): + """AC31: never replace, adopt, merge with, or corrupt the incumbent.""" + registry = _registry() + incumbent = _attach(registry, client="codex", session="s1", generation="gen-1") + before = registry.get(incumbent["identity"]) + + collided = registry.register( + worker_identity=incumbent["identity"], + client_name="gemini", + client_instance_id="inst-other", + session_id="s2", + generation_id="gen-2", + pid=9999, + now=NOW, + ) + self.assertFalse(collided["registered"]) + self.assertTrue(collided["collision"]) + self.assertFalse(collided["mutation_performed"]) + self.assertEqual(collided["blocker_kind"], mwi.BLOCKER_IDENTITY_COLLISION) + self.assertEqual(collided["collision_kind"], "active_worker") + self.assertEqual( + registry.get(incumbent["identity"]), before, "incumbent must be untouched" + ) + + def test_after_collision_a_regenerated_identity_registers(self): + """AC32/AC35: forced collision, safe regeneration, successful replacement.""" + registry = _registry() + fixed = dict(timestamp_ns=99, nonce="deterministic", now=NOW) + forced = mwi.generate_worker_identity("codex", "sess-collide", **fixed) + first = registry.register( + worker_identity=forced, + client_name="codex", + client_instance_id="inst-1", + session_id="sess-collide", + generation_id="gen-1", + now=NOW, + ) + self.assertTrue(first["registered"]) + + # A second worker deriving the same inputs collides deterministically. + again = mwi.generate_worker_identity("codex", "sess-collide", **fixed) + self.assertEqual(again, forced) + self.assertTrue( + registry.register( + worker_identity=again, + client_name="codex", + client_instance_id="inst-2", + session_id="sess-collide", + generation_id="gen-2", + now=NOW, + )["collision"] + ) + + replacement = mwi.generate_worker_identity( + "codex", "sess-collide", timestamp_ns=100, nonce="different", now=NOW + ) + self.assertNotEqual(replacement, forced) + self.assertTrue( + registry.register( + worker_identity=replacement, + client_name="codex", + client_instance_id="inst-2", + session_id="sess-collide", + generation_id="gen-2", + now=NOW, + )["registered"] + ) + + def test_restarted_worker_inherits_nothing(self): + """AC33/AC34: a restart mints a new identity and no prior epoch.""" + registry = _registry() + before = _attach( + registry, client="codex", session="sess-before", generation="gen-1", ttl=60 + ) + later = NOW + timedelta(hours=2) + after = _attach( + registry, client="codex", session="sess-after", generation="gen-2", now=later + ) + self.assertNotEqual(before["identity"], after["identity"]) + self.assertNotEqual( + registry.get(after["identity"])["generation_id"], + registry.get(before["identity"])["generation_id"], + ) + + +class LivenessTests(unittest.TestCase): + def test_stale_session_record_is_not_live(self): + registry = _registry() + worker = _attach(registry, client="codex", session="s", generation="g", ttl=300) + stale = registry.is_live( + registry.get(worker["identity"]), now=NOW + timedelta(hours=1) + ) + self.assertFalse(stale["live"]) + self.assertFalse(stale["heartbeat_fresh"]) + + def test_liveness_is_not_pid_comparison_alone(self): + """AC7: a live PID does not resurrect an expired registration.""" + registry = _registry() + worker = _attach(registry, client="codex", session="s", generation="g", ttl=60) + verdict = registry.is_live( + registry.get(worker["identity"]), + now=NOW + timedelta(hours=1), + pid_alive=True, + ) + self.assertFalse( + verdict["live"], "a live PID must not override a dead heartbeat" + ) + + def test_dead_pid_withdraws_liveness_from_a_fresh_heartbeat(self): + registry = _registry() + worker = _attach(registry, client="codex", session="s", generation="g") + verdict = registry.is_live( + registry.get(worker["identity"]), now=NOW, pid_alive=False + ) + self.assertFalse(verdict["live"]) + + def test_stale_ownership_does_not_permanently_strand_a_daemon(self): + registry = _registry() + stranded = _attach( + registry, client="codex", session="s-old", generation="gen-daemon", ttl=60 + ) + later = NOW + timedelta(hours=3) + rescuer = _attach( + registry, client="claude", session="s-new", generation="gen-tmp", now=later + ) + claim = registry.claim_generation( + worker_identity=rescuer["identity"], + generation_id="gen-daemon", + now=later, + ) + self.assertTrue(claim["claimed"], claim["reasons"]) + self.assertIn(stranded["identity"], claim["superseded_workers"]) + + +class EvidenceTests(unittest.TestCase): + def test_env_flag_alone_does_not_prove_session_ownership(self): + verdict = mwi.assess_provenance( + registry=None, + worker_identity=None, + env={"GITEA_CLIENT_MANAGED": "1", "GITEA_MCP_SANCTIONED_DAEMON": "1"}, + now=NOW, + ) + self.assertFalse(verdict["session_owned"]) + self.assertEqual(verdict["session_ownership"], mwi.OWNERSHIP_UNOWNED) + self.assertTrue(verdict["env_flag_only"]) + self.assertTrue(verdict["fail_closed"]) + self.assertNotIn(mwi.EVIDENCE_ATTACHMENT_RECORD, verdict["evidence"]) + self.assertFalse(verdict["env_signal"]["proves_session_ownership"]) + + def test_env_flag_still_answers_the_launch_question(self): + """The #686 wall is preserved: env decides launch, not ownership.""" + self.assertTrue( + mwi.assess_launch_provenance({"GITEA_CLIENT_MANAGED": "1"})["client_managed"] + ) + self.assertFalse( + mwi.assess_launch_provenance({"GITEA_CLIENT_MANAGED": "0"})["client_managed"] + ) + self.assertFalse( + mwi.assess_launch_provenance({}, stdin_is_tty=True)["client_managed"] + ) + self.assertTrue( + mwi.assess_launch_provenance({"GITEA_MCP_PROFILE": "prgs-author"})[ + "client_managed" + ] + ) + + def test_missing_evidence_is_unproven_not_manual(self): + """A missing proof must not be reported as a hand-launched process.""" + verdict = mwi.assess_provenance( + registry=None, worker_identity=None, env={}, now=NOW + ) + self.assertEqual(verdict["provenance"], mwi.PROVENANCE_UNPROVEN) + self.assertNotEqual(verdict["provenance"], mwi.PROVENANCE_MANUAL) + self.assertTrue(verdict["fail_closed"]) + + def test_declared_manual_launch_is_reported_as_manual(self): + verdict = mwi.assess_provenance( + registry=None, + worker_identity=None, + env={"GITEA_CLIENT_MANAGED": "0"}, + now=NOW, + ) + self.assertEqual(verdict["provenance"], mwi.PROVENANCE_MANUAL) + + def test_fail_closed_refusal_names_its_scope_not_the_profile(self): + """AC17/AC41: no refusal is profile-wide.""" + verdict = mwi.assess_provenance( + registry=None, + worker_identity=None, + env={}, + profile="prgs-reviewer", + role="reviewer", + now=NOW, + ) + self.assertFalse(verdict["scope"]["profile_wide"]) + self.assertEqual(verdict["blocker_kind"], mwi.BLOCKER_NO_ATTACHMENT) + + +class CohortScopingTests(unittest.TestCase): + def test_shared_profile_with_distinct_identities_does_not_block(self): + """AC40: profile is not a singleton identity.""" + registry = _registry() + _attach( + registry, + client="codex", + session="s1", + generation="g1", + profile="prgs-reviewer", + ) + _attach( + registry, + client="gemini", + session="s2", + generation="g2", + profile="prgs-reviewer", + ) + + cohort = mwi.classify_cohort(registry.list_workers(), now=NOW) + self.assertFalse(cohort["blocked"], cohort["reasons"]) + self.assertEqual(cohort["blocker_kind"], mwi.BLOCKER_NONE) + self.assertIn("prgs-reviewer", cohort["shared_profiles"]) + self.assertTrue(cohort["profile_sharing_permitted"]) + self.assertEqual(cohort["blocked_worker_identities"], []) + + def test_duplicate_cohort_records_block_only_the_offenders(self): + registry = _registry() + _attach(registry, client="codex", session="s1", generation="gen-contested") + _attach(registry, client="gemini", session="s2", generation="gen-contested") + _attach(registry, client="claude", session="s3", generation="gen-fine") + + cohort = mwi.classify_cohort(registry.list_workers(), now=NOW) + self.assertTrue(cohort["blocked"]) + self.assertEqual(cohort["contested_generations"], ["gen-contested"]) + self.assertEqual(len(cohort["blocked_worker_identities"]), 2) + + def test_mixed_runtime_generations_are_scoped_independently(self): + """AC17: one stale generation does not wall unrelated healthy ones.""" + registry = _registry() + stale = _attach( + registry, client="codex", session="s1", generation="gen-stale", ttl=60 + ) + healthy_a = _attach(registry, client="gemini", session="s2", generation="gen-a") + healthy_b = _attach(registry, client="claude", session="s3", generation="gen-b") + + scoped = mwi.scope_runtime_failure( + failure_kind="stale-runtime", + worker_identity=stale["identity"], + profile="prgs-reviewer", + all_live_workers=registry.list_workers(), + ) + self.assertFalse(scoped["profile_wide"]) + self.assertFalse(scoped["fleet_wide"]) + self.assertEqual(len(scoped["affected_workers"]), 1) + self.assertEqual(scoped["unaffected_worker_count"], 2) + unaffected = {w["worker_identity"] for w in scoped["unaffected_workers"]} + self.assertEqual(unaffected, {healthy_a["identity"], healthy_b["identity"]}) + + +class HardcodedClientRegressionTests(unittest.TestCase): + def test_unknown_client_does_not_resolve_to_codex(self): + for name in ("gemini", "antigravity", "grok", "some_new_llm", "", None): + with self.subTest(client=name): + self.assertNotEqual( + mcp_client_reconnect.normalize_client(name), + "codex", + "an unidentified client must never be handed Codex UI steps", + ) + + def test_known_clients_still_get_their_own_steps(self): + self.assertEqual(mcp_client_reconnect.normalize_client("codex"), "codex") + self.assertEqual( + mcp_client_reconnect.normalize_client("claude_code"), "claude_code" + ) + + def test_generic_steps_do_not_name_a_specific_vendor(self): + steps = " ".join(mcp_client_reconnect.operator_ui_steps("gemini")) + self.assertNotIn("Codex", steps) + + def test_reconnect_client_is_derived_from_the_attachment_record(self): + registry = _registry() + worker = _attach(registry, client="antigravity", session="s", generation="g") + verdict = mwi.assess_provenance( + registry=registry, worker_identity=worker["identity"], env={}, now=NOW + ) + self.assertEqual(mwi.reconnect_client_for(verdict), "antigravity") + + def test_reconnect_client_is_unknown_rather_than_guessed(self): + self.assertEqual(mwi.reconnect_client_for({}), mwi.UNKNOWN_CLIENT) + + +class RemoteBindingTests(unittest.TestCase): + def test_explicit_prgs_selection_is_honoured(self): + resolved = mwi.resolve_bound_remote( + requested_remote="prgs", bound_remote="prgs", default_remote="dadeschools" + ) + self.assertEqual(resolved["remote"], "prgs") + self.assertFalse(resolved["drifted"]) + + def test_omitted_remote_uses_the_binding_not_the_library_default(self): + """The reported dadeschools host drift.""" + resolved = mwi.resolve_bound_remote( + requested_remote=None, bound_remote="prgs", default_remote="dadeschools" + ) + self.assertEqual(resolved["remote"], "prgs") + self.assertNotEqual(resolved["remote"], "dadeschools") + self.assertEqual(resolved["resolved_from"], "session_binding") + + def test_contradicting_the_binding_is_refused(self): + resolved = mwi.resolve_bound_remote( + requested_remote="dadeschools", + bound_remote="prgs", + default_remote="dadeschools", + ) + self.assertEqual(resolved["remote"], "prgs") + self.assertTrue(resolved["drifted"]) + self.assertFalse(resolved["honoured_request"]) + + def test_unbound_session_falls_back_and_says_so(self): + resolved = mwi.resolve_bound_remote( + requested_remote=None, bound_remote=None, default_remote="dadeschools" + ) + self.assertEqual(resolved["remote"], "dadeschools") + self.assertEqual(resolved["resolved_from"], "library_default") + self.assertTrue(resolved["reasons"]) + + +class SurfaceAgreementTests(unittest.TestCase): + """The reproduced contradiction: two surfaces, one process, two answers.""" + + def test_namespace_health_and_direct_assessment_agree(self): + registry = _registry() + worker = _attach( + registry, + client="gemini", + session="sess-agree", + generation="gen-agree", + profile="prgs-reviewer", + ) + env = {"GITEA_MCP_PROFILE": "prgs-reviewer", "GITEA_CLIENT_MANAGED": "1"} + + direct = mwi.assess_provenance( + registry=registry, + worker_identity=worker["identity"], + env=env, + profile="prgs-reviewer", + ) + health = mcp_namespace_health.classify_namespace_probe( + "gitea-reviewer", + configured=True, + registered_tools=["gitea_whoami"], + probe_result={"success": True}, + probe_source="client_namespace", + process={"pid": 4242, "profile": "prgs-reviewer", "env": env}, + registry=registry, + worker_identity=worker["identity"], + ) + + self.assertEqual(health["provenance"], direct["provenance"]) + self.assertEqual(health["is_client_managed"], direct["is_client_managed"]) + self.assertEqual(health["worker_identity"], direct["worker_identity"]) + self.assertEqual(health["session_id"], "sess-agree") + self.assertEqual(health["client_name"], "gemini") + + def test_namespace_health_can_report_client_managed_at_all(self): + """The old derivation was structurally incapable of this.""" + env = {"GITEA_CLIENT_MANAGED": "1", "GITEA_MCP_PROFILE": "prgs-author"} + health = mcp_namespace_health.classify_namespace_probe( + "gitea-author", + configured=True, + registered_tools=["gitea_whoami"], + probe_result={"success": True}, + probe_source="client_namespace", + process={"pid": 1234, "profile": "prgs-author", "env": env}, + ) + self.assertTrue( + health["is_client_managed"], + "a client-managed launch must be reportable as client-managed", + ) + + def test_namespace_health_without_attachment_fails_closed(self): + health = mcp_namespace_health.classify_namespace_probe( + "gitea-author", + configured=True, + registered_tools=["gitea_whoami"], + probe_result={"success": True}, + probe_source="client_namespace", + process={"pid": 1234, "profile": "prgs-author", "env": {}}, + ) + self.assertTrue(health["provenance_fail_closed"]) + self.assertEqual(health["provenance"], mwi.PROVENANCE_UNPROVEN) + self.assertIsNone(health["session_id"]) + + def test_no_false_reconnect_loop_for_an_owned_session(self): + """A proven owner must not be told to reconnect.""" + registry = _registry() + worker = _attach(registry, client="claude", session="s", generation="g") + verdict = mwi.assess_provenance( + registry=registry, worker_identity=worker["identity"], env={}, now=NOW + ) + self.assertFalse(verdict["fail_closed"]) + self.assertEqual(verdict["blocker_kind"], mwi.BLOCKER_NONE) + self.assertEqual(verdict["reasons"], []) + + +if __name__ == "__main__": + unittest.main()