feat(memory): add semantic backend contract

coding-agent/src/coding_agent/agent.toml

@@ -48,6 +48,10 @@ enabled = true
 read_enabled = true
 write_enabled = true
 
+[memory.semantic]
+enabled = false
+backend = "fake"
+
 [kb]
 db_path = "kb"
 embedding_model = "text-embedding-3-small"

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

@@ -5,7 +5,7 @@
 import uuid
 from collections.abc import Mapping
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
@@ -39,16 +39,36 @@
 from coding_agent.plugins.summarizer import SummarizerPlugin
 from coding_agent.subagents.coordinator import ChildWorkerCoordinator
 from coding_agent.topics.memory import MemoryReviewStore
+from coding_agent.topics.semantic_backends import (
+    FAKE_SEMANTIC_INDEX_SCHEMA,
+    FakeSemanticMemoryBackend,
+    SemanticIndexSchema,
+)
+from coding_agent.topics.semantic_index import SafeSemanticMemoryIndex
 from coding_agent.tools.web_search import create_web_search_backend
 
 ToolFilter = Any
 
 
+@dataclass(frozen=True)
+class SemanticMemoryConfig:
+    enabled: bool = False
+    backend: str = "fake"
+    schema: SemanticIndexSchema = FAKE_SEMANTIC_INDEX_SCHEMA
+
+    def to_config_dict(self) -> dict[str, object]:
+        return {
+            "enabled": self.enabled,
+            "backend": self.backend,
+        }
+
+
 @dataclass(frozen=True)
 class MemorySwitchConfig:
     enabled: bool = True
     read_enabled: bool = True
     write_enabled: bool = True
+    semantic: SemanticMemoryConfig = field(default_factory=SemanticMemoryConfig)
 
     @property
     def effective_read_enabled(self) -> bool:
@@ -58,13 +78,14 @@ def effective_read_enabled(self) -> bool:
     def effective_write_enabled(self) -> bool:
         return self.enabled and self.write_enabled
 
-    def to_config_dict(self) -> dict[str, bool]:
+    def to_config_dict(self) -> dict[str, object]:
         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,
+            "semantic": self.semantic.to_config_dict(),
         }
 
 
@@ -135,6 +156,7 @@ def _parse_memory_switch_config(extra: Mapping[str, Any]) -> 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"),
+        semantic=_parse_semantic_memory_config(raw_memory_cfg),
     )
 
 
@@ -145,6 +167,28 @@ def _memory_bool(config: Mapping[str, Any], field: str) -> bool:
     return value
 
 
+def _parse_semantic_memory_config(
+    memory_config: Mapping[str, Any],
+) -> SemanticMemoryConfig:
+    raw_semantic_cfg = memory_config.get("semantic", {})
+    if raw_semantic_cfg is None:
+        raw_semantic_cfg = {}
+    if not isinstance(raw_semantic_cfg, Mapping):
+        raise ValueError("[memory.semantic] config must be a table")
+    enabled = raw_semantic_cfg.get("enabled", False)
+    if type(enabled) is not bool:
+        raise ValueError("[memory.semantic].enabled must be a boolean")
+    backend = raw_semantic_cfg.get("backend", "fake")
+    if not isinstance(backend, str):
+        raise ValueError("[memory.semantic].backend must be a string")
+    if enabled and backend != "fake":
+        raise ValueError(f"unknown semantic memory backend: {backend}")
+    return SemanticMemoryConfig(
+        enabled=enabled,
+        backend=backend,
+    )
+
+
 def _child_system_prompt_suffix(tool_filter: ToolFilter) -> str:
     if tool_filter is None:
         return ""
@@ -340,6 +384,13 @@ def create_child_pipeline(
         data_dir / kb_db_path / "reviewed_memory.jsonl",
         candidate_writes_enabled=memory_cfg.effective_write_enabled,
     )
+    semantic_memory_backend = None
+    semantic_memory_index = None
+    if memory_cfg.semantic.enabled:
+        semantic_memory_backend = FakeSemanticMemoryBackend(
+            schema=memory_cfg.semantic.schema
+        )
+        semantic_memory_index = SafeSemanticMemoryIndex(semantic_memory_backend)
     observability_cfg = cfg.extra.get("observability", {})
     if not isinstance(observability_cfg, dict):
         raise ValueError("[observability] config must be a table")
