Extract shared runtime factory for API and MCP

[Atharva-Kanherkar]

Summary

• extract a shared runtime.py factory that builds the event bus, stores, retriever, contradiction detector, forgetting service, and event recorder in one place
• remove the FastAPI service pattern that temporarily mutated config.CHROMA_DB_PATH during initialization
• make Chroma paths and embedding dimensions injectable at runtime, and add focused tests for path isolation and config safety

Why

Issue #58 requires the runtime wiring to move out of api/app.py so FastAPI and the upcoming MCP server can share the same composition path. The old initialization flow mutated module-level config while constructing stores, which was unsafe if multiple runtimes were built concurrently with different Chroma paths.

Validation

• python -m py_compile api/app.py runtime.py demo/cli.py retrieval/retriever.py stores/semantic_store.py stores/episodic_store.py stores/procedural_store.py utils/embeddings.py tests/test_runtime.py
• python -m pytest ... could not be run in this sandbox because pytest is not installed

Issue Links

• Closes MCP-1A: Extract a shared runtime factory and remove global config mutation #58
• Parent: MCP-1: Add an MCP server layer for agentic-memory #55

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
agentic-memory	Ready	Preview, Comment	Mar 29, 2026 7:53pm

[Atharva-Kanherkar]

@claude review

[claude]

Claude finished @Atharva-Kanherkar's task in 1m 50s —— View job

---

PR Review: Extract shared runtime factory for API and MCP

• Gather context
• Read changed files
• Analyze code
• Post review feedback

---

Overall: This is a clean, well-motivated refactor. The core goal — eliminating config.CHROMA_DB_PATH mutation during initialization and centralising composition into a single build_runtime() factory — is achieved correctly. The new tests directly validate the safety properties the PR claims to provide. Below are findings ranging from bugs to minor design notes.

---

Issues

1. Overbroad exception suppression — api/app.py:332

def _safe_contradiction_lookup(...):
    try:
        return active_service.contradiction_detector.find_potential_contradictions(record)
    except (ValueError, Exception):   # Exception already subsumes ValueError
        return []

except (ValueError, Exception) is identical to except Exception. This silently swallows programming errors (e.g. AttributeError, TypeError) that should surface. If the intent is to tolerate retrieval failures gracefully, it should be narrowed to the specific exception(s) find_potential_contradictions is documented to raise.

Fix this →

---

2. Potential IndexError in test assertion — tests/test_runtime.py:32

assert runtime.semantic_store.get_by_id(runtime.semantic_store.get_all_records()[0].id) is not None

If store() on line 29 silently fails or returns before persisting, get_all_records() returns an empty list and [0] raises IndexError rather than a clear AssertionError. Add a length check first:

records = runtime.semantic_store.get_all_records()
assert len(records) == 1
assert runtime.semantic_store.get_by_id(records[0].id) is not None

Fix this →

---

3. Fragile private-attribute introspection — runtime.py:77-82

def _embedder_dimensions(embedder: TextEmbedder) -> int | None:
    for attr in ("_dimensions", "dimensions"):
        value = getattr(embedder, attr, None)
        if isinstance(value, int):
            return value
    return None

Probing _dimensions (a private attribute of GeminiEmbedder) from outside the class is brittle. If the attribute is ever renamed, this silently returns None and the mismatch-guard at runtime.py:97-105 stops working without any error. Since TextEmbedder is a Protocol, the cleanest fix is to add an optional dimensions: int | None property to the protocol and have GeminiEmbedder implement it publicly, or document dimensions as part of the embedder contract.

---

Design Notes

4. MemoryAPIService re-expands MemoryRuntime fields — api/app.py:282-291

runtime = build_runtime(...)
self.media_store = runtime.media_store
self.bus = runtime.bus
# ... 7 more lines

Storing the MemoryRuntime directly (self._runtime = runtime) would mean any new field added to MemoryRuntime doesn't also require an update to MemoryAPIService. The current pattern creates a second place to update whenever MemoryRuntime grows.

---

5. CLI bypasses build_runtime — demo/cli.py

The CLI still constructs stores manually via _make_semantic_store, _make_episodic_store, etc. This is intentional (the CLI uses ConsoleLogger and has no EventRecorder), but it means the two construction paths can diverge. A comment explaining the deliberate omission of build_runtime here would help future contributors avoid confusion.

---

6. Non-atomic lazy service initialisation — api/app.py:398-401

def service() -> MemoryAPIService:
    if app.state.service is None:
        app.state.service = MemoryAPIService(**app.state.service_config)
    return app.state.service

Under concurrent requests two threads can both observe None and construct independent MemoryAPIService instances, resulting in two separate Chroma clients pointing at the same path. This is a pre-existing issue, but it's now more visible. A simple fix is to initialise app.state.service eagerly inside create_app (or use a lock). Worth a follow-up ticket.

---

What's Good

