fix(agent): sanitize tool results before observers

agent/tool_executor.py

@@ -31,6 +31,7 @@ from agent.display import (
     _detect_tool_failure,
 )
 from agent.tool_guardrails import ToolGuardrailDecision
+from agent.tool_result_observers import observer_safe_tool_result
 from agent.tool_dispatch_helpers import (
     _is_destructive_command,
     _is_multimodal_tool_result,
@@ -599,8 +600,9 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
                 logger.error("_invoke_tool raised for %s: %s", function_name, tool_error, exc_info=True)
             duration = time.time() - start
             is_error, _ = _detect_tool_failure(function_name, result)
+            observer_result = observer_safe_tool_result(result, function_name, config=_tool_budget)
             if is_error:
-                logger.info("tool %s failed (%.2fs): %s", function_name, duration, result[:200])
+                logger.info("tool %s failed (%.2fs): %s", function_name, duration, str(observer_result)[:200])
             else:
                 logger.info("tool %s completed (%.2fs, %d chars)", function_name, duration, len(result))
             results[index] = (function_name, function_args, result, duration, is_error, False, middleware_trace)
@@ -794,6 +796,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
     for i, (tc, name, args, middleware_trace, block_result, blocked_by_guardrail) in enumerate(parsed_calls):
         r = results[i]
         blocked = False
+        observer_result = None
         # A worker can finish and write results[i] in the window between the
         # deadline snapshot (timed_out_indices, taken from not_done) and this
         # loop. Prefer that real result over a fabricated timeout message — the
@@ -856,8 +859,12 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
                     failed=is_error,
                 )
 
+            observer_result = observer_safe_tool_result(
+                function_result, function_name, config=_tool_budget,
+            )
+
             if is_error:
-                _err_text = _multimodal_text_summary(function_result)
+                _err_text = _multimodal_text_summary(observer_result)
                 result_preview = _err_text[:200] if len(_err_text) > 200 else _err_text
                 logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview)
 
@@ -877,21 +884,27 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
                     agent.tool_progress_callback(
                         "tool.completed", function_name, None, None,
                         duration=tool_duration, is_error=is_error,
-                        result=function_result,
+                        result=observer_result,
                     )
                 except Exception as cb_err:
                     logging.debug(f"Tool progress callback error: {cb_err}")
 
             if agent.verbose_logging:
                 logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s")
-                logging.debug(f"Tool result ({len(function_result)} chars): {function_result}")
+                _observer_text = _multimodal_text_summary(observer_result)
+                logging.debug(f"Tool result ({len(_observer_text)} chars): {_observer_text}")
+
+        if observer_result is None:
+            observer_result = observer_safe_tool_result(
+                function_result, name, config=_tool_budget,
+            )
 
         # Print cute message per tool
         if agent._should_emit_quiet_tool_messages():
-            cute_msg = _get_cute_tool_message_impl(name, args, tool_duration, result=function_result)
+            cute_msg = _get_cute_tool_message_impl(name, args, tool_duration, result=observer_result)
             agent._safe_print(f"  {cute_msg}")
         elif not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off":
-            _preview_str = _multimodal_text_summary(function_result)
+            _preview_str = _multimodal_text_summary(observer_result)
             if agent.verbose_logging:
                 print(f"  ✅ Tool {i+1} completed in {tool_duration:.2f}s")
                 print(agent._wrap_verbose("Result: ", _preview_str))
@@ -905,7 +918,7 @@ def execute_tool_calls_concurrent(agent, assistant_message, messages: list, effe
         if not blocked and agent.tool_complete_callback:
             try:
                 display_args = _redact_tool_args_for_display(name, args) or args
-                agent.tool_complete_callback(tc.id, name, display_args, function_result)
+                agent.tool_complete_callback(tc.id, name, display_args, observer_result)
             except Exception as cb_err:
                 logging.debug(f"Tool complete callback error: {cb_err}")
 
@@ -1182,7 +1195,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             )
             tool_duration = time.time() - tool_start_time
             if agent._should_emit_quiet_tool_messages():
-                agent._vprint(f"  {_get_cute_tool_message_impl('todo', function_args, tool_duration, result=function_result)}")
+                agent._vprint(f"  {_get_cute_tool_message_impl('todo', function_args, tool_duration, result=observer_safe_tool_result(function_result, function_name, config=_tool_budget))}")
         elif function_name == "session_search":
             def _execute(next_args: dict) -> Any:
                 session_db = agent._get_session_db_for_recall()