@@ -533,6 +584,10 @@ async def _memory_handler(directive: Any) -> None:
         ctx_config["observation_sink"] = observation_sink
     if local_workspace_root is not None:
         ctx_config["workspace_root"] = str(local_workspace_root)
+    if semantic_memory_backend is not None:
+        ctx_config["semantic_memory_backend"] = semantic_memory_backend
+    if semantic_memory_index is not None:
+        ctx_config["semantic_memory_index"] = semantic_memory_index
 
     ctx = PipelineContext(
         tape=tape_fork,

coding-agent/src/coding_agent/topics/semantic_backends/__init__.py

@@ -0,0 +1,20 @@
+"""Semantic memory backend contracts and local test implementations."""
+
+from __future__ import annotations
+
+from .base import (
+    SemanticBackendScope,
+    SemanticIndexSchema,
+    SemanticMemoryBackend,
+    SemanticSchemaMismatch,
+)
+from .fake import FAKE_SEMANTIC_INDEX_SCHEMA, FakeSemanticMemoryBackend
+
+__all__ = [
+    "FAKE_SEMANTIC_INDEX_SCHEMA",
+    "FakeSemanticMemoryBackend",
+    "SemanticBackendScope",
+    "SemanticIndexSchema",
+    "SemanticMemoryBackend",
+    "SemanticSchemaMismatch",
+]

coding-agent/src/coding_agent/topics/semantic_backends/base.py

@@ -0,0 +1,227 @@
+"""Coding Agent semantic memory backend contracts."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+import re
+from typing import Protocol, runtime_checkable
+
+from agentkit.storage.protocols import MemoryHit, MemoryIndex
+from coding_agent.topics.semantic_index import (
+    SemanticDocId,
+    SemanticDocKind,
+    SemanticSourceKind,
+    SemanticSourceRef,
+)
+
+_SEMANTIC_SCOPE_PART_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
+
+
+@dataclass(frozen=True, slots=True)
+class SemanticIndexSchema:
+    schema_version: int
+    embedding_provider_id: str
+    embedding_model: str
+    embedding_dim: int
+    backend_adapter_id: str
+    backend_schema_version: int
+    distance_metric: str
+    score_normalization: str
+
+    def __post_init__(self) -> None:
+        _require_positive_int("semantic schema version", self.schema_version)
+        _require_non_empty_token(
+            "semantic embedding provider id", self.embedding_provider_id
+        )
+        _require_non_empty_token("semantic embedding model", self.embedding_model)
+        _require_positive_int("semantic embedding dimension", self.embedding_dim)
+        _require_non_empty_token(
+            "semantic backend adapter id", self.backend_adapter_id
+        )
+        _require_positive_int(
+            "semantic backend schema version", self.backend_schema_version
+        )
+        _require_non_empty_token("semantic distance metric", self.distance_metric)
+        _require_non_empty_token(
+            "semantic score normalization", self.score_normalization
+        )
+
+
+class SemanticSchemaMismatch(ValueError):
+    """Raised when a backend already contains a different semantic schema."""
+
+    def __init__(
+        self,
+        *,
+        existing: SemanticIndexSchema,
+        requested: SemanticIndexSchema,
+    ) -> None:
+        super().__init__(
+            "semantic memory schema mismatch: "
+            f"existing={existing!r} requested={requested!r}"
+        )
+        self.existing = existing
+        self.requested = requested
+
+
+@dataclass(frozen=True, slots=True)
+class SemanticBackendScope:
+    source_kind: SemanticSourceKind | str
+    source_id: str
+    session_id: str | None = None
+    profile: str | None = None
+
+    def __post_init__(self) -> None:
+        source_kind = _coerce_source_kind(self.source_kind)
+        _require_safe_scope_part("semantic backend scope source id", self.source_id)
+        if self.session_id is not None:
+            _require_safe_scope_part(
+                "semantic backend scope session id", self.session_id
+            )
+        if self.profile is not None:
+            _require_safe_scope_part("semantic backend scope profile", self.profile)
+        object.__setattr__(self, "source_kind", source_kind)
+
+    @classmethod
+    def for_source_ref(
+        cls,
+        source_ref: SemanticSourceRef | str,
+        *,
+        session_id: str | None = None,
+        profile: str | None = None,
+    ) -> SemanticBackendScope:
+        ref = SemanticSourceRef.parse(source_ref)
+        return cls(
+            source_kind=ref.kind,
+            source_id=ref.source_id,
+            session_id=session_id,
+            profile=profile,
+        )
+
+    @property
+    def source_ref(self) -> str:
+        if not isinstance(self.source_kind, SemanticSourceKind):
+            raise TypeError("semantic backend scope source kind was not normalized")
+        return str(
+            SemanticSourceRef(kind=self.source_kind, source_id=self.source_id)
+        )
+
+    def matches(
+        self,
+        *,
+        memory_id: str,
+        metadata: dict[str, object],
+        source_refs: tuple[str, ...],
+    ) -> bool:
+        refs = source_refs or (_primary_source_ref_for_memory_id(memory_id),)
+        if self.source_ref not in refs:
+            return False
+        if self.session_id is not None and metadata.get("session_id") != self.session_id:
+            return False
+        if self.profile is not None:
+            metadata_profile = metadata.get("profile", metadata.get("profile_id"))
+            if metadata_profile != self.profile:
+                return False
+        return True
+
+
+@runtime_checkable
+class SemanticMemoryBackend(MemoryIndex, Protocol):
+    """Coding Agent operational extension for semantic memory backends."""
+
+    async def ensure_schema(
+        self,
+        schema: SemanticIndexSchema,
+        *,
+        allow_rebuild: bool = False,
+    ) -> None: ...
+
+    async def list_ids(
+        self,
+        *,
+        scope: SemanticBackendScope | None = None,
+    ) -> list[str]: ...
+
+    async def delete_scope(self, scope: SemanticBackendScope) -> int: ...
+
+
+def normalize_source_refs(
+    memory_id: str,
+    metadata: dict[str, object],
+) -> tuple[str, ...]:
+    raw_refs = metadata.get("source_refs", ())
+    if raw_refs is None:
+        return (_primary_source_ref_for_memory_id(memory_id),)
+    if isinstance(raw_refs, str):
+        refs = (raw_refs,)
+    elif isinstance(raw_refs, list | tuple):
+        refs = tuple(raw_refs)
+    else:
+        raise TypeError("semantic memory metadata source_refs must be a string list")
+    normalized = tuple(
+        sorted({str(SemanticSourceRef.parse(ref)) for ref in refs})
+    )
+    if not normalized:
+        return (_primary_source_ref_for_memory_id(memory_id),)
+    return normalized
+
+
+def memory_hit_from_document(
+    *,
+    memory_id: str,
+    text: str,
+    score: float,
+    metadata: dict[str, object],
+) -> MemoryHit:
+    return MemoryHit(
+        memory_id=memory_id,
+        text=text,
+        score=score,
+        metadata=dict(metadata),
+        source_refs=normalize_source_refs(memory_id, metadata),
+    )
+
+
+def _primary_source_ref_for_memory_id(memory_id: str) -> str:
+    document_id = SemanticDocId.parse(memory_id)
+    if document_id.kind is SemanticDocKind.TOPIC_SUMMARY:
+        return str(
+            SemanticSourceRef(
+                kind=SemanticSourceKind.TOPIC,
+                source_id=document_id.source_id,
+            )
+        )
+    if document_id.kind is SemanticDocKind.ACCEPTED_REVIEWED_MEMORY:
+        return str(
+            SemanticSourceRef(
+                kind=SemanticSourceKind.ACCEPTED_MEMORY,
+                source_id=document_id.source_id,
+            )
+        )
+    raise ValueError("semantic document kind is not supported")
+
+
+def _coerce_source_kind(value: SemanticSourceKind | str) -> SemanticSourceKind:
+    try:
+        return SemanticSourceKind(value)
+    except ValueError as exc:
+        raise ValueError(
+            f"semantic backend scope source kind is not supported: {value}"
+        ) from exc
+
+
+def _require_safe_scope_part(field_name: str, value: str) -> None:
+    if _SEMANTIC_SCOPE_PART_RE.fullmatch(value) is None:
+        raise ValueError(
+            f"{field_name} must contain only letters, numbers, dot, underscore, or dash"
+        )
+
+
+def _require_non_empty_token(field_name: str, value: str) -> None:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"{field_name} must be a non-empty string")
+
+
+def _require_positive_int(field_name: str, value: int) -> None:
+    if type(value) is not int or value <= 0:
+        raise ValueError(f"{field_name} must be a positive integer")