from __future__ import annotations

import importlib.util
import json
import subprocess
import sys
from pathlib import Path


SCRIPT_ROOT = Path(__file__).resolve().parents[1]
EXPORTER_PATH = SCRIPT_ROOT / "host-runaway-process-exporter.py"
REMEDIATION_PATH = SCRIPT_ROOT / "host-runaway-process-remediation.py"
CONTROLLER_PATH = SCRIPT_ROOT / "host-sustained-load-controller.py"
EVIDENCE_PATH = SCRIPT_ROOT / "host-sustained-load-evidence.py"


def load_exporter():
    spec = importlib.util.spec_from_file_location("host_runaway_process_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_classifies_orphan_stockplatform_headless_group() -> None:
    exporter = load_exporter()
    rows = exporter.parse_ps_rows(
        """
        100 1 100 100 7200 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa
        101 100 100 100 7190 55.0 S chromium /opt/chrome/chromium --type=renderer /tmp/stockplatform-review-bulk-ux-aa
        200 10 200 200 600 90.0 S node pnpm --filter @awoooi/web build
        """
    )

    groups = exporter.classify_groups(rows, min_age_seconds=1800, min_cpu_percent=50)

    assert len(groups) == 1
    assert groups[0].rule_id == "stockplatform_headless_smoke"
    assert groups[0].pgid == 100
    assert groups[0].orphan_reason == "ppid_1"
    assert groups[0].cpu_percent == 120.0
    assert len(groups[0].rows) == 2


def test_ignores_non_orphan_or_young_browser_processes() -> None:
    exporter = load_exporter()
    rows = exporter.parse_ps_rows(
        """
        100 99 100 100 7200 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa
        101 100 100 100 7190 55.0 S chromium /opt/chrome/chromium /tmp/stockplatform-review-bulk-ux-aa
        300 1 300 300 60 120.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-bb
        """
    )

    assert exporter.classify_groups(rows, min_age_seconds=1800, min_cpu_percent=50) == []


def test_parses_bsd_elapsed_time_for_local_smoke() -> None:
    exporter = load_exporter()
    rows = exporter.parse_ps_rows(
        """
        100 1 100 100 01:00:00 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa
        101 100 100 100 2-00:00:10 55.0 S chromium /opt/chrome/chromium /tmp/stockplatform-review-bulk-ux-aa
        """
    )

    assert rows[0].etimes == 3600
    assert rows[1].etimes == 172810


def test_renders_ci_load_and_swap_without_authorizing_repair(tmp_path: Path) -> None:
    exporter = load_exporter()
    groups = exporter.classify_groups(
        exporter.parse_ps_rows(
            "100 1 100 100 7200 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa"
        ),
        min_age_seconds=1800,
        min_cpu_percent=50,
    )
    metrics = exporter.render_metrics(
        host="110",
        groups=groups,
        active_action_containers=3,
        active_action_process_load=exporter.ActiveCiLoad(
            group_count=2,
            process_count=4,
            cpu_percent=188.5,
            oldest_age_seconds=240,
        ),
        min_age_seconds=1800,
        min_cpu_percent=50,
        now=123,
        load_ratio=1.25,
        swap_ratio=1.0,
    )

    assert 'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1' in metrics
    assert 'awoooi_host_gitea_actions_active_container_count{host="110"} 3' in metrics
    assert 'awoooi_host_gitea_actions_active_process_group_count{host="110"} 2' in metrics
    assert 'awoooi_host_gitea_actions_active_process_count{host="110"} 4' in metrics
    assert 'awoooi_host_gitea_actions_active_process_cpu_percent{host="110"} 188.500000' in metrics
    assert 'awoooi_host_gitea_actions_active_process_oldest_age_seconds{host="110"} 240' in metrics
    assert 'awoooi_host_swap_used_ratio{host="110"} 1.000000' in metrics
    assert 'awoooi_host_runaway_process_remediation_authorized{host="110"} 0' in metrics
    assert 'rule="stockplatform_headless_smoke"' in metrics


def test_counts_modern_gitea_action_container_names(tmp_path: Path) -> None:
    exporter = load_exporter()
    docker_file = tmp_path / "docker.txt"
    docker_file.write_text(
        "\n".join(
            [
                "GITEA-ACTIONS-TASK-123",
                "awoooi-cd-5901-1-e2e-smoke",
                "awoooi-cd-5873-1-source-link-smoke",
                "awoooi-code-review-3323-1-ai-code-review",
                "gitea",
                "stockplatform-v2-api-1",
            ]
        ),
        encoding="utf-8",
    )

    assert exporter.active_gitea_action_containers(docker_file) == 4


def test_counts_buildkit_runner_process_load() -> None:
    exporter = load_exporter()
    rows = exporter.parse_ps_rows(
        """
        100 10 100 100 240 0.0 S bash bash --noprofile --norc -e -o pipefail /home/wooo/.cache/act/14cc/act/workflow/8.sh
        101 100 100 100 239 1.0 S docker docker build -f apps/web/Dockerfile .
        102 101 100 100 239 2.0 S docker-buildx /home/wooo/.docker/cli-plugins/docker-buildx buildx build -f apps/web/Dockerfile .
        200 150 200 200 210 12.5 S turbo turbo build --filter=@awoooi/web --concurrency=1
        201 200 200 200 200 145.0 S node node /app/apps/web/node_modules/.bin/../next/dist/bin/next build
        300 1 300 300 9999 0.1 S act_runner act_runner daemon --config /config.yaml
        400 1 400 400 120 30.0 S node node apps/web/server.js
        """
    )

    load = exporter.active_gitea_action_process_load(rows)

    assert load.group_count == 2
    assert load.process_count == 5
    assert load.cpu_percent == 160.5
    assert load.oldest_age_seconds == 240


def test_ignores_the_host_pressure_gate_process_group() -> None:
    exporter = load_exporter()
    rows = exporter.parse_ps_rows(
        """
        100 10 100 100 240 0.0 S bash bash --noprofile --norc -e -o pipefail /home/wooo/.cache/act/14cc/act/workflow/2.sh
        101 100 100 100 239 0.0 S bash bash scripts/ci/wait-host-web-build-pressure.sh
        102 101 100 100 238 0.0 S sleep sleep 10
        200 150 200 200 210 12.5 S turbo turbo build --filter=@awoooi/web --concurrency=1
        """
    )

    load = exporter.active_gitea_action_process_load(rows)

    assert load.group_count == 1
    assert load.process_count == 1
    assert load.cpu_percent == 12.5
    assert load.oldest_age_seconds == 210


def test_remediation_defaults_to_dry_run(tmp_path: Path) -> None:
    ps_file = tmp_path / "ps.txt"
    ps_file.write_text(
        "999999 1 999999 999999 7200 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa\n",
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(REMEDIATION_PATH),
            "--ps-file",
            str(ps_file),
            "--rule",
            "stockplatform_headless_smoke",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    assert '"mode": "dry_run"' in result.stdout
    assert '"runtime_gate": 0' in result.stdout
    assert '"action": "dry_run"' in result.stdout


def test_remediation_refuses_apply_without_gates(tmp_path: Path) -> None:
    ps_file = tmp_path / "ps.txt"
    ps_file.write_text(
        "999999 1 999999 999999 7200 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa\n",
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(REMEDIATION_PATH),
            "--ps-file",
            str(ps_file),
            "--apply",
            "--rule",
            "stockplatform_headless_smoke",
        ],
        capture_output=True,
        text=True,
    )

    assert result.returncode != 0
    assert "Refusing apply" in result.stderr
    assert "--controlled-apply-id" in result.stderr
    assert "--confirm-apply" in result.stderr
    assert "--post-apply-verifier" in result.stderr


def test_remediation_accepts_controlled_apply_gate_without_owner_gate(tmp_path: Path) -> None:
    ps_file = tmp_path / "ps.txt"
    ps_file.write_text(
        "100 1 1 100 7200 65.0 S chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa\n",
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(REMEDIATION_PATH),
            "--ps-file",
            str(ps_file),
            "--apply",
            "--confirm-apply",
            "--rule",
            "stockplatform_headless_smoke",
            "--controlled-apply-id",
            "CAP-20260701-HOSTLOAD",
            "--evidence-ref",
            "HostLoadAverageSustainedHigh:110",
            "--post-apply-verifier",
            "scripts/ops/host-sustained-load-controller.py --host 110 --json",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    assert '"mode": "apply_sigterm"' in result.stdout
    assert '"runtime_gate": 1' in result.stdout
    assert '"controlled_apply_id": "CAP-20260701-HOSTLOAD"' in result.stdout
    assert '"owner_approval_id": ""' in result.stdout
    assert '"blocked_reason": "unsafe_pgid"' in result.stdout
    assert '"missing_process_group_count": 0' in result.stdout
    assert '"signal_error_count": 0' in result.stdout
    assert '"signaled_process_group_count": 0' in result.stdout


def test_sustained_load_controller_routes_orphan_browser_to_controlled_remediation(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.2',
                'awoooi_host_swap_used_ratio{host="110"} 0.1',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 0',
                'awoooi_host_gitea_actions_active_container_count{host="110"} 0',
                'awoooi_host_gitea_actions_active_process_group_count{host="110"} 0',
                'awoooi_host_runaway_browser_orphan_group_count{host="110",rule="stockplatform_headless_smoke",min_age_seconds="1800",min_cpu_percent="50"} 1',
                'awoooi_host_runaway_browser_orphan_cpu_percent{host="110",rule="stockplatform_headless_smoke",min_age_seconds="1800",min_cpu_percent="50"} 155.5',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--json",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    payload = json.loads(result.stdout)
    assert payload["classification"] == "controlled_orphan_browser_remediation_ready"
    assert payload["controlled_apply_allowed"] is True
    assert "host-runaway-process-remediation.py --rule stockplatform_headless_smoke" in payload["commands"]["dry_run"]
    assert "--controlled-apply-id" in payload["commands"]["controlled_apply"]
    assert payload["operation_boundaries"]["process_signal_performed"] is False


def test_sustained_load_controller_keeps_ci_saturation_on_runner_path(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.0',
                'awoooi_host_swap_used_ratio{host="110"} 0.1',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 0',
                'awoooi_host_gitea_actions_active_container_count{host="110"} 2',
                'awoooi_host_gitea_actions_active_process_group_count{host="110"} 1',
                'awoooi_host_gitea_actions_active_process_cpu_percent{host="110"} 180.0',
                'awoooi_host_gitea_actions_active_process_oldest_age_seconds{host="110"} 1900',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--json",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    payload = json.loads(result.stdout)
    assert payload["classification"] == "controlled_ci_runner_saturation_guarded"
    assert payload["controlled_apply_allowed"] is True
    assert "fail_closed" in payload["commands"]["controlled_apply"]
    assert "process_kill" not in payload["commands"]["controlled_apply"]


def test_sustained_load_controller_blocks_monitor_authority_violation(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.0',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 1',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--json",
        ],
        capture_output=True,
        text=True,
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_monitor_authority_violation"
    assert payload["controlled_apply_allowed"] is False


def test_sustained_load_controller_routes_gitea_backlog_from_docker_metrics(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.5',
                'awoooi_host_swap_used_ratio{host="110"} 0.1',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 0',
                'awoooi_host_gitea_actions_active_container_count{host="110"} 0',
                'awoooi_host_gitea_actions_active_process_group_count{host="110"} 0',
                'awoooi_host_runaway_browser_orphan_group_count{host="110",rule="stockplatform_headless_smoke",min_age_seconds="1800",min_cpu_percent="50"} 0',
            ]
        ),
        encoding="utf-8",
    )
    docker_file = tmp_path / "docker.prom"
    docker_file.write_text(
        "\n".join(
            [
                'docker_container_cpu_cores{host="110",container_name="gitea"} 3.4',
                'docker_container_cpu_cores{host="110",container_name="redis"} 0.2',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--docker-stats-file",
            str(docker_file),
            "--json",
        ],
        capture_output=True,
        text=True,
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_gitea_queue_or_hook_backlog_requires_playbook"
    assert payload["readback"]["top_container_cpu"]["container_name"] == "gitea"
    assert payload["controlled_apply_allowed"] is False
    assert "host-sustained-load-evidence.py" in payload["commands"]["dry_run"]


def test_sustained_load_controller_ignores_stale_docker_stats_attribution(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.5',
                'awoooi_host_swap_used_ratio{host="110"} 0.1',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 0',
                'awoooi_host_gitea_actions_active_container_count{host="110"} 0',
                'awoooi_host_gitea_actions_active_process_group_count{host="110"} 0',
                'awoooi_host_runaway_browser_orphan_group_count{host="110",rule="stockplatform_headless_smoke",min_age_seconds="1800",min_cpu_percent="50"} 0',
                'node_textfile_mtime_seconds{file="/host/home/wooo/node_exporter_textfiles/docker_stats.prom"} 1000',
                'node_time_seconds 5000',
            ]
        ),
        encoding="utf-8",
    )
    docker_file = tmp_path / "docker.prom"
    docker_file.write_text(
        "\n".join(
            [
                'docker_container_cpu_cores{host="110",container_name="gitea"} 3.4',
                'docker_container_cpu_cores{host="110",container_name="redis"} 0.2',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--docker-stats-file",
            str(docker_file),
            "--json",
        ],
        capture_output=True,
        text=True,
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_unknown_sustained_load_requires_source_specific_playbook"
    assert payload["readback"]["docker_stats"]["fresh"] is False
    assert payload["readback"]["top_container_cpu"] is None
    assert payload["readback"]["top_container_cpu_untrusted"]["container_name"] == "gitea"
    assert payload["controlled_apply_allowed"] is False


def test_sustained_load_controller_routes_unknown_load_to_sanitized_evidence(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.0',
                'awoooi_host_swap_used_ratio{host="110"} 0.1',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 0',
                'awoooi_host_gitea_actions_active_container_count{host="110"} 0',
                'awoooi_host_gitea_actions_active_process_group_count{host="110"} 0',
                'awoooi_host_runaway_browser_orphan_group_count{host="110",rule="stockplatform_headless_smoke",min_age_seconds="1800",min_cpu_percent="50"} 0',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--json",
        ],
        capture_output=True,
        text=True,
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert payload["classification"] == "blocked_unknown_sustained_load_requires_source_specific_playbook"
    assert payload["controlled_apply_allowed"] is False
    assert "host-sustained-load-evidence.py" in payload["commands"]["dry_run"]
    assert payload["operation_boundaries"]["process_signal_performed"] is False


def test_sustained_load_evidence_emits_sanitized_gitea_recommendation(tmp_path: Path) -> None:
    ps_file = tmp_path / "ps.txt"
    ps_file.write_text(
        "\n".join(
            [
                "100 1 100 7200 280.0 1.0 gitea /usr/local/bin/gitea web --config /home/wooo/gitea/app.ini",
                "200 1 200 180 15.0 0.5 systemd systemctl show gitea-act-runner-host.service",
            ]
        ),
        encoding="utf-8",
    )
    docker_file = tmp_path / "docker.prom"
    docker_file.write_text(
        'docker_container_cpu_cores{host="110",container_name="gitea"} 3.4\n',
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(EVIDENCE_PATH),
            "--host",
            "110",
            "--ps-file",
            str(ps_file),
            "--docker-stats-file",
            str(docker_file),
            "--json",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    payload = json.loads(result.stdout)
    assert payload["schema_version"] == "host_sustained_load_sanitized_evidence_v1"
    assert payload["recommendation"] == "gitea_queue_or_hook_backlog_playbook"
    assert payload["redaction"]["raw_command_lines_emitted"] is False
    assert payload["operation_boundaries"]["host_write_performed"] is False
    assert "/home/wooo" not in result.stdout


def test_sustained_load_evidence_keeps_stale_container_samples_untrusted(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'node_textfile_mtime_seconds{file="/host/home/wooo/node_exporter_textfiles/docker_stats.prom"} 1000',
                'node_time_seconds 5000',
            ]
        ),
        encoding="utf-8",
    )
    docker_file = tmp_path / "docker.prom"
    docker_file.write_text(
        'docker_container_cpu_cores{host="110",container_name="gitea"} 3.4\n',
        encoding="utf-8",
    )
    ps_file = tmp_path / "ps.txt"
    ps_file.write_text(
        "100 1 100 120 5.0 1.0 python python monitor.py\n",
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(EVIDENCE_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--ps-file",
            str(ps_file),
            "--docker-stats-file",
            str(docker_file),
            "--json",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    payload = json.loads(result.stdout)
    assert payload["recommendation"] != "gitea_queue_or_hook_backlog_playbook"
    assert payload["readback"]["docker_stats"]["fresh"] is False
    assert payload["top_containers"] == []
    assert payload["top_containers_untrusted"][0]["container_name"] == "gitea"
    assert payload["operation_boundaries"]["host_write_performed"] is False


def test_sustained_load_controller_routes_unknown_load_to_sanitized_evidence(tmp_path: Path) -> None:
    metrics_file = tmp_path / "host.prom"
    metrics_file.write_text(
        "\n".join(
            [
                'awoooi_host_runaway_process_monitor_up{host="110",mode="read_only"} 1',
                'awoooi_host_load5_per_core{host="110"} 2.4',
                'awoooi_host_swap_used_ratio{host="110"} 0.1',
                'awoooi_host_runaway_process_remediation_authorized{host="110"} 0',
            ]
        ),
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(CONTROLLER_PATH),
            "--host",
            "110",
            "--metrics-file",
            str(metrics_file),
            "--json",
        ],
        capture_output=True,
        text=True,
    )

    assert result.returncode == 75
    payload = json.loads(result.stdout)
    assert (
        payload["classification"]
        == "blocked_unknown_sustained_load_requires_source_specific_playbook"
    )
    assert payload["controlled_apply_allowed"] is False
    assert "host-sustained-load-evidence.py" in payload["commands"]["dry_run"]
    assert payload["operation_boundaries"]["host_write_performed"] is False


def test_sustained_load_evidence_sanitizes_process_details(tmp_path: Path) -> None:
    ps_file = tmp_path / "ps.txt"
    ps_file.write_text(
        "\n".join(
            [
                "101 1 101 7200 65.0 2.5 chrome /opt/chrome/chrome --headless --user-data-dir=/tmp/stockplatform-review-bulk-ux-aa --url=https://example.invalid/token",
                "102 1 102 3600 20.0 1.0 node node /srv/private/app/server.js --api-key=SECRET",
            ]
        ),
        encoding="utf-8",
    )
    docker_stats_file = tmp_path / "docker.prom"
    docker_stats_file.write_text(
        'docker_container_cpu_cores{host="110",container_name="gitea"} 3.2\n',
        encoding="utf-8",
    )

    result = subprocess.run(
        [
            sys.executable,
            str(EVIDENCE_PATH),
            "--host",
            "110",
            "--ps-file",
            str(ps_file),
            "--docker-stats-file",
            str(docker_stats_file),
            "--json",
        ],
        check=True,
        capture_output=True,
        text=True,
    )

    payload = json.loads(result.stdout)
    assert payload["schema_version"] == "host_sustained_load_sanitized_evidence_v1"
    assert payload["recommendation"] == "gitea_queue_or_hook_backlog_playbook"
    assert payload["redaction"]["raw_command_lines_emitted"] is False
    assert payload["redaction"]["workspace_paths_emitted"] is False
    assert payload["redaction"]["urls_emitted"] is False
    assert payload["operation_boundaries"]["host_write_performed"] is False
    assert "https://example.invalid/token" not in result.stdout
    assert "/tmp/stockplatform-review-bulk-ux-aa" not in result.stdout
    assert "SECRET" not in result.stdout
    assert {item["family"] for item in payload["top_process_families"]} >= {
        "headless_browser",
        "node_service",
    }