"""#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()