fix(recovery): preserve reboot evidence across delayed startup
This commit is contained in:
ogt
@@ -0,0 +1,21 @@
|
||||
[Unit]
|
||||
Description=AWOOOI external L0 outage and recovery supervisor
|
||||
Documentation=file:/opt/awoooi/reboot-recovery/docs/runbooks/REBOOT-RECOVERY-SOP.md
|
||||
Wants=network-online.target
|
||||
After=network-online.target
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
User=awoooi-l0
|
||||
Group=awoooi-l0
|
||||
WorkingDirectory=/opt/awoooi/reboot-recovery
|
||||
ExecStart=/usr/bin/python3 /opt/awoooi/reboot-recovery/scripts/reboot-recovery/external-l0-recovery-supervisor.py --apply --config /etc/awoooi/external-l0-recovery-supervisor.json --state-file /var/lib/awoooi/external-l0/state.json --artifact-dir /var/lib/awoooi/external-l0/evidence
|
||||
NoNewPrivileges=true
|
||||
PrivateTmp=true
|
||||
ProtectHome=true
|
||||
ProtectSystem=strict
|
||||
ReadWritePaths=/var/lib/awoooi/external-l0
|
||||
TimeoutStartSec=55
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
@@ -0,0 +1,12 @@
|
||||
[Unit]
|
||||
Description=Probe AWOOOI external L0 recovery state every 15 seconds
|
||||
|
||||
[Timer]
|
||||
OnBootSec=15s
|
||||
OnUnitInactiveSec=15s
|
||||
AccuracySec=2s
|
||||
Persistent=true
|
||||
Unit=awoooi-external-l0-recovery-supervisor.service
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
344
scripts/reboot-recovery/external-l0-recovery-supervisor.py
Normal file
344
scripts/reboot-recovery/external-l0-recovery-supervisor.py
Normal file
@@ -0,0 +1,344 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Off-LAN outage state machine for the Windows 99 recovery dependency chain."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
SCHEMA_VERSION = "external_l0_recovery_supervisor_v1"
|
||||
STATE_SCHEMA_VERSION = "external_l0_recovery_state_v1"
|
||||
VALID_LIFECYCLE_STATES = {
|
||||
"healthy",
|
||||
"suspected_outage",
|
||||
"outage_confirmed",
|
||||
"recovering",
|
||||
"recovered",
|
||||
}
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("--config", required=True, type=Path)
|
||||
parser.add_argument("--state-file", required=True, type=Path)
|
||||
parser.add_argument("--artifact-dir", required=True, type=Path)
|
||||
parser.add_argument("--samples-json", type=Path)
|
||||
parser.add_argument("--now")
|
||||
mode = parser.add_mutually_exclusive_group(required=True)
|
||||
mode.add_argument("--check", action="store_true")
|
||||
mode.add_argument("--apply", action="store_true")
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def now_utc(value: str | None) -> datetime:
|
||||
if not value:
|
||||
return datetime.now(timezone.utc)
|
||||
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
if parsed.tzinfo is None:
|
||||
parsed = parsed.replace(tzinfo=timezone.utc)
|
||||
return parsed.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def isoformat(value: datetime) -> str:
|
||||
return value.replace(microsecond=0).isoformat().replace("+00:00", "Z")
|
||||
|
||||
|
||||
def read_json(path: Path) -> dict[str, Any]:
|
||||
payload = json.loads(path.read_text(encoding="utf-8"))
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError(f"JSON root must be an object: {path}")
|
||||
return payload
|
||||
|
||||
|
||||
def write_json(path: Path, payload: dict[str, Any]) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
temporary = path.with_suffix(path.suffix + ".tmp")
|
||||
temporary.write_text(json.dumps(payload, ensure_ascii=True, indent=2) + "\n", encoding="utf-8")
|
||||
os.replace(temporary, path)
|
||||
|
||||
|
||||
def validate_config(config: dict[str, Any]) -> None:
|
||||
if config.get("schema_version") != "external_l0_recovery_supervisor_config_v1":
|
||||
raise ValueError("unsupported config schema_version")
|
||||
targets = config.get("targets")
|
||||
if not isinstance(targets, list) or len(targets) < 2:
|
||||
raise ValueError("at least two external targets are required")
|
||||
target_ids: set[str] = set()
|
||||
for target in targets:
|
||||
if not isinstance(target, dict):
|
||||
raise ValueError("each target must be an object")
|
||||
target_id = str(target.get("id") or "")
|
||||
url = str(target.get("url") or "")
|
||||
if not target_id or target_id in target_ids:
|
||||
raise ValueError("target ids must be present and unique")
|
||||
if not url.startswith("https://"):
|
||||
raise ValueError("external targets must use https")
|
||||
target_ids.add(target_id)
|
||||
|
||||
thresholds = config.get("thresholds") or {}
|
||||
for key in (
|
||||
"outage_consecutive_samples",
|
||||
"recovery_consecutive_samples",
|
||||
"minimum_failed_targets",
|
||||
"minimum_healthy_targets",
|
||||
):
|
||||
value = thresholds.get(key)
|
||||
if not isinstance(value, int) or value < 1:
|
||||
raise ValueError(f"threshold must be a positive integer: {key}")
|
||||
if thresholds["minimum_failed_targets"] > len(targets):
|
||||
raise ValueError("minimum_failed_targets exceeds target count")
|
||||
if thresholds["minimum_healthy_targets"] > len(targets):
|
||||
raise ValueError("minimum_healthy_targets exceeds target count")
|
||||
|
||||
allow_root = Path(str(config.get("callback_allowlist_root") or ""))
|
||||
if not allow_root.is_absolute():
|
||||
raise ValueError("callback_allowlist_root must be absolute")
|
||||
callbacks = config.get("callbacks") or {}
|
||||
if not isinstance(callbacks, dict):
|
||||
raise ValueError("callbacks must be an object")
|
||||
for lifecycle, actions in callbacks.items():
|
||||
if lifecycle not in VALID_LIFECYCLE_STATES:
|
||||
raise ValueError(f"unsupported callback lifecycle: {lifecycle}")
|
||||
if not isinstance(actions, list):
|
||||
raise ValueError("callback lifecycle entries must be arrays")
|
||||
for action in actions:
|
||||
if not isinstance(action, dict) or not action.get("id") or not action.get("path"):
|
||||
raise ValueError("callbacks require id and path")
|
||||
action_path = Path(str(action["path"]))
|
||||
if not action_path.is_absolute() or not action_path.is_relative_to(allow_root):
|
||||
raise ValueError(f"callback path outside allowlist root: {action_path}")
|
||||
|
||||
|
||||
def default_state(observed_at: str) -> dict[str, Any]:
|
||||
return {
|
||||
"schema_version": STATE_SCHEMA_VERSION,
|
||||
"lifecycle_state": "healthy",
|
||||
"incident_id": None,
|
||||
"failure_streak": 0,
|
||||
"success_streak": 0,
|
||||
"first_failure_at": None,
|
||||
"last_transition_at": observed_at,
|
||||
"last_observed_at": observed_at,
|
||||
}
|
||||
|
||||
|
||||
def load_state(path: Path, observed_at: str) -> dict[str, Any]:
|
||||
if not path.exists():
|
||||
return default_state(observed_at)
|
||||
state = read_json(path)
|
||||
if state.get("schema_version") != STATE_SCHEMA_VERSION:
|
||||
raise ValueError("unsupported state schema_version")
|
||||
if state.get("lifecycle_state") not in VALID_LIFECYCLE_STATES:
|
||||
raise ValueError("invalid lifecycle state")
|
||||
return state
|
||||
|
||||
|
||||
def probe_target(target: dict[str, Any]) -> dict[str, Any]:
|
||||
expected = {int(value) for value in target.get("expected_statuses", [200])}
|
||||
timeout_seconds = float(target.get("timeout_seconds", 5))
|
||||
request = urllib.request.Request(
|
||||
str(target["url"]),
|
||||
headers={"User-Agent": "AWOOOI-External-L0-Recovery-Supervisor/1.0"},
|
||||
)
|
||||
status = 0
|
||||
error_class = ""
|
||||
try:
|
||||
with urllib.request.urlopen(request, timeout=timeout_seconds) as response:
|
||||
status = int(response.status)
|
||||
except urllib.error.HTTPError as exc:
|
||||
status = int(exc.code)
|
||||
error_class = "http_error"
|
||||
except urllib.error.URLError:
|
||||
error_class = "url_error"
|
||||
except TimeoutError:
|
||||
error_class = "timeout"
|
||||
return {
|
||||
"id": str(target["id"]),
|
||||
"url": str(target["url"]),
|
||||
"status": status,
|
||||
"healthy": status in expected,
|
||||
"error_class": error_class,
|
||||
"response_body_read": False,
|
||||
}
|
||||
|
||||
|
||||
def collect_results(config: dict[str, Any], samples_json: Path | None) -> list[dict[str, Any]]:
|
||||
if samples_json:
|
||||
payload = read_json(samples_json)
|
||||
results = payload.get("target_results")
|
||||
if not isinstance(results, list):
|
||||
raise ValueError("samples-json requires target_results array")
|
||||
configured_ids = {str(target["id"]) for target in config["targets"]}
|
||||
result_ids = {str(result.get("id") or "") for result in results if isinstance(result, dict)}
|
||||
if configured_ids != result_ids:
|
||||
raise ValueError("samples-json target ids do not match config")
|
||||
return [dict(result) for result in results]
|
||||
return [probe_target(target) for target in config["targets"]]
|
||||
|
||||
|
||||
def advance_state(
|
||||
state: dict[str, Any],
|
||||
results: list[dict[str, Any]],
|
||||
thresholds: dict[str, int],
|
||||
observed_at: str,
|
||||
) -> tuple[dict[str, Any], str | None, dict[str, Any]]:
|
||||
previous = str(state["lifecycle_state"])
|
||||
failed = [str(result["id"]) for result in results if not bool(result.get("healthy"))]
|
||||
healthy = [str(result["id"]) for result in results if bool(result.get("healthy"))]
|
||||
outage_sample = len(failed) >= thresholds["minimum_failed_targets"]
|
||||
recovery_sample = len(healthy) >= thresholds["minimum_healthy_targets"]
|
||||
transition: str | None = None
|
||||
|
||||
if previous in {"healthy", "recovered", "suspected_outage"}:
|
||||
if outage_sample:
|
||||
state["failure_streak"] = int(state.get("failure_streak", 0)) + 1
|
||||
state["success_streak"] = 0
|
||||
state["first_failure_at"] = state.get("first_failure_at") or observed_at
|
||||
desired = "outage_confirmed" if state["failure_streak"] >= thresholds["outage_consecutive_samples"] else "suspected_outage"
|
||||
else:
|
||||
state["failure_streak"] = 0
|
||||
state["success_streak"] = 0
|
||||
state["first_failure_at"] = None
|
||||
desired = "healthy" if previous != "recovered" else "recovered"
|
||||
else:
|
||||
if recovery_sample:
|
||||
state["success_streak"] = int(state.get("success_streak", 0)) + 1
|
||||
state["failure_streak"] = 0
|
||||
desired = "recovered" if state["success_streak"] >= thresholds["recovery_consecutive_samples"] else "recovering"
|
||||
else:
|
||||
state["success_streak"] = 0
|
||||
desired = "outage_confirmed"
|
||||
|
||||
if desired != previous:
|
||||
transition = desired
|
||||
state["lifecycle_state"] = desired
|
||||
state["last_transition_at"] = observed_at
|
||||
if desired == "outage_confirmed" and not state.get("incident_id"):
|
||||
compact = observed_at.replace("-", "").replace(":", "").replace("T", "-").replace("Z", "")
|
||||
state["incident_id"] = f"INC-L0-{compact}"
|
||||
if desired == "healthy":
|
||||
state["incident_id"] = None
|
||||
state["last_observed_at"] = observed_at
|
||||
sample = {
|
||||
"outage_candidate": outage_sample,
|
||||
"recovery_candidate": recovery_sample,
|
||||
"failed_target_ids": failed,
|
||||
"healthy_target_ids": healthy,
|
||||
}
|
||||
return state, transition, sample
|
||||
|
||||
|
||||
def validate_required_callbacks(config: dict[str, Any], lifecycle: str) -> list[str]:
|
||||
configured = {str(item["id"]) for item in (config.get("callbacks") or {}).get(lifecycle, [])}
|
||||
required = {str(item) for item in (config.get("required_callback_ids") or {}).get(lifecycle, [])}
|
||||
return sorted(required - configured)
|
||||
|
||||
|
||||
def run_callbacks(
|
||||
config: dict[str, Any],
|
||||
lifecycle: str,
|
||||
event_file: Path,
|
||||
apply: bool,
|
||||
) -> tuple[list[dict[str, Any]], list[str]]:
|
||||
receipts: list[dict[str, Any]] = []
|
||||
blockers: list[str] = []
|
||||
missing = validate_required_callbacks(config, lifecycle)
|
||||
blockers.extend(f"required_callback_missing:{callback_id}" for callback_id in missing)
|
||||
for action in (config.get("callbacks") or {}).get(lifecycle, []):
|
||||
action_id = str(action["id"])
|
||||
action_path = Path(str(action["path"]))
|
||||
if not action_path.is_file() or not os.access(action_path, os.X_OK):
|
||||
blockers.append(f"callback_not_executable:{action_id}")
|
||||
receipts.append({"id": action_id, "status": "blocked_not_executable", "path": str(action_path)})
|
||||
continue
|
||||
if not apply:
|
||||
receipts.append({"id": action_id, "status": "check_only", "path": str(action_path)})
|
||||
continue
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
[str(action_path), "--event-file", str(event_file)],
|
||||
stdin=subprocess.DEVNULL,
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
timeout=int(action.get("timeout_seconds", 30)),
|
||||
check=False,
|
||||
)
|
||||
status = "executed" if completed.returncode == 0 else "failed"
|
||||
receipts.append({"id": action_id, "status": status, "exit_code": completed.returncode})
|
||||
if completed.returncode != 0:
|
||||
blockers.append(f"callback_failed:{action_id}")
|
||||
except subprocess.TimeoutExpired:
|
||||
receipts.append({"id": action_id, "status": "timeout"})
|
||||
blockers.append(f"callback_timeout:{action_id}")
|
||||
return receipts, blockers
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parse_args()
|
||||
observed = now_utc(args.now)
|
||||
observed_at = isoformat(observed)
|
||||
config = read_json(args.config)
|
||||
validate_config(config)
|
||||
state = load_state(args.state_file, observed_at)
|
||||
previous_state = str(state["lifecycle_state"])
|
||||
results = collect_results(config, args.samples_json)
|
||||
thresholds = {key: int(value) for key, value in config["thresholds"].items()}
|
||||
state, transition, sample = advance_state(state, results, thresholds, observed_at)
|
||||
run_id = f"external-l0-{observed.strftime('%Y%m%dT%H%M%SZ')}"
|
||||
artifact_file = args.artifact_dir / f"{run_id}.json"
|
||||
event: dict[str, Any] = {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"run_id": run_id,
|
||||
"observed_at": observed_at,
|
||||
"mode": "apply" if args.apply else "check",
|
||||
"previous_state": previous_state,
|
||||
"lifecycle_state": state["lifecycle_state"],
|
||||
"transition": transition,
|
||||
"incident_id": state.get("incident_id"),
|
||||
"sample": sample,
|
||||
"target_results": results,
|
||||
"callback_receipts": [],
|
||||
"blockers": [],
|
||||
"secret_value_read": False,
|
||||
"arbitrary_shell_allowed": False,
|
||||
"remote_write_performed": False,
|
||||
}
|
||||
write_json(artifact_file, event)
|
||||
callback_receipts: list[dict[str, Any]] = []
|
||||
blockers: list[str] = []
|
||||
if transition:
|
||||
callback_receipts, blockers = run_callbacks(config, transition, artifact_file, args.apply)
|
||||
event["callback_receipts"] = callback_receipts
|
||||
event["blockers"] = blockers
|
||||
event["runtime_ready"] = not blockers
|
||||
event["remote_write_performed"] = args.apply and any(
|
||||
receipt.get("status") == "executed" for receipt in callback_receipts
|
||||
)
|
||||
write_json(artifact_file, event)
|
||||
write_json(args.state_file, state)
|
||||
print(json.dumps({
|
||||
"run_id": run_id,
|
||||
"state": state["lifecycle_state"],
|
||||
"transition": transition,
|
||||
"incident_id": state.get("incident_id"),
|
||||
"runtime_ready": event["runtime_ready"],
|
||||
"artifact": str(artifact_file),
|
||||
}, ensure_ascii=True))
|
||||
return 2 if args.apply and blockers else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
raise SystemExit(main())
|
||||
except (OSError, ValueError, json.JSONDecodeError) as exc:
|
||||
print(f"external_l0_supervisor_error={exc}", file=sys.stderr)
|
||||
raise SystemExit(2)
|
||||
@@ -4,6 +4,7 @@ set -u
|
||||
DURATION_SECONDS="${DURATION_SECONDS:-900}"
|
||||
INTERVAL_SECONDS="${INTERVAL_SECONDS:-5}"
|
||||
CONNECT_TIMEOUT_SECONDS="${CONNECT_TIMEOUT_SECONDS:-3}"
|
||||
TCP_PROBE_PYTHON="${TCP_PROBE_PYTHON:-python3}"
|
||||
RUN_ID="${RUN_ID:-$(date -u '+%Y%m%dT%H%M%SZ')}"
|
||||
ARTIFACT_DIR="${ARTIFACT_DIR:-/tmp/awoooi-full-host-reboot-drill/${RUN_ID}}"
|
||||
|
||||
@@ -65,6 +66,11 @@ if ! is_positive_integer "$DURATION_SECONDS" || ! is_positive_integer "$INTERVAL
|
||||
exit 64
|
||||
fi
|
||||
|
||||
if ! command -v "$TCP_PROBE_PYTHON" >/dev/null 2>&1; then
|
||||
printf 'error=tcp_probe_python_not_found:%s\n' "$TCP_PROBE_PYTHON" >&2
|
||||
exit 69
|
||||
fi
|
||||
|
||||
mkdir -p "$ARTIFACT_DIR/state"
|
||||
SAMPLES_TSV="$ARTIFACT_DIR/samples.tsv"
|
||||
EVENTS_TSV="$ARTIFACT_DIR/events.tsv"
|
||||
@@ -78,7 +84,16 @@ sanitize_name() {
|
||||
|
||||
probe_host() {
|
||||
local observed_at="$1" alias="$2" address="$3" port="$4" output="$5"
|
||||
if nc -z -w "$CONNECT_TIMEOUT_SECONDS" "$address" "$port" >/dev/null 2>&1; then
|
||||
if "$TCP_PROBE_PYTHON" - "$address" "$port" "$CONNECT_TIMEOUT_SECONDS" >/dev/null 2>&1 <<'PY'; then
|
||||
import socket
|
||||
import sys
|
||||
|
||||
address = sys.argv[1]
|
||||
port = int(sys.argv[2])
|
||||
timeout_seconds = float(sys.argv[3])
|
||||
with socket.create_connection((address, port), timeout=timeout_seconds):
|
||||
pass
|
||||
PY
|
||||
printf '%s\thost\t%s\tup\ttcp_%s_open\n' "$observed_at" "$alias" "$port" >"$output"
|
||||
else
|
||||
printf '%s\thost\t%s\tdown\ttcp_%s_closed_or_unreachable\n' "$observed_at" "$alias" "$port" >"$output"
|
||||
|
||||
@@ -50,10 +50,28 @@ def test_fresh_reboot_claim_requires_fresh_artifact_and_zero_blockers() -> None:
|
||||
assert 'name = "fresh_reboot_window_observed"' in control
|
||||
assert 'name = "scorecard_can_claim_slo"' in control
|
||||
assert 'name = "scorecard_has_no_blockers"' in control
|
||||
assert "$artifactEpoch -ge ($startedEpoch - $CoordinatorConfig.artifactClockSkewSeconds)" in control
|
||||
assert 'name = "scorecard_non_slo_blockers_clear"' in control
|
||||
assert "SCORECARD_BLOCKED_BY_REBOOT_WINDOW_ONLY" in control
|
||||
assert "$artifactEpoch -ge ($evidenceNotBeforeEpoch - $CoordinatorConfig.artifactClockSkewSeconds)" in control
|
||||
assert "-not $requiresFreshWindow -or $rebootSloClaimed" in control
|
||||
|
||||
|
||||
def test_reboot_event_survives_a_killed_first_recovery_and_can_close_late() -> None:
|
||||
control = read("agent99-control-plane.ps1")
|
||||
tasks = read("agent99-register-tasks.ps1")
|
||||
|
||||
assert "pendingRecovery = $pendingRecovery" in control
|
||||
assert 'host_reboot_recovery_pending' in control
|
||||
assert "Update-AgentBootRecoveryState" in control
|
||||
assert '"slo_breached_recovery_pending"' in control
|
||||
assert '"recovered_late"' in control
|
||||
assert "$bootState.pendingRecovery" in control
|
||||
assert "$coordinatorEvidenceNotBefore = [datetime]$bootState.bootTime" in control
|
||||
assert "$recoverySettings" in tasks
|
||||
assert "New-TimeSpan -Minutes 20" in tasks
|
||||
assert '-Settings $recoverySettings' in tasks
|
||||
|
||||
|
||||
def test_runtime_deployer_migrates_recovery_coordinator_defaults() -> None:
|
||||
deployer = read("agent99-deploy.ps1")
|
||||
config = read("agent99.config.99.example.json")
|
||||
|
||||
@@ -0,0 +1,105 @@
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[3]
|
||||
SCRIPT = ROOT / "scripts" / "reboot-recovery" / "external-l0-recovery-supervisor.py"
|
||||
EXAMPLE_CONFIG = ROOT / "ops" / "reboot-recovery" / "external-l0-recovery-supervisor.example.json"
|
||||
|
||||
|
||||
def write_samples(path: Path, healthy: bool) -> None:
|
||||
ids = ("awoooi-api", "awoooi-web", "gitea", "stock-freshness")
|
||||
path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"target_results": [
|
||||
{
|
||||
"id": target_id,
|
||||
"url": f"https://{target_id}.example.invalid/",
|
||||
"status": 200 if healthy else 0,
|
||||
"healthy": healthy,
|
||||
"error_class": "" if healthy else "timeout",
|
||||
"response_body_read": False,
|
||||
}
|
||||
for target_id in ids
|
||||
]
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
|
||||
def run_once(tmp_path: Path, samples: Path, observed_at: str) -> dict[str, object]:
|
||||
completed = subprocess.run(
|
||||
[
|
||||
sys.executable,
|
||||
str(SCRIPT),
|
||||
"--check",
|
||||
"--config",
|
||||
str(EXAMPLE_CONFIG),
|
||||
"--state-file",
|
||||
str(tmp_path / "state.json"),
|
||||
"--artifact-dir",
|
||||
str(tmp_path / "evidence"),
|
||||
"--samples-json",
|
||||
str(samples),
|
||||
"--now",
|
||||
observed_at,
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
return json.loads(completed.stdout)
|
||||
|
||||
|
||||
def test_external_l0_state_machine_is_deduplicated_and_recovers(tmp_path: Path) -> None:
|
||||
failed = tmp_path / "failed.json"
|
||||
healthy = tmp_path / "healthy.json"
|
||||
write_samples(failed, healthy=False)
|
||||
write_samples(healthy, healthy=True)
|
||||
|
||||
first = run_once(tmp_path, failed, "2026-07-11T08:00:00Z")
|
||||
assert first["state"] == "suspected_outage"
|
||||
assert first["transition"] == "suspected_outage"
|
||||
|
||||
second = run_once(tmp_path, failed, "2026-07-11T08:00:15Z")
|
||||
assert second["state"] == "outage_confirmed"
|
||||
assert second["transition"] == "outage_confirmed"
|
||||
assert str(second["incident_id"]).startswith("INC-L0-")
|
||||
assert second["runtime_ready"] is False
|
||||
|
||||
evidence = json.loads(Path(str(second["artifact"])).read_text(encoding="utf-8"))
|
||||
assert sorted(evidence["blockers"]) == [
|
||||
"callback_not_executable:maintenance_activate",
|
||||
"callback_not_executable:power_restore_99",
|
||||
"callback_not_executable:telegram_down",
|
||||
]
|
||||
|
||||
third = run_once(tmp_path, failed, "2026-07-11T08:00:30Z")
|
||||
assert third["state"] == "outage_confirmed"
|
||||
assert third["transition"] is None
|
||||
|
||||
fourth = run_once(tmp_path, healthy, "2026-07-11T08:00:45Z")
|
||||
assert fourth["state"] == "recovering"
|
||||
assert fourth["transition"] == "recovering"
|
||||
|
||||
fifth = run_once(tmp_path, healthy, "2026-07-11T08:01:00Z")
|
||||
assert fifth["state"] == "recovered"
|
||||
assert fifth["transition"] == "recovered"
|
||||
|
||||
|
||||
def test_external_l0_contract_is_off_lan_and_never_reads_response_bodies() -> None:
|
||||
config = json.loads(EXAMPLE_CONFIG.read_text(encoding="utf-8"))
|
||||
text = SCRIPT.read_text(encoding="utf-8")
|
||||
|
||||
assert len(config["targets"]) == 4
|
||||
assert all(target["url"].startswith("https://") for target in config["targets"])
|
||||
assert "192.168.0." not in EXAMPLE_CONFIG.read_text(encoding="utf-8")
|
||||
assert "response_body_read" in text
|
||||
assert "TELEGRAM_BOT_TOKEN" not in text
|
||||
assert "shell=True" not in text
|
||||
assert "callback_allowlist_root" in text
|
||||
assert '"power_restore_99"' in EXAMPLE_CONFIG.read_text(encoding="utf-8")
|
||||
@@ -26,6 +26,10 @@ def test_observer_is_read_only_and_records_transitions() -> None:
|
||||
assert "probe_host" in text
|
||||
assert "probe_route" in text
|
||||
assert "record_transition" in text
|
||||
assert "socket.create_connection" in text
|
||||
assert 'TCP_PROBE_PYTHON="${TCP_PROBE_PYTHON:-python3}"' in text
|
||||
assert "nc -z" not in text
|
||||
assert "telnet://" not in text
|
||||
assert "samples.tsv" in text
|
||||
assert "events.tsv" in text
|
||||
assert "secret_value_read=false" in text
|
||||
|
||||
Reference in New Issue
Block a user