from __future__ import annotations import json from datetime import UTC, datetime, timedelta from pathlib import Path import pytest from scripts.execute_nutricoach_v150_live_upgrade import ( JsonValue, LiveTarget, UpgradeDenied, clean_boundary, inspect_package, load_object, object_at, ) from scripts.prepare_nutricoach_v150_live_upgrade import ( cleanup_clone, prepare_package, ) from scripts.nutricoach_v150_live_upgrade_state import tree_snapshot ROOT = Path(__file__).resolve().parents[1] MANIFEST = ( ROOT / ".omo/evidence/nutricoach-v150-combined/task-1-candidate/manifest.json" ) def _json(path: Path, value: JsonValue) -> None: path.parent.mkdir(parents=True, exist_ok=True) _ = path.write_text( json.dumps(value, separators=(",", ":")) + "\n", encoding="utf-8" ) def _fixture(tmp_path: Path) -> LiveTarget: profiles = tmp_path / "profiles" profile = profiles / "dualcoachtest" expired = (datetime.now(UTC) - timedelta(hours=1)).isoformat() _json( profile / "customers/registry.json", { "version": 1, "registry_mode": "ordinary_v1", "diagnostic_session_digest": None, "owner": {"user_id": "1", "chat_id": "1", "topic_id": "0"}, "customers": [{"customer_key": "legacy", "enabled": True}], }, ) onboarding = profile / "data/onboarding" _json( onboarding / "telegram-customer-bootstrap-v1/ledger.json", {"sessions": [{"state": "ACTIVE", "expires_at": expired}]}, ) outbox = onboarding / "telegram-publication-outbox-v1" _json(outbox / "ledger.json", {"records": [{"state": "COMMITTED"}]}) _json(outbox / "emergency.json", {"records": []}) deliveries = profile / "data/scheduled-deliveries.jsonl" deliveries.parent.mkdir(parents=True, exist_ok=True) _ = deliveries.write_text( "".join( json.dumps({"reservation_id": "r", "state": state}) + "\n" for state in ("prepared", "sending", "delivered", "sent_audited") ), encoding="utf-8", ) _json( profile / "data/customers/legacy/nutrition-onboarding/session.json", {"sessions": {"session": {"state": "COMMITTED"}}}, ) other = profiles / "physique-coach" other.mkdir() _ = (other / "legacy.txt").write_text("stable\n") unit = tmp_path / "systemd/hermes-gateway-dualcoachtest.service" unit.parent.mkdir() _ = unit.write_text("[Service]\n") dropin = Path(f"{unit}.d") dropin.mkdir() credential = tmp_path / "authority/candidate" credential.parent.mkdir() _ = credential.write_text("credential bytes must not leak\n") _ = (dropin / "authority.conf").write_text( f"[Service]\nLoadCredential=candidate:{credential}\n" ) return LiveTarget( profile, profiles, "hermes-gateway-dualcoachtest.service", unit, dropin ) def _service(_name: str) -> dict[str, str]: return { "ActiveState": "active", "SubState": "running", "MainPID": "571685", "ExecMainStartTimestampMonotonic": "2234547821498", } def test_prepare_is_read_only_and_controller_awaits_once(tmp_path: Path) -> None: target = _fixture(tmp_path) output = tmp_path / "preflight" registry = (target.profile_root / "customers/registry.json").read_bytes() receipt = prepare_package(target, MANIFEST, output, _service, datetime.now(UTC)) assert receipt["status"] == "AWAITING_AUTHORIZATION" assert receipt["candidate_digest"] == ( "066a794d44861d2cd0fe8c1ea14c0050e00219b38a2d2026b389772783056dad" ) assert (target.profile_root / "customers/registry.json").read_bytes() == registry assert not tuple(output.parent.glob(".nutricoach-v150-rehearsal-*")) inspected = inspect_package(output / "package.json", None, None, _service) assert inspected["status"] == "AWAITING_AUTHORIZATION" with pytest.raises(UpgradeDenied, match="wrong_approval"): _ = inspect_package(output / "package.json", None, "WRONG", _service) _ = (output / "authorization-consumed.json").write_text("{}\n") with pytest.raises(UpgradeDenied, match="authorization_already_used"): _ = inspect_package(output / "package.json", None, None, _service) @pytest.mark.parametrize( ("case", "reason"), [ ("bootstrap", "pending_onboarding"), ("publication", "pending_provider_outcome"), ("delivery", "pending_send"), ("onboarding", "pending_onboarding"), ("emergency", "unknown_provider_outcome"), ("migration", "pending_migration"), ], ) def test_pending_workflows_are_denied(tmp_path: Path, case: str, reason: str) -> None: profile = _fixture(tmp_path).profile_root outbox = profile / "data/onboarding/telegram-publication-outbox-v1" if case == "bootstrap": future = (datetime.now(UTC) + timedelta(hours=1)).isoformat() _json( profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json", {"sessions": [{"state": "ACTIVE", "expires_at": future}]}, ) elif case == "publication": _json(outbox / "ledger.json", {"records": [{"state": "RECEIPTED"}]}) elif case == "delivery": _ = (profile / "data/scheduled-deliveries.jsonl").write_text( '{"reservation_id":"r","state":"sending"}\n' ) elif case == "onboarding": _json( profile / "data/customers/legacy/nutrition-onboarding/session.json", {"sessions": [{"state": "IN_PROGRESS"}]}, ) elif case == "emergency": _json(outbox / "emergency.json", {"records": [{"state": "RECEIPTED"}]}) else: _json(profile / "data/migrations/pending.json", {"state": "pending"}) with pytest.raises(UpgradeDenied, match=reason): _ = clean_boundary(profile, datetime.now(UTC)) def test_tamper_old_seal_and_stale_protected_bytes_are_denied(tmp_path: Path) -> None: target = _fixture(tmp_path) output = tmp_path / "preflight" _ = prepare_package(target, MANIFEST, output, _service, datetime.now(UTC)) package = output / "package.json" original = package.read_bytes() _ = package.write_bytes(original.replace(b"AWAITING_AUTHORIZATION", b"TAMPERED")) with pytest.raises(UpgradeDenied, match="package_tamper"): _ = inspect_package(package, None, None, _service) _ = package.write_bytes(original) old = ROOT / ".omo/evidence/nutricoach-v150-combined/historical" old_manifest = next(old.glob("superseded-*/candidate/manifest.json")) with pytest.raises(UpgradeDenied, match="old_or_wrong_candidate"): _ = inspect_package(package, old_manifest, None, _service) _ = (target.profile_root / "SOUL.md").write_text("drift\n") with pytest.raises(UpgradeDenied, match="stale_protected_bytes"): _ = inspect_package(package, None, None, _service) def test_disposable_read_only_clone_is_removed(tmp_path: Path) -> None: clone = tmp_path / "clone" readonly = clone / "runtime" readonly.mkdir(parents=True) _ = (readonly / "artifact").write_text("sealed\n") readonly.chmod(0o555) cleanup_clone(clone) assert not clone.exists() def test_active_gateway_logs_may_append_during_read_only_preflight( tmp_path: Path, ) -> None: target = _fixture(tmp_path) logs = target.profile_root / "logs" logs.mkdir() agent_log = logs / "agent.log" errors_log = logs / "errors.log" gateway_log = logs / "gateway.log" _ = agent_log.write_text("before\n") _ = errors_log.write_text("before\n") _ = gateway_log.write_text("before\n") calls = 0 def service_with_log_append(_name: str) -> dict[str, str]: nonlocal calls calls += 1 if calls == 1: with agent_log.open("a") as stream: _ = stream.write("active append\n") with errors_log.open("a") as stream: _ = stream.write("active append\n") with gateway_log.open("a") as stream: _ = stream.write("active append\n") return _service(_name) output = tmp_path / "preflight-v2" _ = prepare_package( target, MANIFEST, output, service_with_log_append, datetime.now(UTC), ) package = load_object(output / "package.json") payload = object_at(package.get("payload"), "payload") snapshots = object_at(payload.get("snapshots"), "snapshots") raw_changed = snapshots.get("volatile_changed_paths") if not isinstance(raw_changed, list) or not all( isinstance(value, str) for value in raw_changed ): raise AssertionError changed = [value for value in raw_changed if isinstance(value, str)] assert changed == [ "dualcoachtest/logs/agent.log", "dualcoachtest/logs/errors.log", "dualcoachtest/logs/gateway.log", ] def test_service_lifecycle_files_are_volatile(tmp_path: Path) -> None: paths = ( "auth.json", "data/onboarding/telegram-staff-membership-v1/events.jsonl", "gateway.pid", "gateway_state.json", "logs/gateway-exit-diag.log", "logs/gateway-shutdown-diag.log", "logs/new-runtime.log", "state.db", "state.db-shm", "state.db-wal", ) for relative in paths: path = tmp_path / relative path.parent.mkdir(parents=True, exist_ok=True) _ = path.write_text("before\n") snapshot = tree_snapshot(tmp_path, classify_volatile=True) volatile = snapshot["volatile"] if not isinstance(volatile, list): raise AssertionError assert [entry["path"] for entry in volatile if isinstance(entry, dict)] == list( paths ) def test_permission_package_rejects_altered_controller_derivation( tmp_path: Path, ) -> None: target = _fixture(tmp_path) output = tmp_path / "preflight-v3" _ = prepare_package(target, MANIFEST, output, _service, datetime.now(UTC)) with pytest.raises(UpgradeDenied, match="controller_derivation"): _ = inspect_package( output / "package.json", service_reader=_service, expected_controller_derivation="altered-controller", )