feat(memory): add read/write memory switches

coding-agent/src/coding_agent/agent.toml

@@ -43,6 +43,11 @@ timeout = 30.0
 max_entries = 100
 keep_recent = 20
 
+[memory]
+enabled = true
+read_enabled = true
+write_enabled = true
+
 [kb]
 db_path = "kb"
 embedding_model = "text-embedding-3-small"

coding-agent/src/coding_agent/core/app.py

@@ -5,6 +5,7 @@
 import uuid
 from collections.abc import Mapping
 from contextlib import contextmanager
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -43,6 +44,30 @@
 ToolFilter = Any
 
 
+@dataclass(frozen=True)
+class MemorySwitchConfig:
+    enabled: bool = True
+    read_enabled: bool = True
+    write_enabled: bool = True
+
+    @property
+    def effective_read_enabled(self) -> bool:
+        return self.enabled and self.read_enabled
+
+    @property
+    def effective_write_enabled(self) -> bool:
+        return self.enabled and self.write_enabled
+
+    def to_config_dict(self) -> dict[str, bool]:
+        return {
+            "enabled": self.enabled,
+            "read_enabled": self.read_enabled,
+            "write_enabled": self.write_enabled,
+            "effective_read_enabled": self.effective_read_enabled,
+            "effective_write_enabled": self.effective_write_enabled,
+        }
+
+
 def _local_workspace_root(environment: Environment) -> Path | None:
     local_root = environment.workspace_summary().local_root
     if local_root is None:
