From 5a3941a286b8eb5152e224eb8699553910c460d4 Mon Sep 17 00:00:00 2001 From: Public Source Publisher Date: Sat, 10 Oct 2026 04:20:16 +0000 Subject: [PATCH] Publish curated Unraid-AppOps-Gateway source from 1a5159c66078983299a1c2acb8837e4b5a8a607a --- PUBLIC_SOURCE_EXPORT.md | 2 +- PUBLIC_SOURCE_MANIFEST.json | 27 +- backend/app/services/action_execution.py | 10 +- backend/app/services/deployment_broker.py | 11 +- .../test_controller_hub_release_volume.py | 45 +++ .../tests/test_runner_resource_envelope.py | 291 +++++++++++++++++- ...unner_specialization_controller_surface.py | 3 +- .../test_runner_specialization_transport.py | 77 +++++ 8 files changed, 443 insertions(+), 23 deletions(-) create mode 100644 backend/tests/test_runner_specialization_transport.py diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index 8bf75d6..b356317 100644 --- a/PUBLIC_SOURCE_EXPORT.md +++ b/PUBLIC_SOURCE_EXPORT.md @@ -1,3 +1,3 @@ # Public source export -This source snapshot was generated from the private canonical repository at revision `5f1a0fc2fdd07082637995b2c8b4514a1ecbbbb8`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. +This source snapshot was generated from the private canonical repository at revision `1a5159c66078983299a1c2acb8837e4b5a8a607a`. It contains no private Git history or operational evidence. Changes are published from the canonical repository. diff --git a/PUBLIC_SOURCE_MANIFEST.json b/PUBLIC_SOURCE_MANIFEST.json index 0988db8..cb51218 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "5f1a0fc2fdd07082637995b2c8b4514a1ecbbbb8", + "sourceRevision": "1a5159c66078983299a1c2acb8837e4b5a8a607a", "files": [ { "path": ".dockerignore", @@ -474,8 +474,8 @@ }, { "path": "backend/app/services/action_execution.py", - "sha256": "31e54c9a8fec9d79ad95045679ed3ace944c44e1c123921088c2d3b709d8de8d", - "bytes": 106444 + "sha256": "6db35539e942183fb4f4af319c8ea8f33f908aa0a6d91543f352f766013e2f7c", + "bytes": 106523 }, { "path": "backend/app/services/action_registry.py", @@ -539,8 +539,8 @@ }, { "path": "backend/app/services/deployment_broker.py", - "sha256": "7eaf302c579a3084e1084415236f16291cbeea1bbc80c7f78f7995dc5e22d4c6", - "bytes": 46123 + "sha256": "9d01d9e3d3b3218824ee6ef3ad561f3805f8f7b6493cfd2cebf62111b39395e0", + "bytes": 46623 }, { "path": "backend/app/services/deployment_control_plane.py", @@ -999,8 +999,8 @@ }, { "path": "backend/tests/test_controller_hub_release_volume.py", - "sha256": "abdc5eab49e59dc62fedc925bbd93fdffe907df7522e6a3c538fb2711906984b", - "bytes": 167055 + "sha256": "f9a2c497c213eaab2002eb56b99b720cf373e6af610080066b3ae330d868361d", + "bytes": 168861 }, { "path": "backend/tests/test_controller_root_procedure_contract.py", @@ -1659,8 +1659,8 @@ }, { "path": "backend/tests/test_runner_resource_envelope.py", - "sha256": "42333a8bc138319e98e9756f2b3909d0c55dac1efa348cd1d6bd7c918098ae73", - "bytes": 53395 + "sha256": "113b95a0316ae0a519c57a9032cff267376fd069c6b439f559399753e2609f75", + "bytes": 66785 }, { "path": "backend/tests/test_runner_scheduler_liveness_recovery.py", @@ -1674,8 +1674,13 @@ }, { "path": "backend/tests/test_runner_specialization_controller_surface.py", - "sha256": "1885820b4f4306c863c104eb1a46409a93358acfccb95db2728386abbb78b369", - "bytes": 13370 + "sha256": "edec2512f9e25195a12d86d3bf9d8578ea56790f2718c848bd972e826e5b4e35", + "bytes": 13473 + }, + { + "path": "backend/tests/test_runner_specialization_transport.py", + "sha256": "e40860e2ce7ca3ddd2bdb4742a0fb8f1188032ce912aeab3ebbd3b581f69de8a", + "bytes": 3644 }, { "path": "backend/tests/test_runner_statistics_binding.py", diff --git a/backend/app/services/action_execution.py b/backend/app/services/action_execution.py index f5474bd..c1cbb18 100644 --- a/backend/app/services/action_execution.py +++ b/backend/app/services/action_execution.py @@ -1797,7 +1797,7 @@ def _execute_runner_fleet_upgrade_v2( ) ) except Exception as exc: - result = {"status": "unknown" if acceptance == "quiesce" else "error", "effectsPerformed": None} + result = {"status": "unknown" if acceptance == "quiesce" or specialized else "error", "effectsPerformed": None} status = result["status"] success = False message = f"{type(exc).__name__}: fixed controller operation failed" @@ -1818,7 +1818,7 @@ def _execute_runner_fleet_upgrade_v2( risk=risk, policy_decision="allowed", reason=reason, - result="success" if success else "unknown" if acceptance and status in {"unknown", "in_progress", "partial"} else "failed", + result="success" if success else "unknown" if (acceptance or specialized) and status in {"unknown", "in_progress", "partial"} else "failed", execution_plan={"steps": steps, "jobId": completed.id}, after_state={ "status": status, @@ -1837,7 +1837,7 @@ def _execute_runner_fleet_upgrade_v2( return { "success": success, "status": "success" if success else status, - "result": "success" if success else "unknown" if acceptance and status in {"unknown", "in_progress", "partial"} else "failed", + "result": "success" if success else "unknown" if (acceptance or specialized) and status in {"unknown", "in_progress", "partial"} else "failed", "message": message, "jobId": completed.id, "auditId": audit.id, @@ -1849,8 +1849,8 @@ def _execute_runner_fleet_upgrade_v2( "projectbrainDeliveryRequired": result.get("projectbrainDeliveryRequired"), "blockers": result.get("blockers") if acceptance else None, "preservedContainers": result.get("preservedContainers") if acceptance else None, - "effectsPerformed": result.get("effectsPerformed") if acceptance else None, - "noMutation": result.get("noMutation") if acceptance else None, + "effectsPerformed": result.get("effectsPerformed") if acceptance or specialized else None, + "noMutation": result.get("noMutation") if acceptance or specialized else None, "maintenancePhase": result.get("maintenancePhase") if acceptance else None, } diff --git a/backend/app/services/deployment_broker.py b/backend/app/services/deployment_broker.py index 8d5fa74..ac2c156 100644 --- a/backend/app/services/deployment_broker.py +++ b/backend/app/services/deployment_broker.py @@ -979,10 +979,19 @@ class DeploymentBrokerClient: def runner_fleet_specialize(self, *, reason: str) -> dict[str, Any]: """Request the fixed protected-source CI specialization migration.""" - return self._request( + result = self._request( "POST", "/v1/appops/runner-fleet/specialize", secret=settings.mcp_token or "", payload={"reason": reason[:1000]}, + request_timeout=httpx.Timeout(connect=5, write=5, pool=5, read=350), ) + # The controller can outlive a lost HTTP response. Never translate a + # transport failure into proof of no host effects or permission to retry. + if result.get("status") == "unreachable": + return { + "status": "unknown", "reason": "runner_specialization_outcome_unknown", + "noMutation": False, "effectsPerformed": None, + } + return result def runner_statistics_configure(self, *, reason: str) -> dict[str, Any]: """Provision the fixed independent read-only ProjectBrain capability.""" diff --git a/backend/tests/test_controller_hub_release_volume.py b/backend/tests/test_controller_hub_release_volume.py index 3a2d91a..e727fea 100644 --- a/backend/tests/test_controller_hub_release_volume.py +++ b/backend/tests/test_controller_hub_release_volume.py @@ -10,6 +10,7 @@ import stat import subprocess import sys import tarfile +import threading import types from pathlib import Path @@ -22,6 +23,50 @@ spec = importlib.util.spec_from_file_location( module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) REVISION = "a" * 40 + + +@pytest.mark.skipif(os.name != "posix", reason="requires real POSIX child/lock contention") +def test_native_probe_cleanup_child_can_wait_beyond_default_command_deadline(monkeypatch, tmp_path): + import fcntl + + real_run = subprocess.run + observed = [] + + def scaled_run(arguments, **options): + observed.append(options["timeout"]) + # Keep the real deadline ratio, with a real child and kernel flock; + # the original 45-second budget expires before this owner releases. + options["timeout"] = options["timeout"] / 200 + return real_run(arguments, **options) + + monkeypatch.setattr(module.subprocess, "run", scaled_run) + path = tmp_path / "probe.lock" + with path.open("w") as owner: + fcntl.flock(owner.fileno(), fcntl.LOCK_EX) + release = threading.Timer(0.4, lambda: fcntl.flock(owner.fileno(), fcntl.LOCK_UN)) + release.start() + try: + output = module._run([ + sys.executable, "-c", + ("import fcntl,sys; f=open(sys.argv[1],'a'); " + "fcntl.flock(f.fileno(),fcntl.LOCK_EX); print('released')"), + str(path), + ], max_output=4096, diagnostic_action="probe_cleanup") + finally: + release.join(timeout=2) + assert output == b"released\n" + assert observed == [300] + + +@pytest.mark.parametrize("action", [None, "writer_copy", "compose_stop"]) +def test_unrelated_promotion_commands_keep_existing_deadline(monkeypatch, action): + observed = [] + def run(*_, **options): + observed.append(options["timeout"]) + return subprocess.CompletedProcess([], 0, b"", b"") + monkeypatch.setattr(module.subprocess, "run", run) + module._run(["fixed-command"], diagnostic_action=action) + assert observed == [45] OLD_REVISION = "c" * 40 IMAGE_ID = "sha256:" + "b" * 64 ORIGINAL_COMPOSE = b"name: gitea-deploy-runners\nservices:\n deploy-control:\n image: immutable-controller\n" diff --git a/backend/tests/test_runner_resource_envelope.py b/backend/tests/test_runner_resource_envelope.py index 9e55535..dc89618 100644 --- a/backend/tests/test_runner_resource_envelope.py +++ b/backend/tests/test_runner_resource_envelope.py @@ -1,11 +1,14 @@ from __future__ import annotations +import concurrent.futures import copy import importlib.util import json import os import subprocess import sys +import threading +from contextlib import contextmanager from pathlib import Path from types import SimpleNamespace @@ -15,7 +18,7 @@ ROOT = Path(__file__).resolve().parents[2] ENVELOPE_PATH = ROOT / "ops" / "gitea-deploy-broker" / "runner_resource_envelope.py" -def _module(): +def _module(*, real_lock=False): broker_dir = str(ENVELOPE_PATH.parent) if broker_dir not in sys.path: sys.path.insert(0, broker_dir) @@ -24,9 +27,281 @@ def _module(): module = importlib.util.module_from_spec(spec) sys.modules[spec.name] = module spec.loader.exec_module(module) + if not real_lock: + # Existing Docker fixtures exercise identity/cleanup independently of + # the protected volume. Native lifecycle tests below retain the lock. + @contextmanager + def fixture_lock(**_kwargs): + yield lambda: None + module._host_probe_lifecycle_lock = fixture_lock return module +class _ProbeLockStore: + """Two independently opened fixture views, with real descriptor/inode checks.""" + + def __init__(self, logical: Path, host: Path | None = None): + self.paths = (logical, host or logical) + self.logical = SimpleNamespace(fd=os.open(logical, os.O_RDONLY | os.O_DIRECTORY)) + self.host = SimpleNamespace(fd=os.open(host or logical, os.O_RDONLY | os.O_DIRECTORY)) + + def assert_current(self): + identities = [] + for view, path in zip((self.logical, self.host), self.paths, strict=True): + opened, current = os.fstat(view.fd), path.stat(follow_symlinks=False) + assert (opened.st_dev, opened.st_ino) == (current.st_dev, current.st_ino) + identities.append((opened.st_dev, opened.st_ino)) + if identities[0] != identities[1]: + raise RuntimeError("private-canary: mismatched volume views") + + def __enter__(self): + return self + + def __exit__(self, *_args): + os.close(self.host.fd) + os.close(self.logical.fd) + + +@pytest.mark.parametrize("mutation", [None, "remote", "options", "labels", "root", "malformed"]) +@pytest.mark.parametrize("native_cleanup", [False, True]) +def test_probe_lock_store_uses_only_attested_fixed_local_volume(monkeypatch, mutation, native_cleanup): + envelope = _module(real_lock=True) + import mcp_hub_release_store as store + document = [{"Name": store.VOLUME_NAME, "Driver": "local", "Scope": "local", + "Mountpoint": store.VOLUME_HOST_ROOT.as_posix(), "Options": {}, + "Labels": dict(store.EXPECTED_LABELS)}] + root = store.DOCKER_ROOT.as_posix() + if mutation == "remote": + document[0]["Driver"] = "nfs" + elif mutation == "options": + document[0]["Options"] = {"device": "private-canary"} + elif mutation == "labels": + document[0]["Labels"] = {} + elif mutation == "root": + root = "/srv/example/private-canary" + calls = [] + def run(args, **kwargs): + calls.append(args) + assert kwargs == {"timeout": 15} + if args == ["docker", "volume", "inspect", store.VOLUME_NAME]: + return "private-canary" if mutation == "malformed" else json.dumps(document) + assert args == ["docker", "info", "--format", "{{.DockerRootDir}}"] + return root + sentinel = object() + def verify(metadata, docker_root): + store.validate_volume_metadata(metadata, docker_root) + return sentinel + monkeypatch.setattr(envelope, "_run", run) + monkeypatch.setattr(envelope, "open_fixed_release_store", verify) + class PinnedFixture: + @staticmethod + def open(path): + assert path == store.VOLUME_HOST_ROOT + return sentinel + monkeypatch.setattr(envelope, "PinnedDirectory", PinnedFixture) + if mutation is None: + result = envelope._open_host_probe_lock_store(native_cleanup=native_cleanup) + if native_cleanup: + assert result.host is result.logical is sentinel + else: + assert result is sentinel + assert len(calls) == 2 + else: + with pytest.raises(RuntimeError, match="^resource_cgroup_probe_lock_store_untrusted$") as error: + envelope._open_host_probe_lock_store(native_cleanup=native_cleanup) + assert error.value.__cause__ is None + + +def test_native_cleanup_store_rechecks_only_fixed_physical_volume_and_closes_once(): + envelope = _module(real_lock=True) + calls = [] + directory = SimpleNamespace(assert_current=lambda path: calls.append(path), close=lambda: calls.append("close")) + with envelope._NativeProbeLockStore(directory) as store: + assert store.host is store.logical is directory + store.assert_current() + assert calls == [envelope.VOLUME_HOST_ROOT, "close"] + + +@pytest.fixture +def native_probe_lock(tmp_path, monkeypatch): + if os.name != "posix" or os.geteuid() != 0: + pytest.skip("real protected flock contract requires root/POSIX") + envelope = _module(real_lock=True) + root = tmp_path / "private-lock-root" + root.mkdir(mode=0o700) + monkeypatch.setattr(envelope, "_open_host_probe_lock_store", lambda **_kwargs: _ProbeLockStore(root)) + return envelope, root + + +def test_probe_lock_is_private_stable_and_released_after_failure(native_probe_lock): + envelope, root = native_probe_lock + with pytest.raises(ValueError, match="fixture failure"), envelope._host_probe_lifecycle_lock() as guard: + guard() + first = (root / envelope._HOST_PROBE_LOCK_NAME).stat() + assert first.st_mode & 0o777 == 0o600 + raise ValueError("fixture failure") + with envelope._host_probe_lifecycle_lock() as guard: + guard() + second = (root / envelope._HOST_PROBE_LOCK_NAME).stat() + assert (second.st_dev, second.st_ino) == (first.st_dev, first.st_ino) + + +@pytest.mark.parametrize("mutation", ["mode", "owner", "symlink", "hardlink", "nonempty", "directory"]) +def test_probe_lock_refuses_untrusted_leaf_without_adopting(native_probe_lock, mutation): + envelope, root = native_probe_lock + path = root / envelope._HOST_PROBE_LOCK_NAME + if mutation == "directory": + path.mkdir(mode=0o700) + elif mutation == "symlink": + target = root / "foreign" + target.write_bytes(b"") + target.chmod(0o600) + path.symlink_to(target) + else: + path.write_bytes(b"private-canary" if mutation == "nonempty" else b"") + path.chmod(0o644 if mutation == "mode" else 0o600) + if mutation == "owner": + os.chown(path, 1000, 1000) + elif mutation == "hardlink": + os.link(path, root / "foreign-link") + with pytest.raises(RuntimeError, match="^resource_cgroup_probe_lock_untrusted$"), envelope._host_probe_lifecycle_lock(): + pytest.fail("unsafe lock adopted") + + +def test_probe_lock_refuses_replacement_and_different_volume_alias(native_probe_lock, monkeypatch, tmp_path): + envelope, root = native_probe_lock + with envelope._host_probe_lifecycle_lock() as guard: + path = root / envelope._HOST_PROBE_LOCK_NAME + path.rename(root / "original-owned-lock") + path.touch(mode=0o600) + with pytest.raises(RuntimeError, match="^resource_cgroup_probe_lock_untrusted$"): + guard() + other = tmp_path / "foreign-volume" + other.mkdir(mode=0o700) + monkeypatch.setattr(envelope, "_open_host_probe_lock_store", lambda **_kwargs: _ProbeLockStore(root, other)) + with pytest.raises(RuntimeError, match="^resource_cgroup_probe_lock_store_untrusted$"), envelope._host_probe_lifecycle_lock(): + pytest.fail("unrelated aliases accepted") + + +@pytest.mark.parametrize("wait", [0.05, 3]) +def test_probe_lock_excludes_independent_process_through_independent_view(native_probe_lock, wait): + envelope, root = native_probe_lock + # Independent interpreter and opens: no inherited Python lock or lock FD. + child_source = ( + "import importlib.util,json,os,sys\n" + "from pathlib import Path\n" + "from types import SimpleNamespace\n" + "sys.path.insert(0,sys.argv[1])\n" + "spec=importlib.util.spec_from_file_location('probe_lock_child',Path(sys.argv[1])/'runner_resource_envelope.py')\n" + "m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)\n" + "class Store:\n" + " def __enter__(self):\n" + " self.logical=SimpleNamespace(fd=os.open(sys.argv[2],os.O_RDONLY|os.O_DIRECTORY))\n" + " self.host=SimpleNamespace(fd=os.open(sys.argv[2],os.O_RDONLY|os.O_DIRECTORY));return self\n" + " def assert_current(self):\n" + " assert os.fstat(self.logical.fd).st_ino==os.fstat(self.host.fd).st_ino\n" + " def __exit__(self,*args):os.close(self.logical.fd);os.close(self.host.fd)\n" + "m._open_host_probe_lock_store=lambda **kw:Store();m._HOST_PROBE_LOCK_WAIT_SECONDS=float(sys.argv[3])\n" + "print('started',flush=True)\n" + "try:\n" + " with m._host_probe_lifecycle_lock(native_cleanup=True) as guard:guard();print('acquired',flush=True)\n" + "except RuntimeError as error:print(str(error),flush=True)\n" + ) + with envelope._host_probe_lifecycle_lock(): + child = subprocess.Popen([sys.executable, "-c", child_source, str(ENVELOPE_PATH.parent), str(root), str(wait)], + stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) + try: + assert child.stdout.readline().strip() == "started" + if wait < 1: + output, error = child.communicate(timeout=3) + assert output.strip() == "resource_cgroup_probe_lock_busy" + assert error == "" and child.returncode == 0 + else: + assert child.poll() is None + finally: + if wait < 1 and child.poll() is None: + child.kill() + child.communicate() + if wait >= 1: + output, error = child.communicate(timeout=5) + assert output.strip() == "acquired" + assert error == "" and child.returncode == 0 + + +def test_concurrent_probe_lifecycles_both_complete_without_busy_or_stolen_cleanup(native_probe_lock, monkeypatch): + envelope, _root = native_probe_lock + probe, image_id, code_source = _host_probe_fixture(envelope) + started, release = threading.Event(), threading.Event() + state = {"exists": False, "starts": 0, "removals": 0} + monkeypatch.setattr(envelope, "_controller_runtime_identity", lambda: (image_id, code_source)) + monkeypatch.setattr(envelope, "_host_probe_image_labels", lambda _: {}) + monkeypatch.setattr(envelope, "_inspect_host_probe_by_name", lambda: copy.deepcopy(probe) if state["exists"] else None) + + def run(args, **_kwargs): + if args[:2] == ["docker", "create"]: + assert not state["exists"] + state["exists"] = True + return probe["Id"] + if args[:2] == ["docker", "inspect"]: + assert state["exists"] + return json.dumps([probe]) + assert args == ["docker", "rm", probe["Id"]] + assert state["exists"] and probe["State"]["Running"] is False + state["exists"] = False + state["removals"] += 1 + return "" + + def start(args, **_kwargs): + assert args == ["docker", "start", "--attach", "--interactive", probe["Id"]] + state["starts"] += 1 + probe["State"]["Running"] = True + if state["starts"] == 1: + started.set() + assert release.wait(3) + probe["State"]["Running"] = False + return SimpleNamespace(returncode=0, stdout=b'{"ready":true,"result":{"verified":true}}') + + monkeypatch.setattr(envelope, "_run", run) + monkeypatch.setattr(envelope.subprocess, "run", start) + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(envelope._run_host_probe, "snapshot") + assert started.wait(3) + second = pool.submit(envelope._run_host_probe, "snapshot") + try: + assert state["starts"] == 1 and not second.done() + finally: + release.set() + assert first.result(timeout=5) == second.result(timeout=5) == {"verified": True} + assert state["removals"] == 2 and state["exists"] is False + + +def test_normal_probe_and_promotion_cleanup_share_same_complete_lock(monkeypatch): + envelope = _module(real_lock=True) + stages = [] + @contextmanager + def lock(**kwargs): + assert kwargs in ({}, {"native_cleanup": True}) + stages.append("acquire") + yield lambda: stages.append("guard") + stages.append("release") + monkeypatch.setattr(envelope, "_host_probe_lifecycle_lock", lock) + def normal(mode, payload, guard): + assert mode == "family" and payload is None + guard() + stages.append("normal") + return {"verified": True} + def cleanup(**kwargs): + kwargs["assert_lock"]() + stages.append("cleanup") + return True + monkeypatch.setattr(envelope, "_run_host_probe_locked", normal) + monkeypatch.setattr(envelope, "_cleanup_stopped_host_probe_locked", cleanup) + assert envelope.inspect_host_cgroup_family() == {"verified": True} + assert envelope.cleanup_stopped_host_probe(image_id="sha256:" + "a" * 64, + code_source=f"{envelope._HOST_VOLUME_ROOT}/controller-artifacts/{'b' * 40}") + assert stages == ["acquire", "guard", "normal", "release", "acquire", "guard", "cleanup", "release"] + + def _group(path: str, cpu_max: str, memory_max: int) -> dict[str, object]: return { "path": path, @@ -747,9 +1022,17 @@ def test_real_probe_cleanup_cli_exposes_only_finite_child_failure(tmp_path, fail "elif a==['rm',v['probe']['Id']]: print(v['failure'],file=sys.stderr); sys.exit(1)\n" + "else: sys.exit(99)\n") shim.chmod(0o700) - result = subprocess.run([sys.executable, str(ENVELOPE_PATH), "--cleanup-stopped-host-probe", - image_id, code_source], capture_output=True, timeout=15, - env={**os.environ, "PATH": str(tmp_path) + os.pathsep + os.environ.get("PATH", "")}) + # Keep the actual CLI branch/diagnostics, isolating only its volume lock. + entry = tmp_path / "cleanup-entry.py" + source = ENVELOPE_PATH.read_text(encoding="utf-8") + source = source.replace('if __name__ == "__main__":', + '@contextmanager\ndef _host_probe_lifecycle_lock(**_kwargs):\n' + ' yield lambda: None\n\nif __name__ == "__main__":') + entry.write_text(source, encoding="utf-8") + result = subprocess.run([sys.executable, str(entry), "--cleanup-stopped-host-probe", + image_id, code_source], capture_output=True, timeout=15, check=False, + env={**os.environ, "PATH": str(tmp_path) + os.pathsep + os.environ.get("PATH", ""), + "PYTHONPATH": str(ENVELOPE_PATH.parent)}) assert result.returncode == 1 assert result.stdout == b"" assert result.stderr == ("resource_cgroup_probe_cleanup_" + condition + "\n").encode() diff --git a/backend/tests/test_runner_specialization_controller_surface.py b/backend/tests/test_runner_specialization_controller_surface.py index e6c3155..b0a31f5 100644 --- a/backend/tests/test_runner_specialization_controller_surface.py +++ b/backend/tests/test_runner_specialization_controller_surface.py @@ -93,7 +93,7 @@ def test_pending_protected_validation_blocks_specialization(controller, monkeypa ("runner_fleet_specialization_entrypoint_untrusted\nprivate-token-value", "runner_fleet_specialization_script_execution_failed"), ]) -def test_specialization_failure_response_contains_only_finite_safe_reason(controller, monkeypatch, private, expected): +def test_specialization_failure_response_contains_only_finite_safe_reason(controller, monkeypatch, capsys, private, expected): instance, responses, _ = handler(controller, monkeypatch) monkeypatch.setattr(controller.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0, stdout=b"exact source script")) @@ -103,6 +103,7 @@ def test_specialization_failure_response_contains_only_finite_safe_reason(contro monkeypatch.setattr(controller, "execute_runner_fleet_specialize", fail) instance.do_POST() + assert capsys.readouterr().out == "runner_fleet_specialization_failure:" + expected + "\n" assert responses == [(409, {"status": "upgrade_failed", "revision": "a" * 40, "reason": expected})] diff --git a/backend/tests/test_runner_specialization_transport.py b/backend/tests/test_runner_specialization_transport.py new file mode 100644 index 0000000..b0ec4ad --- /dev/null +++ b/backend/tests/test_runner_specialization_transport.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import httpx +import pytest +from sqlmodel import Session, SQLModel, create_engine + +from backend.app.db.models import AuditEvent +from backend.app.services import action_execution, deployment_broker + + +@pytest.mark.parametrize("outcome", ["success", "refused", "timeout"]) +def test_actual_signed_specialization_transport_preserves_outcome(monkeypatch, outcome): + client = deployment_broker.DeploymentBrokerClient() + client.base_url = "https://controller.invalid" + monkeypatch.setattr(client, "is_configured", lambda: True) + monkeypatch.setattr(client, "_verify_option", lambda: True) + monkeypatch.setattr(deployment_broker.settings, "mcp_token", "x" * 64) + observed = [] + real_client = httpx.Client + + def respond(request): + assert request.method == "POST" + assert request.url.path == "/v1/appops/runner-fleet/specialize" + assert len(request.headers["X-ITWorx-Deployment-Signature"]) == 64 + if outcome == "timeout": + raise httpx.ReadTimeout("Bearer SECRET-CANARY", request=request) + return httpx.Response(200 if outcome == "success" else 409, json={ + "status": "specialized" if outcome == "success" else "upgrade_failed", + "reason": "runner_fleet_specialization_cgroup_inventory_unavailable", + }) + + def transport_client(**options): + observed.append(options) + return real_client(**options, transport=httpx.MockTransport(respond)) + + monkeypatch.setattr(deployment_broker.httpx, "Client", transport_client) + result = client.runner_fleet_specialize(reason="qualify exact source") + assert len(observed) == 1 # No automatic replay after transport loss/refusal. + assert observed[0]["verify"] is True + assert observed[0]["timeout"].as_dict() == { + "connect": 5, "read": 350, "write": 5, "pool": 5, + } + assert "SECRET-CANARY" not in str(result) + if outcome == "timeout": + assert result == { + "status": "unknown", "reason": "runner_specialization_outcome_unknown", + "noMutation": False, "effectsPerformed": None, + } + else: + assert result["status"] == ("specialized" if outcome == "success" else "upgrade_failed") + assert result["reason"] == "runner_fleet_specialization_cgroup_inventory_unavailable" + + +@pytest.mark.parametrize("raised", [False, True]) +def test_specialization_unknown_is_preserved_in_job_result_and_audit(tmp_path, monkeypatch, raised): + def request(*_, **__): + if raised: + raise RuntimeError("Bearer SECRET-CANARY") + return {"status": "unknown", "reason": "runner_specialization_outcome_unknown", + "noMutation": False, "effectsPerformed": None} + + monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_fleet_specialize", request) + engine = create_engine(f"sqlite:///{tmp_path / 'unknown.db'}") + SQLModel.metadata.create_all(engine) + with Session(engine) as session: + result = action_execution._execute_runner_fleet_upgrade_v2( + session, source="chatgpt-mcp", risk="high", reason="qualify exact source", + original_policy_decision="allowed", specialized=True, + ) + assert result["success"] is False + assert result["controllerStatus"] == "unknown" and result["result"] == "unknown" + assert result["effectsPerformed"] is None + if not raised: + assert result["noMutation"] is False + audit = session.get(AuditEvent, result["auditId"]) + assert audit is not None and audit.result == "unknown" + assert "SECRET-CANARY" not in str(result) + str(audit)