mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(migration): read thread owners from the configured database (#6450)
* fix(migration): read thread owners from the configured database
migrate_user_isolation.py 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 map was always empty
and every legacy thread was moved to users/default/.
Read threads_meta through the configured database's sync SQLAlchemy URL
(SQLite or PostgreSQL, honoring postgres_schema). When legacy threads
exist but owners cannot be read (no database file, a memory backend, a
missing driver, or a failed query), stop before migrating anything;
--allow-missing-thread-owners keeps the old assign-to-default behavior
for installs that never recorded owners.
* docs(changelog): reference #6450
* fix(migration): document recovery and report unusable database URLs
Installs that already ran the old script have every legacy thread under
users/default/threads/, and re-running finds no threads/ to migrate. The
provisioner README and AUTH_DESIGN now give the recovery: move the owned
threads back to threads/ and re-run; an existing owner directory is never
overwritten and the old copy goes to migration-conflicts/. A test pins
that re-run behavior.
create_engine errors other than ImportError (e.g. ArgumentError from an
empty postgres_url) now stop with the same "Nothing was migrated." message
instead of a traceback.
The memory AGENTS.md note said threads go to --user-id; they go to their
threads_meta owner.
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
340a5b63fa
commit
c15ecb688d
12
CHANGELOG.md
12
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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -420,6 +420,10 @@ PYTHONPATH=. python scripts/migrate_user_isolation.py --user-id <target-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}`,需手动合并。
|
||||
|
||||
## 安全不变量
|
||||
|
||||
必须长期保持的不变量:
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
@ -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())
|
||||
|
||||
@ -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')
|
||||
|
||||
@ -218,7 +218,15 @@ PYTHONPATH=. python scripts/migrate_user_isolation.py --dry-run
|
||||
PYTHONPATH=. python scripts/migrate_user_isolation.py --user-id <target-user-id>
|
||||
```
|
||||
|
||||
This moves legacy `threads/{thread_id}/user-data` data under `users/<target-user-id>/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/<owner>/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. `<owner>` 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/<owner>/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.
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user