fix(openviking): gate memory writes and add viking_forget
Mirror built-in memory writes to external providers only after the native memory tool succeeds and is not staged for approval. Keep OpenViking's built-in memory mirroring add-only, since Hermes native memory entries do not yet have stable OpenViking file URIs for replace/remove. Add a narrow viking_forget tool for exact user memory file deletion and document the current OpenViking write/delete behavior.
This commit is contained in:
parent
38c56a1e86
commit
70e7132e2f
9 changed files with 560 additions and 45 deletions
61
agent/memory_write_bridge.py
Normal file
61
agent/memory_write_bridge.py
Normal file
|
|
@ -0,0 +1,61 @@
|
|||
"""Helpers for mirroring built-in memory writes to external providers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from typing import Any, Dict, List
|
||||
|
||||
_MIRRORED_MEMORY_ACTIONS = {"add", "replace", "remove"}
|
||||
|
||||
|
||||
def _memory_tool_result_succeeded(result: Any) -> bool:
|
||||
if isinstance(result, str):
|
||||
try:
|
||||
result = json.loads(result)
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
if isinstance(result, dict):
|
||||
if result.get("success") is False:
|
||||
return False
|
||||
if result.get("staged") is True:
|
||||
return False
|
||||
if "error" in result and result.get("success") is not True:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def collect_memory_write_notifications(
|
||||
tool_result: Any,
|
||||
tool_args: Dict[str, Any],
|
||||
) -> List[Dict[str, str]]:
|
||||
"""Return provider notifications for a successful built-in memory write."""
|
||||
if not _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"),
|
||||
}]
|
||||
|
||||
notifications: List[Dict[str, str]] = []
|
||||
for op in raw_operations:
|
||||
if not isinstance(op, dict):
|
||||
continue
|
||||
action = str(op.get("action") or "")
|
||||
if action not in _MIRRORED_MEMORY_ACTIONS:
|
||||
continue
|
||||
notifications.append({
|
||||
"action": action,
|
||||
"target": target,
|
||||
"content": str(op.get("content") or ""),
|
||||
"old_text": str(op.get("old_text") or ""),
|
||||
})
|
||||
return notifications
|
||||
Loading…
Add table
Add a link
Reference in a new issue