From 2dc68b242fd253b32a1106a2b4c213f4b7041e3b Mon Sep 17 00:00:00 2001 From: paer <162977848+pamir-yuan@users.noreply.github.com> Date: Thu, 12 Nov 2026 14:39:25 +0800 Subject: [PATCH] fix(subagents): prevent concurrent file-backed creates from overwriting definitions (#8132) * docs(subagents): describe create-only file publication * fix(subagents): prevent concurrent create overwrites * test(subagents): cover independent create races * docs(persistence): record create-only publication * docs(config): explain managed subagent file claims * docs(persistence): retain create-only publication rule * docs(subagents): retain shared file create semantics * fix(gateway): probe managed subagent hard links on startup * test(gateway): cover managed subagent volume probes * test(gateway): keep volume probes off the event loop * docs(config): describe startup hard-link warnings * docs(changelog): reference managed subagent fix #8132 * docs(changelog): keep the fix entry clear of append conflicts --- CHANGELOG.md | 6 ++ backend/app/gateway/AGENTS.md | 3 + backend/app/gateway/deps.py | 24 +++++ backend/docs/CONFIGURATION.md | 10 ++ .../harness/deerflow/persistence/AGENTS.md | 7 ++ .../persistence/managed_subagents/file.py | 16 +++- .../test_managed_subagent_file_probe.py | 59 ++++++++++++ .../test_managed_subagent_create_race.py | 91 +++++++++++++++++++ .../tests/test_managed_subagent_file_probe.py | 86 ++++++++++++++++++ docs/guide/subagents.md | 2 + 10 files changed, 300 insertions(+), 4 deletions(-) create mode 100644 backend/tests/blocking_io/test_managed_subagent_file_probe.py create mode 100644 backend/tests/test_managed_subagent_create_race.py create mode 100644 backend/tests/test_managed_subagent_file_probe.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 7129601db3..d61f606daf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -91,6 +91,12 @@ This section accumulates work toward the **2.3.0** milestone the run before executing it. Custom `RunStore` implementations used with heartbeats must implement that method. ([#8046]) +- **subagents:** Concurrent creates for the same managed-subagent name in shared + file storage now preserve the winning definition and return HTTP 409 to the + losing request instead of reporting two successes and overwriting one definition. + Gateway startup also probes hard-link support and warns if the storage volume + cannot support managed subagent creation. ([#8132](https://github.com/bytedance/deer-flow/pull/8132)) + - **agents:** A high-risk tool call without a usable id no longer wedges the thread when `hitl_approval` is enabled. A provider that omits a tool-call id (`None`) or repeats one within a response made the middleware pause with an diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index abe1420239..c857239039 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -77,6 +77,9 @@ replay, archive), [run event stream](../../docs/RUN_EVENT_STREAM.md), ## Lifecycle +- **Probe file-backed managed subagent storage once during startup, off the + event loop.** Probe failures warn operators before their first create request; + atomic create stays fail-closed rather than falling back to an overwrite. - **Shutdown order matters.** `lifespan` drains runs and extensions first, then closes MCP, then shuts the sandbox provider down off the event loop. Cleanup code keeps running after cancellation, allocates nothing new, and logs failures diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index e80aaebfba..c7e420dbaa 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -20,6 +20,7 @@ from __future__ import annotations import asyncio import logging import os +import tempfile from collections.abc import AsyncGenerator, Callable from contextlib import AsyncExitStack, asynccontextmanager from pathlib import Path @@ -384,6 +385,27 @@ def _validate_agent_storage(config: AppConfig) -> None: ) +def _probe_managed_subagent_file_storage() -> None: + """Warn early if this volume cannot support create-only publication.""" + from deerflow.config.paths import get_paths + + root = get_paths().managed_subagents_dir + try: + root.mkdir(parents=True, exist_ok=True) + # A private directory keeps probe names out of the JSON catalog and + # lets independent Gateway workers probe without touching each other. + with tempfile.TemporaryDirectory(prefix=".hardlink-probe-", dir=root) as scratch: + source = Path(scratch) / "source" + source.touch() + os.link(source, Path(scratch) / "linked") + except OSError as exc: + logger.warning( + "Managed subagent hard-link probe failed for %s: %s. Managed subagent creation may fail on this volume. Use a volume with working hard links or set agent_storage.backend='db' with SQLite/Postgres.", + root, + exc, + ) + + def _validate_login_throttle_storage(config: AppConfig) -> None: """Warn when a multi-process deployment counts login failures per process. @@ -788,6 +810,8 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen # Reject agent_storage.backend='db' on a non-durable database, and warn on # node-divergent file storage under multi-worker Postgres. _validate_agent_storage(startup_config) + if getattr(getattr(startup_config, "agent_storage", None), "backend", None) == "file": + await asyncio.to_thread(_probe_managed_subagent_file_storage) # Warn when login lockouts stay per-process under several Gateway processes. _validate_login_throttle_storage(startup_config) # IM chat-to-thread bindings need no companion warning: ``resolve_channel_store`` diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 93454716de..2aa1151268 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -933,6 +933,16 @@ agent_storage: - `agent_storage` is restart-required (the backend is captured at Gateway lifespan startup). - In a multi-worker Postgres deployment (`GATEWAY_WORKERS > 1`), leaving `agent_storage.backend: file` logs a startup warning — agents written to one node's local disk are invisible to the others, which is exactly the divergence the `db` backend fixes. +Managed subagents also follow this backend selection. With `file`, their directory +must support hard links: creation atomically publishes a fully written temporary +file without replacing an existing name. Workers sharing that directory therefore +return one HTTP 201 and one HTTP 409 for concurrent same-name creates. Other +filesystem errors propagate rather than falling back to an overwrite. +Gateway startup probes hard-link creation once per worker in that directory, +off the event loop, and logs a warning if the probe fails. Startup continues; +use a volume with working hard links or migrate to `backend: db` before creating +managed subagents on an incompatible volume. Probe files are removed afterward. + Migrating an existing install from `file` to `db`: Run from the repository root using the backend's `uv` environment. The importer diff --git a/backend/packages/harness/deerflow/persistence/AGENTS.md b/backend/packages/harness/deerflow/persistence/AGENTS.md index 8a4110dff8..8c281eedb6 100644 --- a/backend/packages/harness/deerflow/persistence/AGENTS.md +++ b/backend/packages/harness/deerflow/persistence/AGENTS.md @@ -52,6 +52,13 @@ change to a query, a lock or a JSON predicate is verified on both dialects. the unlock completes.** Drain both through cancellation. Otherwise a second bootstrap overlaps a running migration, or a pooled session keeps the lock. +## Managed subagent file creation + +Stage and fsync complete definitions before publishing them. The RLock is local +to one process, so create claims the destination with `os.link` and maps only +`FileExistsError` to a name conflict. Updates retain `os.replace`. Discard staging +with best-effort cleanup, preserving the original publication outcome. + ## Leases, claims and terminal writes - **A scan selects candidates; it grants nothing.** After you take the row locks, diff --git a/backend/packages/harness/deerflow/persistence/managed_subagents/file.py b/backend/packages/harness/deerflow/persistence/managed_subagents/file.py index ebe6cfe550..e165db5270 100644 --- a/backend/packages/harness/deerflow/persistence/managed_subagents/file.py +++ b/backend/packages/harness/deerflow/persistence/managed_subagents/file.py @@ -55,7 +55,10 @@ class FileManagedSubagentStore(ManagedSubagentStore): if path.exists(): raise ManagedSubagentExistsError(f"Managed subagent '{definition.name}' already exists") path.parent.mkdir(parents=True, exist_ok=True) - self._atomic_write(path, definition) + try: + self._atomic_write(path, definition, overwrite=False) + except FileExistsError as exc: + raise ManagedSubagentExistsError(f"Managed subagent '{definition.name}' already exists") from exc def update(self, definition: ManagedSubagentDefinition) -> None: path = get_paths().managed_subagent_file(definition.name) @@ -86,7 +89,7 @@ class FileManagedSubagentStore(ManagedSubagentStore): return tuple(signature) @staticmethod - def _atomic_write(path: Path, definition: ManagedSubagentDefinition) -> None: + def _atomic_write(path: Path, definition: ManagedSubagentDefinition, *, overwrite: bool = True) -> None: payload = definition.model_dump_json(indent=2) + "\n" tmp_path: Path | None = None try: @@ -97,8 +100,13 @@ class FileManagedSubagentStore(ManagedSubagentStore): handle.write(payload) handle.flush() os.fsync(handle.fileno()) - os.replace(tmp_path, path) - tmp_path = None + if overwrite: + os.replace(tmp_path, path) + tmp_path = None + else: + # Claim the name without replacing a peer's complete definition. + # The process-local RLock cannot serialize independent workers. + os.link(tmp_path, path) finally: if tmp_path is not None: cleanup_failed_atomic_write(tmp_path) diff --git a/backend/tests/blocking_io/test_managed_subagent_file_probe.py b/backend/tests/blocking_io/test_managed_subagent_file_probe.py new file mode 100644 index 0000000000..99586ce9bb --- /dev/null +++ b/backend/tests/blocking_io/test_managed_subagent_file_probe.py @@ -0,0 +1,59 @@ +"""The startup volume probe runs once, off the Gateway event loop.""" + +from __future__ import annotations + +import threading + +import pytest +from fastapi import FastAPI + +from app.gateway import deps +from deerflow.config.app_config import AppConfig +from deerflow.config.paths import Paths + + +class _ReachedNextStartupCheck(Exception): + pass + + +@pytest.fixture +def volume_paths(tmp_path, monkeypatch): + paths = Paths(base_dir=tmp_path) + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: paths) + return paths + + +@pytest.mark.asyncio +@pytest.mark.parametrize("backend", [None, "file", "db"], ids=["default-file", "explicit-file", "database"]) +async def test_lifespan_probes_only_file_storage_off_loop(volume_paths, monkeypatch, backend): + for name in ("GATEWAY_WORKERS", "WEB_CONCURRENCY", "DEER_FLOW_MULTI_INSTANCE"): + monkeypatch.delenv(name, raising=False) + paths = volume_paths + config = AppConfig(sandbox={"use": "deerflow.sandbox.local:LocalSandboxProvider"}, database={"backend": "sqlite"}) + if backend is not None: + config.agent_storage.backend = backend + loop_thread = threading.get_ident() + probe_threads = [] + original_link = deps.os.link + + def observed_link(source, destination): + probe_threads.append(threading.get_ident()) + original_link(source, destination) + + def reached_next_check(_config): + raise _ReachedNextStartupCheck + + monkeypatch.setattr(deps.os, "link", observed_link) + # Stop after the real startup probe, before opening any runtime resources. + monkeypatch.setattr(deps, "_validate_login_throttle_storage", reached_next_check) + with pytest.raises(_ReachedNextStartupCheck): + async with deps.langgraph_runtime(FastAPI(), config): + pytest.fail("the following startup check must abort this focused test") + + if backend == "db": + assert probe_threads == [] + assert not paths.managed_subagents_dir.exists() + else: + assert len(probe_threads) == 1 + assert probe_threads[0] != loop_thread + assert list(paths.managed_subagents_dir.iterdir()) == [] diff --git a/backend/tests/test_managed_subagent_create_race.py b/backend/tests/test_managed_subagent_create_race.py new file mode 100644 index 0000000000..25449f9c22 --- /dev/null +++ b/backend/tests/test_managed_subagent_create_race.py @@ -0,0 +1,91 @@ +"""Create-only publication across independent file-store processes.""" + +from __future__ import annotations + +import errno +import multiprocessing + +import pytest + +from deerflow.config.paths import Paths +from deerflow.persistence.managed_subagents import ManagedSubagentDefinition, ManagedSubagentExistsError +from deerflow.persistence.managed_subagents import file as file_module + + +def _definition(name, writer): + return ManagedSubagentDefinition(name=name, description=f"Writer {writer}", system_prompt=f"Instructions from writer {writer}.") + + +def _create_in_process(home, name, writer, staged, results): + # Spawn gives each writer its own module-level RLock and path registry. + paths = Paths(base_dir=home) + definition = _definition(name, writer) + original_fsync = file_module.os.fsync + + def fsync_then_wait(fd): + original_fsync(fd) + # Both complete files are staged after both writers passed exists(). + # Synchronize the real I/O boundary, without fabricating a collision. + staged.wait(timeout=30) + + with pytest.MonkeyPatch.context() as patch: + patch.setattr(file_module, "get_paths", lambda: paths) + patch.setattr(file_module.os, "fsync", fsync_then_wait) + try: + file_module.FileManagedSubagentStore().create(definition) + except ManagedSubagentExistsError: + outcome = "conflict" + else: + outcome = "created" + results.put((writer, outcome)) + + +@pytest.mark.parametrize("same_name", [True, False], ids=["same-name", "different-names"]) +def test_independent_process_creates_never_overwrite_a_winner(tmp_path, monkeypatch, same_name): + paths = Paths(base_dir=tmp_path) + monkeypatch.setattr(file_module, "get_paths", lambda: paths) + names = ["researcher", "researcher" if same_name else "reviewer"] + context = multiprocessing.get_context("spawn") + staged = context.Barrier(2) + results = context.Queue() + workers = [context.Process(target=_create_in_process, args=(tmp_path, name, writer, staged, results)) for writer, name in enumerate(names)] + try: + for worker in workers: + worker.start() + outcomes = [results.get(timeout=45) for _ in workers] + for worker in workers: + worker.join(10) + assert worker.exitcode == 0 + + assert sorted(outcome for _, outcome in outcomes) == (["conflict", "created"] if same_name else ["created", "created"]) + store = file_module.FileManagedSubagentStore() + for writer, outcome in outcomes: + if outcome == "created": + assert store.get(names[writer]) == _definition(names[writer], writer) + assert len(store.list()) == (1 if same_name else 2) + assert sorted(path.name for path in paths.managed_subagents_dir.iterdir()) == sorted({f"{name}.json" for name in names}) + finally: + staged.abort() + for worker in workers: + if worker.pid is not None: + worker.join(5) + if worker.is_alive(): + worker.terminate() + worker.join(5) + worker.close() + results.close() + results.join_thread() + + +def test_create_propagates_publication_errors_and_discards_staging(tmp_path, monkeypatch): + paths = Paths(base_dir=tmp_path) + monkeypatch.setattr(file_module, "get_paths", lambda: paths) + + def denied_link(*args, **kwargs): + raise PermissionError(errno.EPERM, "hard links denied") + + monkeypatch.setattr(file_module.os, "link", denied_link) + with pytest.raises(PermissionError, match="hard links denied"): + file_module.FileManagedSubagentStore().create(_definition("researcher", 0)) + + assert list(paths.managed_subagents_dir.iterdir()) == [] diff --git a/backend/tests/test_managed_subagent_file_probe.py b/backend/tests/test_managed_subagent_file_probe.py new file mode 100644 index 0000000000..b0dec65a72 --- /dev/null +++ b/backend/tests/test_managed_subagent_file_probe.py @@ -0,0 +1,86 @@ +"""Startup diagnostics for the volume used by managed subagent definitions.""" + +from __future__ import annotations + +import errno +import logging +from pathlib import Path + +import pytest + +from app.gateway import deps +from deerflow.config.paths import Paths + + +@pytest.fixture +def paths(tmp_path, monkeypatch): + paths = Paths(base_dir=tmp_path) + monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: paths) + return paths + + +def test_probe_uses_the_managed_volume_and_leaves_existing_definitions_untouched(paths, monkeypatch, caplog): + root = paths.managed_subagents_dir + root.mkdir() + existing = root / "researcher.json" + existing.write_bytes(b'{"name":"researcher"}\n') + original_link = deps.os.link + claims = [] + + def observed_link(source, destination): + source, destination = Path(source), Path(destination) + assert source.parent.parent == root + assert destination.parent == source.parent + original_link(source, destination) + assert source.stat().st_ino == destination.stat().st_ino + claims.append((source, destination)) + + monkeypatch.setattr(deps.os, "link", observed_link) + with caplog.at_level(logging.WARNING): + deps._probe_managed_subagent_file_storage() + + assert len(claims) == 1 + assert list(root.iterdir()) == [existing] + assert existing.read_bytes() == b'{"name":"researcher"}\n' + assert not caplog.records + + +@pytest.mark.parametrize("error", [OSError(errno.EOPNOTSUPP, "hard links unavailable"), PermissionError(errno.EPERM, "hard links denied")]) +def test_probe_warns_on_link_failure_and_cleans_up(paths, monkeypatch, caplog, error): + attempted = [] + + def failed_link(source, destination): + attempted.append((source, destination)) + raise error + + monkeypatch.setattr(deps.os, "link", failed_link) + with caplog.at_level(logging.WARNING): + deps._probe_managed_subagent_file_storage() + + assert len(attempted) == 1 + assert list(paths.managed_subagents_dir.iterdir()) == [] + assert "hard-link probe failed" in caplog.text + assert str(paths.managed_subagents_dir) in caplog.text + assert "creation may fail" in caplog.text + assert "agent_storage.backend='db'" in caplog.text + + +def test_probe_warns_if_the_directory_cannot_be_created(paths, monkeypatch, caplog): + root = paths.managed_subagents_dir + original_mkdir = Path.mkdir + attempted = [] + + def denied_mkdir(path, *args, **kwargs): + if path == root: + attempted.append(path) + raise PermissionError(errno.EACCES, "volume is read-only") + return original_mkdir(path, *args, **kwargs) + + monkeypatch.setattr(Path, "mkdir", denied_mkdir) + with caplog.at_level(logging.WARNING): + deps._probe_managed_subagent_file_storage() + + assert attempted == [root] + assert "hard-link probe failed" in caplog.text + assert "volume is read-only" in caplog.text + assert not root.exists() diff --git a/docs/guide/subagents.md b/docs/guide/subagents.md index bb16eefb06..de28feb615 100644 --- a/docs/guide/subagents.md +++ b/docs/guide/subagents.md @@ -258,6 +258,8 @@ The factory still does not load YAML or create SQL infrastructure: the caller su Administrators can add, edit, disable, and delete reusable worker definitions from **Settings → Subagents**. Built-in and `config.yaml` definitions remain visible there as read-only entries. The default Lead Agent can use every enabled runtime sub-agent; each page-created Custom Agent can instead allow all, none, or a selected set. That selection is enforced both in the model-visible directory and by the server-side `task` tool. Managed definitions are deployment-wide in this version and follow `agent_storage.backend`: atomic files for a local deployment or the shared application database for multiple instances. +When Gateway workers share file storage, concurrent managed-subagent creates for the same name preserve the winning definition and return one HTTP 201 and one HTTP 409. File-backed managed-subagent creation requires a filesystem supporting hard links; see [Agent Storage](../../backend/docs/CONFIGURATION.md#agent-storage). + Native workers support per-subagent `thinking_enabled` and `reasoning_effort` through YAML configuration and the managed-subagent API. Unconfigured workers keep the existing thinking-off request; the selected model's required-thinking policy, supported effort values and aliases still apply. These settings do not inherit the lead agent's thinking preferences. For example, independent read-only research can run concurrently when the wall-clock savings outweigh duplicated discovery and synthesis cost, while a repository refactor with shared files and sequential test feedback remains with the lead agent. When `max_concurrent_subagents` is `1`, parallel and multi-batch routing guidance is disabled; delegation remains available only for material specialist or context-isolation benefit.