diff --git a/backend/src/apis/app_api/memory_spaces/models.py b/backend/src/apis/app_api/memory_spaces/models.py index edf7633d..66cd15a4 100644 --- a/backend/src/apis/app_api/memory_spaces/models.py +++ b/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()] diff --git a/backend/src/apis/app_api/memory_spaces/routes.py b/backend/src/apis/app_api/memory_spaces/routes.py index 14e3555f..108fe661 100644 --- a/backend/src/apis/app_api/memory_spaces/routes.py +++ b/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) diff --git a/backend/src/apis/shared/memory/__init__.py b/backend/src/apis/shared/memory/__init__.py index b510ef51..f84942aa 100644 --- a/backend/src/apis/shared/memory/__init__.py +++ b/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", diff --git a/backend/src/apis/shared/memory/service.py b/backend/src/apis/shared/memory/service.py index fb5e6a9d..a90ea735 100644 --- a/backend/src/apis/shared/memory/service.py +++ b/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( diff --git a/backend/src/apis/shared/memory/store.py b/backend/src/apis/shared/memory/store.py index 41df6fb8..757935fe 100644 --- a/backend/src/apis/shared/memory/store.py +++ b/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).""" diff --git a/backend/tests/apis/app_api/test_memory_spaces_routes.py b/backend/tests/apis/app_api/test_memory_spaces_routes.py index 90ac6ff9..5010d122 100644 --- a/backend/tests/apis/app_api/test_memory_spaces_routes.py +++ b/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) diff --git a/backend/tests/shared/test_memory_spaces.py b/backend/tests/shared/test_memory_spaces.py index 8ebe719d..9613fc39 100644 --- a/backend/tests/shared/test_memory_spaces.py +++ b/backend/tests/shared/test_memory_spaces.py @@ -446,6 +446,87 @@ def test_viewer_can_read_entry(self, space, service): # ==================== service: manifest concurrency (A4) ==================== +class TestConsolidation: + def test_reports_healthy_space(self, space, service): + service.write_entry(space.space_id, OWNER, OWNER_EMAIL, "a", "one") + report = service.consolidate(space.space_id, OWNER, OWNER_EMAIL) + assert report.entry_count == 1 + assert report.over_cap is False + assert report.duplicate_groups == [] + assert report.dead_links == [] + assert report.orphans_deleted == 0 + + def test_gc_deletes_orphaned_objects(self, space, service): + service.write_entry(space.space_id, OWNER, OWNER_EMAIL, "a", "one") + # Simulate a leaked object (crashed write) directly in the store. + orphan_key = service.store.put( + space_id=space.space_id, content=b"leaked", content_type="text/markdown" + ) + assert orphan_key in service.store.list_keys(space.space_id) + + report = service.consolidate(space.space_id, OWNER, OWNER_EMAIL) + assert report.orphans_deleted == 1 + assert orphan_key not in service.store.list_keys(space.space_id) + # the live entry's object is untouched + assert service.read_entry(space.space_id, OWNER, OWNER_EMAIL, "a") == "one" + + def test_gc_can_be_skipped(self, space, service): + service.store.put( + space_id=space.space_id, content=b"leaked", content_type="text/markdown" + ) + report = service.consolidate( + space.space_id, OWNER, OWNER_EMAIL, apply_gc=False + ) + assert report.orphans_deleted == 0 + + def test_reports_duplicate_content_without_merging(self, space, service): + # Two different slugs, byte-identical bodies → same content hash. + service.write_entry(space.space_id, OWNER, OWNER_EMAIL, "a", "same body") + service.write_entry(space.space_id, OWNER, OWNER_EMAIL, "b", "same body") + report = service.consolidate(space.space_id, OWNER, OWNER_EMAIL) + assert report.duplicate_groups == [["a", "b"]] + # both entries still exist — consolidation never auto-merges + assert len(service.list_entries(space.space_id, OWNER, OWNER_EMAIL)) == 2 + + def test_reports_dead_wikilinks(self, space, service): + service.write_entry(space.space_id, OWNER, OWNER_EMAIL, "jane", "hi") + service.update_index( + space.space_id, OWNER, OWNER_EMAIL, "- [[jane]]\n- [[ghost]]\n" + ) + report = service.consolidate(space.space_id, OWNER, OWNER_EMAIL) + assert report.dead_links == ["ghost"] + assert report.stripped_dead_links is False + # index untouched unless stripping is requested + assert "[[ghost]]" in service.read_index(space.space_id, OWNER, OWNER_EMAIL) + + def test_strip_dead_links_unlinks_but_keeps_prose(self, space, service): + service.update_index( + space.space_id, OWNER, OWNER_EMAIL, "See [[ghost]] for details.\n" + ) + report = service.consolidate( + space.space_id, OWNER, OWNER_EMAIL, strip_dead_links=True + ) + assert report.stripped_dead_links is True + text = service.read_index(space.space_id, OWNER, OWNER_EMAIL) + assert "[[ghost]]" not in text + assert "See ghost for details." in text + + def test_over_cap_flag(self, space, service, monkeypatch): + monkeypatch.setenv("MEMORY_SPACE_INDEX_CAP", "2") + for slug in ("a", "b", "c"): + service.write_entry(space.space_id, OWNER, OWNER_EMAIL, slug, slug) + report = service.consolidate(space.space_id, OWNER, OWNER_EMAIL) + assert report.index_cap == 2 + assert report.over_cap is True + # over-cap is reported, never auto-evicted + assert len(service.list_entries(space.space_id, OWNER, OWNER_EMAIL)) == 3 + + def test_requires_editor(self, space, service): + service.share(space.space_id, OWNER, OWNER_EMAIL, STRANGER_EMAIL, "viewer") + with pytest.raises(MemorySpacePermissionError): + service.consolidate(space.space_id, STRANGER, STRANGER_EMAIL) + + class TestManifestConcurrency: def test_version_increments_per_write(self, space, service): service.write_entry(space.space_id, OWNER, OWNER_EMAIL, "a", "1") diff --git a/backend/tests/shared/test_memory_store.py b/backend/tests/shared/test_memory_store.py index edca74ae..d3499642 100644 --- a/backend/tests/shared/test_memory_store.py +++ b/backend/tests/shared/test_memory_store.py @@ -93,6 +93,21 @@ def test_delete_absent_is_noop(self, store): store.delete("spaces/s/missing") +class TestListKeys: + def test_lists_only_the_space_prefix(self, store): + k1 = store.put(space_id="s1", content=b"a", content_type="text/markdown") + k2 = store.put(space_id="s1", content=b"b", content_type="text/markdown") + store.put(space_id="s2", content=b"c", content_type="text/markdown") + assert set(store.list_keys("s1")) == {k1, k2} + + def test_empty_space_returns_empty(self, store): + assert store.list_keys("nope") == [] + + def test_disabled_returns_empty(self, monkeypatch): + monkeypatch.delenv("S3_MEMORY_SPACES_BUCKET_NAME", raising=False) + assert MemorySpaceStore(bucket_name=None).list_keys("s") == [] + + class TestNotConfigured: def test_disabled_when_no_bucket(self, monkeypatch): monkeypatch.delenv("S3_MEMORY_SPACES_BUCKET_NAME", raising=False) diff --git a/docs/specs/user-markdown-memory.md b/docs/specs/user-markdown-memory.md index bf066c46..b0c8eb8e 100644 --- a/docs/specs/user-markdown-memory.md +++ b/docs/specs/user-markdown-memory.md @@ -558,9 +558,16 @@ calls it. save over the A4 endpoints). Signal facade + API service mirror the assistants/schedules pattern; nav entry gated on a live `accessible$` probe (404 = kill switch off → hidden), redesign-tokens throughout. Viewer = read-only. Facade spec green; dev build + tsc clean. -- **A6 — Consolidation.** A maintenance job (scheduled per space, or on index-cap threshold) that - merges duplicate entries, fixes stale ones, and prunes the index. Uses `MemorySpaceService`; runs - on the shipped scheduler. Primitive-side corpus health, not a per-turn concern. +- **A6 — Consolidation.** ✅ (deterministic slice) `MemorySpaceService.consolidate(space_id)` + + `POST /memory/spaces/{id}/consolidate` (editor+) → a `ConsolidationReport`. Auto-fixes only + storage hygiene — orphaned content-addressed objects (no manifest/index ref, from crashed/raced + writes) are GC'd. Everything needing judgment is *reported, not mutated*: duplicate content + across slugs, dead `[[slug]]` wikilinks in MEMORY.md (opt-in `stripDeadLinks` unlinks them, + keeping prose), and over-cap entry counts (`MEMORY_SPACE_INDEX_CAP`, default 200). It never + merges or evicts entries — that's deferred to the **LLM consolidation pass (Workstream B era)**, + which extends this exact `consolidate()` seam once agentic writes create real duplication/staleness + to act on. (Deliberately not auto-run on a schedule/threshold yet — on-demand only; scheduler wiring + and SPA surfacing are follow-ups.) ### Workstream B — Agent / Harness consumption (binds the primitive)