diff --git a/README.md b/README.md
index 03876184..157f01c6 100644
--- a/README.md
+++ b/README.md
@@ -108,6 +108,14 @@ npx @modelcontextprotocol/inspector@latest http://127.0.0.1:8201/sse --cli --met
 **Available tools:**
 - `get_entities(task: str, entity_type: str = "guideline", include_public: bool = False)`: Get relevant entities for a specific task. Set `include_public=True` to merge in public entities from all other namespaces; those results are annotated with `[public: {owner_id}]`.
 - `get_guidelines(task: str)`: Get relevant guidelines for a specific task (backward compatibility alias for `get_entities`).
+- `get_relevant_guidelines(task: str, top_k: int | None, core_support: int | None)`: Retrieve the always-on guideline core plus a task-relevant dosage.
+- `list_entities(...)`: Return structured, filtered, cursor-paginated entity inventory for user and administrative UIs.
+- `get_entity(entity_id: str, user_id: str | None, record_access: bool = True)`: Return one structured entity, enforcing ownership when a caller ID is supplied.
+- `patch_entity_metadata(entity_id: str, metadata_patch: str, user_id: str | None)`: Merge JSON metadata through the memory hook seam.
+- `record_access(entity_ids: list[str], accessed_at: str | None)`: Explicitly stamp the retention engine's `last_accessed` signal.
+- `validate_retention_policy(policy: str)`: Validate and normalize a JSON retention policy without scanning data.
+- `run_retention(policy: str, dry_run: bool = True, as_of: str | None, scan_limit: int | None)`: Dry-run or apply retention and return a structured, entity-linked report.
+- `get_compliance_status()`: Report backend health, retention availability, hook coverage, and configured plugin health.
 - `save_trajectory(trajectory_data: str, task_id: str | None, owner_id: str | None)`: Save a conversation trajectory and generate new guidelines.
 - `create_entity(content: str, entity_type: str, metadata: str | None, enable_conflict_resolution: bool, owner_id: str | None, visibility: str = "private")`: Create a single entity. Pass `visibility="public"` and `owner_id` to make it immediately discoverable by other namespaces.
 - `publish_entity(entity_id: str, user_id: str | None)`: Make an entity publicly visible to all namespaces. Records the caller as owner and stamps `published_at`.