@@ -100,6 +125,26 @@ def _merge_shell_config(
     return merged
 
 
+def _parse_memory_switch_config(extra: Mapping[str, Any]) -> MemorySwitchConfig:
+    raw_memory_cfg = extra.get("memory", {})
+    if raw_memory_cfg is None:
+        raw_memory_cfg = {}
+    if not isinstance(raw_memory_cfg, Mapping):
+        raise ValueError("[memory] config must be a table")
+    return MemorySwitchConfig(
+        enabled=_memory_bool(raw_memory_cfg, "enabled"),
+        read_enabled=_memory_bool(raw_memory_cfg, "read_enabled"),
+        write_enabled=_memory_bool(raw_memory_cfg, "write_enabled"),
+    )
+
+
+def _memory_bool(config: Mapping[str, Any], field: str) -> bool:
+    value = config.get(field, True)
+    if type(value) is not bool:
+        raise ValueError(f"[memory].{field} must be a boolean")
+    return value
+
+
 def _child_system_prompt_suffix(tool_filter: ToolFilter) -> str:
     if tool_filter is None:
         return ""
@@ -242,6 +287,7 @@ def create_child_pipeline(
     shell_cfg = cfg.extra.get("shell", {})
     if not isinstance(shell_cfg, dict):
         raise ValueError("[shell] config must be a table")
+    memory_cfg = _parse_memory_switch_config(cfg.extra)
     policy_str = approval_mode_override or approval_cfg.get("policy", "auto")
     approval_policy_map = {
         "yolo": ApprovalPolicy.YOLO,
@@ -291,7 +337,8 @@ def create_child_pipeline(
     if not isinstance(kb_db_path, str):
         raise ValueError("[kb].db_path must be a string")
     memory_review_store = MemoryReviewStore(
-        data_dir / kb_db_path / "reviewed_memory.jsonl"
+        data_dir / kb_db_path / "reviewed_memory.jsonl",
+        candidate_writes_enabled=memory_cfg.effective_write_enabled,
     )
     observability_cfg = cfg.extra.get("observability", {})
     if not isinstance(observability_cfg, dict):
@@ -331,7 +378,10 @@ def _create_child_with_environment(**kwargs: Any) -> tuple[Any, PipelineContext]
             max_entries=sum_cfg.get("max_entries", 100),
             keep_recent=sum_cfg.get("keep_recent", 20),
         ),
-        "memory": lambda: MemoryPlugin(),
+        "memory": lambda: MemoryPlugin(
+            read_enabled=memory_cfg.effective_read_enabled,
+            write_enabled=memory_cfg.effective_write_enabled,
+        ),
         "shell_session": lambda: shell_session,
     }
 
@@ -421,7 +471,11 @@ async def _memory_handler(directive: Any) -> None:
             memory_plugin.add_memory(directive)
 
     directive_executor = DirectiveExecutor(
-        memory_handler=_memory_handler if memory_plugin is not None else None,
+        memory_handler=(
+            _memory_handler
+            if memory_plugin is not None and memory_cfg.effective_write_enabled
+            else None
+        ),
     )
 
     pipeline = Pipeline(
@@ -466,6 +520,8 @@ async def _memory_handler(directive: Any) -> None:
         "shell": _merge_shell_config(environment_config, shell_cfg),
         "structured_tool_result_scope": structured_tool_result_scope,
         "memory_review_store": memory_review_store,
+        "memory": memory_cfg.to_config_dict(),
+        "topic_recall": {"enabled": memory_cfg.effective_read_enabled},
     }
     if "isolation_policy" in environment_config:
         ctx_config["isolation_policy"] = environment_config["isolation_policy"]

coding-agent/src/coding_agent/plugins/memory.py

@@ -37,8 +37,16 @@ class MemoryPlugin:
 
     state_key = "memory"
 
-    def __init__(self, max_grounding: int = 5) -> None:
+    def __init__(
+        self,
+        max_grounding: int = 5,
+        *,
+        read_enabled: bool = True,
+        write_enabled: bool = True,
+    ) -> None:
         self._max_grounding = max_grounding
+        self._read_enabled = read_enabled
+        self._write_enabled = write_enabled
         self._memories: list[dict[str, Any]] = []
         self._working_memories: list[dict[str, Any]] = []
         self._topic_file_tags: set[str] = set()
@@ -63,14 +71,19 @@ def do_mount(self, **kwargs: Any) -> dict[str, Any]:
             if isinstance(storage_state, dict):
                 self._storage_plugin = storage_state.get("plugin")
 
-        if self._storage_plugin is not None and self._session_id is not None:
+        if (
+            self._read_enabled
+            and self._storage_plugin is not None
+            and self._session_id is not None
+        ):
             persisted = self._storage_plugin.load_memory_records(self._session_id)
             self._memories = [
                 self._apply_importance_decay(record) for record in persisted
             ]
-            self._storage_plugin.replace_memory_records(
-                self._session_id, self._memories
-            )
+            if self._write_enabled:
+                self._storage_plugin.replace_memory_records(
+                    self._session_id, self._memories
+                )
 
         return {
             "memories": self._memories,
@@ -101,6 +114,8 @@ def on_session_event(
         self, event_type: str = "", payload: dict[str, Any] | None = None, **kwargs: Any
     ) -> None:
         payload = payload or {}
+        if not self._write_enabled:
+            return
         if event_type != "topic_end":
             return
 
@@ -126,6 +141,8 @@ def build_context(
         If topic file tags are available, filter memories to those with
         overlapping tags. Falls back to importance-sorted top-N otherwise.
         """
+        if not self._read_enabled:
+            return []
         if not self._memories:
             return []
 
@@ -188,6 +205,8 @@ def on_turn_end(
           - tags: Extracted topic tags
           - importance: Heuristic score (0-1)
         """
+        if not self._write_enabled:
+            return None
         if tape is None or len(tape) == 0:
             return None
 
@@ -221,6 +240,8 @@ def on_turn_end(
         return record
 
     def add_memory(self, record: MemoryRecord) -> None:
+        if not self._write_enabled:
+            return
         tags = list(record.tags)
         self._working_memories.append(
             {

coding-agent/src/coding_agent/topics/lifecycle.py

@@ -71,11 +71,13 @@ def __init__(
         now: Callable[[], datetime] | None = None,
         topic_id_factory: Callable[[], str] | None = None,
         memory_review_store: MemoryReviewStore | None = None,
+        memory_write_enabled: bool = True,
     ) -> None:
         self._store = store
         self._now = now or (lambda: datetime.now(UTC))
         self._topic_id_factory = topic_id_factory or _new_topic_id
         self._memory_review_store = memory_review_store
+        self._memory_write_enabled = memory_write_enabled
 
     async def create_topic(
         self,
@@ -207,7 +209,12 @@ async def _close_topic(
         except Exception:
             _remove_anchor(tape, seq=seq, entry_id=anchor.id)
             raise
-        if close_status == "finalized" and self._memory_review_store is not None:
+        if (
+            close_status == "finalized"
+            and self._memory_write_enabled
+            and self._memory_review_store is not None
+            and self._memory_review_store.candidate_writes_enabled
+        ):
             candidate = propose_memory_candidate_from_topic(stored)
             if candidate is not None:
                 self._memory_review_store.add_candidate(candidate)

coding-agent/src/coding_agent/topics/memory.py

@@ -108,15 +108,25 @@ def to_dict(self) -> JSONObject:
 class MemoryReviewStore:
     """Local deterministic review store for topic-derived memory candidates."""
 
-    def __init__(self, path: Path | str | None = None) -> None:
+    def __init__(
+        self,
+        path: Path | str | None = None,
+        *,
+        candidate_writes_enabled: bool = True,
+    ) -> None:
         self._path = Path(path) if path is not None else None
+        self._candidate_writes_enabled = candidate_writes_enabled
         self._records: dict[str, ReviewedMemoryRecord] = {}
         self.load()
 
     @property
     def path(self) -> Path | None:
         return self._path
 
+    @property
+    def candidate_writes_enabled(self) -> bool:
+        return self._candidate_writes_enabled
+
     def load(self) -> None:
         if self._path is None:
             return
@@ -172,6 +182,8 @@ def add_candidate(
         self,
         candidate: TopicDerivedMemoryCandidate,
     ) -> ReviewedMemoryRecord:
+        if not self._candidate_writes_enabled:
+            return ReviewedMemoryRecord(candidate=candidate, status="candidate")
         candidate_id = _candidate_id_value(candidate)
         existing = self._records.get(candidate_id)
         if existing is not None: