diff --git a/CHANGELOG.md b/CHANGELOG.md index d595794216..2b7c3fd6e7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -804,6 +804,17 @@ This release closes that milestone with **439 merged pull requests**. ### Fixed +- **persistence:** `scripts/migrate_user_isolation.py` now moves each legacy + thread to the user who owns it. It looked for thread owners in + `{base_dir}/deer-flow.db`, a file DeerFlow never creates (the database is + `{sqlite_dir}/deerflow.db`, or PostgreSQL), so the owner list was always empty + and every legacy thread was moved to `users/default/`. Owners now come from the + `threads_meta` table of the database configured in `config.yaml`. If legacy + threads exist but that table cannot be read, the script stops before moving + anything; `--allow-missing-thread-owners` assigns every legacy thread to + `default` instead, for installs that never recorded thread owners. Installs + that already ran the old script can recover their threads with the steps in + `docker/provisioner/README.md`. ([#6450]) - **sandbox:** A failed command in the e2b sandbox now keeps its output and exit code. The e2b SDK raises `CommandExitException` on a nonzero exit instead of returning a result, so `E2BSandbox.execute_command` returned @@ -9180,3 +9191,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6445]: https://github.com/bytedance/deer-flow/pull/6445 [#6447]: https://github.com/bytedance/deer-flow/pull/6447 [#6448]: https://github.com/bytedance/deer-flow/pull/6448 +[#6450]: https://github.com/bytedance/deer-flow/pull/6450 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index eb8560107d..d144c31a56 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -694,6 +694,12 @@ ### 修复 +- **持久化：** `scripts/migrate_user_isolation.py` 现在会把每个 legacy 线程移动到其所属用户下。此前脚本在 + `{base_dir}/deer-flow.db` 中查找线程归属，而 DeerFlow 从不创建这个文件（数据库是 `{sqlite_dir}/deerflow.db` + 或 PostgreSQL），因此归属映射始终为空，所有 legacy 线程都被移到 `users/default/`。现在归属从 `config.yaml` + 所配置数据库的 `threads_meta` 表读取；若存在 legacy 线程但无法读取该表，脚本会在移动任何数据前退出。对于 + 从未记录过线程归属的安装，可传 `--allow-missing-thread-owners`，此时所有 legacy 线程归入 `default`。 + 已运行过旧脚本的安装可按 `docker/provisioner/README.md` 中的步骤恢复线程。([#6450]) - **沙箱：** e2b 沙箱中失败的命令现在会保留输出和退出码。e2b SDK 在退出码非零时抛出 `CommandExitException`，而不是返回结果，因此 `E2BSandbox.execute_command` 返回的是 `Error: Command exited with code N and error: ...`：stdout 被丢弃，`Exit Code: N` @@ -7516,3 +7522,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6426]: https://github.com/bytedance/deer-flow/pull/6426 [#6441]: https://github.com/bytedance/deer-flow/pull/6441 [#6448]: https://github.com/bytedance/deer-flow/pull/6448 +[#6450]: https://github.com/bytedance/deer-flow/pull/6450 diff --git a/backend/docs/AUTH_DESIGN.md b/backend/docs/AUTH_DESIGN.md index 1520740a39..b8f026a2f3 100644 --- a/backend/docs/AUTH_DESIGN.md +++ b/backend/docs/AUTH_DESIGN.md @@ -420,6 +420,10 @@ PYTHONPATH=. python scripts/migrate_user_isolation.py --user-id 迁移脚本覆盖 legacy `memory.json`、`threads/` 和 `agents/` 到 per-user layout。 +thread 归属读取 Gateway 所配置数据库（`config.yaml` 的 `database`）中 `threads_meta.user_id`，未记录归属的 thread 归入 `default`。若存在 legacy thread 但无法读取该表（数据库文件不存在、`memory` 后端或查询失败），脚本在移动任何数据前退出；只有从未记录过 thread 归属的安装才应传 `--allow-missing-thread-owners`，此时所有 legacy thread 归入 `default`。 + +修复前的脚本找不到数据库，会把所有 legacy thread 移到 `users/default/threads/`，且因 `threads/` 已不存在，重跑不会生效。恢复方法：停止 Gateway，用 `SELECT thread_id, user_id FROM threads_meta WHERE user_id IS NOT NULL AND user_id <> 'default'` 列出有真实归属的 thread，把存在的 `users/default/threads/{thread_id}` 移回 `threads/{thread_id}`，再重跑脚本（先 `--dry-run`）。若归属用户之后又使用过该 thread，目标目录已存在，脚本不会覆盖，而是把旧副本放到 `migration-conflicts/{thread_id}`，需手动合并。 + ## 安全不变量 必须长期保持的不变量： diff --git a/backend/packages/harness/deerflow/agents/memory/AGENTS.md b/backend/packages/harness/deerflow/agents/memory/AGENTS.md index c3dc99abd1..fc9c0a21ae 100644 --- a/backend/packages/harness/deerflow/agents/memory/AGENTS.md +++ b/backend/packages/harness/deerflow/agents/memory/AGENTS.md @@ -192,8 +192,8 @@ The older isolation migration remains available: PYTHONPATH=. python scripts/migrate_user_isolation.py --dry-run ``` -It assigns legacy `memory.json`, `threads/`, `agents/`, `skills/`, and the global -`USER.md` to `--user-id` (default `default`). +It assigns legacy `memory.json`, `agents/`, `skills/`, and the global `USER.md` +to `--user-id` (default `default`); `threads/` go to their `threads_meta` owner. #### Retrieval diff --git a/backend/scripts/migrate_user_isolation.py b/backend/scripts/migrate_user_isolation.py index 493836f1c6..3bc69717d9 100644 --- a/backend/scripts/migrate_user_isolation.py +++ b/backend/scripts/migrate_user_isolation.py @@ -6,13 +6,22 @@ Usage (from the repository root): Use the installed backend environment and the Gateway's exported runtime selectors; relative paths remain anchored to the caller's directory. +Thread owners come from the ``threads_meta`` table of the Gateway's configured +database (``database`` in config.yaml). If legacy threads exist but that +table cannot be read, the script stops before moving anything; pass +``--allow-missing-thread-owners`` only for installs that never recorded owners. + The script is idempotent — re-running it after a successful migration is a no-op. """ import argparse import logging import shutil +import sys +from pathlib import Path +from deerflow.config.app_config import get_app_config +from deerflow.config.database_config import DatabaseConfig from deerflow.config.paths import Paths, get_paths logger = logging.getLogger(__name__) @@ -285,30 +294,51 @@ def migrate_user_profile( shutil.move(str(legacy_profile), str(dest)) -def _build_owner_map_from_db(paths: Paths) -> dict[str, str]: - """Query threads_meta table for thread_id -> user_id mapping. +class OwnerMapUnavailable(RuntimeError): + """The configured database cannot say who owns the legacy threads.""" - Uses raw sqlite3 to avoid async dependencies. + +def _has_legacy_thread_dirs(paths: Paths) -> bool: + legacy_threads = paths.base_dir / "threads" + return legacy_threads.is_dir() and any(entry.is_dir() for entry in legacy_threads.iterdir()) + + +def _build_owner_map_from_db(database: DatabaseConfig) -> dict[str, str]: + """Query the Gateway database's threads_meta table for thread_id -> user_id. + + Uses a synchronous engine on the same database the Gateway uses + (``app_sync_sqlalchemy_url``) to avoid async dependencies. + + Raises: + OwnerMapUnavailable: no database to read, or threads_meta cannot be queried. """ - import sqlite3 + if database.backend not in ("sqlite", "postgres"): + raise OwnerMapUnavailable(f"database.backend is {database.backend!r}, which does not persist thread owners") + # Connecting would create an empty SQLite file, hiding a wrong path. + if database.backend == "sqlite" and not Path(database.sqlite_path).is_file(): + raise OwnerMapUnavailable(f"no SQLite database at {database.sqlite_path}") - db_path = paths.base_dir / "deer-flow.db" - if not db_path.exists(): - logger.info("No database found at %s — using empty owner map.", db_path) - return {} + from sqlalchemy import create_engine, text + from sqlalchemy.exc import SQLAlchemyError - conn = sqlite3.connect(str(db_path)) try: - cursor = conn.execute("SELECT thread_id, user_id FROM threads_meta WHERE user_id IS NOT NULL") - return {row[0]: row[1] for row in cursor.fetchall()} - except sqlite3.OperationalError as e: - logger.warning("Failed to query threads_meta: %s", e) - return {} + engine = create_engine(database.app_sync_sqlalchemy_url) + except ImportError as e: + raise OwnerMapUnavailable(f"database driver is not installed ({e})") from e + except SQLAlchemyError as e: + # e.g. an empty postgres_url when $DATABASE_URL is unset. + raise OwnerMapUnavailable(f"cannot open the configured database ({e.__class__.__name__}: {e})") from e + try: + with engine.connect() as conn: + rows = conn.execute(text("SELECT thread_id, user_id FROM threads_meta WHERE user_id IS NOT NULL")).all() + except SQLAlchemyError as e: + raise OwnerMapUnavailable(f"failed to query threads_meta ({e.__class__.__name__}: {getattr(e, 'orig', None) or e})") from e finally: - conn.close() + engine.dispose() + return {thread_id: user_id for thread_id, user_id in rows} -def main() -> None: +def main() -> int: parser = argparse.ArgumentParser(description="Migrate DeerFlow data to per-user layout") parser.add_argument("--dry-run", action="store_true", help="Log actions without making changes") parser.add_argument( @@ -319,6 +349,14 @@ def main() -> None: "User ID to claim un-owned legacy data (global memory.json, USER.md profile, and legacy custom agents). Defaults to 'default'. In multi-user installs, set this to the operator account that should inherit those legacy artifacts." ), ) + parser.add_argument( + "--allow-missing-thread-owners", + action="store_true", + help=( + "Proceed when the configured database cannot provide thread owners (no database file, a 'memory' backend, or an unreadable threads_meta table); " + "every legacy thread is then assigned to 'default'. Only for installs that never recorded thread owners." + ), + ) args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") @@ -328,8 +366,19 @@ def main() -> None: logger.info("Dry run: %s", args.dry_run) logger.info("Claiming un-owned legacy data for user_id=%s", args.user_id) - owner_map = _build_owner_map_from_db(paths) - logger.info("Found %d thread ownership records in DB", len(owner_map)) + owner_map: dict[str, str] = {} + if _has_legacy_thread_dirs(paths): + try: + owner_map = _build_owner_map_from_db(get_app_config().database) + except OwnerMapUnavailable as e: + if not args.allow_missing_thread_owners: + logger.error( + "Cannot read thread owners: %s. Run from the Gateway's working directory with its runtime selectors and config, or pass --allow-missing-thread-owners if this install never recorded thread owners. Nothing was migrated.", + e, + ) + return 1 + logger.warning("Cannot read thread owners: %s. Assigning every legacy thread to 'default' (--allow-missing-thread-owners).", e) + logger.info("Found %d thread ownership records in DB", len(owner_map)) report = migrate_thread_dirs(paths, owner_map, dry_run=args.dry_run) migrate_memory(paths, user_id=args.user_id, dry_run=args.dry_run) @@ -378,6 +427,8 @@ def main() -> None: args.user_id, ) + return 0 + if __name__ == "__main__": - main() + sys.exit(main()) diff --git a/backend/tests/test_migration_user_isolation.py b/backend/tests/test_migration_user_isolation.py index 33a3710a52..0bd8fe880a 100644 --- a/backend/tests/test_migration_user_isolation.py +++ b/backend/tests/test_migration_user_isolation.py @@ -1,11 +1,21 @@ """Tests for per-user data migration.""" import json +import logging +import os +import sqlite3 +import sys +import uuid from pathlib import Path +from types import SimpleNamespace import pytest +from sqlalchemy import create_engine +from deerflow.config.database_config import DatabaseConfig from deerflow.config.paths import Paths +from deerflow.persistence.base import Base +from deerflow.persistence.thread_meta.model import ThreadMetaRow @pytest.fixture @@ -341,3 +351,171 @@ class TestMigrateUserProfile: assert not (base_dir / "USER.legacy.md").exists() assert dest.read_text(encoding="utf-8") == "# current\n" assert legacy.exists() + + +def _write_thread_owners(database: DatabaseConfig, owners: dict[str, str | None]) -> None: + if database.backend == "sqlite": + Path(database.sqlite_dir).mkdir(parents=True, exist_ok=True) + engine = create_engine(database.app_sync_sqlalchemy_url) + try: + Base.metadata.create_all(engine, tables=[ThreadMetaRow.__table__]) + with engine.begin() as conn: + conn.execute(ThreadMetaRow.__table__.insert(), [{"thread_id": thread_id, "user_id": user_id} for thread_id, user_id in owners.items()]) + finally: + engine.dispose() + + +def _run_migration(monkeypatch, base_dir: Path, database: DatabaseConfig, *argv: str) -> int: + import scripts.migrate_user_isolation as migration + + monkeypatch.setattr(migration, "get_paths", lambda: Paths(base_dir)) + monkeypatch.setattr(migration, "get_app_config", lambda: SimpleNamespace(database=database)) + monkeypatch.setattr(sys, "argv", ["migrate_user_isolation", *argv]) + return migration.main() + + +def _make_legacy_threads(base_dir: Path, *thread_ids: str) -> None: + for thread_id in thread_ids: + (base_dir / "threads" / thread_id / "user-data").mkdir(parents=True) + + +class TestThreadOwnersFromDatabase: + """``main()`` reads owners from the Gateway's configured database, or stops.""" + + def test_threads_follow_owners_in_configured_sqlite_database(self, base_dir: Path, monkeypatch): + # sqlite_dir is deliberately outside base_dir: the owner map must come + # from the configured database, not from a path guessed under base_dir. + database = DatabaseConfig(backend="sqlite", sqlite_dir=str(base_dir.parent / "db")) + _write_thread_owners(database, {"t-alice": "alice", "t-bob": "bob", "t-anon": None}) + _make_legacy_threads(base_dir, "t-alice", "t-bob", "t-anon") + + assert _run_migration(monkeypatch, base_dir, database) == 0 + + assert (base_dir / "users" / "alice" / "threads" / "t-alice").is_dir() + assert (base_dir / "users" / "bob" / "threads" / "t-bob").is_dir() + assert (base_dir / "users" / "default" / "threads" / "t-anon").is_dir() + assert not (base_dir / "threads").exists() + + @pytest.mark.parametrize("argv", [(), ("--dry-run",)]) + def test_missing_sqlite_database_stops_before_moving_anything(self, base_dir: Path, monkeypatch, argv): + database = DatabaseConfig(backend="sqlite", sqlite_dir=str(base_dir / "data")) + # The directory exists, so a careless connect would create the file. + Path(database.sqlite_dir).mkdir() + _make_legacy_threads(base_dir, "t-alice") + (base_dir / "memory.json").write_text("{}", encoding="utf-8") + + assert _run_migration(monkeypatch, base_dir, database, *argv) == 1 + + assert (base_dir / "threads" / "t-alice").is_dir() + assert (base_dir / "memory.json").exists() + assert not (base_dir / "users").exists() + # The lookup must not create the database it failed to find. + assert not Path(database.sqlite_path).exists() + + def test_unreadable_threads_meta_stops_before_moving_anything(self, base_dir: Path, monkeypatch): + database = DatabaseConfig(backend="sqlite", sqlite_dir=str(base_dir / "data")) + Path(database.sqlite_dir).mkdir() + sqlite3.connect(database.sqlite_path).close() # a database without threads_meta + _make_legacy_threads(base_dir, "t-alice") + + assert _run_migration(monkeypatch, base_dir, database) == 1 + + assert (base_dir / "threads" / "t-alice").is_dir() + assert not (base_dir / "users").exists() + + def test_memory_backend_stops_before_moving_anything(self, base_dir: Path, monkeypatch): + _make_legacy_threads(base_dir, "t-alice") + + assert _run_migration(monkeypatch, base_dir, DatabaseConfig(backend="memory")) == 1 + + assert (base_dir / "threads" / "t-alice").is_dir() + assert not (base_dir / "users").exists() + + def test_missing_postgres_driver_stops_before_moving_anything(self, base_dir: Path, monkeypatch): + def _no_driver(*_args, **_kwargs): + raise ModuleNotFoundError("No module named 'psycopg'") + + monkeypatch.setattr("sqlalchemy.create_engine", _no_driver) + database = DatabaseConfig(backend="postgres", postgres_url="postgresql://user@localhost/deerflow") + _make_legacy_threads(base_dir, "t-alice") + + assert _run_migration(monkeypatch, base_dir, database) == 1 + + assert (base_dir / "threads" / "t-alice").is_dir() + + def test_unusable_database_url_stops_before_moving_anything(self, base_dir: Path, monkeypatch, caplog): + # An unset $DATABASE_URL leaves postgres_url empty; create_engine raises + # ArgumentError, which must get the same handling as the other failures. + database = DatabaseConfig(backend="postgres", postgres_url="") + _make_legacy_threads(base_dir, "t-alice") + + with caplog.at_level(logging.ERROR, logger="scripts.migrate_user_isolation"): + assert _run_migration(monkeypatch, base_dir, database) == 1 + + assert (base_dir / "threads" / "t-alice").is_dir() + assert "Nothing was migrated." in caplog.text + + def test_rerun_recovers_threads_moved_back_from_default(self, base_dir: Path, monkeypatch): + # Documented recovery for installs the pre-fix script sent to default: + # move the owned threads back to threads/ and run the script again. + database = DatabaseConfig(backend="sqlite", sqlite_dir=str(base_dir / "data")) + _write_thread_owners(database, {"t-alice": "alice", "t-bob": "bob"}) + _make_legacy_threads(base_dir, "t-alice", "t-bob") + for thread_id in ("t-alice", "t-bob"): + (base_dir / "threads" / thread_id / "user-data" / "legacy.txt").write_text(thread_id, encoding="utf-8") + # bob kept using the thread after the bad migration. + newer = base_dir / "users" / "bob" / "threads" / "t-bob" / "user-data" + newer.mkdir(parents=True) + (newer / "newer.txt").write_text("newer", encoding="utf-8") + + assert _run_migration(monkeypatch, base_dir, database) == 0 + + assert (base_dir / "users" / "alice" / "threads" / "t-alice" / "user-data" / "legacy.txt").read_text(encoding="utf-8") == "t-alice" + # An existing destination is never overwritten; the old copy is set aside. + assert (newer / "newer.txt").read_text(encoding="utf-8") == "newer" + assert not (newer / "legacy.txt").exists() + assert (base_dir / "migration-conflicts" / "t-bob" / "user-data" / "legacy.txt").read_text(encoding="utf-8") == "t-bob" + + def test_allow_missing_thread_owners_assigns_default(self, base_dir: Path, monkeypatch): + database = DatabaseConfig(backend="sqlite", sqlite_dir=str(base_dir / "data")) + _make_legacy_threads(base_dir, "t-legacy") + + assert _run_migration(monkeypatch, base_dir, database, "--allow-missing-thread-owners") == 0 + + assert (base_dir / "users" / "default" / "threads" / "t-legacy").is_dir() + + def test_owner_lookup_is_skipped_without_legacy_threads(self, base_dir: Path, monkeypatch): + import scripts.migrate_user_isolation as migration + + def _unexpected_config(): + raise AssertionError("no legacy threads, so the database must not be consulted") + + (base_dir / "memory.json").write_text("{}", encoding="utf-8") + monkeypatch.setattr(migration, "get_paths", lambda: Paths(base_dir)) + monkeypatch.setattr(migration, "get_app_config", _unexpected_config) + monkeypatch.setattr(sys, "argv", ["migrate_user_isolation"]) + + assert migration.main() == 0 + + assert (base_dir / "users" / "default" / "memory.json").exists() + + def test_threads_follow_owners_in_configured_postgres_schema(self, base_dir: Path, monkeypatch): + url = os.getenv("DEERFLOW_TEST_POSTGRES_URL") or os.getenv("TEST_POSTGRES_URI") + if not url: + pytest.skip("set TEST_POSTGRES_URI or DEERFLOW_TEST_POSTGRES_URL to run live PostgreSQL tests") + psycopg = pytest.importorskip("psycopg") + url = url.replace("postgresql+asyncpg://", "postgresql://") + schema = f"migrate_owners_{uuid.uuid4().hex[:12]}" + with psycopg.connect(url, autocommit=True) as admin: + admin.execute(f'CREATE SCHEMA "{schema}"') + try: + database = DatabaseConfig(backend="postgres", postgres_url=url, postgres_schema=schema) + _write_thread_owners(database, {"t-alice": "alice"}) + _make_legacy_threads(base_dir, "t-alice", "t-anon") + + assert _run_migration(monkeypatch, base_dir, database) == 0 + + assert (base_dir / "users" / "alice" / "threads" / "t-alice").is_dir() + assert (base_dir / "users" / "default" / "threads" / "t-anon").is_dir() + finally: + admin.execute(f'DROP SCHEMA "{schema}" CASCADE') diff --git a/docker/provisioner/README.md b/docker/provisioner/README.md index 98cd9d92d8..1b964053d3 100644 --- a/docker/provisioner/README.md +++ b/docker/provisioner/README.md @@ -218,7 +218,15 @@ PYTHONPATH=. python scripts/migrate_user_isolation.py --dry-run PYTHONPATH=. python scripts/migrate_user_isolation.py --user-id ``` -This moves legacy `threads/{thread_id}/user-data` data under `users//threads/{thread_id}/user-data`, which matches the new provisioner PVC subPath when the gateway base directory is mounted at `deer-flow/` on the PVC. Use `default` as the target user only when the legacy data should remain in the default no-auth user namespace. Run the migration while no gateway or sandbox Pods are writing to those paths. +This moves legacy `threads/{thread_id}/user-data` data under `users//threads/{thread_id}/user-data`, which matches the new provisioner PVC subPath when the gateway base directory is mounted at `deer-flow/` on the PVC. `` is the thread's `user_id` in the `threads_meta` table of the gateway's configured database (`default` when none is recorded), so run the script with the gateway's `config.yaml` and database settings; if that table cannot be read, it stops without moving anything. Pass `--allow-missing-thread-owners` only when the install never recorded thread owners. `--user-id` picks who inherits un-owned global data (memory, `USER.md`, legacy custom agents). Run the migration while no gateway or sandbox Pods are writing to those paths. + +Before this fix, the script never found the database and moved every legacy thread to `users/default/threads/`. Re-running it does nothing then, because `threads/` is gone. To recover, stop the gateway and list the threads with a real owner: + +```sql +SELECT thread_id, user_id FROM threads_meta WHERE user_id IS NOT NULL AND user_id <> 'default'; +``` + +Move each listed `users/default/threads/{thread_id}` that exists back to `threads/{thread_id}`, then run the script again (dry run first). Threads whose owner is `default` or not recorded stay where they are. If an owner has used the thread since, `users//threads/{thread_id}` already exists; the script never overwrites it and sets the old copy aside in `migration-conflicts/{thread_id}` for you to merge by hand. In hostPath mode, the gateway materializes enabled-only views under `skills_view/public` and `users/{user_id}/skills_view/{custom,legacy}` beneath `DEER_FLOW_HOST_BASE_DIR`; the provisioner mounts those stable directories. A lead Agent with an explicit skills policy supplies all four category mounts from `users/{user_id}/threads/{thread_id}/skills_view`; those overrides replace the default hostPath or root skills-PVC mount. When `USERDATA_PVC_NAME` is configured, the provisioner mounts these thread categories from that PVC using `deer-flow/users/{user_id}/threads/{thread_id}/skills_view/{category}` subpaths. For unrestricted threads, operators can still set `SKILLS_PVC_NAME` and optionally configure `SKILLS_PVC_SUBPATH_TEMPLATE`; leaving the template empty mounts the skills PVC root unchanged. The gateway does not populate the unrestricted `SKILLS_PVC_SUBPATH_TEMPLATE` layout dynamically.