fix(recovery): drain host cpu pressure gates

docs/LOGBOOK.md

@@ -51115,3 +51115,35 @@ production browser smoke:
 
 **下一步**：
 - commit / push 後讀回最新 queue；若 CD / Harbor repair 仍 Waiting，下一個主線是 non110 controlled runner readiness receipt / registration evidence，而不是恢復 110 generic runner。
+
+## 2026-07-01 — 08:22 P0 188 CD/API test drain 與 110 systemd exporter 降壓修法
+
+**完成內容**：
+- 188 高 CPU 二次定位：補完 conversation event indexes 後，`k3s-postgres-recovery` 再度衝高不是索引失敗，而是 `pg_dump` 正在執行 `COPY public.asset_coverage_snapshot ... TO stdout`；DB 端 `application_name=pg_dump`、host 端 pid `3170704`，對應 process group `3170701`。
+- 已受控 `TERM` 該 `pg_dump` process group；這是讀取型備份 / 匯出，不是 DB write。post-check：`pg_dump` 消失，長跑 `COPY asset_coverage_snapshot` 消失。
+- 188 隨後出現 `awoooi-cd-7359/7360-1-api-tests`，代表 active non-110 runner 仍會重生 API tests 並間接打 production PostgreSQL；已停止 `awoooi-non110-runner.service`、`awoooi-non110-runner-autostart.path`、`stockplatform-ubuntu-runner` 與現有 `awoooi-cd-*api-tests` 壓力來源。
+- 188 post-drain readback：`awoooi-non110-runner.service=inactive`、autostart path `inactive`、沒有 `awoooi-cd-*`、沒有 `pg_dump`；即時 `docker stats` 顯示 `k3s-postgres-recovery=7.81%`，後續 `docker_stats.prom` 顯示 `docker_container_cpu_cores{k3s-postgres-recovery}=0.406900`。
+- `scripts/ci/wait-host-web-build-pressure.sh` 新增 Docker textfile pressure gate：當 `k3s-postgres-recovery` CPU cores 超過 `HOST_WEB_BUILD_PRESSURE_MAX_POSTGRES_CPU_CORES`（預設 `2.0`）時，CD 在 API tests 前 fail-fast，不再把 188 production PostgreSQL 打滿。
+- 110 高 CPU / 控制面定位：SSH command 持續 timeout，journal 出現 `systemd-logind` / DBus pending replies 爆滿、`systemctl list-*` timeout、`sshd MaxStartups throttling`；node exporter 顯示 `systemd_units.prom` 仍由舊版 exporter 產出 5 個 `systemctl show ... timed out after 10 seconds` label。
+- `scripts/ops/systemd-units-textfile-exporter.py` 新增 non-overlap lock、`systemctl show` 預設 timeout 降到 `2s`、錯誤 label 壓成固定 `timeout` / `exit_N`、連續 timeout 達預算後本輪剩餘 units 標 `scrape_skipped`，避免 exporter 重疊把 DBus/logind 打爆。
+- 新增 receipt：`docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json`。
+
+**本地驗證結果**：
+- `python3.11 -m py_compile scripts/ops/systemd-units-textfile-exporter.py`：passed。
+- `bash -n scripts/ci/wait-host-web-build-pressure.sh scripts/ops/systemd-units-textfile-exporter.py`：passed。
+- `python3.11 -m pytest scripts/ops/tests/test_systemd_units_textfile_exporter.py -q`：`2 passed`。
+- `python3.11 -m pytest scripts/ops/tests/test_systemd_units_textfile_exporter.py ops/runner/test_cd_controlled_runtime_profile.py -q`：`34 passed`。
+- pressure gate fixture：`k3s-postgres-recovery=6.5 cores` 時正確拒絕啟動。
+- `git diff --check`：passed。
+
+**live truth**：
+- 188 已降載；runner 暫時 drained，待 source gate 推送後才能安全恢復。
+- 110 load5 從約 `39.14` 降到 `18.85`，`node_procs_running=6`，但仍高於門檻；`gitea=3.4019` cores 且 SSH control path 仍 timeout，所以 110 live exporter 修法尚未套用。
+
+**仍維持**：
+- 沒有讀 secret / token / `.env` / raw sessions / SQLite / auth；沒有讀 `.runner` 內容。
+- 沒有使用 GitHub / gh / GitHub API / GitHub Actions。
+- 沒有重啟主機，沒有 Docker / Nginx / K3s / DB restart，沒有 workflow_dispatch，沒有 DROP / TRUNCATE / restore / prune。
+
+**下一步**：
+- rebase / push source 修法；110 控制面恢復後第一個 runtime 動作是套用新版 `systemd-units-textfile-exporter.py` 或暫停舊 exporter，再回讀 DBus/logind、systemd_units.prom 與 Gitea CPU。188 runner 只在 host pressure gate / queue readback 綠燈後恢復。

docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json

@@ -0,0 +1,99 @@
+{
+  "schema_version": "awoooi_host_cpu_pressure_drain_readback_v1",
+  "generated_at": "2026-07-01T08:19:00+08:00",
+  "status": "partial_110_control_path_blocked_188_drained",
+  "scope": {
+    "hosts": ["188", "110"],
+    "incident_family": "post_reboot_host_cpu_pressure",
+    "source_branch": "codex/p0-006-stockplatform-postgres-readback-20260630"
+  },
+  "operation_boundaries": {
+    "github_used": false,
+    "secret_or_runner_token_read": false,
+    "raw_sessions_or_sqlite_read": false,
+    "host_reboot_performed": false,
+    "docker_daemon_restart_performed": false,
+    "nginx_restart_performed": false,
+    "k3s_restart_performed": false,
+    "database_restart_performed": false,
+    "destructive_database_operation_performed": false,
+    "workflow_dispatch_performed": false
+  },
+  "live_actions": [
+    {
+      "host": "188",
+      "target_selector": "pg_dump process group 3170701 / pg_dump pid 3170704",
+      "action": "TERM process group",
+      "reason": "pg_dump COPY public.asset_coverage_snapshot TO stdout caused k3s-postgres-recovery CPU pressure",
+      "rollback": "rerun the controlled backup/export in a low-pressure window",
+      "post_apply_readback": "pg_dump process disappeared and long-running COPY query disappeared"
+    },
+    {
+      "host": "188",
+      "target_selector": "awoooi-non110-runner.service, awoooi-non110-runner-autostart.path, awoooi-cd-*api-tests, stockplatform-ubuntu-runner",
+      "action": "stop runner/test pressure only",
+      "reason": "active CD API tests were repeatedly spawning and driving production PostgreSQL CPU",
+      "rollback": "restart the non-110 runner only after host pressure guard and source CD gate are green",
+      "post_apply_readback": "runner inactive, no awoooi-cd-* containers, k3s-postgres-recovery docker stats 7.81 percent"
+    }
+  ],
+  "source_fixes": [
+    {
+      "path": "scripts/ci/wait-host-web-build-pressure.sh",
+      "change": "block CD startup when docker_stats.prom reports k3s-postgres-recovery CPU cores above HOST_WEB_BUILD_PRESSURE_MAX_POSTGRES_CPU_CORES",
+      "evidence": "fixture with k3s-postgres-recovery=6.5 cores fails fast before API tests"
+    },
+    {
+      "path": "scripts/ops/systemd-units-textfile-exporter.py",
+      "change": "shorten systemctl show timeout, compact error labels, stop probing after timeout budget, and add non-overlap lock",
+      "evidence": "pytest covers timeout budget and busy lock"
+    }
+  ],
+  "readback": {
+    "host_188": {
+      "pre_drain_signals": {
+        "k3s_postgres_recovery_docker_stats_percent": "658.65",
+        "awoooi_cd_api_tests_container": "awoooi-cd-7359-1-api-tests / awoooi-cd-7360-1-api-tests",
+        "active_pg_query": "COPY public.asset_coverage_snapshot TO stdout",
+        "pg_dump_application_name": "pg_dump"
+      },
+      "post_drain_signals": {
+        "runner_service": "inactive",
+        "runner_autostart_path": "inactive",
+        "awoooi_cd_containers": "none",
+        "pg_dump_process": "none",
+        "k3s_postgres_recovery_docker_stats_percent": "7.81",
+        "docker_stats_textfile_k3s_postgres_recovery_cores": "0.406900"
+      }
+    },
+    "host_110": {
+      "signals": {
+        "load5_initial": "39.14",
+        "load5_latest": "18.85",
+        "node_procs_running_latest": "6",
+        "gitea_container_cpu_cores": "3.4019",
+        "ssh_control_path": "timeout",
+        "systemd_dbus_symptom": "systemd-logind pending replies exhausted; systemctl list/show timeout",
+        "systemd_units_exporter_current_live_state": "old exporter still emits 10s systemctl timeout labels"
+      },
+      "not_completed_live": [
+        "could not apply pkill/exporter drain on 110 because SSH command sessions timeout",
+        "could not restart or reexec systemd/logind because that would require a control path that is currently unavailable"
+      ]
+    }
+  },
+  "verification": {
+    "py_compile": "passed: scripts/ops/systemd-units-textfile-exporter.py",
+    "bash_n": "passed: scripts/ci/wait-host-web-build-pressure.sh scripts/ops/systemd-units-textfile-exporter.py",
+    "pytest": [
+      "scripts/ops/tests/test_systemd_units_textfile_exporter.py: 2 passed",
+      "scripts/ops/tests/test_systemd_units_textfile_exporter.py + ops/runner/test_cd_controlled_runtime_profile.py: 34 passed"
+    ],
+    "diff_check": "passed"
+  },
+  "next_actions": [
+    "apply the systemd exporter source fix to 110 once SSH/control path is available",
+    "keep 188 non-110 runner drained until the pushed CD gate is in place and host pressure readback is green",
+    "continue 110 Gitea queue / awoooi-host controlled lane recovery without generic runner restore"
+  ]
+}

