diff --git a/agent/auxiliary_client.py b/agent/auxiliary_client.py index 4f8c9a0a464af..e24c384d16d61 100644 --- a/agent/auxiliary_client.py +++ b/agent/auxiliary_client.py @@ -1668,7 +1668,7 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = ""): # ── Custom endpoint (OPENAI_BASE_URL + OPENAI_API_KEY) ─────────── if provider == "custom": if explicit_base_url: - custom_base = explicit_base_url.strip() + custom_base = _to_openai_base_url(explicit_base_url).strip() custom_key = ( (explicit_api_key or "").strip() or os.getenv("OPENAI_API_KEY", "").strip() @@ -1681,7 +1681,7 @@ def _wrap_if_needed(client_obj, final_model_str: str, base_url_str: str = ""): ) return None, None final_model = _normalize_resolved_model( - model or _read_main_model() or "gpt-4o-mini", + model or main_runtime.get("model") or "gpt-4o-mini", provider, ) extra = {} diff --git a/agent/memory_manager.py b/agent/memory_manager.py index 2435c3f248399..e3a537523865f 100644 --- a/agent/memory_manager.py +++ b/agent/memory_manager.py @@ -4,7 +4,10 @@ Single integration point in run_agent.py. Replaces scattered per-backend code with one manager that delegates to registered providers. -The BuiltinMemoryProvider is always registered first and cannot be removed. +The built-in memory is managed directly by MemoryStore (in tools/memory_tool.py) +and does NOT go through the MemoryProvider plugin system. The MemoryManager +coordinates external providers (Holographic, Honcho, etc.) only — it is NOT +responsible for built-in memory reads/writes. Only ONE external (non-builtin) provider is allowed at a time — attempting to register a second external provider is rejected with a warning. This prevents tool schema bloat and conflicting memory backends. diff --git a/agent/title_generator.py b/agent/title_generator.py index 99c771cb5095e..d3f9af0231dfa 100644 --- a/agent/title_generator.py +++ b/agent/title_generator.py @@ -19,10 +19,16 @@ ) -def generate_title(user_message: str, assistant_response: str, timeout: float = 30.0) -> Optional[str]: +def generate_title( + user_message: str, + assistant_response: str, + timeout: float = 30.0, + main_runtime: dict = None, +) -> Optional[str]: """Generate a session title from the first exchange. - Uses the auxiliary LLM client (cheapest/fastest available model). + Uses the main runtime's model when available, falling back to the + auxiliary LLM client (cheapest/fastest available model). Returns the title string or None on failure. """ # Truncate long messages to keep the request small @@ -41,6 +47,7 @@ def generate_title(user_message: str, assistant_response: str, timeout: float = max_tokens=500, temperature=0.3, timeout=timeout, + main_runtime=main_runtime, ) title = (response.choices[0].message.content or "").strip() # Clean up: remove quotes, trailing punctuation, prefixes like "Title: " @@ -61,6 +68,7 @@ def auto_title_session( session_id: str, user_message: str, assistant_response: str, + main_runtime: dict = None, ) -> None: """Generate and set a session title if one doesn't already exist. @@ -81,7 +89,7 @@ def auto_title_session( except Exception: return - title = generate_title(user_message, assistant_response) + title = generate_title(user_message, assistant_response, main_runtime=main_runtime) if not title: return @@ -98,6 +106,7 @@ def maybe_auto_title( user_message: str, assistant_response: str, conversation_history: list, + main_runtime: dict = None, ) -> None: """Fire-and-forget title generation after the first exchange. @@ -118,7 +127,7 @@ def maybe_auto_title( thread = threading.Thread( target=auto_title_session, - args=(session_db, session_id, user_message, assistant_response), + args=(session_db, session_id, user_message, assistant_response, main_runtime), daemon=True, name="auto-title", ) diff --git a/cli.py b/cli.py index a289e3ab2374c..f67ceef23fb4e 100644 --- a/cli.py +++ b/cli.py @@ -8641,6 +8641,13 @@ def run_agent(): message, response, self.conversation_history, + main_runtime={ + "model": self.model, + "provider": self.provider, + "base_url": self.base_url, + "api_key": self.api_key, + "api_mode": self.api_mode, + }, ) except Exception: pass diff --git a/gateway/run.py b/gateway/run.py index a024649cbdd10..47ac962b0a8bc 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -660,6 +660,7 @@ def __init__(self, config: Optional[GatewayConfig] = None): self._pending_messages: Dict[str, str] = {} # Queued messages during interrupt self._busy_ack_ts: Dict[str, float] = {} # last busy-ack timestamp per session (debounce) self._session_run_generation: Dict[str, int] = {} + self._restart_caller_key: str = None # session_key of the agent that triggered /restart # Cache AIAgent instances per session to preserve prompt caching. # Without this, a new AIAgent is created per message, rebuilding the @@ -1626,7 +1627,7 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session return True - async def _drain_active_agents(self, timeout: float) -> tuple[Dict[str, Any], bool]: + async def _drain_active_agents(self, timeout: float, exclude_key: str = None) -> tuple[Dict[str, Any], bool]: snapshot = self._snapshot_running_agents() last_active_count = self._running_agent_count() last_status_at = 0.0 @@ -1634,13 +1635,22 @@ async def _drain_active_agents(self, timeout: float) -> tuple[Dict[str, Any], bo def _maybe_update_status(force: bool = False) -> None: nonlocal last_active_count, last_status_at now = asyncio.get_running_loop().time() - active_count = self._running_agent_count() + # Count agents, excluding the restart caller so it can't block drain. + active_count = sum( + 1 for k in self._running_agents + if k != exclude_key and self._running_agents.get(k) is not _AGENT_PENDING_SENTINEL + ) if force or active_count != last_active_count or (now - last_status_at) >= 1.0: self._update_runtime_status("draining") last_active_count = active_count last_status_at = now - if not self._running_agents: + # Build a filtered view that excludes the restart caller. + filtered_agents = { + k: v for k, v in self._running_agents.items() if k != exclude_key + } if exclude_key else dict(self._running_agents) + + if not filtered_agents: _maybe_update_status(force=True) return snapshot, False @@ -1649,15 +1659,21 @@ def _maybe_update_status(force: bool = False) -> None: return snapshot, True deadline = asyncio.get_running_loop().time() + timeout - while self._running_agents and asyncio.get_running_loop().time() < deadline: + while filtered_agents and asyncio.get_running_loop().time() < deadline: + # Re-filter each iteration in case agents drop out. + filtered_agents = { + k: v for k, v in self._running_agents.items() if k != exclude_key + } _maybe_update_status() await asyncio.sleep(0.1) - timed_out = bool(self._running_agents) + timed_out = bool(filtered_agents) _maybe_update_status(force=True) return snapshot, timed_out - def _interrupt_running_agents(self, reason: str) -> None: + def _interrupt_running_agents(self, reason: str, exclude_key: str = None) -> None: for session_key, agent in list(self._running_agents.items()): + if session_key == exclude_key: + continue if agent is _AGENT_PENDING_SENTINEL: continue try: @@ -2572,7 +2588,9 @@ async def _stop_impl() -> None: await self._notify_active_sessions_of_shutdown() timeout = self._restart_drain_timeout - active_agents, timed_out = await self._drain_active_agents(timeout) + active_agents, timed_out = await self._drain_active_agents( + timeout, exclude_key=self._restart_caller_key + ) if timed_out: logger.warning( "Gateway drain timed out after %.1fs with %d active agent(s); interrupting remaining work.", @@ -2614,7 +2632,8 @@ async def _stop_impl() -> None: _sk[:20], _e, ) self._interrupt_running_agents( - _INTERRUPT_REASON_GATEWAY_RESTART if self._restart_requested else _INTERRUPT_REASON_GATEWAY_SHUTDOWN + _INTERRUPT_REASON_GATEWAY_RESTART if self._restart_requested else _INTERRUPT_REASON_GATEWAY_SHUTDOWN, + exclude_key=self._restart_caller_key, ) interrupt_deadline = asyncio.get_running_loop().time() + 5.0 while self._running_agents and asyncio.get_running_loop().time() < interrupt_deadline: @@ -5280,6 +5299,10 @@ async def _handle_restart_command(self, event: MessageEvent) -> str: # doesn't work under systemd because KillMode=mixed kills all # processes in the cgroup, including the detached helper. _under_service = bool(os.environ.get("INVOCATION_ID")) # systemd sets this + # Record the caller's session_key so drain can exclude it — the restart + # triggerer's agent cannot exit until it receives the HTTP response, so + # excluding it prevents the drain-from-itself deadlock. + self._restart_caller_key = self._session_key_for_source(event.source) if _under_service: self.request_restart(detached=False, via_service=True) else: @@ -10197,12 +10220,21 @@ def _approval_notify_sync(approval_data: dict) -> None: try: from agent.title_generator import maybe_auto_title all_msgs = result_holder[0].get("messages", []) if result_holder[0] else [] + # Build main_runtime from the agent that handled this run + _title_agent = agent_holder[0] maybe_auto_title( self._session_db, effective_session_id, message, final_response, all_msgs, + main_runtime={ + "model": getattr(_title_agent, "model", None), + "provider": getattr(_title_agent, "provider", None), + "base_url": getattr(_title_agent, "base_url", None), + "api_key": getattr(_title_agent, "api_key", None), + "api_mode": getattr(_title_agent, "api_mode", None), + } if _title_agent else None, ) except Exception: pass diff --git a/hermes_state.py b/hermes_state.py index 0ea9815b5a157..132bcc9c95eb2 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -1135,9 +1135,9 @@ def _preserve_quoted(m: re.Match) -> str: # quotes. FTS5's tokenizer splits on dots and hyphens, turning # ``chat-send`` into ``chat AND send`` and ``P2.2`` into ``p2 AND 2``. # Quoting preserves phrase semantics. A single pass avoids the - # double-quoting bug that would occur if dotted and hyphenated + # double-quoting bug that would occur if dotted, hyphenated and underscored # patterns were applied sequentially (e.g. ``my-app.config``). - sanitized = re.sub(r"\b(\w+(?:[.-]\w+)+)\b", r'"\1"', sanitized) + sanitized = re.sub(r"\b(\w+(?:[._-]\w+)+)\b", r'"\1"', sanitized) # Step 6: Restore preserved quoted phrases for i, quoted in enumerate(_quoted_parts): diff --git a/plugins/memory/holographic/__init__.py b/plugins/memory/holographic/__init__.py index cd4ef07b44c70..1ff8fc20c96bd 100644 --- a/plugins/memory/holographic/__init__.py +++ b/plugins/memory/holographic/__init__.py @@ -206,11 +206,87 @@ def prefetch(self, query: str, *, session_id: str = "") -> str: if not self._retriever or not query: return "" try: - results = self._retriever.search(query, min_trust=self._min_trust, limit=5) - if not results: + # Parallel dual-path retrieval: + # Path A — FTS5 keyword + Jaccard + HRR rerank (shallow, fast) + # Path B — HRR reason() algebraic bind/unbind (deep, semantic) + # Both paths are independent; results are merged and deduped by fact_id. + import concurrent.futures + + fts_results: list[dict] = [] + hrr_results: list[dict] = [] + + def run_fts(): + return self._retriever.search( + query, min_trust=0.0, limit=8 + ) + + def run_hrr(): + # reason() does not accept min_trust; trust scoring is applied + # during score fusion in the merge step below. + tokens = [t.strip(".,!?;:\"'()[]{}-") for t in query.lower().split()] + tokens = [t for t in tokens if t] + return self._retriever.reason( + tokens if tokens else [query.lower().strip()], + limit=8, + ) + + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor: + fts_future = executor.submit(run_fts) + hrr_future = executor.submit(run_hrr) + try: + fts_results = fts_future.result(timeout=2.0) + except Exception: + fts_results = [] + try: + hrr_results = hrr_future.result(timeout=2.0) + except Exception: + hrr_results = [] + + # Score quality gate: + # HRR produces near-random ~0.25 with dim=1024, n=21 when no entity + # signal is present. Use HRR only when: + # (a) FTS has results AND HRR clearly beats them (ratio-gated), OR + # (b) FTS returns nothing — HRR is the only signal, accept if above + # a higher floor since we have no calibration baseline. + fts_best = fts_results[0].get("score", 0) if fts_results else 0.0 + _FTS_FAIL_FLOOR = 0.10 # below this FTS is unreliable + _HRR_NOISE_FLOOR = 0.27 # empirical: ~random for this corpus size/dim + + # Merge: FTS primary, HRR supplement + seen: dict[int, dict] = {} + for r in fts_results: + fid = r.get("fact_id") + if fid is not None and fid not in seen: + seen[fid] = r + + # Adopt HRR results when FTS is weak or absent + for r in hrr_results: + fid = r.get("fact_id") + if fid is None or fid in seen: + continue + hrr_score = r.get("score", 0) + # FTS strong enough to judge HRR on ratio? + if fts_best >= _FTS_FAIL_FLOOR: + # Yes: require HRR beat FTS by meaningful margin + if hrr_score > fts_best * 1.05 and hrr_score > _HRR_NOISE_FLOOR: + seen[fid] = r + else: + # No FTS signal: accept HRR only if above higher floor + # (no calibration baseline, must be clearly non-noise) + if hrr_score > 0.29: + seen[fid] = r + + # Sort by score descending, take top 5 + merged = sorted(seen.values(), key=lambda x: x.get("score", 0), reverse=True)[:5] + + # Apply trust threshold after score fusion + merged = [r for r in merged if r.get("trust_score", 0) >= self._min_trust] + + if not merged: return "" + lines = [] - for r in results: + for r in merged: trust = r.get("trust_score", r.get("trust", 0)) lines.append(f"- [{trust:.1f}] {r.get('content', '')}") return "## Holographic Memory\n" + "\n".join(lines) @@ -242,12 +318,19 @@ def on_session_end(self, messages: List[Dict[str, Any]]) -> None: def on_memory_write(self, action: str, target: str, content: str) -> None: """Mirror built-in memory writes as facts.""" - if action == "add" and self._store and content: + if not self._store or not content: + return + if action == "add": try: category = "user_pref" if target == "user" else "general" self._store.add_fact(content, category=category) except Exception as e: logger.debug("Holographic memory_write mirror failed: %s", e) + elif action == "remove": + try: + self._store.remove_fact_by_content(content) + except Exception as e: + logger.debug("Holographic memory_write remove mirror failed: %s", e) def shutdown(self) -> None: self._store = None diff --git a/plugins/memory/holographic/retrieval.py b/plugins/memory/holographic/retrieval.py index a673dcef846ce..705aba6a7851b 100644 --- a/plugins/memory/holographic/retrieval.py +++ b/plugins/memory/holographic/retrieval.py @@ -494,9 +494,32 @@ def _fts_candidates( # Build query - FTS5 rank is negative (lower = better match) # We need to join facts_fts with facts to get all columns + # For multi-token queries, use OR to match any token (FTS5 tokenization + # splits Chinese on character boundaries, so each character or word becomes + # a separate token). Single-token queries are passed as-is. + # For tokens containing ASCII characters (English/case-sensitive), use + # prefix matching so "vegf" matches "VEGF通路" in the index. + import re + tokens = [t.strip(".,!?;:\"'()[]{}-") for t in query.lower().split()] + tokens = [t for t in tokens if t] + if len(tokens) > 1: + fts_terms = [] + for token in tokens: + if re.search(r"[a-zA-Z0-9]", token): + fts_terms.append(token + "*") + else: + fts_terms.append(token) + fts_query = " OR ".join(fts_terms) + elif tokens: + # Single token: still apply prefix matching for ASCII tokens + token = tokens[0] + fts_query = token + "*" if re.search(r"[a-zA-Z0-9]", token) else token + else: + fts_query = query + params: list = [] where_clauses = ["facts_fts MATCH ?"] - params.append(query) + params.append(fts_query) if category: where_clauses.append("f.category = ?") diff --git a/plugins/memory/holographic/store.py b/plugins/memory/holographic/store.py index 3dc66d68648c7..27b6f2e6d8be9 100644 --- a/plugins/memory/holographic/store.py +++ b/plugins/memory/holographic/store.py @@ -316,6 +316,18 @@ def remove_fact(self, fact_id: int) -> bool: self._rebuild_bank(row["category"]) return True + def remove_fact_by_content(self, content: str) -> bool: + """Delete a fact by exact content match. Returns True if a row was deleted.""" + if not content: + return False + with self._lock: + row = self._conn.execute( + "SELECT fact_id FROM facts WHERE content = ?", (content,) + ).fetchone() + if row is None: + return False + return self.remove_fact(row["fact_id"]) + def list_facts( self, category: str | None = None, diff --git a/run_agent.py b/run_agent.py index eaafac5b43f78..cfd008f87592b 100644 --- a/run_agent.py +++ b/run_agent.py @@ -7448,6 +7448,13 @@ def flush_memories(self, messages: list = None, min_turns: int = None): tools=[memory_tool_def], temperature=_flush_temperature, max_tokens=5120, + main_runtime={ + "model": self.model, + "provider": self.provider, + "base_url": self.base_url, + "api_key": self.api_key, + "api_mode": self.api_mode, + }, # timeout resolved from auxiliary.flush_memories.timeout config ) except RuntimeError: @@ -7740,13 +7747,16 @@ def _invoke_tool(self, function_name: str, function_args: dict, effective_task_i store=self._memory_store, ) # Bridge: notify external memory provider of built-in memory writes - if self._memory_manager and function_args.get("action") in ("add", "replace"): + if self._memory_manager and function_args.get("action") in ("add", "replace", "remove"): try: - self._memory_manager.on_memory_write( - function_args.get("action", ""), - target, - function_args.get("content", ""), + action = function_args.get("action", "") + # remove: pass the actual deleted entry so provider can clean up + # add/replace: pass the content + write_content = ( + result.get("deleted_entry", "") if action == "remove" + else function_args.get("content", "") ) + self._memory_manager.on_memory_write(action, target, write_content) except Exception: pass return result @@ -8251,13 +8261,14 @@ def _execute_tool_calls_sequential(self, assistant_message, messages: list, effe store=self._memory_store, ) # Bridge: notify external memory provider of built-in memory writes - if self._memory_manager and function_args.get("action") in ("add", "replace"): + if self._memory_manager and function_args.get("action") in ("add", "replace", "remove"): try: - self._memory_manager.on_memory_write( - function_args.get("action", ""), - target, - function_args.get("content", ""), + action = function_args.get("action", "") + write_content = ( + function_result.get("deleted_entry", "") if action == "remove" + else function_args.get("content", "") ) + self._memory_manager.on_memory_write(action, target, write_content) except Exception: pass tool_duration = time.time() - tool_start_time diff --git a/tests/agent/test_title_generator.py b/tests/agent/test_title_generator.py index 98fb8fb21310d..a8220a1af3ac7 100644 --- a/tests/agent/test_title_generator.py +++ b/tests/agent/test_title_generator.py @@ -150,7 +150,7 @@ def test_fires_on_first_exchange(self): # Wait for the daemon thread to complete import time time.sleep(0.3) - mock_auto.assert_called_once_with(db, "sess-1", "hello", "hi there") + mock_auto.assert_called_once_with(db, "sess-1", "hello", "hi there", None) def test_skips_if_no_response(self): db = MagicMock() diff --git a/tests/gateway/restart_test_helpers.py b/tests/gateway/restart_test_helpers.py index 6332a194fe293..fa9f8ca5aae4e 100644 --- a/tests/gateway/restart_test_helpers.py +++ b/tests/gateway/restart_test_helpers.py @@ -62,6 +62,7 @@ def make_restart_runner( runner._restart_detached = False runner._restart_via_service = False runner._restart_drain_timeout = DEFAULT_GATEWAY_RESTART_DRAIN_TIMEOUT + runner._restart_caller_key = None runner._stop_task = None runner._busy_input_mode = "interrupt" runner._update_prompt_pending = {} diff --git a/tests/gateway/test_clean_shutdown_marker.py b/tests/gateway/test_clean_shutdown_marker.py index 1a476bc49a575..d2410affc20cf 100644 --- a/tests/gateway/test_clean_shutdown_marker.py +++ b/tests/gateway/test_clean_shutdown_marker.py @@ -206,6 +206,7 @@ def test_marker_written_on_restart_stop(self, tmp_path, monkeypatch): runner._background_tasks = set() runner._shutdown_event = MagicMock() runner._restart_drain_timeout = 5 + runner._restart_caller_key = None runner._exit_code = None runner._exit_reason = None runner.adapters = {} diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index f405cf8bd51b8..06b59df430ced 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -581,6 +581,30 @@ def test_sanitize_fts5_quotes_dotted_terms(self): assert s('my-app.config') == '"my-app.config"' assert s('my-app.config.ts') == '"my-app.config.ts"' + def test_sanitize_fts5_quotes_underscored_terms(self): + """Underscored terms should be wrapped in quotes for exact matching. + + FTS5 default tokenizer splits 'sp_new1' into tokens 'sp' and 'new1'. + Without quoting, a search for 'sp_new' becomes an AND query + ('sp AND new') that fails to match rows indexed as 'sp_new1'. + """ + from hermes_state import SessionDB + s = SessionDB._sanitize_fts5_query + # Simple underscored term + assert s('sp_new') == '"sp_new"' + # Multiple underscores + assert s('a_b_c') == '"a_b_c"' + # Mixed underscores and hyphens/dots — single pass avoids double-quoting + assert s('sp_new1') == '"sp_new1"' + assert s('docker-compose_up') == '"docker-compose_up"' + assert s('my.app_config.ts') == '"my.app_config.ts"' + # Already-quoted — no double quoting + assert s('"sp_new"') == '"sp_new"' + # Mixed with other words + result = s('sp_new and 血管瘤') + assert '"sp_new"' in result + assert '血管瘤' in result + # ========================================================================= # CJK (Chinese/Japanese/Korean) LIKE fallback diff --git a/tools/memory_tool.py b/tools/memory_tool.py index eef64e709669b..761d3e4e9c13b 100644 --- a/tools/memory_tool.py +++ b/tools/memory_tool.py @@ -350,11 +350,15 @@ def remove(self, target: str, old_text: str) -> Dict[str, Any]: # All identical -- safe to remove just the first idx = matches[0][0] + deleted_entry = entries[idx] entries.pop(idx) self._set_entries(target, entries) self.save_to_disk(target) - return self._success_response(target, "Entry removed.") + return {"success": True, "target": target, "deleted_entry": deleted_entry, + "message": "Entry removed.", "entries": self._entries_for(target), + "usage": f"{self._char_count(target)}/{self._char_limit(target)} chars", + "entry_count": len(self._entries_for(target))} def format_for_system_prompt(self, target: str) -> Optional[str]: """ diff --git a/website/docs/user-guide/features/memory-providers.md b/website/docs/user-guide/features/memory-providers.md index afbdac5fca410..185a92aad41d4 100644 --- a/website/docs/user-guide/features/memory-providers.md +++ b/website/docs/user-guide/features/memory-providers.md @@ -51,7 +51,7 @@ AI-native cross-session user modeling with dialectic reasoning, session-scoped c | **Data storage** | Honcho Cloud or self-hosted | | **Cost** | Honcho pricing (cloud) / free (self-hosted) | -**Tools (5):** `honcho_profile` (read/update peer card), `honcho_search` (semantic search), `honcho_context` (session context — summary, representation, card, messages), `honcho_reasoning` (LLM-synthesized), `honcho_conclude` (create/delete conclusions) +**Tools (6):** `honcho_profile` (read/update peer card), `honcho_search` (semantic search), `honcho_context` (session context — summary, representation, card, messages), `honcho_reasoning` (LLM-synthesized), `honcho_conclude` (create/delete conclusions), `honcho_sync` (sync peer card) **Architecture:** Two-layer context injection — a base layer (session summary + representation + peer card, refreshed on `contextCadence`) plus a dialectic supplement (LLM reasoning, refreshed on `dialecticCadence`). The dialectic automatically selects cold-start prompts (general user facts) vs. warm prompts (session-scoped context) based on whether base context exists.