OpenViking plugin memory write bug fixes

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #50833 fix(openviking): gate memory writes behind MemoryManager + add viking_forget

  merged to main

  files: agent/agent_runtime_helpers.py, agent/memory_manager.py, agent/tool_executor.py, plugins/memory/openviking/README.md, plugins/memory/openviking/__init__.py, tests/agent/test_memory_provider.py, tests/agent/test_memory_write_bridge.py, tests/plugins/memory/test_openviking_provider.py, tests/run_agent/test_run_agent.py

  Show the diff

  • agent/agent_runtime_helpers.py

    @@ -1838,32 +1838,18 @@ def _execute(next_args: dict) -> Any:
                     operations=operations,
                     store=agent._memory_store,
                 )
    -            # Bridge: notify external memory provider of built-in memory writes.
    -            # Covers both the single-op shape and each add/replace inside a batch.
    +            # Mirror successful built-in memory writes to external providers.
    +            # All gating/op-expansion lives behind the manager interface
    +            # (MemoryManager.notify_memory_tool_write).
                 if agent._memory_manager:
    -                if operations:
    -                    _mem_ops = [
    -                        op for op in operations
    -                        if isinstance(op, dict) and op.get("action") in {"add", "replace"}
    -                    ]
    -                else:
    -                    _mem_ops = (
    -                        [{"action": next_args.get("action"), "content": next_args.get("content")}]
    -                        if next_args.get("action") in {"add", "replace"} else []
    -                    )
    -                for _op in _mem_ops:
    -                    try:
    -                        agent._memory_manager.on_memory_write(
    -                            _op.get("action", ""),
    -                            target,
    -                            _op.get("content", "") or "",
    -                            metadata=agent._build_memory_write_metadata(
    -                                task_id=effective_task_id,
    -                                tool_call_id=tool_call_id,
    -                            ),
    -                        )
    -                    except Exception:
    -                        pass
    +                agent._memory_manager.notify_memory_tool_write(
    +                    result,
    +                    next_args,
    +                    build_metadata=lambda: agent._build_memory_write_metadata(
    +                        task_id=effective_task_id,
    +                        tool_call_id=tool_call_id,
    +                    ),
    +                )
                 return _finish_agent_tool(result, next_args)
         elif agent._memory_manager and agent._memory_manager.has_tool(function_name):
             def _execute(next_args: dict) -> Any:

  • agent/memory_manager.py

    @@ -25,12 +25,13 @@
     
     from __future__ import annotations
     
    +import json
     import logging
     import re
     import inspect
     import threading
     from concurrent.futures import ThreadPoolExecutor
    -from typing import Any, Dict, List, Optional
    +from typing import Any, Callable, Dict, List, Optional
     
     from agent.memory_provider import MemoryProvider
     from agent.skill_commands import extract_user_instruction_from_skill_message
    @@ -850,6 +851,87 @@ def on_memory_write(
                         provider.name, e,
                     )
     
    +    # Actions the bridge mirrors to external providers. The built-in memory
    +    # tool can also return non-mutating shapes (errors, staged-for-approval
    +    # records); those are filtered out by ``notify_memory_tool_write`` before
    +    # we ever reach a provider.
    +    _MIRRORED_MEMORY_ACTIONS = {"add", "replace", "remove"}
    +
    +    @staticmethod
    +    def _memory_tool_result_succeeded(result: Any) -> bool:
    +        """True only when the built-in memory tool actually committed a write.
    +
    +        Fails closed: a string that isn't JSON, a non-dict result, a missing
    +        ``success``, or a write staged for approval (``staged is True``) all
    +        return False so external providers are never told about a write that
    +        did not land.
    +        """
    +        if isinstance(result, str):
    +            try:
    +                result = json.loads(result)
    +            except Exception:
    +                return False
    +        if not isinstance(result, dict):
    +            return False
    +        return result.get("success") is True and result.get("staged") is not True
    +
    +    def notify_memory_tool_write(
    +        self,
    +        tool_result: Any,
    +        tool_args: Dict[str, Any],
    +        *,
    +        build_metadata: Optional[Callable[[], Dict[str, Any]]] = None,
    +    ) -> None:
    +        """Mirror a built-in memory tool call to external providers.
    +
    +        This is the single entry point the agent loop calls after running the
    +        built-in ``memory`` tool. All the decisions about *whether* and *what*
    +        to mirror live here, behind the manager interface — the loop only hands
    +        over the raw tool result and args:
    +
    +        * gate on a committed (non-staged, successful) write,
    +        * expand the single-op and batched (``operations``) shapes,
    +        * keep only mutating actions (add/replace/remove),
    +        * build per-op provenance metadata and forward ``old_text``.
    +
    +        ``build_metadata`` is an optional agent-side callable (the loop knows
    +        session/task/tool-call provenance the manager does not) invoked once per
    +        mirrored op.
    +        """
    +        if not self._memory_tool_result_succeeded(tool_result):
    +            return
    +
    +        target = str(tool_args.get("target") or "memory")
    +        operations = tool_args.get("operations")
    +        if isinstance(operations, list) and operations:
    +            raw_operations = operations
    +        else:
    +            raw_operations = [{
    +                "action": tool_args.get("action"),
    +                "content": tool_args.get("content"),
    +                "old_text": tool_args.get("old_text"),
    +            }]
    +
    +        for op in raw_operations:
    +            if not isinstance(op, dict):
    +                continue
    +            action = str(op.get("action") or "")
    +            if action not in self._MIRRORED_MEMORY_ACTIONS:
    +                continue
    +            try:
    +                metadata = dict(build_metadata() if build_metadata else {})
    +                old_text = op.get("old_text")
    +                if old_text:
    +                    metadata["old_text"] = str(old_text)
    +                self.on_memory_write(
    +                    action,
    +                    target,
    +                    str(op.get("content") or ""),
    +                    metadata=metadata,
    +                )
    +            except Exception as e:
    +                logger.debug("notify_memory_tool_write failed for op %s: %s", action, e)
    +
         def on_delegation(self, task: str, result: str, *,
                           child_session_id: str = "", **kwargs) -> None:
             """Notify all providers that a subagent completed."""

  • agent/tool_executor.py

    @@ -1046,32 +1046,18 @@ def _execute(next_args: dict) -> Any:
                         operations=operations,
                         store=agent._memory_store,
                     )
    -                # Bridge: notify external memory provider of built-in memory writes.
    -                # Covers both the single-op shape and each add/replace inside a batch.
    +                # Mirror successful built-in memory writes to external
    +                # providers. All gating/op-expansion lives behind the manager
    +                # interface (MemoryManager.notify_memory_tool_write).
                     if agent._memory_manager:
    -                    if operations:
    -                        _mem_ops = [
    -                            op for op in operations
    -                            if isinstance(op, dict) and op.get("action") in {"add", "replace"}
    -                        ]
    -                    else:
    -                        _mem_ops = (
    -                            [{"action": next_args.get("action"), "content": next_args.get("content")}]
    -                            if next_args.get("action") in {"add", "replace"} else []
    -                        )
    -                    for _op in _mem_ops:
    -                        try:
    -                            agent._memory_manager.on_memory_write(
    -                                _op.get("action", ""),
    -                                target,
    -                                _op.get("content", "") or "",
    -                                metadata=agent._build_memory_write_metadata(
    -                                    task_id=effective_task_id,
    -                                    tool_call_id=getattr(tool_call, "id", None),
    -                                ),
    -                            )
    -                        except Exception:
    -                            pass
    +                    agent._memory_manager.notify_memory_tool_write(
    +                        result,
    +                        next_args,
    +                        build_metadata=lambda: agent._build_memory_write_metadata(
    +                            task_id=effective_task_id,
    +                            tool_call_id=getattr(tool_call, "id", None),
    +                        ),
    +                    )
                     return result
                 function_result, function_args = _run_agent_tool_execution_middleware(
                     agent,

  • plugins/memory/openviking/README.md

    @@ -47,5 +47,37 @@ Hermes sends `OPENVIKING_ACCOUNT` and `OPENVIKING_USER` as identity headers.
     | `viking_search` | Semantic search with fast/deep/auto modes |
     | `viking_read` | Read content at a viking:// URI (abstract/overview/full) |
     | `viking_browse` | Filesystem-style navigation (list/tree/stat) |
    -| `viking_remember` | Store a fact for extraction on session commit |
    +| `viking_remember` | Store a fact directly with OpenViking `content/write` |
    +| `viking_forget` | Delete one exact `viking://` memory file URI |
     | `viking_add_resource` | Ingest URLs/docs into the knowledge base |
    +
    +## Memory Writes And Deletes
    +
    +`viking_remember` writes directly to OpenViking with `POST /api/v1/content/write`
    +and `mode=create`. It creates peer-scoped memory files under
    +`viking://user/peers/${OPENVIKING_AGENT}/memories/...`; OpenViking may return a
    +canonical user-scoped form such as
    +`viking://user/default/peers/${OPENVIKING_AGENT}/memories/...` in API-key mode.
    +Explicit remembers do not depend on session commit extraction.
    +
    +Hermes built-in `memory` tool additions are mirrored to OpenViking after the
    +local memory operation succeeds:
    +
    +| Hermes action | OpenViking operation |
    +|---------------|----------------------|
    +| `add` | `content/write` with `mode=create` under the configured peer memory namespace |
    +
    +Built-in `replace` and `remove` operations are not mirrored because Hermes
    +native memory entries do not yet carry stable OpenViking file URIs. Use
    +`viking_forget` when the user explicitly asks to delete a specific OpenViking
    +memory URI.
    +
    +`viking_forget` is intentionally narrow. It only accepts concrete user memory
    +file URIs, such as
    +`viking://user/peers/hermes/memories/preferences/mem_abc123.md` or the canonical
    +`viking://user/default/peers/hermes/memories/preferences/mem_abc123.md`. Files
    +directly under `memories/`, such as `viking://user/default/memories/profile.md`,
    +are also allowed because OpenViking supports them. The tool rejects directories,
    +resources, skills, sessions, generated summary files, and URIs with query
    +strings or fragments. Use OpenViking's MCP, CLI, or admin APIs for broader
    +resource and directory cleanup.

  • plugins/memory/openviking/__init__.py

    @@ -91,6 +91,12 @@
         "user": "preferences",
         "memory": "patterns",
     }
    +# OpenViking-generated markdown summaries. Non-.md sidecars such as
    +# .relations.json are rejected earlier by the exact memory-file check.
    +_GENERATED_MEMORY_SUMMARY_FILENAMES = {
    +    ".abstract.md",
    +    ".overview.md",
    +}
     _LOCAL_OPENVIKING_HOSTS = {"localhost", "127.0.0.1", "::1"}
     _LOCAL_OPENVIKING_AUTOSTART_TIMEOUT = 60.0
     _OPENVIKING_SERVER_LOG_RELATIVE_PATH = Path("logs") / "openviking-server.log"
    @@ -320,6 +326,13 @@ def post(self, path: str, payload: dict = None, **kwargs) -> dict:
                 )
             )
     
    +    def delete(self, path: str, **kwargs) -> dict:
    +        return self._send_with_trusted_identity_retry(
    +            lambda headers: self._httpx.delete(
    +                self._url(path), headers=headers, timeout=_TIMEOUT, **kwargs
    +            )
    +        )
    +
         def upload_temp_file(self, file_path: Path) -> str:
             mime_type = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream"
     
    @@ -460,6 +473,26 @@ def validate_root_access(self) -> dict:
         },
     }
     
    +FORGET_SCHEMA = {
    +    "name": "viking_forget",
    +    "description": (
    +        "Delete one OpenViking memory file by exact viking:// URI. "
    +        "Use only when the user explicitly asks to forget or delete a specific "
    +        "memory and you have the exact memory file URI. Resources, skills, "
    +        "sessions, directories, generated summaries, and broad deletes are rejected."
    +    ),
    +    "parameters": {
    +        "type": "object",
    +        "properties": {
    +            "uri": {
    +                "type": "string",
    +                "description": "Exact viking:// memory file URI ending in .md.",
    +            },
    +        },
    +        "required": ["uri"],
    +    },
    +}
    +
     ADD_RESOURCE_SCHEMA = {
         "name": "viking_add_resource",
         "description": (
    @@ -552,6 +585,46 @@ def _is_remote_resource_source(value: str) -> bool:
         return value.startswith(_REMOTE_RESOURCE_PREFIXES)
     
     
    +def _memory_segment_index(parts: List[str]) -> Optional[int]:
    +    if len(parts) >= 2 and parts[0] == "user" and parts[1] == "memories":
    +        return 1
    +    if len(parts) >= 3 and parts[0] == "user" and parts[2] == "memories":
    +        return 2
    +    if len(parts) >= 4 and parts[0] == "user" and parts[1] == "peers" and parts[3] == "memories":
    +        return 3
    +    if len(parts) >= 5 and parts[0] == "user" and parts[2] == "peers" and parts[4] == "memories":
    +        return 4
    +    return None
    +
    +
    +def _validate_forget_memory_uri(raw_uri: Any) -> tuple[Optional[str], Optional[str]]:
    +    if not isinstance(raw_uri, str):
    +        return None, "uri is required"
    +
    +    uri = raw_uri.strip()
    +    if not uri:
    +        return None, "uri is required"
    +
    +    parsed = urlparse(uri)
    +    if parsed.scheme != "viking" or not uri.startswith("viking://"):
    +        return None, "viking_forget only accepts viking:// memory file URIs"
    +    if parsed.query or parsed.fragment:
    +        return None, "viking_forget requires an exact URI without query or fragment"
    +    if uri.endswith("/") or not uri.endswith(".md"):
    +        return None, "viking_forget only deletes concrete .md memory files"
    +
    +    parts = [part for part in uri[len("viking://") :].split("/") if part]
    +    memories_idx = _memory_segment_index(parts)
    +    if memories_idx is None or len(parts) < memories_idx + 2:
    +        return None, "viking_forget only deletes user memory file URIs"
    +
    +    filename = uri.rsplit("/", 1)[-1]
    +    if filename in _GENERATED_MEMORY_SUMMARY_FILENAMES:
    +        return None, "viking_forget cannot delete generated memory summary files"
    +
    +    return uri, None
    +
    +
     def _is_local_path_reference(value: str) -> bool:
         if not value or "\n" in value or "\r" in value:
             return False
    @@ -1719,6 +1792,8 @@ def __init__(self):
             self._prefetch_thread: Optional[threading.Thread] = None
             self._runtime_start_lock = threading.Lock()
             self._runtime_start_thread: Optional[threading.Thread] = None
    +        self._memory_write_lock = threading.Lock()
    +        self._memory_write_threads: Set[threading.Thread] = set()
             # All prefetch threads ever spawned (daemon, short-lived). Tracked so
             # shutdown() can drain them and rapid re-queues don't orphan a still-
             # running thread by overwriting the single _prefetch_thread slot.
    @@ -2047,7 +2122,8 @@ def system_prompt_block(self) -> str:
                     f"Active. Endpoint: {self._endpoint}\n"
                     "Use viking_search to find information, viking_read for details "
                     "(abstract/overview/full), viking_browse to explore.\n"
    -                "Use viking_remember to store facts, viking_add_resource to index URLs/docs."
    +                "Use viking_remember to store facts, viking_forget to delete exact memory "
    +                "file URIs, and viking_add_resource to index URLs/docs."
                 )
             except Exception as e:
                 logger.warning("OpenViking system_prompt_block failed: %s", e)
    @@ -2055,7 +2131,7 @@ def system_prompt_block(self) -> str:
                     "# OpenViking Knowledge Base\n"
                     f"Active. Endpoint: {self._endpoint}\n"
                     "Use viking_search, viking_read, viking_browse, "
    -                "viking_remember, viking_add_resource."
    +                "viking_remember, viking_forget, viking_add_resource."
                 )
     
         def prefetch(self, query: str, *, session_id: str = "") -> str:
    @@ -2806,7 +2882,7 @@ def on_memory_write(
             content: str,
             metadata: Optional[Dict[str, Any]] = None,
         ) -> None:
    -        """Mirror built-in memory writes to OpenViking via content/write."""
    +        """Mirror successful built-in memory additions to OpenViking."""
             if not self._client or action != "add" or not content:
                 return
     
    @@ -2826,12 +2902,30 @@ def _write():
                     })
                 except Exception as e:
                     logger.debug("OpenViking memory mirror failed: %s", e)
    +            finally:
    +                with self._memory_write_lock:
    +                    self._memory_write_threads.discard(threading.current_thread())
     
             t = threading.Thread(target=_write, daemon=True, name="openviking-memwrite")
    -        t.start()
    +        with self._memory_write_lock:
    +            if self._shutting_down:
    +                return
    +            self._memory_write_threads.add(t)
    +            try:
    +                t.start()
    +            except Exception as e:
    +                self._memory_write_threads.discard(t)
    +                logger.debug("OpenViking memory mirror worker failed to start: %s", e)
     
         def get_tool_schemas(self) -> List[Dict[str, Any]]:
    -        return [SEARCH_SCHEMA, READ_SCHEMA, BROWSE_SCHEMA, REMEMBER_SCHEMA, ADD_RESOURCE_SCHEMA]
    +        return [
    +            SEARCH_SCHEMA,
    +            READ_SCHEMA,
    +            BROWSE_SCHEMA,
    +            REMEMBER_SCHEMA,
    +            FORGET_SCHEMA,
    +            ADD_RESOURCE_SCHEMA,
    +        ]
     
         def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
             if not self._client:
    @@ -2846,6 +2940,8 @@ def handle_tool_call(self, tool_name: str, args: dict, **kwargs) -> str:
                     return self._tool_browse(args)
                 elif tool_name == "viking_remember":
                     return self._tool_remember(args)
    +            elif tool_name == "viking_forget":
    +                return self._tool_forget(args)
                 elif tool_name == "viking_add_resource":
                     return self._tool_add_resource(args)
                 return tool_error(f"Unknown tool: {tool_name}")
    @@ -2865,6 +2961,8 @@ def shutdown(self) -> None:
                 deferred_workers = list(self._deferred_commit_threads)
             with self._prefetch_lock:
                 prefetch_workers = list(self._prefetch_threads)
    +        with self._memory_write_lock:
    +            memory_write_workers = list(self._memory_write_threads)
             for t in all_workers:
                 if t.is_alive():
                     t.join(timeout=5.0)
    @@ -2874,6 +2972,9 @@ def shutdown(self) -> None:
             for t in prefetch_workers:
                 if t.is_alive():
                     t.join(timeout=5.0)
    +        for t in memory_write_workers:
    +            if t.is_alive():
    +                t.join(timeout=5.0)
             # Clear atexit reference so it doesn't double-commit.
             global _last_active_provider
             if _last_active_provider is self:
    @@ -3097,6 +3198,31 @@ def _tool_remember(self, args: dict) -> str:
                 logger.error("OpenViking content/write failed: %s", e)
                 return tool_error(f"Failed to store memory: {e}")
     
    +    def _tool_forget(self, args: dict) -> str:
    +        uri, error = _validate_forget_memory_uri(args.get("uri"))
    +        if error:
    +            return tool_error(error)
    +
    +        resp = self._client.delete(
    +            "/api/v1/fs",
    +            params={"uri": uri, "recursive": False},
    +        )
    +        result = self._unwrap_result(resp)
    +        payload: Dict[str, Any] = {"status": "deleted", "uri": uri}
    +        if isinstance(result, dict):
    +            payload["uri"] = result.get("uri") or uri
    +            for key in (
    +                "estimated_deleted_count",
    +                "memory_cleanup",
    +                "semantic_root_uri",
    +                "semantic_status",
    +                "queue_status",
    +            ):
    +                if key in result:
    +                    payload[key] = result[key]
    +
    +        return json.dumps(payload, ensure_ascii=False)
    +
         def _tool_add_resource(self, args: dict) -> str:
             url = args.get("url", "")
             if not url:

  • tests/agent/test_memory_provider.py

    @@ -1172,16 +1172,12 @@ def test_on_memory_write_replace(self):
             mgr.on_memory_write("replace", "user", "updated pref")
             assert p.memory_writes == [("replace", "user", "updated pref")]
     
    -    def test_on_memory_write_remove_not_bridged(self):
    -        """The bridge intentionally skips 'remove' — only add/replace notify."""
    -        # This tests the contract that run_agent.py checks:
    -        #   function_args.get("action") in ("add", "replace")
    +    def test_on_memory_write_remove_supported_by_manager(self):
    +        """The manager forwards remove actions when a caller elects to bridge them."""
             mgr = MemoryManager()
             p = FakeMemoryProvider("ext")
             mgr.add_provider(p)
     
    -        # Manager itself doesn't filter — run_agent.py does.
    -        # But providers should handle remove gracefully.
             mgr.on_memory_write("remove", "memory", "old fact")
             assert p.memory_writes == [("remove", "memory", "old fact")]
     

  • tests/agent/test_memory_write_bridge.py

    @@ -0,0 +1,145 @@
    +"""Behavior tests for the built-in memory → external provider bridge.
    +
    +The bridge lives behind the MemoryManager interface
    +(``MemoryManager.notify_memory_tool_write``): the agent loop hands over the raw
    +built-in memory tool result + args, and the manager decides whether/what to
    +mirror to external providers. These tests drive that method with a fake
    +external provider and assert which ``on_memory_write`` calls land.
    +"""
    +
    +import json
    +
    +import pytest
    +
    +from agent.memory_manager import MemoryManager
    +from agent.memory_provider import MemoryProvider
    +
    +
    +class _RecordingProvider(MemoryProvider):
    +    """Minimal external provider that records on_memory_write calls."""
    +
    +    def __init__(self) -> None:
    +        self.calls = []
    +
    +    @property
    +    def name(self) -> str:
    +        return "recording"
    +
    +    def is_available(self) -> bool:
    +        return True
    +
    +    def initialize(self, session_id: str, **kwargs) -> None:
    +        pass
    +
    +    def get_tool_schemas(self):
    +        return []
    +
    +    def shutdown(self) -> None:
    +        pass
    +
    +    def on_memory_write(self, action, target, content, metadata=None):
    +        self.calls.append({
    +            "action": action,
    +            "target": target,
    +            "content": content,
    +            "metadata": dict(metadata or {}),
    +        })
    +
    +
    +def _manager_with_provider():
    +    mgr = MemoryManager()
    +    provider = _RecordingProvider()
    +    mgr.add_provider(provider)
    +    return mgr, provider
    +
    +
    +def test_notifies_remove_with_old_text_after_success():
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        json.dumps({"success": True}),
    +        {"action": "remove", "target": "memory", "old_text": "stale preference entry"},
    +    )
    +    assert provider.calls == [
    +        {
    +            "action": "remove",
    +            "target": "memory",
    +            "content": "",
    +            "metadata": {"old_text": "stale preference entry"},
    +        }
    +    ]
    +
    +
    +def test_skips_failed_memory_write():
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        json.dumps({"success": False, "error": "No entry matched"}),
    +        {"action": "remove", "target": "memory", "old_text": "stale preference entry"},
    +    )
    +    assert provider.calls == []
    +
    +
    +def test_skips_staged_memory_write():
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        json.dumps({"success": True, "staged": True, "pending_id": "abc123"}),
    +        {"action": "remove", "target": "memory", "old_text": "stale preference entry"},
    +    )
    +    assert provider.calls == []
    +
    +
    [redacted:email]("tool_result", [None, [], object(), "not-json"])
    +def test_skips_unrecognized_tool_result_shape(tool_result):
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        tool_result,
    +        {"action": "add", "target": "memory", "content": "new fact"},
    +    )
    +    assert provider.calls == []
    +
    +
    +def test_preserves_old_text_for_replace_and_remove_batch():
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        json.dumps({"success": True}),
    +        {
    +            "target": "user",
    +            "operations": [
    +                {"action": "replace", "old_text": "old preference", "content": "updated"},
    +                {"action": "remove", "old_text": "obsolete preference"},
    +                {"action": "add", "content": "new fact"},
    +            ],
    +        },
    +    )
    +    assert provider.calls == [
    +        {"action": "replace", "target": "user", "content": "updated",
    +         "metadata": {"old_text": "old preference"}},
    +        {"action": "remove", "target": "user", "content": "",
    +         "metadata": {"old_text": "obsolete preference"}},
    +        {"action": "add", "target": "user", "content": "new fact", "metadata": {}},
    +    ]
    +
    +
    +def test_non_mutating_actions_are_not_mirrored():
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        json.dumps({"success": True}),
    +        {"action": "read", "target": "memory"},
    +    )
    +    assert provider.calls == []
    +
    +
    +def test_build_metadata_callback_is_merged_per_op():
    +    mgr, provider = _manager_with_provider()
    +    mgr.notify_memory_tool_write(
    +        json.dumps({"success": True}),
    +        {"action": "add", "target": "memory", "content": "fact"},
    +        build_metadata=lambda: {"session_id": "s1", "tool_name": "memory"},
    +    )
    +    assert provider.calls == [
    +        {
    +            "action": "add",
    +            "target": "memory",
    +            "content": "fact",
    +            "metadata": {"session_id": "s1", "tool_name": "memory"},
    +        }
    +    ]

  • tests/plugins/memory/test_openviking_provider.py

    @@ -1459,6 +1459,137 @@ def test_tool_add_resource_sends_git_remote_sources_as_path(url):
         })
     
     
    +def test_get_tool_schemas_includes_narrow_forget_tool():
    +    provider = OpenVikingMemoryProvider()
    +
    +    names = [schema["name"] for schema in provider.get_tool_schemas()]
    +
    +    assert "viking_forget" in names
    +
    +
    +def test_handle_tool_call_forget_deletes_exact_memory_file_uri():
    +    uri = "viking://user/peers/hermes/memories/preferences/mem_abc123.md"
    +    provider = OpenVikingMemoryProvider()
    +    provider._client = MagicMock()
    +    provider._client.delete.return_value = {
    +        "status": "ok",
    +        "result": {"uri": uri, "estimated_deleted_count": 1},
    +    }
    +
    +    result = json.loads(provider.handle_tool_call("viking_forget", {"uri": uri}))
    +
    +    provider._client.delete.assert_called_once_with(
    +        "/api/v1/fs",
    +        params={"uri": uri, "recursive": False},
    +    )
    +    assert result == {
    +        "status": "deleted",
    +        "uri": uri,
    +        "estimated_deleted_count": 1,
    +    }
    +
    +
    +def test_handle_tool_call_forget_deletes_exact_memory_file_under_memories_root():
    +    uri = "viking://user/default/memories/profile.md"
    +    provider = OpenVikingMemoryProvider()
    +    provider._client = MagicMock()
    +    provider._client.delete.return_value = {
    +        "status": "ok",
    +        "result": {"uri": uri, "estimated_deleted_count": 1},
    +    }
    +
    +    result = json.loads(provider.handle_tool_call("viking_forget", {"uri": uri}))
    +
    +    provider._client.delete.assert_called_once_with(
    +        "/api/v1/fs",
    +        params={"uri": uri, "recursive": False},
    +    )
    +    assert result == {
    +        "status": "deleted",
    +        "uri": uri,
    +        "estimated_deleted_count": 1,
    +    }
    +
    +
    +def test_handle_tool_call_forget_allows_non_generated_dot_md_memory_file():
    +    uri = "viking://user/default/memories/preferences/.full.md"
    +    provider = OpenVikingMemoryProvider()
    +    provider._client = MagicMock()
    +    provider._client.delete.return_value = {
    +        "status": "ok",
    +        "result": {"uri": uri, "estimated_deleted_count": 1},
    +    }
    +
    +    result = json.loads(provider.handle_tool_call("viking_forget", {"uri": uri}))
    +
    +    provider._client.delete.assert_called_once_with(
    +        "/api/v1/fs",
    +        params={"uri": uri, "recursive": False},
    +    )
    +    assert result == {
    +        "status": "deleted",
    +        "uri": uri,
    +        "estimated_deleted_count": 1,
    +    }
    +
    +
    [redacted:email]("uri", [
    +    "",
    +    "https://example.com/mem.md",
    +    "viking:/user/memories/preferences/mem_abc123.md",
    +    "viking://resources/project/doc.md",
    +    "viking://resources/project/memories/mem_abc123.md",
    +    "viking://memories/preferences/mem_abc123.md",
    +    "viking://agent/hermes/memories/preferences/mem_abc123.md",
    +    "viking://user/skills/example/SKILL.md",
    +    "viking://user/sessions/session-1/messages.jsonl",
    +    "viking://user/memories/preferences/",
    +    "viking://user/memories/preferences/.overview.md",
    +    "viking://user/memories/preferences/.abstract.md",
    +    "viking://user/memories/preferences/mem_abc123.md?recursive=true",
    +])
    +def test_handle_tool_call_forget_rejects_non_memory_file_uris(uri):
    +    provider = OpenVikingMemoryProvider()
    +    provider._client = MagicMock()
    +
    +    result = json.loads(provider.handle_tool_call("viking_forget", {"uri": uri}))
    +
    +    assert "error" in result
    +    provider._client.delete.assert_not_called()
    +
    +
    +def test_viking_client_delete_uses_identity_headers(monkeypatch):
    +    client = _VikingClient(
    +        "https://example.com",
    +        api_key="[redacted:secret-assignment]",
    +        account="acct",
    +        user="alice",
    +        agent="hermes",
    +    )
    +    captured = {}
    +
    +    def capture_delete(url, **kwargs):
    +        captured["url"] = url
    +        captured["kwargs"] = kwargs
    +        return SimpleNamespace(
    +            status_code=200,
    +            text="",
    +            json=lambda: {"status": "ok", "result": {"uri": "viking://user/memories/x.md"}},
    +            raise_for_status=lambda: None,
    +        )
    +
    +    monkeypatch.setattr(client._httpx, "delete", capture_delete)
    +
    +    assert client.delete("/api/v1/fs", params={"uri": "viking://user/memories/x.md"}) == {
    +        "status": "ok",
    +        "result": {"uri": "viking://user/memories/x.md"},
    +    }
    +    assert captured["url"] == "https://example.com/api/v1/fs"
    +    assert captured["kwargs"]["params"] == {"uri": "viking://user/memories/x.md"}
    +    assert captured["kwargs"]["headers"]["Authorization"] == "Bearer test-key"
    +    assert captured["kwargs"]["headers"]["X-OpenViking-Actor-Peer"] == "hermes"
    +
    +
     def test_viking_client_upload_temp_file_uses_multipart_identity_headers(tmp_path, monkeypatch):
         sample = tmp_path / "sample.md"
         sample.write_text("# Local resource\n", encoding="utf-8")
    @@ -2637,6 +2768,94 @@ def post(self, path, payload=None, **kwargs):
         )
     
     
    +def test_shutdown_waits_for_memory_write_worker(monkeypatch):
    +    import threading
    +
    +    provider = OpenVikingMemoryProvider()
    +    provider._client = MagicMock()
    +    provider._endpoint = "http://test"
    +    provider._api_key = ""
    +    provider._account = "acct"
    +    provider._user = "usr"
    +    provider._agent = "hermes"
    +
    +    worker_started = threading.Event()
    +    release_worker = threading.Event()
    +    worker_finished = threading.Event()
    +    shutdown_returned = threading.Event()
    +
    +    class StubClient:
    +        def __init__(self, *a, **kw):
    +            pass
    +
    +        def post(self, path, payload=None, **kwargs):
    +            assert path == "/api/v1/content/write"
    +            worker_started.set()
    +            release_worker.wait(timeout=2.0)
    +            worker_finished.set()
    +            return {}
    +
    +    monkeypatch.setattr(openviking_module, "_VikingClient", StubClient)
    +
    +    provider.on_memory_write("add", "user", "remember this")
    +    assert worker_started.wait(timeout=2.0), "worker never entered post()"
    +
    +    shutdown_thread = threading.Thread(
    +        target=lambda: (provider.shutdown(), shutdown_returned.set()),
    +        daemon=True,
    +    )
    +    shutdown_thread.start()
    +
    +    returned_before_worker_finished = shutdown_returned.wait(timeout=0.1)
    +    release_worker.set()
    +    assert shutdown_returned.wait(timeout=2.0), "shutdown did not return after worker finished"
    +    shutdown_thread.join(timeout=2.0)
    +
    +    assert not returned_before_worker_finished
    +    assert worker_finished.is_set()
    +    assert provider._memory_write_threads == set()
    +
    +
    [redacted:email](
    +    ("action", "content"),
    +    [
    +        ("replace", "updated memory"),
    +        ("remove", ""),
    +        ("forget", ""),
    +        ("delete", ""),
    +    ],
    +)
    +def test_on_memory_write_ignores_non_add_actions(action, content, monkeypatch):
    +    provider = OpenVikingMemoryProvider()
    +    provider._client = MagicMock()
    +    provider._endpoint = "http://test"
    +    provider._api_key = ""
    +    provider._account = "acct"
    +    provider._user = "usr"
    +    provider._agent = "hermes"
    +    uri = "viking://user/peers/hermes/memories/preferences/mem_abc123.md"
    +    spawned = []
    +
    +    class StubThread:
    +        def __init__(self, *args, **kwargs):
    +            spawned.append((args, kwargs))
    +
    +        def start(self):
    +            raise AssertionError("non-URI remove should not spawn a mirror thread")
    +
    +    import plugins.memory.openviking as _mod
    +    monkeypatch.setattr(_mod.threading, "Thread", StubThread)
    +
    +    provider.on_memory_write(
    +        action,
    +        "memory",
    +        content,
    +        metadata={"uri": uri, "old_text": "stale fact"},
    +    )
    +
    +    assert spawned == []
    +
    +
     # ---------------------------------------------------------------------------
     # Prefetch staleness: a prefetch worker that finishes AFTER a session switch
     # must drop its result instead of repopulating the new session with stale

  • tests/run_agent/test_run_agent.py

    @@ -23,6 +23,7 @@
     import run_agent
     from run_agent import AIAgent
     from agent.error_classifier import FailoverReason
    +from agent.memory_manager import MemoryManager
     from agent.prompt_builder import DEFAULT_AGENT_IDENTITY
     
     
    @@ -2082,6 +2083,41 @@ def test_single_tool_executed(self, agent):
             assert messages[0]["role"] == "tool"
             assert "search result" in messages[0]["content"]
     
    +    def test_sequential_memory_remove_notifies_provider_with_tool_result(self, agent):
    +        old_text = "stale preference entry"
    +        tc = _mock_tool_call(
    +            name="memory",
    +            arguments=json.dumps({
    +                "action": "remove",
    +                "target": "memory",
    +                "old_text": old_text,
    +            }),
    +            call_id="mem-1",
    +        )
    +        mock_msg = _mock_assistant_msg(content="", tool_calls=[tc])
    +        messages = []
    +        calls = []
    +
    +        class FakeMemoryManager(MemoryManager):
    +            def has_tool(self, tool_name):
    +                return False
    +
    +            def on_memory_write(self, action, target, content, metadata=None):
    +                calls.append((action, target, content, metadata or {}))
    +
    +        agent._memory_manager = FakeMemoryManager()
    +        agent._memory_store = object()
    +
    +        with patch("tools.memory_tool.memory_tool", return_value=json.dumps({"success": True})):
    +            agent._execute_tool_calls_sequential(mock_msg, messages, "task-1")
    +
    +        assert len(calls) == 1
    +        action, target, content, metadata = calls[0]
    +        assert (action, target, content) == ("remove", "memory", "")
    +        assert metadata["old_text"] == old_text
    +        assert metadata["tool_call_id"] == "mem-1"
    +        assert messages[-1]["tool_call_id"] == "mem-1"
    +
         def test_keyboard_interrupt_emits_cancelled_post_tool_hook(self, agent, monkeypatch):
             tc = _mock_tool_call(name="web_search", arguments='{"q":"test"}', call_id="c1")
             mock_msg = _mock_assistant_msg(content="", tool_calls=[tc])
    @@ -2797,6 +2833,68 @@ def test_blocked_memory_tool_does_not_reset_counter(self, agent, monkeypatch):
             assert json.loads(result) == {"error": "Blocked"}
             assert agent._turns_since_memory == 5
     
    +    def test_invoke_tool_memory_remove_notifies_provider_with_old_text(self, agent, monkeypatch):
    +        monkeypatch.setattr(
    +            "hermes_cli.plugins.get_pre_tool_call_block_message",
    +            lambda *args, **kwargs: None,
    +        )
    +        calls = []
    +
    +        class FakeMemoryManager(MemoryManager):
    +            def has_tool(self, tool_name):
    +                return False
    +
    +            def on_memory_write(self, action, target, content, metadata=None):
    +                calls.append((action, target, content, metadata or {}))
    +
    +        old_text = "stale preference entry"
    +        agent._memory_manager = FakeMemoryManager()
    +        agent._memory_store = object()
    +
    +        with patch("tools.memory_tool.memory_tool", return_value=json.dumps({"success": True})):
    +            agent._invoke_tool(
    +                "memory",
    +                {"action": "remove", "target": "memory", "old_text": old_text},
    +                "task-1",
    +                tool_call_id="mem-1",
    +            )
    +
    +        assert len(calls) == 1
    +        action, target, content, metadata = calls[0]
    +        assert (action, target, content) == ("remove", "memory", "")
    +        assert metadata["old_text"] == old_text
    +        assert metadata["tool_call_id"] == "mem-1"
    +
    +    def test_invoke_tool_memory_failed_remove_skips_provider_notification(self, agent, monkeypatch):
    +        monkeypatch.setattr(
    +            "hermes_cli.plugins.get_pre_tool_call_block_message",
    +            lambda *args, **kwargs: None,
    +        )
    +        notify = MagicMock(side_effect=AssertionError("should not notify"))
    +
    +        class FakeMemoryManager(MemoryManager):
    +            def has_tool(self, tool_name):
    +                return False
    +
    +            on_memory_write = notify
    +
    +        manager = FakeMemoryManager()
    +        agent._memory_manager = manager
    +        agent._memory_store = object()
    +
    +        with patch(
    +            "tools.memory_tool.memory_tool",
    +            return_value=json.dumps({"success": False, "error": "No entry matched"}),
    +        ):
    +            agent._invoke_tool(
    +                "memory",
    +                {"action": "remove", "target": "memory", "old_text": "missing"},
    +                "task-1",
    +                tool_call_id="mem-1",
    +            )
    +
    +        notify.assert_not_called()
    +
         def test_concurrent_blocked_write_skips_checkpoint(self, agent, monkeypatch):
             """Concurrent path: blocked write_file should not trigger checkpoint."""
             tc1 = _mock_tool_call(name="write_file",