From ab3fa553f0935906d3d05b0ae4511af53a056673 Mon Sep 17 00:00:00 2001 From: Public Source Publisher Date: Wed, 7 Oct 2026 23:44:08 +0000 Subject: [PATCH] Publish curated Unraid-AppOps-Gateway source from bc001a6df859f7c9836fa145afb739d1ceebc524 --- PUBLIC_SOURCE_EXPORT.md | 2 +- PUBLIC_SOURCE_MANIFEST.json | 56 +- backend/app/connectors/gitea_actions.py | 470 ++++-- backend/app/mcp_server.py | 33 + backend/app/services/deployment_broker.py | 22 + ...tifact_bootstrap_created_child_recovery.py | 627 ++++++++ backend/tests/test_deploy_broker.py | 12 +- backend/tests/test_gitea_actions.py | 4 + backend/tests/test_hub_candidate_import.py | 90 ++ ...test_projectbrain_delivery_revalidation.py | 1265 +++++++++++++++++ .../test_runner_fleet_scheduler_recovery.py | 779 +++++++++- backend/tests/test_runner_fleet_specialize.py | 101 +- ...test_runner_scheduler_liveness_recovery.py | 431 +++++- src/pages/Overview.tsx | 11 +- src/types/appops.ts | 11 + 15 files changed, 3687 insertions(+), 227 deletions(-) create mode 100644 backend/tests/test_controller_artifact_bootstrap_created_child_recovery.py create mode 100644 backend/tests/test_projectbrain_delivery_revalidation.py diff --git a/PUBLIC_SOURCE_EXPORT.md b/PUBLIC_SOURCE_EXPORT.md index f2d5206..90ce737 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 `2e6cf771e11a9cdc8b2e2453586a16d518adf40b`. 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 `bc001a6df859f7c9836fa145afb739d1ceebc524`. 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 f6e17bc..221fc0d 100644 --- a/PUBLIC_SOURCE_MANIFEST.json +++ b/PUBLIC_SOURCE_MANIFEST.json @@ -1,6 +1,6 @@ { "schemaVersion": 1, - "sourceRevision": "2e6cf771e11a9cdc8b2e2453586a16d518adf40b", + "sourceRevision": "bc001a6df859f7c9836fa145afb739d1ceebc524", "files": [ { "path": ".dockerignore", @@ -294,8 +294,8 @@ }, { "path": "backend/app/connectors/gitea_actions.py", - "sha256": "b367bebc60e2542c7d7763c78e4eabfa98bca5d2be753e387c8faa269fdbe330", - "bytes": 18440 + "sha256": "3c0be8c895079c007b9c04d2eaab152cc30d10ea8708b6d3914a86a3b1b35a65", + "bytes": 29286 }, { "path": "backend/app/connectors/gpu_metrics.py", @@ -404,8 +404,8 @@ }, { "path": "backend/app/mcp_server.py", - "sha256": "650edf520d95c61751422e1c775ef0e77a3c1d7340ca83c00aeb667ed64e106a", - "bytes": 46224 + "sha256": "24817f3e7566b441876b3e12c0bffe52cf596cf27d098e1f8579bce8a28c65ec", + "bytes": 47931 }, { "path": "backend/app/runner_statistics_mcp.py", @@ -539,8 +539,8 @@ }, { "path": "backend/app/services/deployment_broker.py", - "sha256": "80ee3b6ed5b724633934e00b37ddfa9bac9db7d4e12e56909eff5a31364c164a", - "bytes": 40896 + "sha256": "452e6a2c3316501a7828990cb9fe7f689edfd1fa07e68e20e612126cd387bc5f", + "bytes": 42391 }, { "path": "backend/app/services/deployment_control_plane.py", @@ -932,6 +932,11 @@ "sha256": "3389f7cc8225ff2f03b46aa79a3a70dae7a6198d1ab52fd0879638bcf4c92f83", "bytes": 80172 }, + { + "path": "backend/tests/test_controller_artifact_bootstrap_created_child_recovery.py", + "sha256": "a9613e4bad2d66cdf51a840b96674b857b6b82ef125a11345ab9fa2a61735ffd", + "bytes": 21205 + }, { "path": "backend/tests/test_controller_artifact_bootstrap_failed_child_recovery.py", "sha256": "5da80eed7ef420bee44cec4e20fe45bd4918245c6236a4481aeffe71c939d82f", @@ -989,8 +994,8 @@ }, { "path": "backend/tests/test_deploy_broker.py", - "sha256": "8d92404b6719159d1ce2e5b7df73f925ba37ce2fff6f7bbaf9bcb780c46a789c", - "bytes": 415499 + "sha256": "6b1ad652cc7f06ce94105846e15110741fb9514e833101ae670279d5a1711deb", + "bytes": 415988 }, { "path": "backend/tests/test_deployment_auto_rollback.py", @@ -1134,8 +1139,8 @@ }, { "path": "backend/tests/test_gitea_actions.py", - "sha256": "7c4bebd43553b6de6738bdcb70153640a8ec6f2e855c09eac35a40345995b1d8", - "bytes": 12826 + "sha256": "85d386abe5ba6f09971793431e00684ef5708f3c94518a44f89fafe30fa58d8a", + "bytes": 13153 }, { "path": "backend/tests/test_gitea_project_auto_registration.py", @@ -1164,8 +1169,8 @@ }, { "path": "backend/tests/test_hub_candidate_import.py", - "sha256": "7ffc9b4f10d589f092c16bd23cf2cafabac1a6a26fb1a3fb19dbbd5ccf2f04bc", - "bytes": 42194 + "sha256": "c5eb8845b6fba09cb8d49edd06b8fad1420a8a14eac0e732198fdce0b6a0a333", + "bytes": 45629 }, { "path": "backend/tests/test_hub_candidate_transport.py", @@ -1382,6 +1387,11 @@ "sha256": "ef07dc476d12084d55a0ef36a787d91365d743b0b73bcc011bc2f8c0265c42d2", "bytes": 12310 }, + { + "path": "backend/tests/test_projectbrain_delivery_revalidation.py", + "sha256": "579550ac68438da05d9472196961062f33364fe734fc8c4dfc8f6f912ab286c3", + "bytes": 50954 + }, { "path": "backend/tests/test_prowlarr_connector.py", "sha256": "da2a5b9421ced902cf75c7922f15431be0e5db06ef2e350ed5ab2d2a33f8749b", @@ -1579,13 +1589,13 @@ }, { "path": "backend/tests/test_runner_fleet_scheduler_recovery.py", - "sha256": "cc2e5c2b1076df701e4124e62fc9173b0ea3badb148e99ea5d54f84f40619568", - "bytes": 50645 + "sha256": "2b6ad687919a7c4724cf6fe00b968b7e39032df355d71be85da5b36237014f53", + "bytes": 72416 }, { "path": "backend/tests/test_runner_fleet_specialize.py", - "sha256": "3329a6ac547afa73b0804caad9001a0cf7de0755b6d355e3139cb46d550fccf5", - "bytes": 111208 + "sha256": "9de45904f4a4fd254d5e3b4c502b3d2c235f2ce220fd8f122a5ed840388b2eb1", + "bytes": 113743 }, { "path": "backend/tests/test_runner_fleet_v2_execution_surface.py", @@ -1604,8 +1614,8 @@ }, { "path": "backend/tests/test_runner_scheduler_liveness_recovery.py", - "sha256": "86e9a1e336206506f48e149fedf21e79f9c7ad54771dea4f53ec039ac81548af", - "bytes": 3824 + "sha256": "40e69e3d0161fb0932a35889ccc7c51ab42bc74c1e0b39c0f5c31a7a76bfe1c3", + "bytes": 19338 }, { "path": "backend/tests/test_runner_specialization_controller_surface.py", @@ -3094,8 +3104,8 @@ }, { "path": "src/pages/Overview.tsx", - "sha256": "35d51ee59859278cdb649f2c6561ca391b6c1feecb425b9236ce025af035f93d", - "bytes": 30125 + "sha256": "ce94af3836db8be4a66163fcc23be7901854150e465fb527a9f4e2c8267ac73e", + "bytes": 30555 }, { "path": "src/pages/Playbooks.tsx", @@ -3199,8 +3209,8 @@ }, { "path": "src/types/appops.ts", - "sha256": "cffa840d76327eadf663ee7c896a366ad56d6cf860a0e0e9687ff7b3e14e2f3b", - "bytes": 74110 + "sha256": "75e999b241d6196950fbf0b9decc82339e044fd6ced75513d462b9187a85ab24", + "bytes": 74346 }, { "path": "src/utils/toast.ts", diff --git a/backend/app/connectors/gitea_actions.py b/backend/app/connectors/gitea_actions.py index f592da9..509806d 100644 --- a/backend/app/connectors/gitea_actions.py +++ b/backend/app/connectors/gitea_actions.py @@ -20,8 +20,14 @@ _CRITICAL_AGE_SECONDS = 600 _ACTIVE_WARNING_AGE_SECONDS = 1_800 _ACTIVE_CRITICAL_AGE_SECONDS = 7_200 _QUEUE_SAMPLE_LIMIT = 50 +_RECOVERY_MIN_OBSERVATION_SECONDS = 600 +_RECOVERY_MAX_SAMPLE_GAP_SECONDS = 45 _CACHE_LOCK = Lock() _CACHE: tuple[float, dict[str, Any]] | None = None +_RECOVERY_OBSERVATIONS: dict[tuple[Any, ...], tuple[float, float]] = {} +_RECOVERY_SOURCE: str | None = None +_SNAPSHOT_GENERATION = 0 +_COMMITTED_GENERATION = 0 _DEFAULT_RUNNER_TARGETS = { "linux-validation": "itworx-validation-runner-2", "docker-validation": "itworx-docker-runner", @@ -44,6 +50,166 @@ _DEFAULT_RUNNER_TARGET_CANDIDATES = { } +def _is_count(value: Any) -> bool: + return type(value) is int and value >= 0 + + +def _is_identifier(value: Any) -> bool: + return type(value) is int and value > 0 + + +def _is_label_list(value: Any) -> bool: + def valid_label(text: Any) -> bool: + return ( + isinstance(text, str) + and len(text) <= 120 + and bool(text.strip()) + and text == text.strip() + ) + + return isinstance(value, list) and all( + valid_label(item) + or (isinstance(item, dict) and valid_label(item.get("name"))) + for item in value + ) + + +def _complete_rows(payload: Any, row_key: str) -> list[dict[str, Any]] | None: + if not isinstance(payload, dict): + return None + rows = payload.get(row_key) + total = payload.get("total_count") + if not isinstance(rows, list) or not _is_count(total) or total != len(rows): + return None + if not all(isinstance(row, dict) for row in rows): + return None + return rows + + +def _valid_runner_inventory(payload: Any) -> list[dict[str, Any]] | None: + runners = _complete_rows(payload, "runners") + if runners is None: + return None + ids: set[int] = set() + for runner in runners: + runner_id = runner.get("id") + if ( + not _is_identifier(runner_id) + or runner_id in ids + or not isinstance(runner.get("name"), str) + or not runner["name"].strip() + or not isinstance(runner.get("status"), str) + or not runner["status"].strip() + or type(runner.get("busy")) is not bool + or type(runner.get("disabled")) is not bool + or not _is_label_list(runner.get("labels")) + ): + return None + ids.add(runner_id) + return runners + + +def _valid_job_inventory( + payload: Any, expected_status: str, seen_ids: set[int] +) -> list[dict[str, Any]] | None: + jobs = _complete_rows(payload, "jobs") + if jobs is None: + return None + local_ids: set[int] = set() + for job in jobs: + job_id = job.get("id") + if ( + not _is_identifier(job_id) + or job_id in local_ids + or job_id in seen_ids + or not ( + job.get("status") == expected_status + or ( + expected_status == "pending" + and job.get("status") == "requested" + ) + ) + or not _is_label_list(job.get("labels")) + ): + return None + for field in ( + "run_id", + "run_number", + "workflow_id", + "attempt", + "run_attempt", + ): + value = job.get(field) + if value is not None and not _is_identifier(value): + return None + attempt = job.get("attempt") + run_attempt = job.get("run_attempt") + if attempt is not None and run_attempt is not None and attempt != run_attempt: + return None + if expected_status == "in_progress" and not _is_identifier( + job.get("runner_id") + ): + return None + local_ids.add(job_id) + seen_ids.update(local_ids) + return jobs + + +def _reset_recovery_state_locked(*, clear_cache: bool = True) -> None: + global _CACHE + _RECOVERY_OBSERVATIONS.clear() + if clear_cache: + _CACHE = None + + +def _observe_recovery_candidates( + generation: int, + origin: str, + candidates: list[dict[str, Any]], + sample_started: float, + sample_finished: float, + *, + complete: bool, +) -> list[dict[str, Any]] | None: + global _RECOVERY_SOURCE, _COMMITTED_GENERATION + with _CACHE_LOCK: + if generation != _SNAPSHOT_GENERATION: + return None + _COMMITTED_GENERATION = generation + if _RECOVERY_SOURCE != origin: + _RECOVERY_SOURCE = origin + _reset_recovery_state_locked(clear_cache=False) + if ( + not complete + or sample_finished < sample_started + or sample_finished - sample_started > _RECOVERY_MAX_SAMPLE_GAP_SECONDS + or len(candidates) > 500 + ): + _reset_recovery_state_locked(clear_cache=False) + return [] + + observations: dict[tuple[Any, ...], tuple[float, float]] = {} + matured: list[dict[str, Any]] = [] + for candidate in candidates: + key = candidate["key"] + previous = _RECOVERY_OBSERVATIONS.get(key) + gap = sample_finished - previous[1] if previous is not None else None + if gap is not None and 0 <= gap <= _RECOVERY_MAX_SAMPLE_GAP_SECONDS: + first_seen = previous[0] + else: + first_seen = sample_finished + observations[key] = (first_seen, sample_finished) + observed_seconds = max(0, int(sample_finished - first_seen)) + if observed_seconds >= _RECOVERY_MIN_OBSERVATION_SECONDS: + matured.append( + { + **candidate, + "observedClaimableSeconds": observed_seconds, + } + ) + _RECOVERY_OBSERVATIONS.clear() + _RECOVERY_OBSERVATIONS.update(observations) + return matured class GiteaActionsConnector(BaseConnector): """Read a bounded Gitea Actions control-plane snapshot; never execute jobs.""" @@ -171,37 +337,70 @@ class GiteaActionsConnector(BaseConnector): return candidate return None + @staticmethod + def _empty_snapshot(status: str, severity: str) -> dict[str, Any]: + return { + "status": status, + "severity": severity, + "generatedAt": datetime.now(UTC).isoformat(), + "runners": [], + "queue": { + "queued": 0, + "inProgress": 0, + "oldestQueuedAgeSeconds": None, + "oldestInProgressAgeSeconds": None, + "sampleTruncated": False, + "blockedByRunnerLabels": [], + "staleActiveRunnerLabels": [], + "staleAssignmentRunnerLabels": [], + "items": [], + }, + "recommendedAction": None, + "cache": {"hit": False, "ttlSeconds": _CACHE_SECONDS}, + } + def snapshot(self, *, refresh: bool = False) -> dict[str, Any]: - global _CACHE + global _CACHE, _RECOVERY_SOURCE, _SNAPSHOT_GENERATION, _COMMITTED_GENERATION + if not self.is_configured(): + with _CACHE_LOCK: + _SNAPSHOT_GENERATION += 1 + _COMMITTED_GENERATION = _SNAPSHOT_GENERATION + _RECOVERY_SOURCE = None + _reset_recovery_state_locked() + return self._empty_snapshot("missing_config", "unknown") + + try: + origin = self._base_url() + except RuntimeError as exc: + with _CACHE_LOCK: + _SNAPSHOT_GENERATION += 1 + _COMMITTED_GENERATION = _SNAPSHOT_GENERATION + _RECOVERY_SOURCE = None + _reset_recovery_state_locked() + result = self._empty_snapshot("error", "warning") + result["reason"] = redact_message(str(exc))[:240] + return result + now_monotonic = monotonic() with _CACHE_LOCK: + if _RECOVERY_SOURCE is not None and _RECOVERY_SOURCE != origin: + _RECOVERY_SOURCE = origin + _reset_recovery_state_locked() if ( not refresh and _CACHE is not None and now_monotonic - _CACHE[0] < _CACHE_SECONDS ): cached = copy.deepcopy(_CACHE[1]) + action = cached.get("recommendedAction") + if isinstance(action, dict) and action.get("intent") == "runner.recover": + cached["recommendedAction"] = None + cached.get("queue", {})["staleAssignmentRunnerLabels"] = [] cached["cache"] = {"hit": True, "ttlSeconds": _CACHE_SECONDS} return cached - - if not self.is_configured(): - return { - "status": "missing_config", - "severity": "unknown", - "generatedAt": datetime.now(UTC).isoformat(), - "runners": [], - "queue": { - "queued": 0, - "inProgress": 0, - "oldestQueuedAgeSeconds": None, - "oldestInProgressAgeSeconds": None, - "sampleTruncated": False, - "blockedByRunnerLabels": [], - "staleAssignmentRunnerLabels": [], - }, - "recommendedAction": None, - "cache": {"hit": False, "ttlSeconds": _CACHE_SECONDS}, - } + _SNAPSHOT_GENERATION += 1 + generation = _SNAPSHOT_GENERATION + sample_started = now_monotonic try: runner_payload = self._request( @@ -219,7 +418,7 @@ class GiteaActionsConnector(BaseConnector): ) # Keep pending inventory in the all-or-error fetch, but it is not # runnable queue pressure and must not drive runner diagnostics. - self._request( + pending_payload = self._request( "/api/v1/admin/actions/jobs", { "status": "pending", @@ -234,30 +433,55 @@ class GiteaActionsConnector(BaseConnector): {"status": "in_progress", "page": 1, "limit": 100}, ) except RuntimeError as exc: - return { - "status": "error", - "severity": "warning", - "reason": redact_message(str(exc))[:240], - "generatedAt": datetime.now(UTC).isoformat(), - "runners": [], - "queue": { - "queued": 0, - "inProgress": 0, - "oldestQueuedAgeSeconds": None, - "oldestInProgressAgeSeconds": None, - "sampleTruncated": False, - "blockedByRunnerLabels": [], - "staleAssignmentRunnerLabels": [], - }, - "recommendedAction": None, - "cache": {"hit": False, "ttlSeconds": _CACHE_SECONDS}, - } + with _CACHE_LOCK: + if generation == _SNAPSHOT_GENERATION: + _COMMITTED_GENERATION = generation + _RECOVERY_SOURCE = origin + _reset_recovery_state_locked() + result = self._empty_snapshot("error", "warning") + result["reason"] = redact_message(str(exc))[:240] + return result + + sample_finished = monotonic() + + recovery_runners = _valid_runner_inventory(runner_payload) + seen_job_ids: set[int] = set() + recovery_queued_jobs = _valid_job_inventory( + queued_payload, "queued", seen_job_ids + ) + recovery_pending_jobs = _valid_job_inventory( + pending_payload, "pending", seen_job_ids + ) + recovery_active_jobs = _valid_job_inventory( + active_payload, "in_progress", seen_job_ids + ) + recovery_inventory_complete = all( + rows is not None + for rows in ( + recovery_runners, + recovery_queued_jobs, + recovery_pending_jobs, + recovery_active_jobs, + ) + ) + runner_payload = runner_payload if isinstance(runner_payload, dict) else {} + queued_payload = queued_payload if isinstance(queued_payload, dict) else {} + active_payload = active_payload if isinstance(active_payload, dict) else {} + + def display_int(value: Any) -> int: + try: + return int(value or 0) + except (TypeError, ValueError, OverflowError): + return 0 + + def display_count(value: Any, fallback: int = 0) -> int: + return value if _is_count(value) else fallback raw_runners = runner_payload.get("runners") runner_items = raw_runners if isinstance(raw_runners, list) else [] runners = [ { - "id": int(item.get("id", 0) or 0), + "id": display_int(item.get("id", 0)), "name": str(item.get("name") or "unknown")[:160], "status": str(item.get("status") or "unknown")[:32], "busy": bool(item.get("busy")), @@ -276,11 +500,12 @@ class GiteaActionsConnector(BaseConnector): ) if isinstance(item, dict) ] - active_jobs = ( + raw_active_jobs = ( active_payload.get("jobs") if isinstance(active_payload.get("jobs"), list) else [] ) + active_jobs = [item for item in raw_active_jobs if isinstance(item, dict)] now = datetime.now(UTC) queued_times = [self._timestamp(item.get("created_at")) for item in queued_jobs] queued_times = [item for item in queued_times if item is not None] @@ -314,8 +539,12 @@ class GiteaActionsConnector(BaseConnector): if item["status"] == "online" and not item["disabled"] ] blocked_jobs = [] - stale_assignment_jobs: list[dict[str, Any]] = [] + eligible_recovery_candidates: list[dict[str, Any]] = [] queue_items = [] + targets = self._restart_targets() + active_runner_ids = { + job["runner_id"] for job in recovery_active_jobs or [] + } for job in queued_jobs: required = set(self._labels(job.get("labels"))) matching = [ @@ -337,25 +566,60 @@ class GiteaActionsConnector(BaseConnector): if all_matching_busy else "awaiting_scheduler_assignment" ) - if ( - required - and matching - and not all_matching_busy - and age is not None - and age >= _CRITICAL_AGE_SECONDS - ): - stale_assignment_jobs.append( - { - "job": job, - "required": sorted(required), - "matching": matching, - "ageSeconds": age, - } - ) + if recovery_inventory_complete and required: + for label in sorted(required): + target = self._target_for_label(label, online_runners, targets) + if not target: + continue + exact_runners = [ + runner for runner in runners if runner["name"] == target + ] + if len(exact_runners) != 1: + continue + target_runner = exact_runners[0] + target_labels = tuple(sorted(target_runner["labels"])) + if ( + not required.issubset(set(target_labels)) + or target_runner["status"] != "online" + or target_runner["disabled"] is not False + or target_runner["busy"] is not False + or target_runner["id"] in active_runner_ids + ): + continue + key = ( + origin, + job["id"], + tuple( + (field, job[field]) + for field in ( + "run_id", + "run_number", + "workflow_id", + "attempt", + "run_attempt", + ) + if job.get(field) is not None + ), + tuple(sorted(required)), + label, + target, + target_runner["id"], + target_runner["name"], + target_labels, + ) + eligible_recovery_candidates.append( + { + "key": key, + "job": job, + "required": sorted(required), + "label": label, + "target": target, + } + ) if len(queue_items) < _QUEUE_SAMPLE_LIMIT: queue_items.append( { - "id": int(job.get("id", 0) or 0), + "id": display_int(job.get("id", 0)), "name": str(job.get("name") or "job")[:160], "labels": sorted(required)[:20], "ageSeconds": age, @@ -364,6 +628,22 @@ class GiteaActionsConnector(BaseConnector): } ) + try: + current_origin = self._base_url() if self.is_configured() else None + except RuntimeError: + current_origin = None + source_matches = current_origin == origin + matured_recovery_candidates = _observe_recovery_candidates( + generation, + origin, + eligible_recovery_candidates, + sample_started, + sample_finished, + complete=recovery_inventory_complete and source_matches, + ) + if matured_recovery_candidates is None: + matured_recovery_candidates = [] + severity = "healthy" if ( (oldest_age or 0) >= _CRITICAL_AGE_SECONDS @@ -378,7 +658,6 @@ class GiteaActionsConnector(BaseConnector): ): severity = "warning" - targets = self._restart_targets() blocked_labels = [ label for job in blocked_jobs for label in self._labels(job.get("labels")) ] @@ -390,43 +669,33 @@ class GiteaActionsConnector(BaseConnector): stale_assignment_labels = sorted( { label - for candidate in stale_assignment_jobs + for candidate in matured_recovery_candidates for label in candidate["required"] } ) recommended_action: dict[str, Any] | None = None for candidate in sorted( - stale_assignment_jobs, - key=lambda item: int(item["ageSeconds"]), + matured_recovery_candidates, + key=lambda item: item["observedClaimableSeconds"], reverse=True, ): job = candidate["job"] - matching_by_name = { - str(runner.get("name") or ""): runner for runner in candidate["matching"] + recommended_action = { + "intent": "runner.recover", + "target": candidate["target"], + "requiresApproval": False, + "autoExecute": False, + "reason": "Gitea Actions scheduler left a compatible idle runner unassigned for at least ten minutes.", + "evidence": { + "jobId": display_int(job.get("id", 0)), + "label": candidate["label"], + "observedClaimableSeconds": candidate[ + "observedClaimableSeconds" + ], + }, } - for label in candidate["required"]: - target = self._target_for_label(label, candidate["matching"], targets) - if not target: - continue - target_runner = matching_by_name.get(target) - if target_runner is not None and bool(target_runner.get("busy")): - continue - recommended_action = { - "intent": "runner.recover", - "target": target, - "requiresApproval": False, - "autoExecute": True, - "reason": "Gitea Actions scheduler left a compatible idle runner unassigned for at least ten minutes.", - "evidence": { - "jobId": int(job.get("id", 0) or 0), - "label": label, - "minimumAgeSeconds": _CRITICAL_AGE_SECONDS, - }, - } - break - if recommended_action is not None: - break + break if recommended_action is None: restart_target = None @@ -465,14 +734,14 @@ class GiteaActionsConnector(BaseConnector): "offline": sum(item["status"] != "online" for item in runners), }, "queue": { - "queued": int(queued_payload.get("total_count", 0) or 0), - "inProgress": int( - active_payload.get("total_count", len(active_jobs)) or 0 + "queued": display_count(queued_payload.get("total_count", 0)), + "inProgress": display_count( + active_payload.get("total_count", len(active_jobs)) ), "oldestQueuedAgeSeconds": oldest_age, "oldestInProgressAgeSeconds": oldest_active_age, "sampleTruncated": ( - int(queued_payload.get("total_count", 0) or 0) + display_count(queued_payload.get("total_count", 0)) > len( queued_payload.get("jobs") if isinstance(queued_payload.get("jobs"), list) @@ -487,6 +756,25 @@ class GiteaActionsConnector(BaseConnector): "recommendedAction": recommended_action, "cache": {"hit": False, "ttlSeconds": _CACHE_SECONDS}, } + try: + final_origin = self._base_url() if self.is_configured() else None + except RuntimeError: + final_origin = None with _CACHE_LOCK: - _CACHE = (now_monotonic, copy.deepcopy(result)) + if final_origin != origin and generation == _SNAPSHOT_GENERATION: + _SNAPSHOT_GENERATION += 1 + _COMMITTED_GENERATION = _SNAPSHOT_GENERATION + _RECOVERY_SOURCE = final_origin + _reset_recovery_state_locked() + if ( + final_origin == origin + and generation == _SNAPSHOT_GENERATION + and _COMMITTED_GENERATION == generation + ): + _CACHE = (sample_finished, copy.deepcopy(result)) + else: + action = result.get("recommendedAction") + if isinstance(action, dict) and action.get("intent") == "runner.recover": + result["recommendedAction"] = None + result["queue"]["staleAssignmentRunnerLabels"] = [] return result diff --git a/backend/app/mcp_server.py b/backend/app/mcp_server.py index 4fdfb0a..eb8226c 100644 --- a/backend/app/mcp_server.py +++ b/backend/app/mcp_server.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib import json import os +import re from collections import defaultdict from datetime import UTC, datetime from functools import wraps @@ -759,6 +760,38 @@ def appops_registered_legacy_delivery(repository: str, revision: str, request_id })) +@appops_mcp.tool(name="appops.projectbrain_delivery.preflight", annotations=_READ_ONLY_TOOL) +@_tracked("appops.projectbrain_delivery.preflight") +def appops_projectbrain_delivery_preflight(revision: str) -> dict[str, Any]: + """Read the fixed ProjectBrain retry evidence without changing delivery or deployment state.""" + if not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision): + raise ValueError("ProjectBrain delivery revision is invalid") + return _json(DeploymentBrokerClient().projectbrain_delivery_operation( + "preflight", {"revision": revision}, + )) + + +@appops_mcp.tool(name="appops.projectbrain_delivery.revalidate", annotations=_OPERATION_RUN_TOOL) +@_tracked("appops.projectbrain_delivery.revalidate") +def appops_projectbrain_delivery_revalidate( + revision: str, expected_preflight_sha256: str, idempotency_key: str, +) -> dict[str, Any]: + """Queue only the existing ProjectBrain main push after the fixed evidence fingerprint is rechecked.""" + if (not isinstance(revision, str) or not re.fullmatch(r"[0-9a-f]{40}", revision) + or not isinstance(expected_preflight_sha256, str) + or not re.fullmatch(r"[0-9a-f]{64}", expected_preflight_sha256) + or not isinstance(idempotency_key, str) + or not re.fullmatch(r"[A-Za-z0-9._:-]{8,128}", idempotency_key)): + raise ValueError("ProjectBrain revalidation contract is invalid") + return _json(DeploymentBrokerClient().projectbrain_delivery_operation( + "revalidate", { + "revision": revision, + "expectedPreflightSha256": expected_preflight_sha256, + "idempotencyKey": idempotency_key, + }, + )) + + @appops_mcp.tool(name="server.operation.execute", annotations=_LEGACY_APPROVAL_EXECUTION_TOOL) @_tracked("server.operation.execute") def execute_operation( diff --git a/backend/app/services/deployment_broker.py b/backend/app/services/deployment_broker.py index 87dfff3..837455b 100644 --- a/backend/app/services/deployment_broker.py +++ b/backend/app/services/deployment_broker.py @@ -294,6 +294,28 @@ class DeploymentBrokerClient: error_response_fields=frozenset({"noMutation"}), ) + def projectbrain_delivery_operation(self, stage: str, payload: dict[str, Any]) -> dict[str, Any]: + """Use the fixed ProjectBrain retry gate; callers cannot select a repository or runtime.""" + if stage not in {"preflight", "revalidate"} or not isinstance(payload, dict): + raise ValueError("ProjectBrain delivery stage is invalid") + if stage == "preflight": + if (set(payload) != {"revision"} or not isinstance(payload.get("revision"), str) + or not re.fullmatch(r"[0-9a-f]{40}", payload["revision"])): + raise ValueError("ProjectBrain preflight contract is invalid") + elif (set(payload) != {"revision", "expectedPreflightSha256", "idempotencyKey"} + or not isinstance(payload.get("revision"), str) + or not re.fullmatch(r"[0-9a-f]{40}", payload["revision"]) + or not isinstance(payload.get("expectedPreflightSha256"), str) + or not re.fullmatch(r"[0-9a-f]{64}", payload["expectedPreflightSha256"]) + or not isinstance(payload.get("idempotencyKey"), str) + or not re.fullmatch(r"[A-Za-z0-9._:-]{8,128}", payload["idempotencyKey"])): + raise ValueError("ProjectBrain revalidation contract is invalid") + return self._request( + "POST", f"/v1/appops/projectbrain-delivery/{stage}", + secret=settings.mcp_token or "", payload=payload, + error_response_fields=frozenset({"noMutation"}), + ) + def registered_legacy_operation(self, stage: str, payload: dict[str, Any]) -> dict[str, Any]: """Signed fixed-stage metadata adoption/recovery for existing registered targets.""" if stage not in {"observe", "baseline", "preflight", "adopt", "status", "rollback", "delivery"}: diff --git a/backend/tests/test_controller_artifact_bootstrap_created_child_recovery.py b/backend/tests/test_controller_artifact_bootstrap_created_child_recovery.py new file mode 100644 index 0000000..bf4cf5f --- /dev/null +++ b/backend/tests/test_controller_artifact_bootstrap_created_child_recovery.py @@ -0,0 +1,627 @@ +from __future__ import annotations + +import copy +from typing import Any + +import pytest + +from backend.tests import test_controller_artifact_bootstrap as original +from scripts import bootstrap_controller_artifact_promoter as bootstrap + +FAILED_REVISION = "1" * 40 +FAILED_TREE = "2" * 40 +FAILED_HELPER_ID = "3" * 64 +CURRENT_HELPER_ID = "4" * 64 +DRIFTED_HELPER_ID = "9" * 64 +ZERO_DOCKER_TIME = "0001-01-01T00:00:00Z" + + +class Git(original.FakeProtectedGit): + def __init__(self, helper_source: str): + super().__init__(helper_source) + + def verify_failed_modern_ancestor( + self, revision: str, current_revision: str + ) -> str: + if ( + revision != FAILED_REVISION + or current_revision != self.revision + or not self.retired_tree_valid + ): + raise bootstrap.BootstrapError( + "controller_promoter_failed_helper_not_ancestor" + ) + return FAILED_TREE + + def read_blob(self, revision: str, path: str) -> bytes: + if revision == FAILED_REVISION: + self.blobs_read.append((revision, path)) + if path != bootstrap.MODERN_HELPER_SOURCE_PATH: + raise bootstrap.BootstrapError( + "controller_promoter_source_blob_invalid" + ) + return self.helper_source.encode("utf-8") + return super().read_blob(revision, path) + + +class Docker(original.FakeDocker): + def __init__(self, existing_child: dict[str, Any]): + super().__init__(original._legacy_helper_spec(), existing_child=existing_child) + self.name_reads = 0 + self.name_drift_at: int | None = None + self.name_drift: str | None = None + self.id_reads = 0 + self.id_drift_at: int | None = None + self.id_drift: str | None = None + self.create_attempts: list[tuple[str, dict[str, Any]]] = [] + self.log_reads: list[str] = [] + self.created_log_bytes = b"" + self.created_log_failure = False + self.complete_existing_on_start = False + self.existing_exit_code = 0 + + def inspect_container(self, reference: str) -> dict[str, Any] | None: + if ( + reference == bootstrap.MODERN_HELPER_NAME + and self.existing_child is not None + ): + self.name_reads += 1 + snapshot = copy.deepcopy(self.existing_child) + if self.name_reads == self.name_drift_at: + if self.name_drift == "id": + snapshot["Id"] = DRIFTED_HELPER_ID + elif self.name_drift == "name": + snapshot["Name"] = "/itworx-controller-artifact-promoter-foreign" + return snapshot + if self.existing_child is not None and reference == self.existing_child.get( + "Id" + ): + self.id_reads += 1 + snapshot = copy.deepcopy(self.existing_child) + if self.id_reads == self.id_drift_at: + if self.id_drift == "id": + snapshot["Id"] = DRIFTED_HELPER_ID + elif self.id_drift == "name": + snapshot["Name"] = "/itworx-controller-artifact-promoter-foreign" + return snapshot + return super().inspect_container(reference) + + def create_container(self, name: str, spec: dict[str, Any]) -> str: + self.create_attempts.append((name, copy.deepcopy(spec))) + return super().create_container(name, spec) + + def remove_retired_failed_container(self, identifier: str) -> None: + assert self.existing_child is not None + assert identifier == self.existing_child["Id"] + self.removed.append(identifier) + self.existing_child = None + + def start_container(self, identifier: str) -> None: + self.started.append(identifier) + if self.existing_child is not None and identifier == self.existing_child["Id"]: + if self.complete_existing_on_start: + self.existing_child["State"].update( + Running=False, + Status="exited", + ExitCode=self.existing_exit_code, + ) + source_revision = self.existing_child["Config"]["Labels"][ + "itworx.source.revision" + ] + self.containers[bootstrap.CONTROLLER_NAME] = ( + original._modern_controller_inspect( + source_revision, identifier="e" * 64 + ) + ) + else: + self.existing_child["State"].update(Running=True, Status="running") + return + child = self.containers.get(identifier) + if child is None: + return + child["State"].update(Running=True, Status="running") + source_revision = self.created[-1]["Labels"]["itworx.source.revision"] + self.containers[bootstrap.CONTROLLER_NAME] = ( + original._modern_controller_inspect(source_revision, identifier="e" * 64) + ) + + def container_logs(self, identifier: str, *, tail: int, max_bytes: int) -> bytes: + self.log_reads.append(identifier) + assert tail <= 40 + assert max_bytes <= bootstrap.MAX_DOCKER_LOG_BYTES + return self.log_bytes[:max_bytes] + + def container_logs_for_created_container( + self, identifier: str, *, tail: int, max_bytes: int + ) -> bytes: + if self.created_log_failure: + raise bootstrap.BootstrapError("controller_promoter_docker_logs_failed") + if self.existing_child is not None and identifier == self.existing_child["Id"]: + return self.created_log_bytes + return b"" + + +def _created_state() -> dict[str, Any]: + return { + "Status": "created", + "Running": False, + "Paused": False, + "Restarting": False, + "Dead": False, + "Pid": 0, + "OOMKilled": False, + "ExitCode": 0, + "StartedAt": ZERO_DOCKER_TIME, + "FinishedAt": ZERO_DOCKER_TIME, + "Error": "", + } + + +def _inputs(*, current_child: bool) -> tuple[Docker, Git, dict[str, Any], str]: + helper_source = (original.ROOT / bootstrap.MODERN_HELPER_SOURCE_PATH).read_text( + encoding="utf-8" + ) + child_revision = original.REVISION if current_child else FAILED_REVISION + child_id = CURRENT_HELPER_ID if current_child else FAILED_HELPER_ID + baseline = bootstrap.MODERN_CONTROLLER_BASELINE + spec = bootstrap.render_modern_helper_spec( + helper_source, + revision=child_revision, + image_id=baseline.image_id, + current_controller_id=baseline.container_id, + current_controller_revision=baseline.revision, + current_controller_tree=baseline.tree, + current_controller_version=baseline.version, + ) + child = original._retired_failed_child(helper_source) + child["Id"] = child_id + child["Image"] = baseline.image_id + child["State"] = _created_state() + child["RestartCount"] = 0 + + docker = Docker(child) + image_config = docker.images[baseline.image_id]["Config"] + image_config.update( + { + "Env": ["PATH=/usr/bin"], + "WorkingDir": "/tmp", + "Entrypoint": None, + "Shell": ["/bin/bash", "-c"], + } + ) + base_image_labels = bootstrap._inspect_controller_base_image(docker) + child["Config"] = { + "Hostname": docker.host_name, + "Domainname": "", + "User": spec["User"], + "AttachStdin": False, + "AttachStdout": False, + "AttachStderr": False, + "Tty": False, + "OpenStdin": False, + "StdinOnce": False, + "Cmd": copy.deepcopy(spec["Cmd"]), + "Image": spec["Image"], + "Labels": {**base_image_labels, **spec["Labels"]}, + "Env": list(image_config["Env"]), + "Volumes": None, + "WorkingDir": image_config["WorkingDir"], + "Entrypoint": image_config["Entrypoint"], + "OnBuild": None, + } + child["HostConfig"] = { + **copy.deepcopy(bootstrap.FAILED_HELPER_HOST_DEFAULTS), + **copy.deepcopy(spec["HostConfig"]), + "RestartPolicy": {"Name": "no", "MaximumRetryCount": 0}, + } + docker.containers[bootstrap.CONTROLLER_NAME] = ( + original._modern_controller_source_inspect(baseline.revision) + ) + return docker, Git(helper_source), child, child_revision + + +def _run( + docker: Docker, + git: Git, + *, + health: dict[str, Any] | None = None, + health_reader=None, +) -> dict[str, Any]: + payload = health or original._modern_baseline_health() + return bootstrap.run_bootstrap( + bootstrap.SOURCE_REPOSITORY, + git.revision, + git.tree, + docker=docker, + protected_git=git, + hostname=original.HOST_HOSTNAME, + health_reader=health_reader or (lambda: copy.deepcopy(payload)), + monotonic=lambda: 0.0, + sleeper=lambda _seconds: None, + child_timeout_seconds=0.0, + ) + + +def test_current_created_child_resumes_after_exact_baseline_and_identity_rechecks(): + docker, git, child, _revision = _inputs(current_child=True) + + result = _run(docker, git) + + assert result["status"] == "outcome_unknown" + assert docker.started == [child["Id"]] + assert docker.created == [] + assert docker.create_attempts == [] + assert docker.removed == [] + + +def test_current_created_child_success_is_reported_as_promoted(): + docker, git, child, _revision = _inputs(current_child=True) + docker.complete_existing_on_start = True + health_reads = 0 + + def health_reader() -> dict[str, Any]: + nonlocal health_reads + health_reads += 1 + if health_reads <= 3: + return original._modern_baseline_health() + return original._health(revision=git.revision, tree=git.tree) + + result = _run(docker, git, health_reader=health_reader) + + assert result["status"] == "promoted" + assert result["revision"] == git.revision + assert docker.started == [child["Id"]] + assert docker.log_reads == [child["Id"]] + assert docker.create_attempts == [] + assert docker.created == [] + assert docker.removed == [] + + +def test_resumed_child_failure_wins_even_if_health_has_reached_target(): + docker, git, child, _revision = _inputs(current_child=True) + docker.complete_existing_on_start = True + docker.existing_exit_code = 17 + health_reads = 0 + + def health_reader() -> dict[str, Any]: + nonlocal health_reads + health_reads += 1 + if health_reads <= 3: + return original._modern_baseline_health() + return original._health(revision=git.revision, tree=git.tree) + + result = _run(docker, git, health_reader=health_reader) + + assert result["status"] == "failed" + assert result["exitCode"] == 17 + assert docker.started == [child["Id"]] + assert docker.log_reads == [child["Id"]] + assert docker.create_attempts == [] + assert docker.created == [] + + +def test_noncurrent_created_child_is_retired_without_starting_its_old_revision(): + docker, git, child, _revision = _inputs(current_child=False) + + result = _run(docker, git) + + assert result["status"] == "outcome_unknown" + assert docker.removed == [child["Id"]] + assert docker.started == [original.MODERN_HELPER_ID] + assert docker.created[0]["Labels"]["itworx.source.revision"] == git.revision + assert len(docker.create_attempts) == 1 + + +def test_current_created_name_collision_uses_the_same_recovery_proof(): + docker, git, child, _revision = _inputs(current_child=True) + docker.existing_child = None + collision = copy.deepcopy(child) + + def collide(_name: str, _spec: dict[str, Any]) -> str: + docker.create_attempts.append((_name, copy.deepcopy(_spec))) + docker.existing_child = collision + raise bootstrap.BootstrapError( + "controller_promoter_modern_helper_create_failed" + ) + + docker.create_container = collide + + result = _run(docker, git) + + assert result["status"] == "outcome_unknown" + assert docker.started == [child["Id"]] + assert docker.created == [] + assert len(docker.create_attempts) == 1 + assert docker.removed == [] + + +@pytest.mark.parametrize( + "state", + [ + {"Status": "created", "Running": False, "ExitCode": 0}, + {"Status": "exited", "Running": True, "ExitCode": 0}, + {"Status": "exited", "Running": False, "ExitCode": True}, + {"Status": "exited", "Running": False, "ExitCode": "0"}, + ], +) +def test_completed_child_result_checks_terminal_integer_state_before_logs( + state: dict[str, Any], +): + docker, _git, child, _revision = _inputs(current_child=True) + child["State"] = state + + with pytest.raises(bootstrap.BootstrapError): + bootstrap._completed_child_result( + child, + docker=docker, + health_reader=original._health, + source_revision=original.REVISION, + source_tree=original.TREE, + already_applied=True, + ) + + assert docker.log_reads == [] + + +@pytest.mark.parametrize("current_child", [True, False]) +@pytest.mark.parametrize( + "mutation", + [ + "status", + "exit_code_bool", + "exit_code_nonzero", + "exit_code_string", + "paused", + "restarting", + "dead", + "started_at", + "finished_at", + "started_at_type", + "finished_at_type", + "error", + "error_type", + "pid", + "pid_type", + "oom_killed", + "restart_count", + "restart_count_type", + ], +) +def test_created_child_requires_exact_never_started_state( + current_child: bool, mutation: str +): + docker, git, _child, _revision = _inputs(current_child=current_child) + state = docker.existing_child["State"] + if mutation == "status": + state["Status"] = "created " + elif mutation == "exit_code_bool": + state["ExitCode"] = False + elif mutation == "exit_code_nonzero": + state["ExitCode"] = 1 + elif mutation == "exit_code_string": + state["ExitCode"] = "0" + elif mutation in {"paused", "restarting", "dead"}: + state[mutation.capitalize()] = True + elif mutation == "started_at": + state["StartedAt"] = "2026-10-07T09:00:00Z" + elif mutation == "finished_at": + state["FinishedAt"] = "2026-10-07T09:00:00Z" + elif mutation == "started_at_type": + state["StartedAt"] = 0 + elif mutation == "finished_at_type": + state["FinishedAt"] = 0 + elif mutation == "pid": + state["Pid"] = 32 + elif mutation == "pid_type": + state["Pid"] = False + elif mutation == "oom_killed": + state["OOMKilled"] = True + elif mutation == "restart_count": + docker.existing_child["RestartCount"] = 1 + elif mutation == "restart_count_type": + docker.existing_child["RestartCount"] = False + elif mutation == "error_type": + state["Error"] = None + else: + state["Error"] = "start failed" + + try: + result = _run(docker, git) + except bootstrap.BootstrapError: + pass + else: + assert result["status"] == "outcome_unknown" + + assert docker.started == [] + assert docker.removed == [] + assert docker.created == [] + + +@pytest.mark.parametrize("current_child", [True, False]) +@pytest.mark.parametrize( + "mutation", + [ + "command", + "environment", + "label", + "mount", + "logs", + "log_error", + "target_baseline", + "health_artifact", + "main", + "controller", + "appops", + ], +) +def test_created_child_profile_or_frozen_baseline_drift_prevents_action( + current_child: bool, mutation: str +): + docker, git, _child, _revision = _inputs(current_child=current_child) + health = original._modern_baseline_health() + if mutation == "command": + docker.existing_child["Config"]["Cmd"][2] += "\necho changed" + elif mutation == "environment": + docker.existing_child["Config"]["Env"].append("CUSTOM_VARIABLE=extra") + elif mutation == "label": + docker.existing_child["Config"]["Labels"]["itworx.unexpected"] = "true" + elif mutation == "mount": + docker.existing_child["Mounts"][0]["RW"] = not docker.existing_child["Mounts"][ + 0 + ]["RW"] + elif mutation == "logs": + docker.created_log_bytes = b"unexpected pre-start output\n" + elif mutation == "log_error": + docker.created_log_failure = True + elif mutation in {"target_baseline", "health_artifact"}: + health = original._health() + if mutation == "health_artifact": + health = original._modern_baseline_health() + health["artifact"]["artifactSha256"] = "f" * 64 + elif mutation == "main": + git.fail_current_main_at = 2 + elif mutation == "controller": + docker.containers[bootstrap.CONTROLLER_NAME]["Id"] = DRIFTED_HELPER_ID + else: + docker.containers[original.APPOPS_CONTAINER_ID]["Image"] = "sha256:" + "8" * 64 + + try: + result = _run(docker, git, health=health) + except bootstrap.BootstrapError: + pass + else: + assert result["status"] == "outcome_unknown" + + assert docker.started == [] + assert docker.removed == [] + assert docker.created == [] + + +@pytest.mark.parametrize("current_child", [True, False]) +@pytest.mark.parametrize( + ("reference", "read_number", "drift"), + [ + ("name", 2, "id"), + ("name", 2, "name"), + ("name", 3, "id"), + ("name", 3, "name"), + ("id", 1, "id"), + ("id", 1, "name"), + ("id", 2, "id"), + ("id", 2, "name"), + ], +) +def test_created_child_name_or_id_drift_on_reinspection_prevents_action( + current_child: bool, reference: str, read_number: int, drift: str +): + docker, git, _child, _revision = _inputs(current_child=current_child) + if reference == "name": + # The first fixed-name lookup precedes the two proof snapshot pairs. + docker.name_drift_at = read_number + docker.name_drift = drift + else: + docker.id_drift_at = read_number + docker.id_drift = drift + + try: + result = _run(docker, git) + except bootstrap.BootstrapError: + pass + else: + assert result["status"] == "outcome_unknown" + + assert docker.started == [] + assert docker.removed == [] + assert docker.created == [] + assert docker.create_attempts == [] + + +def test_running_current_helper_remains_unknown_and_is_never_started_again(): + docker, git, _child, _revision = _inputs(current_child=True) + docker.existing_child["State"].update(Running=True, Status="running") + + result = _run(docker, git) + + assert result["status"] == "outcome_unknown" + assert docker.started == [] + assert docker.removed == [] + assert docker.created == [] + + +def test_created_container_log_request_is_escaped_and_bounded(): + client = object.__new__(bootstrap.DockerSocketClient) + requests: list[tuple[str, str, int]] = [] + + def request(method: str, path: str, *, max_bytes: int) -> tuple[int, bytes]: + requests.append((method, path, max_bytes)) + return 200, b"" + + client._request = request # type: ignore[method-assign] + + assert ( + client.container_logs_for_created_container( + "abc/def", tail=500, max_bytes=bootstrap.MAX_DOCKER_LOG_BYTES * 2 + ) + == b"" + ) + assert requests == [ + ( + "GET", + "/containers/abc%2Fdef/logs?stdout=1&stderr=1&tail=40", + bootstrap.MAX_DOCKER_LOG_BYTES, + ) + ] + + +@pytest.mark.parametrize("status", [204, 404, 500]) +def test_created_container_log_request_rejects_non_200_status(status: int): + client = object.__new__(bootstrap.DockerSocketClient) + client._request = lambda *_args, **_kwargs: (status, b"") # type: ignore[method-assign] + + with pytest.raises(bootstrap.BootstrapError, match="docker_logs_failed"): + client.container_logs_for_created_container( + "a" * 64, tail=40, max_bytes=bootstrap.MAX_DOCKER_LOG_BYTES + ) + + +def test_created_container_log_request_keeps_response_size_guard( + monkeypatch: pytest.MonkeyPatch, +): + read_sizes: list[int] = [] + + class Response: + status = 200 + + def read(self, size: int) -> bytes: + read_sizes.append(size) + return b"x" * size + + class Connection: + def __init__(self, _socket_path: str): + pass + + def request(self, *_args: Any, **_kwargs: Any) -> None: + pass + + def getresponse(self) -> Response: + return Response() + + def close(self) -> None: + pass + + monkeypatch.setattr(bootstrap, "_UnixHTTPConnection", Connection) + client = bootstrap.DockerSocketClient() + + with pytest.raises(bootstrap.BootstrapError, match="response_too_large"): + client.container_logs_for_created_container( + "a" * 64, tail=40, max_bytes=bootstrap.MAX_DOCKER_LOG_BYTES * 2 + ) + + assert read_sizes == [bootstrap.MAX_DOCKER_LOG_BYTES + 1] + + +def test_finished_child_logs_keep_existing_non_200_empty_behavior(): + client = object.__new__(bootstrap.DockerSocketClient) + client._request = lambda *_args, **_kwargs: (404, b"") # type: ignore[method-assign] + + assert client.container_logs("a" * 64, tail=40, max_bytes=8192) == b"" diff --git a/backend/tests/test_deploy_broker.py b/backend/tests/test_deploy_broker.py index faecb6f..24112f5 100644 --- a/backend/tests/test_deploy_broker.py +++ b/backend/tests/test_deploy_broker.py @@ -9,6 +9,7 @@ import io import ipaddress import json import os +import re import socket import ssl import stat @@ -8237,9 +8238,18 @@ def test_appops_uses_push_validation_without_a_deploy_runner_workflow(): managed_validation = (workflow_root / "managed-validation.yml").read_text(encoding="utf-8") assert "push:" in managed_validation assert "branches: [main]" in managed_validation - assert "runs-on: linux-validation" in managed_validation assert "unraid-deploy" not in managed_validation + # Keep all jobs on the installed legacy fleet during specialized-runner bootstrap. + expected_runner = "linux-validation" + for job in ("preflight", "backend", "frontend", "security", "polyglot", "full"): + match = re.search( + rf"(?ms)^ {re.escape(job)}:\n(?P.*?)(?=^ [a-z][a-z0-9_-]*:\n|\Z)", + managed_validation, + ) + assert match is not None, f"missing managed-validation job {job}" + assert re.search(rf"(?m)^ runs-on: {re.escape(expected_runner)}\s*$", match["body"]) + def test_chimera_example_requires_exact_source_webhook_auth_and_live_proof(): projects = json.loads( diff --git a/backend/tests/test_gitea_actions.py b/backend/tests/test_gitea_actions.py index eda5b8c..808f3f0 100644 --- a/backend/tests/test_gitea_actions.py +++ b/backend/tests/test_gitea_actions.py @@ -40,6 +40,10 @@ def _configure(monkeypatch): '{"unraid-deploy":"itworx-security-runner"}', ) monkeypatch.setattr(gitea_actions, "_CACHE", None) + monkeypatch.setattr(gitea_actions, "_RECOVERY_OBSERVATIONS", {}, raising=False) + monkeypatch.setattr(gitea_actions, "_RECOVERY_SOURCE", None, raising=False) + monkeypatch.setattr(gitea_actions, "_SNAPSHOT_GENERATION", 0, raising=False) + monkeypatch.setattr(gitea_actions, "_COMMITTED_GENERATION", 0, raising=False) def test_runner_snapshot_recommends_only_approval_gated_restart_for_stalled_queue( diff --git a/backend/tests/test_hub_candidate_import.py b/backend/tests/test_hub_candidate_import.py index 1312b48..72f3edb 100644 --- a/backend/tests/test_hub_candidate_import.py +++ b/backend/tests/test_hub_candidate_import.py @@ -1098,6 +1098,96 @@ def test_tar_preflight_rejects_extension_headers_before_tarfile(modules, tmp_pat importer._preflight_uncompressed_tar(path) +_CUSTOM_DB_REPO_DIGESTS = { + "postgres": "gitea.example.com/jens/itworx-mcp-hub/postgres@sha256:a60d81e06987671ae903757f1b71cdfbbff4d481b42610e4de2b7dcfa32aad73", + "redis": "gitea.example.com/jens/itworx-mcp-hub/redis@sha256:7d826b44f245ad32e8fe4bf40d42eae95855c3942c5f2e42784ee8fd3f9547a2", +} + + +@pytest.mark.parametrize("service", ("postgres", "redis")) +def test_custom_external_database_accepts_exact_custom_repo_digest( + modules, monkeypatch, service +): + importer = modules.importer + image_ref, archive_ref = importer.EXTERNAL_IMAGES[service][:2] + custom_repo_digest = _CUSTOM_DB_REPO_DIGESTS[service] + image_id = "sha256:" + "a" * 64 + calls = [] + + def inspect(reference): + calls.append(reference) + if reference == archive_ref: + return {"Id": image_id} + if reference == image_ref: + return {"Id": image_id, "RepoDigests": [custom_repo_digest]} + return None + + monkeypatch.setattr(importer, "_inspect_image", inspect) + result = importer._require_loaded_image( + service, + {"imageRef": image_ref, "archiveRef": archive_ref, "imageId": image_id}, + external=True, + ) + + assert result["Id"] == image_id + assert calls == [archive_ref, image_ref] + + +@pytest.mark.parametrize("service", ("postgres", "redis")) +@pytest.mark.parametrize( + "failure", ("upstream", "wrong_custom", "missing_repo_digests", "foreign_image_id") +) +def test_custom_external_database_rejects_nonmatching_repo_digest_or_image_id( + modules, monkeypatch, service, failure +): + importer = modules.importer + image_ref, archive_ref = importer.EXTERNAL_IMAGES[service][:2] + custom_repo_digest = _CUSTOM_DB_REPO_DIGESTS[service] + image_id = "sha256:" + "a" * 64 + foreign_id = "sha256:" + "f" * 64 + if failure == "upstream": + repo_digests = { + "postgres": ( + "postgres@sha256:0377e72c5289ed2f98cf61b1a9c2db9eb9d300317fe14244492fbc94343b3d04", + "docker.io/library/postgres@sha256:0377e72c5289ed2f98cf61b1a9c2db9eb9d300317fe14244492fbc94343b3d04", + ), + "redis": ( + "redis@sha256:7ef5b5cec96495a04ca7feff88a9492efeab8053fb284d24bdd73344c9245a48", + "docker.io/library/redis@sha256:7ef5b5cec96495a04ca7feff88a9492efeab8053fb284d24bdd73344c9245a48", + ), + }[service] + elif failure == "wrong_custom": + repository = custom_repo_digest.rsplit("@", 1)[0] + repo_digests = [repository + "@sha256:" + "b" * 64] + elif failure == "missing_repo_digests": + repo_digests = None + else: + repo_digests = [custom_repo_digest] + + def inspect(reference): + if reference == archive_ref: + return {"Id": image_id} + if reference == image_ref: + return { + "Id": foreign_id if failure == "foreign_image_id" else image_id, + "RepoDigests": repo_digests, + } + return None + + monkeypatch.setattr(importer, "_inspect_image", inspect) + error = ( + "Pinned external image ID differs" + if failure == "foreign_image_id" + else "Pinned external digest is not present" + ) + with pytest.raises(importer.CandidateRejected, match=error): + importer._require_loaded_image( + service, + {"imageRef": image_ref, "archiveRef": archive_ref, "imageId": image_id}, + external=True, + ) + + def test_remote_head_without_completed_candidate_stays_awaiting(modules, tmp_path, monkeypatch): controller = modules.controller prepare = modules.prepare diff --git a/backend/tests/test_projectbrain_delivery_revalidation.py b/backend/tests/test_projectbrain_delivery_revalidation.py new file mode 100644 index 0000000..fcd5a4f --- /dev/null +++ b/backend/tests/test_projectbrain_delivery_revalidation.py @@ -0,0 +1,1265 @@ +from __future__ import annotations + +import hashlib +import io +import json +import subprocess +import sys +import time +import uuid +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from email.message import Message +from pathlib import Path +from types import SimpleNamespace + +import pytest +from test_deploy_broker import BROKER_ROOT, _load_module + +REPOSITORY = "jens/projectbrain" +REVISION = "2713a23f6240bdabc1c6c45805a6264b9aaedad4" +PREFIX = "/v1/appops/projectbrain-delivery/" +PREFLIGHT_SHA = "a" * 64 +ORIGINAL_FILE_SHA = "b" * 64 +DELIVERY_SHA = "c" * 64 + + +@pytest.fixture +def controller(monkeypatch): + module = _load_module( + "projectbrain_delivery_revalidation_test", + BROKER_ROOT / "status_server.py", + ) + monkeypatch.setattr(module, "_self_upgrade_signature_is_valid", lambda *_a, **_k: True) + monkeypatch.setattr(module, "_self_upgrade_secret", lambda: b"test-key-" + b"x" * 48) + monkeypatch.setattr(module, "_require_controller_release", lambda: {"status": "ready"}) + return module + + +def invoke(module, stage: str, payload: dict[str, object], *, trusted: bool = True): + handler = module.Handler.__new__(module.Handler) + body = json.dumps(payload, separators=(",", ":")).encode() + handler.path = PREFIX + stage + handler.headers = Message() + handler.headers["Content-Length"] = str(len(body)) + handler.headers["Content-Type"] = "application/json" + handler.rfile = io.BytesIO(body) + handler._trusted_client = lambda: trusted + responses = [] + handler._send_json = lambda status, data: responses.append((status, json.loads(data))) + handler._single_header = lambda name: ( + "0" * 32 if name == "X-ITWorx-Deployment-Nonce" + else "0" if name in {"X-ITWorx-Deployment-Timestamp", "X-ITWorx-Deployment-Signature"} + else handler.headers.get(name) + ) + handler._handle_projectbrain_delivery() + return responses[0] + + +def snapshot(): + return { + "status": "ready", + "repository": REPOSITORY, + "revision": REVISION, + "originalDelivery": DELIVERY_SHA, + "liveRevision": "246c0a7f35e32f54fa1bf8a3f720b01bb00dd9af", + "liveReceiptSha256": "d" * 64, + "imageId": "sha256:" + "e" * 64, + "validationRunId": 22470, + "validationRunAttempt": 1, + "validationJobCount": 14, + "preflightSha256": PREFLIGHT_SHA, + "_originalFileSha256": ORIGINAL_FILE_SHA, + "_semanticDigest": DELIVERY_SHA, + "_stateSha256": "1" * 64, + "_receiptSha256": "d" * 64, + "_runtimeSha256": "2" * 64, + "_profileSha256": "3" * 64, + "_targetScriptSha256": "6" * 64, + "_controllerArtifactSha256": "7" * 64, + "_environmentFingerprint": "4" * 64, + "_validationEvidenceSha256": "5" * 64, + } + + +def test_preflight_is_signed_read_only_and_returns_only_bounded_snapshot_metadata( + controller, monkeypatch +): + calls = [] + monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda revision: calls.append(revision) or snapshot()) + monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", _null_context) + monkeypatch.setattr( + controller, + "_enqueue_terminal_revalidation", + lambda *_a, **_k: pytest.fail("preflight must not write a revalidation record"), + ) + + code, result = invoke(controller, "preflight", {"revision": REVISION}) + + assert code == 200 + assert result == { + "status": "ready", + "repository": REPOSITORY, + "revision": REVISION, + "originalDelivery": DELIVERY_SHA, + "liveRevision": "246c0a7f35e32f54fa1bf8a3f720b01bb00dd9af", + "liveReceiptSha256": "d" * 64, + "imageId": "sha256:" + "e" * 64, + "validationRunId": 22470, + "validationRunAttempt": 1, + "validationJobCount": 14, + "preflightSha256": PREFLIGHT_SHA, + } + assert calls == [REVISION] + + +def test_preflight_authenticates_before_reading_project_state(controller, monkeypatch): + monkeypatch.setattr(controller, "_self_upgrade_signature_is_valid", lambda *_a, **_k: False) + monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda *_a: pytest.fail("unauthenticated read")) + + assert invoke(controller, "preflight", {"revision": REVISION})[0] == 403 + + +def test_preflight_reports_busy_without_sampling_during_deployment(controller, monkeypatch): + monkeypatch.setattr( + controller, + "_projectbrain_delivery_mutation_lock", + lambda: _raise_busy(controller), + ) + monkeypatch.setattr( + controller, + "_projectbrain_delivery_snapshot", + lambda *_a: pytest.fail("preflight must not read an incoherent deployment snapshot"), + ) + + code, result = invoke(controller, "preflight", {"revision": REVISION}) + + assert code == 409 + assert result == {"status": "blocked", "reason": "projectbrain_deployment_active"} + + +def test_revalidate_rechecks_fingerprint_and_binds_only_idempotency_digest( + controller, monkeypatch +): + observed = [] + queued = [] + monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda revision: observed.append(revision) or snapshot()) + monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", lambda: _null_context()) + monkeypatch.setattr( + controller, + "_request_terminal_revalidation", + lambda *args, **kwargs: queued.append((args, kwargs)) or { + "status": "awaiting_validation", "delivery": DELIVERY_SHA, + }, + ) + key = "pb-revalidate-2026-10-07" + + code, result = invoke( + controller, + "revalidate", + {"revision": REVISION, "expectedPreflightSha256": PREFLIGHT_SHA, "idempotencyKey": key}, + ) + + assert code == 202 + assert result["status"] == "awaiting_validation" + assert result["repository"] == REPOSITORY + assert result["revision"] == REVISION + assert result["delivery"] == DELIVERY_SHA + assert observed == [REVISION] + assert queued[0][0][0:2] == (REPOSITORY, REVISION) + assert queued[0][1]["expected_original_sha256"] == ORIGINAL_FILE_SHA + assert queued[0][1]["projectbrain_delivery_binding"]["idempotencyKeySha256"] == hashlib.sha256(key.encode()).hexdigest() + assert queued[0][1]["allow_projectbrain_terminal_idempotent"] is True + assert key not in json.dumps(queued) + + +def test_revalidate_rejects_stale_fingerprint_without_queueing(controller, monkeypatch): + queued = [] + monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda _revision: snapshot()) + monkeypatch.setattr(controller, "_projectbrain_delivery_mutation_lock", lambda: _null_context()) + monkeypatch.setattr(controller, "_request_terminal_revalidation", lambda *a, **k: queued.append((a, k))) + + code, result = invoke( + controller, + "revalidate", + {"revision": REVISION, "expectedPreflightSha256": "f" * 64, "idempotencyKey": "pb-revalidate-2026-10-07"}, + ) + + assert code == 409 + assert result == {"status": "blocked", "reason": "projectbrain_delivery_preflight_changed"} + assert queued == [] + + +def test_revalidate_rejects_unknown_or_unbounded_fields_before_snapshot(controller, monkeypatch): + monkeypatch.setattr(controller, "_projectbrain_delivery_snapshot", lambda *_a: pytest.fail("invalid input read state")) + + assert invoke(controller, "preflight", {"revision": REVISION, "repository": REPOSITORY})[0] == 400 + assert invoke( + controller, + "revalidate", + {"revision": REVISION, "expectedPreflightSha256": PREFLIGHT_SHA, "idempotencyKey": "short"}, + )[0] == 400 + + +def _null_context(): + from contextlib import nullcontext + + return nullcontext() + + +def _raise_busy(module): + raise module.ProjectBrainDeliveryRefused("projectbrain_deployment_active") + + +def _git(cwd: Path, *arguments: str) -> str: + result = subprocess.run( + ["git", *arguments], cwd=cwd, capture_output=True, text=True, check=True, + ) + return result.stdout.strip() + + +def _snapshot_fixture(controller, monkeypatch, tmp_path, *, target_root_env=False): + repository = tmp_path / "source" + repository.mkdir() + _git(repository, "init", "-b", "main") + _git(repository, "config", "user.email", "projectbrain-test@example.invalid") + _git(repository, "config", "user.name", "ProjectBrain test") + (repository / "scripts").mkdir() + deploy_script = repository / "scripts" / "deploy-unraid-single.sh" + deploy_script.write_text("#!/bin/sh\nprintf 'live\\n'\n", encoding="utf-8") + (repository / "README.md").write_text("live\n", encoding="utf-8") + _git(repository, "add", ".") + _git(repository, "commit", "-m", "live") + live_revision = _git(repository, "rev-parse", "HEAD") + deploy_script.write_text("#!/bin/sh\nprintf 'target\\n'\n", encoding="utf-8") + (repository / "README.md").write_text("target\n", encoding="utf-8") + if target_root_env: + (repository / ".env").write_text( + "PROJECTBRAIN_PRIVATE_TEST=fixture-only\n", + encoding="utf-8", + ) + _git(repository, "add", ".") + _git(repository, "commit", "-m", "target") + revision = _git(repository, "rev-parse", "HEAD") + repository_path = repository / ".git" + + validation = { + "validation_gate_version": 1, + "required": True, + "contexts": ["Managed validation / full (push)"], + "timeout_seconds": 1800, + "poll_interval_seconds": 5, + "max_poll_interval_seconds": 30, + } + project = { + "slug": "projectbrain", + "branch": "main", + "source_dir": "/srv/example/appdata/brain", + "deploy_script": "scripts/deploy-unraid-single.sh", + "archive_deploy_script": None, + "containers": ["projectbrain"], + "compose_project": "projectbrain", + "compose_files": ["docker-compose.yml"], + "gitea_repository_id": 7, + "environment": { + "PROJECTBRAIN_DEPLOY_SKIP_PULL": "true", + "PROJECTBRAIN_DEPLOY_REF": "", + "VITE_PROJECTBRAIN_COMMIT": "{revision}", + }, + "gitea_webhook_secret_file": "jens__projectbrain", + "verify_oci_revision": False, + "health_timeout": 180, + "validation": validation, + } + inbox = tmp_path / "inbox" + inbox.mkdir(mode=0o700) + monkeypatch.setattr(controller, "WEBHOOK_INBOX_ROOT", inbox) + monkeypatch.setattr(controller, "_load_projects", lambda: {REPOSITORY: project}) + monkeypatch.setattr(controller, "_is_current_branch_head", lambda *_a: True) + monkeypatch.setattr(controller, "_repository_path_for_delivery", lambda *_a: repository_path) + controller_release = { + "artifactSha256": "8" * 64, + "sourceRepository": "jens/unraid-appops-gateway", + "sourceRevision": "9" * 40, + "sourceTree": "a" * 40, + "builtAt": "2026-10-07T12:00:00.000Z", + "controllerVersion": "test", + "apiVersion": "itworx.controller-release/v1", + } + monkeypatch.setattr(controller, "_require_controller_release", lambda: controller_release) + + semantic = controller._webhook_semantic_digest(7, "refs/heads/main", revision) + now = int(time.time()) + original = { + "schema": controller._WEBHOOK_SCHEMA, + "semantic_sha256": semantic, + "gitea_repository_id": 7, + "ref": "refs/heads/main", + "repository": REPOSITORY, + "revision": revision, + "source": "gitea_webhook", + "status": "validation_timeout", + "reason": "validation_timeout", + "delivery": str(uuid.uuid4()), + "body_sha256": "f" * 64, + "validation": controller._validation_snapshot( + controller._validation_policy(REPOSITORY, project), now, + ), + "audit": [{"at": now, "stage": "validation_timeout", "reason": "validation_timeout"}], + "received_at": now - 1800, + "updated_at": now, + } + original_path = inbox / f"{semantic}.json" + controller._write_new_inbox_record(original_path, original) + + tree = _git(repository, "rev-parse", f"{live_revision}^{{tree}}") + tree_digest = hashlib.sha256(f"git-tree:{tree}".encode()).hexdigest() + image_id = "sha256:" + "e" * 64 + built_at = "2026-10-07T11:00:00.000Z" + receipt = { + "schema": "gitea-deploy-receipt.v1", + "repository": REPOSITORY, + "revision": live_revision, + "tree": tree, + "treeDigest": tree_digest, + "build": { + "id": live_revision[:12], + "repository": REPOSITORY, + "revision": live_revision, + "tree": tree, + "treeDigest": tree_digest, + "builtAt": built_at, + }, + "containers": {"projectbrain": {"image_id": image_id}}, + "live": None, + } + state = { + "schema_version": 3, + "repository": REPOSITORY, + "live_revision": live_revision, + "live_status": "healthy", + "live_legacy_provenance": False, + "live_receipt": receipt, + } + monkeypatch.setattr(controller, "_deployment_state", lambda _slug: state) + container = { + "Id": "d" * 64, + "Image": image_id, + "Name": "/projectbrain", + "State": {"Running": True, "Status": "running", "Health": {"Status": "healthy"}}, + "Config": {"Image": "projectbrain-unraid:codex-oidc-20260826"}, + "Mounts": [{ + "Type": "bind", + "Source": "/srv/example/appdata/projectbrain", + "Destination": "/var/lib/projectbrain", + "RW": True, + }], + } + monkeypatch.setattr(controller, "_container_inventory", lambda: {"projectbrain": container}) + + run_id = 22470 + run = { + "id": run_id, + "run_number": 381, + "run_attempt": 1, + "created_at": "2026-10-07T12:00:00Z", + "path": "managed-validation.yml@refs/heads/main", + "event": "push", + "head_branch": "main", + "head_sha": revision, + "repository": {"full_name": REPOSITORY}, + "head_repository": {"full_name": REPOSITORY}, + "pull_requests": [], + "status": "completed", + "conclusion": "success", + } + jobs = [{ + "id": 39165, + "name": "full", + "run_id": run_id, + "run_attempt": 1, + "head_sha": revision, + "head_branch": "main", + "status": "completed", + "conclusion": "success", + }] + + def api_json(path, *, query): + if path.endswith("/branches/main"): + return {"protected": True, "commit": {"id": revision}} + if path.endswith("/actions/runs"): + return {"total_count": 1, "workflow_runs": [run]} + if path.endswith(f"/actions/runs/{run_id}/jobs"): + return {"total_count": len(jobs), "jobs": jobs} + raise AssertionError(f"unexpected bounded validation API path: {path}; query={query}") + + adoption = _load_module( + "projectbrain_delivery_validation_evidence_test", + BROKER_ROOT / "registered_legacy_adoption.py", + ) + monkeypatch.setitem(sys.modules, "registered_legacy_adoption", adoption) + monkeypatch.setattr(controller, "_registered_legacy_api_reader", lambda _repository: api_json) + + deploy = _load_module("projectbrain_delivery_snapshot_deploy_test", BROKER_ROOT / "deploy.py") + def snapshot_binding(_project, observed_container): + mounts = observed_container.get("Mounts") or [] + mount = mounts[0] if len(mounts) == 1 else {} + return { + "app_dir": "/srv/example/appdata/brain", + "data_path": mount.get("Source", ""), + "container_name": "projectbrain", + "image_ref": str(dict(observed_container.get("Config") or {}).get("Image") or ""), + "image_id": str(observed_container.get("Image") or ""), + } + + monkeypatch.setattr(deploy, "projectbrain_snapshot_binding", snapshot_binding) + monkeypatch.setattr(deploy, "projectbrain_rollback_environment", lambda *_a: {}) + environment = {"PROJECTBRAIN_OWNER_PASSWORD": "test-only-secret"} + monkeypatch.setattr(deploy, "static_projectbrain_environment", lambda _path: dict(environment)) + monkeypatch.setitem(sys.modules, "deploy", deploy) + run_process = subprocess.run + + def bounded_run(command, **kwargs): + if command[0] == "docker": + return subprocess.CompletedProcess(command, 0, stdout=image_id + "\n", stderr="") + return run_process(command, **kwargs) + + monkeypatch.setattr(controller.subprocess, "run", bounded_run) + return { + "project": project, + "state": state, + "receipt": receipt, + "container": container, + "jobs": jobs, + "environment": environment, + "original_path": original_path, + "original": original, + "original_bytes": original_path.read_bytes(), + "inbox": inbox, + "repository_path": repository_path, + "live_revision": live_revision, + "revision": revision, + "semantic": semantic, + "image_id": image_id, + "controller_release": controller_release, + } + + +@pytest.mark.parametrize("exact_source", ["omitted", False]) +def test_projectbrain_fixed_profile_accepts_only_the_existing_source_mode( + controller, monkeypatch, tmp_path, exact_source, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + project = dict(fixture["project"]) + if exact_source == "omitted": + project.pop("exact_source", None) + else: + project["exact_source"] = exact_source + + accepted = controller._projectbrain_delivery_project(project) + + assert accepted["_repository_name"] == REPOSITORY + assert accepted["source_dir"] == "/srv/example/appdata/brain" + if exact_source == "omitted": + assert "exact_source" not in accepted + else: + assert accepted["exact_source"] is False + + +@pytest.mark.parametrize("exact_source", [True, "false", 0, None]) +def test_projectbrain_fixed_profile_rejects_other_source_modes( + controller, monkeypatch, tmp_path, exact_source, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + project = {**fixture["project"], "exact_source": exact_source} + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_project(project) + assert failure.value.code == "projectbrain_target_contract_changed" + + +@pytest.mark.parametrize( + ("key", "value"), + [ + ("env_source", None), + ("env_source", False), + ("env_source", 0), + ("env_source", ""), + ("compose_dir", None), + ("compose_dir", False), + ("compose_dir", 0), + ("compose_dir", ""), + ], +) +def test_projectbrain_profile_requires_unconfigured_external_environment_and_compose_dir( + controller, monkeypatch, tmp_path, key, value, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + project = {**fixture["project"], key: value} + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_project(project) + assert failure.value.code == "projectbrain_target_contract_changed" + + +@pytest.mark.parametrize( + ("key", "value"), + [ + ("PROJECTBRAIN_DEPLOY_SKIP_PULL", "false"), + ("PROJECTBRAIN_DEPLOY_SKIP_PULL", False), + ("PROJECTBRAIN_DEPLOY_SKIP_PULL", 1), + ("PROJECTBRAIN_DEPLOY_REF", "main"), + ("PROJECTBRAIN_DEPLOY_REF", None), + ("PROJECTBRAIN_DEPLOY_REF", False), + ("PROJECTBRAIN_DEPLOY_REF", 0), + ("VITE_PROJECTBRAIN_COMMIT", "a" * 40), + ("VITE_PROJECTBRAIN_COMMIT", None), + ("VITE_PROJECTBRAIN_COMMIT", False), + ], +) +def test_projectbrain_profile_rejects_execution_override_drift( + controller, monkeypatch, tmp_path, key, value, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + project = { + **fixture["project"], + "environment": dict(fixture["project"]["environment"]), + } + project["environment"][key] = value + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_project(project) + assert failure.value.code == "projectbrain_target_contract_changed" + + +@pytest.mark.parametrize("deploy_ref", ["omitted", ""]) +def test_projectbrain_profile_accepts_only_absent_or_empty_deploy_ref( + controller, monkeypatch, tmp_path, deploy_ref, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + project = { + **fixture["project"], + "environment": dict(fixture["project"]["environment"]), + } + if deploy_ref == "omitted": + project["environment"].pop("PROJECTBRAIN_DEPLOY_REF") + else: + project["environment"]["PROJECTBRAIN_DEPLOY_REF"] = deploy_ref + + assert controller._projectbrain_delivery_project(project)["environment"] == ( + project["environment"] + ) + + +def test_actual_projectbrain_profile_passes_real_deploy_source_validation_and_binding( + controller, tmp_path, +): + repository = tmp_path / "repository" + repository.mkdir() + _git(repository, "init", "-b", "main") + _git(repository, "config", "user.email", "projectbrain-test@example.invalid") + _git(repository, "config", "user.name", "ProjectBrain test") + (repository / "scripts").mkdir() + (repository / "scripts" / "deploy-unraid-single.sh").write_text("#!/bin/sh\n", encoding="utf-8") + (repository / "docker-compose.yml").write_text("services: {}\n", encoding="utf-8") + _git(repository, "add", ".") + _git(repository, "commit", "-m", "live") + revision = _git(repository, "rev-parse", "HEAD") + project = { + "slug": "projectbrain", + "branch": "main", + "source_dir": "/srv/example/appdata/brain", + "deploy_script": "scripts/deploy-unraid-single.sh", + "archive_deploy_script": None, + "containers": ["projectbrain"], + "compose_project": "projectbrain", + "compose_files": ["docker-compose.yml"], + "gitea_repository_id": 7, + "environment": { + "PROJECTBRAIN_DEPLOY_SKIP_PULL": "true", + "PROJECTBRAIN_DEPLOY_REF": "", + "VITE_PROJECTBRAIN_COMMIT": "{revision}", + }, + } + deploy = _load_module( + "projectbrain_delivery_deploy_profile_test", + BROKER_ROOT / "deploy.py", + ) + deploy.validate_project_config(repository / ".git", project, revision) + + mount_source = str(tmp_path / "projectbrain-data") + container = { + "Config": {"Image": "projectbrain-unraid:codex-oidc-20260826"}, + "Image": "sha256:" + "a" * 64, + "Mounts": [{ + "Type": "bind", + "Source": mount_source, + "Destination": "/var/lib/projectbrain", + "RW": True, + }], + } + binding = deploy.projectbrain_snapshot_binding( + {**project, "_repository_name": REPOSITORY}, + container, + ) + + assert binding["app_dir"] == str(Path(project["source_dir"]).resolve()) + assert binding["data_path"] == mount_source + assert binding["image_id"] == container["Image"] + assert controller._projectbrain_delivery_project(project)["source_dir"] == project["source_dir"] + + exact_source_project = {**project, "exact_source": True} + with pytest.raises(deploy.DeployError, match="Exact source directory is outside"): + deploy.validate_project_config(repository / ".git", exact_source_project, revision) + + +def test_snapshot_binds_full_profile_target_script_controller_artifact_and_real_push_evidence( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + + first = controller._projectbrain_delivery_snapshot(fixture["revision"]) + fixture["project"]["health_timeout"] = 181 + second = controller._projectbrain_delivery_snapshot(fixture["revision"]) + fixture["controller_release"]["artifactSha256"] = "9" * 64 + third = controller._projectbrain_delivery_snapshot(fixture["revision"]) + + assert first["status"] == second["status"] == third["status"] == "ready" + assert first["_profileSha256"] != second["_profileSha256"] + assert first["preflightSha256"] != second["preflightSha256"] + assert second["_controllerArtifactSha256"] != third["_controllerArtifactSha256"] + assert second["preflightSha256"] != third["preflightSha256"] + assert first["_targetScriptSha256"] == controller._projectbrain_delivery_target_script_sha256( + fixture["repository_path"], fixture["revision"], + ) + assert first["_targetScriptSha256"] != controller._projectbrain_delivery_target_script_sha256( + fixture["repository_path"], fixture["live_revision"], + ) + assert first["_controllerArtifactSha256"] == "8" * 64 + assert first["validationRunId"] == 22470 + assert first["validationJobCount"] == 1 + assert "test-only-secret" not in json.dumps(controller._projectbrain_delivery_public_snapshot(first)) + + +def test_snapshot_rejects_target_revision_tracking_private_root_environment_file( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture( + controller, + monkeypatch, + tmp_path, + target_root_env=True, + ) + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_snapshot(fixture["revision"]) + + assert failure.value.code == "projectbrain_source_tracks_private_environment" + + +def test_snapshot_rejects_incomplete_full_job_identity_and_extra_required_context( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + adoption = sys.modules["registered_legacy_adoption"] + monkeypatch.setattr(adoption, "_validation_evidence", lambda *_a: { + "revision": fixture["revision"], + "branch": "main", + "event": "push", + "runId": 22470, + "runNumber": 381, + "runAttempt": 1, + "workflowPath": "managed-validation.yml@refs/heads/main", + "validationContext": "Managed validation / full (push)", + "jobCount": 1, + "jobIds": [], + }) + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_snapshot(fixture["revision"]) + assert failure.value.code == "projectbrain_full_push_validation_invalid" + + monkeypatch.setattr(adoption, "_validation_evidence", lambda *_args: { + "revision": fixture["revision"], + "branch": "main", + "event": "push", + "runId": 22470, + "runNumber": 381, + "runAttempt": 1, + "workflowPath": "managed-validation.yml@refs/heads/main", + "validationContext": "Managed validation / full (push)", + "jobCount": 1, + "jobIds": [39165], + }) + fixture["project"]["validation"]["contexts"].append("Extra required status (push)") + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_snapshot(fixture["revision"]) + assert failure.value.code == "projectbrain_validation_policy_unavailable" + + +@pytest.mark.parametrize( + ("change", "expected_code"), + [ + ("unhealthy", "projectbrain_runtime_unhealthy"), + ("mount", "projectbrain_runtime_binding_changed"), + ("receipt_image", "projectbrain_live_image_changed"), + ], +) +def test_snapshot_rejects_unhealthy_runtime_mount_or_unreceipted_image( + controller, monkeypatch, tmp_path, change, expected_code, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + if change == "unhealthy": + fixture["container"]["State"]["Health"]["Status"] = "unhealthy" + elif change == "mount": + fixture["container"]["Mounts"][0]["Source"] = "/srv/example/appdata/other" + else: + fixture["receipt"]["containers"]["projectbrain"]["image_id"] = "sha256:" + "f" * 64 + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_snapshot(fixture["revision"]) + assert failure.value.code == expected_code + + +def test_snapshot_binds_private_environment_and_rejects_deployment_history_in_original( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + first = controller._projectbrain_delivery_snapshot(fixture["revision"]) + fixture["environment"]["PROJECTBRAIN_OWNER_PASSWORD"] = "changed-test-only-secret" + second = controller._projectbrain_delivery_snapshot(fixture["revision"]) + assert first["preflightSha256"] != second["preflightSha256"] + assert "changed-test-only-secret" not in json.dumps(controller._projectbrain_delivery_public_snapshot(second)) + + changed_original = dict(fixture["original"]) + changed_original["audit"] = [ + *changed_original["audit"], + {"at": int(time.time()), "stage": "deployment_started"}, + ] + fixture["original_path"].write_bytes( + controller._projectbrain_delivery_canonical(changed_original) + b"\n", + ) + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_snapshot(fixture["revision"]) + assert failure.value.code == "projectbrain_original_delivery_not_revalidatable" + + +def test_snapshot_rejects_private_vite_revision_override_without_exposing_value( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + fixture["environment"]["VITE_PROJECTBRAIN_COMMIT"] = "test-only-private-value" + + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_delivery_snapshot(fixture["revision"]) + + assert failure.value.code == "projectbrain_private_environment_overrides_revision" + assert "test-only-private-value" not in str(failure.value) + + +def test_deploy_gate_binds_original_bytes_and_signed_sibling_without_rewriting_history( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + initial = controller._projectbrain_delivery_snapshot(fixture["revision"]) + idempotency_hash = hashlib.sha256(b"pb-revalidate-2026-10-07").hexdigest() + binding = controller._projectbrain_delivery_signed_binding(initial, idempotency_hash) + attempt = { + "schema": controller._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": fixture["semantic"], + "revalidation_of": fixture["semantic"], + "gitea_repository_id": 7, + "ref": "refs/heads/main", + "repository": REPOSITORY, + "revision": fixture["revision"], + "status": "deploying", + "projectbrain_delivery_revalidation": binding, + } + marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" + controller._write_new_inbox_record(marker, attempt) + + assert controller._projectbrain_delivery_deploy_gate( + REPOSITORY, fixture["revision"], initial["preflightSha256"], + ) + assert fixture["original_path"].read_bytes() == fixture["original_bytes"] + + changed_original = dict(fixture["original"]) + changed_original["operator_note"] = "whole-record drift" + fixture["original_path"].write_bytes( + controller._projectbrain_delivery_canonical(changed_original) + b"\n", + ) + assert not controller._projectbrain_delivery_deploy_gate( + REPOSITORY, fixture["revision"], initial["preflightSha256"], + ) + + fixture["original_path"].write_bytes(fixture["original_bytes"]) + fixture["original_path"].write_bytes(fixture["original_bytes"] + b" ") + with pytest.raises(controller.ProjectBrainDeliveryRefused): + controller._projectbrain_read_original_delivery(fixture["semantic"]) + + fixture["original_path"].write_bytes(fixture["original_bytes"]) + attempt["projectbrain_delivery_revalidation"] = {**binding, "hmac": "0" * 64} + marker.write_bytes(controller._projectbrain_delivery_canonical(attempt) + b"\n") + assert not controller._projectbrain_delivery_deploy_gate( + REPOSITORY, fixture["revision"], initial["preflightSha256"], + ) + + +def test_same_idempotency_key_replays_progressed_marker_and_changed_key_conflicts( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + preflight = snapshot() + preflight["_semanticDigest"] = fixture["semantic"] + preflight["revision"] = fixture["revision"] + key = "pb-revalidate-2026-10-07" + key_hash = hashlib.sha256(key.encode()).hexdigest() + binding = controller._projectbrain_delivery_signed_binding(preflight, key_hash) + record = { + "schema": controller._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": fixture["semantic"], + "revalidation_of": fixture["semantic"], + "gitea_repository_id": 7, + "ref": "refs/heads/main", + "repository": REPOSITORY, + "revision": fixture["revision"], + "status": "deployed", + "projectbrain_delivery_revalidation": binding, + } + marker = fixture["inbox"] / f"{fixture['semantic']}.revalidation.json" + controller._write_new_inbox_record(marker, record) + + assert controller._projectbrain_existing_revalidation( + fixture["revision"], key_hash, PREFLIGHT_SHA, + ) == {"status": "deployed", "delivery": fixture["semantic"]} + with pytest.raises(controller.ProjectBrainDeliveryRefused) as failure: + controller._projectbrain_existing_revalidation( + fixture["revision"], hashlib.sha256(b"another-key-2026").hexdigest(), PREFLIGHT_SHA, + ) + assert failure.value.code == "projectbrain_idempotency_conflict" + + +def test_same_key_replay_returns_the_existing_deploying_phase(controller, monkeypatch, tmp_path): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + preflight = snapshot() + preflight["_semanticDigest"] = fixture["semantic"] + preflight["revision"] = fixture["revision"] + key = "pb-revalidate-active-2026" + key_hash = hashlib.sha256(key.encode()).hexdigest() + binding = controller._projectbrain_delivery_signed_binding(preflight, key_hash) + record = { + "schema": controller._WEBHOOK_REVALIDATION_SCHEMA, + "semantic_sha256": fixture["semantic"], + "revalidation_of": fixture["semantic"], + "gitea_repository_id": 7, + "ref": "refs/heads/main", + "repository": REPOSITORY, + "revision": fixture["revision"], + "status": "deploying", + "projectbrain_delivery_revalidation": binding, + } + controller._write_new_inbox_record( + fixture["inbox"] / f"{fixture['semantic']}.revalidation.json", record, + ) + + assert controller._projectbrain_existing_revalidation( + fixture["revision"], key_hash, PREFLIGHT_SHA, + ) == {"status": "deploying", "delivery": fixture["semantic"]} + + +def test_concurrent_enqueue_creates_one_additive_attempt_and_preserves_original( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + preflight = controller._projectbrain_delivery_snapshot(fixture["revision"]) + binding = controller._projectbrain_delivery_signed_binding( + preflight, hashlib.sha256(b"pb-revalidate-concurrent-2026").hexdigest(), + ) + project = fixture["project"] + + def enqueue(): + return controller._enqueue_terminal_revalidation( + REPOSITORY, + fixture["revision"], + project, + expected_original_sha256=preflight["_originalFileSha256"], + projectbrain_delivery_binding=binding, + ) + + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(lambda _index: enqueue(), range(2))) + + assert results == [("awaiting_validation", fixture["semantic"])] * 2 + assert fixture["original_path"].read_bytes() == fixture["original_bytes"] + attempt_files = list(fixture["inbox"].glob("*.revalidation.json")) + assert len(attempt_files) == 1 + record = controller._read_inbox_record(attempt_files[0]) + assert record["projectbrain_delivery_revalidation"] == binding + + +def test_generic_terminal_revalidation_cannot_occupy_projectbrain_sibling( + controller, monkeypatch, tmp_path, +): + fixture = _snapshot_fixture(controller, monkeypatch, tmp_path) + + result = controller._enqueue_terminal_revalidation( + REPOSITORY, + fixture["revision"], + fixture["project"], + ) + + assert result == ("invalid", "") + assert fixture["original_path"].read_bytes() == fixture["original_bytes"] + assert not (fixture["inbox"] / f"{fixture['semantic']}.revalidation.json").exists() + + +def test_deploy_rejection_precedes_source_sync_under_project_and_global_locks( + monkeypatch, tmp_path, +): + deploy = _load_module("projectbrain_delivery_deploy_gate_test", BROKER_ROOT / "deploy.py") + revision = "a" * 40 + project = { + "slug": "projectbrain", + "branch": "main", + "source_dir": str(tmp_path / "source"), + "exact_source": True, + "deploy_script": "scripts/deploy-unraid-single.sh", + "containers": ["projectbrain"], + "compose_project": "projectbrain", + "compose_files": ["docker-compose.yml"], + } + active_locks = [] + project_lock = deploy.STATE_ROOT / "locks" / "projectbrain.lock" + + @contextmanager + def lock(path, _description): + active_locks.append(path) + try: + yield + finally: + active_locks.pop() + + monkeypatch.setattr(deploy, "exclusive_lock", lock) + monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: lock( + deploy.GLOBAL_DEPLOY_LOCK, "global", + )) + monkeypatch.setattr(deploy, "load_projects", lambda: {REPOSITORY: dict(project)}) + monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, config: config) + monkeypatch.setattr(deploy, "repository_path", lambda *_a: tmp_path) + monkeypatch.setattr(deploy, "bind_appops_self_upgrade_recovery", lambda *_a: None) + monkeypatch.setattr(deploy, "bind_registered_legacy_request", lambda *_a: None) + monkeypatch.setattr(deploy, "validate_commit", lambda *_a: revision) + monkeypatch.setattr(deploy, "deployment_metadata", lambda *_a: {"revision": revision}) + monkeypatch.setattr(deploy, "validate_project_config", lambda *_a: None) + monkeypatch.setattr( + deploy, + "sync_source", + lambda *_a, **_k: pytest.fail("source sync must not run after a stale preflight"), + ) + monkeypatch.setattr( + deploy, + "copy_environment", + lambda *_a, **_k: pytest.fail("private environment copy must not run after a stale preflight"), + ) + gate_calls = [] + + def gate(_repository, _revision, _fingerprint): + gate_calls.append(list(active_locks)) + return False + + monkeypatch.setitem(sys.modules, "status_server", SimpleNamespace( + _projectbrain_delivery_deploy_gate=gate, + )) + monkeypatch.setenv(deploy.PROJECTBRAIN_DELIVERY_PREFLIGHT_ENV, "b" * 64) + monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", REPOSITORY, revision]) + + with pytest.raises(deploy.DeployError, match="preflight changed before deployment"): + deploy.main() + + assert gate_calls == [[project_lock, deploy.GLOBAL_DEPLOY_LOCK]] + + +@pytest.mark.parametrize("restore_fails", [False, True]) +def test_second_projectbrain_preflight_failure_restores_tracked_source_and_records_live_state( + monkeypatch, tmp_path, restore_fails, +): + deploy = _load_module("projectbrain_delivery_second_gate_test", BROKER_ROOT / "deploy.py") + repository_path = tmp_path / "repository" + repository_path.mkdir() + _git(repository_path, "init", "-b", "main") + _git(repository_path, "config", "user.email", "projectbrain-test@example.invalid") + _git(repository_path, "config", "user.name", "ProjectBrain test") + source_path = repository_path / "README.md" + source_path.write_text("prior live source\n", encoding="utf-8") + _git(repository_path, "add", ".") + _git(repository_path, "commit", "-m", "prior") + prior_revision = _git(repository_path, "rev-parse", "HEAD") + source_path.write_text("candidate source\n", encoding="utf-8") + _git(repository_path, "add", ".") + _git(repository_path, "commit", "-m", "candidate") + target_revision = _git(repository_path, "rev-parse", "HEAD") + repository_git_dir = repository_path / ".git" + source_dir = tmp_path / "source" + deploy.sync_source(repository_git_dir, prior_revision, source_dir, "projectbrain") + private_env_path = source_dir / ".env" + private_env_path.write_text("PROJECTBRAIN_PRIVATE_TEST=fixture-only\n", encoding="utf-8") + private_env_before = private_env_path.read_bytes() + + project = { + "slug": "projectbrain", + "branch": "main", + "source_dir": str(source_dir), + "deploy_script": "scripts/deploy-unraid-single.sh", + "containers": ["projectbrain"], + "compose_project": "projectbrain", + "compose_files": ["docker-compose.yml"], + } + previous_state = { + "schema_version": 3, + "repository": REPOSITORY, + "live_revision": prior_revision, + "live_status": "healthy", + "live_legacy_provenance": False, + "live_artifacts": {"containers": {"projectbrain": {"image_id": "sha256:" + "a" * 64}}}, + "live_receipt": {"revision": prior_revision, "testOnly": True}, + } + active_locks = [] + project_lock = deploy.STATE_ROOT / "locks" / "projectbrain.lock" + + @contextmanager + def lock(path, _description): + active_locks.append(path) + try: + yield + finally: + active_locks.pop() + + monkeypatch.setattr(deploy, "exclusive_lock", lock) + monkeypatch.setattr(deploy, "exclusive_global_deploy_lock", lambda: lock( + deploy.GLOBAL_DEPLOY_LOCK, "global", + )) + monkeypatch.setattr(deploy, "load_projects", lambda: {REPOSITORY: dict(project)}) + monkeypatch.setattr(deploy, "effective_mcp_hub_project", lambda _repo, config: config) + monkeypatch.setattr(deploy, "repository_path", lambda *_a: repository_git_dir) + monkeypatch.setattr(deploy, "repository_path_from_project", lambda _project: repository_git_dir) + monkeypatch.setattr(deploy, "bind_appops_self_upgrade_recovery", lambda *_a: None) + monkeypatch.setattr(deploy, "bind_registered_legacy_request", lambda *_a: None) + monkeypatch.setattr(deploy, "mcp_hub_prepare_current_receipt_model", lambda *_a: None) + monkeypatch.setattr(deploy, "validate_commit", lambda *_a: target_revision) + monkeypatch.setattr(deploy, "deployment_metadata", lambda *_a: { + "revision": target_revision, + "tree": "f" * 40, + }) + monkeypatch.setattr(deploy, "validate_project_config", lambda *_a: None) + monkeypatch.setattr(deploy, "read_state", lambda _project: dict(previous_state)) + monkeypatch.setattr(deploy, "verified_current_receipt", lambda *_a: None) + monkeypatch.setattr(deploy, "validate_forward_revision", lambda *_a: None) + monkeypatch.setattr(deploy, "_mcp_hub_assert_no_unresolved_migration_intent", lambda *_a: None) + + source_syncs = [] + actual_sync_source = deploy.sync_source + + def record_sync(repo, revision, target, slug, clean_untracked_paths=None, *, exact=False): + source_syncs.append((revision, tuple(active_locks))) + if restore_fails and revision == prior_revision: + raise deploy.DeployError("fixture source restore failure") + return actual_sync_source( + repo, revision, target, slug, clean_untracked_paths, exact=exact, + ) + + monkeypatch.setattr(deploy, "sync_source", record_sync) + environment_copies = [] + actual_copy_environment = deploy.copy_environment + + def record_copy_environment(config, target): + environment_copies.append((tuple(active_locks), config.get("env_source"))) + return actual_copy_environment(config, target) + + monkeypatch.setattr(deploy, "copy_environment", record_copy_environment) + + gate_calls = [] + gate_results = iter((True, False)) + + def gate(_repository, _revision, _fingerprint): + gate_calls.append(tuple(active_locks)) + return next(gate_results) + + monkeypatch.setitem(sys.modules, "status_server", SimpleNamespace( + _projectbrain_delivery_deploy_gate=gate, + )) + state_updates = [] + monkeypatch.setattr( + deploy, + "write_state", + lambda repository, config, revision, status, message, **kwargs: state_updates.append({ + "repository": repository, + "project": config, + "revision": revision, + "status": status, + "message": message, + **kwargs, + }), + ) + monkeypatch.setattr( + deploy, + "capture_compose_rollback", + lambda *_a: pytest.fail("production snapshot must not run after a stale second gate"), + ) + monkeypatch.setattr( + deploy, + "deploy_project", + lambda *_a: pytest.fail("production mutation must not run after a stale second gate"), + ) + monkeypatch.setenv(deploy.PROJECTBRAIN_DELIVERY_PREFLIGHT_ENV, "b" * 64) + monkeypatch.setattr(sys, "argv", ["deploy.py", "deploy", REPOSITORY, target_revision]) + + with pytest.raises(deploy.DeployError, match="preflight changed before deployment"): + deploy.main() + + assert len(gate_calls) == 2 + assert gate_calls == [ + (project_lock, deploy.GLOBAL_DEPLOY_LOCK), + (project_lock, deploy.GLOBAL_DEPLOY_LOCK), + ] + assert source_syncs == [ + (target_revision, (project_lock,)), + (prior_revision, (project_lock, deploy.GLOBAL_DEPLOY_LOCK)), + ] + assert environment_copies == [((project_lock,), None)] + ( + [] if restore_fails else [((project_lock, deploy.GLOBAL_DEPLOY_LOCK), None)] + ) + assert source_dir.joinpath("README.md").read_text(encoding="utf-8") == ( + "candidate source\n" if restore_fails else "prior live source\n" + ) + assert private_env_path.read_bytes() == private_env_before + assert len(state_updates) == 1 + state = state_updates[0] + assert state["status"] == "failed" + assert state["live_revision"] == prior_revision + assert state["live_artifacts"] == previous_state["live_artifacts"] + assert state["live_receipt"] == previous_state["live_receipt"] + assert state["live_status"] == ("unknown" if restore_fails else "healthy") + if restore_fails: + assert "source rollback error: fixture source restore failure" in state["message"] + + +def test_snapshot_lock_is_nonblocking_and_uses_project_then_global_flock( + controller, monkeypatch, tmp_path, +): + project_lock = tmp_path / "project.lock" + global_lock = tmp_path / "global.lock" + controller.PROJECTBRAIN_DELIVERY_PROJECT_LOCK_PATH = project_lock + controller.PROJECTBRAIN_DELIVERY_LOCK_PATH = global_lock + descriptors = {11: project_lock, 12: global_lock} + opened = [] + acquired = [] + released = [] + closed = [] + busy = False + + fake_os = SimpleNamespace( + name="posix", + O_RDWR=1, + O_NOFOLLOW=2, + open=lambda path, _flags: _open_fake(path, descriptors, opened), + fstat=lambda _descriptor: SimpleNamespace( + st_mode=0o100600, + st_nlink=1, + st_uid=0, + ), + geteuid=lambda: 0, + close=lambda descriptor: closed.append(descriptor), + ) + + def flock(descriptor, flags): + if flags == 6 and busy and descriptors[descriptor] == global_lock: + raise BlockingIOError + if flags == 6: + acquired.append(descriptors[descriptor]) + elif flags == 8: + released.append(descriptors[descriptor]) + + monkeypatch.setattr(controller, "os", fake_os) + monkeypatch.setitem(sys.modules, "fcntl", SimpleNamespace( + LOCK_EX=2, + LOCK_NB=4, + LOCK_UN=8, + flock=flock, + )) + + with controller._projectbrain_delivery_mutation_lock(), pytest.raises( + controller.ProjectBrainDeliveryRefused, + ) as thread_busy: + controller._projectbrain_delivery_mutation_lock().__enter__() + assert thread_busy.value.code == "projectbrain_delivery_operation_busy" + assert opened == [project_lock, global_lock] + assert acquired == [project_lock, global_lock] + assert released == [global_lock, project_lock] + assert closed == [12, 11] + + busy = True + opened.clear() + acquired.clear() + released.clear() + closed.clear() + with pytest.raises(controller.ProjectBrainDeliveryRefused) as process_busy: + controller._projectbrain_delivery_mutation_lock().__enter__() + assert process_busy.value.code == "projectbrain_deployment_active" + assert opened == [project_lock, global_lock] + assert acquired == [project_lock] + assert released == [project_lock] + assert closed == [12, 11] + + +def test_deployment_broker_client_and_mcp_tool_keep_fixed_projectbrain_contract(monkeypatch): + from backend.app import mcp_server + from backend.app.services.deployment_broker import DeploymentBrokerClient, settings + + monkeypatch.setattr(settings, "mcp_token", "t" * 48) + client = DeploymentBrokerClient(base_url="https://controller.example", verify_tls=True) + requests = [] + monkeypatch.setattr(client, "_request", lambda *args, **kwargs: requests.append((args, kwargs)) or {"status": "ready"}) + + assert client.projectbrain_delivery_operation("preflight", {"revision": REVISION}) == {"status": "ready"} + assert requests[-1][0][:2] == ("POST", PREFIX + "preflight") + assert requests[-1][1]["payload"] == {"revision": REVISION} + revalidate_payload = { + "revision": REVISION, + "expectedPreflightSha256": PREFLIGHT_SHA, + "idempotencyKey": "pb-revalidate-2026-10-07", + } + assert client.projectbrain_delivery_operation("revalidate", revalidate_payload) == {"status": "ready"} + assert requests[-1][0][:2] == ("POST", PREFIX + "revalidate") + assert requests[-1][1]["payload"] == revalidate_payload + with pytest.raises(ValueError): + client.projectbrain_delivery_operation("preflight", {"revision": REVISION, "repository": REPOSITORY}) + + tool_calls = [] + monkeypatch.setattr( + mcp_server, + "DeploymentBrokerClient", + lambda: SimpleNamespace(projectbrain_delivery_operation=lambda stage, payload: tool_calls.append((stage, payload)) or {"status": "ready"}), + ) + assert mcp_server.appops_projectbrain_delivery_preflight(REVISION) == {"status": "ready"} + with pytest.raises(ValueError): + mcp_server.appops_projectbrain_delivery_revalidate( + REVISION, PREFLIGHT_SHA, "short", + ) + assert mcp_server.appops_projectbrain_delivery_revalidate( + REVISION, PREFLIGHT_SHA, "pb-revalidate-2026-10-07", + ) == {"status": "ready"} + assert tool_calls == [ + ("preflight", {"revision": REVISION}), + ("revalidate", { + "revision": REVISION, + "expectedPreflightSha256": PREFLIGHT_SHA, + "idempotencyKey": "pb-revalidate-2026-10-07", + }), + ] + + +def _open_fake(path, descriptor_paths, opened): + descriptor = 11 if not opened else 12 + opened.append(Path(path)) + descriptor_paths[descriptor] = Path(path) + return descriptor diff --git a/backend/tests/test_runner_fleet_scheduler_recovery.py b/backend/tests/test_runner_fleet_scheduler_recovery.py index 36dd021..605ed25 100644 --- a/backend/tests/test_runner_fleet_scheduler_recovery.py +++ b/backend/tests/test_runner_fleet_scheduler_recovery.py @@ -5,6 +5,7 @@ import sys import types from datetime import UTC, datetime from pathlib import Path +from threading import Event, Thread, current_thread import pytest @@ -107,6 +108,45 @@ def _healthy_snapshot(): +def _jobs_page(jobs=()): + rows = list(jobs) + return {"total_count": len(rows), "jobs": rows} + + +def _scheduler_api(state, error_type): + def fake_gitea(path, params=None): + params = params or {} + if path.endswith("/runners"): + runners = state.get( + "runners", + [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": False, + "disabled": False, + "labels": ["linux-validation"], + } + ], + ) + return {"total_count": len(runners), "runners": runners} + status = params.get("status") + if status == state.get("error_status"): + raise error_type("runner_fleet_gitea_test_error") + rows = state.get( + {"queued": "queued", "pending": "pending", "in_progress": "active"}.get( + status, "" + ), + [], + ) + if status == state.get("incomplete_status"): + return {"total_count": len(rows) + 1, "jobs": rows} + return _jobs_page(rows) + + return fake_gitea + + def test_container_projection_includes_restart_policy(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() payload = [ @@ -780,9 +820,11 @@ def test_scheduler_candidate_uses_live_gitea_queue_and_fixed_label_target( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() - now = 1_800_000_000 - created_at = datetime.fromtimestamp(now - 900, UTC).isoformat() - monkeypatch.setattr(fleet_module.time, "time", lambda: now) + monotonic = [1_000.0] + created_at = datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat() + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) def fake_gitea(path, params=None): params = params or {} @@ -801,33 +843,533 @@ def test_scheduler_candidate_uses_live_gitea_queue_and_fixed_label_target( ] } if params.get("status") == "queued": - return { - "jobs": [ + return _jobs_page( + [ { "id": 22931, + "status": "queued", + "run_id": 3001, + "attempt": 1, "created_at": created_at, "labels": ["linux-validation"], } ] - } - return {"jobs": []} + ) + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) - assert service._scheduler_recovery_candidate() == { + # An old created_at is not evidence of an old runnable/claimable interval. + assert service._scheduler_recovery_candidate() is None + candidate = None + for _ in range(fleet_module.SCHEDULER_STALL_SECONDS // 60): + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + + assert candidate == { "target": "itworx-validation-runner-2", "jobId": 22931, "label": "linux-validation", - "ageSeconds": 900, + "runnerId": 17, + "labels": ("linux-validation",), "reason": "stale_scheduler_assignment", + "runIdentity": (("run_id", 3001), ("attempt", 1)), + "ageSeconds": fleet_module.SCHEDULER_STALL_SECONDS, } +def test_pending_age_is_not_carried_into_newly_queued_claimability( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + monotonic = [1_000.0] + created_at = datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat() + pending_job = { + "id": 22931, + "status": "requested", + "created_at": created_at, + "labels": ["linux-validation"], + } + queued_job = {**pending_job, "status": "queued"} + state = {"pending": [pending_job], "queued": [], "active": []} + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + + assert service._scheduler_recovery_candidate() is None + state["pending"] = [] + state["queued"] = [queued_job] + monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS + # The first queued observation starts the timer even though created_at is old. + assert service._scheduler_recovery_candidate() is None + for _ in range(fleet_module.SCHEDULER_STALL_SECONDS // 60): + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + + assert candidate is not None + assert candidate["jobId"] == 22931 + assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS + + +@pytest.mark.parametrize("uncertainty", ["error", "truncated"]) +def test_scheduler_claimability_resets_after_unknown_or_incomplete_observation( + fleet_module, monkeypatch, uncertainty +): + service = fleet_module.RunnerFleetService() + monotonic = [2_000.0] + old_job = { + "id": 22931, + "status": "queued", + "created_at": datetime.fromtimestamp(1_800_000_000 - 7_200, UTC).isoformat(), + "labels": ["linux-validation"], + } + state = {"queued": [old_job], "pending": [], "active": []} + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + assert service._scheduler_recovery_candidate() is None + + monotonic[0] += 60 + if uncertainty == "error": + state["error_status"] = "pending" + else: + state["incomplete_status"] = "queued" + with pytest.raises(fleet_module.RunnerFleetError): + service._scheduler_recovery_candidate() + + state.pop("error_status", None) + state.pop("incomplete_status", None) + monotonic[0] += 60 + assert service._scheduler_recovery_candidate() is None + for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 2): + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + # The pre-error timer would have matured at this instant if unknown time + # had been bridged instead of starting over at the first fresh sample. + assert candidate is None + for _ in range(2): + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + assert candidate is not None + assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS + + +@pytest.mark.parametrize("loss", ["busy", "active"]) +def test_scheduler_claimability_resets_when_target_becomes_unavailable( + fleet_module, monkeypatch, loss +): + service = fleet_module.RunnerFleetService() + monotonic = [3_000.0] + job = { + "id": 22931, + "status": "queued", + "labels": ["linux-validation"], + } + state = {"queued": [job], "pending": [], "active": []} + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + assert service._scheduler_recovery_candidate() is None + + monotonic[0] += 60 + if loss == "busy": + state["runners"] = [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": True, + "disabled": False, + "labels": ["linux-validation"], + } + ] + else: + state["active"] = [ + {"id": 22932, "status": "in_progress", "runner_id": 17} + ] + assert service._scheduler_recovery_candidate() is None + + state.pop("runners", None) + state["active"] = [] + monotonic[0] += 60 + assert service._scheduler_recovery_candidate() is None + for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 1): + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + assert candidate is None + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + assert candidate is not None + assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS + + +def test_scheduler_claimability_gap_restarts_the_observation_window( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + monotonic = [4_000.0] + state = { + "queued": [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}], + "pending": [], + "active": [], + } + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + assert service._scheduler_recovery_candidate() is None + + monotonic[0] += fleet_module.SCHEDULER_OBSERVATION_MAX_GAP_SECONDS + 1 + assert service._scheduler_recovery_candidate() is None + monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS + # A second long gap resets again instead of treating elapsed wall time as observed. + assert service._scheduler_recovery_candidate() is None + + +def test_older_eligible_sample_cannot_repopulate_after_newer_busy_sample( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + monotonic = [1_000.0] + queued_job = { + "id": 22931, + "status": "queued", + "run_id": 3001, + "attempt": 1, + "labels": ["linux-validation"], + } + idle_state = {"queued": [queued_job], "pending": [], "active": []} + live_state = {"queued": [queued_job], "pending": [], "active": []} + stale_request = _scheduler_api(idle_state, fleet_module.RunnerFleetError) + live_request = _scheduler_api(live_state, fleet_module.RunnerFleetError) + older_sample_entered = Event() + release_older_sample = Event() + older_result: dict[str, object] = {} + + def fake_gitea(path, params=None): + if current_thread().name == "older-idle-sample": + if not older_sample_entered.is_set(): + older_sample_entered.set() + assert release_older_sample.wait(timeout=5) + return stale_request(path, params) + return live_request(path, params) + + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr(service, "_gitea_request", fake_gitea) + assert service._scheduler_recovery_candidate() is None + + monotonic[0] = 1_010.0 + older_sample = Thread( + name="older-idle-sample", + target=lambda: older_result.setdefault( + "candidate", service._scheduler_recovery_candidate() + ), + ) + older_sample.start() + assert older_sample_entered.wait(timeout=5) + + monotonic[0] = 1_020.0 + live_state["runners"] = [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": True, + "disabled": False, + "labels": ["linux-validation"], + } + ] + assert service._scheduler_recovery_candidate() is None + + monotonic[0] = 1_030.0 + release_older_sample.set() + older_sample.join(timeout=5) + assert not older_sample.is_alive() + assert older_result["candidate"] is None + + live_state.pop("runners") + monotonic[0] = 1_031.0 + assert service._scheduler_recovery_candidate() is None + for _ in range(9): + monotonic[0] += 60 + candidate = service._scheduler_recovery_candidate() + + # The rejected stale sample must not have started the ten-minute dwell. + assert monotonic[0] == 1_571.0 + assert candidate is None + monotonic[0] += 59 + assert service._scheduler_recovery_candidate() is None + monotonic[0] += 1 + candidate = service._scheduler_recovery_candidate() + assert candidate is not None + assert candidate["ageSeconds"] == fleet_module.SCHEDULER_STALL_SECONDS + + +def test_scheduler_claimability_is_process_local_and_lost_on_restart( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + monotonic = [5_000.0] + state = { + "queued": [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}], + "pending": [], + "active": [], + } + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + assert service._scheduler_recovery_candidate() is None + + fleet_module.RunnerFleetService._clear_scheduler_claimability() + monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS + assert service._scheduler_recovery_candidate() is None + + +@pytest.mark.parametrize("identity_change", ["origin", "job", "run", "labels", "runner"]) +def test_scheduler_claimability_restarts_when_job_or_runner_identity_changes( + fleet_module, monkeypatch, identity_change +): + service = fleet_module.RunnerFleetService() + monotonic = [6_000.0] + job = { + "id": 22931, + "run_id": 3001, + "attempt": 1, + "status": "queued", + "labels": ["linux-validation"], + } + state = {"queued": [job], "pending": [], "active": []} + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + assert service._scheduler_recovery_candidate() is None + for _ in range((fleet_module.SCHEDULER_STALL_SECONDS // 60) - 1): + monotonic[0] += 60 + assert service._scheduler_recovery_candidate() is None + + if identity_change == "origin": + monkeypatch.setattr( + fleet_module, "GITEA_API_BASE", "http://other-gitea:3000" + ) + elif identity_change == "job": + job["id"] = 22932 + elif identity_change == "run": + job["run_id"] = 3002 + elif identity_change == "labels": + job["labels"] = ["ubuntu-latest"] + state["runners"] = [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": False, + "disabled": False, + "labels": ["linux-validation", "ubuntu-latest"], + } + ] + else: + state["runners"] = [ + { + "id": 18, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": False, + "disabled": False, + "labels": ["linux-validation"], + } + ] + monotonic[0] += 60 + + assert service._scheduler_recovery_candidate() is None + + +def test_scheduler_candidate_pages_jobs_at_gitea_fifty_row_limit( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + queued = [ + {"id": job_id, "status": "queued", "labels": ["unmapped-label"]} + for job_id in range(1, 51) + ] + queued.append( + {"id": 51, "status": "queued", "labels": ["linux-validation"]} + ) + job_calls: list[dict[str, object]] = [] + + def fake_gitea(path, params=None): + params = params or {} + if path.endswith("/runners"): + return { + "total_count": 1, + "runners": [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": False, + "disabled": False, + "labels": ["linux-validation"], + } + ], + } + status = params["status"] + if status == "queued": + job_calls.append(params) + start = (int(params["page"]) - 1) * 50 + return {"total_count": 51, "jobs": queued[start : start + 50]} + return _jobs_page() + + monkeypatch.setattr(service, "_gitea_request", fake_gitea) + + assert service._scheduler_recovery_candidate() is None + assert [(call["page"], call["limit"]) for call in job_calls] == [ + (1, fleet_module.GITEA_JOB_PAGE_LIMIT), + (2, fleet_module.GITEA_JOB_PAGE_LIMIT), + ] + + +def test_scheduler_candidate_rejects_job_inventory_over_one_thousand_rows( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + + def fake_gitea(path, params=None): + if path.endswith("/runners"): + return {"total_count": 0, "runners": []} + return {"total_count": 1_001, "jobs": []} + + monkeypatch.setattr(service, "_gitea_request", fake_gitea) + + with pytest.raises( + fleet_module.RunnerFleetError, + match="runner_fleet_gitea_invalid_job_inventory", + ): + service._scheduler_recovery_candidate() + + +def test_scheduler_candidate_fails_closed_when_job_appears_in_two_statuses( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + monotonic = [7_000.0] + job = {"id": 22931, "labels": ["linux-validation"]} + state = { + "queued": [{**job, "status": "queued"}], + "pending": [{**job, "status": "requested"}], + "active": [], + } + monkeypatch.setattr( + fleet_module, "_scheduler_monotonic_now", lambda: monotonic[0] + ) + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + + with pytest.raises( + fleet_module.RunnerFleetError, + match="runner_fleet_gitea_job_status_changed_during_read", + ): + service._scheduler_recovery_candidate() + state["pending"] = [] + monotonic[0] += fleet_module.SCHEDULER_STALL_SECONDS + assert service._scheduler_recovery_candidate() is None + + +def test_scheduler_candidate_does_not_classify_offline_target_as_claimable( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + state = { + "queued": [ + { + "id": 22931, + "status": "queued", + "created_at": datetime.fromtimestamp( + 1_800_000_000 - 7_200, UTC + ).isoformat(), + "labels": ["linux-validation"], + } + ], + "pending": [], + "active": [], + "runners": [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "offline", + "busy": False, + "disabled": False, + "labels": ["linux-validation"], + } + ], + } + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + + assert service._scheduler_recovery_candidate() is None + + +@pytest.mark.parametrize( + ("label_source", "expected_error"), + [ + ("job", "runner_fleet_gitea_invalid_queued_job"), + ("runner", "runner_fleet_gitea_invalid_runner_inventory"), + ], +) +def test_scheduler_candidate_rejects_labels_that_would_change_when_trimmed( + fleet_module, monkeypatch, label_source, expected_error +): + service = fleet_module.RunnerFleetService() + state = { + "queued": [ + {"id": 22931, "status": "queued", "labels": ["linux-validation"]} + ], + "pending": [], + "active": [], + } + if label_source == "job": + state["queued"][0]["labels"] = [" linux-validation"] + else: + state["runners"] = [ + { + "id": 17, + "name": "itworx-validation-runner-2", + "status": "online", + "busy": False, + "disabled": False, + "labels": ["linux-validation "], + } + ] + monkeypatch.setattr( + service, "_gitea_request", _scheduler_api(state, fleet_module.RunnerFleetError) + ) + + with pytest.raises(fleet_module.RunnerFleetError, match=expected_error): + service._scheduler_recovery_candidate() + + def test_scheduler_candidate_never_targets_busy_runner(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() - now = 1_800_000_000 - created_at = datetime.fromtimestamp(now - 900, UTC).isoformat() - monkeypatch.setattr(fleet_module.time, "time", lambda: now) def fake_gitea(path, params=None): params = params or {} @@ -846,16 +1388,10 @@ def test_scheduler_candidate_never_targets_busy_runner(fleet_module, monkeypatch ] } if params.get("status") == "queued": - return { - "jobs": [ - { - "id": 22931, - "created_at": created_at, - "labels": ["linux-validation"], - } - ] - } - return {"jobs": []} + return _jobs_page( + [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] + ) + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) @@ -866,9 +1402,6 @@ def test_scheduler_candidate_never_restarts_runner_with_in_progress_job( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() - now = 1_800_000_000 - created_at = datetime.fromtimestamp(now - 900, UTC).isoformat() - monkeypatch.setattr(fleet_module.time, "time", lambda: now) def fake_gitea(path, params=None): params = params or {} @@ -885,10 +1418,14 @@ def test_scheduler_candidate_never_restarts_runner_with_in_progress_job( }], } if params.get("status") == "queued": - return {"jobs": [{"id": 22931, "created_at": created_at, "labels": ["linux-validation"]}]} + return _jobs_page( + [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] + ) if params.get("status") == "in_progress": - return {"jobs": [{"id": 22932, "runner_id": 17, "status": "in_progress"}]} - return {"jobs": []} + return _jobs_page( + [{"id": 22932, "runner_id": 17, "status": "in_progress"}] + ) + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) @@ -897,9 +1434,6 @@ def test_scheduler_candidate_never_restarts_runner_with_in_progress_job( def test_scheduler_candidate_never_hides_disabled_busy_runner(fleet_module, monkeypatch): service = fleet_module.RunnerFleetService() - now = 1_800_000_000 - created_at = datetime.fromtimestamp(now - 900, UTC).isoformat() - monkeypatch.setattr(fleet_module.time, "time", lambda: now) def fake_gitea(path, params=None): params = params or {} @@ -920,16 +1454,10 @@ def test_scheduler_candidate_never_hides_disabled_busy_runner(fleet_module, monk ] } if params.get("status") == "queued": - return { - "jobs": [ - { - "id": 22931, - "created_at": created_at, - "labels": ["linux-validation"], - } - ] - } - return {"jobs": []} + return _jobs_page( + [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] + ) + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) @@ -940,9 +1468,6 @@ def test_scheduler_candidate_reads_later_runner_inventory_pages( fleet_module, monkeypatch ): service = fleet_module.RunnerFleetService() - now = 1_800_000_000 - created_at = datetime.fromtimestamp(now - 900, UTC).isoformat() - monkeypatch.setattr(fleet_module.time, "time", lambda: now) runner_calls: list[dict[str, object]] = [] def fake_gitea(path, params=None): @@ -977,16 +1502,10 @@ def test_scheduler_candidate_reads_later_runner_inventory_pages( ], } if params.get("status") == "queued": - return { - "jobs": [ - { - "id": 22931, - "created_at": created_at, - "labels": ["linux-validation"], - } - ] - } - return {"jobs": []} + return _jobs_page( + [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] + ) + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) @@ -1002,7 +1521,7 @@ def test_scheduler_candidate_fails_closed_on_malformed_runner_inventory( def fake_gitea(path, params=None): if path.endswith("/runners"): return {"total_count": 1} - return {"jobs": []} + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) @@ -1018,9 +1537,6 @@ def test_scheduler_candidate_vetoes_any_unsafe_duplicate_fixed_runner( fleet_module, monkeypatch, unsafe_first ): service = fleet_module.RunnerFleetService() - now = 1_800_000_000 - created_at = datetime.fromtimestamp(now - 900, UTC).isoformat() - monkeypatch.setattr(fleet_module.time, "time", lambda: now) unsafe = { "id": 17, "name": "itworx-validation-runner-2", @@ -1044,22 +1560,149 @@ def test_scheduler_candidate_vetoes_any_unsafe_duplicate_fixed_runner( runners = [unsafe, idle] if unsafe_first else [idle, unsafe] return {"total_count": 2, "runners": runners} if params.get("status") == "queued": - return { - "jobs": [ - { - "id": 22931, - "created_at": created_at, - "labels": ["linux-validation"], - } - ] - } - return {"jobs": []} + return _jobs_page( + [{"id": 22931, "status": "queued", "labels": ["linux-validation"]}] + ) + return _jobs_page() monkeypatch.setattr(service, "_gitea_request", fake_gitea) assert service._scheduler_recovery_candidate() is None +def test_automatic_cycle_rechecks_the_same_scheduler_candidate_before_restart( + fleet_module, monkeypatch +): + service = fleet_module.RunnerFleetService() + commands: list[list[str]] = [] + events: list[dict[str, object]] = [] + candidate = { + "target": "itworx-validation-runner-2", + "jobId": 22931, + "label": "linux-validation", + "runnerId": 17, + "labels": ("linux-validation",), + "runIdentity": (("run_id", 3001), ("attempt", 1)), + "ageSeconds": 600, + "reason": "stale_scheduler_assignment", + } + changed = {**candidate, "runIdentity": (("run_id", 3002), ("attempt", 1))} + candidates = iter([candidate, changed]) + monkeypatch.setattr(service, "snapshot", _healthy_snapshot) + monkeypatch.setattr( + service, "_scheduler_recovery_candidate", lambda: next(candidates) + ) + monkeypatch.setattr(service, "_recovery_available", lambda _target: True) + monkeypatch.setattr( + service, + "_container", + lambda _target: { + "name": "itworx-validation-runner-2", + "running": True, + "health": "healthy", + }, + ) + monkeypatch.setattr( + service, "_run", lambda args, timeout=20: commands.append(args) or "" + ) + monkeypatch.setattr(service, "_record", lambda **event: events.append(event) or True) + + result = service.automatic_cycle() + + assert commands == [] + assert result["schedulerRecovery"] == { + "status": "ineligible", + "target": "itworx-validation-runner-2", + "jobId": 22931, + "reason": "scheduler_claimability_changed", + } + assert any( + event.get("action") == "restart.result" + and event.get("result") == "cancelled" + for event in events + ) + + +@pytest.mark.parametrize( + ("unsafe_source", "expected_reason"), + [ + ("host_guard", "host_resource_guard_paused"), + ("pressure", "runner_capacity_pressure_critical"), + ], +) +def test_automatic_cycle_rechecks_host_safety_after_final_scheduler_read( + fleet_module, monkeypatch, unsafe_source, expected_reason +): + service = fleet_module.RunnerFleetService() + commands: list[list[str]] = [] + events: list[dict[str, object]] = [] + safety_changed = [False] + candidate = { + "target": "itworx-validation-runner-2", + "jobId": 22931, + "label": "linux-validation", + "runnerId": 17, + "labels": ("linux-validation",), + "runIdentity": (("run_id", 3001), ("attempt", 1)), + "ageSeconds": 600, + "reason": "stale_scheduler_assignment", + } + scheduler_reads = [0] + + def read_candidate(): + scheduler_reads[0] += 1 + if scheduler_reads[0] == 2: + safety_changed[0] = True + return candidate + + def read_host_guard(): + if safety_changed[0] and unsafe_source == "host_guard": + return "paused", "host_resource_guard_paused" + return "clear", "" + + def read_capacity(_self): + capacity = _healthy_capacity_snapshot() + if safety_changed[0] and unsafe_source == "pressure": + capacity["decision"]["pressure"] = "critical" + return capacity + + monkeypatch.setattr(service, "snapshot", _healthy_snapshot) + monkeypatch.setattr(service, "_scheduler_recovery_candidate", read_candidate) + monkeypatch.setattr(service, "_recovery_available", lambda _target: True) + monkeypatch.setattr( + service, + "_container", + lambda _target: { + "name": "itworx-validation-runner-2", + "running": True, + "health": "healthy", + }, + ) + monkeypatch.setattr(fleet_module, "_host_guard_state", read_host_guard) + monkeypatch.setattr( + fleet_module.RunnerCapacityController, "snapshot", read_capacity + ) + monkeypatch.setattr( + service, "_run", lambda args, timeout=20: commands.append(args) or "" + ) + monkeypatch.setattr( + service, "_record", lambda **event: events.append(event) or True + ) + + result = service.automatic_cycle() + + assert result["status"] == "deferred" + assert result["reason"] == expected_reason + assert commands == [] + # The durable intent remains in the budget even though the final fresh + # safety check cancelled the Docker mutation. + assert any( + event.get("action") == "restart" and event.get("result") == "intent" + for event in events + ) + assert not any(event.get("action") == "restart.result" for event in events) + + def test_automatic_cycle_restarts_only_exact_known_idle_runner_for_stale_scheduler( fleet_module, monkeypatch ): diff --git a/backend/tests/test_runner_fleet_specialize.py b/backend/tests/test_runner_fleet_specialize.py index 3c01781..527948b 100644 --- a/backend/tests/test_runner_fleet_specialize.py +++ b/backend/tests/test_runner_fleet_specialize.py @@ -2349,8 +2349,8 @@ def test_legacy_queue_gate_reads_queued_and_requested_admin_inventories(speciali specialize._check_legacy_queued_jobs() assert len(calls) == 2 - assert "status=pending&limit=100&page=1&sort=id&order=asc" in calls[0] - assert "status=queued&limit=100&page=1&sort=id&order=asc" in calls[1] + assert "status=pending&limit=50&page=1&sort=id&order=asc" in calls[0] + assert "status=queued&limit=50&page=1&sort=id&order=asc" in calls[1] def test_legacy_queue_gate_blocks_unsupported_requested_dependency_job(specialize, monkeypatch): @@ -2407,7 +2407,7 @@ def test_legacy_queue_gate_catches_requested_to_queued_transition_between_filter assert "status=queued" in calls[1] -def test_legacy_queue_gate_paginates_and_blocks_unsupported_page_two_job(specialize, monkeypatch): +def test_legacy_queue_gate_paginates_51_jobs_and_blocks_unsupported_page_two_job(specialize, monkeypatch): calls = [] def api(path): @@ -2416,15 +2416,15 @@ def test_legacy_queue_gate_paginates_and_blocks_unsupported_page_two_job(special page = int(path.rsplit("page=", 1)[1].split("&", 1)[0]) if page == 1: return { - "total_count": 101, + "total_count": 51, "jobs": [ _legacy_queue_job(index + 1, "queued", ["itworx-ci-backend"]) - for index in range(100) + for index in range(50) ], } return { - "total_count": 101, - "jobs": [_legacy_queue_job(101, "queued", ["mcp-validation", "unraid"])], + "total_count": 51, + "jobs": [_legacy_queue_job(51, "queued", ["mcp-validation", "unraid"])], } if "status=pending" in path: return {"total_count": 0, "jobs": []} @@ -2438,7 +2438,39 @@ def test_legacy_queue_gate_paginates_and_blocks_unsupported_page_two_job(special ): specialize._check_legacy_queued_jobs() - assert any("status=queued&limit=100&page=2" in path for path in calls) + assert any("status=queued&limit=50&page=2" in path for path in calls) + + +def test_legacy_queue_gate_paginates_101_jobs_and_checks_unsupported_last_page(specialize, monkeypatch): + calls = [] + + def api(path): + calls.append(path) + if "status=queued" in path: + page = int(path.rsplit("page=", 1)[1].split("&", 1)[0]) + start_id = (page - 1) * 50 + 1 + page_size = min(50, 101 - (page - 1) * 50) + labels = ["mcp-validation", "unraid"] if page == 3 else ["itworx-ci-backend"] + return { + "total_count": 101, + "jobs": [ + _legacy_queue_job(start_id + offset, "queued", labels) + for offset in range(page_size) + ], + } + if "status=pending" in path: + return {"total_count": 0, "jobs": []} + raise AssertionError(f"unexpected fixed Gitea API path: {path}") + + monkeypatch.setattr(specialize, "_api_json", api) + + with pytest.raises( + specialize.RunnerFleetSpecializeError, + match="runner_fleet_specialization_legacy_queue_present", + ): + specialize._check_legacy_queued_jobs() + + assert any("status=queued&limit=50&page=3" in path for path in calls) def test_legacy_queue_gate_rejects_total_count_changes_between_pages(specialize, monkeypatch): @@ -2447,15 +2479,15 @@ def test_legacy_queue_gate_rejects_total_count_changes_between_pages(specialize, page = int(path.rsplit("page=", 1)[1].split("&", 1)[0]) if page == 1: return { - "total_count": 101, + "total_count": 51, "jobs": [ _legacy_queue_job(index + 1, "queued", ["itworx-ci-backend"]) - for index in range(100) + for index in range(50) ], } return { "total_count": 102, - "jobs": [_legacy_queue_job(101, "queued", ["itworx-ci-backend"])], + "jobs": [_legacy_queue_job(51, "queued", ["itworx-ci-backend"])], } if "status=pending" in path: return {"total_count": 0, "jobs": []} @@ -2470,6 +2502,48 @@ def test_legacy_queue_gate_rejects_total_count_changes_between_pages(specialize, specialize._check_legacy_queued_jobs() +def test_legacy_queue_gate_accepts_maximum_1000_job_inventory(specialize, monkeypatch): + calls = [] + + def api(path): + calls.append(path) + page = int(path.rsplit("page=", 1)[1].split("&", 1)[0]) + start_id = (page - 1) * 50 + 1 + return { + "total_count": 1000, + "jobs": [ + _legacy_queue_job(start_id + offset, "queued", ["itworx-ci-backend"]) + for offset in range(50) + ], + } + + monkeypatch.setattr(specialize, "_api_json", api) + + rows = specialize._legacy_action_jobs("queued") + + assert len(rows) == 1000 + assert len(calls) == 20 + assert "limit=50&page=20" in calls[-1] + + +def test_legacy_queue_gate_rejects_inventory_over_1000_jobs(specialize, monkeypatch): + calls = [] + + def api(path): + calls.append(path) + return {"total_count": 1001, "jobs": []} + + monkeypatch.setattr(specialize, "_api_json", api) + + with pytest.raises( + specialize.RunnerFleetSpecializeError, + match="runner_fleet_specialization_legacy_queue_untrusted", + ): + specialize._legacy_action_jobs("queued") + + assert len(calls) == 1 + + @pytest.mark.parametrize( "payload", [ @@ -2539,7 +2613,10 @@ def test_image_workflow_builds_from_protected_appops_main_with_gitea_checkout_ur frontend = (ROOT / "ops" / "runner-task-images" / "frontend" / "Dockerfile").read_text(encoding="utf-8") assert "github-server-url: http://gitea-runner-internal:3000" in workflow - assert "bash scripts/build_ci_task_images.sh" in workflow + assert "runs-on: linux-validation" in workflow + assert 'bash scripts/build_ci_task_images.sh "$GITHUB_SHA" "$GITHUB_RUN_ID" "$output"' in workflow + assert "runs-on: itworx-ci-build" not in workflow + assert "--job-local" not in workflow assert "jens/unraid-appops-gateway" in build assert "Jens/Unraid-AppOps-Gateway" in build assert "corepack prepare pnpm@11.15.0 --activate" in frontend diff --git a/backend/tests/test_runner_scheduler_liveness_recovery.py b/backend/tests/test_runner_scheduler_liveness_recovery.py index 5750131..176932c 100644 --- a/backend/tests/test_runner_scheduler_liveness_recovery.py +++ b/backend/tests/test_runner_scheduler_liveness_recovery.py @@ -1,4 +1,7 @@ from datetime import UTC, datetime, timedelta +from threading import Event, Thread, current_thread + +import pytest from backend.app.connectors import gitea_actions from backend.app.connectors.gitea_actions import GiteaActionsConnector @@ -15,64 +18,124 @@ def _configure(monkeypatch) -> GiteaActionsConnector: '"mcp-validation":"itworx-security-runner"}', ) monkeypatch.setattr(gitea_actions, "_CACHE", None) + monkeypatch.setattr(gitea_actions, "_RECOVERY_OBSERVATIONS", {}, raising=False) + monkeypatch.setattr(gitea_actions, "_RECOVERY_SOURCE", None, raising=False) + monkeypatch.setattr(gitea_actions, "_SNAPSHOT_GENERATION", 0, raising=False) + monkeypatch.setattr(gitea_actions, "_COMMITTED_GENERATION", 0, raising=False) return GiteaActionsConnector() -def _requests(created_at: str, *, busy: bool = False): +def _requests(created_at: str, state: dict[str, object] | None = None): + state = state if state is not None else {} + def fake_request(path, params=None): params = params or {} + if state.get("fail"): + raise RuntimeError("fixture offline") if path.endswith("/runners"): + runners = list(state.get("runners", [ + { + "id": state.get("runner_id", 17), + "name": state.get("runner_name", "itworx-validation-runner-2"), + "status": state.get("runner_status", "online"), + "busy": state.get("runner_busy", False), + "disabled": state.get("runner_disabled", False), + "labels": state.get("runner_labels", ["linux-validation", "ubuntu-latest"]), + } + ])) + if state.get("duplicate_runner") and runners: + runners.append(dict(runners[0])) return { - "total_count": 1, - "runners": [ - { - "id": 17, - "name": "validation-secondary", - "status": "online", - "busy": busy, - "disabled": False, - "labels": ["linux-validation", "ubuntu-latest"], - } - ], + "total_count": state.get("runner_total", len(runners)), + "runners": runners, } - if params.get("status") == "queued": + status = params.get("status") + if status == "queued": + jobs = [] if state.get("queued") is False else [ + { + "id": state.get("job_id", 91), + "run_id": state.get("run_id", 81), + "run_number": state.get("run_number"), + "workflow_id": state.get("workflow_id"), + "attempt": state.get("attempt"), + "run_attempt": state.get("run_attempt", 1), + "name": "full", + "status": "queued", + "created_at": created_at, + "labels": state.get("job_labels", ["linux-validation"]), + } + ] + if state.get("duplicate_job"): + jobs.append(dict(jobs[0])) + if state.get("queued_wrong_status") and jobs: + jobs[0]["status"] = "pending" return { - "total_count": 1, - "jobs": [ - { - "id": 91, - "name": "full", - "status": "queued", - "created_at": created_at, - "labels": ["linux-validation"], - } - ], + "total_count": state.get("queued_total", len(jobs)), + "jobs": jobs, } - return {"total_count": 0, "jobs": []} + if status == "pending": + jobs = [] + if state.get("pending"): + jobs = [{"id": 91, "run_id": 81, "run_attempt": 1, + "status": state.get("pending_status", "requested"), "created_at": created_at, + "labels": ["linux-validation"]}] + return {"total_count": len(jobs), "jobs": jobs} + if status == "in_progress": + jobs = [] + if state.get("active_runner_id") is not None: + jobs = [{"id": 92, "run_id": 82, "run_attempt": 1, + "status": "in_progress", "runner_id": state["active_runner_id"], + "created_at": created_at, "started_at": created_at, + "labels": ["linux-validation"]}] + if state.get("active_malformed"): + jobs = [None] + return {"total_count": len(jobs), "jobs": jobs} + raise AssertionError(f"Unexpected Gitea request: {path} {params}") return fake_request -def test_old_unassigned_job_on_idle_matching_runner_is_recovery_candidate(monkeypatch): +def test_old_created_job_first_seen_as_queued_is_not_immediately_a_recovery_candidate(monkeypatch): connector = _configure(monkeypatch) created_at = (datetime.now(UTC) - timedelta(minutes=12)).isoformat() monkeypatch.setattr(connector, "_request", _requests(created_at)) result = connector.snapshot(refresh=True) - assert result["severity"] == "critical" assert result["queue"]["items"][0]["reason"] == "awaiting_scheduler_assignment" + assert result["queue"]["staleAssignmentRunnerLabels"] == [] + assert result["recommendedAction"] is None + + +def test_old_created_job_requires_ten_minutes_of_fresh_queued_and_idle_observations(monkeypatch): + connector = _configure(monkeypatch) + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + now = {"seconds": 1_000.0} + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", _requests(created_at)) + + first = connector.snapshot(refresh=True) + assert first["recommendedAction"] is None + assert first["queue"]["oldestQueuedAgeSeconds"] >= 7_200 + + for sample in range(1, 20): + now["seconds"] = 1_000.0 + sample * 30 + result = connector.snapshot(refresh=True) + assert result["recommendedAction"] is None + + now["seconds"] = 1_600.0 + result = connector.snapshot(refresh=True) assert result["queue"]["staleAssignmentRunnerLabels"] == ["linux-validation"] assert result["recommendedAction"] == { "intent": "runner.recover", "target": "itworx-validation-runner-2", "requiresApproval": False, - "autoExecute": True, + "autoExecute": False, "reason": "Gitea Actions scheduler left a compatible idle runner unassigned for at least ten minutes.", "evidence": { "jobId": 91, "label": "linux-validation", - "minimumAgeSeconds": 600, + "observedClaimableSeconds": 600, }, } @@ -92,10 +155,320 @@ def test_recent_unassigned_job_does_not_trigger_recovery(monkeypatch): def test_busy_matching_runner_is_never_restarted_for_waiting_queue(monkeypatch): connector = _configure(monkeypatch) created_at = (datetime.now(UTC) - timedelta(minutes=12)).isoformat() - monkeypatch.setattr(connector, "_request", _requests(created_at, busy=True)) + monkeypatch.setattr(connector, "_request", _requests(created_at, {"runner_busy": True})) result = connector.snapshot(refresh=True) assert result["queue"]["items"][0]["reason"] == "all_matching_runners_are_busy" assert result["queue"]["staleAssignmentRunnerLabels"] == [] assert result["recommendedAction"] is None + + +def test_cache_hit_never_replays_auto_recovery_recommendation(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr( + connector, + "_request", + _requests((datetime.now(UTC) - timedelta(hours=2)).isoformat()), + ) + + for seconds in range(0, 601, 30): + now["seconds"] = 1_000.0 + seconds + fresh = connector.snapshot(refresh=True) + assert fresh["recommendedAction"]["intent"] == "runner.recover" + + cached = connector.snapshot() + + assert cached["cache"]["hit"] is True + assert cached["recommendedAction"] is None + assert cached["queue"]["staleAssignmentRunnerLabels"] == [] + + +def test_error_clears_cached_auto_recovery_candidate(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + state: dict[str, object] = {} + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", _requests(created_at, state)) + + for seconds in range(0, 601, 30): + now["seconds"] = 1_000.0 + seconds + result = connector.snapshot(refresh=True) + assert result["recommendedAction"]["intent"] == "runner.recover" + + now["seconds"] = 1_601.0 + state["fail"] = True + assert connector.snapshot(refresh=True)["status"] == "error" + state.clear() + now["seconds"] = 1_602.0 + refreshed = connector.snapshot() + + assert refreshed["cache"]["hit"] is False + assert refreshed["recommendedAction"] is None + assert refreshed["queue"]["staleAssignmentRunnerLabels"] == [] + + +def test_gap_longer_than_poll_window_restarts_observation_period(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr( + connector, + "_request", + _requests((datetime.now(UTC) - timedelta(hours=2)).isoformat()), + ) + + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_046.0 + assert connector.snapshot(refresh=True)["recommendedAction"] is None + for sample in range(1, 21): + now["seconds"] = 1_046.0 + sample * 30 + result = connector.snapshot(refresh=True) + assert result["recommendedAction"]["intent"] == "runner.recover" + + +@pytest.mark.parametrize( + "interruption", + [ + {"runner_busy": True}, + {"runner_status": "offline"}, + {"runner_disabled": True}, + {"runner_name": "renamed-but-not-mapped"}, + {"active_runner_id": 17}, + {"active_malformed": True}, + {"queued": False}, + {"queued": False, "pending": True}, + {"queued_total": 2}, + {"runner_total": 2}, + {"duplicate_runner": True}, + {"runner_total": True}, + {"attempt": 2}, + {"job_labels": ["linux-validation" + ("x" * 120)]}, + {"job_labels": ["linux-validation" + (" " * 105)]}, + {"job_labels": [{"name": "linux-validation" + (" " * 105)}]}, + {"job_labels": [" linux-validation"]}, + {"job_labels": ["linux-validation "]}, + {"job_labels": [{"name": " linux-validation"}]}, + {"runner_labels": [" linux-validation", "ubuntu-latest"]}, + {"runner_labels": ["linux-validation ", "ubuntu-latest"]}, + {"runner_labels": ["linux-validation", "x" * 121]}, + {"duplicate_job": True}, + {"queued_wrong_status": True}, + {"runner_busy": "false"}, + ], + ids=[ + "busy", + "offline", + "disabled", + "exact-target-missing", + "active-assignment", + "malformed-active-inventory", + "job-absent", + "pending-only", + "incomplete-queued-count", + "incomplete-runner-count", + "duplicate-runner-id", + "boolean-runner-count", + "conflicting-attempt-fields", + "overlong-job-label", + "raw-overlong-job-label-with-whitespace", + "raw-overlong-dict-job-label-with-whitespace", + "prefixed-job-label-does-not-match-runner-label", + "suffixed-job-label-does-not-match-runner-label", + "prefixed-dict-job-label-does-not-match-runner-label", + "prefixed-runner-label-does-not-match-job-label", + "suffixed-runner-label-does-not-match-job-label", + "overlong-runner-label", + "duplicate-job-id", + "wrong-status", + "malformed-runner-busy-type", + ], +) +def test_nonclaimable_or_incomplete_sample_resets_dwell(monkeypatch, interruption): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + state: dict[str, object] = {} + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", _requests(created_at, state)) + + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_015.0 + state.update(interruption) + interrupted = connector.snapshot(refresh=True) + action = interrupted["recommendedAction"] + assert not (isinstance(action, dict) and action.get("intent") == "runner.recover") + assert interrupted["queue"]["staleAssignmentRunnerLabels"] == [] + + state.clear() + now["seconds"] = 1_030.0 + assert connector.snapshot(refresh=True)["recommendedAction"] is None + for seconds in range(60, 601, 30): + now["seconds"] = 1_000.0 + seconds + result = connector.snapshot(refresh=True) + assert result["recommendedAction"] is None + now["seconds"] = 1_630.0 + assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover" + + +def test_failed_refresh_resets_recovery_observation(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + state: dict[str, object] = {} + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", _requests(created_at, state)) + + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_015.0 + state["fail"] = True + failed = connector.snapshot(refresh=True) + assert failed["status"] == "error" + assert failed["recommendedAction"] is None + + state.clear() + now["seconds"] = 1_030.0 + assert connector.snapshot(refresh=True)["recommendedAction"] is None + for seconds in range(60, 601, 30): + now["seconds"] = 1_000.0 + seconds + result = connector.snapshot(refresh=True) + assert result["recommendedAction"] is None + now["seconds"] = 1_630.0 + assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover" + + +def test_missing_configuration_resets_recovery_observation(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", _requests(created_at)) + + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_015.0 + monkeypatch.setitem(settings.apis, "GITEA_ADMIN_TOKEN", None) + missing = connector.snapshot(refresh=True) + assert missing["status"] == "missing_config" + monkeypatch.setitem(settings.apis, "GITEA_ADMIN_TOKEN", "fixture-secret") + now["seconds"] = 1_030.0 + assert connector.snapshot(refresh=True)["recommendedAction"] is None + for seconds in range(60, 601, 30): + now["seconds"] = 1_000.0 + seconds + result = connector.snapshot(refresh=True) + assert result["recommendedAction"] is None + now["seconds"] = 1_630.0 + assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover" + + +def test_switching_gitea_origin_does_not_preserve_old_source_dwell(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr( + connector, + "_request", + _requests((datetime.now(UTC) - timedelta(hours=2)).isoformat()), + ) + + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_015.0 + monkeypatch.setitem(settings.apis, "GITEA_URL", "https://other-gitea.test") + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_030.0 + monkeypatch.setitem(settings.apis, "GITEA_URL", "https://gitea.test") + assert connector.snapshot(refresh=True)["recommendedAction"] is None + for seconds in range(60, 601, 30): + now["seconds"] = 1_000.0 + seconds + result = connector.snapshot(refresh=True) + assert result["recommendedAction"] is None + now["seconds"] = 1_630.0 + assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover" + + +@pytest.mark.parametrize( + "changed_identity", + [ + {"run_attempt": 2}, + {"run_attempt": None, "attempt": 2}, + {"run_id": 82}, + {"run_number": 10}, + {"workflow_id": 20}, + {"runner_id": 18}, + {"runner_name": "renamed-but-not-mapped"}, + {"runner_labels": ["linux-validation", "ubuntu-latest", "new-label"]}, + {"job_labels": ["linux-validation", "ubuntu-latest"]}, + ], + ids=[ + "attempt", + "legacy-attempt", + "run-id", + "run-number", + "workflow-id", + "mapped-runner-id", + "exact-target-name", + "runner-signature", + "job-labels", + ], +) +def test_changed_job_or_exact_runner_identity_restarts_dwell(monkeypatch, changed_identity): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + state: dict[str, object] = {} + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", _requests(created_at, state)) + + assert connector.snapshot(refresh=True)["recommendedAction"] is None + now["seconds"] = 1_015.0 + state.update(changed_identity) + assert connector.snapshot(refresh=True)["recommendedAction"] is None + + state.clear() + now["seconds"] = 1_030.0 + assert connector.snapshot(refresh=True)["recommendedAction"] is None + for seconds in range(60, 601, 30): + now["seconds"] = 1_000.0 + seconds + result = connector.snapshot(refresh=True) + assert result["recommendedAction"] is None + now["seconds"] = 1_630.0 + assert connector.snapshot(refresh=True)["recommendedAction"]["intent"] == "runner.recover" + + +def test_slower_older_refresh_cannot_advance_newer_observation(monkeypatch): + connector = _configure(monkeypatch) + now = {"seconds": 1_000.0} + entered_slow_request = Event() + release_slow_request = Event() + results = {} + created_at = (datetime.now(UTC) - timedelta(hours=2)).isoformat() + normal_request = _requests(created_at) + + def delayed_request(path, params=None): + if current_thread().name == "slow-old" and not entered_slow_request.is_set(): + entered_slow_request.set() + assert release_slow_request.wait(timeout=5) + return normal_request(path, params) + + monkeypatch.setattr(gitea_actions, "monotonic", lambda: now["seconds"]) + monkeypatch.setattr(connector, "_request", delayed_request) + slow = Thread( + name="slow-old", + target=lambda: results.setdefault("old", connector.snapshot(refresh=True)), + ) + slow.start() + assert entered_slow_request.wait(timeout=5) + + now["seconds"] = 1_700.0 + results["new"] = connector.snapshot(refresh=True) + release_slow_request.set() + slow.join(timeout=5) + + assert not slow.is_alive() + assert results["old"]["recommendedAction"] is None + assert results["new"]["recommendedAction"] is None + now["seconds"] = 1_701.0 + latest = connector.snapshot(refresh=True) + assert latest["recommendedAction"] is None diff --git a/src/pages/Overview.tsx b/src/pages/Overview.tsx index 669aaf9..6ed8fde 100644 --- a/src/pages/Overview.tsx +++ b/src/pages/Overview.tsx @@ -269,7 +269,7 @@ const Overview = () => {

{giteaActions.queue.queued} queued · {giteaActions.queue.inProgress} actief - {giteaActions.queue.oldestQueuedAgeSeconds !== null && ` · oudste ${Math.ceil(giteaActions.queue.oldestQueuedAgeSeconds / 60)} min`} + {giteaActions.queue.oldestQueuedAgeSeconds !== null && ` · oudste job sinds aanmaak ${Math.ceil(giteaActions.queue.oldestQueuedAgeSeconds / 60)} min`}

{giteaActions.queue.blockedByRunnerLabels.length > 0 && (

@@ -284,12 +284,19 @@ const Overview = () => { )} - {giteaActions.recommendedAction ? ( + {giteaActions.recommendedAction?.intent === 'container.restart' ? ( + ) : giteaActions.recommendedAction?.intent === 'runner.recover' ? ( + + Schedulerherstel aanbevolen + ) : (