Publish curated Unraid-AppOps-Gateway source from bc001a6df859f7c9836fa145afb739d1ceebc524

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.

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",

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

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(

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"}:

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""

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<body>.*?)(?=^  [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(

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(

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

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
 ):

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

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

src/pages/Overview.tsx

@@ -269,7 +269,7 @@ const Overview = () => {
                   </h3>
                   <p className="mt-1 text-sm text-slate-400">
                     {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`}
                   </p>
                   {giteaActions.queue.blockedByRunnerLabels.length > 0 && (
                     <p className="mt-2 text-xs font-semibold text-rose-300">
@@ -284,12 +284,19 @@ const Overview = () => {
                   )}
                 </div>
               </div>
-              {giteaActions.recommendedAction ? (
+              {giteaActions.recommendedAction?.intent === 'container.restart' ? (
                 <Button asChild className="bg-amber-600 text-white hover:bg-amber-500">
                   <Link to={`/command-center?query=${encodeURIComponent(`container.restart ${giteaActions.recommendedAction.target}`)}`}>
                     Controleer runnerrestart <ArrowUpRight className="ml-2 h-4 w-4" />
                   </Link>
                 </Button>
+              ) : giteaActions.recommendedAction?.intent === 'runner.recover' ? (
+                <span
+                  className="rounded border border-amber-500/30 bg-amber-500/10 px-3 py-1 text-xs font-bold text-amber-200"
+                  title={giteaActions.recommendedAction.reason}
+                >
+                  Schedulerherstel aanbevolen
+                </span>
               ) : (
                 <span className={cn(
                   "rounded border px-3 py-1 text-xs font-bold uppercase",

src/types/appops.ts

@@ -366,6 +366,17 @@ export interface GiteaActionsStatus {
     requiresApproval: true;
     autoExecute: false;
     reason: string;
+  } | {
+    intent: 'runner.recover';
+    target: string;
+    requiresApproval: false;
+    autoExecute: false;
+    reason: string;
+    evidence: {
+      jobId: number;
+      label: string;
+      observedClaimableSeconds: number;
+    };
   } | null;
   cache: { hit: boolean; ttlSeconds: number };
 }