@@ -1211,7 +1224,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             )
             tool_duration = time.time() - tool_start_time
             if agent._should_emit_quiet_tool_messages():
-                agent._vprint(f"  {_get_cute_tool_message_impl('session_search', function_args, tool_duration, result=function_result)}")
+                agent._vprint(f"  {_get_cute_tool_message_impl('session_search', function_args, tool_duration, result=observer_safe_tool_result(function_result, function_name, config=_tool_budget))}")
         elif function_name == "memory":
             def _execute(next_args: dict) -> Any:
                 target = next_args.get("target", "memory")
@@ -1248,7 +1261,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             )
             tool_duration = time.time() - tool_start_time
             if agent._should_emit_quiet_tool_messages():
-                agent._vprint(f"  {_get_cute_tool_message_impl('memory', function_args, tool_duration, result=function_result)}")
+                agent._vprint(f"  {_get_cute_tool_message_impl('memory', function_args, tool_duration, result=observer_safe_tool_result(function_result, function_name, config=_tool_budget))}")
         elif function_name == "clarify":
             def _execute(next_args: dict) -> Any:
                 from tools.clarify_tool import clarify_tool as _clarify_tool
@@ -1267,7 +1280,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             )
             tool_duration = time.time() - tool_start_time
             if agent._should_emit_quiet_tool_messages():
-                agent._vprint(f"  {_get_cute_tool_message_impl('clarify', function_args, tool_duration, result=function_result)}")
+                agent._vprint(f"  {_get_cute_tool_message_impl('clarify', function_args, tool_duration, result=observer_safe_tool_result(function_result, function_name, config=_tool_budget))}")
         elif function_name == "read_terminal":
             def _execute(next_args: dict) -> Any:
                 from tools.read_terminal_tool import read_terminal_tool as _read_terminal_tool
@@ -1286,7 +1299,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             )
             tool_duration = time.time() - tool_start_time
             if agent._should_emit_quiet_tool_messages():
-                agent._vprint(f"  {_get_cute_tool_message_impl('read_terminal', function_args, tool_duration, result=function_result)}")
+                agent._vprint(f"  {_get_cute_tool_message_impl('read_terminal', function_args, tool_duration, result=observer_safe_tool_result(function_result, function_name, config=_tool_budget))}")
         elif function_name == "delegate_task":
             tasks_arg = function_args.get("tasks")
             if tasks_arg and isinstance(tasks_arg, list):
@@ -1320,7 +1333,10 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
             finally:
                 agent._delegate_spinner = None
                 tool_duration = time.time() - tool_start_time
-                cute_msg = _get_cute_tool_message_impl('delegate_task', function_args, tool_duration, result=_delegate_result)
+                cute_msg = _get_cute_tool_message_impl(
+                    'delegate_task', function_args, tool_duration,
+                    result=observer_safe_tool_result(_delegate_result, function_name, config=_tool_budget),
+                )
                 if spinner:
                     spinner.stop(cute_msg)
                 elif agent._should_emit_quiet_tool_messages():
@@ -1353,7 +1369,10 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                 logger.error("context_engine.handle_tool_call raised for %s: %s", function_name, tool_error, exc_info=True)
             finally:
                 tool_duration = time.time() - tool_start_time
-                cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_ce_result)
+                cute_msg = _get_cute_tool_message_impl(
+                    function_name, function_args, tool_duration,
+                    result=observer_safe_tool_result(_ce_result, function_name, config=_tool_budget),
+                )
                 if spinner:
                     spinner.stop(cute_msg)
                 elif agent._should_emit_quiet_tool_messages():
@@ -1387,7 +1406,10 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                 logger.error("memory_manager.handle_tool_call raised for %s: %s", function_name, tool_error, exc_info=True)
             finally:
                 tool_duration = time.time() - tool_start_time
-                cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_mem_result)
+                cute_msg = _get_cute_tool_message_impl(
+                    function_name, function_args, tool_duration,
+                    result=observer_safe_tool_result(_mem_result, function_name, config=_tool_budget),
+                )
                 if spinner:
                     spinner.stop(cute_msg)
                 elif agent._should_emit_quiet_tool_messages():