diff --git a/altk_evolve/backend/base.py b/altk_evolve/backend/base.py
index b62bfe81..1f875d91 100644
--- a/altk_evolve/backend/base.py
+++ b/altk_evolve/backend/base.py
@@ -78,6 +78,33 @@ def search_entities(
         results = self._search_entities_impl(namespace_id, query, filters, limit)
         return dispatch_memory_post_read(self, namespace_id, results, query=query, filters=filters)
 
+    def scan_entities(self, namespace_id: str, filters: dict | None = None, limit: int = 100) -> list[RecordedEntity]:
+        """Read entities for administrative work without firing read hooks."""
+        return self._search_entities_impl(namespace_id, query=None, filters=filters, limit=limit)
+
+    def set_entity_created_at(self, namespace_id: str, entity_id: str, created_at: datetime.datetime) -> RecordedEntity:
+        """Set an imported entity's persisted creation time without a public-read hook.
+
+        This is intentionally an administrative import operation. It uses the
+        backend's normal patch/persist path without firing public-read hooks.
+        """
+        entities = self.scan_entities(namespace_id, filters={"id": entity_id}, limit=1)
+        if not entities:
+            raise EvolveException(f"Entity '{entity_id}' not found in namespace '{namespace_id}'")
+        entity = entities[0]
+        self._patch_entity(
+            namespace_id,
+            entity_id,
+            entity.type,
+            serialize_content(entity.content),
+            int(created_at.timestamp()),
+            dict(entity.metadata or {}),
+        )
+        updated = self.scan_entities(namespace_id, filters={"id": entity_id}, limit=1)
+        if not updated:
+            raise EvolveException(f"Entity '{entity_id}' disappeared after timestamp update")
+        return updated[0]
+
     @abstractmethod
     def _search_entities_impl(
         self, namespace_id: str, query: str | None = None, filters: dict | None = None, limit: int = 10
diff --git a/altk_evolve/frontend/client/evolve_client.py b/altk_evolve/frontend/client/evolve_client.py
index ee9dfa5e..9a995b95 100644
--- a/altk_evolve/frontend/client/evolve_client.py
+++ b/altk_evolve/frontend/client/evolve_client.py
@@ -114,6 +114,21 @@ def get_all_entities(self, namespace_id: str, filters: dict | None = None, limit
         """Get all entities from a namespace."""
         return self.search_entities(namespace_id, query=None, filters=filters, limit=limit)
 
+    def scan_entities(self, namespace_id: str, filters: dict | None = None, limit: int = 100) -> list[RecordedEntity]:
+        """Read entities for an administrative scan without recording access.
+
+        Retention and inventory scans are not user recalls and must not refresh
+        ``metadata.last_accessed`` through ``memory_post_read``. This method
+        intentionally uses the backend's administrative scan seam, so callers must
+        not use it for ordinary memory retrieval where read transforms and
+        access auditing are expected.
+        """
+        return self.backend.scan_entities(namespace_id, filters=filters, limit=limit)
+
+    def set_entity_created_at(self, namespace_id: str, entity_id: str, created_at: datetime.datetime) -> RecordedEntity:
+        """Administratively stamp an imported entity and persist the result."""
+        return self.backend.set_entity_created_at(namespace_id, entity_id, created_at)
+
     def delete_entity_by_id(self, namespace_id: str, entity_id: str) -> None:
         """Delete a specific entity by its ID."""
         self.backend.delete_entity_by_id(namespace_id, entity_id)
@@ -127,7 +142,7 @@ def patch_entity_metadata(self, namespace_id: str, entity_id: str, metadata_upda
         """Merge metadata_updates into an entity without touching content or ID."""
         return self.backend.update_entity_metadata(namespace_id, entity_id, metadata_updates)
 
-    def record_access(self, namespace_id: str, entity_ids: list[str], when: datetime.datetime | None = None) -> None:
+    def record_access(self, namespace_id: str, entity_ids: list[str], when: datetime.datetime | None = None) -> list[str]:
         """Stamp ``metadata.last_accessed`` (ISO-8601 UTC) on the given entities.
 
         This is the *explicit* half of the access-stamping story and the signal
@@ -145,17 +160,22 @@ def record_access(self, namespace_id: str, entity_ids: list[str], when: datetime
 
         Failures on individual ids are logged and skipped rather than raised —
         recording an access must never break the flow it rode in on.
+
+        Returns the IDs that were stamped successfully.
         """
         # Imported here (not at module scope) to keep the client import cheap;
         # build_access_stamps is a pure function with no cpex dependency.
         from altk_evolve.hooks.plugins.access_stamp import build_access_stamps
 
         moment = when if when is not None else datetime.datetime.now(datetime.UTC)
+        updated_ids: list[str] = []
         for entity_id, patch in build_access_stamps([{"id": eid} for eid in entity_ids], now=lambda: moment):
             try:
                 self.patch_entity_metadata(namespace_id, entity_id, patch)
+                updated_ids.append(entity_id)
             except Exception:
                 logger.warning("record_access: failed to stamp entity %s", entity_id, exc_info=True)
+        return updated_ids
 
     def get_public_entities(
         self,
@@ -254,7 +274,7 @@ def consolidate_guidelines(self, namespace_id: str, threshold: float | None = No
         # llm_pre_call; the write-back still fires memory_pre_write and the
         # deletes still fire memory_pre_delete.
         limit = 10000
-        entities = self.backend._search_entities_impl(namespace_id, query=None, filters={"type": "guideline"}, limit=limit)
+        entities = self.backend.scan_entities(namespace_id, filters={"type": "guideline"}, limit=limit)
         if len(entities) >= limit:
             logger.warning(
                 "Fetched %d entities (hit limit=%d); consolidation may be incomplete. Consider increasing the limit.",
diff --git a/altk_evolve/frontend/mcp/mcp_server.py b/altk_evolve/frontend/mcp/mcp_server.py
index 73728e7b..13f7cd49 100644
--- a/altk_evolve/frontend/mcp/mcp_server.py
+++ b/altk_evolve/frontend/mcp/mcp_server.py
@@ -4,13 +4,19 @@
 This server provides a tool to get task-relevant guidelines.
 """
 
+import base64
+import datetime
 import json
 import logging
 import threading
 import uuid
 import os
+from dataclasses import asdict
+from importlib.metadata import PackageNotFoundError, version
+from pathlib import Path
 from typing import Any
 
+import yaml
 from fastmcp import FastMCP
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
@@ -237,6 +243,82 @@ def _parse_metadata(metadata: str | None) -> dict[str, Any]:
     return parsed
 
 
+def _json_response(payload: Any) -> str:
+    """Serialize MCP responses consistently, including datetimes and errors."""
+    return json.dumps(payload, default=str)
+
+
+def _entity_payload(entity: RecordedEntity, *, include_content: bool = True) -> dict[str, Any]:
+    content = entity.content
+    preview_source = content if isinstance(content, str) else _json_response(content)
+    payload: dict[str, Any] = {
+        "id": entity.id,
+        "type": entity.type,
+        "content_preview": preview_source[:240],
+        "created_at": entity.created_at.isoformat(),
+        "metadata": entity.metadata or {},
+    }
+    if include_content:
+        payload["content"] = content
+    return payload
+
+
+def _encode_cursor(offset: int) -> str:
+    return base64.urlsafe_b64encode(str(offset).encode()).decode().rstrip("=")
+
+
+def _decode_cursor(cursor: str | None) -> int:
+    if not cursor:
+        return 0
+    try:
+        padded = cursor + "=" * (-len(cursor) % 4)
+        offset = int(base64.urlsafe_b64decode(padded).decode())
+    except (ValueError, UnicodeDecodeError) as exc:
+        raise ValueError("cursor is invalid") from exc
+    if offset < 0:
+        raise ValueError("cursor is invalid")
+    return offset
+
+
+def _parse_datetime(value: str | None, *, field_name: str) -> datetime.datetime | None:
+    if value is None:
+        return None
+    try:
+        parsed = datetime.datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError(f"{field_name} must be an ISO-8601 datetime") from exc
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=datetime.UTC)
+    return parsed
+
+
+def _entity_owned_by(
+    entity: RecordedEntity,
+    user_id: str | None,
+    agent_id: str | None = None,
+) -> bool:
+    metadata = entity.metadata or {}
+    if user_id is not None:
+        attributed_ids = {str(value) for value in (metadata.get("owner_id"), metadata.get("user_id")) if value}
+        if user_id not in attributed_ids:
+            return False
+    if agent_id is not None and str(metadata.get("agent_id") or "") != agent_id:
+        return False
+    return True
+
+
+def _record_entity_access(client: EvolveClient, namespace_id: str, entities: list[RecordedEntity]) -> list[RecordedEntity]:
+    if not entities:
+        return entities
+    moment = datetime.datetime.now(datetime.UTC)
+    updated_ids = set(client.record_access(namespace_id, [entity.id for entity in entities], when=moment))
+    stamp = moment.isoformat()
+    return [
+        entity.model_copy(update={"metadata": {**(entity.metadata or {}), "last_accessed": stamp}}) if entity.id in updated_ids else entity
+        for entity in entities
+    ]
+
+
 def _persist_entities(
     namespace_id: str | None,
     entities: list[Entity],
@@ -373,6 +455,385 @@ def get_relevant_guidelines(
     return format_selection(selection)
 
 
+@mcp.tool()
+def list_entities(
+    entity_types: list[str] | None = None,
+    user_id: str | None = None,
+    agent_id: str | None = None,
+    session_id: str | None = None,
+    metadata_filters: str | None = None,
+    cursor: str | None = None,
+    limit: int = 50,
+    include_content: bool = False,
+    record_access: bool = False,
+    namespace_id: str | None = None,
+) -> str:
+    """Return a structured, paginated entity inventory.
+
+    This is the UI/admin counterpart to ``get_entities``, whose prose response
+    is intentionally optimized for prompt injection. Administrative scans do
+    not count as memory use by default; set ``record_access=True`` for a
+    user-facing read that should refresh ``metadata.last_accessed``.
+    """
+    try:
+        filters = _parse_metadata(metadata_filters)
+        offset = _decode_cursor(cursor)
+    except ValueError as exc:
+        return _json_response({"error": str(exc)})
+
+    resolved_ns = _resolve_namespace(namespace_id)
+    page_size = max(1, min(limit, 200))
+    scan_limit = 100_000
+    client = get_client()
+    candidates = client.scan_entities(resolved_ns, limit=scan_limit)
+
+    wanted_types = set(entity_types or [])
+
+    def matches(entity: RecordedEntity) -> bool:
+        metadata = entity.metadata or {}
+        if wanted_types and entity.type not in wanted_types:
+            return False
+        if user_id and user_id not in {metadata.get("user_id"), metadata.get("owner_id")}:
+            return False
+        if agent_id and metadata.get("agent_id") != agent_id:
+            return False
+        if session_id and session_id not in {metadata.get("session_id"), metadata.get("thread_id")}:
+            return False
+        return all(metadata.get(key) == value for key, value in filters.items())
+
+    matched = [entity for entity in candidates if matches(entity)]
+    matched.sort(key=lambda entity: (entity.created_at, entity.id), reverse=True)
+    page = matched[offset : offset + page_size]
+    consumed = len(page)
+    if record_access and page:
+        transformed_page = []
+        for entity in page:
+            transformed = client.get_entity_by_id(resolved_ns, entity.id)
+            if transformed is not None:
+                transformed_page.append(transformed)
+        page = _record_entity_access(client, resolved_ns, transformed_page)
+    next_offset = offset + consumed
+    facets: dict[str, int] = {}
+    for entity in matched:
+        facets[entity.type] = facets.get(entity.type, 0) + 1
+
+    return _json_response(
+        {
+            "items": [_entity_payload(entity, include_content=include_content) for entity in page],
+            "next_cursor": _encode_cursor(next_offset) if next_offset < len(matched) else None,
+            "total": len(matched),
+            "facets": {"entity_types": facets},
+            "truncated": len(candidates) >= scan_limit,
+            "namespace_id": resolved_ns,
+        }
+    )
+
+
+@mcp.tool()
+def get_entity(
+    entity_id: str,
+    user_id: str | None = None,
+    agent_id: str | None = None,
+    record_access: bool = True,
+    namespace_id: str | None = None,
+) -> str:
+    """Return one structured entity, optionally recording a user-facing read."""
+    resolved_ns = _resolve_namespace(namespace_id)
+    client = get_client()
+    matches = client.scan_entities(resolved_ns, filters={"id": entity_id}, limit=1)
+    entity = matches[0] if matches else None
+    if entity is None:
+        return _json_response({"error": f"Entity {entity_id} not found"})
+    if not _entity_owned_by(entity, user_id, agent_id):
+        return _json_response({"error": "Permission denied: caller is not the owner of this entity"})
+    if record_access:
+        refreshed = client.get_entity_by_id(resolved_ns, entity_id)
+        if refreshed is not None:
+            entity = refreshed
+        entity = _record_entity_access(client, resolved_ns, [entity])[0]
+    return _json_response(_entity_payload(entity))
+
+
+@mcp.tool()
+def patch_entity_metadata(
+    entity_id: str,
+    metadata_patch: str,
+    user_id: str | None = None,
+    agent_id: str | None = None,
+    namespace_id: str | None = None,
+) -> str:
+    """Merge metadata into an owned entity through the memory hook seam."""
+    try:
+        patch = _parse_metadata(metadata_patch)
+    except ValueError as exc:
+        return _json_response({"error": str(exc)})
+
+    resolved_ns = _resolve_namespace(namespace_id)
+    client = get_client()
+    matches = client.scan_entities(resolved_ns, filters={"id": entity_id}, limit=1)
+    entity = matches[0] if matches else None
+    if entity is None:
+        return _json_response({"error": f"Entity {entity_id} not found"})
+    if not _entity_owned_by(entity, user_id, agent_id):
+        return _json_response({"error": "Permission denied: caller is not the owner of this entity"})
+    try:
+        updated = client.patch_entity_metadata(resolved_ns, entity_id, patch)
+        return _json_response(_entity_payload(updated))
+    except EvolveException as exc:
+        return _json_response({"error": str(exc)})
+
+
+@mcp.tool()
+def record_access(
+    entity_ids: list[str],
+    accessed_at: str | None = None,
+    user_id: str | None = None,
+    agent_id: str | None = None,
+    namespace_id: str | None = None,
+) -> str:
+    """Explicitly stamp memories as used without requiring a retrieval query."""
+    try:
+        moment = _parse_datetime(accessed_at, field_name="accessed_at")
+    except ValueError as exc:
+        return _json_response({"error": str(exc)})
+
+    resolved_ns = _resolve_namespace(namespace_id)
+    client = get_client()
+    allowed: list[str] = []
+    denied: list[str] = []
+    missing: list[str] = []
+    for entity_id in dict.fromkeys(entity_ids):
+        matches = client.scan_entities(resolved_ns, filters={"id": entity_id}, limit=1)
+        entity = matches[0] if matches else None
+        if entity is None:
+            missing.append(entity_id)
+        elif not _entity_owned_by(entity, user_id, agent_id):
+            denied.append(entity_id)
+        else:
+            allowed.append(entity_id)
+
+    moment = moment or datetime.datetime.now(datetime.UTC)
+    updated = client.record_access(resolved_ns, allowed, when=moment) if allowed else []
+    return _json_response(
+        {
+            "updated_ids": updated,
+            "denied_ids": denied,
+            "missing_ids": missing,
+            "accessed_at": moment.isoformat(),
+            "namespace_id": resolved_ns,
+        }
+    )
+
+
+def _retention_item_payload(item: Any, entity: RecordedEntity | None, *, dry_run: bool) -> dict[str, Any]:
+    payload = asdict(item)
+    applied_outcomes = {"flag": "flagged", "delete": "deleted", "skip": "skipped"}
+    payload["outcome"] = f"would_{item.action}" if dry_run else applied_outcomes.get(item.action, item.action)
+    if entity is not None:
+        snapshot = _entity_payload(entity, include_content=False)
+        metadata = entity.metadata or {}
+        payload.update(
+            {
+                "created_at": snapshot["created_at"],
+                "content_preview": snapshot["content_preview"],
+                "metadata": metadata,
+                "user_id": metadata.get("user_id") or metadata.get("owner_id"),
+                "agent_id": metadata.get("agent_id"),
+                "session_id": metadata.get("session_id") or metadata.get("thread_id"),
+                "source_task_id": metadata.get("source_task_id"),
+            }
+        )
+    return payload
+
+
+@mcp.tool()
+def validate_retention_policy(policy: str) -> str:
+    """Validate and normalize a JSON retention policy without scanning data."""
+    from pydantic import ValidationError
+
+    from altk_evolve.retention import RetentionPolicy
+
+    try:
+        parsed = _parse_metadata(policy)
+        normalized = RetentionPolicy.from_mapping(parsed)
+    except (ValueError, ValidationError) as exc:
+        errors: list[dict[str, Any]] = (
+            [dict(error) for error in exc.errors()] if isinstance(exc, ValidationError) else [{"message": str(exc)}]
+        )
+        return _json_response({"valid": False, "normalized_policy": None, "errors": errors, "warnings": []})
+    return _json_response(
+        {
+            "valid": True,
+            "normalized_policy": normalized.model_dump(mode="json"),
+            "errors": [],
+            "warnings": [],
+        }
+    )
+
+
+@mcp.tool()
+def run_retention(
+    policy: str,
+    dry_run: bool = True,
+    as_of: str | None = None,
+    scan_limit: int | None = None,
+    run_id: str | None = None,
+    namespace_id: str | None = None,
+    metadata_filters: str | None = None,
+) -> str:
+    """Evaluate or apply a retention policy and return a structured report.
+
+    ``as_of`` exists for deterministic audits and demonstrations. Applied
+    deletes still flow through ``memory_pre_delete``, so legal-hold plugins can
+    veto individual entities without aborting the run. ``metadata_filters`` is
+    a JSON object of metadata key/value pairs used to scope shared namespaces.
+    """
+    from pydantic import ValidationError
+
+    from altk_evolve.retention import RetentionEngine, RetentionPolicy
+
+    try:
+        parsed_policy = _parse_metadata(policy)
+        normalized_policy = RetentionPolicy.from_mapping(parsed_policy)
+        now = _parse_datetime(as_of, field_name="as_of")
+        metadata_filter_values = _parse_metadata(metadata_filters) if metadata_filters else None
+        backend_filters = {f"metadata.{key}": value for key, value in metadata_filter_values.items()} if metadata_filter_values else None
+    except (ValueError, ValidationError) as exc:
+        errors: list[dict[str, Any]] = (
+            [dict(error) for error in exc.errors()] if isinstance(exc, ValidationError) else [{"message": str(exc)}]
+        )
+        return _json_response({"error": "Invalid retention request", "details": errors})
+
+    if scan_limit is not None and scan_limit <= 0:
+        return _json_response({"error": "scan_limit must be greater than zero"})
+
+    resolved_ns = _resolve_namespace(namespace_id)
+    client = get_client()
+    started_at = datetime.datetime.now(datetime.UTC)
+    engine = RetentionEngine(client)
+    report = engine.apply(
+        resolved_ns,
+        normalized_policy,
+        now=now,
+        dry_run=dry_run,
+        scan_limit=scan_limit,
+        filters=backend_filters,
+    )
+    snapshots = {entity.id: entity for entity in engine.last_scanned_entities}
+    completed_at = datetime.datetime.now(datetime.UTC)
+    return _json_response(
+        {
+            "run_id": run_id or str(uuid.uuid4()),
+            "namespace_id": resolved_ns,
+            "started_at": started_at.isoformat(),
+            "completed_at": completed_at.isoformat(),
+            "as_of": (now or completed_at).isoformat(),
+            "dry_run": report.dry_run,
+            "policy": normalized_policy.model_dump(mode="json"),
+            "metadata_filters": metadata_filter_values,
+            "summary": report.summary(),
+            "flagged": [_retention_item_payload(item, snapshots.get(item.entity_id), dry_run=dry_run) for item in report.flagged],
+            "deleted": [_retention_item_payload(item, snapshots.get(item.entity_id), dry_run=dry_run) for item in report.deleted],
+            "skipped": [_retention_item_payload(item, snapshots.get(item.entity_id), dry_run=dry_run) for item in report.skipped],
+            "errors": report.errors,
+            "warnings": report.warnings,
+        }
+    )
+
+
+def _protection_class(name: str, kind: str, hooks: list[str]) -> str:
+    searchable = f"{name} {kind}".lower()
+    if "secret" in searchable:
+        return "secrets"
+    if "pii" in searchable or "readi" in searchable:
+        return "pii"
+    if "access" in searchable:
+        return "access"
+    if "normalizer" in searchable or "provenance" in searchable:
+        return "provenance"
+    if "memory_pre_delete" in hooks:
+        return "deletion_protection"
+    return "memory_policy"
+
+
+def _configured_hook_plugins() -> list[dict[str, Any]]:
+    from altk_evolve.config.hooks import discover_hooks_config_path
+
+    specs = [spec.model_dump(mode="json") for spec in evolve_config.hooks.plugins]
+    yaml_path = evolve_config.hooks.plugins_yaml
+    if not yaml_path and not specs:
+        yaml_path = discover_hooks_config_path()
+    if yaml_path:
+        loaded = yaml.safe_load(Path(yaml_path).read_text(encoding="utf-8")) or {}
+        if not isinstance(loaded, dict):
+            raise ValueError(f"hooks config {yaml_path} must hold a mapping with a 'plugins' list")
+        yaml_specs = loaded.get("plugins", []) or []
+        if not isinstance(yaml_specs, list) or any(not isinstance(spec, dict) for spec in yaml_specs):
+            raise ValueError(f"hooks config {yaml_path} must contain a 'plugins' list of mappings")
+        specs = yaml_specs + specs
+    return specs
+
+
+@mcp.tool()
+def get_compliance_status(namespace_id: str | None = None) -> str:
+    """Report Evolve backend, retention, and configured memory-hook health."""
+    from altk_evolve.hooks.manager import get_plugin_manager, hooks_active
+    from altk_evolve.hooks.types import HookType, engine_available
+
+    resolved_ns = _resolve_namespace(namespace_id)
+    client = get_client()
+    manager = get_plugin_manager()
+    try:
+        specs = _configured_hook_plugins()
+    except (OSError, ValueError, yaml.YAMLError) as exc:
+        return _json_response({"healthy": False, "error": f"Unable to read hook configuration: {exc}"})
+
+    plugins = []
+    for spec in specs:
+        hooks = list(spec.get("hooks", []) or [])
+        mode = str(spec.get("mode", "sequential"))
+        enabled = mode != "disabled"
+        plugins.append(
+            {
+                "name": str(spec.get("name") or spec.get("kind") or "unnamed"),
+                "kind": str(spec.get("kind") or ""),
+                "protection_class": _protection_class(
+                    str(spec.get("name") or ""),
+                    str(spec.get("kind") or ""),
+                    hooks,
+                ),
+                "hooks": hooks,
+                "enabled": enabled,
+                "healthy": bool(manager is not None and enabled and all(manager.has_hooks_for(hook) for hook in hooks)),
+            }
+        )
+
+    try:
+        package_version = version("altk-evolve")
+    except PackageNotFoundError:
+        package_version = "unknown"
+
+    hook_coverage = {hook.value: hooks_active(hook) for hook in HookType}
+    hook_engine_available = engine_available()
+    retention_available = (callable(getattr(client, "scan_entities", None)) or callable(getattr(client, "get_all_entities", None))) and all(
+        callable(getattr(client, method, None)) for method in ("patch_entity_metadata", "delete_entity_by_id")
+    )
+    healthy = client.ready() and retention_available and all(not plugin["enabled"] or plugin["healthy"] for plugin in plugins)
+    return _json_response(
+        {
+            "healthy": healthy,
+            "evolve_version": package_version,
+            "backend": evolve_config.backend,
+            "namespace_id": resolved_ns,
+            "retention_available": retention_available,
+            "hooks_enabled": manager is not None,
+            "hook_engine_available": hook_engine_available,
+            "hook_coverage": hook_coverage,
+            "plugins": plugins,
+        }
+    )
+
+
 def _empty_store_user_facts_response(user_id: str) -> str:
     return json.dumps({"user_id": user_id, "stored_count": 0, "updates": []})
 
@@ -682,6 +1143,7 @@ def create_entity(
     owner_id: str | None = None,
     visibility: str = "private",
     namespace_id: str | None = None,
+    created_at: str | None = None,
 ) -> str:
     """
     Create a single entity in the namespace.
@@ -694,12 +1156,23 @@ def create_entity(
         owner_id: Optional user ID to record as the owner of this entity
         visibility: Visibility of the entity — 'private' (default) or 'public'
         namespace_id: Optional namespace override. Falls back to the configured default.
+        created_at: Optional ISO-8601 timestamp for administrative fixture/import data.
 
     Returns:
         JSON string with the entity update details (ADD/UPDATE/DELETE/NONE) and entity ID
     """
     logger.info(f"Creating entity of type: {entity_type} (namespace override: {namespace_id})")
     try:
+        parsed_created_at = None
+        if created_at:
+            from datetime import UTC, datetime
+
+            try:
+                parsed_created_at = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+            except ValueError:
+                return json.dumps({"error": "Invalid created_at", "message": "created_at must be ISO-8601"})
+            if parsed_created_at.tzinfo is None:
+                parsed_created_at = parsed_created_at.replace(tzinfo=UTC)
         if visibility not in ("private", "public"):
             return json.dumps({"error": f"Invalid visibility '{visibility}': must be 'private' or 'public'"})
         if visibility == "public" and not owner_id:
@@ -734,7 +1207,7 @@ def create_entity(
 
         entity = Entity(type=entity_type, content=content, metadata=metadata_dict)
 
-        updates, _ = _persist_entities(
+        updates, resolved_ns = _persist_entities(
             namespace_id=namespace_id,
             entities=[entity],
             enable_conflict_resolution=enable_conflict_resolution,
@@ -742,8 +1215,26 @@ def create_entity(
 
         if updates:
             update = updates[0]
+            if parsed_created_at and update.event == "ADD":
+                readback = get_client().set_entity_created_at(namespace_id=resolved_ns, entity_id=update.id, created_at=parsed_created_at)
+                return json.dumps(
+                    {
+                        "event": update.event,
+                        "id": readback.id,
+                        "type": readback.type,
+                        "content": readback.content,
+                        "metadata": readback.metadata,
+                        "created_at": readback.created_at.isoformat(),
+                    }
+                )
             return json.dumps(
-                {"event": update.event, "id": update.id, "type": update.type, "content": update.content, "metadata": update.metadata}
+                {
+                    "event": update.event,
+                    "id": update.id,
+                    "type": update.type,
+                    "content": update.content,
+                    "metadata": update.metadata,
+                }
             )
         else:
             return json.dumps({"error": "Entity creation failed"})
@@ -839,7 +1330,12 @@ def unpublish_entity(entity_id: str, user_id: str | None = None, namespace_id: s
 
 
 @mcp.tool()
-def delete_entity(entity_id: str, user_id: str | None = None, namespace_id: str | None = None) -> str:
+def delete_entity(
+    entity_id: str,
+    user_id: str | None = None,
+    agent_id: str | None = None,
+    namespace_id: str | None = None,
+) -> str:
     """
     Delete a specific entity by its ID.
 
@@ -859,8 +1355,7 @@ def delete_entity(entity_id: str, user_id: str | None = None, namespace_id: str
         if entity is None:
             return json.dumps({"success": False, "error": f"Entity {entity_id} not found"})
 
-        existing_owner = (entity.metadata or {}).get("owner_id")
-        if existing_owner is not None and user_id != existing_owner:
+        if not _entity_owned_by(entity, user_id, agent_id):
             logger.info(f"Delete denied for entity={entity_id} namespace={resolved_ns}: caller is not owner")
             return json.dumps({"error": "Permission denied: caller is not the owner of this entity"})
 
diff --git a/altk_evolve/frontend/ui/package-lock.json b/altk_evolve/frontend/ui/package-lock.json
index 86c7df29..12994e5e 100644
--- a/altk_evolve/frontend/ui/package-lock.json
+++ b/altk_evolve/frontend/ui/package-lock.json
@@ -11,7 +11,8 @@
         "lucide-react": "^0.575.0",
         "react": "^19.2.0",
         "react-dom": "^19.2.0",
-        "react-router-dom": "^7.15.0",
+        "react-router": "7.18.1",
+        "react-router-dom": "7.18.1",
         "recharts": "^3.7.0"
       },
       "devDependencies": {
@@ -3793,9 +3794,9 @@
       }
     },
     "node_modules/react-router": {
-      "version": "7.17.0",
-      "resolved": "https://registry.npmjs.org/react-router/-/react-router-7.17.0.tgz",
-      "integrity": "sha512-FDELK7rTMlCHO5+reyXsPlmfr7N1F91lPHsWYfMEGQm/KQ+F4JFM8jGoeQDmDvdTs93Fw9aSilH+uKRb4/jXvQ==",
+      "version": "7.18.1",
+      "resolved": "https://registry.npmjs.org/react-router/-/react-router-7.18.1.tgz",
+      "integrity": "sha512-GDLgg3i3uM0aeJO3Fm+TCS+sDQ7gu12T6x0qdTEzcwqEfleci7JwugVNIF3U//0FWKnJT7ptG+20B2jfDqnZAg==",
       "license": "MIT",
       "dependencies": {
         "cookie": "^1.0.1",
@@ -3815,12 +3816,12 @@
       }
     },
     "node_modules/react-router-dom": {
-      "version": "7.17.0",
-      "resolved": "https://registry.npmjs.org/react-router-dom/-/react-router-dom-7.17.0.tgz",
-      "integrity": "sha512-fyU2yjGups/hE6Xz0I5ZYbVL8Gx29eCjgpHaRaTaVU+OOAdfRX05KsvyRm0GO8YQwOkhpU3MurW1jyMUJn+zSw==",
+      "version": "7.18.1",
+      "resolved": "https://registry.npmjs.org/react-router-dom/-/react-router-dom-7.18.1.tgz",
+      "integrity": "sha512-KaZh+X/6UtEp28x51AUYZDMg9NGoz2ja3dNHa+ta/tk40vCzKhQ/RypCWBMLbmDr6//E24Vv5uPsrqXFozdkAg==",
       "license": "MIT",
       "dependencies": {
-        "react-router": "7.17.0"
+        "react-router": "7.18.1"
       },
       "engines": {
         "node": ">=20.0.0"
diff --git a/altk_evolve/frontend/ui/package.json b/altk_evolve/frontend/ui/package.json
index b450ba5f..82e78f66 100644
--- a/altk_evolve/frontend/ui/package.json
+++ b/altk_evolve/frontend/ui/package.json
@@ -14,7 +14,8 @@
     "lucide-react": "^0.575.0",
     "react": "^19.2.0",
     "react-dom": "^19.2.0",
-    "react-router-dom": "^7.15.0",
+    "react-router": "7.18.1",
+    "react-router-dom": "7.18.1",
     "recharts": "^3.7.0"
   },
   "devDependencies": {
diff --git a/altk_evolve/hooks/manager.py b/altk_evolve/hooks/manager.py
index b7149e77..b20adcf9 100644
--- a/altk_evolve/hooks/manager.py
+++ b/altk_evolve/hooks/manager.py
@@ -388,9 +388,9 @@ def _native_adapter_cls() -> Any:
         return _NATIVE_ADAPTER_CLS
 
     from cpex.framework import Plugin
-    from cpex.framework.models import PluginResult
+    from cpex.framework.models import PluginResult, PluginViolation
 
-    from altk_evolve.hooks.plugin import HookContext
+    from altk_evolve.hooks.plugin import HookContext, HookPolicyViolation
     from altk_evolve.hooks.types import HOOK_PAYLOADS, active_payload_cls
 
     class _NativePluginAdapter(Plugin):
@@ -444,7 +444,18 @@ async def _run(self, hook_type: HookType, cpex_payload: Any, context: Any) -> An
                 state=state,
                 request_id=getattr(gc, "request_id", "") if gc else "",
             )
-            out = method(plain, hook_ctx)  # native, sync; raises to halt (propagates -> on_error)
+            try:
+                out = method(plain, hook_ctx)
+            except HookPolicyViolation as violation:
+                return PluginResult(
+                    continue_processing=False,
+                    violation=PluginViolation(
+                        reason=violation.reason,
+                        description=violation.reason,
+                        code=violation.code,
+                        details=violation.details,
+                    ),
+                )
             if out is None:
                 return PluginResult(continue_processing=True)
             engine_cls: Any = active_payload_cls(hook_type)
diff --git a/altk_evolve/hooks/plugin.py b/altk_evolve/hooks/plugin.py
index 946020a3..a6c738eb 100644
--- a/altk_evolve/hooks/plugin.py
+++ b/altk_evolve/hooks/plugin.py
@@ -55,6 +55,22 @@ class HookContext:
     request_id: str = ""
 
 
+class HookPolicyViolation(Exception):
+    """Engine-neutral signal that a native plugin intentionally blocked an operation."""
+
+    def __init__(
+        self,
+        reason: str,
+        *,
+        code: str = "",
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(reason)
+        self.reason = reason
+        self.code = code
+        self.details = dict(details or {})
+
+
 @runtime_checkable
 class HookPlugin(Protocol):
     """Structural contract for a native hook plugin.
@@ -113,4 +129,4 @@ def llm_pre_call(self, payload: Any, context: HookContext) -> Any | None:
         return None
 
 
-__all__ = ["HookContext", "HookPlugin", "HookPluginBase"]
+__all__ = ["HookContext", "HookPlugin", "HookPluginBase", "HookPolicyViolation"]
diff --git a/altk_evolve/hooks/plugins/legal_hold.py b/altk_evolve/hooks/plugins/legal_hold.py
new file mode 100644
index 00000000..72740084
--- /dev/null
+++ b/altk_evolve/hooks/plugins/legal_hold.py
@@ -0,0 +1,19 @@
+"""Legal-hold policy for memory deletion."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from altk_evolve.hooks.plugin import HookContext, HookPluginBase, HookPolicyViolation
+
+
+class LegalHoldMemoryPlugin(HookPluginBase):
+    """Block deletion when stored metadata contains ``legal_hold: true``."""
+
+    def memory_pre_delete(self, payload: Any, context: HookContext) -> None:
+        if (payload.metadata or {}).get("legal_hold") is True:
+            raise HookPolicyViolation(
+                "entity is under legal hold",
+                code="LEGAL_HOLD",
+                details={"entity_id": payload.entity_id},
+            )
diff --git a/altk_evolve/retention/engine.py b/altk_evolve/retention/engine.py
index 972bcd56..dc8e6016 100644
--- a/altk_evolve/retention/engine.py
+++ b/altk_evolve/retention/engine.py
@@ -121,6 +121,7 @@ class RetentionEngine:
 
     def __init__(self, client: Any) -> None:
         self.client = client
+        self.last_scanned_entities: list[RecordedEntity] = []
 
     # ── signal helpers ────────────────────────────────────────────────
 
@@ -194,6 +195,7 @@ def evaluate(
         warnings: list[str] | None = None,
         skipped: list[RetentionItem] | None = None,
         scan_limit: int | None = None,
+        filters: dict | None = None,
     ) -> list[RetentionItem]:
         """Compute the actions a policy implies, without mutating anything.
 
@@ -206,11 +208,15 @@ def evaluate(
         *scan_limit* caps how many entities are fetched from the namespace in
         one call; it defaults to :attr:`FETCH_LIMIT`. When the fetch returns
         exactly the limit a warning is emitted, since entities beyond it were
-        not evaluated (and therefore not cascaded).
+        not evaluated (and therefore not cascaded). *filters* are forwarded to
+        the administrative scan so a shared namespace can be scoped by owner,
+        agent, or other metadata.
         """
         now = _as_aware(now) if now else datetime.datetime.now(datetime.UTC)
         limit = self.FETCH_LIMIT if scan_limit is None else scan_limit
-        entities = self.client.get_all_entities(namespace_id, limit=limit)
+        scan = getattr(self.client, "scan_entities", None) or self.client.get_all_entities
+        entities = scan(namespace_id, filters=filters, limit=limit)
+        self.last_scanned_entities = entities
         if warnings is not None and len(entities) >= limit:
             warnings.append(
                 f"scan hit the fetch limit of {limit} entities in namespace {namespace_id!r}; entities beyond it "
@@ -238,6 +244,8 @@ def record(item: RetentionItem) -> None:
             existing = actions.get(item.entity_id)
             if existing is not None and (existing.action == "delete" or item.action == "flag"):
                 return
+            if item.action == "delete" and skipped is not None:
+                skipped[:] = [entry for entry in skipped if entry.entity_id != item.entity_id]
             actions[item.entity_id] = item
 
         unstamped = 0
@@ -262,6 +270,9 @@ def record(item: RetentionItem) -> None:
             if degraded and rule.action == "delete":
                 choice = rule.on_missing_access_signal
                 if choice == "skip":
+                    existing = actions.get(e.id)
+                    if existing is not None and existing.action == "delete":
+                        continue
                     if skipped is not None:
                         skipped.append(
                             RetentionItem(
@@ -316,13 +327,15 @@ def apply(
         now: datetime.datetime | None = None,
         dry_run: bool = True,
         scan_limit: int | None = None,
+        filters: dict | None = None,
     ) -> RetentionReport:
         """Evaluate and — unless *dry_run* — flag/delete the matched entities.
 
         Dry run is the default: nothing is mutated and the report describes what
         *would* happen. *scan_limit* caps how many entities are fetched per
         namespace (defaults to :attr:`FETCH_LIMIT`); a boundary hit is surfaced
-        in ``report.warnings``.
+        in ``report.warnings``. *filters* use the backend's structured filter
+        syntax and constrain both direct matches and cascade candidates.
         """
         now = _as_aware(now) if now else datetime.datetime.now(datetime.UTC)
         report = RetentionReport(dry_run=dry_run)
@@ -333,6 +346,7 @@ def apply(
             warnings=report.warnings,
             skipped=report.skipped,
             scan_limit=scan_limit,
+            filters=filters,
         )
         flagged_at = now.isoformat()
 
diff --git a/docs/guides/memory-hooks.md b/docs/guides/memory-hooks.md
index bba49601..f5bfdfe3 100644
--- a/docs/guides/memory-hooks.md
+++ b/docs/guides/memory-hooks.md
@@ -99,6 +99,7 @@ Notes:
 |---|---|---|---|
 | `MetadataNormalizerPlugin` (native) | `memory_pre_write` | transform | Copies `task_id` → `trace_id` when only the former is present (MCP-saved trajectories vs Phoenix-synced ones) and stamps `created_at` |
 | `AccessStampPlugin` (native) | `memory_post_read` | fire_and_forget | Stamps `last_accessed` (ISO-8601 UTC) on read entities via the metadata-patch path |
+| `LegalHoldMemoryPlugin` (native) | `memory_pre_delete` | sequential | Refuses deletion while stored metadata contains `legal_hold: true` |
 | `PIIFilterMemoryPlugin` (**raw CPEX**) | `memory_pre_write`, `llm_pre_call` | sequential | Regex PII method (adapts the external `cpex-pii-filter` plugin onto Evolve's hook types); requires `pip install 'altk-evolve[pii-regex]'` |
 | `ReadiSemanticPIIPlugin` (native) | `memory_pre_write`, `llm_pre_call` | sequential | Semantic (NER) PII method via IBM READI — catches **names**, locations and organizations that regex cannot; requires `pip install 'altk-evolve[pii-semantic]'` |
 | `SecretsFilterMemoryPlugin` (native) | `memory_pre_write`, `llm_pre_call` | sequential | Structured **secrets** method (native plugin wrapping `cpex-secrets-detection`'s Rust core) — catches **credentials/tokens** (AWS keys, GitHub/Slack tokens, Stripe secrets, private-key blocks) that neither PII method targets; requires `pip install 'altk-evolve[secrets]'` |
diff --git a/docs/guides/retention.md b/docs/guides/retention.md
index 93fa6e7f..ee745d41 100644
--- a/docs/guides/retention.md
+++ b/docs/guides/retention.md
@@ -24,6 +24,20 @@ for item in [*report.deleted, *report.flagged]:
     print(item.action, item.entity_id, item.reason, "—", item.detail)
 ```
 
+Or through the MCP server:
+
+```text
+validate_retention_policy(policy='{"rules":[...]}')
+run_retention(policy='{"rules":[...]}')                    # dry run
+run_retention(policy='{"rules":[...]}', dry_run=false)     # enforce
+```
+
+The MCP report includes the retention item plus a pre-action entity snapshot
+(`content_preview`, attribution metadata, and session/provenance identifiers),
+so an audit UI can still explain an applied deletion after the entity is gone.
+`as_of` is an optional ISO-8601 clock override for deterministic audits and
+demonstrations.
+
 [`examples/retention_demo.py`](https://github.com/AgentToolkit/altk-evolve/blob/main/examples/retention_demo.py) is a runnable end-to-end walkthrough.
 
 ## Policy format
@@ -118,6 +132,7 @@ Read the dry run before you apply. That is the whole point of it.
 
 - **`AccessStampPlugin`** (shipped with the [hook seam](memory-hooks.md)) stamps `last_accessed` on every entity returned by a public `search_entities`, via `memory_post_read`. This is the automatic path, and **enabling it is what makes an unused rule mean anything**. Note its cost: fire-and-forget tasks are awaited before the read returns, so every public read pays one metadata write per returned entity (~3.7 ms vs ~0.1 ms for a 10-entity filesystem read).
 - **`EvolveClient.record_access(namespace_id, entity_ids)`** is the explicit path, for callers that do not run hooks, or that want to record a *use* that was not a store read — a memory pulled from a cache and actually acted on, say. It goes through the same core function as the plugin (`build_access_stamps`), so the key, the format, and the one-stamp-per-batch behaviour are identical. Running both is harmless.
+- **The MCP `record_access` tool** exposes that explicit path to remote integrations. The MCP `list_entities` tool defaults to an administrative scan that does not stamp access; set `record_access=true` when the list is a genuine user-facing recall.
 
 **If neither is in play, the signal does not exist.** The engine then falls back to `created_at` — and says so, rather than quietly pretending it measured disuse. Every affected item's `detail` names the fallback, and the report carries a run-level warning:
 
@@ -193,6 +208,7 @@ In scope: private entities under `.evolve/entities/` and session files under `.e
 - One failing entity does not abort the sweep — the failure is logged and recorded in `report.errors`, and the remaining entities are processed. The CLI exits non-zero when `errors` is non-empty.
 - Naive `created_at` values are treated as UTC.
 - Retention is not a hook: it is a periodic sweep you schedule. There is no automatic expiry on write or read.
+- Retention and administrative inventory use the client's non-access scan path, so a compliance sweep does not refresh every entity's `last_accessed` timestamp.
 
 ## Known limitations
 
diff --git a/examples/cuga_compliance_poc_hooks.yaml b/examples/cuga_compliance_poc_hooks.yaml
new file mode 100644
index 00000000..3800163c
--- /dev/null
+++ b/examples/cuga_compliance_poc_hooks.yaml
@@ -0,0 +1,42 @@
+plugins:
+  - name: metadata_normalizer
+    kind: altk_evolve.hooks.plugins.normalizer.MetadataNormalizerPlugin
+    hooks:
+      - memory_pre_write
+    mode: transform
+    priority: 40
+    on_error: fail
+    config:
+      stamp_trace_id: true
+      stamp_created_at: true
+
+  - name: access_stamp
+    kind: altk_evolve.hooks.plugins.access_stamp.AccessStampPlugin
+    hooks:
+      - memory_post_read
+    mode: fire_and_forget
+    priority: 50
+    on_error: fail
+
+  - name: legal_hold
+    kind: altk_evolve.hooks.plugins.legal_hold.LegalHoldMemoryPlugin
+    hooks:
+      - memory_pre_delete
+    mode: sequential
+    priority: 5
+    on_error: fail
+
+  - name: pii_filter_memory
+    kind: altk_evolve.hooks.plugins.pii.PIIFilterMemoryPlugin
+    hooks:
+      - memory_pre_write
+      - llm_pre_call
+    mode: sequential
+    priority: 10
+    on_error: fail
+    config:
+      detect_email: true
+      detect_ssn: true
+      detect_phone: true
+      default_mask_strategy: redact
+      redaction_text: "[REDACTED]"
diff --git a/examples/hooks_plugins.yaml b/examples/hooks_plugins.yaml
index ab32deeb..dabf465c 100644
--- a/examples/hooks_plugins.yaml
+++ b/examples/hooks_plugins.yaml
@@ -55,6 +55,16 @@ plugins:
     priority: 50
     on_error: fail         # fail-closed default; the plugin still guards its own stamping so the read never fails
 
+  # Refuses deletion while stored metadata has legal_hold: true.
+  - name: legal_hold
+    kind: altk_evolve.hooks.plugins.legal_hold.LegalHoldMemoryPlugin
+    description: Delete veto for records under legal hold
+    hooks:
+      - memory_pre_delete
+    mode: sequential
+    priority: 5
+    on_error: fail
+
   # Regex PII method on memory writes and LLM egress.
   # Requires: pip install 'altk-evolve[pii-regex]'  ([pii] is a back-compat alias)
   - name: pii_filter_memory
diff --git a/tests/unit/test_combine_guidelines.py b/tests/unit/test_combine_guidelines.py
index e1467824..2c48a9cf 100644
--- a/tests/unit/test_combine_guidelines.py
+++ b/tests/unit/test_combine_guidelines.py
@@ -260,10 +260,10 @@ def test_consolidate_guidelines_deletes_originals_and_inserts_new(self, mock_com
         ]
 
         mock_backend = MagicMock()
-        # Consolidation reads via the INTERNAL seam (_search_entities_impl), not
+        # Consolidation reads via the administrative scan seam, not
         # public search_entities, so a redacting post_read plugin can't make it
         # persist the redacted view back to the store.
-        mock_backend._search_entities_impl.return_value = entities_cluster
+        mock_backend.scan_entities.return_value = entities_cluster
         client = _make_client(mock_backend)
 
         with patch.object(client, "_cluster_guideline_entities", return_value=[entities_cluster]):
@@ -271,7 +271,7 @@ def test_consolidate_guidelines_deletes_originals_and_inserts_new(self, mock_com
 
         # The read must NOT go through the public (post_read-firing) API.
         mock_backend.search_entities.assert_not_called()
-        mock_backend._search_entities_impl.assert_called_once()
+        mock_backend.scan_entities.assert_called_once_with("test-ns", filters={"type": "guideline"}, limit=10000)
 
         # Verify insert was called with correct args
         assert mock_backend.update_entities.call_count == 1
diff --git a/tests/unit/test_filesystem_backend.py b/tests/unit/test_filesystem_backend.py
index 44398c0b..4491ff24 100644
--- a/tests/unit/test_filesystem_backend.py
+++ b/tests/unit/test_filesystem_backend.py
@@ -1,6 +1,8 @@
 from pathlib import Path
 
 import pytest
+from datetime import UTC, datetime
+from altk_evolve.schema.core import Entity
 
 from altk_evolve.backend.filesystem import FilesystemEntityBackend
 from altk_evolve.config.evolve import EvolveConfig
@@ -67,3 +69,16 @@ def test_save_tolerates_stale_shared_tmp(backend: FilesystemEntityBackend, tmp_p
 
     target = tmp_path / "ns_busy.json"
     assert target.exists() and target.stat().st_size > 0
+
+
+@pytest.mark.unit
+def test_set_entity_created_at_persists_and_reads_back(client: EvolveClient):
+    client.ensure_namespace("imports")
+    update = client.update_entities("imports", [Entity(type="fact", content="fixture", metadata={})], enable_conflict_resolution=False)[0]
+    expected = datetime(2020, 1, 2, tzinfo=UTC)
+
+    stamped = client.set_entity_created_at("imports", update.id, expected)
+    readback = client.scan_entities("imports", filters={"id": update.id}, limit=1)[0]
+
+    assert stamped.created_at == expected
+    assert readback.created_at == expected
diff --git a/tests/unit/test_hooks_seam.py b/tests/unit/test_hooks_seam.py
index 5b14bed4..97a09d30 100644
--- a/tests/unit/test_hooks_seam.py
+++ b/tests/unit/test_hooks_seam.py
@@ -663,6 +663,33 @@ def test_legal_hold_on_external_delete_raises_and_preserves_entity(client: Evolv
     assert client.get_entity_by_id("ns", other.id) is None
 
 
+@pytest.mark.unit
+def test_shipped_legal_hold_plugin_blocks_delete_with_stable_code(client: EvolveClient):
+    enable_hooks(
+        specs=[
+            HookPluginSpec(
+                name="legal_hold",
+                kind="altk_evolve.hooks.plugins.legal_hold.LegalHoldMemoryPlugin",
+                hooks=[HookType.MEMORY_PRE_DELETE.value],
+                mode="sequential",
+                priority=5,
+                on_error="fail",
+            )
+        ]
+    )
+    client.create_namespace("ns")
+    _write(client, "ns", "keep me", {"legal_hold": True})
+    entity = client.search_entities("ns", limit=1)[0]
+
+    with pytest.raises(
+        MemoryPolicyViolation,
+        match=r"\[LEGAL_HOLD\] entity is under legal hold",
+    ):
+        client.delete_entity_by_id("ns", entity.id)
+
+    assert client.get_entity_by_id("ns", entity.id) is not None
+
+
 @pytest.mark.unit
 def test_external_delete_payload_carries_fetched_metadata(client: EvolveClient):
     from altk_evolve.schema.exceptions import EvolveException
diff --git a/tests/unit/test_mcp_compliance_tools.py b/tests/unit/test_mcp_compliance_tools.py
new file mode 100644
index 00000000..037bd599
--- /dev/null
+++ b/tests/unit/test_mcp_compliance_tools.py
@@ -0,0 +1,416 @@
+import datetime
+import json
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from altk_evolve.frontend.mcp.mcp_server import (
+    delete_entity,
+    get_compliance_status,
+    get_entity,
+    list_entities,
+    patch_entity_metadata,
+    record_access,
+    run_retention,
+    validate_retention_policy,
+)
+from altk_evolve.schema.core import RecordedEntity
+
+pytestmark = pytest.mark.unit
+
+NOW = datetime.datetime(2026, 7, 24, 12, 0, tzinfo=datetime.UTC)
+
+
+def _entity(
+    entity_id: str,
+    *,
+    entity_type: str = "fact",
+    created_days_ago: int = 0,
+    metadata: dict | None = None,
+) -> RecordedEntity:
+    return RecordedEntity(
+        id=entity_id,
+        type=entity_type,
+        content=f"Memory {entity_id}",
+        metadata=metadata or {},
+        created_at=NOW - datetime.timedelta(days=created_days_ago),
+    )
+
+
+@pytest.fixture
+def client():
+    with (
+        patch("altk_evolve.frontend.mcp.mcp_server._resolve_namespace", return_value="tenant-a"),
+        patch("altk_evolve.frontend.mcp.mcp_server.get_client") as get_client,
+    ):
+        yield get_client.return_value
+
+
+def test_list_entities_returns_filtered_paginated_inventory_without_recording_access(client):
+    client.scan_entities.return_value = [
+        _entity("old", created_days_ago=10, metadata={"user_id": "user-1", "agent_id": "agent-a"}),
+        _entity("new", created_days_ago=1, metadata={"user_id": "user-1", "agent_id": "agent-a"}),
+        _entity("other-user", metadata={"user_id": "user-2", "agent_id": "agent-a"}),
+    ]
+
+    first = json.loads(
+        list_entities(
+            entity_types=["fact"],
+            user_id="user-1",
+            agent_id="agent-a",
+            limit=1,
+            namespace_id="tenant-a",
+        )
+    )
+    second = json.loads(
+        list_entities(
+            entity_types=["fact"],
+            user_id="user-1",
+            agent_id="agent-a",
+            cursor=first["next_cursor"],
+            limit=1,
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert first["total"] == 2
+    assert first["items"][0]["id"] == "new"
+    assert "content" not in first["items"][0]
+    assert second["items"][0]["id"] == "old"
+    assert second["next_cursor"] is None
+    client.scan_entities.assert_called_with("tenant-a", limit=100_000)
+    client.get_all_entities.assert_not_called()
+
+
+def test_list_entities_can_record_user_facing_access(client):
+    entity = _entity("one", metadata={"user_id": "user-1"})
+    client.scan_entities.return_value = [entity]
+    client.get_entity_by_id.return_value = entity
+    client.record_access.return_value = ["one"]
+
+    result = json.loads(list_entities(user_id="user-1", record_access=True, namespace_id="tenant-a"))
+
+    assert result["items"][0]["id"] == "one"
+    assert result["items"][0]["metadata"]["last_accessed"]
+    client.scan_entities.assert_called_once_with("tenant-a", limit=100_000)
+    client.get_entity_by_id.assert_called_once_with("tenant-a", "one")
+    client.record_access.assert_called_once()
+    assert client.record_access.call_args.args == ("tenant-a", ["one"])
+    assert client.record_access.call_args.kwargs["when"].tzinfo is datetime.UTC
+
+
+def test_list_entities_cursor_counts_scanned_rows_when_access_read_disappears(client):
+    entities = [
+        _entity("new", created_days_ago=1),
+        _entity("old", created_days_ago=2),
+    ]
+    client.scan_entities.return_value = entities
+    client.get_entity_by_id.side_effect = [None, entities[1]]
+    client.record_access.return_value = ["old"]
+
+    result = json.loads(list_entities(limit=1, record_access=True, namespace_id="tenant-a"))
+
+    assert result["items"] == []
+    assert result["next_cursor"] is not None
+    assert json.loads(list_entities(cursor=result["next_cursor"], limit=1, namespace_id="tenant-a"))["items"][0]["id"] == "old"
+
+
+@pytest.mark.parametrize("metadata", [{"owner_id": "user-1"}, {}])
+def test_get_entity_enforces_attributed_owner(client, metadata):
+    client.scan_entities.return_value = [_entity("one", metadata=metadata)]
+
+    denied = json.loads(
+        get_entity(
+            "one",
+            user_id="user-2",
+            record_access=False,
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert denied["error"].startswith("Permission denied")
+
+
+def test_get_entity_denies_non_owner_before_access_stamping(client):
+    client.scan_entities.return_value = [_entity("one", metadata={"owner_id": "user-1"})]
+
+    denied = json.loads(get_entity("one", user_id="user-2", namespace_id="tenant-a"))
+
+    assert denied["error"].startswith("Permission denied")
+    client.get_entity_by_id.assert_not_called()
+    client.record_access.assert_not_called()
+
+
+def test_get_entity_enforces_agent_scope(client):
+    client.scan_entities.return_value = [
+        _entity(
+            "one",
+            metadata={"owner_id": "user-1", "agent_id": "agent-a"},
+        )
+    ]
+
+    denied = json.loads(
+        get_entity(
+            "one",
+            user_id="user-1",
+            agent_id="agent-b",
+            record_access=False,
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert denied["error"].startswith("Permission denied")
+
+
+def test_patch_entity_metadata_routes_through_client_hook_seam(client):
+    original = _entity("one", metadata={"owner_id": "user-1"})
+    updated = original.model_copy(update={"metadata": {"owner_id": "user-1", "legal_hold": True}})
+    client.scan_entities.return_value = [original]
+    client.patch_entity_metadata.return_value = updated
+
+    result = json.loads(
+        patch_entity_metadata(
+            "one",
+            json.dumps({"legal_hold": True}),
+            user_id="user-1",
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert result["metadata"]["legal_hold"] is True
+    client.patch_entity_metadata.assert_called_once_with("tenant-a", "one", {"legal_hold": True})
+
+
+def test_patch_entity_metadata_enforces_agent_scope(client):
+    client.scan_entities.return_value = [
+        _entity(
+            "one",
+            metadata={"owner_id": "user-1", "agent_id": "agent-a"},
+        )
+    ]
+
+    denied = json.loads(
+        patch_entity_metadata(
+            "one",
+            json.dumps({"title": "Changed"}),
+            user_id="user-1",
+            agent_id="agent-b",
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert denied["error"].startswith("Permission denied")
+    client.patch_entity_metadata.assert_not_called()
+
+
+def test_record_access_reports_updated_denied_and_missing_ids(client):
+    owned = _entity("owned", metadata={"user_id": "user-1"})
+    denied = _entity("denied", metadata={"user_id": "user-2"})
+    client.scan_entities.side_effect = [[owned], [denied], []]
+    client.record_access.return_value = ["owned"]
+
+    result = json.loads(
+        record_access(
+            ["owned", "denied", "missing"],
+            accessed_at=NOW.isoformat(),
+            user_id="user-1",
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert result["updated_ids"] == ["owned"]
+    assert result["denied_ids"] == ["denied"]
+    assert result["missing_ids"] == ["missing"]
+    client.record_access.assert_called_once_with("tenant-a", ["owned"], when=NOW)
+
+
+def test_record_access_skips_backend_write_when_every_id_is_denied_or_missing(client):
+    denied = _entity("denied", metadata={"user_id": "user-2"})
+    client.scan_entities.side_effect = [[denied], []]
+
+    result = json.loads(record_access(["denied", "missing"], user_id="user-1", namespace_id="tenant-a"))
+
+    assert result["updated_ids"] == []
+    client.record_access.assert_not_called()
+
+
+def test_record_access_enforces_agent_scope(client):
+    other_agent = _entity(
+        "other-agent",
+        metadata={"user_id": "user-1", "agent_id": "agent-a"},
+    )
+    client.scan_entities.return_value = [other_agent]
+
+    result = json.loads(
+        record_access(
+            ["other-agent"],
+            user_id="user-1",
+            agent_id="agent-b",
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert result["denied_ids"] == ["other-agent"]
+    client.record_access.assert_not_called()
+
+
+def test_delete_entity_enforces_user_and_agent_scope(client):
+    entity = _entity(
+        "one",
+        metadata={"owner_id": "user-1", "agent_id": "agent-a"},
+    )
+    client.get_entity_by_id.return_value = entity
+
+    denied = json.loads(
+        delete_entity(
+            "one",
+            user_id="user-1",
+            agent_id="agent-b",
+            namespace_id="tenant-a",
+        )
+    )
+
+    assert denied["error"].startswith("Permission denied")
+    client.delete_entity_by_id.assert_not_called()
+
+
+def test_validate_retention_policy_normalizes_valid_policy():
+    result = json.loads(
+        validate_retention_policy(
+            json.dumps(
+                {
+                    "rules": [
+                        {
+                            "name": "stale-facts",
+                            "entity_type": "fact",
+                            "max_age_days": 90,
+                            "action": "flag",
+                        }
+                    ]
+                }
+            )
+        )
+    )
+
+    assert result["valid"] is True
+    assert result["normalized_policy"]["rules"][0]["on_missing_access_signal"] == "skip"
+
+
+def test_validate_retention_policy_returns_field_errors():
+    result = json.loads(validate_retention_policy(json.dumps({"rules": [{"name": "invalid"}]})))
+
+    assert result["valid"] is False
+    assert result["errors"]
+
+
+def test_run_retention_returns_real_entity_references_and_predelete_snapshot(client):
+    entity = _entity(
+        "old-session",
+        entity_type="trajectory",
+        created_days_ago=400,
+        metadata={
+            "user_id": "user-1",
+            "agent_id": "agent-a",
+            "session_id": "thread-9",
+            "task_id": "trace-9",
+        },
+    )
+    client.scan_entities.return_value = [entity]
+    result = json.loads(
+        run_retention(
+            policy=json.dumps(
+                {
+                    "rules": [
+                        {
+                            "name": "old-sessions",
+                            "entity_type": "trajectory",
+                            "max_age_days": 365,
+                            "action": "delete",
+                            "cascade_derived": True,
+                        }
+                    ]
+                }
+            ),
+            dry_run=False,
+            as_of=NOW.isoformat(),
+            run_id="run-1",
+            namespace_id="tenant-a",
+            metadata_filters=json.dumps({"agent_id": "agent-a"}),
+        )
+    )
+
+    deleted = result["deleted"][0]
+    assert result["run_id"] == "run-1"
+    assert result["metadata_filters"] == {"agent_id": "agent-a"}
+    client.scan_entities.assert_called_once_with(
+        "tenant-a",
+        filters={"metadata.agent_id": "agent-a"},
+        limit=100_000,
+    )
+    assert deleted["entity_id"] == "old-session"
+    assert deleted["outcome"] == "deleted"
+    assert deleted["session_id"] == "thread-9"
+    assert deleted["content_preview"] == "Memory old-session"
+
+
+def test_get_compliance_status_reports_configured_plugin_health(client):
+    client.ready.return_value = True
+    specs = [
+        {
+            "name": "access-stamp",
+            "kind": "altk_evolve.hooks.plugins.access_stamp.AccessStampPlugin",
+            "hooks": ["memory_post_read"],
+            "mode": "fire_and_forget",
+        }
+    ]
+
+    with (
+        patch("altk_evolve.frontend.mcp.mcp_server._configured_hook_plugins", return_value=specs),
+        patch("altk_evolve.hooks.manager.get_plugin_manager", return_value=MagicMock()),
+        patch("altk_evolve.hooks.manager.hooks_active", return_value=True),
+        patch("altk_evolve.hooks.types.engine_available", return_value=True),
+        patch("altk_evolve.frontend.mcp.mcp_server.version", return_value="1.1.5"),
+    ):
+        result = json.loads(get_compliance_status(namespace_id="tenant-a"))
+
+    assert result["healthy"] is True
+    assert result["retention_available"] is True
+    assert result["plugins"][0]["protection_class"] == "access"
+    assert result["plugins"][0]["healthy"] is True
+
+
+def test_get_compliance_status_marks_unregistered_plugin_unhealthy(client):
+    client.ready.return_value = True
+    manager = MagicMock()
+    manager.has_hooks_for.return_value = False
+    specs = [
+        {
+            "name": "legal-hold",
+            "kind": "example.LegalHoldPlugin",
+            "hooks": ["memory_pre_delete"],
+            "mode": "sequential",
+        }
+    ]
+
+    with (
+        patch("altk_evolve.frontend.mcp.mcp_server._configured_hook_plugins", return_value=specs),
+        patch("altk_evolve.hooks.manager.get_plugin_manager", return_value=manager),
+        patch("altk_evolve.hooks.manager.hooks_active", return_value=False),
+        patch("altk_evolve.hooks.types.engine_available", return_value=True),
+        patch("altk_evolve.frontend.mcp.mcp_server.version", return_value="1.1.5"),
+    ):
+        result = json.loads(get_compliance_status(namespace_id="tenant-a"))
+
+    assert result["healthy"] is False
+    assert result["plugins"][0]["healthy"] is False
+
+
+def test_get_compliance_status_handles_non_mapping_hooks_config(client, tmp_path):
+    config_path = tmp_path / "hooks.yaml"
+    config_path.write_text("- malformed\n", encoding="utf-8")
+
+    with patch("altk_evolve.frontend.mcp.mcp_server.evolve_config.hooks.plugins_yaml", str(config_path)):
+        result = json.loads(get_compliance_status(namespace_id="tenant-a"))
+
+    assert result["healthy"] is False
+    assert "Unable to read hook configuration" in result["error"]
diff --git a/tests/unit/test_mcp_server.py b/tests/unit/test_mcp_server.py
index c164672e..72d76122 100644
--- a/tests/unit/test_mcp_server.py
+++ b/tests/unit/test_mcp_server.py
@@ -92,6 +92,20 @@ def test_create_entity_metadata_injection_manual_policy(mock_get_client):
     assert entity.metadata["creation_mode"] == "manual"
 
 
+def test_create_entity_import_timestamp_is_read_back(mock_get_client):
+    stamped = datetime.datetime(2020, 1, 2, tzinfo=datetime.UTC)
+    mock_get_client.update_entities.return_value = [EntityUpdate(id="123", type="fact", content="fake", event="ADD", metadata={})]
+    mock_get_client.set_entity_created_at.return_value = RecordedEntity(
+        id="123", type="fact", content="fake", created_at=stamped, metadata={}
+    )
+
+    result = json.loads(create_entity(content="fake", entity_type="fact", created_at="2020-01-02T00:00:00Z"))
+
+    assert result["id"] == "123"
+    assert result["created_at"].startswith("2020-01-02T00:00:00")
+    mock_get_client.set_entity_created_at.assert_called_once()
+
+
 def test_create_entity_no_metadata_injection_for_other_types(mock_get_client):
     mock_update = EntityUpdate(id="123", type="log", content="App started", event="ADD", metadata={})
     mock_get_client.update_entities.return_value = [mock_update]
diff --git a/tests/unit/test_retention.py b/tests/unit/test_retention.py
index aaf0b6e3..5c78a4e2 100644
--- a/tests/unit/test_retention.py
+++ b/tests/unit/test_retention.py
@@ -65,6 +65,69 @@ def test_age_flag_marks_old_entities_without_deleting():
     assert "retention_flagged_at" not in client.store["2"].metadata
 
 
+def test_retention_prefers_non_access_administrative_scan():
+    class ScanAwareClient(FakeClient):
+        def __init__(self, entities):
+            super().__init__(entities)
+            self.scan_calls = 0
+            self.read_calls = 0
+
+        def scan_entities(self, namespace_id, filters=None, limit=100):
+            self.scan_calls += 1
+            return list(self.store.values())
+
+        def get_all_entities(self, namespace_id, filters=None, limit=100):
+            self.read_calls += 1
+            return list(self.store.values())
+
+    client = ScanAwareClient([_entity("1", created_days_ago=100)])
+    policy = RetentionPolicy(rules=[RetentionRule(name="stale", max_age_days=90, action="flag")])
+
+    RetentionEngine(client).apply("ns", policy, now=NOW, dry_run=True)
+
+    assert client.scan_calls == 1
+    assert client.read_calls == 0
+
+
+def test_retention_forwards_administrative_scope_filters():
+    class FilterAwareClient(FakeClient):
+        def __init__(self, entities):
+            super().__init__(entities)
+            self.filters = None
+
+        def scan_entities(self, namespace_id, filters=None, limit=100):
+            self.filters = filters
+            return list(self.store.values())
+
+    client = FilterAwareClient([_entity("1", created_days_ago=100)])
+    policy = RetentionPolicy(rules=[RetentionRule(name="stale", max_age_days=90, action="flag")])
+
+    RetentionEngine(client).apply(
+        "ns",
+        policy,
+        now=NOW,
+        dry_run=True,
+        filters={"metadata.agent_id": "agent-a"},
+    )
+
+    assert client.filters == {"metadata.agent_id": "agent-a"}
+
+
+def test_retention_scan_does_not_require_get_all_entities():
+    class ScanOnlyClient:
+        def scan_entities(self, namespace_id, filters=None, limit=100):
+            return [_entity("1", created_days_ago=100)]
+
+        def patch_entity_metadata(self, namespace_id, entity_id, metadata_patch):
+            pass
+
+    policy = RetentionPolicy(rules=[RetentionRule(name="stale", max_age_days=90, action="flag")])
+
+    report = RetentionEngine(ScanOnlyClient()).apply("ns", policy, now=NOW, dry_run=True)
+
+    assert len(report.flagged) == 1
+
+
 def test_age_delete_removes_old_entities():
     client = FakeClient([_entity("1", created_days_ago=400)])
     policy = RetentionPolicy(rules=[RetentionRule(name="old", max_age_days=365, action="delete")])
@@ -444,6 +507,40 @@ def test_cascade_delete_supersedes_a_flag_from_an_earlier_rule():
     assert report.flagged == []
 
 
+def test_cascade_delete_supersedes_a_degraded_skip_from_an_earlier_rule():
+    trajectory = _entity(
+        "traj",
+        type="trajectory",
+        created_days_ago=400,
+        metadata={"trace_id": "T1", "last_accessed": (NOW - datetime.timedelta(days=400)).isoformat()},
+    )
+    derived = _entity("g1", type="guideline", created_days_ago=220, metadata={"source_task_id": "T1"})
+    policy = RetentionPolicy(
+        rules=[
+            RetentionRule(
+                name="unused-guidelines",
+                entity_type="guideline",
+                max_unused_days=180,
+                action="delete",
+                on_missing_access_signal="skip",
+            ),
+            RetentionRule(
+                name="old-sessions",
+                entity_type="trajectory",
+                max_age_days=365,
+                action="delete",
+                cascade_derived=True,
+            ),
+        ]
+    )
+
+    for entities in ([derived, trajectory], [trajectory, derived]):
+        report = RetentionEngine(FakeClient(entities)).apply("ns", policy, now=NOW, dry_run=True)
+
+        assert {item.entity_id for item in report.deleted} == {"traj", "g1"}
+        assert report.skipped == []
+
+
 # ── record_access: the explicit half of the access signal ─────────────