Стабилизация переключения Bridge и Quick Connect
Безопасно восстанавливает управляющее подключение и локальные checkpoint без повторных команд сканеру. Добавляет PCAP/Bridge guardrails и регрессионные проверки одношагового переподключения. Известный дефект: после второго подключения интерфейс не присоединяется к новой генерации preview правой камеры. В живой Quick Connect-сессии STOP был принят, но READY не подтвердился до таймаута; автоматический повтор STOP запрещён.
This commit is contained in:
DCCONSTRUCTIONS
Executable
+233
@@ -0,0 +1,233 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fail closed if Quick Connect work drifts from K1 protocol or Bridge baselines.
|
||||
|
||||
This checker performs local filesystem, Git, and synthetic-test validation only.
|
||||
It never opens BLE, CoreWLAN, MQTT, RTSP, or a socket to the scanner.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
||||
BASELINE_COMMIT = "1001a316385a742114523c9996c43034898c115d"
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class CaptureOracle:
|
||||
relative_path: str
|
||||
size_bytes: int
|
||||
sha256: str
|
||||
|
||||
|
||||
CAPTURE_ORACLES = (
|
||||
CaptureOracle(
|
||||
"sessions/iphone-k1-observation/"
|
||||
"20260716T151239Z_lixelgo-clean-cycle_d547/"
|
||||
"captures/iphone-network.pcapng",
|
||||
126_962_848,
|
||||
"e6e98e22156ba2ce041c5f6ff32a3116446aaae8b617c90c16b3f00de8d5100f",
|
||||
),
|
||||
CaptureOracle(
|
||||
"sessions/iphone-k1-observation/"
|
||||
"20260716T151239Z_lixelgo-clean-cycle_d547/"
|
||||
"captures/iphone-network.pcap",
|
||||
116_785_317,
|
||||
"45ebaebbba8c8f62ebc405eb0f7cdc496d84dcdf80c01decba61c56b4514a26f",
|
||||
),
|
||||
CaptureOracle(
|
||||
"sessions/iphone-k1-observation/"
|
||||
"20260716T152407Z_lixelgo-controls-stop_9be0/"
|
||||
"captures/iphone-network.pcapng",
|
||||
254_161_132,
|
||||
"59133299bb1c4dc2157e50401ddef4c131acd2196d7d2e2ccddd5c51b95e57cb",
|
||||
),
|
||||
CaptureOracle(
|
||||
"sessions/iphone-k1-observation/"
|
||||
"20260716T152407Z_lixelgo-controls-stop_9be0/"
|
||||
"captures/iphone-network.pcap",
|
||||
234_691_027,
|
||||
"6e267f0ab9b213e5dde732b94cf3f4b2f32fcf66f5569bcb511a13c952883606",
|
||||
),
|
||||
)
|
||||
|
||||
FROZEN_PATHS = (
|
||||
"src/k1link/device_plugins/xgrids_k1/protocol",
|
||||
"src/k1link/device_plugins/xgrids_k1/ble/ap_activation.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/ble/wifi_provisioning.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/connection_supervisor.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/physical_command_coordinator.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/physical_command_ledger.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/mqtt",
|
||||
"src/k1link/device_plugins/xgrids_k1/camera.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/viewer/runtime.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/archive.py",
|
||||
)
|
||||
|
||||
PYTEST_TARGETS = (
|
||||
"tests/test_xgrids_application_bootstrap.py",
|
||||
(
|
||||
"tests/test_xgrids_application_mqtt.py::"
|
||||
"test_unanswered_optional_status_does_not_steal_same_identity_required_refresh"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_application_mqtt.py::"
|
||||
"test_already_arrived_optional_status_is_consumed_before_required_refresh"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_application_mqtt.py::"
|
||||
"test_unbound_optional_status_remains_distinct_from_bound_required_refresh"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_application_session.py::"
|
||||
"test_canonical_stages_require_operator_events_but_device_standby_does_not"
|
||||
),
|
||||
(
|
||||
"tests/test_connection_supervisor.py::"
|
||||
"test_quick_to_bridge_revokes_authority_with_same_target_and_host_epoch"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_applied_bridge_target_requires_explicit_wifi_client_mode"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_post_dispatch_bridge_accepts_exact_fw302_network_name_when_baseline_is_same"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_post_dispatch_bridge_rejects_another_fw302_network_name"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_bridge_read_only_verify_never_shortcuts_fresh_ble_status"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_durable_restart_same_bridge_baseline_cannot_resolve_ambiguous_write"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_quick_to_bridge_requires_new_explicit_scan_instead_of_retained_session"
|
||||
),
|
||||
(
|
||||
"tests/test_xgrids_acquisition_lifecycle.py::"
|
||||
"test_quick_to_bridge_does_not_retrieve_old_handle_without_new_scan"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def fail(message: str) -> None:
|
||||
raise RuntimeError(message)
|
||||
|
||||
|
||||
def sha256_file(path: Path) -> str:
|
||||
digest = hashlib.sha256()
|
||||
with path.open("rb") as source:
|
||||
while chunk := source.read(4 * 1024 * 1024):
|
||||
digest.update(chunk)
|
||||
return digest.hexdigest()
|
||||
|
||||
|
||||
def verify_capture_oracles() -> None:
|
||||
for oracle in CAPTURE_ORACLES:
|
||||
path = REPOSITORY_ROOT / oracle.relative_path
|
||||
try:
|
||||
metadata = path.lstat()
|
||||
except FileNotFoundError:
|
||||
fail(f"capture oracle is missing: {oracle.relative_path}")
|
||||
if not stat.S_ISREG(metadata.st_mode) or path.is_symlink():
|
||||
fail(f"capture oracle is not a regular non-symlink file: {oracle.relative_path}")
|
||||
if stat.S_IMODE(metadata.st_mode) != 0o600:
|
||||
fail(f"capture oracle mode is not 0600: {oracle.relative_path}")
|
||||
if metadata.st_size != oracle.size_bytes:
|
||||
fail(
|
||||
"capture oracle size changed: "
|
||||
f"{oracle.relative_path} ({metadata.st_size} != {oracle.size_bytes})"
|
||||
)
|
||||
observed_sha256 = sha256_file(path)
|
||||
if observed_sha256 != oracle.sha256:
|
||||
fail(
|
||||
"capture oracle SHA-256 changed: "
|
||||
f"{oracle.relative_path} ({observed_sha256} != {oracle.sha256})"
|
||||
)
|
||||
print(f"[k1-guardrail] PASS capture oracles: {len(CAPTURE_ORACLES)}")
|
||||
|
||||
|
||||
def git_output(*args: str) -> str:
|
||||
result = subprocess.run(
|
||||
("git", *args),
|
||||
cwd=REPOSITORY_ROOT,
|
||||
check=True,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
)
|
||||
return result.stdout.strip()
|
||||
|
||||
|
||||
def verify_frozen_contour() -> None:
|
||||
git_output("cat-file", "-e", f"{BASELINE_COMMIT}^{{commit}}")
|
||||
changed = git_output("diff", "--name-only", BASELINE_COMMIT, "--", *FROZEN_PATHS)
|
||||
untracked = git_output(
|
||||
"ls-files",
|
||||
"--others",
|
||||
"--exclude-standard",
|
||||
"--",
|
||||
*FROZEN_PATHS,
|
||||
)
|
||||
drift = tuple(line for line in (changed, untracked) if line)
|
||||
if drift:
|
||||
fail("frozen K1 protocol/Bridge contour changed:\n" + "\n".join(drift))
|
||||
print(
|
||||
"[k1-guardrail] PASS frozen protocol/Bridge contour: "
|
||||
f"baseline {BASELINE_COMMIT[:7]}"
|
||||
)
|
||||
|
||||
|
||||
def run_checked(command: tuple[str, ...]) -> None:
|
||||
print("[k1-guardrail] RUN " + " ".join(command), flush=True)
|
||||
subprocess.run(command, cwd=REPOSITORY_ROOT, check=True)
|
||||
|
||||
|
||||
def verify_synthetic_contracts() -> None:
|
||||
python = REPOSITORY_ROOT / ".venv" / "bin" / "python"
|
||||
ruff = REPOSITORY_ROOT / ".venv" / "bin" / "ruff"
|
||||
if not python.is_file():
|
||||
fail("repository Python runtime is missing: .venv/bin/python")
|
||||
if not ruff.is_file():
|
||||
fail("repository Ruff runtime is missing: .venv/bin/ruff")
|
||||
run_checked((str(python), "-m", "pytest", "-q", *PYTEST_TARGETS))
|
||||
run_checked(
|
||||
(
|
||||
str(ruff),
|
||||
"check",
|
||||
"scripts/check_k1_quick_connect_guardrails.py",
|
||||
"src/k1link/device_plugins/xgrids_k1/facade.py",
|
||||
"tests/test_xgrids_acquisition_lifecycle.py",
|
||||
)
|
||||
)
|
||||
run_checked(("git", "diff", "--check"))
|
||||
print("[k1-guardrail] PASS synthetic protocol/Bridge sentinels")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
os.chdir(REPOSITORY_ROOT)
|
||||
try:
|
||||
verify_capture_oracles()
|
||||
verify_frozen_contour()
|
||||
verify_synthetic_contracts()
|
||||
except (OSError, RuntimeError, subprocess.CalledProcessError) as exc:
|
||||
print(f"[k1-guardrail] FAIL {exc}", file=sys.stderr)
|
||||
return 1
|
||||
print("[k1-guardrail] PASS all checks; no scanner I/O was performed")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user