feat(memory): deterministic consolidation health pass (A6)

backend/src/apis/app_api/memory_spaces/models.py

@@ -19,6 +19,7 @@
     ShareRole,
     SpaceMember,
 )
+from apis.shared.memory.service import ConsolidationReport
 from apis.shared.memory.templates import TEMPLATES, SpaceTemplate
 
 
@@ -163,5 +164,38 @@ class MembersListResponse(BaseModel):
     members: List[MemberResponse]
 
 
+class ConsolidateRequest(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    apply_gc: bool = Field(True, alias="applyGc")
+    strip_dead_links: bool = Field(False, alias="stripDeadLinks")
+
+
+class ConsolidationReportResponse(BaseModel):
+    model_config = ConfigDict(populate_by_name=True)
+
+    space_id: str = Field(..., alias="spaceId")
+    entry_count: int = Field(..., alias="entryCount")
+    index_cap: int = Field(..., alias="indexCap")
+    over_cap: bool = Field(..., alias="overCap")
+    orphans_deleted: int = Field(0, alias="orphansDeleted")
+    duplicate_groups: List[List[str]] = Field(default_factory=list, alias="duplicateGroups")
+    dead_links: List[str] = Field(default_factory=list, alias="deadLinks")
+    stripped_dead_links: bool = Field(False, alias="strippedDeadLinks")
+
+    @classmethod
+    def from_report(cls, r: "ConsolidationReport") -> "ConsolidationReportResponse":
+        return cls(
+            space_id=r.space_id,
+            entry_count=r.entry_count,
+            index_cap=r.index_cap,
+            over_cap=r.over_cap,
+            orphans_deleted=r.orphans_deleted,
+            duplicate_groups=r.duplicate_groups,
+            dead_links=r.dead_links,
+            stripped_dead_links=r.stripped_dead_links,
+        )
+
+
 def all_templates() -> List[TemplateResponse]:
     return [TemplateResponse.from_template(t) for t in TEMPLATES.values()]

backend/src/apis/app_api/memory_spaces/routes.py

@@ -43,6 +43,8 @@
 from apis.shared.memory.store import MemorySpaceStoreError
 
 from apis.app_api.memory_spaces.models import (
+    ConsolidateRequest,
+    ConsolidationReportResponse,
     CreateSpaceRequest,
     EntriesListResponse,
     EntryContentResponse,
@@ -265,6 +267,39 @@ def export_space(
     )
 
 
+@router.post("/{space_id}/consolidate", response_model=ConsolidationReportResponse)
+def consolidate_space(
+    space_id: str,
+    request: ConsolidateRequest | None = None,
+    user: User = Depends(require_memory_spaces_user),
+) -> ConsolidationReportResponse:
+    """Run a deterministic consolidation (health) pass on a space (editor+, A6).
+
+    Auto-fixes storage hygiene (orphaned-object GC) and reports issues that
+    need judgment (duplicate content, dead ``[[slug]]`` links, over-cap). Never
+    merges or evicts entries. ``stripDeadLinks`` opts into unlinking dead
+    wikilinks from MEMORY.md.
+    """
+    req = request or ConsolidateRequest()
+    try:
+        report = _svc().consolidate(
+            space_id,
+            user.user_id,
+            user.email,
+            apply_gc=req.apply_gc,
+            strip_dead_links=req.strip_dead_links,
+        )
+    except MemorySpaceError as e:
+        raise _translate(e)
+    except MemorySpaceStoreError as e:
+        logger.error("memory-spaces: consolidate failed for space=%s: %s", space_id, e)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="failed to consolidate memory space storage",
+        )
+    return ConsolidationReportResponse.from_report(report)
+
+
 @router.delete("/{space_id}", status_code=status.HTTP_204_NO_CONTENT)
 def delete_or_leave_space(
     space_id: str, user: User = Depends(require_memory_spaces_user)

backend/src/apis/shared/memory/__init__.py

@@ -22,6 +22,7 @@
 )
 from .repository import MemorySpaceRepository
 from .service import (
+    ConsolidationReport,
     MemorySpaceConcurrencyError,
     MemorySpaceError,
     MemorySpaceExport,
@@ -49,6 +50,7 @@
     "MemorySpaceRepository",
     "MemorySpaceService",
     "MemorySpaceExport",
+    "ConsolidationReport",
     "MemorySpaceConcurrencyError",
     "MemorySpaceError",
     "MemorySpaceNotFoundError",

backend/src/apis/shared/memory/service.py

@@ -17,6 +17,8 @@
 from __future__ import annotations
 
 import logging
+import os
+import re
 import uuid
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -45,9 +47,28 @@
 # exhausts this and surfaces as a conflict.
 _MAX_MANIFEST_RETRIES = 5
 
+# Default soft cap on the number of entries (≈ index lines). Consolidation
+# reports when a space is over it — it never auto-evicts (that's a judgment
+# call for the future LLM pass). ≈ 200 entries ≈ 4k always-loaded tokens/turn.
+_DEFAULT_INDEX_CAP = 200
+
+# Wikilinks in MEMORY.md: [[slug]] pointers into the entry set.
+_WIKILINK_RE = re.compile(r"\[\[([^\[\]]+)\]\]")
+
 _T = TypeVar("_T")
 
 
+def _index_cap() -> int:
+    """Soft entry cap, overridable via ``MEMORY_SPACE_INDEX_CAP``."""
+    raw = os.environ.get("MEMORY_SPACE_INDEX_CAP")
+    if raw:
+        try:
+            return max(1, int(raw))
+        except ValueError:
+            logger.warning("invalid MEMORY_SPACE_INDEX_CAP=%r; using default", raw)
+    return _DEFAULT_INDEX_CAP
+
+
 class MemorySpaceError(RuntimeError):
     """Base class for memory-space service errors (translated by the API layer)."""
 
@@ -85,6 +106,30 @@ class MemorySpaceExport:
     members: List[SpaceMember] = field(default_factory=list)
 
 
+@dataclass
+class ConsolidationReport:
+    """Result of a deterministic consolidation (health) pass over a space (A6).
+
+    The pass auto-fixes only storage hygiene — orphaned content-addressed
+    objects with no manifest/index reference are deleted (``orphans_deleted``).
+    Everything that needs a judgment call is *reported*, not mutated:
+    ``duplicate_groups`` (entries sharing a content hash — which slug survives
+    is semantic), ``dead_links`` (``[[slug]]`` pointers in MEMORY.md with no
+    entry), and ``over_cap`` (entry count past the soft index cap — which entry
+    to drop is semantic). The LLM consolidation pass (Workstream B) extends this
+    seam to act on those reports.
+    """
+
+    space_id: str
+    entry_count: int
+    index_cap: int
+    over_cap: bool
+    orphans_deleted: int = 0
+    duplicate_groups: List[List[str]] = field(default_factory=list)
+    dead_links: List[str] = field(default_factory=list)
+    stripped_dead_links: bool = False
+
+
 def _now_iso() -> str:
     return datetime.now(timezone.utc).isoformat()
 
@@ -301,6 +346,94 @@ def leave_space(
         self.repository.delete_member(space_id, user_email)
         logger.info("memory-spaces: user=%s left space=%s", user_id, space_id)
 
+    # ---- consolidation (A6) --------------------------------------------
+
+    def consolidate(
+        self,
+        space_id: str,
+        user_id: str,
+        user_email: Optional[str] = None,
+        *,
+        apply_gc: bool = True,
+        strip_dead_links: bool = False,
+    ) -> ConsolidationReport:
+        """Deterministic consolidation (health) pass over a space (editor+).
+
+        Auto-fixes storage hygiene — orphaned content-addressed objects (no
+        manifest/index reference) are deleted when ``apply_gc``. Everything that
+        needs judgment is reported, not mutated: duplicate content across slugs,
+        dead ``[[slug]]`` wikilinks in MEMORY.md, and over-cap entry counts. It
+        never merges or evicts entries — that's the LLM pass (Workstream B) that
+        extends this seam. ``strip_dead_links`` opts into one safe edit: unlink
+        dead ``[[slug]]`` pointers (they point nowhere), preserving the prose.
+        """
+        space, _ = self._require(space_id, user_id, user_email, "editor")
+        index = self.repository.get_index(space_id)
+        entries = index.entries
+        slugs = {e.slug for e in entries}
+
+        cap = _index_cap()
+
+        # Duplicate detection: more than one slug sharing a content hash.
+        by_hash: Dict[str, List[str]] = {}
+        for e in entries:
+            by_hash.setdefault(e.content_hash, []).append(e.slug)
+        duplicate_groups = sorted(
+            (sorted(s) for s in by_hash.values() if len(s) > 1),
+            key=lambda g: g[0],
+        )
+
+        # Dead-link detection over the MEMORY.md text.
+        index_text = ""
+        if space.index_s3_key:
+            index_text = self.store.get(space.index_s3_key).decode("utf-8")
+        referenced = {m.strip() for m in _WIKILINK_RE.findall(index_text)}
+        dead_links = sorted(ref for ref in referenced if ref and ref not in slugs)
+
+        stripped = False
+        if strip_dead_links and dead_links and space.index_s3_key:
+            new_text = index_text
+            for ref in dead_links:
+                new_text = new_text.replace(f"[[{ref}]]", ref)
+            if new_text != index_text:
+                self.update_index(space_id, user_id, user_email, new_text)
+                space = self.repository.get_space(space_id) or space
+                stripped = True
+
+        # Orphaned-object GC: keys under the space prefix that no entry or the
+        # index pointer references (leaks from crashed/raced writes). Safe —
+        # unreferenced content is invisible to every read path.
+        orphans_deleted = 0
+        if apply_gc:
+            referenced_keys = {e.s3_key for e in entries}
+            if space.index_s3_key:
+                referenced_keys.add(space.index_s3_key)
+            for key in self.store.list_keys(space_id):
+                if key not in referenced_keys:
+                    self.store.delete(key)
+                    orphans_deleted += 1
+
+        logger.info(
+            "memory-spaces: consolidated space=%s entries=%d orphans=%d "
+            "dups=%d dead_links=%d stripped=%s",
+            space_id,
+            len(entries),
+            orphans_deleted,
+            len(duplicate_groups),
+            len(dead_links),
+            stripped,
+        )
+        return ConsolidationReport(
+            space_id=space_id,
+            entry_count=len(entries),
+            index_cap=cap,
+            over_cap=len(entries) > cap,
+            orphans_deleted=orphans_deleted,
+            duplicate_groups=duplicate_groups,
+            dead_links=dead_links,
+            stripped_dead_links=stripped,
+        )
+
     # ---- sharing -------------------------------------------------------
 
     def share(

backend/src/apis/shared/memory/store.py

@@ -175,6 +175,36 @@ def get(self, s3_key: str) -> bytes:
                 f"failed to read memory file at key '{s3_key}'"
             ) from e
 
+    def list_keys(self, space_id: str) -> list[str]:
+        """List every object key stored under a space's prefix.
+
+        Used by the consolidation pass to find orphaned objects (keys no
+        manifest entry or index pointer references) for garbage collection.
+        Returns an empty list when storage is not configured.
+        """
+        if not self.enabled:
+            return []
+        client = self._client()
+        prefix = f"spaces/{space_id}/"
+        keys: list[str] = []
+        token: Optional[str] = None
+        while True:
+            kwargs = {"Bucket": self.bucket_name, "Prefix": prefix}
+            if token:
+                kwargs["ContinuationToken"] = token
+            try:
+                resp = client.list_objects_v2(**kwargs)
+            except ClientError as e:  # pragma: no cover - network/permission path
+                logger.error("memory-spaces: list failed for space=%s: %s", space_id, e)
+                raise MemorySpaceStoreError(
+                    f"failed to list memory objects for space '{space_id}'"
+                ) from e
+            keys.extend(obj["Key"] for obj in resp.get("Contents", []))
+            if not resp.get("IsTruncated"):
+                break
+            token = resp.get("NextContinuationToken")
+        return keys
+
     def delete(self, s3_key: str) -> None:
         """Delete an object key. Best-effort — never raises on the storage
         miss path (deleting an already-absent object is a no-op in S3)."""

backend/tests/apis/app_api/test_memory_spaces_routes.py

@@ -344,6 +344,42 @@ def test_share_rejects_owner_role(self, service, monkeypatch):
         assert r.status_code == 422
 
 
+class TestConsolidate:
+    def test_owner_consolidate_returns_report(self, service, monkeypatch):
+        client = _client(service, monkeypatch, user=OWNER)
+        sid = client.post("/memory/spaces", json={"name": "X"}).json()["spaceId"]
+        client.put(f"/memory/spaces/{sid}/entries/a", json={"body": "one"})
+        # leak an orphan object to prove GC runs through the route
+        service.store.put(space_id=sid, content=b"leaked", content_type="text/markdown")
+
+        resp = client.post(f"/memory/spaces/{sid}/consolidate", json={})
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["spaceId"] == sid
+        assert body["entryCount"] == 1
+        assert body["orphansDeleted"] == 1
+        assert body["overCap"] is False
+        assert body["duplicateGroups"] == []
+
+    def test_consolidate_no_body(self, service, monkeypatch):
+        client = _client(service, monkeypatch, user=OWNER)
+        sid = client.post("/memory/spaces", json={"name": "X"}).json()["spaceId"]
+        assert client.post(f"/memory/spaces/{sid}/consolidate").status_code == 200
+
+    def test_viewer_cannot_consolidate(self, service, monkeypatch):
+        owner = _client(service, monkeypatch, user=OWNER)
+        sid = owner.post("/memory/spaces", json={"name": "X"}).json()["spaceId"]
+        service.share(sid, OWNER.user_id, OWNER.email, STRANGER.email, "viewer")
+        member = _client(service, monkeypatch, user=STRANGER)
+        assert member.post(f"/memory/spaces/{sid}/consolidate", json={}).status_code == 403
+
+    def test_consolidate_404_when_flag_off(self, service, monkeypatch):
+        client = _client(service, monkeypatch, user=OWNER)
+        sid = client.post("/memory/spaces", json={"name": "X"}).json()["spaceId"]
+        monkeypatch.setenv("MEMORY_SPACES_ENABLED", "false")
+        assert client.post(f"/memory/spaces/{sid}/consolidate", json={}).status_code == 404
+
+
 class TestExport:
     def _seed(self, service, monkeypatch):
         client = _client(service, monkeypatch, user=OWNER)