OpenViking memory plugin session creation and handling

Fixes on record

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

• #10463 fix(openviking): correct account default + commit session on /new and compress

  merged to main

  files: cli.py, plugins/memory/openviking/__init__.py, run_agent.py, scripts/release.py, tests/agent/test_memory_provider.py

  Show the diff

  • cli.py

    @@ -4100,6 +4100,8 @@ def new_session(self, silent=False):
                     self.agent.flush_memories(self.conversation_history)
                 except (Exception, KeyboardInterrupt):
                     pass
    +            # Trigger memory extraction on the old session before session_id rotates.
    +            self.agent.commit_memory_session(self.conversation_history)
                 self._notify_session_boundary("on_session_finalize")
             elif self.agent:
                 # First session or empty history — still finalize the old session

  • plugins/memory/openviking/__init__.py

    @@ -10,8 +10,9 @@
     Config via environment variables (profile-scoped via each profile's .env):
       OPENVIKING_ENDPOINT  — Server URL (default: http://127.0.0.1:1933)
       OPENVIKING_API_KEY   — API key (required for authenticated servers)
    -  OPENVIKING_ACCOUNT   — Tenant account (default: root)
    +  OPENVIKING_ACCOUNT   — Tenant account (default: default)
       OPENVIKING_USER      — Tenant user (default: default)
    +  OPENVIKING_AGENT   — Tenant agent (default: hermes)
     
     Capabilities:
       - Automatic memory extraction on session commit (6 categories)
    @@ -80,11 +81,12 @@ class _VikingClient:
         """Thin HTTP client for the OpenViking REST API."""
     
         def __init__(self, endpoint: str, api_key: str = "",
    -                 account: str = "", user: str = ""):
    +                 account: str = "", user: str = "", agent: str = ""):
             self._endpoint = endpoint.rstrip("/")
             self._api_key = [redacted:secret-assignment]
    -        self._account = account or os.environ.get("OPENVIKING_ACCOUNT", "root")
    +        self._account = account or os.environ.get("OPENVIKING_ACCOUNT", "default")
             self._user = user or os.environ.get("OPENVIKING_USER", "default")
    +        self._agent = agent or os.environ.get("OPENVIKING_AGENT", "hermes")
             self._httpx = _get_httpx()
             if self._httpx is None:
                 raise ImportError("httpx is required for OpenViking: pip install httpx")
    @@ -94,6 +96,7 @@ def _headers(self) -> dict:
                 "Content-Type": "application/json",
                 "X-OpenViking-Account": self._account,
                 "X-OpenViking-User": self._user,
    +            "X-OpenViking-Agent": self._agent,
             }
             if self._api_key:
                 h["X-API-Key"] = self._api_key
    @@ -282,20 +285,44 @@ def get_config_schema(self):
                 },
                 {
                     "key": "api_key",
    -                "description": "OpenViking API key",
    +                "description": "OpenViking API key (leave blank for local dev mode)",
                     "secret": True,
                     "env_var": "OPENVIKING_API_KEY",
                 },
    +            {
    +                "key": "account",
    +                "description": "OpenViking tenant account ID ([default], used when local mode, OPENVIKING_API_KEY is empty)",
    +                "default": "default",
    +                "env_var": "OPENVIKING_ACCOUNT",
    +            },
    +            {
    +                "key": "user",
    +                "description": "OpenViking user ID within the account ([default], used when local mode, OPENVIKING_API_KEY is empty)",
    +                "default": "default",
    +                "env_var": "OPENVIKING_USER",
    +            },
    +            {
    +                "key": "agent",
    +                "description": "OpenViking agent ID within the account ([hermes], useful in multi-agent mode)",
    +                "default": "hermes",
    +                "env_var": "OPENVIKING_AGENT",
    +            },
             ]
     
         def initialize(self, session_id: str, **kwargs) -> None:
             self._endpoint = os.environ.get("OPENVIKING_ENDPOINT", _DEFAULT_ENDPOINT)
             self._api_key = os.environ.get("OPENVIKING_API_KEY", "")
    +        self._account = os.environ.get("OPENVIKING_ACCOUNT", "default")
    +        self._user = os.environ.get("OPENVIKING_USER", "default")
    +        self._agent = os.environ.get("OPENVIKING_AGENT", "hermes")
             self._session_id = session_id
             self._turn_count = 0
     
             try:
    -            self._client = _VikingClient(self._endpoint, self._api_key)
    +            self._client = _VikingClient(
    +                self._endpoint, self._api_key,
    +                account=self._account, user=self._user, agent=self._agent,
    +            )
                 if not self._client.health():
                     logger.warning("OpenViking server at %s is not reachable", self._endpoint)
                     self._client = None
    @@ -325,7 +352,8 @@ def system_prompt_block(self) -> str:
                     "(abstract/overview/full), viking_browse to explore.\n"
                     "Use viking_remember to store facts, viking_add_resource to index URLs/docs."
                 )
    -        except Exception:
    +        except Exception as e:
    +            logger.warning("OpenViking system_prompt_block failed: %s", e)
                 return (
                     "# OpenViking Knowledge Base\n"
                     f"Active. Endpoint: {self._endpoint}\n"
    @@ -351,7 +379,10 @@ def queue_prefetch(self, query: str, *, session_id: str = "") -> None:
     
             def _run():
                 try:
    -                client = _VikingClient(self._endpoint, self._api_key)
    +                client = _VikingClient(
    +                    self._endpoint, self._api_key,
    +                    account=self._account, user=self._user, agent=self._agent,
    +                )
                     resp = client.post("/api/v1/search/find", {
                         "query": query,
                         "top_k": 5,
    @@ -386,7 +417,10 @@ def sync_turn(self, user_content: str, assistant_content: str, *, session_id: st
     
             def _sync():
                 try:
    -                client = _VikingClient(self._endpoint, self._api_key)
    +                client = _VikingClient(
    +                    self._endpoint, self._api_key,
    +                    account=self._account, user=self._user, agent=self._agent,
    +                )
                     sid = self._session_id
     
                     # Add user message
    @@ -442,7 +476,10 @@ def on_memory_write(self, action: str, target: str, content: str) -> None:
     
             def _write():
                 try:
    -                client = _VikingClient(self._endpoint, self._api_key)
    +                client = _VikingClient(
    +                    self._endpoint, self._api_key,
    +                    account=self._account, user=self._user, agent=self._agent,
    +                )
                     # Add as a user message with memory context so the commit
                     # picks it up as an explicit memory during extraction
                     client.post(f"/api/v1/sessions/{self._session_id}/messages", {

  • run_agent.py

    @@ -3040,6 +3040,18 @@ def shutdown_memory_provider(self, messages: list = None) -> None:
                 except Exception:
                     pass
         
    +    def commit_memory_session(self, messages: list = None) -> None:
    +        """Trigger end-of-session extraction without tearing providers down.
    +        Called when session_id rotates (e.g. /new, context compression);
    +        providers keep their state and continue running under the old
    +        session_id — they just flush pending extraction now."""
    +        if not self._memory_manager:
    +            return
    +        try:
    +            self._memory_manager.on_session_end(messages or [])
    +        except Exception:
    +            pass
    +
         def close(self) -> None:
             """Release all resources held by this agent instance.
     
    @@ -6826,6 +6838,8 @@ def _compress_context(self, messages: list, system_message: str, *, approx_token
                 try:
                     # Propagate title to the new session with auto-numbering
                     old_title = self._session_db.get_session_title(self.session_id)
    +                # Trigger memory extraction on the old session before it rotates.
    +                self.commit_memory_session(messages)
                     self._session_db.end_session(self.session_id, "compression")
                     old_session_id = self.session_id
                     self.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:6]}"

  • scripts/release.py

    @@ -197,6 +197,8 @@
         "[redacted:email]": "zhouboli",
         "[redacted:email]": "tomqiaozc",
         "[redacted:email]": "xinbenlv",
    +    "[redacted:email]": "ZaynJarvis",
    +    "[redacted:email]": "ZaynJarvis",
     }
     
     

  • tests/agent/test_memory_provider.py

    @@ -695,3 +695,44 @@ def test_fenced_block_separates_user_from_recall(self):
             fence_end = combined.index("</memory-context>")
             assert "Alice" in combined[fence_start:fence_end]
             assert combined.index("weather") < fence_start
    +
    +
    +# ---------------------------------------------------------------------------
    +# AIAgent.commit_memory_session — routes to MemoryManager.on_session_end
    +# ---------------------------------------------------------------------------
    +
    +
    +class _CommitRecorder(FakeMemoryProvider):
    +    """Provider that records on_session_end calls for assertions."""
    +
    +    def __init__(self, name="recorder"):
    +        super().__init__(name)
    +        self.end_calls = []
    +
    +    def on_session_end(self, messages):
    +        self.end_calls.append(list(messages or []))
    +
    +
    +class TestCommitMemorySessionRouting:
    +    def test_on_session_end_fans_out(self):
    +        mgr = MemoryManager()
    +        builtin = _CommitRecorder("builtin")
    +        external = _CommitRecorder("openviking")
    +        mgr.add_provider(builtin)
    +        mgr.add_provider(external)
    +
    +        msgs = [{"role": "user", "content": "hi"}]
    +        mgr.on_session_end(msgs)
    +
    +        assert builtin.end_calls == [msgs]
    +        assert external.end_calls == [msgs]
    +
    +    def test_on_session_end_tolerates_failure(self):
    +        mgr = MemoryManager()
    +        builtin = FakeMemoryProvider("builtin")
    +        bad = _CommitRecorder("bad-provider")
    +        bad.on_session_end = lambda m: (_ for _ in ()).throw(RuntimeError("boom"))
    +        mgr.add_provider(builtin)
    +        mgr.add_provider(bad)
    +
    +        mgr.on_session_end([])  # must not raise