from __future__ import annotations import json import multiprocessing import os import sys import uuid from contextlib import contextmanager from pathlib import Path import pytest from opensquilla.recovery import ( ProfileOperationLock, RecoveryError, StaleRecoveryTransactionError, inspect_profile, ) from opensquilla.recovery.cleanup import ( _path_identity_payload, abandon_cleanup_transaction, cleanup_apply, cleanup_inspect, ) def _home(path: Path, value: str) -> Path: workspace = path / "workspace" workspace.mkdir(parents=True) (workspace / "SOUL.md").write_text(value + "\n", encoding="utf-8") (path / "config.toml").write_text('workspace_dir = "workspace"\n', encoding="utf-8") return path def _recovery_profile(user_data: Path, recovery_id: str, value: str) -> Path: root = user_data / "recovery-profiles" / recovery_id _home(root / "opensquilla", value) (root / "desktop-credential.json").write_text("{}\n", encoding="utf-8") (root / "logs").mkdir() (root / "logs" / "desktop.log").write_text("synthetic\n", encoding="utf-8") return root def _backup(user_data: Path, primary: Path) -> Path: transaction_id = str(uuid.uuid4()) backup = primary.with_name(f"{primary.name}.backup.{transaction_id}") _home(backup, "backup") history = { "schema_version": 1, "backups": [ { "transaction_id": transaction_id, "committed_at": "2026-07-11T00:00:00+00:00", "source": str(user_data / "synthetic-source"), "target": str(primary), "backup": str(backup), "source_identity": _path_identity_payload(backup), "target_identity": _path_identity_payload(primary), "backup_identity": _path_identity_payload(backup), } ], } (user_data / "profile-replacement-history.json").write_text( json.dumps(history, indent=2) + "\n", encoding="utf-8", ) return backup def _desktop_primary(user_data: Path) -> Path: primary = _home(user_data / "opensquilla", "primary") (user_data / "desktop-credential.json").write_text("{}\n", encoding="utf-8") (user_data / "logs").mkdir() (user_data / "logs" / "desktop.log").write_text("synthetic\n", encoding="utf-8") (user_data / "desktop-profile-context.json").write_text( '{"schema_version":1,"active_profile_kind":"primary"}\n', encoding="utf-8", ) return primary def _hold_profile_lock(home: str, state_root: str, ready, release) -> None: os.environ["OPENSQUILLA_USER_STATE_DIR"] = state_root with ProfileOperationLock(home): ready.set() release.wait(10) def _hold_gateway(home: str, ready, release) -> None: from opensquilla.gateway.pidlock import GatewayPidLock state = Path(home) / "state" lock = GatewayPidLock(state) lock.acquire() try: ready.set() release.wait(10) finally: lock.release() def _hold_recreated_legacy_gateway(home: str, ready, release) -> None: from opensquilla.gateway.pidlock import GatewayPidLock home_path = Path(home) state = home_path / "state" state.mkdir(parents=True) lock = GatewayPidLock(state) lock.acquire() try: workspace = home_path / "workspace" workspace.mkdir() (workspace / "SOUL.md").write_text("recreated by old gateway\n", encoding="utf-8") ready.set() release.wait(10) finally: lock.release() def test_delete_current_primary_preserves_recovery_profiles_and_backups( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) recovery_id = str(uuid.uuid4()) recovery = _recovery_profile(user_data, recovery_id, "recovery") backup = _backup(user_data, primary) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) assert inspected.outcome == "ready" assert all("backup" not in item.kind for item in inspected.items) assert not (tmp_path / "lock-state").exists(), "cleanup inspection must remain read-only" result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete", result.stable_code assert not primary.exists() assert not (user_data / "desktop-credential.json").exists() assert not (user_data / "logs").exists() assert not (user_data / "desktop-profile-context.json").exists() assert recovery.is_dir() assert backup.is_dir() assert (user_data / "profile-replacement-history.json").is_file() def test_delete_current_profile_with_state_but_no_legacy_lock_completes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) state = primary / "state" state.mkdir() (state / "sessions.db").write_bytes(b"synthetic sessions") assert not (state / "gateway.pid.lock").exists() inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete", result.stable_code assert not primary.exists() def test_cleanup_revision_ignores_directory_metadata_but_not_child_changes( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) state = primary / "state" state.mkdir() (state / "gateway.pid.lock").write_bytes(b"synthetic lock authority\n") before = cleanup_module._build_plan( user_data, mode="delete-current-profile", profile_kind="primary", recovery_id=None, ) for directory in (primary, primary / "workspace", state): value = directory.stat() os.utime( directory, ns=(value.st_atime_ns, value.st_mtime_ns + 1_000_000_000), ) metadata_only = cleanup_module._build_plan( user_data, mode="delete-current-profile", profile_kind="primary", recovery_id=None, ) assert metadata_only.revision == before.revision (state / "sessions.db").write_bytes(b"new user data\n") changed = cleanup_module._build_plan( user_data, mode="delete-current-profile", profile_kind="primary", recovery_id=None, ) assert changed.revision != before.revision def test_reset_current_settings_preserves_config_workspace_and_sessions( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) config = primary / "config.toml" config_before = config.read_bytes() workspace = primary / "workspace" / "SOUL.md" workspace_before = workspace.read_bytes() sessions = primary / "state" / "sessions.db" sessions.parent.mkdir() sessions.write_bytes(b"synthetic session database") (user_data / "migration-provider-setup.json").write_text("{}\n", encoding="utf-8") (user_data / "migration-last-result.json").write_text("{}\n", encoding="utf-8") inspected = cleanup_inspect( user_data, mode="reset-current-settings", profile_kind="primary", ) assert inspected.outcome == "ready" assert {item.kind for item in inspected.items} == { "primary-credential", "migration-pending", "migration-result", } result = cleanup_apply( user_data, mode="reset-current-settings", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete", result.stable_code assert not (user_data / "desktop-credential.json").exists() assert not (user_data / "migration-provider-setup.json").exists() assert not (user_data / "migration-last-result.json").exists() assert config.read_bytes() == config_before assert workspace.read_bytes() == workspace_before assert sessions.read_bytes() == b"synthetic session database" assert (user_data / "desktop-profile-context.json").is_file() def test_reset_recovery_settings_only_clears_selected_recovery_credential( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) recovery_id = str(uuid.uuid4()) recovery_root = _recovery_profile(user_data, recovery_id, "recovery") recovery_home = recovery_root / "opensquilla" sessions = recovery_home / "state" / "sessions.db" sessions.parent.mkdir() sessions.write_bytes(b"synthetic recovery sessions") config_before = (recovery_home / "config.toml").read_bytes() workspace_before = (recovery_home / "workspace" / "SOUL.md").read_bytes() inspected = cleanup_inspect( user_data, mode="reset-current-settings", profile_kind="recovery", recovery_id=recovery_id, ) result = cleanup_apply( user_data, mode="reset-current-settings", profile_kind="recovery", recovery_id=recovery_id, transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete" assert not (recovery_root / "desktop-credential.json").exists() assert (recovery_root / "logs" / "desktop.log").is_file() assert (recovery_home / "config.toml").read_bytes() == config_before assert (recovery_home / "workspace" / "SOUL.md").read_bytes() == workspace_before assert sessions.read_bytes() == b"synthetic recovery sessions" assert primary.is_dir() def test_delete_current_recovery_preserves_primary_other_recovery_and_backups( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) selected_id = str(uuid.uuid4()) other_id = str(uuid.uuid4()) selected = _recovery_profile(user_data, selected_id, "selected") other = _recovery_profile(user_data, other_id, "other") backup = _backup(user_data, primary) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="recovery", recovery_id=selected_id, ) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="recovery", recovery_id=selected_id, transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete", result.stable_code assert not selected.exists() assert primary.is_dir() assert other.is_dir() assert backup.is_dir() assert (user_data / "profile-replacement-history.json").is_file() assert not (user_data / "desktop-profile-context.json").exists() def test_delete_all_inventory_and_apply_removes_every_owned_category( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) recovery_id = str(uuid.uuid4()) _recovery_profile(user_data, recovery_id, "recovery") backup = _backup(user_data, primary) (user_data / "migration-provider-setup.json").write_text("{}\n", encoding="utf-8") (user_data / "migration-last-result.json").write_text("{}\n", encoding="utf-8") (user_data / "desktop-locale.json").write_text('{"locale":"zh-CN"}\n', encoding="utf-8") chromium_storage = user_data / "Local Storage" / "leveldb" chromium_storage.mkdir(parents=True) (chromium_storage / "000001.log").write_text("synthetic cache\n", encoding="utf-8") inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) assert inspected.outcome == "ready" kinds = {item.kind for item in inspected.items} assert { "primary-home", "primary-credential", "primary-logs", "profile-context", "profile-backup", "replacement-history", "migration-pending", "migration-result", "recovery-profiles-container", "user-data-entry", } <= kinds residual_paths = {item.path for item in inspected.items if item.kind == "user-data-entry"} assert user_data / "desktop-locale.json" in residual_paths assert user_data / "Local Storage" in residual_paths result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete" assert all(not item.exists for item in result.items) assert not backup.exists() assert list(user_data.iterdir()) == [] def test_delete_all_from_recovery_profile_still_covers_primary_and_every_recovery( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) selected_id = str(uuid.uuid4()) other_id = str(uuid.uuid4()) _recovery_profile(user_data, selected_id, "selected") _recovery_profile(user_data, other_id, "other") backup = _backup(user_data, primary) inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="recovery", recovery_id=selected_id, ) result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="recovery", recovery_id=selected_id, transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete" assert all(not item.exists for item in result.items) assert not primary.exists() assert not backup.exists() assert list(user_data.iterdir()) == [] def test_cleanup_apply_requires_exact_user_data_confirmation( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=tmp_path / "different-user-data", ) assert result.outcome == "blocked" assert result.stable_code == "cleanup_confirmation_mismatch" assert primary.is_dir() def test_delete_all_retains_overlapping_coordination_locks_without_split_brain( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: # Models macOS: user-state/OpenSquilla is also Electron userData A. user_data = tmp_path / "OpenSquilla" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path)) _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) assert inspected.outcome == "ready" assert all(item.path != user_data / "profile-locks" for item in inspected.items) result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete" assert [path.name for path in user_data.iterdir()] == ["profile-locks"] assert (user_data / "profile-locks").is_dir() def test_cleanup_apply_blocks_when_profile_lock_is_busy( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() state_root = tmp_path / "lock-state" monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(state_root)) primary = _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) context = multiprocessing.get_context("spawn" if sys.platform == "win32" else "fork") ready = context.Event() release = context.Event() process = context.Process( target=_hold_profile_lock, args=(str(primary), str(state_root), ready, release), ) process.start() assert ready.wait(5) try: result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) finally: release.set() process.join(10) assert result.outcome == "blocked" assert result.stable_code == "profile_lock_busy" assert primary.is_dir() def test_cleanup_apply_contends_with_replacement_history_authority( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() state_root = tmp_path / "lock-state" monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(state_root)) _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) context = multiprocessing.get_context("spawn" if sys.platform == "win32" else "fork") ready = context.Event() release = context.Event() process = context.Process( target=_hold_profile_lock, args=(str(user_data), str(state_root), ready, release), ) process.start() assert ready.wait(5) try: result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) finally: release.set() process.join(10) assert result.outcome == "blocked" assert result.stable_code == "profile_lock_busy" assert (user_data / "opensquilla").is_dir() def test_cleanup_apply_refuses_running_legacy_gateway( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) (primary / "state").mkdir() inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) context = multiprocessing.get_context("spawn" if sys.platform == "win32" else "fork") ready = context.Event() release = context.Event() process = context.Process( target=_hold_gateway, args=(str(primary), ready, release), ) process.start() assert ready.wait(5) try: result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) finally: release.set() process.join(10) assert result.outcome == "blocked" assert result.stable_code == "legacy_gateway_running" assert primary.is_dir() def test_cleanup_rejects_linked_user_data_root_without_following_it(tmp_path: Path) -> None: real_user_data = tmp_path / "real-user-data" real_user_data.mkdir() sentinel = real_user_data / "keep.txt" sentinel.write_text("keep\n", encoding="utf-8") linked_user_data = tmp_path / "linked-user-data" try: linked_user_data.symlink_to(real_user_data, target_is_directory=True) except OSError: pytest.skip("directory symlink creation is unavailable") report = cleanup_inspect( linked_user_data, mode="delete-all-user-data", profile_kind="primary", ) assert report.outcome == "blocked" assert report.stable_code == "cleanup_user_data_unsafe" assert sentinel.read_text(encoding="utf-8") == "keep\n" def test_cleanup_apply_rejects_stale_recursive_inventory( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) (primary / "workspace" / "SOUL.md").write_text("changed after inspect\n", encoding="utf-8") result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "blocked" assert result.stable_code == "cleanup_inventory_stale" assert primary.is_dir() def test_cleanup_never_deletes_canonical_profile_recreated_after_quarantine( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) original_remove = cleanup_module._remove_no_follow context = multiprocessing.get_context("spawn" if sys.platform == "win32" else "fork") ready = context.Event() release = context.Event() process: multiprocessing.Process | None = None def recreate_canonical_before_deleting_quarantine(path: Path) -> None: nonlocal process if process is None and path.name.startswith(f".{primary.name}.cleanup."): process = context.Process( target=_hold_recreated_legacy_gateway, args=(str(primary), ready, release), ) process.start() assert ready.wait(5) original_remove(path) monkeypatch.setattr( cleanup_module, "_remove_no_follow", recreate_canonical_before_deleting_quarantine, ) try: result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) finally: release.set() if process is not None: process.join(10) assert process is not None assert process.exitcode == 0 assert result.outcome == "partial" assert (primary / "workspace" / "SOUL.md").read_text(encoding="utf-8") == ( "recreated by old gateway\n" ) assert (user_data / "desktop-credential.json").is_file() assert (user_data / "desktop-profile-context.json").is_file() assert (user_data / ".opensquilla.profile-cleanup.json").is_file() def test_cleanup_releases_legacy_handles_before_deleting_profile_tombstone( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module import opensquilla.recovery.locking as locking_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) state = primary / "state" state.mkdir() (state / "gateway.pid.lock").write_bytes(b"synthetic lock authority\n") inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) monkeypatch.setattr( locking_module, "_windows_requires_legacy_lock_handoff", lambda: True, ) real_acquire = cleanup_module.acquire_legacy_gateway_locks captured_locks = [] @contextmanager def capture_locks(*homes, **kwargs): with real_acquire(*homes, **kwargs) as locks: captured_locks.extend(locks) yield locks real_remove = cleanup_module._remove_no_follow observed_tombstone_delete = False def assert_handles_released(path: Path) -> None: nonlocal observed_tombstone_delete if path.name.startswith(f".{primary.name}.cleanup."): observed_tombstone_delete = True moved_state = path / "state" assert not any(lock.holds_state_root(moved_state) for lock in captured_locks) real_remove(path) monkeypatch.setattr(cleanup_module, "acquire_legacy_gateway_locks", capture_locks) monkeypatch.setattr(cleanup_module, "_remove_no_follow", assert_handles_released) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete", result.stable_code assert observed_tombstone_delete is True assert captured_locks assert not primary.exists() def test_cleanup_rolls_back_directory_swapped_at_no_replace_boundary( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module import opensquilla.recovery.locking as locking_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) displaced = user_data / "displaced-original-profile" real_move = cleanup_module.native_move_no_replace swapped = False monkeypatch.setattr( locking_module, "_windows_requires_legacy_lock_handoff", lambda: True, ) def swap_then_move( source: Path, destination: Path, **move_options: object, ) -> None: nonlocal swapped if source == primary and not swapped: swapped = True source.rename(displaced) _home(source, "replacement profile") real_move(source, destination, **move_options) # type: ignore[arg-type] monkeypatch.setattr(cleanup_module, "native_move_no_replace", swap_then_move) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert swapped is True assert result.outcome == "partial" assert (primary / "workspace" / "SOUL.md").read_text(encoding="utf-8") == ( "replacement profile\n" ) assert (displaced / "workspace" / "SOUL.md").read_text(encoding="utf-8") == ( "primary\n" ) assert (user_data / ".opensquilla.profile-cleanup.json").is_file() def test_cleanup_backup_delete_failure_preserves_history_and_cleanup_journal( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) primary = _desktop_primary(user_data) backup = _backup(user_data, primary) history = user_data / "profile-replacement-history.json" history_before = history.read_bytes() inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) original_remove = cleanup_module._remove_no_follow def fail_backup_delete(path: Path) -> None: if path == backup or path.name.startswith(f".{backup.name}.cleanup."): raise OSError("synthetic failure after backup quarantine") original_remove(path) monkeypatch.setattr(cleanup_module, "_remove_no_follow", fail_backup_delete) result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "partial" assert backup.is_dir() assert history.read_bytes() == history_before assert (user_data / ".opensquilla.profile-cleanup.json").is_file() def test_recovery_profile_inspection_sees_global_partial_cleanup_journal( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) recovery_id = str(uuid.uuid4()) recovery_root = _recovery_profile(user_data, recovery_id, "recovery") recovery_home = recovery_root / "opensquilla" inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="recovery", recovery_id=recovery_id, ) original_remove = cleanup_module._remove_no_follow def fail_recovery_delete(path: Path) -> None: if path == recovery_root or path.name.startswith(f".{recovery_root.name}.cleanup."): raise OSError("synthetic partial recovery profile deletion") original_remove(path) monkeypatch.setattr(cleanup_module, "_remove_no_follow", fail_recovery_delete) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="recovery", recovery_id=recovery_id, transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) report = inspect_profile(recovery_home, profile_kind="desktop-recovery") assert result.outcome == "partial" assert recovery_home.is_dir() assert report.outcome == "recovery_required" assert report.stable_code == "cleanup_transaction_incomplete" assert "recover-transaction" not in report.allowed_actions def test_incomplete_cleanup_journal_blocks_fresh_primary_bootstrap(tmp_path: Path) -> None: user_data = tmp_path / "user-data" user_data.mkdir() home = user_data / "opensquilla" (user_data / ".opensquilla.profile-cleanup.json").write_text( json.dumps( { "schema_version": 1, "operation": "profile-cleanup", "transaction_id": str(uuid.uuid4()), "phase": "prepared", "primary_home": str(home), "tombstones": [], } ) + "\n", encoding="utf-8", ) report = inspect_profile(home, profile_kind="desktop-primary") assert report.outcome == "recovery_required" assert report.stable_code == "cleanup_transaction_incomplete" assert "recover-transaction" not in report.allowed_actions def test_primary_cleanup_journal_does_not_block_unaffected_recovery_profile( tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() primary = _desktop_primary(user_data) recovery_id = str(uuid.uuid4()) recovery_root = _recovery_profile(user_data, recovery_id, "recovery") recovery_home = recovery_root / "opensquilla" (recovery_home / "state").mkdir() (user_data / ".opensquilla.profile-cleanup.json").write_text( json.dumps( { "schema_version": 1, "operation": "profile-cleanup", "transaction_id": str(uuid.uuid4()), "phase": "prepared", "primary_home": str(primary), "mode": "delete-current-profile", "profile_kind": "primary", "recovery_id": None, "tombstones": [], } ) + "\n", encoding="utf-8", ) report = inspect_profile(recovery_home, profile_kind="desktop-recovery") assert report.outcome == "recovery_profile" assert report.stable_code != "cleanup_transaction_incomplete" def test_cleanup_journal_does_not_apply_to_an_ordinary_cli_profile(tmp_path: Path) -> None: home = _home(tmp_path / "cli-home", "cli") (home / "state").mkdir() (home.parent / f".{home.name}.profile-cleanup.json").write_text( "malformed guard that belongs only to Desktop\n", encoding="utf-8", ) report = inspect_profile(home, profile_kind="cli-home") assert report.outcome == "ready" assert report.stable_code != "cleanup_transaction_incomplete" def test_cleanup_unlinks_logs_symlink_without_following_external_target( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) logs = user_data / "logs" for child in logs.iterdir(): child.unlink() logs.rmdir() external = tmp_path / "external-logs" external.mkdir() sentinel = external / "keep.txt" sentinel.write_text("keep\n", encoding="utf-8") try: logs.symlink_to(external, target_is_directory=True) except OSError: pytest.skip("symlink creation is unavailable") inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete" assert not os.path.lexists(logs) assert sentinel.read_text(encoding="utf-8") == "keep\n" @pytest.mark.parametrize( ("filename", "content", "stable_code", "mode"), [ ( "profile-replacement-history.json", "not-json\n", "cleanup_history_invalid", "delete-all-user-data", ), ( ".opensquilla.profile-replace.json", '{"schema_version":1,"phase":"prepared"}\n', "cleanup_transaction_incomplete", "delete-current-profile", ), ], ) def test_malformed_history_or_unfinished_journal_blocks_cleanup( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, filename: str, content: str, stable_code: str, mode: str, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) (user_data / filename).write_text(content, encoding="utf-8") report = cleanup_inspect( user_data, mode=mode, profile_kind="primary", ) assert report.outcome == "blocked" assert report.stable_code == stable_code @pytest.mark.parametrize("selected_kind", ["primary", "recovery"]) def test_delete_current_ignores_unrelated_bad_history_and_non_uuid_recovery( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, selected_kind: str, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) selected_id = str(uuid.uuid4()) selected = _recovery_profile(user_data, selected_id, "selected") unrelated = user_data / "recovery-profiles" / "not-a-uuid" unrelated.mkdir() (unrelated / "keep.txt").write_text("keep\n", encoding="utf-8") history = user_data / "profile-replacement-history.json" history.write_text("malformed and unrelated\n", encoding="utf-8") inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind=selected_kind, recovery_id=selected_id if selected_kind == "recovery" else None, ) assert inspected.outcome == "ready" result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind=selected_kind, recovery_id=selected_id if selected_kind == "recovery" else None, transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "complete" if selected_kind == "recovery": assert not selected.exists() assert (user_data / "opensquilla").exists() else: assert not (user_data / "opensquilla").exists() assert selected.exists() assert history.read_text(encoding="utf-8") == "malformed and unrelated\n" assert (unrelated / "keep.txt").read_text(encoding="utf-8") == "keep\n" def test_non_uuid_recovery_entry_blocks_all_cleanup(tmp_path: Path) -> None: user_data = tmp_path / "user-data" user_data.mkdir() _desktop_primary(user_data) (user_data / "recovery-profiles" / "not-a-uuid").mkdir(parents=True) report = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) assert report.outcome == "blocked" assert report.stable_code == "cleanup_recovery_entry_invalid" def test_cleanup_delete_failure_returns_partial_and_never_claims_all( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) credential = user_data / "desktop-credential.json" original_remove = cleanup_module._remove_no_follow def fail_credential(path: Path) -> None: if path == credential: raise OSError("synthetic sharing violation") original_remove(path) monkeypatch.setattr(cleanup_module, "_remove_no_follow", fail_credential) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "partial" assert result.stable_code == "cleanup_partial" assert credential.exists() def test_cleanup_reports_partial_when_missing_known_path_is_recreated_after_apply( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) credential = user_data / "desktop-credential.json" credential.unlink() inspected = cleanup_inspect( user_data, mode="delete-current-profile", profile_kind="primary", ) credential_item = next( item for item in inspected.items if item.kind == "primary-credential" ) assert credential_item.exists is False real_current_item = cleanup_module._current_item recreated = False def recreate_before_final_inventory(planned): nonlocal recreated if not recreated: recreated = True credential.write_text("{}\n", encoding="utf-8") return real_current_item(planned) monkeypatch.setattr(cleanup_module, "_current_item", recreate_before_final_inventory) result = cleanup_apply( user_data, mode="delete-current-profile", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert recreated is True assert result.outcome == "partial" assert credential.is_file() assert (user_data / ".opensquilla.profile-cleanup.json").is_file() def test_delete_all_reports_partial_for_new_unplanned_user_data_entry( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) late_entry = user_data / "late-chromium-cache" / "cache.bin" real_current_item = cleanup_module._current_item recreated = False def create_unplanned_entry_before_final_inventory(planned): nonlocal recreated if not recreated: recreated = True late_entry.parent.mkdir() late_entry.write_bytes(b"synthetic cache") return real_current_item(planned) monkeypatch.setattr( cleanup_module, "_current_item", create_unplanned_entry_before_final_inventory, ) result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert recreated is True assert result.outcome == "partial" assert late_entry.read_bytes() == b"synthetic cache" assert (user_data / ".opensquilla.profile-cleanup.json").is_file() def test_partial_delete_all_can_be_abandoned_then_new_recovery_profile_runs( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) _desktop_primary(user_data) inspected = cleanup_inspect( user_data, mode="delete-all-user-data", profile_kind="primary", ) credential = user_data / "desktop-credential.json" original_remove = cleanup_module._remove_no_follow def fail_credential(path: Path) -> None: if path == credential: raise OSError("synthetic partial cleanup") original_remove(path) monkeypatch.setattr(cleanup_module, "_remove_no_follow", fail_credential) result = cleanup_apply( user_data, mode="delete-all-user-data", profile_kind="primary", transaction_id=inspected.transaction_id, expected_revision=inspected.revision, confirm_user_data=user_data, ) assert result.outcome == "partial" recovery_id = str(uuid.uuid4()) recovery_root = _recovery_profile(user_data, recovery_id, "new recovery") recovery_home = recovery_root / "opensquilla" (recovery_home / "state").mkdir() blocked = inspect_profile(recovery_home, profile_kind="desktop-recovery") assert blocked.stable_code == "cleanup_transaction_incomplete" assert "abandon-cleanup" in blocked.allowed_actions preserved = abandon_cleanup_transaction( user_data, home=recovery_home, profile_kind="desktop-recovery", transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert preserved.is_file() assert ".profile-cleanup.abandoned." in preserved.name assert not (user_data / ".opensquilla.profile-cleanup.json").exists() assert credential.is_file() usable = inspect_profile(recovery_home, profile_kind="desktop-recovery") assert usable.outcome == "recovery_profile" def test_malformed_cleanup_journal_can_be_quarantined_without_deleting_data( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) recovery_id = str(uuid.uuid4()) recovery_root = _recovery_profile(user_data, recovery_id, "safe recovery") recovery_home = recovery_root / "opensquilla" (recovery_home / "state").mkdir() journal = user_data / ".opensquilla.profile-cleanup.json" journal.write_text("malformed cleanup authority\n", encoding="utf-8") before = (recovery_home / "workspace" / "SOUL.md").read_bytes() blocked = inspect_profile(recovery_home, profile_kind="desktop-recovery") preserved = abandon_cleanup_transaction( user_data, home=recovery_home, profile_kind="desktop-recovery", transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert preserved.read_text(encoding="utf-8") == "malformed cleanup authority\n" assert not journal.exists() assert (recovery_home / "workspace" / "SOUL.md").read_bytes() == before assert inspect_profile( recovery_home, profile_kind="desktop-recovery", ).outcome == "recovery_profile" @pytest.mark.parametrize("mutation", ["identity", "digest"]) def test_abandon_cleanup_rejects_journal_swapped_after_inspection( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, mutation: str, ) -> None: user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) home = _desktop_primary(user_data) journal = user_data / ".opensquilla.profile-cleanup.json" first = b"malformed cleanup journal alpha\n" second = b"malformed cleanup journal bravo\n" assert len(first) == len(second) journal.write_bytes(first) blocked = inspect_profile(home, profile_kind="desktop-primary") assert blocked.stable_code == "cleanup_transaction_incomplete" before = journal.lstat() before_identity = ( before.st_dev, before.st_ino, before.st_mode, before.st_size, before.st_mtime_ns, ) if mutation == "identity": replacement = user_data / ".replacement-cleanup-journal.json" replacement.write_bytes(first) os.replace(replacement, journal) after = journal.lstat() after_identity = ( after.st_dev, after.st_ino, after.st_mode, after.st_size, after.st_mtime_ns, ) assert after_identity != before_identity else: journal.write_bytes(second) os.utime(journal, ns=(before.st_atime_ns, before.st_mtime_ns)) after = journal.lstat() after_identity = ( after.st_dev, after.st_ino, after.st_mode, after.st_size, after.st_mtime_ns, ) assert after_identity == before_identity current = journal.read_bytes() swapped = inspect_profile(home, profile_kind="desktop-primary") assert swapped.stable_code == "cleanup_transaction_incomplete" assert swapped.transaction_id == blocked.transaction_id assert swapped.revision != blocked.revision with pytest.raises(StaleRecoveryTransactionError): abandon_cleanup_transaction( user_data, home=home, profile_kind="desktop-primary", transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert journal.read_bytes() == current assert not tuple(user_data.glob(".opensquilla.profile-cleanup.abandoned.*.json")) def test_abandon_cleanup_rechecks_captured_journal_after_internal_inspection( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.engine as engine_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) home = _desktop_primary(user_data) journal = user_data / ".opensquilla.profile-cleanup.json" original = b"original cleanup authority\n" replacement_data = b"replacement cleanup authority\n" journal.write_bytes(original) blocked = inspect_profile(home, profile_kind="desktop-primary") real_inspect = engine_module.inspect_profile def inspect_then_swap(*args, **kwargs): report = real_inspect(*args, **kwargs) replacement = user_data / ".replacement-cleanup-journal.json" replacement.write_bytes(replacement_data) os.replace(replacement, journal) return report monkeypatch.setattr(engine_module, "inspect_profile", inspect_then_swap) with pytest.raises(RecoveryError) as error: abandon_cleanup_transaction( user_data, home=home, profile_kind="desktop-primary", transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert error.value.stable_code == "config_changed" assert journal.read_bytes() == replacement_data assert not tuple(user_data.glob(".opensquilla.profile-cleanup.abandoned.*.json")) def test_abandon_cleanup_rolls_back_journal_swapped_at_move_boundary( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) home = _desktop_primary(user_data) journal = user_data / ".opensquilla.profile-cleanup.json" journal.write_bytes(b"original cleanup authority\n") blocked = inspect_profile(home, profile_kind="desktop-primary") replacement_data = b"replacement cleanup authority\n" real_move = cleanup_module.native_move_no_replace swapped = False def swap_then_move(source: Path, destination: Path) -> None: nonlocal swapped if source == journal and not swapped: swapped = True replacement = user_data / ".replacement-cleanup-journal.json" replacement.write_bytes(replacement_data) os.replace(replacement, journal) real_move(source, destination) monkeypatch.setattr(cleanup_module, "native_move_no_replace", swap_then_move) with pytest.raises(RecoveryError) as error: abandon_cleanup_transaction( user_data, home=home, profile_kind="desktop-primary", transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert error.value.stable_code == "config_changed" assert journal.read_bytes() == replacement_data assert not tuple(user_data.glob(".opensquilla.profile-cleanup.abandoned.*.json")) def test_abandon_cleanup_reports_unknown_when_swapped_move_cannot_be_rolled_back( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: import opensquilla.recovery.cleanup as cleanup_module user_data = tmp_path / "user-data" user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) home = _desktop_primary(user_data) journal = user_data / ".opensquilla.profile-cleanup.json" journal.write_bytes(b"original cleanup authority\n") blocked = inspect_profile(home, profile_kind="desktop-primary") replacement_data = b"replacement cleanup authority\n" recreated_data = b"new canonical cleanup authority\n" real_move = cleanup_module.native_move_no_replace swapped = False def swap_move_and_recreate(source: Path, destination: Path) -> None: nonlocal swapped if source == journal and not swapped: swapped = True replacement = user_data / ".replacement-cleanup-journal.json" replacement.write_bytes(replacement_data) os.replace(replacement, journal) real_move(source, destination) journal.write_bytes(recreated_data) return real_move(source, destination) monkeypatch.setattr( cleanup_module, "native_move_no_replace", swap_move_and_recreate, ) with pytest.raises(RecoveryError) as error: abandon_cleanup_transaction( user_data, home=home, profile_kind="desktop-primary", transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert error.value.stable_code == "atomic_state_unknown" assert journal.read_bytes() == recreated_data preserved = tuple(user_data.glob(".opensquilla.profile-cleanup.abandoned.*.json")) assert len(preserved) == 1 assert preserved[0].read_bytes() == replacement_data @pytest.mark.parametrize("profile_kind", ["desktop-primary", "desktop-recovery"]) def test_abandon_cleanup_rejects_profile_from_another_user_data_root( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, profile_kind: str, ) -> None: user_data = tmp_path / "selected-user-data" user_data.mkdir() foreign_user_data = tmp_path / "foreign-user-data" foreign_user_data.mkdir() monkeypatch.setenv("OPENSQUILLA_USER_STATE_DIR", str(tmp_path / "lock-state")) if profile_kind == "desktop-primary": foreign_home = _desktop_primary(foreign_user_data) else: recovery_id = str(uuid.uuid4()) foreign_home = ( _recovery_profile(foreign_user_data, recovery_id, "foreign recovery") / "opensquilla" ) selected_journal = user_data / ".opensquilla.profile-cleanup.json" foreign_journal = foreign_user_data / ".opensquilla.profile-cleanup.json" selected_before = b"selected cleanup authority\n" foreign_before = b"foreign cleanup authority\n" selected_journal.write_bytes(selected_before) foreign_journal.write_bytes(foreign_before) blocked = inspect_profile(foreign_home, profile_kind=profile_kind) assert blocked.stable_code == "cleanup_transaction_incomplete" with pytest.raises(RecoveryError) as error: abandon_cleanup_transaction( user_data, home=foreign_home, profile_kind=profile_kind, transaction_id=blocked.transaction_id, expected_revision=blocked.revision, ) assert error.value.stable_code == "cleanup_profile_selector_invalid" assert selected_journal.read_bytes() == selected_before assert foreign_journal.read_bytes() == foreign_before assert not tuple(user_data.glob(".opensquilla.profile-cleanup.abandoned.*.json")) assert not tuple(foreign_user_data.glob(".opensquilla.profile-cleanup.abandoned.*.json"))