• Eliminating the config.CHROMA_DB_PATH mutation is the right call — the old pattern was fragile under any concurrent or test-parallel execution.
• MemoryRuntime as a @dataclass(slots=True) is a clean, lightweight value type for passing the composed runtime around.
• The dimension-mismatch guard (runtime.py:97-105) prevents a hard-to-debug class of Chroma collection shape errors at construction time rather than at query time.
• Test coverage directly exercises the safety properties claimed in the PR description (path isolation, config immutability, dimension validation).
• EventRecorder being co-located with the factory makes sense — it's an observer wired to the bus, not a FastAPI concern.

[Copilot]

Pull request overview

This PR extracts a shared runtime.py factory to centralize wiring of the event bus, stores, retriever, contradiction detector, forgetting service, and event recorder, so both FastAPI and the upcoming MCP server can share the same composition path while avoiding unsafe global config mutation.

Changes:

• Added runtime.build_runtime() / MemoryRuntime container and migrated api/app.py to consume it.
• Made Chroma DB paths injectable per store/runtime and made embedding dimensions injectable (and enforced in retriever/vector normalization).
• Added focused runtime tests for Chroma path isolation and embedding-dimension validation.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
runtime.py	New shared runtime/container factory and event recording support.
api/app.py	Replaces in-module wiring/config mutation with build_runtime() and adds embedding_dimensions injection.
stores/semantic_store.py	Allows injecting chroma_path into Chroma PersistentClient construction.
stores/episodic_store.py	Allows injecting chroma_path into Chroma PersistentClient construction.
stores/procedural_store.py	Allows injecting chroma_path into Chroma PersistentClient construction.
retrieval/retriever.py	Adds injectable embedding_dimensions for query-vector validation.
utils/embeddings.py	Adds embedder dimensions surface and makes Gemini embedding dimensionality runtime-configurable.
tests/test_runtime.py	New tests verifying runtime isolation (paths/config) and dimension validation behavior.
tests/helpers.py	Updates test embedder to expose dimensions.
demo/cli.py	Small refactor of local wiring helpers/import order (keeps CLI wiring separate from shared runtime).

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

_embedder_dimensions() assumes every passed embedder has a .dimensions attribute. If a caller provides a duck-typed embedder that only implements embed_text/embed_query (common in existing tests/stubs), build_runtime() will raise AttributeError. To keep the runtime factory more resilient, consider using getattr(embedder, "dimensions", None) (or otherwise gracefully treating missing dimensions as unknown).

Suggested change

	return embedder.dimensions
	return getattr(embedder, "dimensions", None)

[Copilot]

chroma_path or config.CHROMA_DB_PATH treats an empty string the same as None, silently falling back to the global default path. That can accidentally defeat per-runtime path isolation if an empty/blank value is passed through. Prefer an explicit None check (and consider rejecting blank strings) so invalid paths fail fast instead of writing to the shared default DB.

Suggested change

	client = chromadb.PersistentClient(path=chroma_path or config.CHROMA_DB_PATH)
	if chroma_path is None:
	effective_chroma_path = config.CHROMA_DB_PATH
	else:
	if isinstance(chroma_path, str) and chroma_path.strip() == "":
	raise ValueError("chroma_path must not be empty or only whitespace")
	effective_chroma_path = chroma_path
	client = chromadb.PersistentClient(path=effective_chroma_path)

[Copilot]

chroma_path or config.CHROMA_DB_PATH treats an empty string the same as None, silently falling back to the global default path. That can accidentally defeat per-runtime path isolation if an empty/blank value is passed through. Prefer an explicit None check (and consider rejecting blank strings) so invalid paths fail fast instead of writing to the shared default DB.

Suggested change

	client = chromadb.PersistentClient(path=chroma_path or config.CHROMA_DB_PATH)
	if chroma_path is None:
	resolved_chroma_path = config.CHROMA_DB_PATH
	else:
	# Reject empty or whitespace-only paths so they don't silently
	# fall back to the global default Chroma DB path.
	if isinstance(chroma_path, str) and not chroma_path.strip():
	raise ValueError("chroma_path must be a non-empty string or None.")
	resolved_chroma_path = chroma_path
	client = chromadb.PersistentClient(path=resolved_chroma_path)

[Copilot]

chroma_path or config.CHROMA_DB_PATH treats an empty string the same as None, silently falling back to the global default path. That can accidentally defeat per-runtime path isolation if an empty/blank value is passed through. Prefer an explicit None check (and consider rejecting blank strings) so invalid paths fail fast instead of writing to the shared default DB.

Suggested change

	client = chromadb.PersistentClient(path=chroma_path or config.CHROMA_DB_PATH)
	if chroma_path is None:
	db_path = config.CHROMA_DB_PATH
	else:
	# Reject empty or whitespace-only paths to avoid silently falling back
	if isinstance(chroma_path, str) and not chroma_path.strip():
	raise ValueError("chroma_path must be a non-empty path when provided")
	db_path = chroma_path
	client = chromadb.PersistentClient(path=db_path)

[Copilot]

_safe_contradiction_lookup() now only catches ValueError. If Chroma (or any other unexpected error) occurs during contradiction lookup, the API will return a 500 after the semantic record has already been stored, which can lead to client retries and duplicate writes. Consider catching broader exceptions here (at least the likely runtime/storage exceptions) and returning an empty candidate list (or a structured "lookup_status") so the write endpoint remains idempotent/safe on transient failures.