feat(memory): add OpenViking HTTP backend (#4509)

* feat(memory): add OpenViking HTTP backend

* fix(memory): harden OpenViking lifecycle

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
RongfuShuiping 2026-07-28 23:36:25 +08:00 committed by GitHub
parent a5059b8284
commit 2aaf74b0f8
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
19 changed files with 2769 additions and 8 deletions

View File

@ -17,6 +17,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
# 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

View File

@ -960,6 +960,15 @@ Then uncomment the `group: browser` tool entries in `config.yaml` (`browser_navi
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`. Submitted-message watermarks prevent a failed
Session commit from duplicating already accepted messages on retry; see
[OpenViking memory backend](docs/OPENVIKING.md) for configuration and Docker
startup.
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.
Memory updates now skip duplicate fact entries at apply time, so repeated preferences and context do not accumulate endlessly across sessions.

View File

@ -229,7 +229,10 @@ Blocking-IO runtime gate (`tests/blocking_io/`):
offloading the uploads-directory scan off the event loop);
`test_uploads_router.py` (locks Gateway upload/list/delete endpoints
offloading upload directory creation, staged writes, chmod/cleanup,
directory scans/deletes, and remote sandbox sync off the event loop); and
directory scans/deletes, and remote sandbox sync off the event loop);
`test_openviking_memory_backend.py` (locks the OpenViking backend's async
add/context/search entrypoints offloading synchronous HTTP and watermark
filesystem IO); and
`test_workspace_changes_recorder.py` (locks the offload around the snapshot
text cache lifecycle — roots resolution, `mkdtemp`, and the `shutil.rmtree`
on both the capture-failure branch and `record_workspace_changes`' `finally`).
@ -847,6 +850,22 @@ The cached value is reused for both the blocking (`runs.wait`) and streaming (`_
**Workflow**:
- `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
`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 separately records
submitted and committed message IDs: 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. 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.
- `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, prompt injection, manual CRUD primitives, and the updater backend.
- Middleware mode queue debounces (30s default), batches updates, and commits global summaries plus the selected/default agent's fact delta through a user-level lock, optimistic user-memory revisions, per-fact revisions, and a recoverable target-file journal. Only explicitly marked point operations may rebase a stale shared revision, and only while every addressed fact still satisfies its original absent/revision precondition. Snapshot-derived clear/trim/consolidation operations instead reload the complete document and recompute their intent on a manifest conflict, with a bounded retry. Typed manifest/fact conflict subclasses keep that decision independent of exception text, and same-ID creates and stale same-fact writes fail. Scope-lock objects are weakly cached so inactive users do not grow a process-lifetime map. Cache validation does not scale with the fact-file count: its token combines the shared JSON's `(mtime_ns, size, revision)`, so the persisted revision invalidates stale caches even when a coarse-mtime filesystem reports identical metadata for same-size writes; direct out-of-band Markdown edits require `reload()`. Atomic replacement also syncs the parent directory on POSIX so the rename is durable. DeerMem translates private storage conflict/corruption exceptions to the backend-neutral MemoryManager contract; the Gateway maps them to HTTP 409 and a stable HTTP 500 response respectively. A normal default-manager read automatically migrates legacy facts from the global JSON into `__default__`; it also adopts the earlier implicit `lead-agent` fact bucket only when that directory has no custom-agent `config.yaml`, and rejects unexpected files instead of deleting them. The v1-to-v2 migration is one-way for the running application: operators must stop DeerFlow and snapshot the configured storage root before upgrade. Before any destructive v2 write, every migrated JSON source is durably retained as `{manifest_filename}.v1.bak`; a missing-write or mismatched existing backup aborts without modifying v1 data. Legacy per-agent JSON is deleted only after its non-empty summaries are safely adopted or confirmed identical; summary conflicts keep the source file and fail loudly.

View File

@ -4,6 +4,7 @@ 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).
This guide tells you **which files to touch** when you change, swap, or add a memory system. Paths are relative to `backend/` unless noted.

View File

@ -0,0 +1,7 @@
"""OpenViking HTTP memory backend."""
from .openviking_manager import OpenVikingMemoryManager
MANAGER_CLASS = OpenVikingMemoryManager
__all__ = ["OpenVikingMemoryManager"]

View File

@ -0,0 +1,232 @@
"""Thin synchronous HTTP client for the OpenViking server API."""
from __future__ import annotations
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,
)
self._client = httpx.Client(base_url=config.base_url, timeout=timeout, 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(0.05 * (2**attempt))
continue
raise OpenVikingTimeoutError(operation, "OpenViking request timed out") from exc
except httpx.TransportError as exc:
if attempt + 1 < attempts:
time.sleep(0.05 * (2**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(0.05 * (2**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")

View File

@ -0,0 +1,127 @@
"""Configuration for the OpenViking HTTP memory backend."""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any, Literal
from urllib.parse import urlparse
@dataclass(frozen=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.
"""
base_url: str
storage_path: str
auth_mode: Literal["trusted", "dev"]
account: str
api_key: str | None
connect_timeout_seconds: float
read_timeout_seconds: float
write_timeout_seconds: float
pool_timeout_seconds: float
max_retries: int
search_top_k: int
score_threshold: float | None
max_injection_chars: int
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:
cfg = dict(backend_config or {})
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
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_retries=int(cfg.pop("max_retries", 1)),
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)),
injection_query=str(
retrieval.pop(
"injection_query",
"user profile preferences important entities events ongoing goals constraints and prior decisions",
)
).strip(),
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)),
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)])
if unknown:
raise ValueError(f"Unknown OpenViking backend_config fields: {', '.join(unknown)}")
result._validate()
return result
def _validate(self) -> None:
parsed = urlparse(self.base_url)
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 0 <= self.max_retries <= 5:
raise ValueError("OpenViking max_retries must be between 0 and 5")
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 not 256 <= self.max_injection_chars <= 100_000:
raise ValueError("OpenViking retrieval.max_injection_chars must be between 256 and 100000")
if not self.injection_query:
raise ValueError("OpenViking retrieval.injection_query must not be empty")
if self.startup_policy not in {"fail_fast", "warn"}:
raise ValueError("OpenViking startup_policy must be 'fail_fast' or 'warn'")
if self.read_failure_policy not in {"fail_open", "raise"}:
raise ValueError("OpenViking failure_policy.read must be 'fail_open' or 'raise'")
if self.write_failure_policy not in {"log_and_drop", "raise"}:
raise ValueError("OpenViking failure_policy.write must be 'log_and_drop' or 'raise'")
if not 16 <= self.max_seen_message_ids <= 10_000:
raise ValueError("OpenViking max_seen_message_ids must be between 16 and 10000")
def _mapping(value: Any, name: str) -> dict[str, Any]:
if value is None:
return {}
if not isinstance(value, dict):
raise ValueError(f"OpenViking {name} must be a mapping")
return dict(value)
def _optional_float(value: Any) -> float | None:
return None if value is None else float(value)

View File

@ -0,0 +1,63 @@
"""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

View File

@ -0,0 +1,494 @@
"""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.
"""
from __future__ import annotations
import asyncio
import hashlib
import json
import logging
import os
import re
import threading
import time
import weakref
from collections.abc import Callable
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 .client import OpenVikingClientError, OpenVikingHttpClient
from .config import OpenVikingConfig
from .models import OpenVikingIdentity, OpenVikingMessage, OpenVikingSearchHit
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."""
supports_search: ClassVar[bool] = True
_config: OpenVikingConfig = PrivateAttr()
_client: OpenVikingHttpClient = PrivateAttr()
_should_keep_hidden_message: Callable[[Any], bool] | None = 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)
def model_post_init(self, __context: Any) -> None:
self._config = OpenVikingConfig.from_backend_config(self.backend_config)
self._client = OpenVikingHttpClient(self._config)
@classmethod
def from_config(
cls,
backend_config: dict[str, Any] | None = None,
*,
mode: Literal["middleware", "tool"] = "middleware",
**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
return instance
def add(
self,
thread_id: str,
messages: list[Any],
*,
agent_name: str | None = None,
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)
def add_nowait(
self,
thread_id: str,
messages: list[Any],
*,
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)
async def aadd(
self,
thread_id: str,
messages: list[Any],
*,
agent_name: str | None = None,
user_id: str | None = None,
trace_id: str | None = None,
) -> None:
await asyncio.to_thread(
self.add,
thread_id,
messages,
agent_name=agent_name,
user_id=user_id,
trace_id=trace_id,
)
def get_context(
self,
user_id: str | None,
*,
agent_name: str | None = None,
thread_id: str | None = None,
) -> str:
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,
)
except OpenVikingClientError:
if self._config.read_failure_policy == "raise":
raise
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)
finally:
self._end_operation()
async def aget_context(
self,
user_id: str | None,
*,
agent_name: str | None = None,
thread_id: str | None = None,
) -> str:
return await asyncio.to_thread(
self.get_context,
user_id,
agent_name=agent_name,
thread_id=thread_id,
)
def search(
self,
query: str,
top_k: int = 5,
*,
user_id: str | None = None,
agent_name: str | None = None,
category: str | None = None,
) -> list[dict[str, Any]]:
if not query.strip():
return []
if not self._begin_operation():
return []
try:
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:
if self._config.read_failure_policy == "raise":
raise
logger.warning("OpenViking memory search failed; returning no results", exc_info=True)
return []
return [_hit_to_fact(hit) for hit in hits]
finally:
self._end_operation()
async def asearch(
self,
query: str,
top_k: int = 5,
*,
user_id: str | None = None,
agent_name: str | None = None,
category: str | None = None,
) -> list[dict[str, Any]]:
return await asyncio.to_thread(
self.search,
query,
top_k,
user_id=user_id,
agent_name=agent_name,
category=category,
)
def warm(self) -> bool | None:
if not self._begin_operation():
return False
try:
try:
healthy = self._client.health()
except OpenVikingClientError:
if self._config.startup_policy == "fail_fast":
raise
logger.warning("OpenViking health check 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")
if not healthy:
logger.warning("OpenViking health check returned unhealthy; memory will run in degraded mode")
return healthy
finally:
self._end_operation()
def shutdown_flush(self, timeout: float) -> bool:
"""Stop new operations, drain in-flight work, then close the HTTP pool."""
deadline = time.monotonic() + max(0.0, timeout)
with self._lifecycle:
self._closed = 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()
return True
def _write_conversation(
self,
thread_id: str,
messages: list[Any],
*,
agent_name: str | None,
user_id: str | None,
) -> None:
if not self._begin_operation():
logger.warning("OpenViking write ignored after backend shutdown")
return
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", [])))
submitted = set(submitted_ids)
converted = _convert_messages(messages, self._should_keep_hidden_message)
pending = [message for message in converted if message.message_id not in submitted]
if not pending:
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": 2,
"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 :],
"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": 2,
"committed_message_ids": state["submitted_message_ids"],
"last_commit_task_id": commit.task_id,
"last_archive_uri": commit.archive_uri,
}
self._save_state(session_id, state)
finally:
self._end_operation()
def _search_hits(
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,
)
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 _session_lock(self, session_id: str) -> threading.RLock:
with self._session_locks_guard:
return self._session_locks.setdefault(session_id, threading.RLock())
def _begin_operation(self) -> bool:
with self._lifecycle:
if self._closed:
return False
self._active_operations += 1
return True
def _end_operation(self) -> None:
with self._lifecycle:
self._active_operations -= 1
if self._active_operations == 0:
self._lifecycle.notify_all()
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]:
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 {}
def _save_state(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")
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)
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 _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]:
return {
"id": hit.uri,
"content": _hit_content(hit),
"category": hit.category or "memory",
"confidence": hit.score,
"source": hit.uri,
"score": hit.score,
}
def _format_context(hits: list[OpenVikingSearchHit], *, max_chars: int) -> str:
lines: list[str] = []
seen: set[str] = set()
for hit in hits:
content = " ".join(_hit_content(hit).split())
key = content.casefold()
if not content or key in seen:
continue
seen.add(key)
line = f"- [{hit.category or 'memory'}] {content}"
candidate = "\n".join([*lines, line])
if len(candidate) > max_chars:
remaining = max_chars - len("\n".join(lines)) - (1 if lines else 0)
if remaining > 16:
lines.append(f"{line[: max(0, remaining - 1)]}")
break
lines.append(line)
return "\n".join(lines)

View File

@ -0,0 +1,91 @@
"""Regression anchors: OpenViking async memory methods must not block the loop."""
from __future__ import annotations
from pathlib import Path
from typing import Any
import pytest
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
class _BlockingProbeClient:
"""Perform real file IO so Blockbuster can detect missing async offload."""
def __init__(self, probe_path: Path):
self._probe_path = probe_path
def _probe(self) -> None:
self._probe_path.write_text("probe", encoding="utf-8")
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 close(self) -> None:
pass
def _manager(tmp_path: Path) -> OpenVikingMemoryManager:
manager = OpenVikingMemoryManager.from_config(
{
"base_url": "http://openviking:1933",
"storage_path": str(tmp_path),
"auth_mode": "trusted",
"account": "deerflow",
"startup_policy": "warn",
}
)
manager._client = _BlockingProbeClient(tmp_path / "probe.txt") # type: ignore[assignment]
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")]
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.",
"category": "preferences",
"confidence": 0.9,
"source": "viking://user/memories/preferences/test.md",
"score": 0.9,
}
]

View File

@ -0,0 +1,368 @@
from __future__ import annotations
import gc
import json
import threading
import weakref
from pathlib import Path
from typing import Any
import httpx
import pytest
from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
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
from deerflow.agents.memory.manager import _scan_backends, reset_memory_manager
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",
"startup_policy": "warn",
"retrieval": {"top_k": 4, "max_injection_chars": 1000},
}
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 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_backend_is_discovered_by_registered_name() -> None:
reset_memory_manager()
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"
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 ensure_session(self, identity: OpenVikingIdentity, session_id: str) -> None:
self.ensured.append((identity, session_id))
def add_messages(
self,
identity: OpenVikingIdentity,
session_id: str,
messages: list[OpenVikingMessage],
) -> int:
self.added.append((identity, session_id, messages))
return len(messages)
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 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
def _manager(tmp_path: Path) -> tuple[OpenVikingMemoryManager, _FakeClient]:
manager = OpenVikingMemoryManager.from_config(_backend_config(tmp_path))
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)
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"),
]
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
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(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"]
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 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)
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"
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")
assert results == [
{
"id": "viking://user/memories/preferences/editor.md",
"content": "User prefers concise answers.",
"category": "preferences",
"confidence": 0.9,
"source": "viking://user/memories/preferences/editor.md",
"score": 0.9,
}
]
assert context == "- [preferences] User prefers concise answers."
assert client.searches[1][1] == manager._config.injection_query
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_manager_session_locks_do_not_accumulate(tmp_path: Path) -> None:
manager, _ = _manager(tmp_path)
lock = manager._session_lock("session-1")
lock_ref = weakref.ref(lock)
assert len(manager._session_locks) == 1
del lock
gc.collect()
assert lock_ref() is 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 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)
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 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)
client.add_messages = blocking_add # type: ignore[method-assign]
write_thread = threading.Thread(
target=manager.add,
args=("thread-1", [HumanMessage("hello", id="h1")]),
kwargs={"user_id": "alice"},
)
write_thread.start()
assert write_started.wait(1)
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
release_write.set()
write_thread.join(2)
assert not write_thread.is_alive()
assert manager.shutdown_flush(1) is True
assert client.closed is True

View File

@ -1618,7 +1618,7 @@ summarization:
# enabled - Master switch for the memory mechanism (call-site gate)
# injection_enabled - Whether to inject memory into the system prompt (call-site gate)
# shutdown_flush_timeout_seconds - Hard budget (s) to drain pending updates on Gateway graceful shutdown (default: 30)
# manager_class - Backend selector: registered name (deermem/noop) or dotted path
# manager_class - Backend selector: registered name (deermem/noop/openviking) or dotted path
# backend_config - Backend-private config dict (passthrough; each backend self-interprets)
#
# DeerMem-private fields live under ``backend_config`` (NOT at the memory: top level):
@ -1671,6 +1671,27 @@ memory:
mode: middleware
# 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):
#
# backend_config:
# base_url: http://openviking:1933
# auth_mode: trusted
# account: deerflow
# api_key_env: OPENVIKING_API_KEY
# startup_policy: fail_fast
# failure_policy:
# read: fail_open
# write: log_and_drop
# retrieval:
# top_k: 8
# score_threshold: 0.25
# max_injection_chars: 12000
#
# 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
# optional Compose overlay uses the internal http://openviking:1933 address.
backend_config:
storage_path: "" # empty = deer-flow base_dir (factory injects absolute runtime_home); a non-empty path is the root DIRECTORY (per-user memory under {storage_path}/users/{uid}/memory.json)
storage_class: file # file (default) or a dotted MemoryStorage class path; invalid persistent backends fail fast

View File

@ -208,8 +208,8 @@ services:
- PROVISIONER_API_KEY=${PROVISIONER_API_KEY:-}
# Proxy values (HTTP_PROXY/HTTPS_PROXY/ALL_PROXY) are inherited from ../.env via env_file.
# Only NO_PROXY is declared here so internal service hostnames are always exempt from the proxy.
- NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal
- no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal
- NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal
- no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal
env_file:
- ../.env
extra_hosts:

View File

@ -0,0 +1,42 @@
# Optional OpenViking memory service overlay.
#
# First-time initialization:
# 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
#
# Full stack:
# docker compose -f docker/docker-compose.yaml \
# -f docker/docker-compose.openviking.yaml up -d --build
services:
openviking:
image: ${OPENVIKING_IMAGE:-ghcr.io/volcengine/openviking:latest}
container_name: deer-flow-openviking
command: ["--without-bot"]
ports:
- "${OPENVIKING_PORT:-1933}:1933"
volumes:
- openviking-data:/app/.openviking
networks:
- deer-flow
restart: unless-stopped
healthcheck:
test:
[
"CMD",
"python",
"-c",
"import urllib.request; urllib.request.urlopen('http://127.0.0.1:1933/health', timeout=3)",
]
interval: 10s
timeout: 5s
retries: 12
gateway:
depends_on:
openviking:
condition: service_healthy
volumes:
openviking-data:

View File

@ -131,8 +131,8 @@ services:
- DEER_FLOW_SANDBOX_HOST=host.docker.internal
# Proxy values (HTTP_PROXY/HTTPS_PROXY/ALL_PROXY) are inherited from ../.env via env_file.
# Only NO_PROXY is declared here so internal service hostnames are always exempt from the proxy.
- NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal
- no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,host.docker.internal
- NO_PROXY=${NO_PROXY:-}${NO_PROXY:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal
- no_proxy=${no_proxy:-}${no_proxy:+,}localhost,127.0.0.1,::1,gateway,frontend,nginx,provisioner,openviking,host.docker.internal
env_file:
- ../.env
extra_hosts:

212
docs/OPENVIKING.md Normal file
View File

@ -0,0 +1,212 @@
# 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.
## Supported behavior
- 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 message watermarks under DeerFlow's runtime home to avoid resubmitting
the same conversation history in a single-Gateway deployment.
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.
## OpenViking requirements
OpenViking must be configured with:
- a VLM provider;
- an embedding provider;
- persistent workspace storage;
- `server.auth_mode: trusted`;
- a non-empty `server.root_api_key` when exposed beyond localhost.
DeerFlow passes trusted `X-OpenViking-Account` and
`X-OpenViking-User` headers. Do not expose a trusted-mode OpenViking endpoint
directly to untrusted clients.
## Configure DeerFlow
Put the trusted OpenViking key in the repository root `.env`:
```dotenv
OPENVIKING_API_KEY=replace-with-the-same-root-api-key
```
Replace the `memory` section in `config.yaml` with:
```yaml
memory:
enabled: true
injection_enabled: true
shutdown_flush_timeout_seconds: 30
manager_class: openviking
mode: middleware
backend_config:
base_url: http://openviking:1933
auth_mode: trusted
account: deerflow
api_key_env: OPENVIKING_API_KEY
startup_policy: fail_fast
failure_policy:
read: fail_open
write: log_and_drop
retrieval:
top_k: 8
score_threshold: 0.25
max_injection_chars: 12000
```
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`.
## Docker first-time startup
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 <http://localhost:2026> and OpenViking Studio at
<http://localhost:1933/studio>.
## 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:
```bash
openviking-server doctor
openviking-server
curl http://127.0.0.1:1933/health
```
Set `base_url: http://127.0.0.1:1933`, then start DeerFlow normally:
```bash
make doctor
make dev
```
The DeerFlow entrypoint is <http://localhost:2026>.
## Failure behavior
- 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.
- 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.
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.

File diff suppressed because it is too large Load Diff

View File

@ -223,7 +223,7 @@ For Docker or scripted starts, set `UV_EXTRAS=postgres` before installing or bui
memory:
enabled: true
injection_enabled: true
manager_class: deermem # backend selector: deermem | noop | dotted path
manager_class: deermem # backend selector: deermem | noop | openviking | dotted path
mode: middleware # middleware (default) | tool (experimental)
backend_config: # DeerMem-private knobs (the backend self-interprets these)
storage_path: "" # empty = deer-flow base_dir (per-user memory under it)
@ -237,6 +237,13 @@ 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
`docs/OPENVIKING.md` in the repository for the complete configuration and
Docker startup sequence.
## Frontend environment variables
Set these before running `pnpm build` or starting the frontend in production:

View File

@ -45,7 +45,7 @@ memory:
enabled: true
injection_enabled: true
# Backend selector: deermem (default) | noop | a dotted path to a custom
# Backend selector: deermem (default) | noop | openviking | a dotted path to a custom
# MemoryManager subclass. Swap backend = drop a backends/<name>/ folder and
# set this (see backend/.../agents/memory/backends/).
manager_class: deermem
@ -89,6 +89,36 @@ memory:
max_injection_tokens: 2000
```
## OpenViking HTTP 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.
```yaml
memory:
enabled: true
injection_enabled: true
manager_class: openviking
mode: middleware
backend_config:
base_url: http://openviking:1933
auth_mode: trusted
account: deerflow
api_key_env: OPENVIKING_API_KEY
failure_policy:
read: fail_open
write: log_and_drop
retrieval:
top_k: 8
score_threshold: 0.25
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`.
## Global vs per-agent memory
DeerFlow supports two levels of memory: