diff --git a/.env.example b/.env.example index 3f27fc7c9..d1ed25e28 100644 --- a/.env.example +++ b/.env.example @@ -25,7 +25,7 @@ INFOQUEST_API_KEY=your-infoquest-api-key # FIRECRAWL_API_KEY=your-firecrawl-api-key # VOLCENGINE_API_KEY=your-volcengine-api-key # OPENAI_API_KEY=your-openai-api-key -# OPENVIKING_API_KEY=your-openviking-trusted-root-api-key +# OPENVIKING_API_KEY=your-openviking-user-api-key # GEMINI_API_KEY=your-gemini-api-key # DEEPSEEK_API_KEY=your-deepseek-api-key # NOVITA_API_KEY=your-novita-api-key # OpenAI-compatible, see https://novita.ai diff --git a/README.md b/README.md index 958cd4e53..75ef0b8e6 100644 --- a/README.md +++ b/README.md @@ -1028,16 +1028,14 @@ request the binary capability retain the legacy JSON/base64 frame protocol. Most agents forget everything the moment a conversation ends. DeerFlow remembers. -DeerFlow also includes an optional `openviking` memory backend. It connects to -an independent OpenViking server over HTTP, submits completed turns through -OpenViking Sessions, and recalls remote memories for prompt injection while -leaving DeerMem as the default. The initial integration supports -`memory.mode: middleware`. Bounded submitted-message watermarks cover long and -compacted histories and prevent a failed Session commit from duplicating -already accepted messages on retry; the shared HTTP client also has explicit -connection limits and jittered retries. See -[OpenViking memory backend](docs/OPENVIKING.md) for configuration and Docker -startup. +DeerFlow also includes an optional `openviking` memory backend. It uses the +official `langchain-openviking` package to capture completed turns into stable +OpenViking Sessions and recall memory for prompt injection while leaving +DeerMem as the default. The initial integration supports one DeerFlow user with +one credential-bound OpenViking USER API key in `memory.mode: middleware` and +does not inherit arbitrary HTTP headers from `ovcli.conf`. +See [OpenViking memory backend](docs/OPENVIKING.md) for its configuration, +behavior, and current boundaries. Across sessions, DeerFlow builds a persistent memory of your profile, preferences, and accumulated knowledge. The more you use it, the better it knows you — your writing style, your technical stack, your recurring workflows. Memory is stored locally and stays under your control. diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 566baca45..b179fe6fc 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -271,8 +271,8 @@ Blocking-IO runtime gate (`tests/blocking_io/`): skips redundant sandbox sync when thread data is already mounted); `test_channel_outbound_files.py` (locks Feishu, Telegram, and WeCom outbound attachment open/read/hash work off the event loop); - `test_openviking_memory_backend.py` (locks the OpenViking backend's async - add/context/search entrypoints offloading synchronous HTTP and watermark + `test_openviking_memory_backend.py` (locks the official OpenViking backend's + async add/context/search entrypoints offloading synchronous SDK and cursor filesystem IO); and `test_workspace_changes_recorder.py` (locks the offload around the snapshot text cache lifecycle — roots resolution, `mkdtemp`, and the `shutil.rmtree` @@ -1014,26 +1014,26 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_ - `memory.mode: middleware` (default) keeps the passive path: `MemoryMiddleware` filters messages (user inputs + final AI responses), captures `user_id` via `resolve_runtime_user_id(runtime)`, queues conversation with the captured `user_id`, and the debounced background thread invokes the LLM to extract context updates and facts using the stored `user_id`. `DynamicContextMiddleware` passes the same resolved identity to the memory read path. On standalone Agent Server runs, server-owned auth identity is also resolved during lead-agent construction, normalized through `make_safe_user_id` for DeerFlow storage, and explicitly reused for custom-agent config/SOUL, user skills, skill policy, and prompt assembly; ordinary client `user_id` values cannot override `langgraph_auth_user_id`. On the embedded Gateway path, `inject_authenticated_user_context` removes client-supplied `langgraph_auth_user` / `langgraph_auth_user_id` from both RunnableConfig sections before graph construction, so those reserved fields cannot impersonate Agent Server auth. - The optional `openviking` backend under `packages/harness/deerflow/agents/memory/backends/openviking/` is a - remote-only HTTP adapter. Select it with + remote-only adapter built on the maintained `langchain-openviking` package. + Select it with `memory.manager_class: openviking` and keep `memory.mode: middleware`. It - commits filtered turns to OpenViking Sessions and maps remote memory search - results into the shared contract. It hashes `(user_id, agent_name)` into a - safe OpenViking trusted-user identity for hard scope isolation and keeps - bounded message watermarks below - `{storage_path}/openviking/sessions/`. The watermark combines a constant-size - ordered-prefix digest for append-only histories with a bounded recent-ID - fallback for compaction and separately records submitted and committed - progress. Once batch submission succeeds, a later update never resubmits - those messages or retries an ambiguous commit; a future batch can commit the - still-open Session together with new messages. Schema-v2 recent-ID - watermarks migrate without duplicating their anchored history. The shared - HTTP client has explicit total/keep-alive connection limits and jittered - exponential retry delays, and its configuration representation omits the API - key. Session locks are weakly cached, async entrypoints offload synchronous - HTTP and file IO, and graceful shutdown rejects new work before draining all - in-flight client operations within its timeout. It does not implement - DeerMem fact CRUD/import/export and must not import the OpenViking embedded - runtime. + uses one OpenViking USER API key bound to the configured DeerFlow + `owner_user_id`; another DeerFlow user is rejected before remote access. + DeerFlow owns the existing recall/capture timing, fixed injection query and + full-transcript suffix cursor. `langchain-openviking` owns SDK transport, + message conversion, tool-call preservation, batching, partial-write progress + and Session commits. One DeerFlow thread maps to one stable OpenViking + Session, with the default or named agent represented as its actor peer. + Bounded hash-only cursors live below `{storage_path}/openviking/sessions/`; + session locks are weakly cached, async entrypoints offload synchronous SDK + and file IO, and graceful shutdown drains active operations before closing + the recorder-owned client. The recorder receives an explicit empty + `extra_headers` mapping so `ovcli.conf` cannot add arbitrary transport + headers. Do not reintroduce a backend-local HTTP client, + explicitly configured trusted identity headers, root-key data access, or + imports of the OpenViking embedded runtime. Multi-user provisioning, + query-aware refresh policy and new lifecycle scheduling are separate changes, + not part of this backend. - `memory.mode: tool` skips `MemoryMiddleware` and registers `memory_search`, `memory_add`, `memory_update`, and `memory_delete` on the agent. The model decides when to search, add, update, or delete facts; this is opt-in/experimental and should not be described as better than middleware mode without eval evidence. - Both modes share `FileMemoryStorage`, per-user/per-agent isolation, manual CRUD primitives, and the updater backend. Injection is mode-aware: middleware mode injects global `user`/`history` summaries plus the selected agent's facts, while tool mode injects only the global summaries and leaves every agent fact behind `memory_search` to avoid duplicating automatically injected and retrieval-returned context. `memory.injection_enabled: false` suppresses the complete block in either mode. - Middleware extraction classifies proposed facts with extraction-only `scope`/`durability`/`authority` labels. `_apply_updates` accepts only `user` + `durable` + `descriptive` new/consolidated facts, accepts only wholly user-scoped summary prose with `authority=descriptive`, and rejects missing labels per item without aborting unrelated updates. Contradiction removals use object entries with `id`, `scope`, `reason`, and optional zero-based `replacementFactIndex`; task/project removals fail closed, and a paired removal runs only when the referenced replacement survives the scope/confidence gates, deduplication, and max-fact trim under another fact ID. The labels are not persisted, so no storage migration is required. Staleness removals retain their independent candidate/cap guardrails, while tool-mode CRUD remains outside this extraction gate. Custom `memory.backend_config.prompts_dir` templates (including per-agent overrides) must carry the same classification fields; an un-migrated template makes the fail-closed gate reject every extraction-driven write, observable only through `rejected_by_scope_gate` and the >60% fact-rejection warning. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/README.md b/backend/packages/harness/deerflow/agents/memory/backends/README.md index 861762831..e3a9e7c54 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/README.md +++ b/backend/packages/harness/deerflow/agents/memory/backends/README.md @@ -4,7 +4,8 @@ Each subfolder under `agents/memory/backends/` is a pluggable memory backend. Sw - `deermem/` - the default backend (deer-flow's own: structured facts + JSON storage). - `noop/` - an empty backend and the **template** to copy when adding a new one. -- `openviking/` - optional remote OpenViking backend over HTTP (middleware mode). +- `openviking/` - optional remote backend using the official + `langchain-openviking` package (single-user middleware mode). This guide tells you **which files to touch** when you change, swap, or add a memory system. Paths are relative to `backend/` unless noted. diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py index 06feb3998..26a3c8040 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/__init__.py @@ -1,4 +1,4 @@ -"""OpenViking HTTP memory backend.""" +"""OpenViking memory backend using the official LangChain integration.""" from .openviking_manager import OpenVikingMemoryManager diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/client.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/client.py deleted file mode 100644 index 46fdcfdbf..000000000 --- a/backend/packages/harness/deerflow/agents/memory/backends/openviking/client.py +++ /dev/null @@ -1,247 +0,0 @@ -"""Thin synchronous HTTP client for the OpenViking server API.""" - -from __future__ import annotations - -import random -import time -from typing import Any - -import httpx - -from .config import OpenVikingConfig -from .models import OpenVikingCommitResult, OpenVikingIdentity, OpenVikingMessage, OpenVikingSearchHit, OpenVikingSessionContext - - -class OpenVikingClientError(RuntimeError): - """Base error raised by the remote OpenViking adapter.""" - - def __init__(self, operation: str, message: str, *, status_code: int | None = None, code: str | None = None): - super().__init__(message) - self.operation = operation - self.status_code = status_code - self.code = code - - -class OpenVikingAuthenticationError(OpenVikingClientError): - pass - - -class OpenVikingTimeoutError(OpenVikingClientError): - pass - - -class OpenVikingUnavailableError(OpenVikingClientError): - pass - - -class OpenVikingProtocolError(OpenVikingClientError): - pass - - -class OpenVikingHttpClient: - """OpenViking API wrapper with bounded timeout and conservative retries.""" - - def __init__(self, config: OpenVikingConfig, *, transport: httpx.BaseTransport | None = None): - self._config = config - timeout = httpx.Timeout( - connect=config.connect_timeout_seconds, - read=config.read_timeout_seconds, - write=config.write_timeout_seconds, - pool=config.pool_timeout_seconds, - ) - limits = httpx.Limits( - max_connections=config.max_connections, - max_keepalive_connections=config.max_keepalive_connections, - ) - self._client = httpx.Client( - base_url=config.base_url, - timeout=timeout, - limits=limits, - transport=transport, - ) - - def close(self) -> None: - self._client.close() - - def health(self) -> bool: - response = self._request("health", "GET", "/health", identity=None, retryable=True) - if response.status_code != 200: - return False - try: - payload = response.json() - except ValueError: - return False - return payload.get("status") == "ok" - - def ensure_session(self, identity: OpenVikingIdentity, session_id: str) -> None: - self._request_result( - "session.ensure", - "GET", - f"/api/v1/sessions/{session_id}", - identity=identity, - params={"auto_create": "true"}, - retryable=True, - ) - - def add_messages(self, identity: OpenVikingIdentity, session_id: str, messages: list[OpenVikingMessage]) -> int: - if not messages: - return 0 - added = 0 - # OpenViking caps one batch at 100 messages. - for offset in range(0, len(messages), 100): - batch = messages[offset : offset + 100] - result = self._request_result( - "messages.add", - "POST", - f"/api/v1/sessions/{session_id}/messages/batch", - identity=identity, - json={"messages": [message.as_request() for message in batch]}, - retryable=False, - ) - added += int(result.get("added", len(batch))) - return added - - def commit_session(self, identity: OpenVikingIdentity, session_id: str) -> OpenVikingCommitResult: - result = self._request_result( - "session.commit", - "POST", - f"/api/v1/sessions/{session_id}/commit", - identity=identity, - json={"keep_recent_count": 0}, - retryable=False, - ) - return OpenVikingCommitResult( - status=str(result.get("status") or ""), - task_id=str(result["task_id"]) if result.get("task_id") else None, - archive_uri=str(result["archive_uri"]) if result.get("archive_uri") else None, - archived=bool(result.get("archived", False)), - ) - - def search( - self, - identity: OpenVikingIdentity, - query: str, - *, - top_k: int, - category: str | None = None, - session_id: str | None = None, - ) -> list[OpenVikingSearchHit]: - body: dict[str, Any] = { - "query": query, - "target_uri": "viking://user/memories", - "context_type": "memory", - "node_limit": top_k, - } - if session_id: - body["session_id"] = session_id - if self._config.score_threshold is not None: - body["score_threshold"] = self._config.score_threshold - result = self._request_result( - "search", - "POST", - "/api/v1/search/search" if session_id else "/api/v1/search/find", - identity=identity, - json=body, - retryable=True, - ) - values = result.get("memories", []) - if not isinstance(values, list): - raise OpenVikingProtocolError("search", "OpenViking response field result.memories is not a list") - hits = [OpenVikingSearchHit.from_response(value) for value in values if isinstance(value, dict)] - if category: - category_key = category.casefold() - hits = [hit for hit in hits if hit.category.casefold() == category_key] - return hits[:top_k] - - def get_session_context( - self, - identity: OpenVikingIdentity, - session_id: str, - *, - token_budget: int, - ) -> OpenVikingSessionContext: - result = self._request_result( - "session.context", - "GET", - f"/api/v1/sessions/{session_id}/context", - identity=identity, - params={"token_budget": token_budget}, - retryable=True, - ) - messages = result.get("messages", []) - return OpenVikingSessionContext( - latest_archive_overview=str(result.get("latest_archive_overview") or ""), - messages=messages if isinstance(messages, list) else [], - estimated_tokens=int(result.get("estimatedTokens") or 0), - ) - - def _headers(self, identity: OpenVikingIdentity | None) -> dict[str, str]: - headers = {"Accept": "application/json"} - if self._config.api_key: - headers["X-API-Key"] = self._config.api_key - if identity is not None and self._config.auth_mode == "trusted": - headers["X-OpenViking-Account"] = identity.account - headers["X-OpenViking-User"] = identity.user - return headers - - def _request_result(self, operation: str, method: str, path: str, *, identity: OpenVikingIdentity, retryable: bool, **kwargs: Any) -> dict[str, Any]: - response = self._request(operation, method, path, identity=identity, retryable=retryable, **kwargs) - try: - payload = response.json() - except ValueError as exc: - raise OpenVikingProtocolError(operation, "OpenViking returned non-JSON data", status_code=response.status_code) from exc - if not isinstance(payload, dict) or payload.get("status") != "ok": - error = payload.get("error", {}) if isinstance(payload, dict) else {} - code = str(error.get("code") or "UNKNOWN") if isinstance(error, dict) else "UNKNOWN" - message = str(error.get("message") or "OpenViking request failed") if isinstance(error, dict) else "OpenViking request failed" - error_type = OpenVikingAuthenticationError if response.status_code in {401, 403} else OpenVikingProtocolError - raise error_type(operation, message, status_code=response.status_code, code=code) - result = payload.get("result") - if not isinstance(result, dict): - raise OpenVikingProtocolError(operation, "OpenViking response field result is not an object", status_code=response.status_code) - return result - - def _request( - self, - operation: str, - method: str, - path: str, - *, - identity: OpenVikingIdentity | None, - retryable: bool, - **kwargs: Any, - ) -> httpx.Response: - attempts = self._config.max_retries + 1 if retryable else 1 - for attempt in range(attempts): - try: - response = self._client.request(method, path, headers=self._headers(identity), **kwargs) - except httpx.TimeoutException as exc: - if attempt + 1 < attempts: - time.sleep(_retry_delay(attempt)) - continue - raise OpenVikingTimeoutError(operation, "OpenViking request timed out") from exc - except httpx.TransportError as exc: - if attempt + 1 < attempts: - time.sleep(_retry_delay(attempt)) - continue - raise OpenVikingUnavailableError(operation, "OpenViking is unavailable") from exc - if response.status_code in {429, 502, 503, 504} and attempt + 1 < attempts: - time.sleep(_retry_delay(attempt)) - continue - if response.status_code >= 400: - try: - payload = response.json() - except ValueError: - payload = {} - error = payload.get("error", {}) if isinstance(payload, dict) else {} - code = str(error.get("code") or "HTTP_ERROR") if isinstance(error, dict) else "HTTP_ERROR" - message = str(error.get("message") or f"OpenViking HTTP {response.status_code}") if isinstance(error, dict) else f"OpenViking HTTP {response.status_code}" - error_type = OpenVikingAuthenticationError if response.status_code in {401, 403} else OpenVikingProtocolError - raise error_type(operation, message, status_code=response.status_code, code=code) - return response - raise OpenVikingUnavailableError(operation, "OpenViking request failed") - - -def _retry_delay(attempt: int) -> float: - base_delay = 0.05 * (2**attempt) - return base_delay + random.uniform(0.0, base_delay) diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py index 3225ad6e7..d29296c5b 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/config.py @@ -1,69 +1,84 @@ -"""Configuration for the OpenViking HTTP memory backend.""" +"""Validated configuration for the official OpenViking memory adapter.""" from __future__ import annotations import os +import re from dataclasses import dataclass, field +from math import isfinite from typing import Any, Literal from urllib.parse import urlparse +_SAFE_PEER_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") +GENERATED_PEER_PREFIX = "df-agent-" +_REMOVED_CUSTOM_HTTP_FIELDS = frozenset( + { + "connect_timeout_seconds", + "max_connections", + "max_keepalive_connections", + "max_retries", + "pool_timeout_seconds", + "read_timeout_seconds", + "write_timeout_seconds", + } +) -@dataclass(frozen=True) + +@dataclass(frozen=True, slots=True) class OpenVikingConfig: - """Parsed backend-private configuration. - - The backend is intentionally remote-only: DeerFlow talks to an independent - OpenViking server and does not import the OpenViking Python runtime. - """ + """Credential-bound connection settings and existing DeerFlow policy.""" base_url: str storage_path: str - auth_mode: Literal["trusted", "dev"] - account: str - api_key: str | None = field(repr=False) - connect_timeout_seconds: float - read_timeout_seconds: float - write_timeout_seconds: float - pool_timeout_seconds: float - max_connections: int - max_keepalive_connections: int - max_retries: int + owner_user_id: str + api_key: str = field(repr=False) + api_key_env: str + default_peer_id: str + timeout_seconds: float search_top_k: int score_threshold: float | None max_injection_chars: int + content_mode: Literal["auto", "abstract", "overview", "read"] injection_query: str startup_policy: Literal["fail_fast", "warn"] read_failure_policy: Literal["fail_open", "raise"] write_failure_policy: Literal["log_and_drop", "raise"] allow_insecure_http: bool - allow_insecure_dev: bool max_seen_message_ids: int @classmethod - def from_backend_config(cls, backend_config: dict[str, Any] | None) -> OpenVikingConfig: + def from_backend_config( + cls, + backend_config: dict[str, Any] | None, + ) -> OpenVikingConfig: cfg = dict(backend_config or {}) - retrieval = _mapping(cfg.pop("retrieval", {}), "retrieval") - failure_policy = _mapping(cfg.pop("failure_policy", {}), "failure_policy") + if "auth_mode" in cfg or "account" in cfg: + raise ValueError("OpenViking trusted mode is no longer supported by this backend; use a USER API key and configure owner_user_id") + removed_fields = sorted(_REMOVED_CUSTOM_HTTP_FIELDS.intersection(cfg)) + if removed_fields: + raise ValueError("OpenViking custom HTTP client fields are no longer supported: " + ", ".join(removed_fields)) + retrieval = _mapping(cfg.pop("retrieval", {}), "retrieval") + failure_policy = _mapping( + cfg.pop("failure_policy", {}), + "failure_policy", + ) api_key_env = str(cfg.pop("api_key_env", "OPENVIKING_API_KEY")).strip() - api_key = os.environ.get(api_key_env) if api_key_env else None + if not api_key_env: + raise ValueError("OpenViking api_key_env must not be empty") result = cls( base_url=str(cfg.pop("base_url", "http://127.0.0.1:1933")).rstrip("/"), storage_path=str(cfg.pop("storage_path", "")), - auth_mode=str(cfg.pop("auth_mode", "trusted")).lower(), # type: ignore[arg-type] - account=str(cfg.pop("account", "deerflow")).strip(), - api_key=api_key, - connect_timeout_seconds=float(cfg.pop("connect_timeout_seconds", 2.0)), - read_timeout_seconds=float(cfg.pop("read_timeout_seconds", 10.0)), - write_timeout_seconds=float(cfg.pop("write_timeout_seconds", 10.0)), - pool_timeout_seconds=float(cfg.pop("pool_timeout_seconds", 2.0)), - max_connections=int(cfg.pop("max_connections", 100)), - max_keepalive_connections=int(cfg.pop("max_keepalive_connections", 20)), - max_retries=int(cfg.pop("max_retries", 1)), + owner_user_id=str(cfg.pop("owner_user_id", "")).strip(), + api_key=os.environ.get(api_key_env, "").strip(), + api_key_env=api_key_env, + default_peer_id=str(cfg.pop("default_peer_id", "deerflow")).strip(), + timeout_seconds=float(cfg.pop("timeout_seconds", 30.0)), search_top_k=int(retrieval.pop("top_k", 8)), score_threshold=_optional_float(retrieval.pop("score_threshold", None)), max_injection_chars=int(retrieval.pop("max_injection_chars", 12_000)), + content_mode=str(retrieval.pop("content_mode", "overview")).lower(), # type: ignore[arg-type] injection_query=str( retrieval.pop( "injection_query", @@ -73,14 +88,22 @@ class OpenVikingConfig: startup_policy=str(cfg.pop("startup_policy", "fail_fast")).lower(), # type: ignore[arg-type] read_failure_policy=str(failure_policy.pop("read", "fail_open")).lower(), # type: ignore[arg-type] write_failure_policy=str(failure_policy.pop("write", "log_and_drop")).lower(), # type: ignore[arg-type] - allow_insecure_http=bool(cfg.pop("allow_insecure_http", False)), - allow_insecure_dev=bool(cfg.pop("allow_insecure_dev", False)), + allow_insecure_http=_boolean( + cfg.pop("allow_insecure_http", False), + "allow_insecure_http", + ), max_seen_message_ids=int(cfg.pop("max_seen_message_ids", 512)), ) - unknown = sorted([*cfg, *(f"retrieval.{key}" for key in retrieval), *(f"failure_policy.{key}" for key in failure_policy)]) + unknown = sorted( + [ + *cfg, + *(f"retrieval.{key}" for key in retrieval), + *(f"failure_policy.{key}" for key in failure_policy), + ] + ) if unknown: - raise ValueError(f"Unknown OpenViking backend_config fields: {', '.join(unknown)}") + raise ValueError("Unknown OpenViking backend_config fields: " + ", ".join(unknown)) result._validate() return result @@ -89,28 +112,25 @@ class OpenVikingConfig: if parsed.scheme not in {"http", "https"} or not parsed.netloc: raise ValueError("OpenViking base_url must be an absolute http(s) URL") if parsed.scheme == "http" and not self.allow_insecure_http and parsed.hostname not in {"127.0.0.1", "localhost", "openviking"}: - raise ValueError("OpenViking plain HTTP is allowed only for localhost/openviking; set allow_insecure_http=true for a trusted internal network") - if self.auth_mode not in {"trusted", "dev"}: - raise ValueError("OpenViking auth_mode must be 'trusted' or 'dev'") - if self.auth_mode == "dev" and not self.allow_insecure_dev: - raise ValueError("OpenViking auth_mode='dev' requires allow_insecure_dev=true") - if self.auth_mode == "trusted" and not self.account: - raise ValueError("OpenViking trusted auth requires a non-empty account") - for field_name in ("connect_timeout_seconds", "read_timeout_seconds", "write_timeout_seconds", "pool_timeout_seconds"): - if getattr(self, field_name) <= 0: - raise ValueError(f"OpenViking {field_name} must be > 0") - if not 1 <= self.max_connections <= 1000: - raise ValueError("OpenViking max_connections must be between 1 and 1000") - if not 0 <= self.max_keepalive_connections <= self.max_connections: - raise ValueError("OpenViking max_keepalive_connections must be between 0 and max_connections") - if not 0 <= self.max_retries <= 5: - raise ValueError("OpenViking max_retries must be between 0 and 5") + raise ValueError("OpenViking plain HTTP is allowed only for localhost/openviking; set allow_insecure_http=true only for a trusted internal network") + if not self.owner_user_id: + raise ValueError("OpenViking owner_user_id must not be empty") + if not self.api_key: + raise ValueError(f"OpenViking USER API key is missing; set {self.api_key_env}") + if not is_safe_peer_id(self.default_peer_id): + raise ValueError("OpenViking default_peer_id must start with a lowercase letter or digit and contain at most 64 lowercase letters, digits, '_' or '-'") + if self.default_peer_id.startswith(GENERATED_PEER_PREFIX): + raise ValueError(f"OpenViking default_peer_id must not start with the reserved prefix {GENERATED_PEER_PREFIX!r}") + if not isfinite(self.timeout_seconds) or self.timeout_seconds <= 0: + raise ValueError("OpenViking timeout_seconds must be a finite value > 0") if not 1 <= self.search_top_k <= 100: raise ValueError("OpenViking retrieval.top_k must be between 1 and 100") - if self.score_threshold is not None and not 0 <= self.score_threshold <= 1: - raise ValueError("OpenViking retrieval.score_threshold must be between 0 and 1") + if self.score_threshold is not None and (not isfinite(self.score_threshold) or not 0 <= self.score_threshold <= 1): + raise ValueError("OpenViking retrieval.score_threshold must be a finite value between 0 and 1") if not 256 <= self.max_injection_chars <= 100_000: raise ValueError("OpenViking retrieval.max_injection_chars must be between 256 and 100000") + if self.content_mode not in {"auto", "abstract", "overview", "read"}: + raise ValueError("OpenViking retrieval.content_mode must be auto, abstract, overview, or read") if not self.injection_query: raise ValueError("OpenViking retrieval.injection_query must not be empty") if self.startup_policy not in {"fail_fast", "warn"}: @@ -123,6 +143,12 @@ class OpenVikingConfig: raise ValueError("OpenViking max_seen_message_ids must be between 16 and 10000") +def is_safe_peer_id(value: str) -> bool: + """Return whether *value* is valid for an OpenViking actor peer.""" + + return _SAFE_PEER_RE.fullmatch(value) is not None + + def _mapping(value: Any, name: str) -> dict[str, Any]: if value is None: return {} @@ -133,3 +159,15 @@ def _mapping(value: Any, name: str) -> dict[str, Any]: def _optional_float(value: Any) -> float | None: return None if value is None else float(value) + + +def _boolean(value: Any, name: str) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in {"true", "1", "yes", "on"}: + return True + if normalized in {"false", "0", "no", "off"}: + return False + raise ValueError(f"OpenViking {name} must be a boolean") diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/models.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/models.py deleted file mode 100644 index a86fbb4b8..000000000 --- a/backend/packages/harness/deerflow/agents/memory/backends/openviking/models.py +++ /dev/null @@ -1,63 +0,0 @@ -"""Small transport-neutral models used by the OpenViking adapter.""" - -from __future__ import annotations - -from dataclasses import dataclass -from typing import Any - - -@dataclass(frozen=True) -class OpenVikingIdentity: - account: str - user: str - - -@dataclass(frozen=True) -class OpenVikingMessage: - message_id: str - role: str - content: str - - def as_request(self) -> dict[str, Any]: - # OpenViking AddMessageRequest generates its own message ID and rejects - # unknown request fields, so the DeerFlow ID remains adapter-local for - # watermarking and is not sent over the wire. - return {"role": self.role, "content": self.content} - - -@dataclass(frozen=True) -class OpenVikingCommitResult: - status: str - task_id: str | None - archive_uri: str | None - archived: bool - - -@dataclass(frozen=True) -class OpenVikingSearchHit: - uri: str - context_type: str - category: str - score: float - abstract: str - overview: str | None - match_reason: str - - @classmethod - def from_response(cls, value: dict[str, Any]) -> OpenVikingSearchHit: - return cls( - uri=str(value.get("uri") or ""), - context_type=str(value.get("context_type") or "memory"), - category=str(value.get("category") or "memory"), - score=float(value.get("score") or 0.0), - abstract=str(value.get("abstract") or ""), - overview=str(value["overview"]) if value.get("overview") else None, - match_reason=str(value.get("match_reason") or ""), - ) - - -@dataclass(frozen=True) -class OpenVikingSessionContext: - latest_archive_overview: str - messages: list[dict[str, Any]] - estimated_tokens: int diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py index 0bfeca100..13493d734 100644 --- a/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/openviking_manager.py @@ -1,59 +1,90 @@ -"""OpenViking HTTP implementation of the pluggable memory contract. - -This backend deliberately contains no OpenViking extraction or vector logic. -It forwards filtered conversation turns to OpenViking Sessions and maps remote -memory search results back to DeerFlow's backend-neutral shapes. -""" +"""OpenViking memory backend built on the maintained LangChain adapters.""" from __future__ import annotations import asyncio -import hashlib +import copy import json import logging import os -import re import threading import time import weakref -from collections.abc import Callable +from contextlib import AbstractContextManager from pathlib import Path from typing import Any, ClassVar, Literal from pydantic import PrivateAttr -# ABC contract -- the only DeerFlow import in this backend package. -from deerflow.agents.memory.manager import MemoryManager +from deerflow.agents.memory.manager import MemoryManager, MemoryManagerError -from .client import OpenVikingClientError, OpenVikingHttpClient from .config import OpenVikingConfig -from .models import OpenVikingIdentity, OpenVikingMessage, OpenVikingSearchHit +from .session import ( + _advanced_cursor, + _canonical_peer_id, + _captureable_messages, + _matching_prefix_count, + _memory_target_uris, + _message_signature, + _session_id, + _string_list, +) logger = logging.getLogger(__name__) -_DEFAULT_AGENT_SCOPE = "__default__" -_SESSION_NAMESPACE = "deerflow-openviking-v1" -_SAFE_SCOPE_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$") - class OpenVikingMemoryManager(MemoryManager): - """Remote OpenViking memory backend for passive middleware mode.""" + """Single-user OpenViking backend using the official integration package. + + DeerFlow continues to choose when capture and recall occur. The official + adapter owns SDK transport, message conversion, batching, commit retries, + and retrieval behavior. + """ supports_search: ClassVar[bool] = True _config: OpenVikingConfig = PrivateAttr() - _client: OpenVikingHttpClient = PrivateAttr() - _should_keep_hidden_message: Callable[[Any], bool] | None = PrivateAttr(default=None) + _client: Any = PrivateAttr() + _recorder: Any = PrivateAttr() + _retriever: Any = PrivateAttr() + _use_actor_peer: Any = PrivateAttr() + _partial_write_error: type[Exception] = PrivateAttr() + _should_keep_hidden_message: Any = PrivateAttr(default=None) _session_locks: weakref.WeakValueDictionary[str, threading.RLock] = PrivateAttr(default_factory=weakref.WeakValueDictionary) _session_locks_guard: threading.Lock = PrivateAttr(default_factory=threading.Lock) _lifecycle: threading.Condition = PrivateAttr(default_factory=threading.Condition) _active_operations: int = PrivateAttr(default=0) _closed: bool = PrivateAttr(default=False) - _client_closed: bool = PrivateAttr(default=False) + _close_requested: bool = PrivateAttr(default=False) + _resources_closed: bool = PrivateAttr(default=False) + _resource_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock) def model_post_init(self, __context: Any) -> None: self._config = OpenVikingConfig.from_backend_config(self.backend_config) - self._client = OpenVikingHttpClient(self._config) + integration = _load_official_integration() + commit_policy = integration["OpenVikingCommitPolicy"](mode="always") + self._recorder = integration["OpenVikingSessionRecorder"]( + url=self._config.base_url, + api_key=self._config.api_key, + timeout=self._config.timeout_seconds, + # An explicit mapping disables the SDK's ovcli.conf header fallback. + extra_headers={}, + commit_policy=commit_policy, + ) + # The recorder owns one recovery-aware SDK client. Retrieval borrows the + # same handle, so this backend has one connection pool and one owner. + self._client = self._recorder.client + self._retriever = integration["OpenVikingRetriever"]( + client=self._client, + search_mode="find", + limit=self._config.search_top_k, + score_threshold=self._config.score_threshold, + context_types=("memory",), + content_mode=self._config.content_mode, + max_content_chars=self._config.max_injection_chars, + ) + self._use_actor_peer = integration["use_actor_peer"] + self._partial_write_error = integration["OpenVikingPartialWriteError"] @classmethod def from_config( @@ -64,10 +95,10 @@ class OpenVikingMemoryManager(MemoryManager): **host_hooks: Any, ) -> OpenVikingMemoryManager: if mode != "middleware": - raise ValueError("The OpenViking HTTP backend currently supports memory.mode='middleware' only") - instance = cls(backend_config=backend_config, mode=mode) - hook = host_hooks.get("should_keep_hidden_message") - instance._should_keep_hidden_message = hook if callable(hook) else None + raise ValueError("The OpenViking automatic-memory backend supports memory.mode='middleware' only; use OpenViking MCP for explicit model tools") + instance = cls(backend_config=backend_config or {}, mode=mode) + hidden_filter = host_hooks.get("should_keep_hidden_message") + instance._should_keep_hidden_message = hidden_filter if callable(hidden_filter) else None return instance def add( @@ -79,7 +110,13 @@ class OpenVikingMemoryManager(MemoryManager): user_id: str | None = None, trace_id: str | None = None, ) -> None: - self._write_conversation(thread_id, messages, agent_name=agent_name, user_id=user_id) + del trace_id + self._write_conversation( + thread_id, + messages, + agent_name=agent_name, + user_id=user_id, + ) def add_nowait( self, @@ -89,7 +126,14 @@ class OpenVikingMemoryManager(MemoryManager): agent_name: str | None = None, user_id: str | None = None, ) -> None: - self._write_conversation(thread_id, messages, agent_name=agent_name, user_id=user_id) + # Preserve DeerFlow's existing pre-compaction behavior. This backend + # commits every accepted capture, so add_nowait needs no separate mode. + self._write_conversation( + thread_id, + messages, + agent_name=agent_name, + user_id=user_id, + ) async def aadd( self, @@ -119,21 +163,31 @@ class OpenVikingMemoryManager(MemoryManager): if not self._begin_operation(): return "" try: - try: - hits = self._search_hits( - self._config.injection_query, - top_k=self._config.search_top_k, - user_id=user_id, - agent_name=agent_name, - category=None, - thread_id=thread_id, + peer_id = self._resolve_scope(user_id, agent_name) + retriever = copy.copy(self._retriever) + retriever.target_uri = _memory_target_uris(peer_id) + if thread_id: + retriever.search_mode = "search" + retriever.session_id = _session_id( + self._config.owner_user_id, + peer_id, + thread_id, ) - except OpenVikingClientError: + try: + with self._actor_peer_scope(peer_id): + documents = retriever.invoke(self._config.injection_query) + except Exception: if self._config.read_failure_policy == "raise": raise - logger.warning("OpenViking context retrieval failed; continuing without injected memory", exc_info=True) + logger.warning( + "OpenViking context retrieval failed; continuing without injected memory", + exc_info=True, + ) return "" - return _format_context(hits, max_chars=self._config.max_injection_chars) + return _format_documents( + documents, + max_chars=self._config.max_injection_chars, + ) finally: self._end_operation() @@ -160,26 +214,33 @@ class OpenVikingMemoryManager(MemoryManager): agent_name: str | None = None, category: str | None = None, ) -> list[dict[str, Any]]: - if not query.strip(): - return [] - if not self._begin_operation(): + if not query.strip() or not self._begin_operation(): return [] try: + peer_id = self._resolve_scope(user_id, agent_name) + retriever = copy.copy(self._retriever) + retriever.target_uri = _memory_target_uris(peer_id) + retriever.search_mode = "find" + retriever.session_id = None + retriever.limit = max(1, min(int(top_k), 100)) + if category: + retriever.filter = { + "op": "must", + "field": "category", + "conds": [category], + } try: - hits = self._search_hits( - query, - top_k=top_k, - user_id=user_id, - agent_name=agent_name, - category=category, - thread_id=None, - ) - except OpenVikingClientError: + with self._actor_peer_scope(peer_id): + documents = retriever.invoke(query.strip()) + except Exception: if self._config.read_failure_policy == "raise": raise - logger.warning("OpenViking memory search failed; returning no results", exc_info=True) + logger.warning( + "OpenViking memory search failed; returning no results", + exc_info=True, + ) return [] - return [_hit_to_fact(hit) for hit in hits] + return [_document_to_fact(document) for document in documents] finally: self._end_operation() @@ -206,14 +267,18 @@ class OpenVikingMemoryManager(MemoryManager): return False try: try: - healthy = self._client.health() - except OpenVikingClientError: + health = getattr(self._client, "health", None) + healthy = bool(health()) if callable(health) else True + except Exception: if self._config.startup_policy == "fail_fast": raise - logger.warning("OpenViking health check failed; memory will run in degraded mode", exc_info=True) + logger.warning( + "OpenViking startup validation failed; memory will run in degraded mode", + exc_info=True, + ) return False if not healthy and self._config.startup_policy == "fail_fast": - raise RuntimeError("OpenViking health check returned an unhealthy response") + raise MemoryManagerError("OpenViking health check returned an unhealthy response") if not healthy: logger.warning("OpenViking health check returned unhealthy; memory will run in degraded mode") return healthy @@ -221,21 +286,30 @@ class OpenVikingMemoryManager(MemoryManager): self._end_operation() def shutdown_flush(self, timeout: float) -> bool: - """Stop new operations, drain in-flight work, then close the HTTP pool.""" + """Stop new work, drain accepted calls, and close owned resources.""" + deadline = time.monotonic() + max(0.0, timeout) with self._lifecycle: self._closed = True + self._close_requested = True while self._active_operations: remaining = deadline - time.monotonic() if remaining <= 0: return False self._lifecycle.wait(remaining) - if self._client_closed: - return True - self._client_closed = True - self._client.close() + self._close_resources() return True + def close(self) -> None: + """Request idempotent closure after any active operation finishes.""" + + with self._lifecycle: + self._closed = True + self._close_requested = True + can_close = self._active_operations == 0 + if can_close: + self._close_resources() + def _write_conversation( self, thread_id: str, @@ -250,116 +324,147 @@ class OpenVikingMemoryManager(MemoryManager): try: if not thread_id: raise ValueError("OpenViking memory write requires thread_id") - - identity = self._identity(user_id, agent_name) - session_id = _session_id(identity, thread_id) - lock = self._session_lock(session_id) - with lock: - state = self._load_state(session_id) - submitted_ids = list(state.get("submitted_message_ids", state.get("seen_message_ids", []))) - committed_ids = list(state.get("committed_message_ids", state.get("seen_message_ids", []))) - converted = _convert_messages(messages, self._should_keep_hidden_message) - prefix_count = _matching_submitted_prefix_count(state, submitted_ids, converted) - if prefix_count is not None: - pending = converted[prefix_count:] - else: - submitted = set(submitted_ids) - pending = [message for message in converted if message.message_id not in submitted] - if not pending: - if prefix_count is None and converted: - state = { - **state, - "schema_version": 3, - "submitted_prefix_count": len(converted), - "submitted_prefix_digest": _message_sequence_digest(converted), - } - self._save_state(session_id, state) - return - try: - self._client.ensure_session(identity, session_id) - self._client.add_messages(identity, session_id, pending) - except OpenVikingClientError: - if self._config.write_failure_policy == "raise": - raise - logger.error( - "OpenViking memory message submission failed; dropping this update (session=%s, messages=%d)", - session_id, - len(pending), - exc_info=True, - ) - return - submitted_ids.extend(message.message_id for message in pending) - state = { - "schema_version": 3, - "session_id": session_id, - "submitted_message_ids": submitted_ids[-self._config.max_seen_message_ids :], - "committed_message_ids": committed_ids[-self._config.max_seen_message_ids :], - "submitted_prefix_count": len(converted), - "submitted_prefix_digest": _message_sequence_digest(converted), - "committed_prefix_count": state.get("committed_prefix_count"), - "committed_prefix_digest": state.get("committed_prefix_digest"), - "last_commit_task_id": state.get("last_commit_task_id"), - "last_archive_uri": state.get("last_archive_uri"), - } - self._save_state(session_id, state) - - try: - commit = self._client.commit_session(identity, session_id) - except OpenVikingClientError: - if self._config.write_failure_policy == "raise": - raise - logger.error( - "OpenViking memory commit failed; preserving submitted watermark without retry (session=%s)", - session_id, - exc_info=True, - ) - return - - state = { - **state, - "schema_version": 3, - "committed_message_ids": state["submitted_message_ids"], - "committed_prefix_count": state["submitted_prefix_count"], - "committed_prefix_digest": state["submitted_prefix_digest"], - "last_commit_task_id": commit.task_id, - "last_archive_uri": commit.archive_uri, - } - self._save_state(session_id, state) + peer_id = self._resolve_scope(user_id, agent_name) + session_id = _session_id( + self._config.owner_user_id, + peer_id, + thread_id, + ) + with self._session_lock(session_id): + self._capture_locked( + session_id, + peer_id, + _captureable_messages( + messages, + self._should_keep_hidden_message, + ), + ) finally: self._end_operation() - def _search_hits( + def _capture_locked( self, - query: str, - *, - top_k: int, - user_id: str | None, - agent_name: str | None, - category: str | None, - thread_id: str | None, - ) -> list[OpenVikingSearchHit]: - identity = self._identity(user_id, agent_name) - session_id = _session_id(identity, thread_id) if thread_id else None - return self._client.search( - identity, - query, - top_k=max(1, min(top_k, 100)), - category=category, - session_id=session_id, + session_id: str, + peer_id: str, + messages: list[Any], + ) -> None: + state = self._load_cursor(session_id) + signatures = [_message_signature(message) for message in messages] + + if state.get("commit_pending"): + try: + with self._actor_peer_scope(peer_id): + self._recorder.flush(session_id) + except Exception as exc: + self._handle_write_error( + exc, + "OpenViking pending commit retry failed; preserving capture cursor", + session_id, + ) + return + state = {**state, "commit_pending": False} + self._save_cursor(session_id, state) + + start = _matching_prefix_count(state, signatures) + append_only = start is not None + if append_only: + pending = messages[start:] + pending_signatures = signatures[start:] + else: + submitted = set(_string_list(state.get("submitted_signatures"))) + pending_pairs = [(message, signature) for message, signature in zip(messages, signatures, strict=True) if signature not in submitted] + pending = [message for message, _ in pending_pairs] + pending_signatures = [signature for _, signature in pending_pairs] + + if not pending: + self._save_cursor( + session_id, + _advanced_cursor( + state, + signatures, + [], + max_seen=self._config.max_seen_message_ids, + commit_pending=False, + ), + ) + return + + try: + with self._actor_peer_scope(peer_id): + self._recorder.record( + session_id, + pending, + peer_id=peer_id, + ) + except self._partial_write_error as exc: + consumed = max( + 0, + min( + len(pending_signatures), + int(getattr(exc, "input_messages_consumed", 0)), + ), + ) + confirmed = pending_signatures[:consumed] + commit_pending = bool(getattr(exc, "commit_pending", False)) + if confirmed or commit_pending: + confirmed_prefix = signatures[: int(start or 0) + consumed] if append_only else None + self._save_cursor( + session_id, + _advanced_cursor( + state, + confirmed_prefix, + confirmed, + max_seen=self._config.max_seen_message_ids, + commit_pending=commit_pending, + ), + ) + self._handle_write_error( + exc, + "OpenViking partially recorded a conversation; confirmed progress was preserved", + session_id, + ) + return + except Exception as exc: + self._handle_write_error( + exc, + "OpenViking conversation recording failed; capture cursor was not advanced", + session_id, + ) + return + + self._save_cursor( + session_id, + _advanced_cursor( + state, + signatures, + pending_signatures, + max_seen=self._config.max_seen_message_ids, + commit_pending=False, + ), ) - def _identity(self, user_id: str | None, agent_name: str | None) -> OpenVikingIdentity: - raw_user = str(user_id or "anonymous") - agent_scope = _canonical_agent_scope(agent_name) - # OpenViking trusted identity must be a safe path segment. Hashing the - # DeerFlow scope also prevents raw emails/usernames from leaving the - # Gateway and gives each agent a hard-isolated memory namespace. - digest = hashlib.sha256(f"{self._config.account}\0{raw_user}\0{agent_scope}".encode()).hexdigest() - return OpenVikingIdentity(account=self._config.account, user=f"df_{digest[:40]}") + def _resolve_scope( + self, + user_id: str | None, + agent_name: str | None, + ) -> str: + resolved_user = str(user_id or "default") + if resolved_user != self._config.owner_user_id: + raise MemoryManagerError(f"OpenViking USER API key is bound to DeerFlow owner_user_id {self._config.owner_user_id!r}, but this request belongs to {resolved_user!r}. Refusing to share one credential across users.") + return _canonical_peer_id(agent_name, self._config.default_peer_id) + + def _actor_peer_scope( + self, + peer_id: str, + ) -> AbstractContextManager[None]: + return self._use_actor_peer(peer_id) def _session_lock(self, session_id: str) -> threading.RLock: with self._session_locks_guard: - return self._session_locks.setdefault(session_id, threading.RLock()) + return self._session_locks.setdefault( + session_id, + threading.RLock(), + ) def _begin_operation(self) -> bool: with self._lifecycle: @@ -369,172 +474,108 @@ class OpenVikingMemoryManager(MemoryManager): return True def _end_operation(self) -> None: + should_close = False with self._lifecycle: self._active_operations -= 1 if self._active_operations == 0: self._lifecycle.notify_all() + should_close = self._close_requested + if should_close: + try: + self._close_resources() + except Exception: + logger.exception("Failed to close OpenViking memory resources") + + def _close_resources(self) -> None: + with self._resource_lock: + if self._resources_closed: + return + self._recorder.close() + self._resources_closed = True def _state_path(self, session_id: str) -> Path: root = Path(self._config.storage_path or ".") / "openviking" / "sessions" return root / f"{session_id}.json" - def _load_state(self, session_id: str) -> dict[str, Any]: + def _load_cursor(self, session_id: str) -> dict[str, Any]: path = self._state_path(session_id) try: value = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: return {} - except (OSError, ValueError): - logger.warning("Ignoring unreadable OpenViking session watermark: %s", path, exc_info=True) - return {} - return value if isinstance(value, dict) else {} + except (OSError, ValueError) as exc: + raise MemoryManagerError(f"OpenViking capture cursor is unreadable; refusing unsafe replay (session={session_id})") from exc + if not isinstance(value, dict): + raise MemoryManagerError(f"OpenViking capture cursor is invalid; refusing unsafe replay (session={session_id})") + return value - def _save_state(self, session_id: str, state: dict[str, Any]) -> None: + def _save_cursor(self, session_id: str, state: dict[str, Any]) -> None: path = self._state_path(session_id) path.parent.mkdir(parents=True, exist_ok=True) temp_path = path.with_suffix(f".{os.getpid()}.{threading.get_ident()}.tmp") try: - temp_path.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") + temp_path.write_text( + json.dumps(state, ensure_ascii=False, indent=2), + encoding="utf-8", + ) os.replace(temp_path, path) finally: try: temp_path.unlink(missing_ok=True) except OSError: - logger.debug("Failed to remove OpenViking watermark temp file: %s", temp_path, exc_info=True) + logger.debug( + "Failed to remove OpenViking cursor temp file: %s", + temp_path, + exc_info=True, + ) + + def _handle_write_error( + self, + exc: Exception, + message: str, + session_id: str, + ) -> None: + detail = f"{message} (session={session_id})" + if self._config.write_failure_policy == "raise": + raise MemoryManagerError(detail) from exc + logger.error(detail, exc_info=True) -def _canonical_agent_scope(agent_name: str | None) -> str: - if agent_name is None: - return _DEFAULT_AGENT_SCOPE - value = str(agent_name).strip().lower() - if value == _DEFAULT_AGENT_SCOPE or not _SAFE_SCOPE_RE.fullmatch(value): - raise ValueError(f"Invalid OpenViking agent scope: {agent_name!r}") - return value - - -def _session_id(identity: OpenVikingIdentity, thread_id: str) -> str: - digest = hashlib.sha256(f"{_SESSION_NAMESPACE}\0{identity.account}\0{identity.user}\0{thread_id}".encode()).hexdigest() - return f"df_{digest[:48]}" - - -def _matching_submitted_prefix_count( - state: dict[str, Any], - submitted_ids: list[str], - messages: list[OpenVikingMessage], -) -> int | None: - count = state.get("submitted_prefix_count") - digest = state.get("submitted_prefix_digest") - if isinstance(count, int) and 0 <= count <= len(messages) and isinstance(digest, str): - if _message_sequence_digest(messages[:count]) == digest: - return count - return None - - # Schema v2 only retained a recent suffix of submitted IDs. When that - # suffix still appears intact, it safely anchors the append-only prefix and - # avoids a one-time duplicate submission during migration to schema v3. - if submitted_ids and len(submitted_ids) <= len(messages): - message_ids = [message.message_id for message in messages] - width = len(submitted_ids) - for start in range(len(message_ids) - width, -1, -1): - if message_ids[start : start + width] == submitted_ids: - return start + width - return None - - -def _message_sequence_digest(messages: list[OpenVikingMessage]) -> str: - digest = hashlib.sha256() - for message in messages: - encoded = message.message_id.encode() - digest.update(len(encoded).to_bytes(8, "big")) - digest.update(encoded) - return digest.hexdigest() - - -def _convert_messages( - messages: list[Any], - should_keep_hidden_message: Callable[[Any], bool] | None, -) -> list[OpenVikingMessage]: - converted: list[OpenVikingMessage] = [] - for index, message in enumerate(messages): - role = _message_role(message) - if role not in {"user", "assistant"}: - continue - additional_kwargs = _message_value(message, "additional_kwargs", {}) - if not isinstance(additional_kwargs, dict): - additional_kwargs = {} - if additional_kwargs.get("hide_from_ui") and not (should_keep_hidden_message and should_keep_hidden_message(additional_kwargs)): - continue - tool_calls = _message_value(message, "tool_calls", []) - if role == "assistant" and tool_calls: - continue - content = _text_content(_message_value(message, "content", "")) - if not content.strip(): - continue - native_id = _message_value(message, "id", None) - stable_id = str(native_id) if native_id else hashlib.sha256(f"{role}\0{index}\0{content}".encode()).hexdigest() - converted.append(OpenVikingMessage(message_id=f"df_{stable_id}", role=role, content=content.strip())) - return converted - - -def _message_role(message: Any) -> str | None: - value = _message_value(message, "type", None) or _message_value(message, "role", None) - if value in {"human", "user"}: - return "user" - if value in {"ai", "assistant"}: - return "assistant" - name = type(message).__name__.lower() - if "human" in name: - return "user" - if "ai" in name: - return "assistant" - return None - - -def _message_value(message: Any, key: str, default: Any) -> Any: - return message.get(key, default) if isinstance(message, dict) else getattr(message, key, default) - - -def _text_content(content: Any) -> str: - if isinstance(content, str): - return content - if not isinstance(content, list): - return str(content) if content is not None else "" - parts: list[str] = [] - for block in content: - if isinstance(block, str): - parts.append(block) - elif isinstance(block, dict) and block.get("type") in {"text", "input_text", "output_text"}: - text = block.get("text") - if text: - parts.append(str(text)) - return "\n".join(parts) - - -def _hit_content(hit: OpenVikingSearchHit) -> str: - return (hit.overview or hit.abstract).strip() - - -def _hit_to_fact(hit: OpenVikingSearchHit) -> dict[str, Any]: +def _load_official_integration() -> dict[str, Any]: + try: + from langchain_openviking import ( + OpenVikingCommitPolicy, + OpenVikingPartialWriteError, + OpenVikingRetriever, + OpenVikingSessionRecorder, + has_request_actor_peer_support, + ) + from langchain_openviking.actor_peer import use_actor_peer + except ImportError as exc: + raise ImportError("The OpenViking memory backend requires langchain-openviking==0.1.0. Install DeerFlow backend dependencies and retry.") from exc + if not has_request_actor_peer_support(): + raise ImportError("The installed OpenViking SDK lacks request-scoped actor-peer support. Install openviking-sdk>=0.1.6,<0.2 and retry.") return { - "id": hit.uri, - "content": _hit_content(hit), - "category": hit.category or "memory", - "confidence": hit.score, - "source": hit.uri, - "score": hit.score, + "OpenVikingCommitPolicy": OpenVikingCommitPolicy, + "OpenVikingPartialWriteError": OpenVikingPartialWriteError, + "OpenVikingRetriever": OpenVikingRetriever, + "OpenVikingSessionRecorder": OpenVikingSessionRecorder, + "use_actor_peer": use_actor_peer, } -def _format_context(hits: list[OpenVikingSearchHit], *, max_chars: int) -> str: +def _format_documents(documents: list[Any], *, max_chars: int) -> str: lines: list[str] = [] seen: set[str] = set() - for hit in hits: - content = " ".join(_hit_content(hit).split()) + for document in documents: + content = " ".join(str(getattr(document, "page_content", "") or "").split()) key = content.casefold() if not content or key in seen: continue seen.add(key) - line = f"- [{hit.category or 'memory'}] {content}" + metadata = getattr(document, "metadata", {}) or {} + category = metadata.get("openviking_category") or "memory" + line = f"- [{category}] {content}" candidate = "\n".join([*lines, line]) if len(candidate) > max_chars: remaining = max_chars - len("\n".join(lines)) - (1 if lines else 0) @@ -543,3 +584,24 @@ def _format_context(hits: list[OpenVikingSearchHit], *, max_chars: int) -> str: break lines.append(line) return "\n".join(lines) + + +def _document_to_fact(document: Any) -> dict[str, Any]: + metadata = getattr(document, "metadata", {}) or {} + uri = metadata.get("openviking_uri") or metadata.get("source") or "" + score = metadata.get("openviking_score") + return { + "id": uri, + "content": str(getattr(document, "page_content", "") or ""), + "category": metadata.get("openviking_category") or "memory", + "confidence": score, + "source": uri, + "score": score, + } + + +__all__ = [ + "OpenVikingMemoryManager", + "_canonical_peer_id", + "_session_id", +] diff --git a/backend/packages/harness/deerflow/agents/memory/backends/openviking/session.py b/backend/packages/harness/deerflow/agents/memory/backends/openviking/session.py new file mode 100644 index 000000000..293e6a56f --- /dev/null +++ b/backend/packages/harness/deerflow/agents/memory/backends/openviking/session.py @@ -0,0 +1,177 @@ +"""Stable OpenViking session identity and transcript-cursor helpers.""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Mapping +from typing import Any + +from .config import GENERATED_PEER_PREFIX, is_safe_peer_id + +_SESSION_NAMESPACE = "deerflow-openviking-adapter-v1" +_DEFAULT_AGENT_SCOPE = "__default__" +_CURSOR_SCHEMA_VERSION = 1 + + +def _canonical_peer_id( + agent_name: str | None, + default_peer_id: str, +) -> str: + """Map DeerFlow's case-insensitive agent names to disjoint peer IDs.""" + + if agent_name is None: + return default_peer_id + + value = str(agent_name).strip().lower() + if not value or value == _DEFAULT_AGENT_SCOPE: + raise ValueError(f"Invalid OpenViking peer scope: {agent_name!r}") + if is_safe_peer_id(value) and value != default_peer_id and not value.startswith(GENERATED_PEER_PREFIX): + return value + + # The generated namespace is reserved, so compatible names, the default + # peer, and hashed fallbacks cannot alias one another. The 128-bit digest + # also avoids collisions caused by sanitizing or truncating agent names. + digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:32] + return f"{GENERATED_PEER_PREFIX}{digest}" + + +def _session_id( + owner_user_id: str, + peer_id: str, + thread_id: str, +) -> str: + """Derive one stable OpenViking session for one DeerFlow thread.""" + + digest = hashlib.sha256(f"{_SESSION_NAMESPACE}\0{owner_user_id}\0{peer_id}\0{thread_id}".encode()).hexdigest() + return f"df_{digest[:48]}" + + +def _memory_target_uris(peer_id: str) -> list[str]: + """Return the self and current-peer memory roots for a request.""" + + return [ + "viking://user/memories", + f"viking://user/peers/{peer_id}/memories", + ] + + +def _captureable_messages( + messages: list[Any], + should_keep_hidden_message: Any, +) -> list[Any]: + """Drop DeerFlow-only injected context before handing messages to OpenViking.""" + + selected: list[Any] = [] + for message in messages: + additional_kwargs = _message_value( + message, + "additional_kwargs", + {}, + ) + if not isinstance(additional_kwargs, dict): + additional_kwargs = {} + if additional_kwargs.get("hide_from_ui") and not (should_keep_hidden_message and should_keep_hidden_message(additional_kwargs)): + continue + selected.append(message) + return selected + + +def _message_signature(message: Any) -> str: + """Hash stable message semantics without retaining transcript content.""" + + additional_kwargs = _message_value(message, "additional_kwargs", {}) + if not isinstance(additional_kwargs, Mapping): + additional_kwargs = {} + tool_calls = _message_value(message, "tool_calls", None) + if not tool_calls: + tool_calls = additional_kwargs.get("tool_calls") or [] + + value = { + "id": _message_value(message, "id", None), + "role": _message_value(message, "type", None) or _message_value(message, "role", None), + "content": _message_value(message, "content", ""), + "tool_calls": tool_calls, + "tool_call_id": _message_value(message, "tool_call_id", None) or _message_value(message, "tool_id", None), + "tool_name": _message_value(message, "name", None) or _message_value(message, "tool_name", None), + "tool_status": _message_value(message, "status", None) or _message_value(message, "tool_status", None), + } + encoded = json.dumps( + value, + ensure_ascii=False, + sort_keys=True, + default=str, + separators=(",", ":"), + ).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _matching_prefix_count( + state: dict[str, Any], + signatures: list[str], +) -> int | None: + """Return the already submitted prefix, including after compaction.""" + + count = state.get("submitted_prefix_count") + digest = state.get("submitted_prefix_digest") + if isinstance(count, int) and 0 <= count <= len(signatures) and isinstance(digest, str): + if _sequence_digest(signatures[:count]) == digest: + return count + return None + + submitted = _string_list(state.get("submitted_signatures")) + if submitted and len(submitted) <= len(signatures): + width = len(submitted) + for start in range(len(signatures) - width, -1, -1): + if signatures[start : start + width] == submitted: + return start + width + return 0 if not state else None + + +def _advanced_cursor( + previous: dict[str, Any], + prefix_signatures: list[str] | None, + newly_submitted: list[str], + *, + max_seen: int, + commit_pending: bool, +) -> dict[str, Any]: + """Advance confirmed capture progress without persisting message content.""" + + recent = [ + *_string_list(previous.get("submitted_signatures")), + *newly_submitted, + ][-max_seen:] + state: dict[str, Any] = { + "schema_version": _CURSOR_SCHEMA_VERSION, + "submitted_signatures": recent, + "commit_pending": commit_pending, + } + if prefix_signatures is not None: + state["submitted_prefix_count"] = len(prefix_signatures) + state["submitted_prefix_digest"] = _sequence_digest(prefix_signatures) + else: + state["submitted_prefix_count"] = previous.get("submitted_prefix_count") + state["submitted_prefix_digest"] = previous.get("submitted_prefix_digest") + return state + + +def _sequence_digest(signatures: list[str]) -> str: + digest = hashlib.sha256() + for signature in signatures: + encoded = signature.encode() + digest.update(len(encoded).to_bytes(8, "big")) + digest.update(encoded) + return digest.hexdigest() + + +def _string_list(value: Any) -> list[str]: + if not isinstance(value, list): + return [] + return [item for item in value if isinstance(item, str)] + + +def _message_value(message: Any, key: str, default: Any) -> Any: + if isinstance(message, Mapping): + return message.get(key, default) + return getattr(message, key, default) diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 2e547699f..a5255b9b1 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -22,6 +22,7 @@ dependencies = [ "langchain-anthropic>=1.4.1", "langchain-deepseek>=1.0.1", "langchain-mcp-adapters>=0.2.2", + "langchain-openviking==0.1.0", "langchain-openai>=1.2.1", "langfuse>=3.4.1", "langgraph>=1.2.9,<1.3", diff --git a/backend/tests/blocking_io/test_openviking_memory_backend.py b/backend/tests/blocking_io/test_openviking_memory_backend.py index 78ebeede1..7de1f5749 100644 --- a/backend/tests/blocking_io/test_openviking_memory_backend.py +++ b/backend/tests/blocking_io/test_openviking_memory_backend.py @@ -1,85 +1,141 @@ -"""Regression anchors: OpenViking async memory methods must not block the loop.""" +"""Regression anchors: OpenViking async methods must not block the loop.""" from __future__ import annotations +from contextlib import nullcontext from pathlib import Path from typing import Any import pytest +from langchain_core.documents import Document from langchain_core.messages import AIMessage, HumanMessage -from deerflow.agents.memory.backends.openviking.models import OpenVikingCommitResult, OpenVikingSearchHit -from deerflow.agents.memory.backends.openviking.openviking_manager import OpenVikingMemoryManager +from deerflow.agents.memory.backends.openviking.openviking_manager import ( + OpenVikingMemoryManager, +) -class _BlockingProbeClient: - """Perform real file IO so Blockbuster can detect missing async offload.""" +class _CommitPolicy: + def __init__(self, *, mode: str): + self.mode = mode - def __init__(self, probe_path: Path): - self._probe_path = probe_path - def _probe(self) -> None: - self._probe_path.write_text("probe", encoding="utf-8") +class _Client: + supports_request_actor_peer = True - def ensure_session(self, identity, session_id) -> None: - self._probe() - - def add_messages(self, identity, session_id, messages) -> int: - self._probe() - return len(messages) - - def commit_session(self, identity, session_id) -> OpenVikingCommitResult: - self._probe() - return OpenVikingCommitResult(status="accepted", task_id="task-1", archive_uri=None, archived=True) - - def search( - self, - identity, - query: str, - *, - top_k: int, - category: str | None = None, - session_id: str | None = None, - ) -> list[OpenVikingSearchHit]: - self._probe() - return [ - OpenVikingSearchHit( - uri="viking://user/memories/preferences/test.md", - context_type="memory", - category="preferences", - score=0.9, - abstract="Prefers concise answers.", - overview=None, - match_reason="", - ) - ] + def health(self) -> bool: + return True def close(self) -> None: pass -def _manager(tmp_path: Path) -> OpenVikingMemoryManager: +class _BlockingRecorder: + def __init__(self, *, commit_policy: Any, **kwargs: Any): + del kwargs + self.commit_policy = commit_policy + self.client = _Client() + self.probe_path: Path | None = None + + def record( + self, + session_id: str, + messages: list[Any], + peer_id: str | None = None, + ) -> None: + del session_id, messages, peer_id + assert self.probe_path is not None + self.probe_path.write_text("record", encoding="utf-8") + + def flush(self, session_id: str) -> None: + del session_id + + def close(self) -> None: + pass + + +class _BlockingRetriever: + def __init__(self, *, client: Any, **kwargs: Any): + del client + self.__dict__.update(kwargs) + self.filter = None + self.target_uri = "" + self.session_id = None + self.probe_path: Path | None = None + + def __copy__(self) -> _BlockingRetriever: + copied = type(self)(client=None) + copied.__dict__.update(self.__dict__) + return copied + + def invoke(self, query: str) -> list[Document]: + del query + assert self.probe_path is not None + self.probe_path.write_text("retrieve", encoding="utf-8") + return [ + Document( + page_content="Prefers concise answers.", + metadata={ + "openviking_uri": ("viking://user/memories/preferences/test.md"), + "openviking_category": "preferences", + "openviking_score": 0.9, + }, + ) + ] + + +def _manager( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> OpenVikingMemoryManager: + import deerflow.agents.memory.backends.openviking.openviking_manager as module + + monkeypatch.setenv("OPENVIKING_API_KEY", "user-key") + monkeypatch.setattr( + module, + "_load_official_integration", + lambda: { + "OpenVikingCommitPolicy": _CommitPolicy, + "OpenVikingPartialWriteError": RuntimeError, + "OpenVikingRetriever": _BlockingRetriever, + "OpenVikingSessionRecorder": _BlockingRecorder, + "use_actor_peer": lambda peer_id: nullcontext(), + }, + ) manager = OpenVikingMemoryManager.from_config( { "base_url": "http://openviking:1933", "storage_path": str(tmp_path), - "auth_mode": "trusted", - "account": "deerflow", + "owner_user_id": "alice", "startup_policy": "warn", } ) - manager._client = _BlockingProbeClient(tmp_path / "probe.txt") # type: ignore[assignment] + manager._recorder.probe_path = tmp_path / "record.txt" + manager._retriever.probe_path = tmp_path / "retrieve.txt" return manager @pytest.mark.asyncio -async def test_async_openviking_operations_do_not_block_event_loop(tmp_path: Path) -> None: - manager = _manager(tmp_path) - messages: list[Any] = [HumanMessage("hello", id="h1"), AIMessage("hi", id="a1")] +async def test_async_openviking_operations_do_not_block_event_loop( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + manager = _manager(tmp_path, monkeypatch) + messages: list[Any] = [ + HumanMessage("hello", id="h1"), + AIMessage("hi", id="a1"), + ] - await manager.aadd("thread-1", messages, user_id="alice") - assert await manager.aget_context("alice") == "- [preferences] Prefers concise answers." - assert await manager.asearch("answer style", user_id="alice") == [ + await manager.aadd( + "thread-1", + messages, + user_id="alice", + ) + assert await manager.aget_context("alice") == ("- [preferences] Prefers concise answers.") + assert await manager.asearch( + "answer style", + user_id="alice", + ) == [ { "id": "viking://user/memories/preferences/test.md", "content": "Prefers concise answers.", diff --git a/backend/tests/test_openviking_memory_backend.py b/backend/tests/test_openviking_memory_backend.py index 0cf47820e..947edf71f 100644 --- a/backend/tests/test_openviking_memory_backend.py +++ b/backend/tests/test_openviking_memory_backend.py @@ -1,93 +1,239 @@ +"""Tests for the official-package OpenViking memory backend.""" + from __future__ import annotations +import copy import gc import json import threading import weakref +from contextlib import contextmanager +from contextvars import ContextVar from pathlib import Path from typing import Any -import httpx import pytest -from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage +from langchain_core.documents import Document +from langchain_core.messages import AIMessage, HumanMessage -from deerflow.agents.memory.backends.openviking.client import OpenVikingAuthenticationError, OpenVikingHttpClient, OpenVikingUnavailableError from deerflow.agents.memory.backends.openviking.config import OpenVikingConfig -from deerflow.agents.memory.backends.openviking.models import ( - OpenVikingCommitResult, - OpenVikingIdentity, - OpenVikingMessage, - OpenVikingSearchHit, +from deerflow.agents.memory.backends.openviking.openviking_manager import ( + OpenVikingMemoryManager, + _canonical_peer_id, + _session_id, ) -from deerflow.agents.memory.backends.openviking.openviking_manager import OpenVikingMemoryManager -from deerflow.agents.memory.manager import _scan_backends, reset_memory_manager +from deerflow.agents.memory.manager import ( + MemoryManagerError, + _scan_backends, + reset_memory_manager, +) + + +class _CommitPolicy: + def __init__(self, *, mode: str, pending_token_threshold: int = 8_000): + self.mode = mode + self.pending_token_threshold = pending_token_threshold + + +class _PartialWriteError(RuntimeError): + def __init__( + self, + consumed: int, + *, + commit_pending: bool = False, + ) -> None: + super().__init__("partial") + self.input_messages_consumed = consumed + self.commit_pending = commit_pending + + +class _Client: + supports_request_actor_peer = True + + def __init__(self, **kwargs: Any): + self.kwargs = kwargs + self.closed = False + self.healthy = True + + def health(self) -> bool: + return self.healthy + + def close(self) -> None: + self.closed = True + + +class _Recorder: + def __init__( + self, + *, + commit_policy: _CommitPolicy, + url: str, + api_key: str, + timeout: float, + extra_headers: dict[str, str], + ) -> None: + self.commit_policy = commit_policy + self.connection = { + "url": url, + "api_key": api_key, + "timeout": timeout, + "extra_headers": extra_headers, + } + self._client = _Client(**self.connection) + self.calls: list[tuple[str, list[Any], str | None, str | None]] = [] + self.flushes: list[tuple[str, str | None]] = [] + self.failures: list[BaseException] = [] + self.closed = False + + @property + def client(self) -> _Client: + return self._client + + def record( + self, + session_id: str, + messages: list[Any], + peer_id: str | None = None, + ) -> object: + self.calls.append((session_id, list(messages), peer_id, _ACTOR_PEER.get())) + if self.failures: + raise self.failures.pop(0) + return object() + + def flush(self, session_id: str) -> None: + self.flushes.append((session_id, _ACTOR_PEER.get())) + if self.failures: + raise self.failures.pop(0) + + def close(self) -> None: + self.closed = True + self._client.close() + + +class _Retriever: + def __init__(self, *, client: _Client, **kwargs: Any): + self.client = client + self.kwargs = kwargs + self.limit = kwargs["limit"] + self.filter = None + self.session_id = kwargs.get("session_id") + self.target_uri = kwargs.get("target_uri", "") + self.calls: list[dict[str, Any]] = [] + + def __copy__(self) -> _Retriever: + copied = type(self)(client=self.client, **self.kwargs) + copied.calls = self.calls + copied.limit = self.limit + copied.filter = copy.deepcopy(self.filter) + copied.session_id = self.session_id + copied.target_uri = copy.deepcopy(self.target_uri) + return copied + + def invoke(self, query: str) -> list[Document]: + self.calls.append( + { + "query": query, + "actor_peer": _ACTOR_PEER.get(), + "limit": self.limit, + "filter": copy.deepcopy(self.filter), + "session_id": self.session_id, + "target_uri": copy.deepcopy(self.target_uri), + "search_mode": getattr(self, "search_mode", "find"), + } + ) + return [ + Document( + page_content="Prefers concise answers.", + metadata={ + "openviking_uri": ("viking://user/memories/preferences/style.md"), + "openviking_category": "preferences", + "openviking_score": 0.91, + }, + ) + ] + + +_ACTOR_PEER: ContextVar[str | None] = ContextVar( + "test_actor_peer", + default=None, +) + + +@contextmanager +def _use_actor_peer(peer_id: str | None): + token = _ACTOR_PEER.set(peer_id) + try: + yield + finally: + _ACTOR_PEER.reset(token) + + +@pytest.fixture +def official_integration(monkeypatch: pytest.MonkeyPatch) -> None: + import deerflow.agents.memory.backends.openviking.openviking_manager as module + + monkeypatch.setattr( + module, + "_load_official_integration", + lambda: { + "OpenVikingCommitPolicy": _CommitPolicy, + "OpenVikingPartialWriteError": _PartialWriteError, + "OpenVikingRetriever": _Retriever, + "OpenVikingSessionRecorder": _Recorder, + "use_actor_peer": _use_actor_peer, + }, + ) def _backend_config(tmp_path: Path, **overrides: Any) -> dict[str, Any]: config: dict[str, Any] = { "base_url": "http://openviking:1933", "storage_path": str(tmp_path), - "auth_mode": "trusted", - "account": "deerflow", + "owner_user_id": "alice", "startup_policy": "warn", - "retrieval": {"top_k": 4, "max_injection_chars": 1000}, + "retrieval": { + "top_k": 4, + "max_injection_chars": 1_000, + "injection_query": "profile preferences and prior decisions", + }, } config.update(overrides) return config -def test_config_parses_nested_fields_and_rejects_unknown(tmp_path: Path) -> None: - config = OpenVikingConfig.from_backend_config( - _backend_config( - tmp_path, - retrieval={"top_k": 7, "score_threshold": 0.4, "max_injection_chars": 2048}, - failure_policy={"read": "raise", "write": "raise"}, - ) - ) - - assert config.search_top_k == 7 - assert config.score_threshold == 0.4 - assert config.read_failure_policy == "raise" - - with pytest.raises(ValueError, match="Unknown OpenViking"): - OpenVikingConfig.from_backend_config(_backend_config(tmp_path, typo=True)) +def _manager( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + **overrides: Any, +) -> OpenVikingMemoryManager: + monkeypatch.setenv("OPENVIKING_API_KEY", "user-key") + return OpenVikingMemoryManager.from_config(_backend_config(tmp_path, **overrides)) -def test_config_rejects_dev_auth_without_explicit_opt_in(tmp_path: Path) -> None: - with pytest.raises(ValueError, match="allow_insecure_dev"): - OpenVikingConfig.from_backend_config(_backend_config(tmp_path, auth_mode="dev")) - - -def test_config_repr_does_not_expose_api_key(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("OPENVIKING_API_KEY", "super-secret-api-key") +def test_config_uses_single_user_key_and_rejects_legacy_trusted_fields( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("OPENVIKING_API_KEY", raising=False) + with pytest.raises(ValueError, match="USER API key"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path)) + monkeypatch.setenv("OPENVIKING_API_KEY", "secret") config = OpenVikingConfig.from_backend_config(_backend_config(tmp_path)) - assert "super-secret-api-key" not in repr(config) + assert config.owner_user_id == "alice" + assert config.content_mode == "overview" + assert config.injection_query == "profile preferences and prior decisions" + assert "secret" not in repr(config) - -def test_config_parses_and_validates_connection_limits(tmp_path: Path) -> None: - config = OpenVikingConfig.from_backend_config( - _backend_config( - tmp_path, - max_connections=48, - max_keepalive_connections=12, - ) - ) - - assert config.max_connections == 48 - assert config.max_keepalive_connections == 12 - - with pytest.raises(ValueError, match="max_connections"): - OpenVikingConfig.from_backend_config(_backend_config(tmp_path, max_connections=0)) - with pytest.raises(ValueError, match="max_keepalive_connections"): - OpenVikingConfig.from_backend_config( - _backend_config( - tmp_path, - max_connections=10, - max_keepalive_connections=11, - ) - ) + with pytest.raises(ValueError, match="trusted mode is no longer supported"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path, auth_mode="trusted")) + with pytest.raises(ValueError, match="Unknown OpenViking"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path, typo=True)) + with pytest.raises(ValueError, match="reserved prefix"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path, default_peer_id="df-agent-default")) + with pytest.raises(ValueError, match="custom HTTP client fields"): + OpenVikingConfig.from_backend_config(_backend_config(tmp_path, max_connections=10)) def test_backend_is_discovered_by_registered_name() -> None: @@ -95,341 +241,391 @@ def test_backend_is_discovered_by_registered_name() -> None: assert _scan_backends()["openviking"] is OpenVikingMemoryManager -def test_http_client_sends_trusted_identity_and_maps_search(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setenv("OPENVIKING_API_KEY", "secret") - requests: list[httpx.Request] = [] - - def handler(request: httpx.Request) -> httpx.Response: - requests.append(request) - if request.url.path == "/api/v1/search/find": - return httpx.Response( - 200, - json={ - "status": "ok", - "result": { - "memories": [ - { - "uri": "viking://user/memories/preferences/editor.md", - "context_type": "memory", - "category": "preferences", - "score": 0.91, - "abstract": "Uses Vim.", - "overview": None, - } - ], - "resources": [], - "skills": [], - "total": 1, - }, - }, - ) - raise AssertionError(f"unexpected path: {request.url.path}") - - config = OpenVikingConfig.from_backend_config(_backend_config(tmp_path)) - client = OpenVikingHttpClient(config, transport=httpx.MockTransport(handler)) - identity = OpenVikingIdentity(account="deerflow", user="df_user") - - hits = client.search(identity, "editor preference", top_k=3) - - assert [hit.abstract for hit in hits] == ["Uses Vim."] - assert requests[0].headers["X-OpenViking-Account"] == "deerflow" - assert requests[0].headers["X-OpenViking-User"] == "df_user" - assert requests[0].headers["X-API-Key"] == "secret" - assert json.loads(requests[0].content)["target_uri"] == "viking://user/memories" - assert json.loads(requests[0].content)["context_type"] == "memory" - - -def test_http_client_maps_authentication_error(tmp_path: Path) -> None: - def handler(request: httpx.Request) -> httpx.Response: - return httpx.Response( - 401, - json={"status": "error", "error": {"code": "UNAUTHENTICATED", "message": "bad key"}}, - ) - - config = OpenVikingConfig.from_backend_config(_backend_config(tmp_path)) - client = OpenVikingHttpClient(config, transport=httpx.MockTransport(handler)) - - with pytest.raises(OpenVikingAuthenticationError) as exc_info: - client.ensure_session(OpenVikingIdentity(account="deerflow", user="df_user"), "session") - assert exc_info.value.code == "UNAUTHENTICATED" - - -def test_http_client_configures_connection_limits(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - captured: dict[str, Any] = {} - - class _RecordingClient: - def __init__(self, **kwargs: Any) -> None: - captured.update(kwargs) - - monkeypatch.setattr(httpx, "Client", _RecordingClient) - config = OpenVikingConfig.from_backend_config( - _backend_config( - tmp_path, - max_connections=32, - max_keepalive_connections=8, - ) +def test_official_loader_uses_standalone_package() -> None: + from deerflow.agents.memory.backends.openviking.openviking_manager import ( + _load_official_integration, ) - OpenVikingHttpClient(config) + integration = _load_official_integration() - limits = captured["limits"] - assert limits.max_connections == 32 - assert limits.max_keepalive_connections == 8 + assert integration["OpenVikingSessionRecorder"].__module__.startswith("langchain_openviking") + assert integration["OpenVikingRetriever"].__module__.startswith("langchain_openviking") -def test_http_client_adds_jitter_to_retry_delay(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - attempts = 0 - jitter_ranges: list[tuple[float, float]] = [] - sleeps: list[float] = [] +def test_manager_uses_official_recorder_retriever_and_commit_always( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) - def handler(request: httpx.Request) -> httpx.Response: - nonlocal attempts - attempts += 1 - if attempts == 1: - return httpx.Response(503) - return httpx.Response(200, json={"status": "ok"}) - - def fake_uniform(lower: float, upper: float) -> float: - jitter_ranges.append((lower, upper)) - return 0.02 - - monkeypatch.setattr("random.uniform", fake_uniform) - monkeypatch.setattr("time.sleep", sleeps.append) - config = OpenVikingConfig.from_backend_config(_backend_config(tmp_path, max_retries=1)) - client = OpenVikingHttpClient(config, transport=httpx.MockTransport(handler)) - - assert client.health() is True - assert jitter_ranges == [(0.0, 0.05)] - assert sleeps == [pytest.approx(0.07)] + assert manager._recorder.commit_policy.mode == "always" + assert manager._retriever.client is manager._recorder.client + assert manager._retriever.kwargs["content_mode"] == "overview" + assert manager._recorder.connection == { + "url": "http://openviking:1933", + "api_key": "user-key", + "timeout": 30.0, + "extra_headers": {}, + } -class _FakeClient: - def __init__(self) -> None: - self.ensured: list[tuple[OpenVikingIdentity, str]] = [] - self.added: list[tuple[OpenVikingIdentity, str, list[OpenVikingMessage]]] = [] - self.committed: list[tuple[OpenVikingIdentity, str]] = [] - self.searches: list[tuple[OpenVikingIdentity, str, int, str | None]] = [] - self.closed = False +def test_context_preserves_existing_fixed_query_behavior( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) - def ensure_session(self, identity: OpenVikingIdentity, session_id: str) -> None: - self.ensured.append((identity, session_id)) + context = manager.get_context( + "alice", + agent_name="research", + thread_id="thread-1", + ) - def add_messages( - self, - identity: OpenVikingIdentity, - session_id: str, - messages: list[OpenVikingMessage], - ) -> int: - self.added.append((identity, session_id, messages)) - return len(messages) + assert context == "- [preferences] Prefers concise answers." + assert manager._retriever.calls == [ + { + "query": "profile preferences and prior decisions", + "actor_peer": "research", + "limit": 4, + "filter": None, + "session_id": _session_id("alice", "research", "thread-1"), + "target_uri": [ + "viking://user/memories", + "viking://user/peers/research/memories", + ], + "search_mode": "search", + } + ] - def commit_session(self, identity: OpenVikingIdentity, session_id: str) -> OpenVikingCommitResult: - self.committed.append((identity, session_id)) - return OpenVikingCommitResult( - status="accepted", - task_id="task-1", - archive_uri="viking://user/sessions/session/history/archive_001", - archived=True, + +def test_context_without_thread_uses_existing_find_path( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + + context = manager.get_context( + "alice", + agent_name="research", + ) + + assert context == "- [preferences] Prefers concise answers." + assert manager._retriever.calls == [ + { + "query": "profile preferences and prior decisions", + "actor_peer": "research", + "limit": 4, + "filter": None, + "session_id": None, + "target_uri": [ + "viking://user/memories", + "viking://user/peers/research/memories", + ], + "search_mode": "find", + } + ] + + +def test_manager_refuses_to_share_single_user_key_across_deerflow_users( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + + with pytest.raises(MemoryManagerError, match="owner_user_id 'alice'"): + manager.get_context("bob", agent_name="research") + with pytest.raises(MemoryManagerError, match="owner_user_id 'alice'"): + manager.add( + "thread-1", + [HumanMessage("private", id="h1")], + user_id="bob", + agent_name="research", ) - def search( - self, - identity: OpenVikingIdentity, - query: str, - *, - top_k: int, - category: str | None = None, - session_id: str | None = None, - ) -> list[OpenVikingSearchHit]: - self.searches.append((identity, query, top_k, category)) - return [ - OpenVikingSearchHit( - uri="viking://user/memories/preferences/editor.md", - context_type="memory", - category="preferences", - score=0.9, - abstract="User prefers concise answers.", - overview=None, - match_reason="", - ) - ] - - def health(self) -> bool: - return True - - def close(self) -> None: - self.closed = True + assert manager._retriever.calls == [] + assert manager._recorder.calls == [] -def _manager(tmp_path: Path, **overrides: Any) -> tuple[OpenVikingMemoryManager, _FakeClient]: - manager = OpenVikingMemoryManager.from_config(_backend_config(tmp_path, **overrides)) - fake = _FakeClient() - manager._client = fake # type: ignore[assignment] - return manager, fake - - -def test_manager_filters_messages_commits_and_deduplicates(tmp_path: Path) -> None: - manager, client = _manager(tmp_path) +def test_manager_records_only_unseen_suffix( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) messages = [ - SystemMessage("system"), - HumanMessage("Remember that I prefer Vim.", id="human-1"), - AIMessage("", tool_calls=[{"name": "search", "args": {}, "id": "call-1", "type": "tool_call"}]), - ToolMessage("tool output", tool_call_id="call-1"), - AIMessage("I will remember that.", id="ai-1"), + HumanMessage("Remember Vim.", id="h1"), + AIMessage("I will remember.", id="a1"), ] - manager.add("thread-1", messages, user_id="alice", agent_name="research") - manager.add("thread-1", messages, user_id="alice", agent_name="research") + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + messages.append(HumanMessage("Also concise.", id="h2")) + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) - assert len(client.added) == 1 - assert [(message.role, message.content) for message in client.added[0][2]] == [ - ("user", "Remember that I prefer Vim."), - ("assistant", "I will remember that."), + assert len(manager._recorder.calls) == 2 + assert manager._recorder.calls[0][1] == messages[:2] + assert manager._recorder.calls[1][1] == messages[2:] + assert all(call[2:] == ("research", "research") for call in manager._recorder.calls) + + +def test_partial_write_progress_is_not_resubmitted( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + messages = [ + HumanMessage("one", id="h1"), + AIMessage("two", id="a1"), + HumanMessage("three", id="h2"), ] - assert len(client.committed) == 1 - watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) - state = json.loads(watermark.read_text(encoding="utf-8")) - assert state["last_commit_task_id"] == "task-1" - assert state["submitted_message_ids"] == state["committed_message_ids"] + manager._recorder.failures.append(_PartialWriteError(2)) + + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + + assert manager._recorder.calls[0][1] == messages + assert manager._recorder.calls[1][1] == messages[2:] -def test_manager_does_not_resubmit_messages_after_failed_commit(tmp_path: Path) -> None: - manager, client = _manager(tmp_path) - messages = [HumanMessage("hello", id="h1"), AIMessage("hi", id="a1")] - original_commit = client.commit_session - commit_attempts = 0 +def test_pending_commit_is_retried_without_resubmitting_messages( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + messages = [HumanMessage("one", id="h1"), AIMessage("two", id="a1")] + manager._recorder.failures.append(_PartialWriteError(len(messages), commit_pending=True)) - def fail_once(identity: OpenVikingIdentity, session_id: str) -> OpenVikingCommitResult: - nonlocal commit_attempts - commit_attempts += 1 - if commit_attempts == 1: - raise OpenVikingUnavailableError("session.commit", "temporary failure") - return original_commit(identity, session_id) + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) - client.commit_session = fail_once # type: ignore[method-assign] - - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) - failed_state = json.loads(watermark.read_text(encoding="utf-8")) - assert failed_state["submitted_message_ids"] == ["df_h1", "df_a1"] - assert failed_state["committed_message_ids"] == [] - - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - assert len(client.added) == 1 - assert commit_attempts == 1 - - messages.append(HumanMessage("new information", id="h2")) - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - assert len(client.added) == 2 - assert [message.message_id for message in client.added[1][2]] == ["df_h2"] - assert commit_attempts == 2 - recovered_state = json.loads(watermark.read_text(encoding="utf-8")) - assert recovered_state["submitted_message_ids"] == recovered_state["committed_message_ids"] - assert recovered_state["last_commit_task_id"] == "task-1" + session_id = _session_id("alice", "research", "thread-1") + assert manager._recorder.flushes == [(session_id, "research")] + assert len(manager._recorder.calls) == 1 -def test_manager_does_not_resubmit_history_beyond_recent_id_window(tmp_path: Path) -> None: - manager, client = _manager(tmp_path, max_seen_message_ids=16) - messages = [HumanMessage(f"message {index}", id=f"h{index}") for index in range(20)] +def test_search_maps_documents_without_custom_http_models( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) - manager.add("thread-1", messages, user_id="alice", agent_name="research") - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - assert len(client.added) == 1 - - messages.append(HumanMessage("message 20", id="h20")) - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - assert len(client.added) == 2 - assert [message.message_id for message in client.added[1][2]] == ["df_h20"] - watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) - state = json.loads(watermark.read_text(encoding="utf-8")) - assert state["schema_version"] == 3 - assert state["submitted_prefix_count"] == 21 - assert len(state["submitted_message_ids"]) == 16 - - -def test_manager_rebases_watermark_after_history_compaction(tmp_path: Path) -> None: - manager, client = _manager(tmp_path, max_seen_message_ids=16) - messages = [HumanMessage(f"message {index}", id=f"h{index}") for index in range(20)] - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - compacted = [*messages[-8:], HumanMessage("message 20", id="h20")] - manager.add("thread-1", compacted, user_id="alice", agent_name="research") - manager.add("thread-1", compacted, user_id="alice", agent_name="research") - - assert len(client.added) == 2 - assert [message.message_id for message in client.added[1][2]] == ["df_h20"] - watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) - state = json.loads(watermark.read_text(encoding="utf-8")) - assert state["submitted_prefix_count"] == 9 - - -def test_manager_migrates_legacy_recent_id_watermark(tmp_path: Path) -> None: - manager, client = _manager(tmp_path, max_seen_message_ids=16) - messages = [HumanMessage(f"message {index}", id=f"h{index}") for index in range(20)] - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - watermark = next((tmp_path / "openviking" / "sessions").glob("*.json")) - legacy_state = json.loads(watermark.read_text(encoding="utf-8")) - legacy_state["schema_version"] = 2 - legacy_state.pop("submitted_prefix_count", None) - legacy_state.pop("submitted_prefix_digest", None) - legacy_state.pop("committed_prefix_count", None) - legacy_state.pop("committed_prefix_digest", None) - watermark.write_text(json.dumps(legacy_state), encoding="utf-8") - - messages.append(HumanMessage("message 20", id="h20")) - manager.add("thread-1", messages, user_id="alice", agent_name="research") - - assert len(client.added) == 2 - assert [message.message_id for message in client.added[1][2]] == ["df_h20"] - - -def test_manager_identity_is_stable_and_agent_isolated(tmp_path: Path) -> None: - manager, client = _manager(tmp_path) - messages = [HumanMessage("hello", id="h1"), AIMessage("hi", id="a1")] - - manager.add("thread-1", messages, user_id="alice", agent_name="research") - manager.add("thread-1", messages, user_id="alice", agent_name="coding") - - identities = [entry[0].user for entry in client.added] - session_ids = [entry[1] for entry in client.added] - assert identities[0] != identities[1] - assert session_ids[0] != session_ids[1] - assert all(value.startswith("df_") for value in identities) - - -def test_manager_search_and_context_map_remote_results(tmp_path: Path) -> None: - manager, client = _manager(tmp_path) - - results = manager.search("answer style", user_id="alice", agent_name="research") - context = manager.get_context("alice", agent_name="research") + results = manager.search( + "answer style", + top_k=3, + user_id="alice", + agent_name="research", + category="preferences", + ) assert results == [ { - "id": "viking://user/memories/preferences/editor.md", - "content": "User prefers concise answers.", + "id": "viking://user/memories/preferences/style.md", + "content": "Prefers concise answers.", "category": "preferences", - "confidence": 0.9, - "source": "viking://user/memories/preferences/editor.md", - "score": 0.9, + "confidence": 0.91, + "source": "viking://user/memories/preferences/style.md", + "score": 0.91, } ] - assert context == "- [preferences] User prefers concise answers." - assert client.searches[1][1] == manager._config.injection_query + assert manager._retriever.calls[0]["filter"] == { + "op": "must", + "field": "category", + "conds": ["preferences"], + } + assert manager._retriever.calls[0]["search_mode"] == "find" -def test_manager_rejects_tool_mode(tmp_path: Path) -> None: - with pytest.raises(ValueError, match="middleware"): - OpenVikingMemoryManager.from_config(_backend_config(tmp_path), mode="tool") +def test_capture_keeps_tool_history_but_drops_hidden_injected_context( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + messages = [ + HumanMessage("question", id="h1"), + AIMessage( + "", + id="a-tool", + tool_calls=[ + { + "name": "search", + "args": {"query": "OpenViking"}, + "id": "call-1", + "type": "tool_call", + } + ], + ), + HumanMessage( + "injected memory", + id="hidden", + additional_kwargs={"hide_from_ui": True}, + ), + AIMessage("answer", id="a1"), + ] + + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + + assert manager._recorder.calls[0][1] == [ + messages[0], + messages[1], + messages[3], + ] -def test_manager_session_locks_do_not_accumulate(tmp_path: Path) -> None: - manager, _ = _manager(tmp_path) +def test_compacted_history_rebases_without_replaying_known_messages( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager( + tmp_path, + monkeypatch, + max_seen_message_ids=16, + ) + messages = [HumanMessage(f"message {index}", id=f"h{index}") for index in range(20)] + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + compacted = [ + *messages[-8:], + HumanMessage("new", id="h20"), + ] + + manager.add( + "thread-1", + compacted, + user_id="alice", + agent_name="research", + ) + + assert manager._recorder.calls[1][1] == [compacted[-1]] + + +def test_failed_write_does_not_advance_capture_cursor( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + manager._recorder.failures.append(RuntimeError("unavailable")) + messages = [HumanMessage("retry me", id="h1")] + + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + manager.add( + "thread-1", + messages, + user_id="alice", + agent_name="research", + ) + + assert [call[1] for call in manager._recorder.calls] == [ + messages, + messages, + ] + + +def test_corrupt_cursor_fails_closed_instead_of_replaying_history( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + session_id = _session_id("alice", "research", "thread-1") + cursor = manager._state_path(session_id) + cursor.parent.mkdir(parents=True) + cursor.write_text("not-json", encoding="utf-8") + + with pytest.raises(MemoryManagerError, match="refusing unsafe replay"): + manager.add( + "thread-1", + [HumanMessage("private", id="h1")], + user_id="alice", + agent_name="research", + ) + + assert manager._recorder.calls == [] + + +def test_peer_mapping_is_stable_and_namespaces_are_disjoint() -> None: + assert _canonical_peer_id(None, "deerflow") == "deerflow" + assert _canonical_peer_id("Research", "deerflow") == "research" + assert _canonical_peer_id("deerflow", "deerflow").startswith("df-agent-") + assert _canonical_peer_id("-research", "deerflow").startswith("df-agent-") + assert _canonical_peer_id("df-agent-custom", "deerflow").startswith("df-agent-") + assert ( + len( + { + _canonical_peer_id(None, "deerflow"), + _canonical_peer_id("deerflow", "deerflow"), + _canonical_peer_id("-research", "deerflow"), + _canonical_peer_id("df-agent-custom", "deerflow"), + } + ) + == 4 + ) + + +def test_session_locks_do_not_accumulate( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) lock = manager._session_lock("session-1") lock_ref = weakref.ref(lock) @@ -441,67 +637,104 @@ def test_manager_session_locks_do_not_accumulate(tmp_path: Path) -> None: assert len(manager._session_locks) == 0 -@pytest.mark.asyncio -async def test_manager_async_methods_offload_sync_operations(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: - manager, _ = _manager(tmp_path) - event_loop_thread = threading.get_ident() - worker_threads: list[int] = [] +def test_shutdown_uses_existing_manager_lifecycle_contract( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) - def fake_add(self: OpenVikingMemoryManager, *args: Any, **kwargs: Any) -> None: - worker_threads.append(threading.get_ident()) - - def fake_get_context(self: OpenVikingMemoryManager, *args: Any, **kwargs: Any) -> str: - worker_threads.append(threading.get_ident()) - return "context" - - def fake_search(self: OpenVikingMemoryManager, *args: Any, **kwargs: Any) -> list[dict[str, Any]]: - worker_threads.append(threading.get_ident()) - return [{"id": "memory-1"}] - - monkeypatch.setattr(OpenVikingMemoryManager, "add", fake_add) - monkeypatch.setattr(OpenVikingMemoryManager, "get_context", fake_get_context) - monkeypatch.setattr(OpenVikingMemoryManager, "search", fake_search) - - await manager.aadd("thread-1", [], user_id="alice") - assert await manager.aget_context("alice") == "context" - assert await manager.asearch("query", user_id="alice") == [{"id": "memory-1"}] - - assert len(worker_threads) == 3 - assert all(thread_id != event_loop_thread for thread_id in worker_threads) + assert manager.shutdown_flush(1.0) is True + assert manager._recorder.closed is True + assert manager._recorder.client.closed is True + assert manager.shutdown_flush(1.0) is True -def test_manager_shutdown_waits_for_in_flight_write(tmp_path: Path) -> None: - manager, client = _manager(tmp_path) - write_started = threading.Event() - release_write = threading.Event() - original_add = client.add_messages +def test_shutdown_timeout_closes_resources_after_active_write_finishes( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + started = threading.Event() + release = threading.Event() + original_record = manager._recorder.record - def blocking_add( - identity: OpenVikingIdentity, - session_id: str, - messages: list[OpenVikingMessage], - ) -> int: - write_started.set() - assert release_write.wait(2) - return original_add(identity, session_id, messages) + def blocking_record(*args: Any, **kwargs: Any) -> object: + started.set() + assert release.wait(2.0) + return original_record(*args, **kwargs) - client.add_messages = blocking_add # type: ignore[method-assign] - write_thread = threading.Thread( + manager._recorder.record = blocking_record + writer = threading.Thread( target=manager.add, args=("thread-1", [HumanMessage("hello", id="h1")]), - kwargs={"user_id": "alice"}, + kwargs={"user_id": "alice", "agent_name": "research"}, ) - write_thread.start() - assert write_started.wait(1) + writer.start() + assert started.wait(1.0) - assert manager.shutdown_flush(0.01) is False - assert client.closed is False - manager.add("thread-2", [HumanMessage("ignored", id="h2")], user_id="alice") - assert len(client.ensured) == 1 + assert manager.shutdown_flush(0.0) is False + assert manager._recorder.closed is False - release_write.set() - write_thread.join(2) - assert not write_thread.is_alive() + release.set() + writer.join(2.0) - assert manager.shutdown_flush(1) is True - assert client.closed is True + assert not writer.is_alive() + assert manager._recorder.closed is True + assert manager._recorder.client.closed is True + assert manager.shutdown_flush(1.0) is True + + +@pytest.mark.asyncio +async def test_async_operations_run_off_the_event_loop( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + event_loop_thread = threading.get_ident() + worker_threads: list[int] = [] + original_add = OpenVikingMemoryManager.add + + def recording_add( + self: OpenVikingMemoryManager, + *args: Any, + **kwargs: Any, + ) -> None: + worker_threads.append(threading.get_ident()) + original_add(self, *args, **kwargs) + + monkeypatch.setattr(OpenVikingMemoryManager, "add", recording_add) + + await manager.aadd( + "thread-1", + [HumanMessage("hello", id="h1")], + user_id="alice", + agent_name="research", + ) + + assert worker_threads + assert worker_threads[0] != event_loop_thread + + +def test_capture_cursor_contains_no_message_content( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + official_integration: None, +) -> None: + manager = _manager(tmp_path, monkeypatch) + + manager.add( + "thread-1", + [HumanMessage("very private text", id="h1")], + user_id="alice", + agent_name="research", + ) + + cursor = next((tmp_path / "openviking" / "sessions").glob("*.json")) + serialized = cursor.read_text(encoding="utf-8") + state = json.loads(serialized) + + assert "very private text" not in serialized + assert state["submitted_prefix_count"] == 1 diff --git a/backend/uv.lock b/backend/uv.lock index 06b4cfcbd..4172e29a2 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -936,6 +936,7 @@ dependencies = [ { name = "langchain-google-genai" }, { name = "langchain-mcp-adapters" }, { name = "langchain-openai" }, + { name = "langchain-openviking" }, { name = "langfuse" }, { name = "langgraph" }, { name = "langgraph-api" }, @@ -1015,6 +1016,7 @@ requires-dist = [ { name = "langchain-mcp-adapters", specifier = ">=0.2.2" }, { name = "langchain-ollama", marker = "extra == 'ollama'", specifier = ">=0.3.0" }, { name = "langchain-openai", specifier = ">=1.2.1" }, + { name = "langchain-openviking", specifier = "==0.1.0" }, { name = "langfuse", specifier = ">=3.4.1" }, { name = "langgraph", specifier = ">=1.2.9,<1.3" }, { name = "langgraph-api", specifier = ">=0.8.1" }, @@ -2097,6 +2099,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/dc/55/2865b18ee3a3dd11160b8c4b2cf37e75bf2a4a8d1d38868ffffc7b7cc180/langchain_openai-1.2.1-py3-none-any.whl", hash = "sha256:a80732185030d4f453dda6c25feef46f645f665423fdffe38ae3edf1ac3c6c4d", size = 98626, upload-time = "2026-04-24T19:46:41.971Z" }, ] +[[package]] +name = "langchain-openviking" +version = "0.1.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "langchain-core" }, + { name = "openviking-sdk" }, + { name = "pydantic" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/7f/c0/418dba42bddb5093d20d5e36c3e8d9ceff0b27ce92a6bc2f16fb33535fac/langchain_openviking-0.1.0.tar.gz", hash = "sha256:dc4ee47a122527881eae1fbbe6f9ca857cc0269cd60f08c8d3fd3fff94996e9e", size = 48694, upload-time = "2026-08-03T10:45:28.614Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/04/22/87f547ab4a84fa4094462e1e1cfdbadf28a539b5cd0efdb147b754d3b88a/langchain_openviking-0.1.0-py3-none-any.whl", hash = "sha256:ea32f08fa4bed81743e0d11e5f2f52e940c1464de74408d45bd574c081a4dffb", size = 55506, upload-time = "2026-08-03T10:45:27.421Z" }, +] + [[package]] name = "langchain-protocol" version = "0.0.18" @@ -3074,6 +3090,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/eb/a6/83dc2ab6fa397ee66fba04fe2e74bdf7be3b3870005359ceb7689103c058/opentelemetry_semantic_conventions-0.62b1-py3-none-any.whl", hash = "sha256:cf506938103d331fbb78eded0d9788095f7fd59016f2bda813c3324e5a74a93c", size = 231620, upload-time = "2026-04-24T13:15:35.454Z" }, ] +[[package]] +name = "openviking-sdk" +version = "0.1.6" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "httpx" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/6b/65/021ba0a777750d536b70e7ca5c14888d9e906627b80a9332b2c644082480/openviking_sdk-0.1.6.tar.gz", hash = "sha256:7ecfdebdbe3538556584e4348f19f1c990831c739404ee4e06e251f3d8a6641d", size = 40041, upload-time = "2026-08-03T07:15:38.963Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/6b/86/e0a9ff16d5b5eb38871450110b0ba64bff7a46020c617c482d4ce0ea3e7a/openviking_sdk-0.1.6-py3-none-any.whl", hash = "sha256:61e55f51f2733b80950488bb57299c5fd1bface388242ebff02c245ab6e04bd5", size = 25058, upload-time = "2026-08-03T07:15:37.776Z" }, +] + [[package]] name = "orjson" version = "3.11.8" diff --git a/config.example.yaml b/config.example.yaml index 56af387d1..8c3c425ed 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -1699,17 +1699,18 @@ memory: # Backend-private config (a dict), passed verbatim to the backend __init__. # Each backend self-interprets it (DeerMem parses it into DeerMemConfig). # - # OpenViking HTTP example (replace this DeerMem backend_config block when - # manager_class is openviking; OpenViking currently supports middleware mode): + # OpenViking example (replace this DeerMem backend_config block when + # manager_class is openviking). This first official-adapter integration uses + # one credential-bound OpenViking USER key for one DeerFlow user and supports + # middleware mode only. Use owner_user_id: default when DeerFlow auth is off. # # backend_config: # base_url: http://openviking:1933 - # auth_mode: trusted - # account: deerflow + # owner_user_id: default # api_key_env: OPENVIKING_API_KEY - # max_connections: 100 - # max_keepalive_connections: 20 - # max_seen_message_ids: 512 # recent-ID fallback for compacted histories + # timeout_seconds: 30 + # default_peer_id: deerflow + # max_seen_message_ids: 512 # bounded hash-only capture cursor # startup_policy: fail_fast # failure_policy: # read: fail_open @@ -1718,6 +1719,10 @@ memory: # top_k: 8 # score_threshold: 0.25 # max_injection_chars: 12000 + # content_mode: overview + # injection_query: >- + # user profile preferences important entities events ongoing goals + # constraints and prior decisions # # For a host-installed OpenViking used by Docker DeerFlow, set base_url to # http://host.docker.internal:1933 and allow_insecure_http: true. The bundled diff --git a/docs/OPENVIKING.md b/docs/OPENVIKING.md index 6bee05eed..489a6a540 100644 --- a/docs/OPENVIKING.md +++ b/docs/OPENVIKING.md @@ -1,50 +1,73 @@ # OpenViking memory backend DeerFlow can use a remote OpenViking server as an optional long-term memory -backend. The integration is a pluggable `MemoryManager`; DeerMem remains the -default and the agent/Gateway runtime does not import the OpenViking Python -runtime. +backend. DeerMem remains the default. The OpenViking backend uses the maintained +[`langchain-openviking`](https://pypi.org/project/langchain-openviking/) +package instead of implementing OpenViking's HTTP protocol inside DeerFlow. -## Supported behavior +## Current scope -- HTTP connection to an independent OpenViking server. -- Passive `memory.mode: middleware` capture after each completed turn. -- OpenViking Session commit and asynchronous memory extraction. -- Automatic prompt injection from OpenViking memory search. -- Explicit search through the backend-neutral `MemoryManager.search` API. -- Hard isolation by hashing each DeerFlow `(user_id, agent_name)` scope into a - separate OpenViking trusted user identity. -- Local bounded message watermarks under DeerFlow's runtime home. An ordered - prefix digest handles append-only histories of any length, while a recent-ID - window handles history compaction without resubmitting known messages. +The first official-adapter integration deliberately preserves DeerFlow's +existing automatic-memory behavior: -The current backend does not implement DeerMem fact CRUD, import/export, or the -Settings memory document. Keep `mode: middleware`; tool mode is rejected -because its add/update/delete tools require fact CRUD. +- memory is recalled through DeerFlow's existing fixed memory query; +- completed turns are captured by the existing memory middleware; +- messages about to be compacted are captured by the existing summarization hook; +- every accepted capture is committed to the thread's stable OpenViking Session; +- the official adapter handles message conversion, tool calls and results, + 100-message batching, partial-write progress, commit retry, and SDK transport; +- one recorder-owned SDK client is shared with retrieval and closed through + DeerFlow's existing memory shutdown contract. -## OpenViking requirements +This backend supports `memory.mode: middleware`. It does not implement DeerMem +fact CRUD, import/export, or the Settings memory-document view. OpenViking MCP +tools are a separate integration surface and are not enabled by this backend. -OpenViking must be configured with: +## Authentication boundary -- a VLM provider; -- an embedding provider; -- persistent workspace storage; -- `server.auth_mode: trusted`; -- a non-empty `server.root_api_key` when exposed beyond localhost. +This version is for one DeerFlow user backed by one ordinary OpenViking **USER +API key**. OpenViking derives the account and user from that credential. +DeerFlow does not configure trusted account/user headers and must not receive a +root key for normal memory traffic. -DeerFlow passes trusted `X-OpenViking-Account` and -`X-OpenViking-User` headers. Do not expose a trusted-mode OpenViking endpoint -directly to untrusted clients. +The supported server configuration is OpenViking `api_key` mode, where the USER +key determines the account and user. DeerFlow supplies its URL and API key +explicitly, overrides any ambient actor peer during memory operations, and does +not inherit arbitrary HTTP headers from `ovcli.conf`. + +Before enabling this backend, remove legacy `OPENVIKING_ACCOUNT` and +`OPENVIKING_USER` values from DeerFlow's repository-root `.env` and service +environment, and remove `account` and `user` defaults from +`~/.openviking/ovcli.conf`. Those settings belong to trusted-mode +configurations and are outside this adapter's supported setup. + +`owner_user_id` binds the configured key to one DeerFlow identity. Use +`default` when DeerFlow authentication is disabled. In an authenticated +single-user deployment, use that user's DeerFlow ID. A request for another +DeerFlow user is rejected before OpenViking is contacted, preventing one USER +key from silently sharing memory across users. + +Multi-user credential provisioning and storage are intentionally outside this +first adapter PR. + +Existing trusted-mode configurations are not migrated automatically. Configure +the OpenViking server in `api_key` mode, replace `auth_mode`, `account`, and the +root key with `owner_user_id` and a USER key, and remove the legacy ambient +identity settings listed above. Because the credential-bound user and Session +mapping differ from the old trusted-user mapping, previously captured +trusted-mode data remains in its old OpenViking namespace rather than being +silently reassigned. ## Configure DeerFlow -Put the trusted OpenViking key in the repository root `.env`: +Create or select an OpenViking user, then copy its USER API key into DeerFlow's +repository-root `.env`: ```dotenv -OPENVIKING_API_KEY=replace-with-the-same-root-api-key +OPENVIKING_API_KEY=replace-with-an-openviking-user-api-key ``` -Replace the `memory` section in `config.yaml` with: +Select the backend in `config.yaml`: ```yaml memory: @@ -54,13 +77,9 @@ memory: manager_class: openviking mode: middleware backend_config: - base_url: http://openviking:1933 - auth_mode: trusted - account: deerflow + base_url: http://127.0.0.1:1933 + owner_user_id: default api_key_env: OPENVIKING_API_KEY - max_connections: 100 - max_keepalive_connections: 20 - max_seen_message_ids: 512 startup_policy: fail_fast failure_policy: read: fail_open @@ -69,118 +88,22 @@ memory: top_k: 8 score_threshold: 0.25 max_injection_chars: 12000 + content_mode: overview + injection_query: >- + user profile preferences important entities events ongoing goals + constraints and prior decisions ``` -For a locally installed DeerFlow process, use -`http://127.0.0.1:1933`. For a DeerFlow container connecting to OpenViking on -the host, use `http://host.docker.internal:1933` and set -`allow_insecure_http: true`. +For a host-installed OpenViking used by Docker DeerFlow, set `base_url` to +`http://host.docker.internal:1933` and `allow_insecure_http: true`. The optional +Compose overlay uses the internal `http://openviking:1933` address. -`max_connections` and `max_keepalive_connections` bound the shared HTTP -connection pool. `max_seen_message_ids` bounds only the recent-ID fallback used -when a conversation is compacted or rewritten; append-only histories are -tracked by a constant-size prefix digest and do not depend on that window. +The dependency on `langchain-openviking==0.1.0` is declared by DeerFlow's +harness package and is installed by the normal `uv sync` flow. -## Docker first-time startup +## Start the services -Create the standard DeerFlow local files if they no longer exist: - -```bash -make config -cp .env.example .env -cp frontend/.env.example frontend/.env -``` - -Set at least the normal DeerFlow secrets in `.env`, including -`BETTER_AUTH_SECRET`, model provider credentials, and -`OPENVIKING_API_KEY`. - -The production Compose file expects the same path variables normally exported -by `scripts/deploy.sh`. Export them before using the OpenViking overlay -directly: - -```bash -export DEER_FLOW_CONFIG_PATH="$PWD/config.yaml" -export DEER_FLOW_EXTENSIONS_CONFIG_PATH="$PWD/extensions_config.json" -export DEER_FLOW_HOME="$PWD/backend/.deer-flow" -export DEER_FLOW_REPO_ROOT="$PWD" -``` - -Start only OpenViking: - -```bash -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - up -d openviking -``` - -Initialize it interactively: - -```bash -docker exec -it deer-flow-openviking openviking-server init -``` - -Choose trusted authentication, configure the same root API key stored in -`OPENVIKING_API_KEY`, and configure VLM and embedding providers. Validate the -configuration: - -```bash -docker exec -it deer-flow-openviking openviking-server doctor -docker restart deer-flow-openviking -curl http://localhost:1933/health -``` - -Then start DeerFlow with the same overlay: - -```bash -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - up -d --build -``` - -Open DeerFlow at and OpenViking Studio at -. - -## Routine Docker operations - -```bash -# Logs -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - logs -f gateway openviking - -# Stop containers but retain data -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - down - -# Restart -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - up -d - -# Pull a newer OpenViking image and recreate -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - pull openviking -docker compose \ - -f docker/docker-compose.yaml \ - -f docker/docker-compose.openviking.yaml \ - up -d openviking -``` - -Do not add `-v` to `docker compose down` unless you intentionally want to -delete Redis and OpenViking persistent volumes. - -## Local-process startup - -Run OpenViking separately and verify: +For a local OpenViking process, start and verify the server first: ```bash openviking-server doctor @@ -188,36 +111,67 @@ openviking-server curl http://127.0.0.1:1933/health ``` -Set `base_url: http://127.0.0.1:1933`, then start DeerFlow normally: +Then start DeerFlow normally: ```bash make doctor make dev ``` -The DeerFlow entrypoint is . +DeerFlow is available at . OpenViking Studio is +available at . -## Failure behavior +To use the optional Docker service instead: -- Invalid backend configuration fails loudly; DeerFlow never silently writes - to DeerMem instead. -- With `read: fail_open`, retrieval failures produce no injected memory and - the main agent continues. -- With `write: log_and_drop`, a failed OpenViking commit is logged without - failing an already generated assistant response. -- Once a message batch is accepted, DeerFlow persists a submitted-message - watermark before committing the Session. If commit then fails, later updates - do not resubmit those messages or retry the ambiguous commit; a future batch - can commit the still-open Session together with new messages. -- Retried health, session lookup, and search requests use exponential backoff - with jitter so concurrent Gateway workers do not retry in lockstep. -- OpenViking commit is eventually consistent: accepting a commit archives the - messages immediately, while summary and memory extraction finish in a - background task. -- Graceful shutdown stops admitting new memory operations, waits up to - `shutdown_flush_timeout_seconds` for active reads and writes, and closes the - shared HTTP client only after they drain. +```bash +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + up -d openviking + +docker exec -it deer-flow-openviking openviking-server init + +docker compose \ + -f docker/docker-compose.yaml \ + -f docker/docker-compose.openviking.yaml \ + up -d --build +``` + +Configure OpenViking in API-key mode and obtain a USER key through its identity +management flow. Only that USER key belongs in DeerFlow's +`OPENVIKING_API_KEY` variable. + +## Identity and session mapping + +One DeerFlow thread maps deterministically to one OpenViking Session. A commit +creates an archive inside that Session; it does not create a new Session, so a +thread keeps the same identity when the user returns later. + +The default DeerFlow agent uses `default_peer_id` (`deerflow` by default). +Named agents use lowercase OpenViking peer IDs. Names that are not valid peer +IDs, conflict with the default, or enter the reserved `df-agent-` namespace are +mapped to collision-resistant IDs. USER-key identity remains the security +boundary; peers separate memory scopes within that user. + +## Retry and failure behavior + +- `read: fail_open` logs retrieval failures and returns no injected OpenViking + memory. `read: raise` propagates the retrieval failure to its DeerFlow caller. +- `write: log_and_drop` logs capture failures without failing an already + generated answer. `write: raise` propagates them. +- DeerFlow stores only hashes and counters in a bounded local capture cursor + under `{storage_path}/openviking/sessions/`. It never stores message text + there. +- The cursor prevents full LangGraph transcript snapshots from being submitted + again. It also records confirmed progress from partial batches and retries a + failed commit before appending more messages. +- An unreadable cursor fails closed because replaying an unknown prefix could + duplicate private conversation history. +- Graceful shutdown stops new memory work, waits up to + `shutdown_flush_timeout_seconds` for accepted operations, and closes the + recorder-owned SDK client. It does not introduce a new DeerFlow lifecycle or + background worker. For deployments where a lost memory update is unacceptable, a durable outbox -is still required; the initial plugin intentionally does not claim -at-least-once delivery. +is still required. This initial integration does not claim at-least-once +delivery. diff --git a/frontend/src/content/en/application/configuration.mdx b/frontend/src/content/en/application/configuration.mdx index a79c7e8d2..c0ec7b140 100644 --- a/frontend/src/content/en/application/configuration.mdx +++ b/frontend/src/content/en/application/configuration.mdx @@ -237,10 +237,11 @@ memory: # api_key: $OPENAI_API_KEY ``` -The optional `openviking` backend connects to an independent OpenViking HTTP -server and currently supports `mode: middleware`. Its private configuration -uses `base_url`, `auth_mode`, `account`, `api_key_env`, `failure_policy`, and -`retrieval` instead of the DeerMem fields shown above. See +The optional `openviking` backend connects to an independent OpenViking server +through `langchain-openviking` and currently supports one DeerFlow user in +`mode: middleware`. Its private configuration uses `base_url`, +`owner_user_id`, `api_key_env`, `failure_policy`, and `retrieval` with an +ordinary OpenViking USER API key instead of the DeerMem fields shown above. See `docs/OPENVIKING.md` in the repository for the complete configuration and Docker startup sequence. diff --git a/frontend/src/content/en/harness/memory.mdx b/frontend/src/content/en/harness/memory.mdx index 19d70a46d..e0f4ea906 100644 --- a/frontend/src/content/en/harness/memory.mdx +++ b/frontend/src/content/en/harness/memory.mdx @@ -89,11 +89,13 @@ memory: max_injection_tokens: 2000 ``` -## OpenViking HTTP backend +## OpenViking backend Set `manager_class: openviking` to send completed turns to an independent -OpenViking server and recall its memory over HTTP. This backend currently -supports middleware mode only; DeerMem remains the default. +OpenViking server and recall its memory through the official +`langchain-openviking` adapter. This first version supports one DeerFlow user +with one OpenViking USER API key in middleware mode; DeerMem remains the +default. ```yaml memory: @@ -103,8 +105,7 @@ memory: mode: middleware backend_config: base_url: http://openviking:1933 - auth_mode: trusted - account: deerflow + owner_user_id: default api_key_env: OPENVIKING_API_KEY failure_policy: read: fail_open @@ -115,9 +116,11 @@ memory: max_injection_chars: 12000 ``` -The backend hashes each DeerFlow user/agent scope into a distinct OpenViking -trusted identity. Keep trusted-mode OpenViking on an internal network and put -the API key in the server environment, not directly in `config.yaml`. +Use `owner_user_id: default` when DeerFlow authentication is disabled. Put the +USER key in the server environment, not directly in `config.yaml`. Trusted-mode +account headers, root-key memory access, and multi-user key provisioning are not +part of this version. See `docs/OPENVIKING.md` for the full boundary and startup +guide. ## Global vs per-agent memory