diff --git a/backend/docs/blob-storage.md b/backend/docs/blob-storage.md new file mode 100644 index 0000000000..33c66aace2 --- /dev/null +++ b/backend/docs/blob-storage.md @@ -0,0 +1,112 @@ +# Blob storage (content-addressed, cross-instance) + +Resolves the multi-instance half of [#4189](https://github.com/bytedance/deer-flow/issues/4189) item 2: two producers persist blob-shaped data outside the checkpoint payload and address it with a **server-local filesystem path**, which only resolves on the instance that wrote it. + +| Producer | Write | Reads | +|---|---|---| +| Viewed images | `view_image_tool` → `ViewedImageData.actual_path` (`deerflow/agents/thread_state.py:52`) | `ViewImageMiddleware._read_image_as_data_url`, gateway artifact routes, IM channels, `present_file_tool` | +| Externalized tool results | `ToolOutputBudgetMiddleware._externalize` → virtual path under the thread `outputs` tree | model `read_file` via the thread-data mount | + +On a single gateway both are correct. Behind a load balancer, the instance handling the read is frequently not the instance that wrote the file. The blob store replaces **"where on this machine"** with **"which content"**, so every instance that can reach the backing store resolves the same bytes. + +## Contract + +`deerflow/storage/contract.py` + +- `BlobRef` — `{sha256, size, kind, content_type}`. The digest is the address, so writes are idempotent and dedup is free. +- `BlobStore` — plain ABC, tiered like `MemoryStorage`: + - **abstract** — `put_bytes(data, *, kind, content_type=None, thread_id=None) -> BlobRef` (`thread_id` is advisory provenance, see the deletion rule below), `get_bytes(ref) -> bytes` + - **default** — `exists` (probes via `get_bytes`), `delete` (raises; deleting an absent blob is not an error), `close` +- Errors — `BlobStoreError` → `BlobNotConfiguredError` / `BlobWriteError` / `BlobReadError` → `BlobNotFoundError`. A failed `delete` raises the neutral base `BlobStoreError`, not a read error. + +Reads **verify the digest**. A content-addressed store that silently returns wrong bytes is indistinguishable from a corrupt checkpoint, so it fails loudly instead. + +`kind` is validated by `validate_blob_kind` against `^[a-z0-9]([a-z0-9-]*[a-z0-9])?$` (1-64 chars, no leading or trailing hyphen) at the contract level because it is also a path segment in the `local_fs` backend — traversal and separator surprises are ruled out once, not per backend, and the single validator is what a rejected producer sees. + +## Backend + +`deerflow/storage/backends/local_fs/` (default) + +``` +/// the bytes +///.json sidecar: content_type, writer_thread_id, created_at +``` + +- Writes go to a unique temp file in the destination directory then `os.replace` — atomic within a volume, so a concurrent reader never sees a partial file and two instances racing the same blob converge on identical content. +- Single-put size cap: `_MAX_BLOB_BYTES` (64 MiB) in `local_fs_store.py` rejects larger puts with `BlobWriteError`. This is a backend-level defense-in-depth limit, **not** part of the `BlobStore` contract — other backends set their own policy — but the externalized-tool-results follow-up should assume it and stream or split oversized payloads before they reach `put_bytes`. +- The sidecar is GC metadata, **not a read dependency and not a reference count**: losing it must not make content unreadable, and `writer_thread_id` is advisory provenance that cannot stand in for liveness (see the deletion rule above). + +This backend already delivers multi-instance resolution when `root` points at a shared volume (NFS / EFS / a `ReadWriteMany` PVC). The S3/MinIO backend is a later, optional extra implementing the same contract — it is deliberately not part of this change, because it needs a dependency decision (`[tool.uv.sources]`, optional extra) that belongs in its own PR. + +## Configuration + +```yaml +blob_storage: + enabled: true # default false + backend: local_fs # folder name under storage/backends/, or a dotted import path + backend_config: + root: /mnt/shared/deerflow-blobs # default: {runtime_home}/blobs, absolute +``` + +Fail-fast on an unresolvable backend (`ValueError`), mirroring `MemoryConfig.manager_class`: blobs are persistent state, so silently substituting a different backend would strand previously written content. + +The factory checks the effective backend and backend configuration on each access. After a `config.yaml` edit, new accesses use the new store; disabling storage makes the optional accessor return `None` and the required accessor raise `BlobNotConfiguredError`. A store already handed to an in-flight caller remains usable and is closed by `reset_blob_store()` rather than during the switch. When changing a storage root in a multi-instance deployment, coordinate the rollout and keep previously written blobs available until existing references have been migrated. + +## What this PR deliberately does not do + +**No producer is migrated.** `blob_storage.enabled` defaults to `false`, no existing call site changed, and the diff is purely additive — a deployment that never sets the key behaves exactly as before. + +Migration happens in two follow-up PRs, each independently revertible: + +1. **Viewed images.** `ViewedImageData` gains an optional `blob_ref` (`actual_path` is kept); `view_image_tool` writes the blob when the store is enabled; `ViewImageMiddleware._read_image_as_data_url` resolves blob-first, path-second. The gateway artifact routes and IM channels keep their local-path reads until they can be exercised against a multi-instance deployment. +2. **Externalized tool results.** `ToolOutputBudgetMiddleware._externalize` records a ref alongside the virtual path; the sandbox variant (`_externalize_to_sandbox`, issue #3416) stays as-is — sandbox-resident content is a different failure mode one layer down. + +## Interaction with checkpoint retention (#5255) + +Blobs are content-addressed, and `kind` plus the sidecar's `writer_thread_id` +make a sweep *addressable* by kind and thread. That is the whole claim this seam +supports — it is **not** that retention can stop reasoning about references: + +> **Deletion rule.** A blob may be unlinked only after confirming that no +> surviving durable reference names it. `kind` / `writer_thread_id` narrow the +> candidate set; on their own they are never sufficient. + +The narrower rule is the honest one, because identical content collapses to one +object. Two threads that externalize the same bytes share a single file, and the +sidecar can only record whichever writer landed first (`writer_thread_id` is set +by the first write; the idempotent path does not rewrite it). Concretely: thread +A and thread B both view the same uploaded image — same bytes, same sha256, one +file under `viewed-image/`. A sweep for A that trusted the sidecar would delete +bytes B still references, and B's own sweep would never see the blob at all, +because the sidecar names A. + +Recording *every* writer thread id would not repair that. Appending to one +sidecar from two gateway instances is a read-modify-write race with lost +updates, so the list could be missing precisely the writer that still needs the +content — and a reference count that is wrong under concurrency is worse than +none. Hence the contract states the liveness rule instead of a refcount. + +What this means for the two migration PRs: + +- **Viewed images** — an image is shareable by construction (any two threads can + upload the same bytes), so the sweep must check liveness against the + checkpoint references before unlinking; `kind` and `writer_thread_id` bound + the search. +- **Externalized tool results** — per-thread by construction, so a thread-scoped + sweep is safe while that remains true; `kind` is what makes the "while" + checkable. + +This is also the seam [#5188](https://github.com/bytedance/deer-flow/issues/5188) needs: a thread-scoped blob sweep keyed by thread incarnation, rather than by the reusable thread id. + +## Adding a backend + +`packages/harness/deerflow/storage/AGENTS.md` owns the depth — the drop-in +folder contract, the portability rule and the durability rules a new backend has +to keep. In short: + +1. Copy `backends/local_fs/` to `backends//`. +2. Implement `from_config` + `put_bytes` + `get_bytes`; override `delete` if you can support it. +3. Export `STORE_CLASS = ` from `backends//__init__.py`. +4. Set `blob_storage.backend: `; backend knobs go under `blob_storage.backend_config`. + +If the backend needs external libs (boto3, minio), declare them in `packages/harness/pyproject.toml` with `[tool.uv.sources]` — otherwise `uv sync` purges them. diff --git a/backend/packages/harness/deerflow/config/app_config.py b/backend/packages/harness/deerflow/config/app_config.py index 3982bed230..5f1c971f0e 100644 --- a/backend/packages/harness/deerflow/config/app_config.py +++ b/backend/packages/harness/deerflow/config/app_config.py @@ -14,6 +14,7 @@ from deerflow.config.agent_storage_config import AgentStorageConfig from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict from deerflow.config.auth_config import AuthAppConfig from deerflow.config.authorization_config import AuthorizationConfig, load_authorization_config_from_dict +from deerflow.config.blob_storage_config import BlobStorageConfig, load_blob_storage_config_from_dict from deerflow.config.channel_connections_config import ChannelConnectionsConfig from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict from deerflow.config.database_config import DatabaseConfig @@ -247,6 +248,7 @@ class AppConfig(BaseModel): summarization: SummarizationConfig = Field(default_factory=SummarizationConfig, description="Conversation summarization configuration") task_continuity: TaskContinuityConfig = Field(default_factory=TaskContinuityConfig, description="Thread-local notes and compacted-source recall") memory: MemoryConfig = Field(default_factory=MemoryConfig, description="Memory subsystem configuration") + blob_storage: BlobStorageConfig = Field(default_factory=BlobStorageConfig, description="Content-addressed blob store configuration") knowledge_base: KnowledgeBaseConfig = Field( default_factory=KnowledgeBaseConfig, description="Provider-agnostic knowledge capability and custom-agent scope-selection configuration", @@ -484,6 +486,7 @@ class AppConfig(BaseModel): load_title_config_from_dict(config.title.model_dump()) load_summarization_config_from_dict(config.summarization.model_dump()) load_memory_config_from_dict(config.memory.model_dump()) + load_blob_storage_config_from_dict(config.blob_storage.model_dump()) load_agents_api_config_from_dict(config.agents_api.model_dump()) load_subagents_config_from_dict(config.subagents.model_dump()) load_tool_search_config_from_dict(config.tool_search.model_dump()) diff --git a/backend/packages/harness/deerflow/config/blob_storage_config.py b/backend/packages/harness/deerflow/config/blob_storage_config.py new file mode 100644 index 0000000000..e45c0204ed --- /dev/null +++ b/backend/packages/harness/deerflow/config/blob_storage_config.py @@ -0,0 +1,120 @@ +"""Configuration for the blob store (host-shared fields only). + +Backend-private fields live in each backend's own config (parsed from +``backend_config``, a dict the factory passes verbatim to the backend). This +module holds ONLY the host-shared fields every backend / call site / factory +reads: ``enabled`` / ``backend`` / ``backend_config``. Keeping the shared +schema slim is what makes backends swappable and portable (a backend's knobs +do not leak onto the shared contract). Mirrors ``memory_config``: blobs are +persistent state, so the factory fails fast on an unresolvable ``backend`` +instead of silently substituting a different storage backend. +""" + +import logging +from typing import Any + +from pydantic import BaseModel, Field + +logger = logging.getLogger(__name__) + +# Host-shared BlobStorageConfig fields (read by every backend / call site / factory). +_SHARED_FIELDS = frozenset({"enabled", "backend", "backend_config"}) + + +class BlobStorageConfig(BaseModel): + """Host-shared blob store configuration (backend-agnostic).""" + + enabled: bool = Field( + default=False, + description=("Whether to enable the content-addressed blob store. Default False: no producer is migrated yet, so a deployment that never sets this key behaves exactly as it does today (see docs/blob-storage.md)."), + ) + backend: str = Field( + default="local_fs", + description=( + "Blob store backend selector. Either a registered backend name " + "(matching a `storage/backends//` folder that exposes " + "`STORE_CLASS`, e.g. `local_fs`) or a dotted import path to a " + "`BlobStore` subclass. The factory resolves this at " + "`get_blob_store()` time and raises `ValueError` on failure " + "(fail-fast: blobs are persistent state, so an unresolved backend " + "must not be silently substituted with a different one)." + ), + ) + backend_config: dict[str, Any] = Field( + default_factory=dict, + description=( + "Backend-private config (a dict), passed verbatim to the backend by " + "the factory. Each backend self-interprets it (local_fs reads " + "`root`; an empty root defaults to `{runtime_home}/blobs`). Values " + "live in the host config file (`config.yaml` " + "`blob_storage.backend_config`); they do not belong on the shared " + "schema." + ), + ) + + +# Global configuration instance +_blob_storage_config: BlobStorageConfig = BlobStorageConfig() + + +def get_blob_storage_config() -> BlobStorageConfig: + """Get the current blob store configuration. + + ``_blob_storage_config`` is only refreshed as a side effect of + ``get_app_config()`` reloading (via ``_apply_singleton_configs`` -> + ``load_blob_storage_config_from_dict``). A reader that reaches blob config + without going through ``get_app_config()`` first -- e.g. a middleware + deciding whether to externalize a tool result -- would otherwise see a + stale ``blob_storage.enabled`` after a ``config.yaml`` edit, even though + ``blob_storage.*`` is documented as hot-reloadable. Trigger the same + signature-checked reload here so the singleton follows the config file. + + If ``get_app_config()`` has never been called (``_app_config`` is + ``None``), there is no stale config to refresh, so we keep the + pre-existing behaviour of returning the in-memory singleton. This avoids + picking up a config file as a side effect of the first access to + ``get_blob_storage_config()``, which would break callers that expect + module-level defaults (e.g. unit tests). + """ + # Lazy import: app_config imports this module, so a top-level import cycles. + from .app_config import _app_config, get_app_config + + if _app_config is not None: + try: + get_app_config() + except Exception: + # If the config file is transiently broken (invalid YAML, schema + # violation, missing env var, etc.), keep the last-good singleton + # so an in-flight turn completes normally instead of crashing. + logger.warning( + "Failed to reload app config from get_blob_storage_config(); falling back to cached blob storage config.", + exc_info=True, + ) + return _blob_storage_config + + +def set_blob_storage_config(config: BlobStorageConfig) -> None: + """Set the blob store configuration (tests / programmatic setup).""" + global _blob_storage_config + _blob_storage_config = config + + +def load_blob_storage_config_from_dict(config_dict: dict) -> None: + """Load blob store configuration from a dictionary. + + Host-shared fields (``enabled`` / ``backend`` / ``backend_config``) are + read directly. Unknown top-level keys (likely typos) are warned and + ignored -- there are no legacy pre-abstraction fields to migrate (the + blob store is new in this contract). + """ + global _blob_storage_config + config_dict = dict(config_dict or {}) + unknown = sorted(key for key in config_dict if key not in _SHARED_FIELDS) + if unknown: + logger.warning( + "Unknown blob_storage config key(s) %r at top level (shared fields: %s); ignored.", + unknown, + sorted(_SHARED_FIELDS), + ) + config_dict = {key: value for key, value in config_dict.items() if key in _SHARED_FIELDS} + _blob_storage_config = BlobStorageConfig(**config_dict) diff --git a/backend/packages/harness/deerflow/storage/AGENTS.md b/backend/packages/harness/deerflow/storage/AGENTS.md new file mode 100644 index 0000000000..5476fc7759 --- /dev/null +++ b/backend/packages/harness/deerflow/storage/AGENTS.md @@ -0,0 +1,79 @@ +### Blob Store (`packages/harness/deerflow/storage/`) + +**What it owns**: a content-addressed store for bytes that must resolve on every +instance, not just the one that wrote them (issue #4189, item 2). The address is +the SHA-256 of the content, so writes are idempotent, deduplication is free, and +a reader can verify what it got back. Contract: `storage/contract.py` +(`BlobRef`, `BlobStore`, the `BlobStoreError` family, `validate_blob_kind`). +Operator-facing behaviour and the producer migration plan: `docs/blob-storage.md`. +Factory: `storage/manager.py`. + +**Disabled by default**: `blob_storage.enabled` is `false`, and no producer has +been migrated, so an untouched deployment behaves exactly as before. Producers +call `get_blob_store_if_enabled()` and keep their existing local-path code for the +`None` case; only a caller that genuinely requires the store uses +`get_blob_store()`, which raises `BlobNotConfiguredError` when it is disabled. +Both accessors are exported from this package — prefer +`from deerflow.storage import ...` over reaching into submodules. +The factory compares the effective backend and backend config before reusing its +cached store, so hot-reloaded `blob_storage` edits take effect on new accesses. +Replaced stores stay open for callers that already hold them and are closed by +`reset_blob_store()`; changing a root requires preserving existing blob refs +and coordinating deployment instances. + +**Layout** (local_fs, the default):: + + /// the bytes + ///.json sidecar metadata + +The sidecar holds what the bytes cannot say (`content_type`, +`writer_thread_id`, `created_at`). It is **not** a reference count and reads must +not depend on it: losing it cannot make content unreadable. + +**Adding a backend** — the memory backends' drop-in contract, folder name == +backend name == `blob_storage.backend` value: + +1. Create `storage/backends//` with an `__init__.py` that exposes + `STORE_CLASS = `; the factory discovers backends by scanning that + folder, so no registry edit is needed. +2. Subclass `BlobStore` and implement `from_config`, `put_bytes`, `get_bytes`; + `exists` / `delete` / `close` have working defaults (`delete` raises so a + backend that cannot delete says so). Verify the digest on read — a store that + silently returns wrong bytes is indistinguishable from a corrupt checkpoint. +3. Keep the portability rule: a backend talks to the host through the method + arguments and its `backend_config` dict only. The single `deerflow` import a + backend folder needs is the contract line importing `BlobStore`, so a + third-party backend can live outside this tree and be selected with a dotted + path (`pkg.mod.Class` or `pkg.mod:Class`). +4. Reject an unusable config by raising; a store that silently starts on a + different root or backend strands previously written content. +5. Add the backend's semantics tests to `backend/tests/test_blob_store.py`; the + suite runs every backend-agnostic scenario against a temporary root. + +**Durability rules that reviewers have enforced**, worth keeping intact: + +- Validate the `kind` through `validate_blob_kind` instead of re-implementing the + grammar; it is also a path segment in `local_fs`, so traversal and separator + surprises are ruled out once, at the contract level. +- Publish atomically and tolerate a concurrent writer: two gateway instances may + put the same content at the same time (see `LocalFsBlobStore._publish`, which + handles the Windows `os.replace`/`WinError 5` case by recognising that the + destination already holds this content). +- Fail fast on an unresolvable backend: `_resolve_store_class` raises rather than + substituting a default, because blobs are persistent state. +- Reject oversized puts early: `local_fs` enforces `_MAX_BLOB_BYTES` (64 MiB) per + put as a backend-level defense-in-depth limit — not part of the `BlobStore` + contract, so a different backend sets its own policy, and the + externalized-tool-results follow-up must stream or split payloads before + `put_bytes`. + +**Garbage collection is not this module's job.** A blob row may be referenced by +several threads once identical content dedupes, so a sweep must establish +liveness from the durable references (the checkpoint rows that name the blob) and +never from the sidecar's `writer_thread_id`. The exact rule, and how it composes +with the checkpoint retention contract (#5255), is stated in +`docs/blob-storage.md`. + +**Entry points**: `packages/harness/deerflow/storage/contract.py`, +`storage/manager.py`, `storage/backends/local_fs/local_fs_store.py`, +`config/blob_storage_config.py`; tests in `backend/tests/test_blob_store.py`. diff --git a/backend/packages/harness/deerflow/storage/__init__.py b/backend/packages/harness/deerflow/storage/__init__.py new file mode 100644 index 0000000000..a2b80bc87d --- /dev/null +++ b/backend/packages/harness/deerflow/storage/__init__.py @@ -0,0 +1,34 @@ +"""Content-addressed blob store (issue #4189, item 2). + +See ``contract.py`` for the interface and ``AGENTS.md`` for how to add a +backend. Nothing in deer-flow writes here until ``blob_storage.enabled`` is +True and a producer has been migrated to the store. +""" + +from deerflow.storage.contract import ( + BlobNotConfiguredError, + BlobNotFoundError, + BlobReadError, + BlobRef, + BlobStore, + BlobStoreError, + BlobWriteError, + is_valid_blob_kind, + validate_blob_kind, +) +from deerflow.storage.manager import get_blob_store, get_blob_store_if_enabled, reset_blob_store + +__all__ = [ + "BlobNotFoundError", + "BlobNotConfiguredError", + "BlobReadError", + "BlobRef", + "BlobStore", + "BlobStoreError", + "BlobWriteError", + "get_blob_store", + "get_blob_store_if_enabled", + "is_valid_blob_kind", + "reset_blob_store", + "validate_blob_kind", +] diff --git a/backend/packages/harness/deerflow/storage/backends/__init__.py b/backend/packages/harness/deerflow/storage/backends/__init__.py new file mode 100644 index 0000000000..b0dcfdc139 --- /dev/null +++ b/backend/packages/harness/deerflow/storage/backends/__init__.py @@ -0,0 +1,13 @@ +"""Pluggable blob store backends. + +Each subpackage is a self-contained backend that exposes +``STORE_CLASS`` (a :class:`~deerflow.storage.contract.BlobStore` +subclass) in its ``__init__``. The drop-in contract: folder name == +backend name == ``BlobStorageConfig.backend`` value. + +Add a new backend by dropping a new folder here and setting +``blob_storage.backend: `` -- no other deer-flow code changes. +The depth (layout, publish rules, portability, dotted-path escape hatch) +is owned by ``../AGENTS.md``; the retention/GC interaction lives in +``docs/blob-storage.md``. +""" diff --git a/backend/packages/harness/deerflow/storage/backends/local_fs/__init__.py b/backend/packages/harness/deerflow/storage/backends/local_fs/__init__.py new file mode 100644 index 0000000000..7fc38b9e53 --- /dev/null +++ b/backend/packages/harness/deerflow/storage/backends/local_fs/__init__.py @@ -0,0 +1,13 @@ +"""local_fs backend -- the default blob store (content-addressed files). + +Holds the store class (:mod:`local_fs_store`): content-addressed files under +``///`` plus a JSON sidecar carrying what the +bytes cannot say, published atomically and verified on read. +""" + +from .local_fs_store import LocalFsBlobStore + +#: The :class:`~deerflow.storage.contract.BlobStore` subclass this +#: backend exposes. Discovered by the factory's ``_scan_backends`` drop-in +#: mechanism under the folder name ``local_fs``. +STORE_CLASS = LocalFsBlobStore diff --git a/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py b/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py new file mode 100644 index 0000000000..04bce9a331 --- /dev/null +++ b/backend/packages/harness/deerflow/storage/backends/local_fs/local_fs_store.py @@ -0,0 +1,258 @@ +"""Local filesystem blob store -- the default backend. + +Layout (content-addressed, sharded):: + + /// the bytes + ///.json sidecar metadata + +The address is the SHA-256 of the content, so: + +* writes are idempotent -- putting the same bytes again finds the file already + present and does nothing; +* deduplication is free -- two producers that externalize the same output share + one file; +* a reader can verify what it got back against the address, which is what makes + "which content" a safe replacement for "where on this machine". + +The sidecar carries what cannot be derived from the bytes alone +(``content_type``, ``writer_thread_id``, creation time). Reads do not depend on +it: a lost or truncated sidecar must not make blob content unreadable. It is also +**not a reference count** -- ``writer_thread_id`` records whichever writer landed +first, and identical bytes collapse to one file, so a sweep must confirm that no +surviving reference needs the content before unlinking it (the rule in +``docs/blob-storage.md``). + +Concurrency: two gateway instances may put the same blob at the same time. +Both write to a unique temp file in the destination directory and +``os.replace`` it into place, so the last writer wins with identical content +and no reader ever observes a partial file. On Windows ``os.replace`` is +atomic within a volume, which is why the temp file is created in the +destination directory rather than a global temp dir. +""" + +from __future__ import annotations + +import hashlib +import json +import logging +import os +import tempfile +import time +from pathlib import Path +from typing import Any + +from deerflow.storage.contract import ( + BlobNotFoundError, + BlobReadError, + BlobRef, + BlobStore, + BlobStoreError, + BlobWriteError, + validate_blob_kind, +) + +logger = logging.getLogger(__name__) + +_META_SUFFIX = ".json" +_MAX_BLOB_BYTES = 64 * 1024 * 1024 # defense-in-depth cap on a single put +# os.replace retry budget for the concurrent-writer race (see _publish). +_REPLACE_ATTEMPTS = 3 +_REPLACE_BACKOFF_SECONDS = 0.02 + + +class LocalFsBlobStore(BlobStore): + """Content-addressed store on the local filesystem. + + This is the default and the migration bridge: when ``root`` points at a + shared volume (NFS, EFS, a PVC in ``ReadWriteMany``) the store already + gives multi-instance resolution with no object storage dependency. The S3 + backend is a later, optional extra -- it implements the same contract. + """ + + def __init__(self, root: str | os.PathLike[str]) -> None: + self._root = Path(root).expanduser() + try: + self._root.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise BlobWriteError(f"Cannot create blob store root {self._root}: {exc}") from exc + if not os.access(self._root, os.W_OK): + raise BlobWriteError(f"Blob store root is not writable: {self._root}") + + @classmethod + def from_config(cls, backend_config: dict[str, Any]) -> LocalFsBlobStore: + root = backend_config.get("root") + if not root: + raise ValueError("local_fs blob store requires backend_config.root (the factory injects a default)") + return cls(root=str(root)) + + # -- layout ----------------------------------------------------------- + + def _paths_for(self, ref: BlobRef) -> tuple[Path, Path]: + shard_dir = self._root / ref.kind / ref.sha256[:2] + return shard_dir / ref.sha256, shard_dir / f"{ref.sha256}{_META_SUFFIX}" + + # -- contract --------------------------------------------------------- + + def put_bytes( + self, + data: bytes, + *, + kind: str, + content_type: str | None = None, + thread_id: str | None = None, + ) -> BlobRef: + validate_blob_kind(kind) + if len(data) > _MAX_BLOB_BYTES: + raise BlobWriteError(f"Refusing to store {len(data)} bytes (cap is {_MAX_BLOB_BYTES})") + + digest = hashlib.sha256(data).hexdigest() + ref = BlobRef(sha256=digest, size=len(data), kind=kind, content_type=content_type) + data_path, meta_path = self._paths_for(ref) + shard_dir = data_path.parent + + if data_path.is_file(): + # Idempotent re-put: content already present. Ensure the sidecar + # exists (an earlier put may have crashed between the two writes). + if not meta_path.is_file(): + self._write_meta(ref, meta_path, thread_id) + return ref + + try: + shard_dir.mkdir(parents=True, exist_ok=True) + fd, tmp_name = tempfile.mkstemp(prefix=".put-", suffix=".tmp", dir=shard_dir) + try: + with os.fdopen(fd, "wb") as handle: + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + self._publish(tmp_name, data_path, data) + except BaseException: + # Best effort: a stranded temp file must not accumulate. + try: + os.unlink(tmp_name) + except OSError: + pass + raise + self._write_meta(ref, meta_path, thread_id) + except BlobWriteError: + raise + except OSError as exc: + raise BlobWriteError(f"Failed to persist blob {ref.sha256[:12]}: {exc}") from exc + return ref + + def get_bytes(self, ref: BlobRef) -> bytes: + data_path, _ = self._paths_for(ref) + try: + data = data_path.read_bytes() + except FileNotFoundError as exc: + raise BlobNotFoundError(f"Blob {ref.sha256[:12]} not found under kind {ref.kind!r}") from exc + except OSError as exc: + raise BlobReadError(f"Failed to read blob {ref.sha256[:12]}: {exc}") from exc + + # Verify against the address. A content-addressed store that returns + # the wrong bytes silently is indistinguishable from a corrupt + # checkpoint, so fail loudly instead — naming WHICH check failed: + # a size delta points at truncation/partial writes, a digest mismatch + # at same-size corruption (bit rot), and conflating the two sends the + # debugger in the wrong direction ("got 1024 bytes, expected 1024"). + if len(data) != ref.size: + raise BlobReadError(f"Blob {ref.sha256[:12]} size mismatch on read (got {len(data)} bytes, expected {ref.size}); the store may be truncated or corrupted") + digest = hashlib.sha256(data).hexdigest() + if digest != ref.sha256: + raise BlobReadError(f"Blob {ref.sha256[:12]} content digest mismatch on read (content hashes to {digest[:12]}, ref names {ref.sha256[:12]}); the store may be corrupted") + return data + + def exists(self, ref: BlobRef) -> bool: + data_path, _ = self._paths_for(ref) + return data_path.is_file() + + def delete(self, ref: BlobRef) -> None: + data_path, meta_path = self._paths_for(ref) + removed = False + for path in (data_path, meta_path): + try: + path.unlink() + removed = True + except FileNotFoundError: + continue + except OSError as exc: + # The neutral base, not BlobReadError: this error family is + # public API from day one, and a caller matching on a read + # failure must not be handed a failed unlink it cannot read as + # one. + raise BlobStoreError(f"Failed to delete blob {ref.sha256[:12]}: {exc}") from exc + if not removed: + logger.debug("delete on absent blob %s (kind=%s): already gone", ref.sha256[:12], ref.kind) + + def close(self) -> None: # noqa: B027 - nothing to release + """No handles to close; the store is stateless beyond the root path.""" + + # -- internals -------------------------------------------------------- + + def _publish(self, tmp_path: str, data_path: Path, data: bytes) -> None: + """Move the temp file into place, tolerating a concurrent writer. + + ``os.replace`` raises ``PermissionError`` (WinError 5) on Windows when + another handle still holds the destination open -- which is exactly + what happens when two gateway instances put the same blob while a + reader is already serving it. If the destination already holds this + exact content, another writer won the race and the put is complete; + otherwise retry briefly, then surface the error. + """ + for attempt in range(_REPLACE_ATTEMPTS): + try: + os.replace(tmp_path, data_path) + return + except PermissionError: + if self._content_matches(data_path, data): + self._discard(tmp_path) + return + if attempt + 1 == _REPLACE_ATTEMPTS: + raise + time.sleep(_REPLACE_BACKOFF_SECONDS * (attempt + 1)) + + @staticmethod + def _content_matches(path: Path, data: bytes) -> bool: + try: + return path.read_bytes() == data + except OSError: + return False + + @staticmethod + def _discard(path: str) -> None: + try: + os.unlink(path) + except OSError: + pass + + def _write_meta(self, ref: BlobRef, meta_path: Path, thread_id: str | None) -> None: + payload = { + "sha256": ref.sha256, + "size": ref.size, + "kind": ref.kind, + "content_type": ref.content_type, + # Advisory provenance only: the first writer of this content. It is + # deliberately not a "threads that reference this" list -- identical + # bytes dedupe to one file, and appending to a sidecar from two + # gateway instances is a lost-update race, so such a list could not + # be trusted as a reference count anyway. + "writer_thread_id": thread_id, + "created_at": time.time(), + } + fd, tmp_name = tempfile.mkstemp(prefix=".meta-", suffix=".tmp", dir=meta_path.parent) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle) + os.replace(tmp_name, meta_path) + except OSError: + # The sidecar is an optimization for later GC, not a read + # dependency: losing it must not fail a put that already + # succeeded. + try: + os.unlink(tmp_name) + except OSError: + pass + logger.warning("Could not write blob sidecar %s", meta_path, exc_info=True) + + +__all__ = ["LocalFsBlobStore"] diff --git a/backend/packages/harness/deerflow/storage/contract.py b/backend/packages/harness/deerflow/storage/contract.py new file mode 100644 index 0000000000..9c6b889d40 --- /dev/null +++ b/backend/packages/harness/deerflow/storage/contract.py @@ -0,0 +1,198 @@ +"""Content-addressed blob store contract. + +A :class:`BlobRef` addresses **content**, not a location. That is the whole +point of the abstraction: on a multi-gateway deployment a server-local +filesystem path is only meaningful on the instance that wrote the file, +whereas a content digest is meaningful everywhere -- every instance that can +reach the backing store resolves the same bytes. + +Producers (issue #4189, item 2): + +* ``ViewedImageData.actual_path`` -- written by ``view_image_tool``, read by + ``ViewImageMiddleware``, the gateway artifact routes and the IM channels. +* Oversized tool results externalized by ``ToolOutputBudgetMiddleware``. + +Both keep their existing local path when the store is disabled, so the seam is +purely additive. +""" + +from __future__ import annotations + +import re +from abc import ABC, abstractmethod +from typing import Any, ClassVar + +from pydantic import BaseModel, ConfigDict, Field, field_validator + +# Blob kinds are also filesystem path segments in the local_fs backend, so the +# grammar is deliberately narrow: lowercase, digits and hyphens, no leading or +# trailing hyphen, 1-64 chars. That rules out traversal (".."), separators and +# case-sensitivity surprises on NTFS/APFS at the contract level instead of +# leaving every backend to re-check it. +_KIND_PATTERN = re.compile(r"^(?=.{1,64}$)[a-z0-9]([a-z0-9-]*[a-z0-9])?$") +# The grammar as a caller sees it. One text, one place: this message is the only +# thing a producer gets when a kind is rejected, and a second copy of the +# pattern (the local_fs backend had one) drifts from the enforced one. +_KIND_GRAMMAR = "^[a-z0-9]([a-z0-9-]*[a-z0-9])?$ (1-64 chars, no leading or trailing hyphen)" + + +def is_valid_blob_kind(kind: str) -> bool: + """Return whether *kind* is a safe blob-kind identifier.""" + return isinstance(kind, str) and bool(_KIND_PATTERN.match(kind)) + + +def validate_blob_kind(kind: str) -> str: + """Return *kind* unchanged, or raise ``ValueError`` quoting the real grammar. + + Backends must not re-implement the pattern: a store that rejects a kind the + contract accepts -- or accepts one it rejects -- makes content addressable + on one backend and not another. + """ + if not is_valid_blob_kind(kind): + raise ValueError(f"kind must match {_KIND_GRAMMAR} (it is used as a path segment)") + return kind + + +class BlobStoreError(RuntimeError): + """Backend-neutral base error exposed at the blob store boundary.""" + + +class BlobNotConfiguredError(BlobStoreError): + """A caller required a blob store but ``blob_storage.enabled`` is False.""" + + +class BlobWriteError(BlobStoreError): + """A write failed after retries; the content was not persisted.""" + + +class BlobReadError(BlobStoreError): + """A read failed for a reason other than the blob being absent.""" + + +class BlobNotFoundError(BlobReadError): + """The referenced content is not present in the backing store.""" + + +class BlobRef(BaseModel): + """A content-addressed reference to blob bytes. + + ``sha256`` is the address: writes are idempotent (putting the same bytes + twice yields the same ref) and deduplication is free. ``size`` is carried + so a reader can sanity-check what it got back without trusting the + backend, and ``kind`` scopes garbage collection -- retention sweeps can + reason about a kind ("externalized tool outputs for thread X") without + parsing paths. + """ + + model_config = ConfigDict(frozen=True) + + sha256: str = Field(description="Lowercase hex SHA-256 of the blob content (64 chars).") + size: int = Field(ge=0, description="Content length in bytes.") + kind: str = Field(description="Content class, e.g. 'viewed-image' or 'tool-output'.") + content_type: str | None = Field(default=None, description="MIME type when the producer knows it.") + + @field_validator("sha256") + @classmethod + def _sha256_is_hex64(cls, value: str) -> str: + value = (value or "").strip().lower() + if len(value) != 64 or not re.fullmatch(r"[0-9a-f]{64}", value): + raise ValueError("sha256 must be 64 lowercase hex characters") + return value + + @field_validator("kind") + @classmethod + def _kind_is_safe(cls, value: str) -> str: + return validate_blob_kind(value) + + def matches(self, data: bytes) -> bool: + """Return whether *data* really is this blob's content.""" + import hashlib + + return len(data) == self.size and hashlib.sha256(data).hexdigest() == self.sha256 + + +class BlobStore(ABC): + """Backend-neutral blob store contract. + + A plain ABC (like ``MemoryStorage``, not the pydantic ``MemoryManager``): + the contract carries no config fields, so there is nothing for pydantic to + validate and no ``PrivateAttr`` ceremony. Backends are constructed by the + factory through :meth:`from_config`. + + Portability rule (mirrors the memory backends' golden rule): a backend + talks to the host through exactly two channels -- the method arguments and + the ``backend_config`` dict. The only ``deerflow`` import a backend folder + needs is the contract line importing :class:`BlobStore`. + """ + + #: Sentinel attribute each backend package's ``__init__`` exposes so the + #: folder-scan factory can find it (mirrors memory's ``MANAGER_CLASS``). + STORE_CLASS_ATTR: ClassVar[str] = "STORE_CLASS" + + @classmethod + @abstractmethod + def from_config(cls, backend_config: dict[str, Any]) -> BlobStore: + """Build a store from backend-private config. + + ``backend_config`` is the dict from ``blob_storage.backend_config``, + with ``root`` injected by the factory when the host did not set one. + Must raise on an unusable config -- a blob store that silently starts + on the wrong root is worse than one that refuses to start. + """ + + @abstractmethod + def put_bytes( + self, + data: bytes, + *, + kind: str, + content_type: str | None = None, + thread_id: str | None = None, + ) -> BlobRef: + """Persist *data* and return its :class:`BlobRef`. + + Idempotent: putting identical bytes twice returns the same ref and must + not duplicate storage. + + ``thread_id`` is **advisory provenance, never a reference count.** + Content addressing means identical bytes collapse to one object, so two + threads that externalize the same content share it, and a backend can + only record whichever writer reached it first -- on a multi-instance + deployment it cannot even do that reliably, since two gateway instances + appending to one sidecar is a read-modify-write race that loses entries. + Garbage collection must therefore establish liveness from the durable + references (the checkpoint rows that name the blob), not from this + field; see ``docs/blob-storage.md`` for the exact rule. + """ + + @abstractmethod + def get_bytes(self, ref: BlobRef) -> bytes: + """Return the content addressed by *ref*. + + Raises :class:`BlobNotFoundError` when the content is absent and + :class:`BlobReadError` when it is present but unreadable. Backends + SHOULD verify the digest on read: a content-addressed store that + silently returns wrong bytes is indistinguishable from a corrupt + checkpoint. + """ + + def exists(self, ref: BlobRef) -> bool: + """Return whether *ref* is present. Default probes via :meth:`get_bytes`.""" + try: + self.get_bytes(ref) + except BlobNotFoundError: + return False + return True + + def delete(self, ref: BlobRef) -> None: + """Remove the content addressed by *ref*. + + Default raises so a backend that cannot delete says so instead of + leaving a caller to assume a sweep succeeded. Deleting an absent blob + is not an error (idempotent), matching how the retention contract + treats orphans. + """ + raise NotImplementedError(f"{type(self).__name__} does not support delete") + + def close(self) -> None: # noqa: B027 - intentional no-op default + """Release backend resources. Default: nothing to release.""" diff --git a/backend/packages/harness/deerflow/storage/manager.py b/backend/packages/harness/deerflow/storage/manager.py new file mode 100644 index 0000000000..412fb056b4 --- /dev/null +++ b/backend/packages/harness/deerflow/storage/manager.py @@ -0,0 +1,223 @@ +"""Blob store factory: folder-scan backend resolution + process singleton. + +Mirrors ``deerflow/agents/memory/manager.py`` (the pattern maintainers pointed +at for issue #4189 item 2): backends live in ``storage/backends//``, each +package's ``__init__`` exposes ``STORE_CLASS``, and ``blob_storage.backend`` +names one of them -- or a dotted import path for a backend that lives elsewhere. + +Two accessors, so call sites cannot drift: + +* :func:`get_blob_store_if_enabled` -- the one producers should use. Returns + ``None`` when ``blob_storage.enabled`` is False, which is the default, so an + unmigrated call site is a no-op rather than an accident. +* :func:`get_blob_store` -- for callers that require a store; raises + :class:`BlobNotConfiguredError` when the store is disabled. + +Both fail fast on an unresolvable backend. A blob store that silently starts on +a different backend than the operator configured would strand previously +written content, which is worse than refusing to start. +""" + +from __future__ import annotations + +import importlib +import logging +import threading +from copy import deepcopy +from pathlib import Path +from types import ModuleType +from typing import Any + +from deerflow.config.blob_storage_config import BlobStorageConfig, get_blob_storage_config +from deerflow.storage.contract import BlobNotConfiguredError, BlobStore, BlobStoreError + +logger = logging.getLogger(__name__) + +# Backend packages live in /backends//. +_BACKENDS_DIR = Path(__file__).parent / "backends" +# Sentinel attribute each backend's __init__ exposes (a BlobStore subclass). +_STORE_CLASS_ATTR = "STORE_CLASS" + +# Singleton instance + the effective settings it was built with. Keep replaced +# stores alive until reset so callers already holding one can finish using it. +_blob_store: BlobStore | None = None +_blob_store_settings: tuple[str, dict[str, Any]] | None = None +_retired_blob_stores: list[BlobStore] = [] +_backends_cache: dict[str, type[BlobStore]] | None = None +_store_lock = threading.Lock() + + +def _import_backend_module(name: str) -> ModuleType: + """Import ``backends/`` and return its package module.""" + package_dir = _BACKENDS_DIR / name + if not package_dir.is_dir() or not (package_dir / "__init__.py").is_file(): + raise ValueError(f"Unknown blob store backend {name!r}: no storage/backends/{name}/ package") + return importlib.import_module(f"{__package__}.backends.{name}") + + +def _resolve_store_class(backend: str) -> type[BlobStore]: + """Resolve ``blob_storage.backend`` to a :class:`BlobStore` subclass. + + Two forms are accepted, matching the memory backends' contract: + + * a registered backend name -- a folder under ``storage/backends/`` whose + ``__init__`` exposes ``STORE_CLASS``; + * a dotted import path -- ``pkg.mod.Class`` for a backend that lives + outside this package (an extension or a deployment-local backend). + """ + global _backends_cache + + if "." not in backend: + if _backends_cache is None: + _backends_cache = _scan_backends() + cls = _backends_cache.get(backend) + if cls is None: + available = ", ".join(sorted(_backends_cache)) or "(none)" + raise ValueError(f"Unknown blob store backend {backend!r}. Registered backends: {available}. Alternatively pass a dotted import path to a BlobStore subclass.") + return cls + + module_path, _, class_name = backend.partition(":") if ":" in backend else backend.rpartition(".") + if not module_path or not class_name: + raise ValueError(f"Invalid blob storage backend {backend!r}: needs 'pkg.mod.Class' or 'pkg.mod:Class'") + try: + module = importlib.import_module(module_path) + candidate = getattr(module, class_name) + except (ImportError, AttributeError) as exc: + raise ValueError(f"Cannot import blob storage backend {backend!r}: {exc}") from exc + if not (isinstance(candidate, type) and issubclass(candidate, BlobStore)): + raise ValueError(f"Blob storage backend {backend!r} is not a BlobStore subclass") + return candidate + + +def _scan_backends() -> dict[str, type[BlobStore]]: + """Return ``{name: BlobStore subclass}`` for every backend folder. + + Imports through :func:`_import_backend_module` so the "is this a backend + package?" check lives in exactly one place -- a folder that survives the + scan is a folder the factory can resolve by name. + """ + registry: dict[str, type[BlobStore]] = {} + if not _BACKENDS_DIR.is_dir(): + return registry + for entry in sorted(_BACKENDS_DIR.iterdir()): + if not entry.is_dir() or entry.name.startswith("_"): + continue + try: + module = _import_backend_module(entry.name) + except ValueError as exc: + # Not a backend package (no __init__.py); nothing to register. + logger.debug("Skipping %s: %s", entry.name, exc) + continue + except Exception: + # A broken optional backend must not take the whole store down: + # log it and leave it out of the registry so the factory can still + # resolve the default. + logger.exception("Skipping blob store backend %r: import failed", entry.name) + continue + cls = getattr(module, _STORE_CLASS_ATTR, None) + if isinstance(cls, type) and issubclass(cls, BlobStore): + registry[entry.name] = cls + else: + logger.warning( + "Blob store backend folder %r exposes no %s; skipping", + entry.name, + _STORE_CLASS_ATTR, + ) + return registry + + +def _default_backend_config() -> dict[str, Any]: + """Zero-config UX: default the backend root to deer-flow's state dir. + + Absolute and CWD-independent, so every gateway instance in a deployment + that has not configured a shared root still lands in the same logical + place on its own node -- which is exactly the single-host behaviour we + must not change. + """ + from deerflow.config.runtime_paths import runtime_home + + return {"root": str((Path(runtime_home()) / "blobs").resolve())} + + +def _resolve_backend_config(cfg: BlobStorageConfig) -> dict[str, Any]: + backend_config = dict(cfg.backend_config or {}) + root = backend_config.get("root") + if not root: + backend_config.update(_default_backend_config()) + elif not Path(str(root)).is_absolute(): + # A relative root is resolved against runtime_home() (base_dir-relative, + # CWD-independent) in host code, so backends stay free of any + # runtime_home dependency -- same reason the memory factory does it. + from deerflow.config.runtime_paths import runtime_home + + backend_config["root"] = str((Path(runtime_home()) / str(root)).resolve()) + return backend_config + + +def get_blob_store_if_enabled() -> BlobStore | None: + """Return the singleton store, or ``None`` when the store is disabled. + + This is the accessor producers should call: ``None`` is an explicit, + checkable signal that the deployment did not opt in, and the producer then + keeps its existing server-local path. + """ + return _get_blob_store_if_enabled() + + +def get_blob_store() -> BlobStore: + """Return the singleton :class:`BlobStore` for the active config. + + Raises :class:`BlobNotConfiguredError` when ``blob_storage.enabled`` is + False, and ``ValueError`` when the configured backend cannot be resolved. + """ + store = _get_blob_store_if_enabled() + if store is None: + raise BlobNotConfiguredError("blob_storage.enabled is False; no blob store is available. Use get_blob_store_if_enabled() at optional call sites.") + return store + + +def _get_blob_store_if_enabled() -> BlobStore | None: + """Select a store against one config snapshot under the singleton lock.""" + global _blob_store, _blob_store_settings + + with _store_lock: + cfg = get_blob_storage_config() + if not cfg.enabled: + return None + + backend_config = _resolve_backend_config(cfg) + settings = (cfg.backend, deepcopy(backend_config)) + if _blob_store is not None and _blob_store_settings == settings: + return _blob_store + + cls = _resolve_store_class(cfg.backend) + new_store = cls.from_config(backend_config) + if _blob_store is not None: + _retired_blob_stores.append(_blob_store) + _blob_store = new_store + _blob_store_settings = settings + logger.info("Blob store resolved: %s (backend=%r)", type(_blob_store).__name__, cfg.backend) + return _blob_store + + +def reset_blob_store() -> None: + """Close current and replaced stores; clear the singleton and registry.""" + global _blob_store, _blob_store_settings, _backends_cache + with _store_lock: + for store in [*_retired_blob_stores, *([_blob_store] if _blob_store is not None else [])]: + try: + store.close() + except Exception: + logger.warning("Blob store close() raised during reset", exc_info=True) + _retired_blob_stores.clear() + _blob_store = None + _blob_store_settings = None + _backends_cache = None + + +__all__ = [ + "BlobStoreError", + "get_blob_store", + "get_blob_store_if_enabled", + "reset_blob_store", +] diff --git a/backend/tests/test_agent_guidance_check.py b/backend/tests/test_agent_guidance_check.py index bc85b1d74c..8c6e585933 100644 --- a/backend/tests/test_agent_guidance_check.py +++ b/backend/tests/test_agent_guidance_check.py @@ -32,6 +32,7 @@ EXPECTED_GUIDANCE_PATHS = { "backend/packages/harness/deerflow/persistence/user/AGENTS.md", "backend/packages/harness/deerflow/reflection/AGENTS.md", "backend/packages/harness/deerflow/skills/AGENTS.md", + "backend/packages/harness/deerflow/storage/AGENTS.md", "backend/packages/harness/deerflow/subagents/AGENTS.md", "backend/packages/harness/deerflow/tools/AGENTS.md", "backend/packages/harness/deerflow/tracing/AGENTS.md", diff --git a/backend/tests/test_app_config_reload.py b/backend/tests/test_app_config_reload.py index aacad592a5..847d28dd32 100644 --- a/backend/tests/test_app_config_reload.py +++ b/backend/tests/test_app_config_reload.py @@ -23,6 +23,7 @@ from deerflow.config.title_config import get_title_config, load_title_config_fro from deerflow.config.tool_search_config import get_tool_search_config, load_tool_search_config_from_dict from deerflow.runtime.checkpointer import get_checkpointer, reset_checkpointer from deerflow.runtime.store import get_store, reset_store +from deerflow.storage import BlobNotConfiguredError, get_blob_store, get_blob_store_if_enabled, reset_blob_store def _reset_config_singletons() -> None: @@ -804,6 +805,40 @@ def test_get_memory_config_self_syncs_without_prior_get_app_config(tmp_path, mon _reset_config_singletons() +def test_blob_store_follows_config_file_reload(tmp_path, monkeypatch): + """The storage accessor must see config edits without a prior app-config read.""" + config_path = tmp_path / "config.yaml" + extensions_path = tmp_path / "extensions_config.json" + _write_extensions_config(extensions_path) + first_root = tmp_path / "first" + second_root = tmp_path / "second" + _write_config_with_sections(config_path, {"blob_storage": {"enabled": True, "backend_config": {"root": str(first_root)}}}) + + monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(config_path)) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(extensions_path)) + reset_app_config() + reset_blob_store() + + try: + get_app_config() + first_store = get_blob_store() + + _write_config_with_sections(config_path, {"blob_storage": {"enabled": True, "backend_config": {"root": str(second_root)}}}) + second_store = get_blob_store() + assert second_store is not first_store + ref = second_store.put_bytes(b"after reload", kind="tool-output") + assert (second_root / ref.kind / ref.sha256[:2] / ref.sha256).is_file() + assert not (first_root / ref.kind / ref.sha256[:2] / ref.sha256).exists() + + _write_config_with_sections(config_path, {"blob_storage": {"enabled": False}}) + assert get_blob_store_if_enabled() is None + with pytest.raises(BlobNotConfiguredError): + get_blob_store() + finally: + reset_blob_store() + _reset_config_singletons() + + def test_get_memory_config_falls_back_on_broken_config(tmp_path, monkeypatch): """get_memory_config() does not crash on transiently broken config.yaml. diff --git a/backend/tests/test_blob_store.py b/backend/tests/test_blob_store.py new file mode 100644 index 0000000000..c4ea6af00e --- /dev/null +++ b/backend/tests/test_blob_store.py @@ -0,0 +1,400 @@ +"""Tests for the content-addressed blob store contract and the local_fs backend. + +Scope: the seam only. No producer is migrated yet, so these tests never touch +``viewed_images`` or ``ToolOutputBudgetMiddleware`` -- that is the follow-up +wiring PR for issue #4189 item 2. +""" + +from __future__ import annotations + +import hashlib +import json +import threading +from pathlib import Path + +import pytest +from pydantic import ValidationError + +from deerflow.config import blob_storage_config as bsc +from deerflow.storage import ( + BlobNotConfiguredError, + BlobNotFoundError, + BlobReadError, + BlobRef, + BlobStore, + BlobStoreError, + BlobWriteError, + get_blob_store, + get_blob_store_if_enabled, + is_valid_blob_kind, + reset_blob_store, + validate_blob_kind, +) +from deerflow.storage import contract as contract_module +from deerflow.storage.backends.local_fs import LocalFsBlobStore + + +@pytest.fixture(autouse=True) +def _isolated_config(monkeypatch): + """Give every test a pristine config + a fresh singleton.""" + bsc.set_blob_storage_config(bsc.BlobStorageConfig()) + reset_blob_store() + yield + bsc.set_blob_storage_config(bsc.BlobStorageConfig()) + reset_blob_store() + + +@pytest.fixture +def store(tmp_path: Path) -> LocalFsBlobStore: + return LocalFsBlobStore.from_config({"root": str(tmp_path / "blobs")}) + + +# --------------------------------------------------------------------------- +# BlobRef +# --------------------------------------------------------------------------- + + +def test_blob_ref_validates_digest_shape(): + with pytest.raises(ValidationError): + BlobRef(sha256="not-a-digest", size=1, kind="tool-output") + with pytest.raises(ValidationError): + BlobRef(sha256="z" * 64, size=1, kind="tool-output") # right length, not hex + with pytest.raises(ValidationError): + BlobRef(sha256="a" * 63, size=1, kind="tool-output") # too short + ref = BlobRef(sha256=hashlib.sha256(b"x").hexdigest(), size=1, kind="tool-output") + assert ref.sha256 == hashlib.sha256(b"x").hexdigest() + + +def test_blob_ref_kind_is_path_safe(): + for bad in ("../etc", "a/b", "Aupper", "", "-lead", "trailing-", "has space", "x" * 65): + with pytest.raises(ValidationError): + BlobRef(sha256=hashlib.sha256(b"x").hexdigest(), size=1, kind=bad) + for good in ("viewed-image", "tool-output", "a", "0", "a" * 64): + assert is_valid_blob_kind(good) + + +def test_blob_ref_matches_detects_corruption(): + data = b"payload" + ref = BlobRef(sha256=hashlib.sha256(data).hexdigest(), size=len(data), kind="tool-output") + assert ref.matches(data) + assert not ref.matches(b"payload!") + assert not ref.matches(b"") + + +# --------------------------------------------------------------------------- +# local_fs backend semantics +# --------------------------------------------------------------------------- + + +def test_put_then_get_roundtrip(store: LocalFsBlobStore): + data = "外部化内容 🚀".encode() + ref = store.put_bytes(data, kind="tool-output", content_type="text/plain", thread_id="t-1") + assert ref.size == len(data) + assert ref.sha256 == hashlib.sha256(data).hexdigest() + assert ref.content_type == "text/plain" + assert store.get_bytes(ref) == data + + +def test_put_is_idempotent_and_deduplicates(store: LocalFsBlobStore, tmp_path: Path): + data = b"same bytes" + first = store.put_bytes(data, kind="tool-output") + before = sorted(p.name for p in (tmp_path / "blobs").rglob("*") if p.is_file()) + second = store.put_bytes(data, kind="tool-output", thread_id="t-9") + after = sorted(p.name for p in (tmp_path / "blobs").rglob("*") if p.is_file()) + + assert first == second + assert before == after, "a re-put must not create new files" + + +def test_dedup_records_only_the_first_writer_as_advisory_provenance(store: LocalFsBlobStore): + """Identical content is *shared*, so the sidecar cannot be a reference count. + + The GC contract in docs/blob-storage.md says a blob may only be unlinked + after liveness is confirmed from the durable references; this test pins the + sidecar semantics that make that rule necessary, so nobody later "fixes" + dedup by trusting writer_thread_id. + """ + data = b"the same uploaded image" + first = store.put_bytes(data, kind="viewed-image", thread_id="thread-a") + second = store.put_bytes(data, kind="viewed-image", thread_id="thread-b") + + assert first == second, "one address for one content" + data_path = next((store._root / "viewed-image").rglob(first.sha256)) + sidecar = next((store._root / "viewed-image").rglob(f"{first.sha256}.json")) + assert data_path.is_file() + payload = json.loads(sidecar.read_text(encoding="utf-8")) + assert payload["writer_thread_id"] == "thread-a" + assert "thread-b" not in json.dumps(payload), "dedup keeps the first writer only — never a refcount" + + +def test_kind_rejection_quotes_the_enforced_grammar(): + """One validator, one message: the pattern a producer is shown must be the + pattern that is enforced (it was not, before this was centralised).""" + assert validate_blob_kind("tool-output") == "tool-output" + for bad in ("trailing-", "a--", "-lead", "Aupper", "a/b", "", "x" * 65): + with pytest.raises(ValueError) as caught: + validate_blob_kind(bad) + message = str(caught.value) + assert "^[a-z0-9]([a-z0-9-]*[a-z0-9])?$" in message, message + assert "1-64 chars" in message, message + assert "no leading or trailing hyphen" in message, message + assert validate_blob_kind is contract_module.validate_blob_kind + + +def test_same_content_different_kind_lands_separately(store: LocalFsBlobStore): + ref_a = store.put_bytes(b"shared", kind="tool-output") + ref_b = store.put_bytes(b"shared", kind="viewed-image") + assert ref_a.sha256 == ref_b.sha256 + assert ref_a != ref_b + assert store.exists(ref_a) and store.exists(ref_b) + + +def test_get_missing_blob_raises_not_found(store: LocalFsBlobStore): + ref = BlobRef(sha256=hashlib.sha256(b"absent").hexdigest(), size=6, kind="tool-output") + with pytest.raises(BlobNotFoundError): + store.get_bytes(ref) + assert store.exists(ref) is False + + +def test_get_detects_size_mismatch(store: LocalFsBlobStore): + data = b"intact" + ref = store.put_bytes(data, kind="tool-output") + # Corrupt the stored bytes behind the store's back with a DIFFERENT length: + # this is the truncation / partial-write failure mode. + data_path = next((store._root / "tool-output").rglob(ref.sha256)) + data_path.write_bytes(b"tampered") + with pytest.raises(BlobReadError, match="size mismatch"): + store.get_bytes(ref) + + +def test_get_detects_same_size_digest_mismatch(store: LocalFsBlobStore): + """Same-size corruption (bit rot) must hit the digest branch, not the size one. + + ``data[::-1]`` keeps the length and flips every byte, so the size check + passes and the digest comparison is the only guard that can catch it. + """ + data = b"intact" + ref = store.put_bytes(data, kind="tool-output") + data_path = next((store._root / "tool-output").rglob(ref.sha256)) + data_path.write_bytes(data[::-1]) + with pytest.raises(BlobReadError, match="digest mismatch"): + store.get_bytes(ref) + + +def test_delete_is_idempotent(store: LocalFsBlobStore): + ref = store.put_bytes(b"doomed", kind="tool-output") + store.delete(ref) + assert store.exists(ref) is False + store.delete(ref) # absent blob: not an error + with pytest.raises(BlobNotFoundError): + store.get_bytes(ref) + + +def test_delete_failure_raises_the_neutral_base_error(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch): + """A failed unlink is not a read failure. + + The exception family is public API from the first release, so callers that + match on BlobReadError must not receive delete failures they cannot + interpret. + """ + ref = store.put_bytes(b"cannot unlink", kind="tool-output") + + def _boom(self, *args, **kwargs): + raise OSError("device busy") + + monkeypatch.setattr(Path, "unlink", _boom) + with pytest.raises(BlobStoreError) as caught: + store.delete(ref) + assert not isinstance(caught.value, BlobReadError) + assert "Failed to delete blob" in str(caught.value) + + +def test_delete_unsupported_on_contract_default(tmp_path: Path): + class Minimal(BlobStore): + def from_config(cls, backend_config): + raise NotImplementedError + + def put_bytes(self, data, *, kind, content_type=None, thread_id=None): + raise NotImplementedError + + def get_bytes(self, ref): + raise BlobNotFoundError("nope") + + assert Minimal().exists(BlobRef(sha256="0" * 64, size=0, kind="tool-output")) is False + with pytest.raises(NotImplementedError): + Minimal().delete(BlobRef(sha256="0" * 64, size=0, kind="tool-output")) + + +def test_invalid_kind_is_rejected_before_touching_disk(store: LocalFsBlobStore, tmp_path: Path): + with pytest.raises(ValueError): + store.put_bytes(b"x", kind="../escape") + assert not any(tmp_path.rglob("escape")) + + +def test_oversized_put_is_refused(store: LocalFsBlobStore, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr("deerflow.storage.backends.local_fs.local_fs_store._MAX_BLOB_BYTES", 4) + with pytest.raises(BlobWriteError): + store.put_bytes(b"12345", kind="tool-output") + + +def test_concurrent_puts_of_same_content(tmp_path: Path): + """Two instances racing the same blob must not corrupt or duplicate it.""" + store = LocalFsBlobStore.from_config({"root": str(tmp_path / "blobs")}) + data = b"raced content" * 100 + results: list[BlobRef] = [] + errors: list[BaseException] = [] + barrier = threading.Barrier(8) + + def worker() -> None: + try: + barrier.wait(timeout=5) + results.append(store.put_bytes(data, kind="tool-output")) + except BaseException as exc: # pragma: no cover - surfaced below + errors.append(exc) + + threads = [threading.Thread(target=worker) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=30) + + assert not errors, errors + assert len({r.sha256 for r in results}) == 1 + assert store.get_bytes(results[0]) == data + assert len(list((tmp_path / "blobs" / "tool-output").rglob(f"{results[0].sha256}"))) == 1 + + +def test_put_survives_a_lost_sidecar(store: LocalFsBlobStore): + """The sidecar is GC metadata, not a read dependency.""" + data = b"payload-with-sidecar" + ref = store.put_bytes(data, kind="tool-output") + meta = next((store._root / "tool-output").rglob(f"{ref.sha256}.json")) + meta.unlink() + assert store.get_bytes(ref) == data + + +# --------------------------------------------------------------------------- +# Factory + config wiring +# --------------------------------------------------------------------------- + + +def test_package_exports_the_whole_error_family(): + """Callers use `from deerflow.storage import ...`, so the error they must + catch has to be reachable there — and be the same class the contract + raises, not a re-exported copy.""" + assert BlobNotConfiguredError is contract_module.BlobNotConfiguredError + assert BlobStoreError is contract_module.BlobStoreError + for name in ("BlobNotConfiguredError", "BlobStoreError", "BlobNotFoundError", "validate_blob_kind"): + assert name in __import__("deerflow.storage", fromlist=["__all__"]).__all__ + + bsc.set_blob_storage_config(bsc.BlobStorageConfig()) + reset_blob_store() + with pytest.raises(BlobStoreError): + get_blob_store() + + +def test_disabled_by_default(): + bsc.set_blob_storage_config(bsc.BlobStorageConfig()) + reset_blob_store() + assert get_blob_store_if_enabled() is None + with pytest.raises(BlobNotConfiguredError): + get_blob_store() + + +def test_enabled_factory_resolves_folder_backend(tmp_path: Path): + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="local_fs", backend_config={"root": str(tmp_path / "b")})) + reset_blob_store() + store = get_blob_store_if_enabled() + assert isinstance(store, LocalFsBlobStore) + # Singleton: a second call returns the same instance. + assert get_blob_store() is store + + +def test_factory_replaces_store_when_root_changes(tmp_path: Path): + first_root = tmp_path / "first" + second_root = tmp_path / "second" + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(first_root)})) + + first_store = get_blob_store() + first_store.put_bytes(b"before", kind="tool-output") + + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(second_root)})) + second_store = get_blob_store_if_enabled() + assert second_store is not first_store + assert get_blob_store() is second_store + + after = second_store.put_bytes(b"after", kind="tool-output") + assert (second_root / after.kind / after.sha256[:2] / after.sha256).is_file() + assert not (first_root / after.kind / after.sha256[:2] / after.sha256).exists() + + +def test_factory_rejects_cached_store_after_disable(tmp_path: Path): + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(tmp_path)})) + get_blob_store() + + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=False)) + assert get_blob_store_if_enabled() is None + with pytest.raises(BlobNotConfiguredError): + get_blob_store() + + +def test_factory_does_not_keep_cached_store_for_invalid_backend(tmp_path: Path): + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(tmp_path)})) + get_blob_store() + + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="does_not_exist")) + with pytest.raises(ValueError, match="Unknown blob store backend"): + get_blob_store() + + +def test_replaced_store_stays_open_until_reset(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import deerflow.storage.manager as mgr + + closed: list[Path] = [] + + class TrackingStore(LocalFsBlobStore): + def close(self) -> None: + closed.append(self._root) + + monkeypatch.setattr(mgr, "_resolve_store_class", lambda backend: TrackingStore) + first_root = tmp_path / "first" + second_root = tmp_path / "second" + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(first_root)})) + first_store = get_blob_store() + + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend_config={"root": str(second_root)})) + get_blob_store() + assert closed == [] + first_store.put_bytes(b"in flight", kind="tool-output") + + reset_blob_store() + assert set(closed) == {first_root, second_root} + + +def test_unknown_backend_fails_fast(tmp_path: Path): + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="does_not_exist", backend_config={"root": str(tmp_path)})) + reset_blob_store() + with pytest.raises(ValueError, match="Unknown blob store backend"): + get_blob_store() + + +def test_dotted_import_path_backend(tmp_path: Path): + bsc.set_blob_storage_config( + bsc.BlobStorageConfig( + enabled=True, + backend="deerflow.storage.backends.local_fs:LocalFsBlobStore", + backend_config={"root": str(tmp_path / "d")}, + ) + ) + reset_blob_store() + assert isinstance(get_blob_store(), LocalFsBlobStore) + + +def test_relative_root_is_resolved_against_runtime_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import deerflow.storage.manager as mgr + + monkeypatch.setattr(mgr, "_default_backend_config", lambda: {"root": str(tmp_path / "state")}) + bsc.set_blob_storage_config(bsc.BlobStorageConfig(enabled=True, backend="local_fs", backend_config={"root": "relative"})) + reset_blob_store() + store = get_blob_store() + assert Path(str(store._root)).is_absolute() diff --git a/config.example.yaml b/config.example.yaml index b7f06ab58c..bc09520b4b 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -20,7 +20,7 @@ # ============================================================================ # Bump this number when the config schema changes. # Run `make config-upgrade` to merge new fields into your local config.yaml. -config_version: 46 +config_version: 47 # Optional per-model request pacing (inside a models[] entry): # request_admission: @@ -2262,6 +2262,33 @@ memory: # patterns_dir: "" # empty = bundled defaults # prompts_dir: "" # empty = bundled defaults +# ============================================================================ +# Blob Storage Configuration +# ============================================================================ +# Content-addressed blob store: content is addressed by its sha256, so writes +# are idempotent, identical bytes dedupe to one object, and reads verify the +# digest. Mirrors the memory mechanism: a `backend` name matches a +# storage/backends// folder that exposes STORE_CLASS (e.g. local_fs), or +# takes a dotted import path to a BlobStore subclass. The factory resolves the +# backend at get_blob_store() time and fails fast on an unresolvable value -- +# blobs are persistent state, so a broken backend name must not silently swap +# the storage location. +# +# Off by default and purely opt-in: no producer is migrated yet, so a config +# that never sets this key behaves exactly as before. Contract, the two named +# producers, and the migration order: backend/docs/blob-storage.md. +blob_storage: + enabled: false # master switch (inert until the producer follow-ups land) + backend: local_fs # folder name under storage/backends/, or a dotted import path + # Backend-private config (dict), passed verbatim to the backend; each backend + # self-interprets it. local_fs reads `root` -- empty defaults to + # {runtime_home}/blobs; point it at a shared volume (NFS / EFS / a + # ReadWriteMany PVC) for multi-instance deployments: + # + # backend_config: + # root: /mnt/shared/deerflow-blobs + backend_config: {} + # ============================================================================ # Custom Agent Management API # ============================================================================ diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 8eabdfa8dd..06773181e8 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -135,7 +135,7 @@ they resolve from the `secrets` map): ```yaml config: | - config_version: 46 + config_version: 47 models: - name: gpt-4 use: langchain_openai:ChatOpenAI diff --git a/deploy/helm/deer-flow/values.yaml b/deploy/helm/deer-flow/values.yaml index 50472404ac..518fad2e79 100644 --- a/deploy/helm/deer-flow/values.yaml +++ b/deploy/helm/deer-flow/values.yaml @@ -249,7 +249,7 @@ ingress: # -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never # inline literal secret values here. The default enables provisioner sandbox. config: | - config_version: 46 + config_version: 47 log_level: info recursion_limit: 100 max_recursion_limit: 1000