@@ -1438,7 +1460,10 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                 logger.error("handle_function_call raised for %s: %s", function_name, tool_error, exc_info=True)
             finally:
                 tool_duration = time.time() - tool_start_time
-                cute_msg = _get_cute_tool_message_impl(function_name, function_args, tool_duration, result=_spinner_result)
+                cute_msg = _get_cute_tool_message_impl(
+                    function_name, function_args, tool_duration,
+                    result=observer_safe_tool_result(_spinner_result, function_name, config=_tool_budget),
+                )
                 if spinner:
                     spinner.stop(cute_msg)
                 elif agent._should_emit_quiet_tool_messages():
@@ -1520,9 +1545,13 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                 function_result,
                 failed=_is_error_result,
             )
-            result_preview = function_result if agent.verbose_logging else (
-                function_result[:200] if len(function_result) > 200 else function_result
-            )
+        observer_result = observer_safe_tool_result(
+            function_result, function_name, config=_tool_budget,
+        )
+        observer_text = _multimodal_text_summary(observer_result)
+        result_preview = observer_text if agent.verbose_logging else (
+            observer_text[:200] if len(observer_text) > 200 else observer_text
+        )
         if _is_error_result:
             logger.warning("Tool %s returned error (%.2fs): %s", function_name, tool_duration, result_preview)
         else:
@@ -1545,7 +1574,7 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
                 agent.tool_progress_callback(
                     "tool.completed", function_name, None, None,
                     duration=tool_duration, is_error=_is_error_result,
-                    result=function_result,
+                    result=observer_result,
                 )
             except Exception as cb_err:
                 logging.debug(f"Tool progress callback error: {cb_err}")
@@ -1555,13 +1584,13 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
 
         if agent.verbose_logging:
             logging.debug(f"Tool {function_name} completed in {tool_duration:.2f}s")
-            _log_result = _multimodal_text_summary(function_result)
+            _log_result = _multimodal_text_summary(observer_result)
             logging.debug(f"Tool result ({len(_log_result)} chars): {_log_result}")
 
         if not _execution_blocked and agent.tool_complete_callback:
             try:
                 display_args = _redact_tool_args_for_display(function_name, function_args) or function_args
-                agent.tool_complete_callback(tool_call.id, function_name, display_args, function_result)
+                agent.tool_complete_callback(tool_call.id, function_name, display_args, observer_result)
             except Exception as cb_err:
                 logging.debug(f"Tool complete callback error: {cb_err}")
 
@@ -1600,10 +1629,9 @@ def execute_tool_calls_sequential(agent, assistant_message, messages: list, effe
         if not agent.quiet_mode and getattr(agent, "tool_progress_mode", "all") != "off":
             if agent.verbose_logging:
                 print(f"  ✅ Tool {i} completed in {tool_duration:.2f}s")
-                print(agent._wrap_verbose("Result: ", function_result))
+                print(agent._wrap_verbose("Result: ", observer_text))
             else:
-                _fr_str = function_result if isinstance(function_result, str) else str(function_result)
-                response_preview = _fr_str[:agent.log_prefix_chars] + "..." if len(_fr_str) > agent.log_prefix_chars else _fr_str
+                response_preview = observer_text[:agent.log_prefix_chars] + "..." if len(observer_text) > agent.log_prefix_chars else observer_text
                 print(f"  ✅ Tool {i} completed in {tool_duration:.2f}s - {response_preview}")
 
         if agent._interrupt_requested and i < len(assistant_message.tool_calls):

agent/tool_result_observers.py

@@ -0,0 +1,58 @@
+"""Safe projection of tool results for logs and observer callbacks.
+
+Tool results may remain full-fidelity on the internal dispatch/context path, but
+must cross this boundary before being exposed to logs, plugins, progress feeds,
+or completion callbacks.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from agent.redact import redact_sensitive_text
+from tools.budget_config import BudgetConfig, DEFAULT_BUDGET
+from tools.tool_result_storage import maybe_persist_tool_result
+
+
+def observer_safe_tool_result(
+    result: Any,
+    tool_name: str,
+    *,
+    config: BudgetConfig = DEFAULT_BUDGET,
+) -> Any:
+    """Return a redacted, size-bounded copy suitable for external observers.
+
+    Small structured plugin/MCP results retain their JSON shape. Oversized
+    values become the same bounded inline preview used by tool-result context
+    budgeting, without persisting a second (observer-only) copy to disk.
+    Redaction is forced because observer boundaries must stay safe even when
+    model-context redaction has been explicitly disabled.
+    """
+    was_string = isinstance(result, str)
+    if was_string:
+        serialized = result
+    else:
+        try:
+            serialized = json.dumps(result, ensure_ascii=False, default=str)
+        except Exception:
+            serialized = str(result)
+
+    redacted = redact_sensitive_text(serialized, force=True)
+    bounded = maybe_persist_tool_result(
+        content=redacted,
+        tool_name=tool_name,
+        tool_use_id="observer",
+        env=None,
+        config=config,
+        # Observer surfaces are always bounded, including read_file whose
+        # context threshold is intentionally infinite to avoid spill loops.
+        threshold=config.default_result_size,
+    )
+
+    if was_string or bounded != redacted:
+        return bounded
+    try:
+        return json.loads(bounded)
+    except (TypeError, ValueError):
+        return bounded

hermes_cli/backup.py

@@ -257,30 +257,23 @@ def _safe_copy_db(src: Path, dst: Path) -> bool:
     """Copy a SQLite database safely using the backup() API.
 
     Handles WAL mode — produces a consistent snapshot even while
-    the DB is being written to. Fail closed if a consistent snapshot cannot
-    be created: copying only the live main file can omit committed WAL data.
+    the DB is being written to.  Falls back to raw copy on failure.
     """
-    conn = None
-    backup_conn = None
     try:
         conn = sqlite3.connect(f"file:{src}?mode=ro", uri=True)
         backup_conn = sqlite3.connect(str(dst))
         conn.backup(backup_conn)
+        backup_conn.close()
+        conn.close()
         return True
     except Exception as exc:
         logger.warning("SQLite safe copy failed for %s: %s", src, exc)
         try:
-            dst.unlink(missing_ok=True)
-        except OSError:
-            pass
-        return False
-    finally:
-        for connection in (backup_conn, conn):
-            if connection is not None:
-                try:
-                    connection.close()
-                except Exception:
-                    pass
+            shutil.copy2(src, dst)
+            return True
+        except Exception as exc2:
+            logger.error("Raw copy also failed for %s: %s", src, exc2)
+            return False
 
 
 # ---------------------------------------------------------------------------
@@ -436,10 +429,7 @@ def run_backup(args) -> None:
 
     # Summary
     print()
-    if errors:
-        print(f"Backup incomplete: {out_path}")
-    else:
-        print(f"Backup complete: {out_path}")
+    print(f"Backup complete: {out_path}")
     print(f"  Files:       {file_count}")
     print(f"  Original:    {_format_size(total_bytes)}")
     print(f"  Compressed:  {_format_size(zip_size)}")
@@ -471,8 +461,7 @@ def run_backup(args) -> None:
         if len(errors) > 10:
             print(f"  ... and {len(errors) - 10} more")
 
-    if not errors:
-        print(f"\nRestore with: hermes import {out_path.name}")
+    print(f"\nRestore with: hermes import {out_path.name}")
 
 
 # ---------------------------------------------------------------------------
@@ -1188,7 +1177,6 @@ def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]:
     if not files_to_add:
         return None
 
-    sqlite_snapshot_failed = False
     try:
         with zipfile.ZipFile(out_path, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as zf:
             for abs_path, rel_path in files_to_add:
@@ -1203,14 +1191,8 @@ def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]:
                         ) as tmp:
                             tmp_db = Path(tmp.name)
                         try:
-                            if not _safe_copy_db(abs_path, tmp_db):
-                                logger.warning(
-                                    "Full-zip backup aborted: SQLite snapshot failed for %s",
-                                    rel_path,
-                                )
-                                sqlite_snapshot_failed = True
-                                break
-                            zf.write(tmp_db, arcname=str(rel_path))
+                            if _safe_copy_db(abs_path, tmp_db):
+                                zf.write(tmp_db, arcname=str(rel_path))
                         finally:
                             tmp_db.unlink(missing_ok=True)
                     else:
@@ -1227,13 +1209,6 @@ def _write_full_zip_backup(out_path: Path, hermes_root: Path) -> Optional[Path]:
             pass
         return None
 
-    if sqlite_snapshot_failed:
-        try:
-            out_path.unlink(missing_ok=True)
-        except OSError:
-            pass
-        return None
-
     return out_path
 
 

model_tools.py

@@ -971,6 +971,13 @@ def _tool_result_observer_fields(result: Any) -> tuple[str, Optional[str], Optio
     return "ok", None, None
 
 
+def _observer_safe_tool_result(result: Any, function_name: str) -> Any:
+    """Project a raw tool result through the shared observer security boundary."""
+    from agent.tool_result_observers import observer_safe_tool_result
+
+    return observer_safe_tool_result(result, function_name)
+
+
 def _emit_post_tool_call_hook(
     *,
     function_name: str,
@@ -1000,13 +1007,14 @@ def _emit_post_tool_call_hook(
         from hermes_cli.plugins import has_hook, invoke_hook
         if not has_hook("post_tool_call"):
             return
+        observer_result = _observer_safe_tool_result(result, function_name)
         if status is None:
-            status, error_type, error_message = _tool_result_observer_fields(result)
+            status, error_type, error_message = _tool_result_observer_fields(observer_result)
         invoke_hook(
             "post_tool_call",
             tool_name=function_name,
             args=function_args,
-            result=result,
+            result=observer_result,
             task_id=task_id or "",
             session_id=session_id or "",
             tool_call_id=tool_call_id or "",
@@ -1321,12 +1329,13 @@ def handle_function_call(
         try:
             from hermes_cli.plugins import has_hook, invoke_hook
             if has_hook("transform_tool_result"):
-                status, error_type, error_message = _tool_result_observer_fields(result)
+                observer_result = _observer_safe_tool_result(result, function_name)
+                status, error_type, error_message = _tool_result_observer_fields(observer_result)
                 hook_results = invoke_hook(
                     "transform_tool_result",
                     tool_name=function_name,
                     args=function_args,
-                    result=result,
+                    result=observer_result,
                     task_id=task_id or "",
                     session_id=session_id or "",
                     tool_call_id=tool_call_id or "",

tests/agent/test_tool_result_observers.py

@@ -0,0 +1,67 @@
+"""Security boundaries for tool-result logs and observer callbacks."""
+
+import json
+
+import model_tools
+from tools.budget_config import BudgetConfig
+
+
+SECRET = "sk-proj-ABCD1234567890EFGH"
+
+
+def test_observer_safe_result_redacts_and_bounds_string_without_mutating_source():
+    from agent.tool_result_observers import observer_safe_tool_result
+
+    raw = f'token={SECRET}\n' + ("x" * 200)
+    config = BudgetConfig(default_result_size=80, preview_size=40)
+
+    observed = observer_safe_tool_result(raw, "mcp_demo", config=config)
+
+    assert SECRET in raw
+    assert SECRET not in observed
+    assert "Truncated" in observed
+    assert len(observed) < len(raw)
+
+
+def test_observer_safe_result_preserves_small_non_string_shape_and_redacts_nested_values():
+    from agent.tool_result_observers import observer_safe_tool_result
+
+    raw = {"content": [{"type": "text", "text": SECRET}], "count": 1}
+
+    observed = observer_safe_tool_result(raw, "plugin_demo")
+
+    assert isinstance(observed, dict)
+    assert observed["count"] == 1
+    assert SECRET not in json.dumps(observed)
+    assert raw["content"][0]["text"] == SECRET
+
+
+def test_post_and_transform_plugin_observers_receive_safe_result_but_dispatch_result_stays_full(monkeypatch):
+    raw = {"payload": SECRET, "body": "y" * 110_000}
+    observed = []
+
+    monkeypatch.setattr(model_tools.registry, "dispatch", lambda *args, **kwargs: raw)
+    monkeypatch.setattr(model_tools, "_READ_SEARCH_TOOLS", frozenset())
+    monkeypatch.setattr("hermes_cli.plugins.has_hook", lambda name: True)
+
+    def invoke_hook(name, **kwargs):
+        if name in {"post_tool_call", "transform_tool_result"}:
+            observed.append((name, kwargs["result"]))
+        return []
+
+    monkeypatch.setattr("hermes_cli.plugins.invoke_hook", invoke_hook)
+
+    result = model_tools.handle_function_call(
+        "mcp_demo",
+        {},
+        task_id="task",
+        tool_call_id="call",
+        skip_pre_tool_call_hook=True,
+    )
+
+    assert result is raw
+    assert [name for name, _ in observed] == ["post_tool_call", "transform_tool_result"]
+    for _, value in observed:
+        serialized = json.dumps(value) if not isinstance(value, str) else value
+        assert SECRET not in serialized
+        assert len(serialized) < 10_000

tests/hermes_cli/test_backup.py

@@ -19,10 +19,8 @@ def _make_hermes_tree(root: Path) -> None:
     """Create a realistic ~/.hermes directory structure for testing."""
     (root / "config.yaml").write_text("model:\n  provider: openrouter\n")
     (root / ".env").write_text("OPENROUTER_API_KEY=sk-test-123\n")
-    for db_name in ("memory_store.db", "hermes_state.db"):
-        with sqlite3.connect(root / db_name) as conn:
-            conn.execute("CREATE TABLE sample (value TEXT)")
-            conn.execute("INSERT INTO sample VALUES ('test')")
+    (root / "memory_store.db").write_bytes(b"fake-sqlite")
+    (root / "hermes_state.db").write_bytes(b"fake-state")
 
     # Sessions
     (root / "sessions").mkdir(exist_ok=True)
@@ -227,65 +225,6 @@ class TestBackup:
             # Skins
             assert "skins/cyber.yaml" in names
 
-    def test_failed_sqlite_backup_never_raw_copies_live_wal_db(self, tmp_path, monkeypatch, capsys):
-        """A failed backup() must not silently archive the stale main DB file.
-
-        Keep a real, uncheckpointed WAL transaction live so a raw copy of only
-        ``state.db`` would be a valid-looking but torn snapshot.
-        """
-        hermes_home = tmp_path / ".hermes"
-        hermes_home.mkdir()
-        (hermes_home / "config.yaml").write_text("model: test\n")
-        db_path = hermes_home / "state.db"
-
-        writer = sqlite3.connect(db_path)
-        writer.execute("PRAGMA journal_mode=WAL")
-        writer.execute("PRAGMA wal_autocheckpoint=0")
-        writer.execute("CREATE TABLE events (value TEXT)")
-        writer.commit()
-        writer.execute("PRAGMA wal_checkpoint(TRUNCATE)")
-        writer.execute("INSERT INTO events VALUES ('only-in-wal')")
-        writer.commit()
-        assert Path(f"{db_path}-wal").stat().st_size > 0
-
-        monkeypatch.setenv("HERMES_HOME", str(hermes_home))
-        monkeypatch.setattr(Path, "home", lambda: tmp_path)
-
-        import hermes_cli.backup as backup_mod
-        real_connect = backup_mod.sqlite3.connect
-
-        class FailingBackupConnection:
-            def __init__(self, connection):
-                self._connection = connection
-
-            def backup(self, _destination):
-                raise sqlite3.OperationalError("forced backup failure")
-
-            def close(self):
-                self._connection.close()
-
-        def connect_with_failed_backup(database, *args, **kwargs):
-            connection = real_connect(database, *args, **kwargs)
-            if str(database).startswith(f"file:{db_path}"):
-                return FailingBackupConnection(connection)
-            return connection
-
-        monkeypatch.setattr(backup_mod.sqlite3, "connect", connect_with_failed_backup)
-        out_zip = tmp_path / "backup.zip"
-        try:
-            backup_mod.run_backup(Namespace(output=str(out_zip)))
-        finally:
-            writer.close()
-
-        with zipfile.ZipFile(out_zip) as zf:
-            assert "config.yaml" in zf.namelist()
-            assert "state.db" not in zf.namelist()
-
-        output = capsys.readouterr().out
-        assert "Backup incomplete" in output
-        assert "state.db: SQLite safe copy failed" in output
-        assert "Restore with:" not in output
-
     def test_db_snapshots_staged_beside_output_zip(self, tmp_path, monkeypatch):
         """SQLite staging temp files must be created on the output zip's
         filesystem (dir=out_path.parent), NOT the system /tmp default — a
@@ -1929,53 +1868,6 @@ class TestPreUpdateBackup:
     """Tests for create_pre_update_backup — the auto-backup ``hermes update``
     runs before touching anything."""
 
-    def test_failed_sqlite_snapshot_removes_incomplete_archive(self, tmp_path, monkeypatch):
-        """The non-interactive full-zip helper must fail the entire archive
-        rather than return success after omitting a live WAL database."""
-        hermes_home = tmp_path / ".hermes"
-        hermes_home.mkdir()
-        (hermes_home / "config.yaml").write_text("model: test\n")
-        db_path = hermes_home / "state.db"
-
-        writer = sqlite3.connect(db_path)
-        writer.execute("PRAGMA journal_mode=WAL")
-        writer.execute("PRAGMA wal_autocheckpoint=0")
-        writer.execute("CREATE TABLE events (value TEXT)")
-        writer.commit()
-        writer.execute("PRAGMA wal_checkpoint(TRUNCATE)")
-        writer.execute("INSERT INTO events VALUES ('only-in-wal')")
-        writer.commit()
-        assert Path(f"{db_path}-wal").stat().st_size > 0
-
-        import hermes_cli.backup as backup_mod
-        real_connect = backup_mod.sqlite3.connect
-
-        class FailingBackupConnection:
-            def __init__(self, connection):
-                self._connection = connection
-
-            def backup(self, _destination):
-                raise sqlite3.OperationalError("forced backup failure")
-
-            def close(self):
-                self._connection.close()
-
-        def connect_with_failed_backup(database, *args, **kwargs):
-            connection = real_connect(database, *args, **kwargs)
-            if str(database).startswith(f"file:{db_path}"):
-                return FailingBackupConnection(connection)
-            return connection
-
-        monkeypatch.setattr(backup_mod.sqlite3, "connect", connect_with_failed_backup)
-        out_zip = tmp_path / "pre-update.zip"
-        try:
-            result = backup_mod._write_full_zip_backup(out_zip, hermes_home)
-        finally:
-            writer.close()
-
-        assert result is None
-        assert not out_zip.exists()
-
     @pytest.fixture
     def hermes_home(self, tmp_path):
         root = tmp_path / ".hermes"

tests/run_agent/test_run_agent.py

@@ -2955,6 +2955,44 @@ class TestConcurrentToolExecution:
         assert progress[0][2].startswith("sk-pro")
         assert secret not in repr(starts + completes + progress)
 
+    @pytest.mark.parametrize("concurrent", [False, True])
+    def test_result_observers_are_redacted_and_bounded_before_callbacks_and_verbose_output(
+        self, agent, concurrent, capsys, caplog,
+    ):
+        secret = "sk-proj-ABCD1234567890EFGH"
+        raw = f'{{"token":"{secret}","body":"' + ("z" * 110_000) + '"}'
+        calls = [
+            _mock_tool_call(name="mcp_demo", arguments="{}", call_id="c1"),
+            _mock_tool_call(name="mcp_demo", arguments="{}", call_id="c2"),
+        ] if concurrent else [
+            _mock_tool_call(name="mcp_demo", arguments="{}", call_id="c1"),
+        ]
+        mock_msg = _mock_assistant_msg(content="", tool_calls=calls)
+        messages = []
+        progress_results = []
+        complete_results = []
+        agent.verbose_logging = True
+        agent.quiet_mode = False
+        agent.tool_progress_callback = lambda event, name, preview, args, **kw: (
+            progress_results.append(kw["result"]) if event == "tool.completed" else None
+        )
+        agent.tool_complete_callback = lambda _id, _name, _args, result: complete_results.append(result)
+
+        with patch("run_agent.handle_function_call", return_value=raw):
+            if concurrent:
+                agent._execute_tool_calls_concurrent(mock_msg, messages, "task-1")
+            else:
+                agent._execute_tool_calls_sequential(mock_msg, messages, "task-1")
+
+        observer_text = repr(progress_results + complete_results) + capsys.readouterr().out + caplog.text
+        assert secret not in observer_text
+        assert "z" * 10_000 not in observer_text
+        assert progress_results and complete_results
+        assert all("Truncated" in result for result in progress_results + complete_results)
+        # Observer projection must not replace the result used for internal
+        # context persistence/dispatch; that path keeps its own budget policy.
+        assert all(secret in message["content"] for message in messages)
+
     def test_invoke_tool_handles_agent_level_tools(self, agent):
         """_invoke_tool should handle todo tool directly."""
         with patch("tools.todo_tool.todo_tool", return_value='{"ok":true}') as mock_todo: