fix(ops): ignore recovered stale job pods

scripts/reboot-recovery/full-stack-cold-start-check.sh

@@ -912,6 +912,7 @@ fi
   fi
 
   if out=$(host_cmd "wooo@192.168.0.120" '
+set -o pipefail
 kcmd() {
   if [ -n "${REMOTE_SUDO_PASSWORD:-}" ]; then
     printf "%s\n" "$REMOTE_SUDO_PASSWORD" | sudo -S -p "" kubectl "$@"
@@ -921,45 +922,97 @@ kcmd() {
 }
 echo "CRON_120 $(systemctl is-active cron 2>/dev/null || systemctl is-active crond 2>/dev/null || true)"
 kcmd get cronjobs -n awoooi-prod -o json | python3 -c "import json,sys; d=json.load(sys.stdin); items=d.get(\"items\", []); print(\"CRONJOB_COUNT\", len(items)); print(\"CRONJOB_SUSPENDED\", sum(1 for i in items if i.get(\"spec\",{}).get(\"suspend\")))"
-kcmd get jobs -n awoooi-prod -o json | python3 -c "import json,sys,re; d=json.load(sys.stdin)
+if ! job_summary="$(kcmd get jobs -n awoooi-prod -o json | python3 -c "import json,sys; d=json.load(sys.stdin)
+items=d.get(\"items\")
+if not isinstance(items, list):
+  raise ValueError(\"jobs_items_missing_or_invalid\")
 def owner(job):
   for ref in job.get(\"metadata\",{}).get(\"ownerReferences\",[]) or []:
     if ref.get(\"kind\") == \"CronJob\" and ref.get(\"name\"):
       return ref.get(\"name\")
-  name = job.get(\"metadata\",{}).get(\"name\", \"\")
-  return re.sub(r\"-[0-9]+$\", \"\", name)
-def has_condition(job, kind):
-  return any(c.get(\"type\") == kind and c.get(\"status\") == \"True\" for c in job.get(\"status\",{}).get(\"conditions\",[]) or [])
-def job_time(job):
-  status = job.get(\"status\",{})
-  return status.get(\"completionTime\") or status.get(\"startTime\") or \"\"
+  return \"\"
+def condition_time(job, kind):
+  for condition in job.get(\"status\",{}).get(\"conditions\",[]) or []:
+    if condition.get(\"type\") == kind and condition.get(\"status\") == \"True\":
+      return condition.get(\"lastTransitionTime\") or condition.get(\"lastProbeTime\") or \"\"
+  return \"\"
 latest_success = {}
 failed_jobs = []
-for job in d.get(\"items\", []):
+for job in items:
   own = owner(job)
-  ts = job_time(job)
-  if has_condition(job, \"Complete\"):
-    latest_success[own] = max(latest_success.get(own, \"\"), ts)
-  if has_condition(job, \"Failed\"):
-    failed_jobs.append((own, job.get(\"metadata\",{}).get(\"name\", \"\"), ts))
+  status = job.get(\"status\", {})
+  success_ts = condition_time(job, \"Complete\")
+  failed_ts = condition_time(job, \"Failed\")
+  if own and success_ts:
+    latest_success[own] = max(latest_success.get(own, \"\"), success_ts)
+  if failed_ts or status.get(\"failed\", 0):
+    failed_jobs.append((own, job.get(\"metadata\",{}).get(\"name\", \"\"), failed_ts))
 active_failed = 0
 stale_failed = 0
-for own, name, ts in failed_jobs:
-  if ts and latest_success.get(own, \"\") > ts:
+stale_failed_names = []
+for own, name, failed_ts in failed_jobs:
+  if own and failed_ts and latest_success.get(own, \"\") > failed_ts:
     stale_failed += 1
+    stale_failed_names.append(name)
   else:
     active_failed += 1
 print(\"FAILED_JOBS\", len(failed_jobs))
 print(\"STALE_FAILED_JOBS\", stale_failed)
-print(\"ACTIVE_FAILED_JOBS\", active_failed)"
-kcmd get pods -n awoooi-prod --no-headers 2>/dev/null | awk "\$3 !~ /^(Running|Completed)$/ {bad++} END {print \"BAD_PODS\", bad+0}"
+print(\"ACTIVE_FAILED_JOBS\", active_failed)
+print(\"STALE_FAILED_JOB_NAMES\", \",\".join(sorted(stale_failed_names)))")"; then
+  echo "SCHEDULE_CLASSIFIER_ERROR jobs_unavailable_or_invalid"
+  exit 1
+fi
+echo "$job_summary"
+stale_failed_job_names=""
+while IFS= read -r line; do
+  case "$line" in
+    "STALE_FAILED_JOB_NAMES "*)
+      stale_failed_job_names="${line#STALE_FAILED_JOB_NAMES }"
+      break
+      ;;
+  esac
+done <<<"$job_summary"
+if ! pods_json="$(kcmd get pods -n awoooi-prod -o json 2>/dev/null)"; then
+  echo "SCHEDULE_CLASSIFIER_ERROR pods_unavailable_or_invalid"
+  exit 1
+fi
+printf "%s" "$pods_json" | STALE_FAILED_JOB_NAMES="$stale_failed_job_names" python3 -c "import json,os,sys
+d=json.load(sys.stdin)
+items=d.get(\"items\")
+if not isinstance(items, list):
+  raise ValueError(\"pods_items_missing_or_invalid\")
+stale=set(filter(None, os.environ.get(\"STALE_FAILED_JOB_NAMES\", \"\").split(\",\")))
+bad=0
+ignored=0
+for pod in items:
+  status=pod.get(\"status\", {})
+  phase=status.get(\"phase\", \"\")
+  container_statuses=status.get(\"containerStatuses\", []) or []
+  states=[]
+  for item in (status.get(\"initContainerStatuses\", []) or []) + container_statuses:
+    states.append(item.get(\"state\", {}) or {})
+  waiting_reasons={s.get(\"waiting\", {}).get(\"reason\", \"\") for s in states if s.get(\"waiting\")}
+  terminated_reasons={s.get(\"terminated\", {}).get(\"reason\", \"\") for s in states if s.get(\"terminated\")}
+  unready_running=phase == \"Running\" and (not container_statuses or any(item.get(\"ready\") is not True for item in container_statuses))
+  is_bad=phase not in {\"Running\", \"Succeeded\"} or unready_running or bool(waiting_reasons & {\"CrashLoopBackOff\", \"Error\", \"ImagePullBackOff\", \"ErrImagePull\", \"CreateContainerConfigError\", \"CreateContainerError\", \"RunContainerError\"}) or bool(terminated_reasons & {\"Error\", \"OOMKilled\", \"ContainerCannotRun\", \"StartError\", \"DeadlineExceeded\"})
+  if not is_bad:
+    continue
+  owner=next((r.get(\"name\", \"\") for r in pod.get(\"metadata\", {}).get(\"ownerReferences\", []) or [] if r.get(\"kind\") == \"Job\"), \"\")
+  if owner and owner in stale:
+    ignored += 1
+  else:
+    bad += 1
+print(\"RAW_BAD_PODS\", bad + ignored)
+print(\"STALE_BAD_PODS_IGNORED\", ignored)
+print(\"BAD_PODS\", bad)"
 ' 2>&1); then
     echo "$out"
     grep -q "CRON_120 active" <<<"$out" && ok "120 cron active" || warn "120 cron not confirmed"
     awk '/CRONJOB_COUNT / {exit !($2 >= 4)}' <<<"$out" && ok "K8s AWOOOI CronJobs present" || warn "K8s AWOOOI CronJobs missing"
     grep -q "CRONJOB_SUSPENDED 0" <<<"$out" && ok "K8s AWOOOI CronJobs unsuspended" || warn "K8s AWOOOI CronJob suspended"
     grep -q "ACTIVE_FAILED_JOBS 0" <<<"$out" && ok "K8s AWOOOI has no active failed Jobs" || warn "K8s AWOOOI active failed Jobs remain"
-    grep -q "BAD_PODS 0" <<<"$out" && ok "K8s AWOOOI pods Running/Completed only" || warn "K8s AWOOOI bad pod status remains"
+    grep -q "^BAD_PODS 0$" <<<"$out" && ok "K8s AWOOOI has no action-required bad pods" || warn "K8s AWOOOI action-required bad pod status remains"
   else
     warn "120 K8s schedule check unavailable"
     echo "$out"

scripts/reboot-recovery/tests/test_cold_start_monitor_bounded_probes.py

@@ -1,5 +1,9 @@
 from __future__ import annotations
 
+import json
+import os
+import subprocess
+import sys
 from pathlib import Path
 
 
@@ -88,6 +92,191 @@ def test_cold_start_110_backup_splits_current_freshness_from_old_aggregate_failu
     assert "split($7,e,\"=\"); exit !((a[2]+b[2]+c[2]+e[2]) == 0)" in text
 
 
+def test_cold_start_ignores_only_pods_owned_by_recovered_stale_failed_jobs() -> None:
+    text = COLD_START_CHECK.read_text(encoding="utf-8")
+
+    schedule_section = text[text.index('host_cmd "wooo@192.168.0.120"', text.index("check_schedules")) :]
+    assert "set -o pipefail" in schedule_section[:500]
+    assert 'stale_failed_names.append(name)' in text
+    assert 'print(\\"STALE_FAILED_JOB_NAMES\\"' in text
+    assert 'condition.get(\\"lastTransitionTime\\")' in text
+    assert 'items=d.get(\\"items\\")' in text
+    assert 'status.get(\\"failed\\", 0)' in text
+    assert 'if own and failed_ts and latest_success.get(own, \\"\\") > failed_ts' in text
+    assert 'r.get(\\"kind\\") == \\"Job\\"' in text
+    assert 'owner and owner in stale' in text
+    assert 'terminated_reasons' in text
+    assert 'unready_running' in text
+    assert 'print(\\"RAW_BAD_PODS\\"' in text
+    assert 'print(\\"STALE_BAD_PODS_IGNORED\\"' in text
+    assert 'print(\\"BAD_PODS\\", bad)' in text
+    assert 'SCHEDULE_CLASSIFIER_ERROR jobs_unavailable_or_invalid' in text
+    assert 'SCHEDULE_CLASSIFIER_ERROR pods_unavailable_or_invalid' in text
+    assert 'grep -q "^BAD_PODS 0$"' in text
+    assert 'awk "\\$3 !~ /^(Running|Completed)$/' not in text
+
+
+def _extract_schedule_classifier(kind: str) -> str:
+    text = COLD_START_CHECK.read_text(encoding="utf-8")
+    if kind == "jobs":
+        section = text.index('if ! job_summary=')
+        start = text.index('python3 -c "', section) + len('python3 -c "')
+        end = text.index('")"; then', start)
+    else:
+        section = text.index('printf "%s" "$pods_json"')
+        start = text.index('python3 -c "', section) + len('python3 -c "')
+        end = text.index("\"\n' 2>&1); then", start)
+    return text[start:end].replace('\\"', '"')
+
+
+def _condition(kind: str, at: str) -> dict[str, str]:
+    return {"type": kind, "status": "True", "lastTransitionTime": at}
+
+
+def _job(
+    name: str,
+    *,
+    cronjob: str | None = None,
+    failed_at: str | None = None,
+    failed_count: int = 0,
+    completed_at: str | None = None,
+) -> dict[str, object]:
+    conditions = []
+    if failed_at:
+        conditions.append(_condition("Failed", failed_at))
+    if completed_at:
+        conditions.append(_condition("Complete", completed_at))
+    metadata: dict[str, object] = {"name": name}
+    if cronjob:
+        metadata["ownerReferences"] = [{"kind": "CronJob", "name": cronjob}]
+    return {
+        "metadata": metadata,
+        "status": {"conditions": conditions, "failed": failed_count},
+    }
+
+
+def _pod(
+    name: str,
+    *,
+    phase: str,
+    owner_job: str | None = None,
+    states: list[dict[str, object]] | None = None,
+) -> dict[str, object]:
+    metadata: dict[str, object] = {"name": name}
+    if owner_job:
+        metadata["ownerReferences"] = [{"kind": "Job", "name": owner_job}]
+    return {
+        "metadata": metadata,
+        "status": {
+            "phase": phase,
+            "containerStatuses": [
+                {"state": state, "ready": "running" in state}
+                for state in (states or [])
+            ],
+        },
+    }
+
+
+def test_cold_start_schedule_classifier_semantics_with_adversarial_fixtures() -> None:
+    jobs = {
+        "items": [
+            _job(
+                "cron-a-old",
+                cronjob="cron-a",
+                failed_at="2026-07-14T00:20:00Z",
+            ),
+            _job(
+                "cron-a-new",
+                cronjob="cron-a",
+                completed_at="2026-07-14T00:30:00Z",
+            ),
+            _job(
+                "cron-a-overlap",
+                cronjob="cron-a",
+                failed_at="2026-07-14T00:40:00Z",
+            ),
+            _job(
+                "cron-a-manual-123",
+                failed_at="2026-07-14T00:10:00Z",
+            ),
+            _job(
+                "cron-a-conditionless",
+                cronjob="cron-a",
+                failed_count=1,
+            ),
+        ]
+    }
+    jobs_result = subprocess.run(
+        [sys.executable, "-c", _extract_schedule_classifier("jobs")],
+        input=json.dumps(jobs),
+        text=True,
+        capture_output=True,
+        check=True,
+    )
+    assert "FAILED_JOBS 4" in jobs_result.stdout
+    assert "STALE_FAILED_JOBS 1" in jobs_result.stdout
+    assert "ACTIVE_FAILED_JOBS 3" in jobs_result.stdout
+    assert "STALE_FAILED_JOB_NAMES cron-a-old" in jobs_result.stdout
+
+    pods = {
+        "items": [
+            _pod("stale", phase="Failed", owner_job="cron-a-old"),
+            _pod("standalone", phase="Failed", owner_job="cron-a-manual-123"),
+            _pod("overlap", phase="Failed", owner_job="cron-a-overlap"),
+            _pod(
+                "conditionless",
+                phase="Failed",
+                owner_job="cron-a-conditionless",
+            ),
+            _pod(
+                "multi-container-error",
+                phase="Running",
+                states=[{"running": {}}, {"terminated": {"reason": "Error"}}],
+            ),
+            _pod(
+                "pending",
+                phase="Pending",
+                states=[{"waiting": {"reason": "ContainerCreating"}}],
+            ),
+            _pod(
+                "crashloop",
+                phase="Running",
+                states=[{"waiting": {"reason": "CrashLoopBackOff"}}],
+            ),
+            _pod("healthy", phase="Running", states=[{"running": {}}]),
+        ]
+    }
+    env = os.environ.copy()
+    env["STALE_FAILED_JOB_NAMES"] = "cron-a-old"
+    pods_result = subprocess.run(
+        [sys.executable, "-c", _extract_schedule_classifier("pods")],
+        input=json.dumps(pods),
+        text=True,
+        capture_output=True,
+        check=True,
+        env=env,
+    )
+    assert "RAW_BAD_PODS 7" in pods_result.stdout
+    assert "STALE_BAD_PODS_IGNORED 1" in pods_result.stdout
+    assert "BAD_PODS 6" in pods_result.stdout
+
+
+def test_cold_start_schedule_pipeline_failure_is_not_masked() -> None:
+    result = subprocess.run(
+        [
+            "bash",
+            "-o",
+            "pipefail",
+            "-c",
+            "(printf '{}'; exit 9) | python3 -c 'import json,sys; json.load(sys.stdin)'",
+        ],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert result.returncode == 9
+
+
 def test_recovery_scorecard_bounds_offsite_evidence_ssh() -> None:
     text = RECOVERY_SCORECARD.read_text(encoding="utf-8")