Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7e19079b5c | ||
|
|
32ab839289 |
@@ -34,8 +34,10 @@ namespaces.
|
||||
|
||||
1. Multiple active instances **may** share the same `client_type`.
|
||||
2. Every application launch receives a **distinct** `client_instance_id`.
|
||||
3. All five namespace workers of one launch report the **same**
|
||||
`client_instance_id`.
|
||||
3. All namespace workers of one launch report the **same**
|
||||
`client_instance_id`. For a whole-fleet launch that is five workers; for a
|
||||
project-scoped launch it is exactly the namespaces that launch started
|
||||
(#985).
|
||||
4. Each namespace worker has a **distinct** `worker_identity`, process
|
||||
identity, generation, and PID.
|
||||
5. Instance identity is **never** inferred from PID proximity, timestamps, or
|
||||
@@ -46,6 +48,92 @@ namespaces.
|
||||
This deliberately **replaces** any permanent `exactly_one_per_profile` fleet
|
||||
model (#949 assumption) as the operating rule for multi-instance fleets.
|
||||
|
||||
## Project-scoped launches and the runnable entry point (#985)
|
||||
|
||||
Not every project exposes all five namespaces. A project-scoped configuration
|
||||
such as Weekly Briefings intentionally has **author, reviewer, and merger
|
||||
only**, and has no controller or reconciler profile to supply. Requiring all
|
||||
five would force operators to invent profiles that must not exist, so the
|
||||
launcher accepts an explicitly validated subset.
|
||||
|
||||
### Running a launch
|
||||
|
||||
```bash
|
||||
python3 -m mcp_application_launcher \
|
||||
--client claude_code \
|
||||
--namespaces author,reviewer,merger \
|
||||
--profile author=prgs-author \
|
||||
--profile reviewer=prgs-reviewer \
|
||||
--profile merger=prgs-merger
|
||||
```
|
||||
|
||||
That mints one trusted `client_instance_id`, writes a per-launch `mcpServers`
|
||||
config, and executes:
|
||||
|
||||
```text
|
||||
claude --mcp-config <per-launch.json> --strict-mcp-config
|
||||
```
|
||||
|
||||
Add `--dry-run` to print the plan (namespaces, excluded namespaces, minted ID,
|
||||
argv, config path) as JSON and start nothing. Omit `--namespaces` to launch all
|
||||
five exactly as before. Arguments after the flags are forwarded to the client.
|
||||
|
||||
Other supported clients register one entry in
|
||||
`mcp_application_launcher.CLIENT_LAUNCH_SPECS`; because the argv builder only
|
||||
ever receives a config this module wrote, every client necessarily goes through
|
||||
the same mint-once/propagate-to-all mechanism.
|
||||
|
||||
### Why the config is per-launch and never `.mcp.json`
|
||||
|
||||
Persisting a trusted ID into a shared, reused `.mcp.json` would give two
|
||||
concurrent sessions **the same** trusted identity — precisely the reuse case
|
||||
the duplicate gate must reject. The launcher therefore writes a fresh
|
||||
owner-readable-only (`0600`) config per launch. Do not commit one, and do not
|
||||
copy a minted `GITEA_MCP_CLIENT_INSTANCE` into any checked-in configuration.
|
||||
|
||||
### Provenance sealing
|
||||
|
||||
The `inst-…` format is public and reproducible, so format alone cannot
|
||||
establish trust: anyone able to set one environment variable could otherwise
|
||||
hand-write a valid-looking ID and be believed. Trust therefore requires **both**
|
||||
the format and `GITEA_MCP_INSTANCE_PROVENANCE=trusted_launcher`, which only the
|
||||
launcher writes. A well-formed but unsealed identity is classified
|
||||
`unsealed_launcher` and **fails closed** — it is still reported for diagnosis,
|
||||
but never authorizes trusted attribution. Manually asserted trust is not
|
||||
possible.
|
||||
|
||||
### Migration and coordinated relaunch
|
||||
|
||||
Existing configurations that predate this work set no instance key at all, so
|
||||
their workers register under `legacy-pid-*` (`legacy_incomplete`) and remain
|
||||
mutually indistinguishable when two launches share a profile.
|
||||
|
||||
Migrating is a **coordinated relaunch**, not an in-place edit — a running
|
||||
worker cannot acquire an identity it was not started with:
|
||||
|
||||
1. Stop every LLM client currently running Gitea MCP workers. A single
|
||||
surviving legacy cohort keeps the fleet ambiguous.
|
||||
2. Relaunch each application through the command above.
|
||||
3. Repeat per application. Concurrent launches are expected and safe: each
|
||||
receives its own trusted ID.
|
||||
|
||||
### Post-launch verification
|
||||
|
||||
* `gitea_get_runtime_context` → `provenance_assessment.attachment` should show
|
||||
an `inst-…` `client_instance_id` with
|
||||
`instance_id_provenance: trusted_launcher`, not `legacy-pid-*` /
|
||||
`legacy_incomplete`.
|
||||
* Every namespace of the same launch must report that **same** ID; two
|
||||
different launches must report different IDs.
|
||||
* `gitea_resolve_task_capability` should return
|
||||
`exact_safe_next_action: "None; ready for operations."` with no
|
||||
`blocker_kind`. A `runtime_reconnect_required` naming duplicate PIDs per
|
||||
profile means at least one cohort is still on legacy identity, or a second
|
||||
cohort is genuinely running.
|
||||
* `mcp_application_launcher.collect_instance_ids_from_mcp_servers(servers,
|
||||
namespaces=[...])` returns `shared_single_trusted_id` for a built config, and
|
||||
reports `missing_servers` when an expected namespace entry is absent.
|
||||
|
||||
## How five workers join one instance
|
||||
|
||||
1. The host starts one application instance (for example one Codex session).
|
||||
|
||||
+6
-2
@@ -1286,16 +1286,19 @@ def multi_namespace_launcher_entries(
|
||||
profile_by_namespace,
|
||||
*,
|
||||
client_type,
|
||||
namespaces=None,
|
||||
config_path=None,
|
||||
client_instance_id=None,
|
||||
fleet_run_id=None,
|
||||
session_id=None,
|
||||
launch_nonce=None,
|
||||
):
|
||||
"""Build production mcpServers for all five namespaces of one application launch.
|
||||
"""Build production mcpServers for the namespaces of one application launch.
|
||||
|
||||
One shared trusted ``client_instance_id`` is minted (or reused) and
|
||||
propagated to every namespace worker env. See
|
||||
propagated to every namespace worker env. *namespaces* selects a validated
|
||||
project-scoped subset (#985); omitting it launches all five sanctioned
|
||||
namespaces as before. See
|
||||
:func:`mcp_application_launcher.build_application_mcp_servers`.
|
||||
"""
|
||||
import mcp_application_launcher as app_launcher
|
||||
@@ -1303,6 +1306,7 @@ def multi_namespace_launcher_entries(
|
||||
return app_launcher.build_application_mcp_servers(
|
||||
profile_by_namespace,
|
||||
client_type=client_type,
|
||||
namespaces=namespaces,
|
||||
config_path=config_path or DEFAULT_CONFIG_PATH,
|
||||
client_instance_id=client_instance_id,
|
||||
fleet_run_id=fleet_run_id,
|
||||
|
||||
+379
-19
@@ -15,7 +15,7 @@ from __future__ import annotations
|
||||
|
||||
import os
|
||||
import secrets
|
||||
from typing import Any, Mapping
|
||||
from typing import Any, Mapping, Sequence
|
||||
|
||||
import mcp_fleet_snapshot as fleet
|
||||
import mcp_worker_identity as mwi
|
||||
@@ -152,10 +152,54 @@ def namespace_worker_env(
|
||||
return env
|
||||
|
||||
|
||||
def resolve_launch_namespaces(
|
||||
namespaces: "Sequence[str] | None" = None,
|
||||
) -> tuple[str, ...]:
|
||||
"""Validate an explicit project-scoped namespace subset (#985).
|
||||
|
||||
``None`` keeps the historical whole-fleet behaviour and resolves to all five
|
||||
sanctioned namespaces. An explicit sequence is validated against
|
||||
:data:`SANCTIONED_NAMESPACES` and returned in canonical fleet order, so a
|
||||
project-scoped configuration (for example author/reviewer/merger only) can
|
||||
launch without inventing controller or reconciler profiles.
|
||||
|
||||
This is an allow-list, not a filter: anything outside the sanctioned set is
|
||||
refused rather than silently dropped, so a typo can never quietly shrink a
|
||||
launch to fewer workers than the operator intended.
|
||||
"""
|
||||
if namespaces is None:
|
||||
return tuple(SANCTIONED_NAMESPACES)
|
||||
|
||||
requested = [str(ns).strip().lower() for ns in namespaces]
|
||||
if not requested or any(not ns for ns in requested):
|
||||
raise ValueError(
|
||||
"namespaces must be a non-empty sequence of sanctioned namespace "
|
||||
f"names; choose from {list(SANCTIONED_NAMESPACES)}"
|
||||
)
|
||||
|
||||
unknown = sorted({ns for ns in requested if ns not in SANCTIONED_NAMESPACES})
|
||||
if unknown:
|
||||
raise ValueError(
|
||||
f"unsanctioned namespace(s) {unknown} requested; sanctioned "
|
||||
f"namespaces are {list(SANCTIONED_NAMESPACES)}"
|
||||
)
|
||||
|
||||
duplicates = sorted({ns for ns in requested if requested.count(ns) > 1})
|
||||
if duplicates:
|
||||
raise ValueError(
|
||||
f"duplicate namespace(s) {duplicates} requested; each sanctioned "
|
||||
"namespace may appear at most once in one launch"
|
||||
)
|
||||
|
||||
selected = set(requested)
|
||||
return tuple(ns for ns in SANCTIONED_NAMESPACES if ns in selected)
|
||||
|
||||
|
||||
def build_application_mcp_servers(
|
||||
profile_by_namespace: Mapping[str, str],
|
||||
*,
|
||||
client_type: str | None,
|
||||
namespaces: "Sequence[str] | None" = None,
|
||||
config_path: str | None = None,
|
||||
client_instance_id: str | None = None,
|
||||
fleet_run_id: str | None = None,
|
||||
@@ -175,14 +219,22 @@ def build_application_mcp_servers(
|
||||
|
||||
Every namespace server entry receives the **same** instance ID. Separate
|
||||
calls with no supplied ID receive distinct IDs.
|
||||
|
||||
*namespaces* selects an explicitly validated project-scoped subset (#985).
|
||||
Omitting it preserves the original whole-fleet behaviour, so existing
|
||||
five-namespace callers are unaffected. Only the resolved namespaces need a
|
||||
profile, and only they are started; excluded namespaces are reported so a
|
||||
caller can prove a controller/reconciler worker was never launched.
|
||||
"""
|
||||
resolved_namespaces = resolve_launch_namespaces(namespaces)
|
||||
|
||||
missing = [
|
||||
ns for ns in SANCTIONED_NAMESPACES if not profile_by_namespace.get(ns)
|
||||
ns for ns in resolved_namespaces if not profile_by_namespace.get(ns)
|
||||
]
|
||||
if missing:
|
||||
raise ValueError(
|
||||
"build_application_mcp_servers requires a profile for every "
|
||||
f"sanctioned namespace; missing: {missing}"
|
||||
f"requested namespace; missing: {missing}"
|
||||
)
|
||||
|
||||
if client_instance_id is None:
|
||||
@@ -213,6 +265,13 @@ def build_application_mcp_servers(
|
||||
"namespace_server_names": list(NAMESPACE_SERVER_NAMES),
|
||||
}
|
||||
|
||||
# The launch record describes only the namespaces this launch actually
|
||||
# starts, so downstream attribution never implies an unstarted worker.
|
||||
launch["namespaces"] = list(resolved_namespaces)
|
||||
launch["namespace_server_names"] = [
|
||||
f"gitea-{ns}" for ns in resolved_namespaces
|
||||
]
|
||||
|
||||
# Resolve command/args from the production server entry when not provided.
|
||||
if command is None or args is None:
|
||||
import gitea_config
|
||||
@@ -223,7 +282,7 @@ def build_application_mcp_servers(
|
||||
|
||||
servers: dict[str, Any] = {}
|
||||
shared_id = launch["client_instance_id"]
|
||||
for namespace in SANCTIONED_NAMESPACES:
|
||||
for namespace in resolved_namespaces:
|
||||
server_name = f"gitea-{namespace}"
|
||||
profile = profile_by_namespace[namespace]
|
||||
env = namespace_worker_env(
|
||||
@@ -240,14 +299,17 @@ def build_application_mcp_servers(
|
||||
"env": env,
|
||||
}
|
||||
|
||||
excluded = [ns for ns in SANCTIONED_NAMESPACES if ns not in resolved_namespaces]
|
||||
return {
|
||||
"mcpServers": servers,
|
||||
"launch": launch,
|
||||
"client_instance_id": shared_id,
|
||||
"client_type": launch["client_type"],
|
||||
"namespaces": list(SANCTIONED_NAMESPACES),
|
||||
"namespaces": list(resolved_namespaces),
|
||||
"excluded_namespaces": excluded,
|
||||
"project_scoped": bool(excluded),
|
||||
"shared_instance_id_across_namespaces": True,
|
||||
"namespace_count": len(SANCTIONED_NAMESPACES),
|
||||
"namespace_count": len(resolved_namespaces),
|
||||
}
|
||||
|
||||
|
||||
@@ -316,15 +378,33 @@ def launcher_entry_for_profile(
|
||||
|
||||
def collect_instance_ids_from_mcp_servers(
|
||||
mcp_servers: Mapping[str, Any],
|
||||
*,
|
||||
namespaces: "Sequence[str] | None" = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Inspect a production mcpServers map for shared instance attribution.
|
||||
|
||||
Returns the unique set of client_instance_id values across gitea-* servers
|
||||
and whether all five namespaces share exactly one trusted ID.
|
||||
Returns the unique set of client_instance_id values across the gitea-*
|
||||
servers of one launch and whether every worker of that launch shares
|
||||
exactly one trusted ID.
|
||||
|
||||
The inspected set is the launch's own namespaces, not an assumed five
|
||||
(#985): a project-scoped author/reviewer/merger launch must be able to
|
||||
prove shared attribution without a controller or reconciler entry counting
|
||||
as a missing worker. Pass *namespaces* to assert an exact expected subset;
|
||||
omit it to inspect whichever sanctioned namespace servers are present.
|
||||
"""
|
||||
if namespaces is None:
|
||||
expected_servers = [
|
||||
name for name in NAMESPACE_SERVER_NAMES if name in mcp_servers
|
||||
]
|
||||
else:
|
||||
expected_servers = [
|
||||
f"gitea-{ns}" for ns in resolve_launch_namespaces(namespaces)
|
||||
]
|
||||
|
||||
ids: list[str] = []
|
||||
by_server: dict[str, str | None] = {}
|
||||
for name in NAMESPACE_SERVER_NAMES:
|
||||
for name in expected_servers:
|
||||
entry = mcp_servers.get(name) or {}
|
||||
env = entry.get("env") or {}
|
||||
raw = (env.get(CLIENT_INSTANCE_ENV) or "").strip() or None
|
||||
@@ -337,14 +417,19 @@ def collect_instance_ids_from_mcp_servers(
|
||||
for i in unique
|
||||
if fleet.assess_instance_identity(i)["trusted"]
|
||||
]
|
||||
missing_servers = [n for n in expected_servers if n not in mcp_servers]
|
||||
return {
|
||||
"server_instance_ids": by_server,
|
||||
"unique_instance_ids": unique,
|
||||
"trusted_instance_ids": trusted,
|
||||
"inspected_servers": list(expected_servers),
|
||||
"missing_servers": missing_servers,
|
||||
"shared_single_trusted_id": (
|
||||
len(unique) == 1
|
||||
bool(expected_servers)
|
||||
and not missing_servers
|
||||
and len(unique) == 1
|
||||
and len(trusted) == 1
|
||||
and all(by_server.get(n) == unique[0] for n in NAMESPACE_SERVER_NAMES)
|
||||
and all(by_server.get(n) == unique[0] for n in expected_servers)
|
||||
),
|
||||
"namespace_server_count": sum(
|
||||
1 for n in NAMESPACE_SERVER_NAMES if n in mcp_servers
|
||||
@@ -365,20 +450,23 @@ def inherit_or_refuse_client_instance(
|
||||
source = dict(env if env is not None else os.environ)
|
||||
raw = (source.get(CLIENT_INSTANCE_ENV) or "").strip() or None
|
||||
provenance_marker = (source.get(INSTANCE_PROVENANCE_ENV) or "").strip()
|
||||
assessment = fleet.assess_instance_identity(raw)
|
||||
# Ordinary user-supplied values without launcher provenance marker are
|
||||
# still format-checked by assess_instance_identity. When the format is
|
||||
# trusted but the launcher marker is absent, keep the ID but note that
|
||||
# provenance is not launcher-sealed (operator hand-set or legacy config).
|
||||
assessment = dict(fleet.assess_instance_identity(raw))
|
||||
# The ``inst-…`` format is public and reproducible, so format alone cannot
|
||||
# establish trust: anyone able to set one env var could otherwise hand-write
|
||||
# a well-formed ID and be believed. Trust therefore requires the launcher's
|
||||
# own seal as well, and a well-formed but unsealed identity fails closed
|
||||
# rather than degrading to a mere annotation (#985).
|
||||
if assessment["trusted"] and provenance_marker != LAUNCHER_PROVENANCE_VALUE:
|
||||
assessment = dict(assessment)
|
||||
assessment["launcher_sealed"] = False
|
||||
assessment["trusted"] = False
|
||||
assessment["complete"] = False
|
||||
assessment["provenance"] = fleet.INSTANCE_ID_PROVENANCE_UNSEALED
|
||||
assessment["reasons"] = list(assessment.get("reasons") or []) + [
|
||||
f"{INSTANCE_PROVENANCE_ENV} is not {LAUNCHER_PROVENANCE_VALUE!r}; "
|
||||
"identity format is valid but not sealed by the production launcher"
|
||||
"identity format is valid but not sealed by the production "
|
||||
"launcher, so trusted attribution is refused"
|
||||
]
|
||||
else:
|
||||
assessment = dict(assessment)
|
||||
assessment["launcher_sealed"] = bool(
|
||||
assessment["trusted"]
|
||||
and provenance_marker == LAUNCHER_PROVENANCE_VALUE
|
||||
@@ -391,3 +479,275 @@ def inherit_or_refuse_client_instance(
|
||||
(source.get("GITEA_MCP_CLIENT") or "").strip() or None
|
||||
)
|
||||
return assessment
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Runnable launch path (#985)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
#: How each supported LLM client is started with a per-launch MCP config.
|
||||
#:
|
||||
#: The launch interface is deliberately a data-driven registry rather than a
|
||||
#: Claude-specific code path: adding another supported client is one entry, and
|
||||
#: every client necessarily goes through the same mint-once/propagate-to-all
|
||||
#: identity mechanism because the argv builder only ever receives a config file
|
||||
#: this module wrote.
|
||||
CLIENT_LAUNCH_SPECS: dict[str, dict[str, Any]] = {
|
||||
"claude_code": {
|
||||
"command": "claude",
|
||||
"config_args": lambda path: ["--mcp-config", path, "--strict-mcp-config"],
|
||||
"docs": "claude --mcp-config <per-launch.json> --strict-mcp-config",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def supported_launch_clients() -> tuple[str, ...]:
|
||||
"""Client types with a sanctioned runnable launch path."""
|
||||
return tuple(sorted(CLIENT_LAUNCH_SPECS))
|
||||
|
||||
|
||||
def build_client_launch_argv(
|
||||
client_type: str,
|
||||
config_path: str,
|
||||
*,
|
||||
extra_args: "Sequence[str] | None" = None,
|
||||
) -> list[str]:
|
||||
"""Build the argv that starts *client_type* against a per-launch config."""
|
||||
client = mwi.normalize_client_name(client_type)
|
||||
spec = CLIENT_LAUNCH_SPECS.get(client)
|
||||
if spec is None:
|
||||
raise ValueError(
|
||||
f"no sanctioned runnable launch path for client {client!r}; "
|
||||
f"supported clients: {list(supported_launch_clients())}"
|
||||
)
|
||||
argv = [str(spec["command"])] + list(spec["config_args"](str(config_path)))
|
||||
if extra_args:
|
||||
argv.extend(str(a) for a in extra_args)
|
||||
return argv
|
||||
|
||||
|
||||
def write_launch_config(
|
||||
mcp_servers: Mapping[str, Any],
|
||||
*,
|
||||
directory: str | None = None,
|
||||
) -> str:
|
||||
"""Write one launch's ``mcpServers`` config to a fresh per-launch file.
|
||||
|
||||
The file is per-launch and owner-readable only. #985 explicitly rules out
|
||||
persisting an instance ID into a shared, reused ``.mcp.json``: two
|
||||
concurrent sessions reading one static trusted ID is precisely the reuse
|
||||
case the duplicate guard must reject. A distinct file per launch keeps the
|
||||
minted identity bound to the launch that minted it.
|
||||
"""
|
||||
import json
|
||||
import tempfile
|
||||
|
||||
fd, path = tempfile.mkstemp(
|
||||
prefix="gitea-mcp-launch-", suffix=".json", dir=directory
|
||||
)
|
||||
try:
|
||||
os.fchmod(fd, 0o600)
|
||||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||||
json.dump({"mcpServers": dict(mcp_servers)}, handle, indent=2)
|
||||
handle.write("\n")
|
||||
except Exception:
|
||||
try:
|
||||
os.unlink(path)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
return path
|
||||
|
||||
|
||||
def prepare_application_launch(
|
||||
profile_by_namespace: Mapping[str, str],
|
||||
*,
|
||||
client_type: str,
|
||||
namespaces: "Sequence[str] | None" = None,
|
||||
config_path: str | None = None,
|
||||
fleet_run_id: str | None = None,
|
||||
session_id: str | None = None,
|
||||
launch_nonce: str | None = None,
|
||||
extra_args: "Sequence[str] | None" = None,
|
||||
config_directory: str | None = None,
|
||||
now=None,
|
||||
) -> dict[str, Any]:
|
||||
"""Mint one launch and produce everything needed to exec the client.
|
||||
|
||||
Returns the built servers, the per-launch config file path, the argv to
|
||||
exec, and the shared trusted identity — without starting anything, so the
|
||||
same preparation is testable and inspectable via ``--dry-run``.
|
||||
"""
|
||||
built = build_application_mcp_servers(
|
||||
profile_by_namespace,
|
||||
client_type=client_type,
|
||||
namespaces=namespaces,
|
||||
config_path=config_path,
|
||||
fleet_run_id=fleet_run_id,
|
||||
session_id=session_id,
|
||||
launch_nonce=launch_nonce,
|
||||
now=now,
|
||||
)
|
||||
# Fail before writing anything if this client has no runnable path.
|
||||
argv_probe = build_client_launch_argv(
|
||||
client_type, "<pending>", extra_args=extra_args
|
||||
)
|
||||
launch_config_path = write_launch_config(
|
||||
built["mcpServers"], directory=config_directory
|
||||
)
|
||||
argv = build_client_launch_argv(
|
||||
client_type, launch_config_path, extra_args=extra_args
|
||||
)
|
||||
proof = collect_instance_ids_from_mcp_servers(
|
||||
built["mcpServers"], namespaces=built["namespaces"]
|
||||
)
|
||||
return {
|
||||
"client_type": built["client_type"],
|
||||
"client_instance_id": built["client_instance_id"],
|
||||
"namespaces": built["namespaces"],
|
||||
"excluded_namespaces": built["excluded_namespaces"],
|
||||
"project_scoped": built["project_scoped"],
|
||||
"launch_config_path": launch_config_path,
|
||||
"argv": argv,
|
||||
"argv_template": argv_probe,
|
||||
"mcpServers": built["mcpServers"],
|
||||
"launch": built["launch"],
|
||||
"shared_single_trusted_id": proof["shared_single_trusted_id"],
|
||||
"attribution_proof": proof,
|
||||
}
|
||||
|
||||
|
||||
def _parse_profile_assignments(values: "Sequence[str]") -> dict[str, str]:
|
||||
"""Parse ``namespace=profile`` CLI pairs into a mapping."""
|
||||
profiles: dict[str, str] = {}
|
||||
for raw in values or ():
|
||||
text = str(raw)
|
||||
if "=" not in text:
|
||||
raise ValueError(
|
||||
f"invalid --profile {text!r}; expected namespace=profile-name"
|
||||
)
|
||||
namespace, _, profile = text.partition("=")
|
||||
namespace = namespace.strip().lower()
|
||||
profile = profile.strip()
|
||||
if not namespace or not profile:
|
||||
raise ValueError(
|
||||
f"invalid --profile {text!r}; expected namespace=profile-name"
|
||||
)
|
||||
if namespace in profiles:
|
||||
raise ValueError(f"duplicate --profile entry for namespace {namespace!r}")
|
||||
profiles[namespace] = profile
|
||||
return profiles
|
||||
|
||||
|
||||
def main(argv: "Sequence[str] | None" = None) -> int:
|
||||
"""Runnable entry point: start a supported LLM client for one launch.
|
||||
|
||||
Example (project-scoped, three namespaces)::
|
||||
|
||||
python3 -m mcp_application_launcher \\
|
||||
--client claude_code \\
|
||||
--namespaces author,reviewer,merger \\
|
||||
--profile author=prgs-author \\
|
||||
--profile reviewer=prgs-reviewer \\
|
||||
--profile merger=prgs-merger
|
||||
"""
|
||||
import argparse
|
||||
import json
|
||||
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="mcp_application_launcher",
|
||||
description=(
|
||||
"Start a supported LLM client with one trusted, per-launch "
|
||||
"GITEA_MCP_CLIENT_INSTANCE shared by every MCP namespace worker."
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--client",
|
||||
default="claude_code",
|
||||
help=f"client to launch; supported: {list(supported_launch_clients())}",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--namespaces",
|
||||
default=None,
|
||||
help=(
|
||||
"comma-separated sanctioned namespace subset (e.g. "
|
||||
"'author,reviewer,merger'); omit to launch all five"
|
||||
),
|
||||
)
|
||||
parser.add_argument(
|
||||
"--profile",
|
||||
action="append",
|
||||
default=[],
|
||||
metavar="NAMESPACE=PROFILE",
|
||||
help="profile for one namespace; repeat once per launched namespace",
|
||||
)
|
||||
parser.add_argument("--config-path", default=None, help="profiles.json path")
|
||||
parser.add_argument("--fleet-run-id", default=None)
|
||||
parser.add_argument("--session-id", default=None)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="print the launch plan as JSON and exit without starting anything",
|
||||
)
|
||||
parser.add_argument(
|
||||
"client_args",
|
||||
nargs=argparse.REMAINDER,
|
||||
help="arguments forwarded to the client after '--'",
|
||||
)
|
||||
ns = parser.parse_args(list(argv) if argv is not None else None)
|
||||
|
||||
try:
|
||||
profiles = _parse_profile_assignments(ns.profile)
|
||||
requested = (
|
||||
[part.strip() for part in ns.namespaces.split(",") if part.strip()]
|
||||
if ns.namespaces
|
||||
else None
|
||||
)
|
||||
resolved = resolve_launch_namespaces(requested)
|
||||
# Profiles for namespaces that are not being launched are almost always
|
||||
# a mistake (a stale five-namespace invocation), so refuse rather than
|
||||
# silently ignore them.
|
||||
stray = sorted(set(profiles) - set(resolved))
|
||||
if stray:
|
||||
raise ValueError(
|
||||
f"--profile given for namespace(s) {stray} that this launch "
|
||||
f"does not start; launched namespaces are {list(resolved)}"
|
||||
)
|
||||
forwarded = [a for a in (ns.client_args or []) if a != "--"]
|
||||
prepared = prepare_application_launch(
|
||||
profiles,
|
||||
client_type=ns.client,
|
||||
namespaces=requested,
|
||||
config_path=ns.config_path,
|
||||
fleet_run_id=ns.fleet_run_id,
|
||||
session_id=ns.session_id,
|
||||
extra_args=forwarded,
|
||||
)
|
||||
except ValueError as exc:
|
||||
parser.error(str(exc))
|
||||
return 2 # pragma: no cover - argparse.error raises SystemExit
|
||||
|
||||
if ns.dry_run:
|
||||
plan = {
|
||||
key: prepared[key]
|
||||
for key in (
|
||||
"client_type",
|
||||
"client_instance_id",
|
||||
"namespaces",
|
||||
"excluded_namespaces",
|
||||
"project_scoped",
|
||||
"launch_config_path",
|
||||
"argv",
|
||||
"shared_single_trusted_id",
|
||||
)
|
||||
}
|
||||
print(json.dumps(plan, indent=2))
|
||||
return 0
|
||||
|
||||
argv_to_exec = prepared["argv"]
|
||||
os.execvp(argv_to_exec[0], argv_to_exec)
|
||||
return 0 # pragma: no cover - execvp does not return
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover - process entry point
|
||||
raise SystemExit(main())
|
||||
|
||||
@@ -84,6 +84,10 @@ SANCTIONED_NAMESPACES = frozenset(
|
||||
INSTANCE_ID_PROVENANCE_TRUSTED = "trusted_launcher"
|
||||
INSTANCE_ID_PROVENANCE_LEGACY = "legacy_incomplete"
|
||||
INSTANCE_ID_PROVENANCE_MISSING = "missing"
|
||||
#: Well-formed ``inst-…`` identity presented without the launcher's own
|
||||
#: provenance seal (#985). The format alone is public and reproducible, so an
|
||||
#: unsealed value is treated as manually asserted trust and fails closed.
|
||||
INSTANCE_ID_PROVENANCE_UNSEALED = "unsealed_launcher"
|
||||
|
||||
CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE"
|
||||
FLEET_RUN_ENV = "GITEA_MCP_FLEET_RUN_ID"
|
||||
|
||||
@@ -1032,9 +1032,43 @@ class ClientHintsTests(unittest.TestCase):
|
||||
)
|
||||
|
||||
def test_client_hints_trusted_when_set(self):
|
||||
"""Trusted requires the launcher seal as well as the format (#985).
|
||||
|
||||
Before #985 a well-formed ``inst-…`` value alone was accepted as
|
||||
trusted. The format is public and reproducible, so anyone able to set
|
||||
one environment variable could hand-write a valid-looking ID; #985
|
||||
therefore additionally requires the provenance marker that only the
|
||||
production launcher writes. This case now supplies the full sealed env
|
||||
the launcher actually emits.
|
||||
"""
|
||||
import gitea_mcp_server as server
|
||||
|
||||
inst = fleet.generate_client_instance_id("codex", launch_nonce="t", now=NOW)
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
"GITEA_MCP_CLIENT": "codex",
|
||||
"GITEA_MCP_CLIENT_INSTANCE": inst,
|
||||
"GITEA_MCP_INSTANCE_PROVENANCE": (
|
||||
fleet.INSTANCE_ID_PROVENANCE_TRUSTED
|
||||
),
|
||||
"GITEA_MCP_FLEET_RUN_ID": "run-1",
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
hints = server._client_identity_hints()
|
||||
self.assertTrue(hints["instance_identity_trusted"])
|
||||
self.assertTrue(hints["instance_launcher_sealed"])
|
||||
self.assertEqual(hints["client_instance_id"], inst)
|
||||
self.assertEqual(hints["fleet_run_id"], "run-1")
|
||||
|
||||
def test_client_hints_wellformed_but_unsealed_not_trusted(self):
|
||||
"""#985: a valid-looking ID without the launcher seal fails closed."""
|
||||
import gitea_mcp_server as server
|
||||
|
||||
inst = fleet.generate_client_instance_id(
|
||||
"codex", launch_nonce="unsealed", now=NOW
|
||||
)
|
||||
with mock.patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
@@ -1044,10 +1078,16 @@ class ClientHintsTests(unittest.TestCase):
|
||||
},
|
||||
clear=False,
|
||||
):
|
||||
os.environ.pop("GITEA_MCP_INSTANCE_PROVENANCE", None)
|
||||
hints = server._client_identity_hints()
|
||||
self.assertTrue(hints["instance_identity_trusted"])
|
||||
self.assertFalse(hints["instance_identity_trusted"])
|
||||
self.assertFalse(hints["instance_launcher_sealed"])
|
||||
# Still reported for diagnosis rather than silently discarded.
|
||||
self.assertEqual(hints["client_instance_id"], inst)
|
||||
self.assertEqual(hints["fleet_run_id"], "run-1")
|
||||
self.assertEqual(
|
||||
hints["instance_id_provenance"],
|
||||
fleet.INSTANCE_ID_PROVENANCE_UNSEALED,
|
||||
)
|
||||
|
||||
def test_malformed_user_supplied_instance_not_trusted(self):
|
||||
import gitea_mcp_server as server
|
||||
|
||||
@@ -0,0 +1,664 @@
|
||||
"""#985 — trusted client-instance identity for project-scoped LLM launches.
|
||||
|
||||
Covers the acceptance criteria of issue #985:
|
||||
|
||||
* a three-namespace (author/reviewer/merger) launch succeeds with no
|
||||
controller or reconciler profile,
|
||||
* every worker of one launch shares exactly one trusted instance ID,
|
||||
* two concurrent launches receive distinct IDs and stay distinguishable,
|
||||
* reuse, missing, legacy, malformed, unsanctioned, and unsealed identities all
|
||||
fail closed,
|
||||
* the existing five-namespace behaviour is unchanged,
|
||||
* Claude Code has a documented, runnable launch command.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import datetime as _dt
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
import mcp_application_launcher as launcher # noqa: E402
|
||||
import mcp_fleet_snapshot as fleet # noqa: E402
|
||||
|
||||
NOW = _dt.datetime(2026, 7, 31, 12, 0, 0, tzinfo=_dt.timezone.utc)
|
||||
|
||||
PROJECT_NAMESPACES = ("author", "reviewer", "merger")
|
||||
ALL_NAMESPACES = ("author", "reviewer", "merger", "controller", "reconciler")
|
||||
|
||||
#: A project-scoped configuration intentionally has no controller/reconciler
|
||||
#: profile — that absence is the condition under test, not an omission.
|
||||
PROJECT_PROFILES = {ns: f"prgs-{ns}" for ns in PROJECT_NAMESPACES}
|
||||
FULL_PROFILES = {ns: f"prgs-{ns}" for ns in ALL_NAMESPACES}
|
||||
|
||||
|
||||
def _instance_ids(servers):
|
||||
return {
|
||||
name: entry["env"][launcher.CLIENT_INSTANCE_ENV]
|
||||
for name, entry in servers.items()
|
||||
}
|
||||
|
||||
|
||||
class ResolveLaunchNamespacesTests(unittest.TestCase):
|
||||
"""The validated subset allow-list."""
|
||||
|
||||
def test_none_resolves_to_all_five_in_canonical_order(self):
|
||||
self.assertEqual(
|
||||
launcher.resolve_launch_namespaces(None), tuple(ALL_NAMESPACES)
|
||||
)
|
||||
|
||||
def test_project_subset_resolves_in_canonical_order(self):
|
||||
# Deliberately out of order on input; canonical order on output.
|
||||
self.assertEqual(
|
||||
launcher.resolve_launch_namespaces(["merger", "author", "reviewer"]),
|
||||
("author", "reviewer", "merger"),
|
||||
)
|
||||
|
||||
def test_case_and_whitespace_normalised(self):
|
||||
self.assertEqual(
|
||||
launcher.resolve_launch_namespaces([" Author ", "REVIEWER"]),
|
||||
("author", "reviewer"),
|
||||
)
|
||||
|
||||
def test_unsanctioned_namespace_refused(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
launcher.resolve_launch_namespaces(["author", "admin"])
|
||||
self.assertIn("admin", str(ctx.exception))
|
||||
|
||||
def test_typo_is_refused_not_silently_dropped(self):
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.resolve_launch_namespaces(["author", "reviewr"])
|
||||
|
||||
def test_empty_subset_refused(self):
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.resolve_launch_namespaces([])
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.resolve_launch_namespaces([" "])
|
||||
|
||||
def test_duplicate_namespace_refused(self):
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.resolve_launch_namespaces(["author", "author"])
|
||||
|
||||
|
||||
class ProjectScopedLaunchTests(unittest.TestCase):
|
||||
"""AC: three-namespace launch without controller/reconciler profiles."""
|
||||
|
||||
def _build(self, **kwargs):
|
||||
kwargs.setdefault("client_type", "claude_code")
|
||||
kwargs.setdefault("namespaces", PROJECT_NAMESPACES)
|
||||
kwargs.setdefault("command", "python3")
|
||||
kwargs.setdefault("args", ["mcp_server.py"])
|
||||
profiles = kwargs.pop("profiles", PROJECT_PROFILES)
|
||||
return launcher.build_application_mcp_servers(profiles, **kwargs)
|
||||
|
||||
def test_three_namespace_launch_succeeds_without_controller_reconciler(self):
|
||||
built = self._build(launch_nonce="proj-1", now=NOW)
|
||||
servers = built["mcpServers"]
|
||||
self.assertEqual(
|
||||
sorted(servers), ["gitea-author", "gitea-merger", "gitea-reviewer"]
|
||||
)
|
||||
self.assertEqual(built["namespace_count"], 3)
|
||||
self.assertEqual(list(built["namespaces"]), list(PROJECT_NAMESPACES))
|
||||
|
||||
def test_excluded_namespaces_are_not_started(self):
|
||||
built = self._build(launch_nonce="proj-2", now=NOW)
|
||||
self.assertNotIn("gitea-controller", built["mcpServers"])
|
||||
self.assertNotIn("gitea-reconciler", built["mcpServers"])
|
||||
self.assertEqual(
|
||||
sorted(built["excluded_namespaces"]), ["controller", "reconciler"]
|
||||
)
|
||||
self.assertTrue(built["project_scoped"])
|
||||
|
||||
def test_all_workers_of_one_launch_share_one_trusted_id(self):
|
||||
built = self._build(launch_nonce="proj-3", now=NOW)
|
||||
ids = set(_instance_ids(built["mcpServers"]).values())
|
||||
self.assertEqual(len(ids), 1, ids)
|
||||
shared = next(iter(ids))
|
||||
self.assertTrue(fleet.assess_instance_identity(shared)["trusted"])
|
||||
self.assertEqual(built["client_instance_id"], shared)
|
||||
|
||||
def test_every_worker_env_is_launcher_sealed(self):
|
||||
built = self._build(launch_nonce="proj-4", now=NOW)
|
||||
for name, entry in built["mcpServers"].items():
|
||||
env = entry["env"]
|
||||
self.assertEqual(
|
||||
env[launcher.INSTANCE_PROVENANCE_ENV],
|
||||
fleet.INSTANCE_ID_PROVENANCE_TRUSTED,
|
||||
name,
|
||||
)
|
||||
self.assertEqual(env["GITEA_CLIENT_MANAGED"], "1", name)
|
||||
self.assertFalse(
|
||||
env[launcher.CLIENT_INSTANCE_ENV].startswith("legacy-"), name
|
||||
)
|
||||
|
||||
def test_attribution_proof_holds_for_three_namespace_launch(self):
|
||||
"""A subset launch must not read as 'missing two workers'."""
|
||||
built = self._build(launch_nonce="proj-5", now=NOW)
|
||||
proof = launcher.collect_instance_ids_from_mcp_servers(
|
||||
built["mcpServers"], namespaces=PROJECT_NAMESPACES
|
||||
)
|
||||
self.assertTrue(proof["shared_single_trusted_id"], proof)
|
||||
self.assertEqual(proof["missing_servers"], [])
|
||||
# And without an explicit expectation it inspects what is present.
|
||||
inferred = launcher.collect_instance_ids_from_mcp_servers(
|
||||
built["mcpServers"]
|
||||
)
|
||||
self.assertTrue(inferred["shared_single_trusted_id"], inferred)
|
||||
|
||||
def test_missing_profile_for_requested_namespace_refused(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
self._build(profiles={"author": "prgs-author"})
|
||||
self.assertIn("merger", str(ctx.exception))
|
||||
|
||||
def test_profile_for_unlaunched_namespace_is_ignored_not_started(self):
|
||||
"""Extra profiles never widen the launch beyond the validated subset."""
|
||||
built = self._build(profiles=FULL_PROFILES, launch_nonce="proj-6", now=NOW)
|
||||
self.assertEqual(len(built["mcpServers"]), 3)
|
||||
self.assertNotIn("gitea-controller", built["mcpServers"])
|
||||
|
||||
def test_unsanctioned_namespace_refused_at_build(self):
|
||||
with self.assertRaises(ValueError):
|
||||
self._build(
|
||||
profiles={**PROJECT_PROFILES, "admin": "prgs-admin"},
|
||||
namespaces=["author", "admin"],
|
||||
)
|
||||
|
||||
|
||||
class ConcurrentLaunchDistinctionTests(unittest.TestCase):
|
||||
"""AC: concurrent launches are distinct and distinguishable."""
|
||||
|
||||
def _build(self, nonce, namespaces=PROJECT_NAMESPACES, profiles=None):
|
||||
return launcher.build_application_mcp_servers(
|
||||
profiles or PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=namespaces,
|
||||
launch_nonce=nonce,
|
||||
now=NOW,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
|
||||
def test_two_concurrent_launches_receive_different_ids(self):
|
||||
a = self._build("concurrent-a")
|
||||
b = self._build("concurrent-b")
|
||||
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
|
||||
|
||||
def test_same_profile_across_launches_stays_distinguishable(self):
|
||||
"""Same profile, two launches: distinguishable, so not a false duplicate."""
|
||||
a = self._build("dist-a")
|
||||
b = self._build("dist-b")
|
||||
a_author = a["mcpServers"]["gitea-author"]["env"]
|
||||
b_author = b["mcpServers"]["gitea-author"]["env"]
|
||||
# Same profile on purpose — that alone must never imply duplication.
|
||||
self.assertEqual(a_author["GITEA_MCP_PROFILE"], b_author["GITEA_MCP_PROFILE"])
|
||||
self.assertNotEqual(
|
||||
a_author[launcher.CLIENT_INSTANCE_ENV],
|
||||
b_author[launcher.CLIENT_INSTANCE_ENV],
|
||||
)
|
||||
for env in (a_author, b_author):
|
||||
assessed = launcher.inherit_or_refuse_client_instance(env)
|
||||
self.assertTrue(assessed["trusted"], assessed)
|
||||
self.assertTrue(assessed["launcher_sealed"], assessed)
|
||||
|
||||
def test_no_nonce_still_yields_distinct_ids(self):
|
||||
a = self._build(None)
|
||||
b = self._build(None)
|
||||
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
|
||||
|
||||
def test_ids_are_never_persisted_between_launches(self):
|
||||
"""No static per-project ID: two runs of one config differ."""
|
||||
first = self._build("static-check-1")
|
||||
second = self._build("static-check-2")
|
||||
self.assertNotEqual(
|
||||
first["mcpServers"]["gitea-author"]["env"][launcher.CLIENT_INSTANCE_ENV],
|
||||
second["mcpServers"]["gitea-author"]["env"][launcher.CLIENT_INSTANCE_ENV],
|
||||
)
|
||||
|
||||
|
||||
class FailClosedIdentityTests(unittest.TestCase):
|
||||
"""AC: missing / legacy / reused / conflicting / unsealed all fail closed."""
|
||||
|
||||
def test_missing_identity_fails_closed(self):
|
||||
assessed = launcher.inherit_or_refuse_client_instance({})
|
||||
self.assertFalse(assessed["trusted"])
|
||||
self.assertFalse(assessed["launcher_sealed"])
|
||||
self.assertEqual(
|
||||
assessed["provenance"], fleet.INSTANCE_ID_PROVENANCE_MISSING
|
||||
)
|
||||
|
||||
def test_legacy_pid_identity_fails_closed(self):
|
||||
for legacy in ("legacy-pid-40578", "pid-1234", "proc-99"):
|
||||
with self.subTest(legacy=legacy):
|
||||
assessed = launcher.inherit_or_refuse_client_instance(
|
||||
{
|
||||
launcher.CLIENT_INSTANCE_ENV: legacy,
|
||||
launcher.INSTANCE_PROVENANCE_ENV: (
|
||||
fleet.INSTANCE_ID_PROVENANCE_TRUSTED
|
||||
),
|
||||
}
|
||||
)
|
||||
self.assertFalse(assessed["trusted"], legacy)
|
||||
self.assertFalse(assessed["launcher_sealed"], legacy)
|
||||
|
||||
def test_wellformed_but_unsealed_identity_fails_closed(self):
|
||||
"""The decisive case: format alone must not manufacture trust."""
|
||||
forged = fleet.generate_client_instance_id(
|
||||
"claude_code", launch_nonce="forged", now=NOW
|
||||
)
|
||||
# A hand-set env var with a valid-looking ID but no launcher seal.
|
||||
assessed = launcher.inherit_or_refuse_client_instance(
|
||||
{launcher.CLIENT_INSTANCE_ENV: forged}
|
||||
)
|
||||
self.assertFalse(assessed["trusted"], assessed)
|
||||
self.assertFalse(assessed["launcher_sealed"], assessed)
|
||||
self.assertEqual(
|
||||
assessed["provenance"], fleet.INSTANCE_ID_PROVENANCE_UNSEALED
|
||||
)
|
||||
# The value is still reported for diagnosis, not silently dropped.
|
||||
self.assertEqual(assessed["client_instance_id"], forged)
|
||||
|
||||
def test_wrong_provenance_marker_fails_closed(self):
|
||||
forged = fleet.generate_client_instance_id(
|
||||
"claude_code", launch_nonce="wrong-marker", now=NOW
|
||||
)
|
||||
assessed = launcher.inherit_or_refuse_client_instance(
|
||||
{
|
||||
launcher.CLIENT_INSTANCE_ENV: forged,
|
||||
launcher.INSTANCE_PROVENANCE_ENV: "totally_trusted",
|
||||
}
|
||||
)
|
||||
self.assertFalse(assessed["trusted"], assessed)
|
||||
self.assertEqual(
|
||||
assessed["provenance"], fleet.INSTANCE_ID_PROVENANCE_UNSEALED
|
||||
)
|
||||
|
||||
def test_sealed_launcher_env_is_trusted(self):
|
||||
env = launcher.namespace_worker_env(
|
||||
profile_name="prgs-author",
|
||||
client_type="claude_code",
|
||||
client_instance_id=fleet.generate_client_instance_id(
|
||||
"claude_code", launch_nonce="sealed", now=NOW
|
||||
),
|
||||
)
|
||||
assessed = launcher.inherit_or_refuse_client_instance(env)
|
||||
self.assertTrue(assessed["trusted"], assessed)
|
||||
self.assertTrue(assessed["launcher_sealed"], assessed)
|
||||
|
||||
def test_untrusted_id_refused_on_build_and_worker_env(self):
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.build_application_mcp_servers(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
client_instance_id="legacy-pid-1",
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.namespace_worker_env(
|
||||
profile_name="prgs-author",
|
||||
client_type="claude_code",
|
||||
client_instance_id="hand-written",
|
||||
)
|
||||
|
||||
def test_extra_env_cannot_forge_seal_or_identity(self):
|
||||
trusted = fleet.generate_client_instance_id(
|
||||
"claude_code", launch_nonce="seal-guard", now=NOW
|
||||
)
|
||||
env = launcher.namespace_worker_env(
|
||||
profile_name="prgs-author",
|
||||
client_type="claude_code",
|
||||
client_instance_id=trusted,
|
||||
extra_env={
|
||||
launcher.CLIENT_INSTANCE_ENV: "inst-spoof-20260101T000000Z-abcdef",
|
||||
launcher.INSTANCE_PROVENANCE_ENV: "trusted_launcher",
|
||||
"GITEA_CLIENT_MANAGED": "0",
|
||||
},
|
||||
)
|
||||
self.assertEqual(env[launcher.CLIENT_INSTANCE_ENV], trusted)
|
||||
self.assertEqual(env["GITEA_CLIENT_MANAGED"], "1")
|
||||
|
||||
def test_conflicting_ids_across_workers_break_shared_attribution(self):
|
||||
"""One launch whose workers disagree must not read as shared."""
|
||||
built = launcher.build_application_mcp_servers(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
launch_nonce="conflict",
|
||||
now=NOW,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
servers = built["mcpServers"]
|
||||
servers["gitea-merger"]["env"][launcher.CLIENT_INSTANCE_ENV] = (
|
||||
fleet.generate_client_instance_id(
|
||||
"claude_code", launch_nonce="other", now=NOW
|
||||
)
|
||||
)
|
||||
proof = launcher.collect_instance_ids_from_mcp_servers(
|
||||
servers, namespaces=PROJECT_NAMESPACES
|
||||
)
|
||||
self.assertFalse(proof["shared_single_trusted_id"], proof)
|
||||
self.assertEqual(len(proof["unique_instance_ids"]), 2)
|
||||
|
||||
def test_absent_expected_worker_breaks_shared_attribution(self):
|
||||
built = launcher.build_application_mcp_servers(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
launch_nonce="absent",
|
||||
now=NOW,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
servers = dict(built["mcpServers"])
|
||||
servers.pop("gitea-merger")
|
||||
proof = launcher.collect_instance_ids_from_mcp_servers(
|
||||
servers, namespaces=PROJECT_NAMESPACES
|
||||
)
|
||||
self.assertFalse(proof["shared_single_trusted_id"], proof)
|
||||
self.assertEqual(proof["missing_servers"], ["gitea-merger"])
|
||||
|
||||
def test_reused_trusted_id_is_reuse_not_a_fresh_launch(self):
|
||||
"""Resume reuses by design; two *live* launches sharing it is the hazard."""
|
||||
first = launcher.build_application_mcp_servers(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
launch_nonce="reuse-src",
|
||||
now=NOW,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
resumed = launcher.build_application_mcp_servers(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
client_instance_id=first["client_instance_id"],
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
self.assertEqual(
|
||||
first["client_instance_id"], resumed["client_instance_id"]
|
||||
)
|
||||
# Two independent launches must never collide by default.
|
||||
independent = launcher.build_application_mcp_servers(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
self.assertNotEqual(
|
||||
first["client_instance_id"], independent["client_instance_id"]
|
||||
)
|
||||
|
||||
|
||||
class BackwardCompatibilityTests(unittest.TestCase):
|
||||
"""AC: existing full five-namespace behaviour remains compatible."""
|
||||
|
||||
def test_omitting_namespaces_launches_all_five(self):
|
||||
built = launcher.build_application_mcp_servers(
|
||||
FULL_PROFILES,
|
||||
client_type="codex",
|
||||
launch_nonce="compat",
|
||||
now=NOW,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
self.assertEqual(len(built["mcpServers"]), 5)
|
||||
self.assertEqual(built["namespace_count"], 5)
|
||||
self.assertEqual(built["excluded_namespaces"], [])
|
||||
self.assertFalse(built["project_scoped"])
|
||||
ids = set(_instance_ids(built["mcpServers"]).values())
|
||||
self.assertEqual(len(ids), 1)
|
||||
|
||||
def test_five_namespace_missing_profile_still_refused(self):
|
||||
incomplete = {ns: f"prgs-{ns}" for ns in PROJECT_NAMESPACES}
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
launcher.build_application_mcp_servers(
|
||||
incomplete,
|
||||
client_type="codex",
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
message = str(ctx.exception)
|
||||
self.assertIn("controller", message)
|
||||
self.assertIn("reconciler", message)
|
||||
|
||||
def test_legacy_collect_call_without_namespaces_still_works(self):
|
||||
built = launcher.build_application_mcp_servers(
|
||||
FULL_PROFILES,
|
||||
client_type="codex",
|
||||
launch_nonce="compat-proof",
|
||||
now=NOW,
|
||||
command="python3",
|
||||
args=["mcp_server.py"],
|
||||
)
|
||||
proof = launcher.collect_instance_ids_from_mcp_servers(built["mcpServers"])
|
||||
self.assertTrue(proof["shared_single_trusted_id"], proof)
|
||||
self.assertEqual(proof["namespace_server_count"], 5)
|
||||
|
||||
def test_gitea_config_wrapper_supports_subset(self):
|
||||
import gitea_config
|
||||
|
||||
built = gitea_config.multi_namespace_launcher_entries(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
config_path="/cfg/profiles.json",
|
||||
launch_nonce="cfg-subset",
|
||||
)
|
||||
self.assertEqual(built["namespace_count"], 3)
|
||||
self.assertEqual(
|
||||
sorted(built["excluded_namespaces"]), ["controller", "reconciler"]
|
||||
)
|
||||
|
||||
def test_gitea_config_wrapper_default_is_five(self):
|
||||
import gitea_config
|
||||
|
||||
built = gitea_config.multi_namespace_launcher_entries(
|
||||
FULL_PROFILES,
|
||||
client_type="claude_code",
|
||||
config_path="/cfg/profiles.json",
|
||||
launch_nonce="cfg-full",
|
||||
)
|
||||
self.assertEqual(built["namespace_count"], 5)
|
||||
|
||||
|
||||
class RunnableLaunchPathTests(unittest.TestCase):
|
||||
"""AC: Claude Code has a documented supported launch command."""
|
||||
|
||||
def test_claude_code_is_a_supported_client(self):
|
||||
self.assertIn("claude_code", launcher.supported_launch_clients())
|
||||
|
||||
def test_claude_code_argv_uses_per_launch_config(self):
|
||||
argv = launcher.build_client_launch_argv("claude_code", "/tmp/launch.json")
|
||||
self.assertEqual(argv[0], "claude")
|
||||
self.assertIn("--mcp-config", argv)
|
||||
self.assertIn("/tmp/launch.json", argv)
|
||||
self.assertIn("--strict-mcp-config", argv)
|
||||
|
||||
def test_extra_args_are_forwarded(self):
|
||||
argv = launcher.build_client_launch_argv(
|
||||
"claude_code", "/tmp/launch.json", extra_args=["--resume"]
|
||||
)
|
||||
self.assertEqual(argv[-1], "--resume")
|
||||
|
||||
def test_unsupported_client_refused(self):
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.build_client_launch_argv("not_a_client", "/tmp/x.json")
|
||||
|
||||
def test_prepare_launch_writes_private_per_launch_config(self):
|
||||
prepared = launcher.prepare_application_launch(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
launch_nonce="prep-1",
|
||||
now=NOW,
|
||||
)
|
||||
path = prepared["launch_config_path"]
|
||||
self.addCleanup(lambda: os.path.exists(path) and os.unlink(path))
|
||||
self.assertTrue(os.path.exists(path))
|
||||
self.assertEqual(os.stat(path).st_mode & 0o777, 0o600)
|
||||
with open(path, encoding="utf-8") as handle:
|
||||
payload = json.load(handle)
|
||||
self.assertEqual(
|
||||
sorted(payload["mcpServers"]),
|
||||
["gitea-author", "gitea-merger", "gitea-reviewer"],
|
||||
)
|
||||
self.assertTrue(prepared["shared_single_trusted_id"], prepared)
|
||||
self.assertIn(path, prepared["argv"])
|
||||
|
||||
def test_two_prepared_launches_use_distinct_files_and_ids(self):
|
||||
first = launcher.prepare_application_launch(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
now=NOW,
|
||||
)
|
||||
second = launcher.prepare_application_launch(
|
||||
PROJECT_PROFILES,
|
||||
client_type="claude_code",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
now=NOW,
|
||||
)
|
||||
for prepared in (first, second):
|
||||
path = prepared["launch_config_path"]
|
||||
self.addCleanup(lambda p=path: os.path.exists(p) and os.unlink(p))
|
||||
self.assertNotEqual(
|
||||
first["launch_config_path"], second["launch_config_path"]
|
||||
)
|
||||
self.assertNotEqual(
|
||||
first["client_instance_id"], second["client_instance_id"]
|
||||
)
|
||||
|
||||
def test_unsupported_client_writes_no_config_file(self):
|
||||
"""Fail before creating a file, so a bad invocation leaves no residue."""
|
||||
with self.assertRaises(ValueError):
|
||||
launcher.prepare_application_launch(
|
||||
PROJECT_PROFILES,
|
||||
client_type="not_a_client",
|
||||
namespaces=PROJECT_NAMESPACES,
|
||||
now=NOW,
|
||||
)
|
||||
|
||||
|
||||
class CliTests(unittest.TestCase):
|
||||
"""The runnable entry point itself."""
|
||||
|
||||
def _run_dry(self, argv):
|
||||
import contextlib
|
||||
import io
|
||||
|
||||
buffer = io.StringIO()
|
||||
with contextlib.redirect_stdout(buffer):
|
||||
code = launcher.main(argv)
|
||||
return code, buffer.getvalue()
|
||||
|
||||
def test_dry_run_project_scoped_plan(self):
|
||||
code, out = self._run_dry(
|
||||
[
|
||||
"--client",
|
||||
"claude_code",
|
||||
"--namespaces",
|
||||
"author,reviewer,merger",
|
||||
"--profile",
|
||||
"author=prgs-author",
|
||||
"--profile",
|
||||
"reviewer=prgs-reviewer",
|
||||
"--profile",
|
||||
"merger=prgs-merger",
|
||||
"--dry-run",
|
||||
]
|
||||
)
|
||||
self.assertEqual(code, 0)
|
||||
plan = json.loads(out)
|
||||
path = plan["launch_config_path"]
|
||||
self.addCleanup(lambda: os.path.exists(path) and os.unlink(path))
|
||||
self.assertEqual(plan["namespaces"], ["author", "reviewer", "merger"])
|
||||
self.assertEqual(
|
||||
sorted(plan["excluded_namespaces"]), ["controller", "reconciler"]
|
||||
)
|
||||
self.assertTrue(plan["shared_single_trusted_id"])
|
||||
self.assertTrue(
|
||||
fleet.assess_instance_identity(plan["client_instance_id"])["trusted"]
|
||||
)
|
||||
self.assertEqual(plan["argv"][0], "claude")
|
||||
|
||||
def test_cli_refuses_profile_for_unlaunched_namespace(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
self._run_dry(
|
||||
[
|
||||
"--namespaces",
|
||||
"author,reviewer,merger",
|
||||
"--profile",
|
||||
"author=prgs-author",
|
||||
"--profile",
|
||||
"reviewer=prgs-reviewer",
|
||||
"--profile",
|
||||
"merger=prgs-merger",
|
||||
"--profile",
|
||||
"controller=prgs-controller",
|
||||
"--dry-run",
|
||||
]
|
||||
)
|
||||
|
||||
def test_cli_refuses_unsanctioned_namespace(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
self._run_dry(
|
||||
[
|
||||
"--namespaces",
|
||||
"author,admin",
|
||||
"--profile",
|
||||
"author=prgs-author",
|
||||
"--dry-run",
|
||||
]
|
||||
)
|
||||
|
||||
def test_cli_refuses_missing_profile(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
self._run_dry(
|
||||
[
|
||||
"--namespaces",
|
||||
"author,reviewer",
|
||||
"--profile",
|
||||
"author=prgs-author",
|
||||
"--dry-run",
|
||||
]
|
||||
)
|
||||
|
||||
def test_cli_refuses_malformed_profile_pair(self):
|
||||
with self.assertRaises(SystemExit):
|
||||
self._run_dry(
|
||||
["--namespaces", "author", "--profile", "prgs-author", "--dry-run"]
|
||||
)
|
||||
|
||||
def test_two_cli_runs_receive_distinct_ids(self):
|
||||
args = [
|
||||
"--namespaces",
|
||||
"author,reviewer,merger",
|
||||
"--profile",
|
||||
"author=prgs-author",
|
||||
"--profile",
|
||||
"reviewer=prgs-reviewer",
|
||||
"--profile",
|
||||
"merger=prgs-merger",
|
||||
"--dry-run",
|
||||
]
|
||||
_, first = self._run_dry(args)
|
||||
_, second = self._run_dry(args)
|
||||
a = json.loads(first)
|
||||
b = json.loads(second)
|
||||
for plan in (a, b):
|
||||
path = plan["launch_config_path"]
|
||||
self.addCleanup(lambda p=path: os.path.exists(p) and os.unlink(p))
|
||||
self.assertNotEqual(a["client_instance_id"], b["client_instance_id"])
|
||||
|
||||
|
||||
if __name__ == "__main__": # pragma: no cover
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user