feat(tui-v3): migrate legacy utility/maintenance commands to v3 CommandProcessor

src/seedpass/tui_v3/app.py

@@ -63,7 +63,18 @@ class CommandProcessor:
 
         if cmd == "help":
             self.app.notify(
-                "v3 commands: help, stats, atlas, wayfinder, session-status, lock, unlock <password>, refresh, search <query>, search-mode <keyword|hybrid|semantic>, sort <relevance|modified_desc|modified_asc|label_asc|kind|created_desc|most_linked>, filter <all|secrets|docs|keys|2fa>, archived, open <id>, settings, profiles, relay-list, npub, nostr-pubkey, nostr-reset-sync-state, nostr-fresh-namespace, change-password, backup-parent-seed <path> (optional: password), maximize, copy, edit, export, add, seed-plus, archive, restore, delete, ml, mx, db-export <path>, db-import <path>"
+                "v3 commands: help, stats, atlas, wayfinder, session-status, lock, unlock <password>, refresh, "
+                "search <query>, search-mode <keyword|hybrid|semantic>, sort <relevance|modified_desc|modified_asc|label_asc|kind|created_desc|most_linked>, "
+                "filter <all|secrets|docs|keys|2fa>, archived, archive-filter <active|all|archived>, "
+                "open <id>, settings, profiles, relay-list, relay-add <url>, relay-rm <idx>, relay-reset, "
+                "npub, nostr-pubkey, nostr-reset-sync-state, nostr-fresh-namespace, "
+                "sync-now, sync-bg, "
+                "checksum-verify, checksum-update, totp-export [path], "
+                "setting-secret <on|off>, setting-quick-unlock <on|off>, setting-offline <on|off>, "
+                "setting-timeout <s>, setting-kdf-mode <mode>, setting-kdf-iterations <n>, "
+                "density <compact|comfortable>, onboarding, "
+                "change-password, backup-parent-seed <path> (optional: password), maximize, copy, edit, export, "
+                "add, seed-plus, archive, restore, delete, ml, mx, db-export <path>, db-import <path>"
             )
         elif cmd == "stats":
             self.app.notify("Calculating stats...")
@@ -182,6 +193,84 @@ class CommandProcessor:
                 self.app.notify("Usage: db-import <path>", severity="warning")
                 return
             self.app.action_db_import(args[0])
+        elif cmd == "checksum-verify":
+            self.app.action_checksum_verify()
+        elif cmd == "checksum-update":
+            self.app.action_checksum_update()
+        elif cmd == "totp-export":
+            path = args[0] if args else None
+            self.app.action_totp_export(path)
+        elif cmd == "sync-now":
+            self.app.action_sync_now()
+        elif cmd == "sync-bg":
+            self.app.action_sync_bg()
+        elif cmd == "relay-add":
+            if not args:
+                self.app.notify("Usage: relay-add <url>", severity="warning")
+                return
+            self.app.action_relay_add(args[0])
+        elif cmd == "relay-rm":
+            if not args:
+                self.app.notify("Usage: relay-rm <idx>", severity="warning")
+                return
+            try:
+                idx = int(args[0])
+            except ValueError:
+                self.app.notify("relay-rm: index must be an integer", severity="error")
+                return
+            self.app.action_relay_rm(idx)
+        elif cmd == "relay-reset":
+            self.app.action_relay_reset()
+        elif cmd == "setting-secret":
+            if not args:
+                self.app.notify("Usage: setting-secret <on|off>", severity="warning")
+                return
+            self.app.action_setting_secret(args[0])
+        elif cmd == "setting-quick-unlock":
+            if not args:
+                self.app.notify(
+                    "Usage: setting-quick-unlock <on|off>", severity="warning"
+                )
+                return
+            self.app.action_setting_quick_unlock(args[0])
+        elif cmd == "setting-offline":
+            if not args:
+                self.app.notify("Usage: setting-offline <on|off>", severity="warning")
+                return
+            self.app.action_setting_offline(args[0])
+        elif cmd == "setting-timeout":
+            if not args:
+                self.app.notify("Usage: setting-timeout <seconds>", severity="warning")
+                return
+            self.app.action_setting_timeout(args[0])
+        elif cmd == "setting-kdf-mode":
+            if not args:
+                self.app.notify("Usage: setting-kdf-mode <mode>", severity="warning")
+                return
+            self.app.action_setting_kdf_mode(args[0])
+        elif cmd == "setting-kdf-iterations":
+            if not args:
+                self.app.notify(
+                    "Usage: setting-kdf-iterations <n>", severity="warning"
+                )
+                return
+            self.app.action_setting_kdf_iterations(args[0])
+        elif cmd == "archive-filter":
+            if not args:
+                self.app.notify(
+                    "Usage: archive-filter <active|all|archived>", severity="warning"
+                )
+                return
+            self.app.action_archive_filter(args[0])
+        elif cmd == "density":
+            if not args:
+                self.app.notify(
+                    "Usage: density <compact|comfortable>", severity="warning"
+                )
+                return
+            self.app.action_set_density(args[0])
+        elif cmd in {"onboarding", "welcome", "quickstart"}:
+            self.app.action_show_onboarding()
         else:
             self.app.notify(f"Unknown v3 command: {cmd}", severity="warning")
 
@@ -837,6 +926,8 @@ class SeedPassTuiV3(App[None]):
     search_sort = reactive("relevance")
     filter_kind = reactive("all")
     show_archived = reactive(False)
+    filter_archived_only = reactive(False)
+    density_mode = reactive("comfortable")
 
     # Internal state for sensitive actions
     _pending_sensitive_confirm: tuple[str, int, float] | None = None
@@ -1884,3 +1975,231 @@ class SeedPassTuiV3(App[None]):
                     board.reveal_data = data
         except Exception:
             pass
+
+    # ---------------------------------------------------------------------------
+    # Utility / maintenance actions (legacy parity)
+    # ---------------------------------------------------------------------------
+
+    def action_checksum_verify(self) -> None:
+        """Verify the application checksum."""
+        utility = self.services.get("utility")
+        if not utility:
+            self.notify("Utility service offline", severity="error")
+            return
+        try:
+            utility.verify_checksum()
+            self.notify("Checksum verified")
+        except Exception as e:
+            self.notify(f"Checksum verify failed: {e}", severity="error")
+
+    def action_checksum_update(self) -> None:
+        """Update / regenerate the application checksum."""
+        utility = self.services.get("utility")
+        if not utility:
+            self.notify("Utility service offline", severity="error")
+            return
+        try:
+            utility.update_checksum()
+            self.notify("Checksum updated")
+        except Exception as e:
+            self.notify(f"Checksum update failed: {e}", severity="error")
+
+    def action_totp_export(self, path: str | None = None) -> None:
+        """Export all TOTP entries, optionally writing to a file."""
+        entry = self.services.get("entry")
+        if not entry:
+            self.notify("Entry service offline", severity="error")
+            return
+        try:
+            data = entry.export_totp_entries()
+            if path:
+                import json
+                from pathlib import Path
+
+                Path(path).write_text(json.dumps(data, indent=2), encoding="utf-8")
+                self.notify(f"TOTP entries exported to {path}")
+            else:
+                count = len(data) if isinstance(data, dict) else 0
+                self.notify(
+                    f"TOTP export: {count} entries (provide a path to save to file)"
+                )
+        except Exception as e:
+            self.notify(f"TOTP export failed: {e}", severity="error")
+
+    def action_sync_now(self) -> None:
+        """Synchronise vault to Nostr immediately."""
+        sync = self.services.get("sync")
+        if not sync:
+            self.notify("Sync service offline", severity="error")
+            return
+        try:
+            self.notify("Syncing vault to Nostr…")
+            result = sync.sync()
+            if result:
+                self.notify("Sync complete")
+            else:
+                self.notify("Sync returned no events", severity="warning")
+        except Exception as e:
+            self.notify(f"Sync failed: {e}", severity="error")
+
+    def action_sync_bg(self) -> None:
+        """Start a background Nostr vault sync."""
+        sync = self.services.get("sync")
+        if not sync:
+            self.notify("Sync service offline", severity="error")
+            return
+        try:
+            sync.start_background_vault_sync()
+            self.notify("Background sync started")
+        except Exception as e:
+            self.notify(f"Background sync failed: {e}", severity="error")
+
+    def action_relay_add(self, url: str) -> None:
+        """Add a Nostr relay."""
+        nostr = self.services.get("nostr")
+        if not nostr:
+            self.notify("Nostr service offline", severity="error")
+            return
+        try:
+            nostr.add_relay(url)
+            self.notify(f"Relay added: {url}")
+        except Exception as e:
+            self.notify(f"Relay add failed: {e}", severity="error")
+
+    def action_relay_rm(self, idx: int) -> None:
+        """Remove a Nostr relay by index."""
+        nostr = self.services.get("nostr")
+        if not nostr:
+            self.notify("Nostr service offline", severity="error")
+            return
+        try:
+            nostr.remove_relay(idx)
+            self.notify(f"Relay #{idx} removed")
+        except Exception as e:
+            self.notify(f"Relay remove failed: {e}", severity="error")
+
+    def action_relay_reset(self) -> None:
+        """Reset Nostr relays to built-in defaults."""
+        nostr = self.services.get("nostr")
+        if not nostr:
+            self.notify("Nostr service offline", severity="error")
+            return
+        try:
+            relays = nostr.reset_relays()
+            self.notify(f"Relays reset to defaults ({len(relays)} relays)")
+        except Exception as e:
+            self.notify(f"Relay reset failed: {e}", severity="error")
+
+    def action_setting_secret(self, value: str) -> None:
+        """Toggle secret mode on or off."""
+        config = self.services.get("config")
+        if not config:
+            self.notify("Config service offline", severity="error")
+            return
+        try:
+            enabled = value.lower() in {"on", "1", "true", "yes", "y"}
+            delay = config.get_clipboard_clear_delay()
+            config.set_secret_mode(enabled, delay)
+            self.notify(f"Secret mode: {'on' if enabled else 'off'}")
+        except Exception as e:
+            self.notify(f"Setting failed: {e}", severity="error")
+
+    def action_setting_quick_unlock(self, value: str) -> None:
+        """Enable or disable quick-unlock."""
+        config = self.services.get("config")
+        if not config:
+            self.notify("Config service offline", severity="error")
+            return
+        try:
+            config.set("quick_unlock", value)
+            self.notify(f"Quick unlock: {value}")
+        except Exception as e:
+            self.notify(f"Setting failed: {e}", severity="error")
+
+    def action_setting_offline(self, value: str) -> None:
+        """Enable or disable offline mode."""
+        config = self.services.get("config")
+        if not config:
+            self.notify("Config service offline", severity="error")
+            return
+        try:
+            enabled = value.lower() in {"on", "1", "true", "yes", "y"}
+            config.set_offline_mode(enabled)
+            self.notify(f"Offline mode: {'on' if enabled else 'off'}")
+        except Exception as e:
+            self.notify(f"Setting failed: {e}", severity="error")
+
+    def action_setting_timeout(self, seconds: str) -> None:
+        """Set the inactivity auto-lock timeout in seconds."""
+        config = self.services.get("config")
+        if not config:
+            self.notify("Config service offline", severity="error")
+            return
+        try:
+            config.set("inactivity_timeout", seconds)
+            self.notify(f"Inactivity timeout: {seconds}s")
+        except Exception as e:
+            self.notify(f"Setting failed: {e}", severity="error")
+
+    def action_setting_kdf_mode(self, mode: str) -> None:
+        """Set the KDF algorithm (e.g. argon2id, pbkdf2)."""
+        config = self.services.get("config")
+        if not config:
+            self.notify("Config service offline", severity="error")
+            return
+        try:
+            config.set("kdf_mode", mode)
+            self.notify(f"KDF mode: {mode}")
+        except Exception as e:
+            self.notify(f"Setting failed: {e}", severity="error")
+
+    def action_setting_kdf_iterations(self, n: str) -> None:
+        """Set the number of KDF iterations."""
+        config = self.services.get("config")
+        if not config:
+            self.notify("Config service offline", severity="error")
+            return
+        try:
+            config.set("kdf_iterations", n)
+            self.notify(f"KDF iterations: {n}")
+        except Exception as e:
+            self.notify(f"Setting failed: {e}", severity="error")
+
+    def action_archive_filter(self, mode: str) -> None:
+        """Set the archive visibility scope: active, all, or archived."""
+        normalized = mode.lower()
+        if normalized not in {"active", "all", "archived"}:
+            self.notify(
+                "Usage: archive-filter <active|all|archived>", severity="warning"
+            )
+            return
+        if normalized == "archived":
+            self.show_archived = True
+            self.filter_archived_only = True
+        elif normalized == "all":
+            self.show_archived = True
+            self.filter_archived_only = False
+        else:
+            self.show_archived = False
+            self.filter_archived_only = False
+        self.notify(f"Archive filter: {normalized}")
+        self.action_refresh()
+
+    def action_set_density(self, mode: str) -> None:
+        """Switch display density between compact and comfortable."""
+        normalized = mode.lower()
+        if normalized not in {"compact", "comfortable"}:
+            self.notify(
+                "Usage: density <compact|comfortable>", severity="warning"
+            )
+            return
+        self.density_mode = normalized
+        self.notify(f"Density: {normalized}")
+
+    def action_show_onboarding(self) -> None:
+        """Show onboarding / quickstart guidance."""
+        self.notify(
+            "Onboarding: add your first entry with 'add', reveal it with 'v', "
+            "inspect fields in the inspector pane, and use Ctrl+P to open the "
+            "command palette for all available operations."
+        )

src/tests/test_tui_v3_parity.py

@@ -164,6 +164,73 @@ class MockVaultService:
         }
 
 
+class MockUtilityService:
+    def __init__(self):
+        self.checksum_verified = False
+        self.checksum_updated = False
+
+    def verify_checksum(self):
+        self.checksum_verified = True
+
+    def update_checksum(self):
+        self.checksum_updated = True
+
+
+class MockSyncService:
+    def __init__(self):
+        self.synced = False
+        self.bg_synced = False
+
+    def sync(self):
+        self.synced = True
+        return {"events": 1}
+
+    def start_background_vault_sync(self, summary=None):
+        self.bg_synced = True
+
+
+class MockNostrService:
+    def __init__(self):
+        self.relays: list[str] = ["wss://relay.example.com"]
+        self.added: list[str] = []
+        self.removed: list[int] = []
+        self.reset_called = False
+
+    def add_relay(self, url: str):
+        self.added.append(url)
+        self.relays.append(url)
+
+    def remove_relay(self, idx: int):
+        self.removed.append(idx)
+
+    def reset_relays(self) -> list[str]:
+        self.reset_called = True
+        self.relays = ["wss://default.relay"]
+        return list(self.relays)
+
+
+class MockConfigService:
+    def __init__(self):
+        self._cfg: dict = {}
+        self.secret_mode_calls: list[tuple] = []
+        self.offline_mode_calls: list[bool] = []
+
+    def get(self, key: str):
+        return self._cfg.get(key)
+
+    def set(self, key: str, value: str):
+        self._cfg[key] = value
+
+    def get_clipboard_clear_delay(self) -> int:
+        return 30
+
+    def set_secret_mode(self, enabled: bool, delay: int):
+        self.secret_mode_calls.append((enabled, delay))
+
+    def set_offline_mode(self, enabled: bool):
+        self.offline_mode_calls.append(enabled)
+
+
 @pytest.mark.anyio
 async def test_v3_reveal_copy_parity():
     app = SeedPassTuiV3(
@@ -312,3 +379,183 @@ async def test_v3_action_bar_only_shows_context_actions_for_selected_kind() -> N
         assert "Load" in rendered
         assert "Delete" in rendered
         assert "Export" not in rendered
+
+
+# ---------------------------------------------------------------------------
+# Parity tests: legacy utility/maintenance commands migrated to v3
+# ---------------------------------------------------------------------------
+
+
+def _make_full_app(**extra_services):
+    """Build a SeedPassTuiV3 with all mock services wired up."""
+    entry_svc = MockEntryService()
+    vault_svc = MockVaultService()
+    utility_svc = MockUtilityService()
+    sync_svc = MockSyncService()
+    nostr_svc = MockNostrService()
+    config_svc = MockConfigService()
+
+    app = SeedPassTuiV3(
+        entry_service_factory=lambda: entry_svc,
+        vault_service_factory=lambda: vault_svc,
+        utility_service_factory=lambda: utility_svc,
+        sync_service_factory=lambda: sync_svc,
+        nostr_service_factory=lambda: nostr_svc,
+        config_service_factory=lambda: config_svc,
+    )
+    return app, entry_svc, vault_svc, utility_svc, sync_svc, nostr_svc, config_svc
+
+
+@pytest.mark.anyio
+async def test_v3_checksum_commands() -> None:
+    app, *_, utility_svc, _, _, _ = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.action_checksum_verify()
+        assert utility_svc.checksum_verified is True
+
+        app.action_checksum_update()
+        assert utility_svc.checksum_updated is True
+
+
+@pytest.mark.anyio
+async def test_v3_totp_export_no_path() -> None:
+    app, entry_svc, *_ = _make_full_app()
+    # Provide a basic export_totp_entries method
+    entry_svc.export_totp_entries = lambda: {"1": "otpauth://totp/label?secret=ABC"}
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        # Should not raise; notifies count
+        app.action_totp_export(None)
+
+
+@pytest.mark.anyio
+async def test_v3_sync_commands() -> None:
+    app, _, _, _, sync_svc, _, _ = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.action_sync_now()
+        assert sync_svc.synced is True
+
+        app.action_sync_bg()
+        assert sync_svc.bg_synced is True
+
+
+@pytest.mark.anyio
+async def test_v3_relay_commands() -> None:
+    app, _, _, _, _, nostr_svc, _ = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+
+        app.action_relay_add("wss://new.relay")
+        assert "wss://new.relay" in nostr_svc.added
+
+        app.action_relay_rm(0)
+        assert 0 in nostr_svc.removed
+
+        app.action_relay_reset()
+        assert nostr_svc.reset_called is True
+
+
+@pytest.mark.anyio
+async def test_v3_setting_secret() -> None:
+    app, _, _, _, _, _, config_svc = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.action_setting_secret("on")
+        assert config_svc.secret_mode_calls[-1][0] is True
+
+        app.action_setting_secret("off")
+        assert config_svc.secret_mode_calls[-1][0] is False
+
+
+@pytest.mark.anyio
+async def test_v3_setting_offline() -> None:
+    app, _, _, _, _, _, config_svc = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.action_setting_offline("on")
+        assert config_svc.offline_mode_calls[-1] is True
+
+        app.action_setting_offline("off")
+        assert config_svc.offline_mode_calls[-1] is False
+
+
+@pytest.mark.anyio
+async def test_v3_setting_config_keys() -> None:
+    app, _, _, _, _, _, config_svc = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.action_setting_quick_unlock("on")
+        assert config_svc._cfg.get("quick_unlock") == "on"
+
+        app.action_setting_timeout("600")
+        assert config_svc._cfg.get("inactivity_timeout") == "600"
+
+        app.action_setting_kdf_mode("argon2id")
+        assert config_svc._cfg.get("kdf_mode") == "argon2id"
+
+        app.action_setting_kdf_iterations("200000")
+        assert config_svc._cfg.get("kdf_iterations") == "200000"
+
+
+@pytest.mark.anyio
+async def test_v3_archive_filter() -> None:
+    app = SeedPassTuiV3(
+        entry_service_factory=lambda: MockEntryService(),
+        vault_service_factory=lambda: MockVaultService(),
+    )
+    async with app.run_test() as pilot:
+        await pilot.pause()
+
+        app.action_archive_filter("active")
+        assert app.show_archived is False
+        assert app.filter_archived_only is False
+
+        app.action_archive_filter("all")
+        assert app.show_archived is True
+        assert app.filter_archived_only is False
+
+        app.action_archive_filter("archived")
+        assert app.show_archived is True
+        assert app.filter_archived_only is True
+
+
+@pytest.mark.anyio
+async def test_v3_density() -> None:
+    app = SeedPassTuiV3(
+        entry_service_factory=lambda: MockEntryService(),
+        vault_service_factory=lambda: MockVaultService(),
+    )
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        app.action_set_density("compact")
+        assert app.density_mode == "compact"
+
+        app.action_set_density("comfortable")
+        assert app.density_mode == "comfortable"
+
+
+@pytest.mark.anyio
+async def test_v3_command_processor_routes_utility_commands() -> None:
+    """Verify the CommandProcessor correctly routes new utility commands."""
+    app, *_, utility_svc, sync_svc, nostr_svc, _ = _make_full_app()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        proc = app.processor
+
+        proc.execute("checksum-verify")
+        assert utility_svc.checksum_verified is True
+
+        proc.execute("sync-now")
+        assert sync_svc.synced is True
+
+        proc.execute("relay-add wss://proc.relay")
+        assert "wss://proc.relay" in nostr_svc.added
+
+        proc.execute("archive-filter archived")
+        assert app.show_archived is True
+        assert app.filter_archived_only is True
+
+        proc.execute("density compact")
+        assert app.density_mode == "compact"