refactor(memory): isolate store write path from facade

docs/plans/2026-05-04-062-store-mainline-facade-boundary-cleanup-plan.md

@@ -0,0 +1,106 @@
+---
+status: completed
+created: 2026-05-04
+origin: user request
+scope: store mainline CortexMemory facade boundary cleanup
+---
+
+# Store Mainline Facade Boundary Cleanup
+
+## Problem
+
+The normal `/api/v1/memory/store` path now has a clear staged flow in
+`MemoryWriteService.add`, but several write-path helper services still reach
+through `CortexMemory` for storage, filesystem, embedding, record projection,
+signals, and URI helpers. That makes the store mainline look like it still
+depends on the top-level compatibility facade rather than on the write domain
+boundary.
+
+The goal is to let the store mainline depend on `MemoryWriteService` as its
+local boundary. `CortexMemory` should keep compatibility wrappers, but the
+normal store helpers should stop calling `self._write_service._orch` directly.
+
+## Scope
+
+In scope:
+
+- Add narrow dependency accessors on
+  `src/opencortex/services/memory_write_service.py` for store helpers:
+  storage, collection, filesystem, embedder, config, signal bus, entity index,
+  record/URI helpers, and derived projection sync.
+- Update normal store-path helper services to call the write-service boundary
+  instead of `CortexMemory`:
+  - `src/opencortex/services/memory_write_context_builder.py`
+  - `src/opencortex/services/memory_write_derive_service.py`
+  - `src/opencortex/services/memory_write_embed_service.py`
+  - `src/opencortex/services/memory_write_dedup_service.py`
+  - `src/opencortex/services/memory_directory_record_service.py`
+  - `src/opencortex/services/memory_store_record_service.py`
+- Remove `_orch` properties from those helper services when no longer needed.
+- Preserve `CortexMemory` public and compatibility wrappers.
+- Preserve `/api/v1/memory/store` behavior, dedup behavior, projection sync,
+  parent directory records, signals, and CortexFS fire-and-forget semantics.
+
+Out of scope:
+
+- Update/remove mutation cleanup in `MemoryMutationService`.
+- Document ingest cleanup in `MemoryDocumentWriteService`.
+- Batch/document benchmark paths.
+- Deleting `CortexMemory` compatibility methods.
+- Changing storage filter, TTL, projection, or entity-index semantics.
+
+## Implementation Units
+
+### 1. Add Write-Service Boundary Methods
+
+Add focused accessors/delegates to `MemoryWriteService` for exactly the
+dependencies needed by the normal store helpers:
+
+- initialization and collection lookup
+- storage and filesystem
+- embedder and config
+- memory signal bus and entity index
+- URI/category helpers
+- abstract/object payload helpers
+- derive helpers used by write derive
+- record loading for explicit URI/dedup
+- TTL and anchor projection sync
+
+These delegates may still call `CortexMemory` internally. The cleanup target is
+that store helper services depend on `MemoryWriteService`, not the top-level
+facade.
+
+### 2. Move Store Helpers Off `_orch`
+
+Change the normal store helper services listed in scope to call the new
+write-service boundary. Keep argument order and result payloads unchanged.
+
+### 3. Update Tests to Assert the New Boundary
+
+Adjust focused tests such as `tests/test_memory_store_record_service.py` so the
+test double represents `MemoryWriteService` directly instead of a nested
+`_orch` object. Existing behavioral assertions should remain the same.
+
+## Test Plan
+
+Focused tests:
+
+- `uv run --group dev pytest tests/test_memory_store_record_service.py tests/test_memory_write_context_builder.py tests/test_memory_write_derive_service.py tests/test_memory_write_embed_service.py tests/test_memory_write_dedup_service.py tests/test_memory_directory_record_service.py -q`
+- `uv run --group dev pytest tests/test_write_dedup.py tests/test_http_server.py -q`
+
+Static checks:
+
+- `uv run --group dev ruff format --check src/opencortex/services/memory_write_service.py src/opencortex/services/memory_write_context_builder.py src/opencortex/services/memory_write_derive_service.py src/opencortex/services/memory_write_embed_service.py src/opencortex/services/memory_write_dedup_service.py src/opencortex/services/memory_directory_record_service.py src/opencortex/services/memory_store_record_service.py tests/test_memory_store_record_service.py`
+- `uv run --group dev ruff check src/opencortex/services/memory_write_service.py src/opencortex/services/memory_write_context_builder.py src/opencortex/services/memory_write_derive_service.py src/opencortex/services/memory_write_embed_service.py src/opencortex/services/memory_write_dedup_service.py src/opencortex/services/memory_directory_record_service.py src/opencortex/services/memory_store_record_service.py tests/test_memory_store_record_service.py`
+
+## Risks
+
+- Some focused tests may intentionally construct helper services with
+  `SimpleNamespace(_orch=...)`; update those tests only where the helper's
+  boundary changes.
+- Dedup and parent-directory writes share storage and embedding behavior with
+  normal store. Preserve collection names, filter shapes, and best-effort
+  filesystem behavior.
+- This is a dependency-boundary cleanup, not a semantics refactor. Do not
+  change scoring, derive output, TTL values, projection payloads, or signal
+  payloads.

src/opencortex/services/cortex_memory_services.py

@@ -6,9 +6,9 @@
 from typing import TYPE_CHECKING, Callable, TypeVar
 
 if TYPE_CHECKING:
+    from opencortex.cortex_memory import CortexMemory
     from opencortex.lifecycle.background_tasks import BackgroundTaskManager
     from opencortex.lifecycle.bootstrapper import SubsystemBootstrapper
-    from opencortex.cortex_memory import CortexMemory
     from opencortex.services.derivation_service import DerivationService
     from opencortex.services.knowledge_service import KnowledgeService
     from opencortex.services.memory_admin_stats_service import (
@@ -45,11 +45,42 @@ def _cached(self, attr_name: str, factory: Callable[[], T]) -> T:
     def memory_service(self) -> "MemoryService":
         """Lazy-built MemoryService for delegated CRUD/query/scoring methods."""
         from opencortex.services.memory_service import MemoryService
+        from opencortex.services.memory_write_service import MemoryWriteDependencies
 
         return self._cached(
             "_memory_service_instance",
-            lambda: MemoryService(self._orch),
+            lambda: self._build_memory_service(
+                MemoryService,
+                MemoryWriteDependencies,
+            ),
+        )
+
+    def _build_memory_service(
+        self,
+        memory_service_type: type["MemoryService"],
+        dependencies_type: type,
+    ) -> "MemoryService":
+        """Construct MemoryService and bind explicit write-path dependencies."""
+        service = memory_service_type(self._orch)
+        if not hasattr(self._orch, "_config"):
+            return service
+        service.configure_write_dependencies(
+            dependencies_type(
+                config=self._orch._config,
+                storage=self._orch._storage,
+                fs=self._orch._fs,
+                embedder=self._orch._embedder,
+                memory_signal_bus=getattr(self._orch, "_memory_signal_bus", None),
+                entity_index=getattr(self._orch, "_entity_index", None),
+                memory_record_service=self._orch._memory_record_service,
+                derivation_service=self._orch._derivation_service,
+                session_lifecycle_service=self._orch._session_lifecycle_service,
+                ensure_init=self._orch._ensure_init,
+                get_collection=self._orch._get_collection,
+                feedback=service.feedback,
+            )
         )
+        return service
 
     @property
     def derivation_service(self) -> "DerivationService":

src/opencortex/services/memory_directory_record_service.py

@@ -26,10 +26,6 @@ def __init__(self, write_service: "MemoryWriteService") -> None:
         """Bind the directory service to a write service facade."""
         self._write_service = write_service
 
-    @property
-    def _orch(self) -> Any:
-        return self._write_service._orch
-
     async def ensure_parent_records(self, parent_uri: str) -> None:
         """Ensure all ancestor directory records exist in the vector store."""
         to_create = await self._collect_missing_ancestors(parent_uri)
@@ -49,7 +45,6 @@ async def ensure_parent_records(self, parent_uri: str) -> None:
 
     async def _collect_missing_ancestors(self, parent_uri: str) -> List[str]:
         """Walk upward and return directory URIs missing from storage."""
-        orch = self._orch
         uri = parent_uri
         to_create: List[str] = []
 
@@ -59,8 +54,8 @@ async def _collect_missing_ancestors(self, parent_uri: str) -> List[str]:
             except ValueError:
                 break
 
-            existing = await orch._storage.filter(
-                orch._get_collection(),
+            existing = await self._write_service._storage.filter(
+                self._write_service._get_collection(),
                 FilterExpr.eq("uri", uri).to_dict(),
                 limit=1,
             )
@@ -84,10 +79,9 @@ async def _create_directory_record(
         effective_user: UserIdentifier,
     ) -> None:
         """Build and upsert one directory record."""
-        orch = self._orch
         dir_ctx = Context(
             uri=dir_uri,
-            parent_uri=orch._derive_parent_uri(dir_uri),
+            parent_uri=self._write_service._derive_parent_uri(dir_uri),
             is_leaf=False,
             abstract="",
             user=effective_user,
@@ -106,12 +100,14 @@ async def _create_directory_record(
         record["mergeable"] = False
         record["session_id"] = ""
         record["ttl_expires_at"] = ""
-        await orch._storage.upsert(orch._get_collection(), record)
+        await self._write_service._storage.upsert(
+            self._write_service._get_collection(), record
+        )
         logger.debug("[MemoryService] Created directory record: %s", dir_uri)
 
     async def _embed_directory_name(self, *, dir_ctx: Context, uri: str) -> Any:
         """Embed the directory basename and attach its dense vector."""
-        embedder = self._orch._embedder
+        embedder = self._write_service._embedder
         dir_name = uri.rstrip("/").rsplit("/", 1)[-1]
         if not embedder or not dir_name:
             return None

src/opencortex/services/memory_service.py

@@ -55,7 +55,10 @@
     from opencortex.cortex_memory import CortexMemory
     from opencortex.services.memory_query_service import MemoryQueryService
     from opencortex.services.memory_scoring_service import MemoryScoringService
-    from opencortex.services.memory_write_service import MemoryWriteService
+    from opencortex.services.memory_write_service import (
+        MemoryWriteDependencies,
+        MemoryWriteService,
+    )
 
 _BATCH_ADD_CONCURRENCY = 8
 _BATCH_ADD_TASK_CHUNK_SIZE = _BATCH_ADD_CONCURRENCY * 4
@@ -79,6 +82,14 @@ def __init__(self, orchestrator: "CortexMemory") -> None:
                 at call time. Stored as ``self._orch``; not validated.
         """
         self._orch = orchestrator
+        self._write_dependencies: "MemoryWriteDependencies | None" = None
+
+    def configure_write_dependencies(
+        self,
+        dependencies: "MemoryWriteDependencies",
+    ) -> None:
+        """Bind explicit write-path dependencies from the service registry."""
+        self._write_dependencies = dependencies
 
     @property
     def _memory_write_service(self) -> "MemoryWriteService":
@@ -87,7 +98,9 @@ def _memory_write_service(self) -> "MemoryWriteService":
 
         cached = getattr(self, "_memory_write_service_instance", None)
         if cached is None:
-            cached = MemoryWriteService(self)
+            if self._write_dependencies is None:
+                raise RuntimeError("Memory write dependencies are not configured")
+            cached = MemoryWriteService(self, self._write_dependencies)
             self._memory_write_service_instance = cached
         return cached
 

src/opencortex/services/memory_store_record_service.py

@@ -34,10 +34,6 @@ def __init__(self, write_service: "MemoryWriteService") -> None:
         """Bind the persistence service to a write service facade."""
         self._write_service = write_service
 
-    @property
-    def _orch(self) -> Any:
-        return self._write_service._orch
-
     async def persist_context_record(
         self,
         *,
@@ -57,7 +53,6 @@ async def persist_context_record(
         is_leaf: bool,
     ) -> StoredRecordResult:
         """Assemble and persist a normal store record."""
-        orch = self._orch
         record = ctx.to_dict()
         if ctx.vector:
             record["vector"] = ctx.vector
@@ -85,9 +80,11 @@ async def persist_context_record(
         self._populate_flattened_source_fields(record, meta)
 
         upsert_started = asyncio.get_running_loop().time()
-        await orch._storage.upsert(orch._get_collection(), record)
+        await self._write_service._storage.upsert(
+            self._write_service._get_collection(), record
+        )
         upsert_ms = int((asyncio.get_running_loop().time() - upsert_started) * 1000)
-        await orch._sync_anchor_projection_records(
+        await self._write_service._sync_anchor_projection_records(
             source_record=record,
             abstract_json=abstract_json,
         )
@@ -120,15 +117,18 @@ def _ttl_for_record(
         meta: Dict[str, Any],
     ) -> str:
         """Return the TTL string for short-lived record kinds."""
-        orch = self._orch
         if context_type == "staging":
-            return orch._ttl_from_hours(orch._config.immediate_event_ttl_hours)
+            return self._write_service._ttl_from_hours(
+                self._write_service._config.immediate_event_ttl_hours
+            )
         if (
             (context_type or "memory") == "memory"
             and effective_category == "events"
             and meta.get("layer") == "merged"
         ):
-            return orch._ttl_from_hours(orch._config.merged_event_ttl_hours)
+            return self._write_service._ttl_from_hours(
+                self._write_service._config.merged_event_ttl_hours
+            )
         return ""
 
     @staticmethod
@@ -156,7 +156,7 @@ def _publish_memory_stored(
         effective_category: str,
     ) -> None:
         """Publish the post-store lifecycle signal when a bus exists."""
-        signal_bus = getattr(self._orch, "_memory_signal_bus", None)
+        signal_bus = self._write_service._memory_signal_bus
         if signal_bus is None:
             return
         signal_bus.publish_nowait(
@@ -179,9 +179,11 @@ def _sync_entity_index(
         entities: List[str],
     ) -> None:
         """Sync the entity index for entity-bearing records."""
-        entity_index = getattr(self._orch, "_entity_index", None)
+        entity_index = self._write_service._entity_index
         if entity_index and entities:
-            entity_index.add(self._orch._get_collection(), str(record["id"]), entities)
+            entity_index.add(
+                self._write_service._get_collection(), str(record["id"]), entities
+            )
 
     def _schedule_cortexfs_write(
         self,
@@ -207,7 +209,7 @@ def _on_fs_done(task: asyncio.Task[Any]) -> None:
                 )
 
         fs_task = asyncio.create_task(
-            self._orch._fs.write_context(
+            self._write_service._fs.write_context(
                 uri=uri,
                 content=content,
                 abstract=abstract,