core/tools: session on_todo_change and MemoryViewStore seed helpers

src/plyngent/tools/context.py

@@ -10,7 +10,7 @@ from typing import TYPE_CHECKING, Any
 from plyngent.tools.view import PersistentDataView, session_data_view
 
 if TYPE_CHECKING:
-    from collections.abc import Generator
+    from collections.abc import Callable, Generator
     from pathlib import Path
 
     from plyngent.agent.todo_stack import TodoStack
@@ -58,6 +58,9 @@ class SessionState:
     data: PersistentDataView[Any] = field(default_factory=_default_session_data)
     # Live domain object during migration (also under data["todo"] when views bind).
     todo: TodoStack | None = None
+    # Host hook after todo tools mutate (e.g. CLI memory persist). Prefer this over
+    # process-global set_todo_stack(on_change=...) when a session is bound.
+    on_todo_change: Callable[[], None] | None = None
     # Soft-confirm grants live map: tool_name → True (Phase 1 key is tool name).
     # Durable copy lives under data["grants"] (see plyngent.tools.grants).
     grants: dict[str, bool] = field(default_factory=dict)

src/plyngent/tools/todo.py

@@ -32,6 +32,13 @@ def get_todo_stack() -> TodoStack | None:
 
 
 def _notify() -> None:
+    """Fire host persist hooks: prefer session.on_todo_change, else process bind."""
+    from plyngent.tools.context import get_session
+
+    session = get_session()
+    if session is not None and session.on_todo_change is not None:
+        session.on_todo_change()
+        return
     if _on_change is not None:
         _on_change()
 
@@ -79,7 +86,7 @@ async def _with_todo_stack(mutator: Callable[[TodoStack], str]) -> str:
         # Keep view domain + buffer in sync for durable commit.
         data["todo"].store(stack)
         result = mutator(stack)
-    # View commit serialized to_raw; process on_change may still persist CLI memory.
+    # View commit serialized to_raw; host on_todo_change may persist CLI memory.
     _notify()
     return result
 

src/plyngent/tools/view.py

@@ -29,6 +29,24 @@ class MemoryViewStore:
     async def store(self, root: object) -> None:
         self._root = root
 
+    def peek(self) -> object:
+        """Return the current root without a transaction (sync hosts / tests)."""
+        return self._root
+
+    def replace(self, root: object) -> None:
+        """Replace the root without a transaction (sync hosts / tests)."""
+        self._root = root
+
+    def merge_key(self, key: str, value: object) -> None:
+        """Set *key* on a dict root (creates a dict root if needed)."""
+        root = self._root
+        if not isinstance(root, dict):
+            self._root = {key: value}
+            return
+        updated: dict[object, object] = dict(cast("dict[object, object]", root))
+        updated[key] = value
+        self._root = updated
+
 
 @dataclass
 class _TxnState: