diff --git a/backend/scripts/migrate_paths_to_workspace.py b/backend/scripts/migrate_paths_to_workspace.py new file mode 100644 index 00000000..f4750290 --- /dev/null +++ b/backend/scripts/migrate_paths_to_workspace.py @@ -0,0 +1,212 @@ +"""One-time migration: lift per-user thread / memory / agent dirs into per-workspace layout. + +PR4 introduced ``{base_dir}/users/{user_id}/...`` for per-user isolation. PR6 +adds the multi-tenant top dimension: ``{base_dir}/workspaces/{wid}/...``, +with per-user state nested under each workspace +(``{base_dir}/workspaces/{wid}/users/{uid}/memory.json`` etc.). This script +walks the old PR4 layout, looks up each user's ``default_workspace_id`` +from the ``users`` table, and rewrites the path. + +Usage: + PYTHONPATH=. python scripts/migrate_paths_to_workspace.py [--dry-run] [--default-workspace WID] + +The script is idempotent — re-running it after a successful migration is a no-op. +A ``--dry-run`` invocation must not write anything; it only logs what would happen. + +Mapping rules: + +- ``{base_dir}/users/{uid}/threads/{tid}/`` -> ``{base_dir}/workspaces/{wid}/threads/{tid}/`` +- ``{base_dir}/users/{uid}/memory.json`` -> ``{base_dir}/workspaces/{wid}/users/{uid}/memory.json`` +- ``{base_dir}/users/{uid}/agents/{name}/`` -> ``{base_dir}/workspaces/{wid}/users/{uid}/agents/{name}/`` + +``{wid}`` is read from ``users.default_workspace_id``. Users without one +fall through to ``--default-workspace`` (defaults to the special bucket +``"legacy_workspace"`` so they can be triaged manually). Pre-existing +destinations are preserved; legacy copies are moved to +``{base_dir}/migration-conflicts/<...>`` for human review. +""" + +from __future__ import annotations + +import argparse +import logging +import shutil +import sqlite3 +from pathlib import Path + +from deerflow.config.paths import Paths, get_paths + +logger = logging.getLogger(__name__) + +LEGACY_WORKSPACE_FALLBACK = "legacy_workspace" + + +def _load_user_workspaces(paths: Paths) -> dict[str, str | None]: + """Read ``user_id -> default_workspace_id`` from the local sqlite DB. + + Returns an empty dict when the database does not exist (fresh install + / Postgres-only deployments). The Postgres path is out of scope for + this script — the operator should run it with ``--default-workspace`` + set to the target workspace and skip the DB lookup. + """ + db_path = paths.base_dir / "deer-flow.db" + if not db_path.exists(): + logger.info("No sqlite database at %s — every user will use the fallback workspace.", db_path) + return {} + + conn = sqlite3.connect(str(db_path)) + try: + try: + cursor = conn.execute("SELECT id, default_workspace_id FROM users") + except sqlite3.OperationalError as e: + logger.warning("Failed to query users.default_workspace_id: %s", e) + return {} + return {row[0]: row[1] for row in cursor.fetchall()} + finally: + conn.close() + + +def _resolve_workspace(user_id: str, user_workspaces: dict[str, str | None], fallback: str) -> str: + wid = user_workspaces.get(user_id) + return wid or fallback + + +def _move(src: Path, dest: Path, conflict_root: Path, *, dry_run: bool, label: str) -> str: + """Move ``src`` to ``dest``; on conflict, divert legacy under ``conflict_root``. + + Returns a short string describing what happened (for the report). + """ + if dest.exists(): + conflict_dest = conflict_root / label / src.name + if not dry_run: + conflict_dest.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(src), str(conflict_dest)) + logger.warning("Conflict for %s/%s: legacy copy diverted to %s", label, src.name, conflict_dest) + return f"conflict -> {conflict_dest}" + + if not dry_run: + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.move(str(src), str(dest)) + return f"moved -> {dest}" + + +def migrate_user_tree( + paths: Paths, + user_id: str, + workspace_id: str, + *, + dry_run: bool, +) -> list[dict]: + """Lift one user's tree under a workspace. Returns per-asset report rows.""" + report: list[dict] = [] + legacy_user_dir = paths.base_dir / "users" / user_id + if not legacy_user_dir.exists(): + return report + + conflict_root = paths.base_dir / "migration-conflicts" / "workspace-migration" + + # 1. threads/{tid}/ -> workspaces/{wid}/threads/{tid}/ + legacy_threads = legacy_user_dir / "threads" + if legacy_threads.exists(): + for thread_dir in sorted(legacy_threads.iterdir()): + if not thread_dir.is_dir(): + continue + dest = paths.thread_dir(thread_dir.name, workspace_id=workspace_id) + action = _move(thread_dir, dest, conflict_root, dry_run=dry_run, label=f"threads/{workspace_id}") + report.append({"asset": "thread", "user_id": user_id, "workspace_id": workspace_id, "name": thread_dir.name, "action": action}) + if not dry_run and legacy_threads.exists() and not any(legacy_threads.iterdir()): + legacy_threads.rmdir() + + # 2. memory.json -> workspaces/{wid}/users/{uid}/memory.json + legacy_mem = legacy_user_dir / "memory.json" + if legacy_mem.exists(): + dest = paths.user_memory_file(user_id, workspace_id=workspace_id) + action = _move(legacy_mem, dest, conflict_root, dry_run=dry_run, label=f"users/{user_id}/memory") + report.append({"asset": "memory", "user_id": user_id, "workspace_id": workspace_id, "name": "memory.json", "action": action}) + + # 3. agents/{name}/ -> workspaces/{wid}/users/{uid}/agents/{name}/ + legacy_agents = legacy_user_dir / "agents" + if legacy_agents.exists(): + for agent_dir in sorted(legacy_agents.iterdir()): + if not agent_dir.is_dir(): + continue + dest = paths.user_agent_dir(user_id, agent_dir.name, workspace_id=workspace_id) + action = _move(agent_dir, dest, conflict_root, dry_run=dry_run, label=f"users/{user_id}/agents") + report.append({"asset": "agent", "user_id": user_id, "workspace_id": workspace_id, "name": agent_dir.name, "action": action}) + if not dry_run and legacy_agents.exists() and not any(legacy_agents.iterdir()): + legacy_agents.rmdir() + + # 4. If the user dir is now empty, remove it so lifespan warnings clear. + if not dry_run and legacy_user_dir.exists() and not any(legacy_user_dir.iterdir()): + legacy_user_dir.rmdir() + + return report + + +def migrate( + paths: Paths, + *, + user_workspaces: dict[str, str | None], + fallback_workspace: str, + dry_run: bool, +) -> list[dict]: + """Top-level entry point: iterate every user under ``{base_dir}/users``.""" + legacy_users = paths.base_dir / "users" + if not legacy_users.exists(): + logger.info("No legacy ``users/`` directory under %s — nothing to migrate.", paths.base_dir) + return [] + + report: list[dict] = [] + for user_dir in sorted(legacy_users.iterdir()): + if not user_dir.is_dir(): + continue + user_id = user_dir.name + workspace_id = _resolve_workspace(user_id, user_workspaces, fallback_workspace) + logger.info("Migrating user %s -> workspace %s", user_id, workspace_id) + report.extend(migrate_user_tree(paths, user_id, workspace_id, dry_run=dry_run)) + + if not dry_run and legacy_users.exists() and not any(legacy_users.iterdir()): + legacy_users.rmdir() + + return report + + +def main() -> None: + parser = argparse.ArgumentParser(description="Lift legacy per-user paths into per-workspace layout (PR6).") + parser.add_argument("--dry-run", action="store_true", help="Log actions without making changes.") + parser.add_argument( + "--default-workspace", + default=LEGACY_WORKSPACE_FALLBACK, + metavar="WID", + help=(f"Workspace id to use for users without a ``default_workspace_id`` in the DB. Defaults to ``{LEGACY_WORKSPACE_FALLBACK}`` (matches the orphan-row bucket used by PR5 backfill)."), + ) + args = parser.parse_args() + + logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") + + paths = get_paths() + logger.info("Base directory: %s", paths.base_dir) + logger.info("Dry run: %s", args.dry_run) + logger.info("Fallback workspace: %s", args.default_workspace) + + user_workspaces = _load_user_workspaces(paths) + logger.info("Loaded %d user->workspace mappings from DB", len(user_workspaces)) + + report = migrate( + paths, + user_workspaces=user_workspaces, + fallback_workspace=args.default_workspace, + dry_run=args.dry_run, + ) + + if not report: + logger.info("Nothing to migrate.") + return + + logger.info("Migration report (%d entries):", len(report)) + for entry in report: + logger.info(" asset=%s user=%s workspace=%s name=%s action=%s", entry["asset"], entry["user_id"], entry["workspace_id"], entry["name"], entry["action"]) + + +if __name__ == "__main__": + main() diff --git a/backend/tests/test_migrate_paths_to_workspace.py b/backend/tests/test_migrate_paths_to_workspace.py new file mode 100644 index 00000000..2d429034 --- /dev/null +++ b/backend/tests/test_migrate_paths_to_workspace.py @@ -0,0 +1,155 @@ +"""PR6 T6.13 — workspace-path migration script tests. + +Builds the legacy ``users/{uid}/...`` tree under a temp base dir, +points a populated sqlite ``users`` table at it via the conventional +``deer-flow.db`` location, and asserts the migration produces the new +``workspaces/{wid}/...`` layout. Covers: + +- threads / memory.json / custom agents all rewritten under the workspace +- ``--dry-run`` writes nothing +- pre-existing destinations get diverted to ``migration-conflicts/`` +- users without a ``default_workspace_id`` fall back to the explicit flag +- empty legacy dirs are cleaned up +""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest + +from deerflow.config.paths import Paths +from scripts.migrate_paths_to_workspace import ( + LEGACY_WORKSPACE_FALLBACK, + _load_user_workspaces, + migrate, +) + + +def _build_legacy_tree(base: Path, *, user_id: str, thread_ids: tuple[str, ...] = (), with_memory: bool = False, agent_names: tuple[str, ...] = ()) -> None: + user_root = base / "users" / user_id + user_root.mkdir(parents=True, exist_ok=True) + for tid in thread_ids: + (user_root / "threads" / tid / "user-data" / "workspace").mkdir(parents=True, exist_ok=True) + (user_root / "threads" / tid / "user-data" / "workspace" / "marker.txt").write_text(f"{user_id}/{tid}", encoding="utf-8") + if with_memory: + (user_root / "memory.json").write_text(f'{{"user_id": "{user_id}"}}', encoding="utf-8") + for name in agent_names: + (user_root / "agents" / name).mkdir(parents=True, exist_ok=True) + (user_root / "agents" / name / "SOUL.md").write_text(f"# {name}", encoding="utf-8") + + +def _seed_db(base: Path, *, users: dict[str, str | None]) -> None: + db_path = base / "deer-flow.db" + conn = sqlite3.connect(str(db_path)) + try: + conn.execute("CREATE TABLE users (id TEXT PRIMARY KEY, default_workspace_id TEXT)") + conn.executemany("INSERT INTO users (id, default_workspace_id) VALUES (?, ?)", list(users.items())) + conn.commit() + finally: + conn.close() + + +@pytest.fixture +def base(tmp_path: Path) -> Path: + return tmp_path + + +def test_migrate_threads_under_workspace(base: Path): + _build_legacy_tree(base, user_id="alice", thread_ids=("t1",)) + _seed_db(base, users={"alice": "ws-alpha"}) + paths = Paths(base) + + report = migrate(paths, user_workspaces=_load_user_workspaces(paths), fallback_workspace=LEGACY_WORKSPACE_FALLBACK, dry_run=False) + + assert (base / "workspaces" / "ws-alpha" / "threads" / "t1" / "user-data" / "workspace" / "marker.txt").read_text(encoding="utf-8") == "alice/t1" + assert not (base / "users" / "alice" / "threads").exists() + assert {entry["asset"] for entry in report} == {"thread"} + + +def test_migrate_memory_and_agents_nested_under_workspace_and_user(base: Path): + _build_legacy_tree(base, user_id="alice", with_memory=True, agent_names=("code-reviewer",)) + _seed_db(base, users={"alice": "ws-alpha"}) + paths = Paths(base) + + migrate(paths, user_workspaces=_load_user_workspaces(paths), fallback_workspace=LEGACY_WORKSPACE_FALLBACK, dry_run=False) + + assert (base / "workspaces" / "ws-alpha" / "users" / "alice" / "memory.json").exists() + assert (base / "workspaces" / "ws-alpha" / "users" / "alice" / "agents" / "code-reviewer" / "SOUL.md").exists() + + +def test_dry_run_writes_nothing(base: Path): + _build_legacy_tree(base, user_id="alice", thread_ids=("t1",), with_memory=True, agent_names=("a1",)) + _seed_db(base, users={"alice": "ws-alpha"}) + paths = Paths(base) + + report = migrate(paths, user_workspaces=_load_user_workspaces(paths), fallback_workspace=LEGACY_WORKSPACE_FALLBACK, dry_run=True) + + # Source unchanged + assert (base / "users" / "alice" / "threads" / "t1" / "user-data" / "workspace" / "marker.txt").exists() + assert (base / "users" / "alice" / "memory.json").exists() + assert (base / "users" / "alice" / "agents" / "a1" / "SOUL.md").exists() + # No destination created + assert not (base / "workspaces").exists() + # Report still populated so operator sees what *would* happen + assert len(report) == 3 + + +def test_fallback_workspace_used_when_user_has_no_default(base: Path): + _build_legacy_tree(base, user_id="alice", thread_ids=("t1",)) + _seed_db(base, users={"alice": None}) + paths = Paths(base) + + migrate(paths, user_workspaces=_load_user_workspaces(paths), fallback_workspace="legacy_workspace", dry_run=False) + + assert (base / "workspaces" / "legacy_workspace" / "threads" / "t1" / "user-data" / "workspace" / "marker.txt").exists() + + +def test_conflict_routes_legacy_to_migration_conflicts(base: Path): + # Pre-create the destination with a different marker so the move sees a conflict. + _build_legacy_tree(base, user_id="alice", thread_ids=("t1",)) + (base / "workspaces" / "ws-alpha" / "threads" / "t1" / "user-data" / "workspace").mkdir(parents=True) + (base / "workspaces" / "ws-alpha" / "threads" / "t1" / "user-data" / "workspace" / "marker.txt").write_text("preexisting", encoding="utf-8") + _seed_db(base, users={"alice": "ws-alpha"}) + paths = Paths(base) + + report = migrate(paths, user_workspaces=_load_user_workspaces(paths), fallback_workspace=LEGACY_WORKSPACE_FALLBACK, dry_run=False) + + assert (base / "workspaces" / "ws-alpha" / "threads" / "t1" / "user-data" / "workspace" / "marker.txt").read_text(encoding="utf-8") == "preexisting" + conflict_marker = base / "migration-conflicts" / "workspace-migration" / "threads/ws-alpha" / "t1" + assert conflict_marker.exists() + assert any("conflict" in entry["action"] for entry in report) + + +def test_empty_users_dir_removed_after_full_migration(base: Path): + _build_legacy_tree(base, user_id="alice", thread_ids=("t1",)) + _seed_db(base, users={"alice": "ws-alpha"}) + paths = Paths(base) + + migrate(paths, user_workspaces=_load_user_workspaces(paths), fallback_workspace=LEGACY_WORKSPACE_FALLBACK, dry_run=False) + + assert not (base / "users").exists(), "Empty legacy users/ dir should be cleaned up" + + +def test_no_users_directory_is_noop(base: Path): + paths = Paths(base) + report = migrate(paths, user_workspaces={}, fallback_workspace=LEGACY_WORKSPACE_FALLBACK, dry_run=False) + assert report == [] + + +def test_missing_db_returns_empty_mapping(base: Path): + paths = Paths(base) + assert _load_user_workspaces(paths) == {} + + +def test_db_without_users_table_returns_empty(base: Path): + db_path = base / "deer-flow.db" + conn = sqlite3.connect(str(db_path)) + try: + conn.execute("CREATE TABLE other_table (x INTEGER)") + conn.commit() + finally: + conn.close() + paths = Paths(base) + assert _load_user_workspaces(paths) == {}