docs/runbooks/FULL-STACK-COLD-START-SOP.md

@@ -2365,3 +2365,40 @@ bash -n scripts/reboot-recovery/awoooi-startup.sh
 DATABASE_URL=sqlite+aiosqlite:////tmp/awoooi-codex-api-test.db PYTHONPATH=apps/api \
   python3.11 -m pytest scripts/reboot-recovery/tests/test_188_host_hygiene_checklist.py -q
 ```
+
+### 16.2 188 CD/API test 與 110 systemd exporter 壓力不得混成一般 CPU 高
+
+2026-07-01 post-reboot pressure 回讀補充兩個獨立模式：
+
+1. 188 `k3s-postgres-recovery` 高 CPU 可能不是 DB service 壞掉，而是 CD / backup / export 正在讀 production DB。
+2. 110 `systemd-logind` / DBus 卡住時，`systemd_units.prom` exporter 反覆 `systemctl show` runner units 會放大控制面壓力，讓 SSH / `systemctl list-*` timeout。
+
+188 判讀順序：
+
+| Evidence | Decision |
+|----------|----------|
+| `pg_stat_activity.application_name=pg_dump` 且 query 為 `COPY ... TO stdout` | 視為備份 / 匯出壓力；先定位 host process group，只允許停止該讀取型 export，不重啟 PostgreSQL / Docker。 |
+| `docker ps` 出現 `awoooi-cd-*-api-tests` 且 `k3s-postgres-recovery` 同步高 CPU | 視為 CD API test 壓力；先 drain non-prod runner / test container，再把 CD gate 補上 188 Docker metrics。 |
+| `docker_stats.prom` 中 `docker_container_cpu_cores{container_name="k3s-postgres-recovery"}` 超過 `2.0` | `scripts/ci/wait-host-web-build-pressure.sh` 必須 fail-fast，不得啟動新 API tests。 |
+
+110 判讀順序：
+
+| Evidence | Decision |
+|----------|----------|
+| journal 出現 `systemd-logind` pending replies exhausted / `systemctl list-*` timeout / `sshd MaxStartups throttling` | 視為 control-plane saturation；停止並行 SSH，改成單線控制或 Prometheus readback。 |
+| `systemd_units.prom` 仍出現多個 `systemctl show ... timed out after 10 seconds` label | 視為舊 exporter 放大 DBus 壓力；套用新版 exporter 或暫停舊 exporter，不要先重啟主機。 |
+| `gitea` 固定接近 CPU cap 且 no active Actions containers | 視為 Gitea queue / hook backlog；不要恢復 generic runner，不要把 host pressure guard 改 warn-only。 |
+
+新的 source-of-truth：
+
+- `scripts/ci/wait-host-web-build-pressure.sh`
+- `scripts/ops/systemd-units-textfile-exporter.py`
+- `scripts/ops/tests/test_systemd_units_textfile_exporter.py`
+- `docs/operations/host-cpu-pressure-drain-readback-2026-07-01.snapshot.json`
+
+驗證命令：
+
+```bash
+bash -n scripts/ci/wait-host-web-build-pressure.sh scripts/ops/systemd-units-textfile-exporter.py
+python3.11 -m pytest scripts/ops/tests/test_systemd_units_textfile_exporter.py ops/runner/test_cd_controlled_runtime_profile.py -q
+```

scripts/ci/wait-host-web-build-pressure.sh

@@ -34,7 +34,11 @@ MAX_CI_CPU_PERCENT="${HOST_WEB_BUILD_PRESSURE_MAX_CI_CPU_PERCENT:-250}"
 MAX_ACTIVE_CI_PROCESS_GROUPS="${HOST_WEB_BUILD_PRESSURE_MAX_ACTIVE_CI_PROCESS_GROUPS:-1}"
 MAX_ACTIVE_CI_CONTAINERS="${HOST_WEB_BUILD_PRESSURE_MAX_ACTIVE_CI_CONTAINERS:-1}"
 MAX_ORPHAN_BROWSER_GROUPS="${HOST_WEB_BUILD_PRESSURE_MAX_ORPHAN_BROWSER_GROUPS:-0}"
+MAX_POSTGRES_CPU_CORES="${HOST_WEB_BUILD_PRESSURE_MAX_POSTGRES_CPU_CORES:-2.0}"
+POSTGRES_CONTAINER_NAME="${HOST_WEB_BUILD_PRESSURE_POSTGRES_CONTAINER:-k3s-postgres-recovery}"
 METRICS_FILE="${HOST_RUNAWAY_PROCESS_METRICS_FILE:-${HOST_WEB_BUILD_PRESSURE_METRICS_FILE:-/home/wooo/node_exporter_textfiles/host_runaway_process.prom}}"
+DEFAULT_DOCKER_METRICS_FILE="/home/$(id -un)/node_exporter_textfiles/docker_stats.prom"
+DOCKER_METRICS_FILE="${HOST_WEB_BUILD_PRESSURE_DOCKER_METRICS_FILE:-$DEFAULT_DOCKER_METRICS_FILE}"
 EXPORTER="${HOST_RUNAWAY_PROCESS_EXPORTER:-/home/wooo/scripts/host-runaway-process-exporter.py}"
 
 default_ps_command() {
@@ -111,6 +115,24 @@ metric_sum() {
   ' "$METRICS_FILE"
 }
 
+docker_metric_labeled_value() {
+  local name="$1"
+  local label_key="$2"
+  local label_value="$3"
+  if [ ! -r "$DOCKER_METRICS_FILE" ]; then
+    return 1
+  fi
+  awk -v metric="$name" -v key="$label_key" -v val="$label_value" '
+    $1 ~ ("^" metric "\\{") && $0 ~ (key "=\"" val "\"") {
+      value = $NF
+    }
+    END {
+      if (value != "") print value
+      else exit 1
+    }
+  ' "$DOCKER_METRICS_FILE"
+}
+
 load5_per_core() {
   metric_value "awoooi_host_load5_per_core" 2>/dev/null || awk '
     BEGIN {
@@ -150,7 +172,7 @@ list_headless_smoke_pressure() {
 
 pressure_report() {
   local report=""
-  local load_ratio active_ci_cpu active_ci_groups active_ci_containers orphan_groups
+  local load_ratio active_ci_cpu active_ci_groups active_ci_containers orphan_groups postgres_cpu_cores
   load_ratio="$(load5_per_core 2>/dev/null || echo 0)"
   active_ci_cpu="$(metric_value "awoooi_host_gitea_actions_active_process_cpu_percent" 2>/dev/null || echo 0)"
   active_ci_groups="$(metric_value "awoooi_host_gitea_actions_active_process_group_count" 2>/dev/null || echo 0)"
@@ -160,6 +182,10 @@ pressure_report() {
       || metric_sum "awoooi_host_orphan_browser_group_count" 2>/dev/null \
       || echo 0
   )"
+  postgres_cpu_cores="$(
+    docker_metric_labeled_value "docker_container_cpu_cores" "container_name" "$POSTGRES_CONTAINER_NAME" 2>/dev/null \
+      || echo 0
+  )"
 
   if greater_than "$load_ratio" "$MAX_LOAD5_PER_CORE"; then
     report="${report}host load5/core ${load_ratio} > ${MAX_LOAD5_PER_CORE}"$'\n'
@@ -176,6 +202,9 @@ pressure_report() {
   if greater_than "$orphan_groups" "$MAX_ORPHAN_BROWSER_GROUPS"; then
     report="${report}orphan browser/smoke groups ${orphan_groups} > ${MAX_ORPHAN_BROWSER_GROUPS}"$'\n'
   fi
+  if greater_than "$postgres_cpu_cores" "$MAX_POSTGRES_CPU_CORES"; then
+    report="${report}${POSTGRES_CONTAINER_NAME} CPU cores ${postgres_cpu_cores} > ${MAX_POSTGRES_CPU_CORES}"$'\n'
+  fi
 
   local smoke_pressure
   smoke_pressure="$(list_headless_smoke_pressure || true)"

scripts/ops/systemd-units-textfile-exporter.py

@@ -11,6 +11,7 @@ needs direct visibility into systemd restarts, watchdog settings, and limits.
 from __future__ import annotations
 
 import os
+import fcntl
 import re
 import subprocess
 import tempfile
@@ -20,12 +21,15 @@ from pathlib import Path
 TEXTFILE_DIR = Path(os.environ.get("NODE_EXPORTER_TEXTFILE_DIR", "/var/lib/node_exporter/textfile_collector"))
 OUTPUT_NAME = "systemd_units.prom"
 HOST_LABEL = os.environ.get("AIOPS_HOST_LABEL", os.uname().nodename)
+LOCK_PATH = Path(os.environ.get("AIOPS_SYSTEMD_EXPORTER_LOCK", f"/tmp/{OUTPUT_NAME}.lock"))
 UNIT_NAMES = [
     unit.strip()
     for unit in os.environ.get("AIOPS_SYSTEMD_UNITS", "").split(",")
     if unit.strip()
 ]
 LABEL_RE = re.compile(r'["\\\n]')
+DEFAULT_SYSTEMCTL_TIMEOUT_SECONDS = 2.0
+DEFAULT_MAX_TIMEOUTS = 1
 
 
 def _escape_label(value: str) -> str:
@@ -61,6 +65,34 @@ def _parse_bytes(value: str) -> float:
         return 0.0
 
 
+def _systemctl_timeout_seconds() -> float:
+    raw = os.environ.get("AIOPS_SYSTEMD_SHOW_TIMEOUT_SECONDS", "")
+    if not raw:
+        return DEFAULT_SYSTEMCTL_TIMEOUT_SECONDS
+    try:
+        return max(0.25, float(raw))
+    except ValueError:
+        return DEFAULT_SYSTEMCTL_TIMEOUT_SECONDS
+
+
+def _max_timeouts() -> int:
+    raw = os.environ.get("AIOPS_SYSTEMD_MAX_TIMEOUTS", "")
+    if not raw:
+        return DEFAULT_MAX_TIMEOUTS
+    try:
+        return max(1, int(raw))
+    except ValueError:
+        return DEFAULT_MAX_TIMEOUTS
+
+
+def _compact_error(exc: Exception) -> str:
+    if isinstance(exc, subprocess.TimeoutExpired):
+        return "timeout"
+    if isinstance(exc, subprocess.CalledProcessError):
+        return f"exit_{exc.returncode}"
+    return exc.__class__.__name__
+
+
 def _show_unit(unit: str) -> dict[str, str]:
     result = subprocess.run(
         [
@@ -83,7 +115,7 @@ def _show_unit(unit: str) -> dict[str, str]:
         check=True,
         capture_output=True,
         text=True,
-        timeout=10,
+        timeout=_systemctl_timeout_seconds(),
     )
     values = {}
     for line in result.stdout.splitlines():
@@ -104,16 +136,32 @@ def collect() -> str:
         "# TYPE systemd_unit_cpu_quota_cores gauge",
         "# HELP systemd_unit_memory_max_bytes Systemd unit MemoryMax in bytes, 0 when unlimited.",
         "# TYPE systemd_unit_memory_max_bytes gauge",
+        "# HELP systemd_unit_exporter_timeout_budget_exhausted Whether this scrape stopped probing after repeated systemctl timeouts.",
+        "# TYPE systemd_unit_exporter_timeout_budget_exhausted gauge",
     ]
-    for unit in UNIT_NAMES:
+    timeout_count = 0
+    timeout_budget_exhausted = False
+    for index, unit in enumerate(UNIT_NAMES):
+        labels = f'host="{_escape_label(HOST_LABEL)}",unit="{_escape_label(unit)}"'
         try:
             values = _show_unit(unit)
         except Exception as exc:
-            labels = f'host="{_escape_label(HOST_LABEL)}",unit="{_escape_label(unit)}"'
-            lines.append(f'systemd_unit_info{{{labels},active_state="scrape_error",sub_state="{_escape_label(str(exc))}"}} 0')
+            reason = _compact_error(exc)
+            lines.append(f'systemd_unit_info{{{labels},active_state="scrape_error",sub_state="{_escape_label(reason)}"}} 0')
+            if reason == "timeout":
+                timeout_count += 1
+                if timeout_count >= _max_timeouts():
+                    timeout_budget_exhausted = True
+                    for skipped_unit in UNIT_NAMES[index + 1 :]:
+                        skipped_labels = (
+                            f'host="{_escape_label(HOST_LABEL)}",unit="{_escape_label(skipped_unit)}"'
+                        )
+                        lines.append(
+                            f'systemd_unit_info{{{skipped_labels},active_state="scrape_skipped",sub_state="systemctl_timeout_budget_exhausted"}} 0'
+                        )
+                    break
             continue
 
-        labels = f'host="{_escape_label(HOST_LABEL)}",unit="{_escape_label(unit)}"'
         active_state = values.get("ActiveState", "")
         sub_state = values.get("SubState", "")
         restarts = int(values.get("NRestarts") or 0)
@@ -127,20 +175,30 @@ def collect() -> str:
         lines.append(f"systemd_unit_watchdog_seconds{{{labels}}} {watchdog_seconds:.6f}")
         lines.append(f"systemd_unit_cpu_quota_cores{{{labels}}} {cpu_quota:.6f}")
         lines.append(f"systemd_unit_memory_max_bytes{{{labels}}} {memory_max:.0f}")
+    host_labels = f'host="{_escape_label(HOST_LABEL)}"'
+    lines.append(
+        f"systemd_unit_exporter_timeout_budget_exhausted{{{host_labels}}} {1 if timeout_budget_exhausted else 0}"
+    )
     return "\n".join(lines) + "\n"
 
 
 def main() -> None:
     if not UNIT_NAMES:
         raise SystemExit("AIOPS_SYSTEMD_UNITS is required")
-    TEXTFILE_DIR.mkdir(parents=True, exist_ok=True)
-    payload = collect()
-    with tempfile.NamedTemporaryFile("w", dir=TEXTFILE_DIR, delete=False, encoding="utf-8") as tmp:
-        tmp.write(payload)
-        tmp_path = Path(tmp.name)
-    output_path = TEXTFILE_DIR / OUTPUT_NAME
-    tmp_path.replace(output_path)
-    output_path.chmod(0o644)
+    LOCK_PATH.parent.mkdir(parents=True, exist_ok=True)
+    with LOCK_PATH.open("w") as lock_handle:
+        try:
+            fcntl.flock(lock_handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError:
+            return
+        TEXTFILE_DIR.mkdir(parents=True, exist_ok=True)
+        payload = collect()
+        with tempfile.NamedTemporaryFile("w", dir=TEXTFILE_DIR, delete=False, encoding="utf-8") as tmp:
+            tmp.write(payload)
+            tmp_path = Path(tmp.name)
+        output_path = TEXTFILE_DIR / OUTPUT_NAME
+        tmp_path.replace(output_path)
+        output_path.chmod(0o644)
 
 
 if __name__ == "__main__":

scripts/ops/tests/test_systemd_units_textfile_exporter.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import fcntl
+import importlib.util
+import subprocess
+import sys
+from pathlib import Path
+
+
+SCRIPT_ROOT = Path(__file__).resolve().parents[1]
+EXPORTER_PATH = SCRIPT_ROOT / "systemd-units-textfile-exporter.py"
+
+
+def load_exporter():
+    spec = importlib.util.spec_from_file_location("systemd_units_textfile_exporter", EXPORTER_PATH)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_stops_probe_after_systemctl_timeout_budget(monkeypatch) -> None:
+    exporter = load_exporter()
+    monkeypatch.setattr(exporter, "UNIT_NAMES", ["runner-a.service", "runner-b.service", "runner-c.service"])
+    monkeypatch.setenv("AIOPS_SYSTEMD_MAX_TIMEOUTS", "1")
+    calls: list[str] = []
+
+    def fake_show_unit(unit: str) -> dict[str, str]:
+        calls.append(unit)
+        raise subprocess.TimeoutExpired(cmd=["systemctl", "show", unit], timeout=2)
+
+    monkeypatch.setattr(exporter, "_show_unit", fake_show_unit)
+
+    metrics = exporter.collect()
+
+    assert calls == ["runner-a.service"]
+    assert 'unit="runner-a.service",active_state="scrape_error",sub_state="timeout"' in metrics
+    assert (
+        'unit="runner-b.service",active_state="scrape_skipped",'
+        'sub_state="systemctl_timeout_budget_exhausted"'
+    ) in metrics
+    assert (
+        'unit="runner-c.service",active_state="scrape_skipped",'
+        'sub_state="systemctl_timeout_budget_exhausted"'
+    ) in metrics
+    assert "Command '['" not in metrics
+    assert "systemd_unit_exporter_timeout_budget_exhausted" in metrics
+    assert metrics.rstrip().endswith(" 1")
+
+
+def test_main_returns_when_exporter_lock_is_busy(tmp_path: Path, monkeypatch) -> None:
+    exporter = load_exporter()
+    lock_path = tmp_path / "systemd_units.prom.lock"
+    output_dir = tmp_path / "textfiles"
+    monkeypatch.setattr(exporter, "LOCK_PATH", lock_path)
+    monkeypatch.setattr(exporter, "TEXTFILE_DIR", output_dir)
+    monkeypatch.setattr(exporter, "UNIT_NAMES", ["runner-a.service"])
+
+    lock_path.parent.mkdir(parents=True, exist_ok=True)
+    with lock_path.open("w") as lock_handle:
+        fcntl.flock(lock_handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+        exporter.main()
+
+    assert not (output_dir / exporter.OUTPUT_NAME).exists()