Function bodies 288 total

    def summary(self, entry: MemoryEntry) -> str:
        """Return a single-line provenance summary string."""
        tier = self._tracker.get_tier(entry)
        rel = self._tracker.get_reliability(entry)
        tagged_at = entry.metadata.get("provenance_tagged_at", "unknown")
        note = entry.metadata.get("provenance_note", "")
        parts = [
            f"source={entry.source.value}",
            f"tier={tier}",
            f"reliability={rel:.2f}",
            f"tagged_at={tagged_at}",
        ]
        if note:
            parts.append(f"note={note!r}")
        return " | ".join(parts)

    def as_dict(self, entry: MemoryEntry) -> dict[str, str | float]:
        """Return provenance data as a structured dictionary."""
        return {
            "memory_id": entry.memory_id,
            "source": entry.source.value,
            "tier": self._tracker.get_tier(entry),
            "reliability": self._tracker.get_reliability(entry),
            "tagged_at": entry.metadata.get("provenance_tagged_at", ""),
            "note": entry.metadata.get("provenance_note", ""),
        }

    def tier_label(self, source: MemorySource) -> str:
        """Return a human-readable reliability tier label."""
        score = self.score(source)
        if score >= 0.90:
            return "VERIFIED_TOOL"
        if score >= 0.80:
            return "VERIFIED_DOCUMENT"
        if score >= 0.65:
            return "USER_INPUT"
        if score >= 0.45:
            return "AGENT_INFERENCE"
        return "UNKNOWN"

    def tag(
        self,
        entry: MemoryEntry,
        note: str = "",
    ) -> MemoryEntry:
        """Return a new entry with provenance metadata applied from its source."""
        rel_score = self._reliability.score(entry.source)
        tier = self._reliability.tier_label(entry.source)
        now_iso = datetime.now(timezone.utc).isoformat()
        updated_meta = dict(entry.metadata)
        updated_meta[self._RELIABILITY_KEY] = str(round(rel_score, 4))
        updated_meta[self._TIER_KEY] = tier
        updated_meta[self._TAGGED_AT_KEY] = now_iso
        if note:
            updated_meta[self._SOURCE_NOTE_KEY] = note
        return entry.model_copy(update={"metadata": updated_meta})

    def get_reliability(self, entry: MemoryEntry) -> float:
        """Return the stored reliability score, or re-compute from source."""
        raw = entry.metadata.get(self._RELIABILITY_KEY)
        if raw is not None:
            try:
                return float(raw)
            except ValueError:
                pass
        return self._reliability.score(entry.source)

    def override_reliability(
        self,
        entry: MemoryEntry,
        new_source: MemorySource,
    ) -> MemoryEntry:
        """Re-tag an entry with a different source, updating all provenance fields."""
        updated = entry.model_copy(update={"source": new_source})
        return self.tag(updated)

    def __post_init__(self) -> None:
        total = self.vector + self.graph + self.recency
        if abs(total - 1.0) > 0.01:
            raise ValueError(
                f"FusionWeights must sum to 1.0, got {total:.4f} "
                f"(vector={self.vector}, graph={self.graph}, recency={self.recency})"
            )
        for name, value in (("vector", self.vector), ("graph", self.graph), ("recency", self.recency)):
            if value < 0.0:
                raise ValueError(f"Weight '{name}' must be >= 0.0, got {value}")

    def fuse(
        self,
        vector_results: list[RetrievalResult],
        graph_results: list[RetrievalResult],
        recency_results: list[RetrievalResult],
        top_k: int = 10,
    ) -> list[RetrievalResult]:
        """Fuse results from all three backends into a single ranked list.

        Parameters
        ----------
        vector_results:
            Results from the vector store (source tag ``"vector"``).
        graph_results:
            Results from the graph retriever (source tag ``"graph"``).
        recency_results:
            Results from the recency scorer (source tag ``"recency"``).
        top_k:
            Maximum number of fused results to return.

        Returns
        -------
        list[RetrievalResult]
            Merged results with ``source="hybrid"`` sorted by fused score
            descending, up to *top_k* items.
        """
        if top_k <= 0:
            return []

        # Index each result list by content string
        vector_

    def __init__(self, hop_decay: float = 0.8) -> None:
        if not (0.0 < hop_decay <= 1.0):
            raise ValueError(f"hop_decay must be in (0, 1], got {hop_decay}")
        self._hop_decay = hop_decay
        # Adjacency list: node_label -> list of (relation, target_label, weight)
        self._graph: dict[str, list[tuple[str, str, float]]] = {}
        # Reverse adjacency for undirected search (also traverse incoming edges)
        self._reverse_graph: dict[str, list[tuple[str, str, float]]] = {}

    def add_edge(
        self,
        source: str,
        relation: str,
        target: str,
        weight: float = 1.0,
    ) -> None:
        """Add a directed, weighted edge to the knowledge graph.

        Parameters
        ----------
        source:
            Label of the source node.
        relation:
            Relation type string (e.g. ``"capital_of"``, ``"knows"``).
        target:
            Label of the target node.
        weight:
            Edge weight in ``(0.0, 1.0]``.  Higher weight means greater
            relevance of this edge.

        Raises
        ------
        ValueError
            If *weight* is not in ``(0.0, 1.0]``.
        """
        if not (0.0 < weight <= 1.0):
            raise ValueError(f"weight must be in (0, 1], got {weight}")

        if source not in self._graph:
            self._graph[source] = []
        self._graph[source].append((relation, target, weight))

        # Ensure target node exists in the graph even if it has no outgo

    def retrieve(
        self,
        query: str,
        max_hops: int = 2,
        top_k: int = 5,
        bidirectional: bool = True,
    ) -> list[RetrievalResult]:
        """Retrieve nodes reachable from seed nodes matching *query*.

        Seed selection: any node whose label shares at least one token
        with the query (case-insensitive).

        BFS traversal up to *max_hops* hops collects candidates.  Each
        candidate is scored as ``seed_weight × hop_decay^hop_distance``
        where ``seed_weight`` is the maximum edge weight on any path from
        the seed to the candidate.

        Parameters
        ----------
        query:
            Free-text query used to identify seed nodes.
        max_hops:
            Maximum number of hops from a seed node.
        top_k:
            Maximum number of results to return.
        bidirectional:
            When ``True`` also traverse reverse edges so that nodes
            mentioning any query token as a *target* c

    def __init__(
        self,
        vector_store: VectorStoreProtocol | None = None,
        graph: GraphRetriever | None = None,
        recency: RecencyScorer | None = None,
        fusion: ScoreFusion | None = None,
    ) -> None:
        self._vector_store = vector_store
        self._graph = graph if graph is not None else GraphRetriever()
        self._recency = recency if recency is not None else RecencyScorer()
        self._fusion = fusion if fusion is not None else ScoreFusion()

    def retrieve(self, query: str, top_k: int = 10) -> list[RetrievalResult]:
        """Retrieve the most relevant results for *query* across all signals.

        Parameters
        ----------
        query:
            Free-text query string.
        top_k:
            Maximum number of results to return.

        Returns
        -------
        list[RetrievalResult]
            Results ranked by fused score, highest first.
        """
        vector_results: list[RetrievalResult] = []
        graph_results: list[RetrievalResult] = []
        recency_results: list[RetrievalResult] = []

        # --- Vector signal ---
        if self._vector_store is not None:
            try:
                vector_results = self._vector_store.query(query, top_k=top_k)
            except Exception:
                vector_results = []

        # --- Graph signal ---
        try:
            graph_results = self._graph.retrieve(query, top_k=top_k)
        except Exception:
            graph_results =

    def _recency_from_existing(
        self,
        existing: list[RetrievalResult],
        top_k: int,
    ) -> list[RetrievalResult]:
        """Produce recency-scored versions of already-retrieved results.

        Extracts ``"timestamp"`` from each result's metadata (if present)
        and scores it with the recency scorer.  Results without a timestamp
        receive a score of 0.0.

        Parameters
        ----------
        existing:
            Results from vector and graph retrievers.
        top_k:
            Maximum results to return.

        Returns
        -------
        list[RetrievalResult]
        """
        from datetime import datetime, timezone

        scored: list[RetrievalResult] = []
        for result in existing:
            ts_raw = result.metadata.get("timestamp")
            if isinstance(ts_raw, datetime):
                timestamp = ts_raw
            elif isinstance(ts_raw, str):
                try:
                    timestamp = datetime.fro

    def score(self, entry: MemoryEntry, query: str) -> float:
        """Compute a combined relevance score in [0, 1].

        Parameters
        ----------
        entry:
            The memory entry to score.
        query:
            The original search query.

        Returns
        -------
        float
            Weighted score in [0.0, 1.0].
        """
        weights = self._weights
        total_weight = weights.total
        if total_weight == 0.0:
            return 0.0

        text_score = self._text_relevance(entry.content, query) * weights.text_relevance
        importance_score = entry.importance_score * weights.importance
        freshness_score = entry.freshness_score * weights.freshness
        provenance_score = self._provenance_score(entry) * weights.provenance

        raw = (text_score + importance_score + freshness_score + provenance_score) / total_weight
        return round(min(1.0, max(0.0, raw)), 6)

    def rank(
        self,
        entries: list[tuple[MemoryEntry, str]],
    ) -> list[tuple[float, MemoryEntry]]:
        """Rank a list of (entry, query) pairs.

        Parameters
        ----------
        entries:
            Each element is ``(entry, query)`` where query is the original
            search string used to retrieve this entry.

        Returns
        -------
        list[tuple[float, MemoryEntry]]
            ``(score, entry)`` pairs sorted by score descending.
        """
        scored = [(self.score(entry, query), entry) for entry, query in entries]
        scored.sort(key=lambda x: x[0], reverse=True)
        return scored

    def _text_relevance(self, content: str, query: str) -> float:
        """TF-style overlap score between query tokens and content."""
        query_tokens = _tokenise(query)
        if not query_tokens:
            return 0.0
        content_tokens = _tokenise(content)
        if not content_tokens:
            return 0.0

        total_tf = 0.0
        for token in query_tokens:
            tf = content_tokens.count(token) / len(content_tokens)
            total_tf += tf

        # Average TF across query tokens, then apply a scaling function
        avg_tf = total_tf / len(query_tokens)
        # Sigmoid-style saturation: map small TF values to meaningful scores
        return round(_sigmoid_scale(avg_tf), 6)

    def _provenance_score(self, entry: MemoryEntry) -> float:
        """Extract provenance reliability from metadata, or estimate from source."""
        raw = entry.metadata.get("provenance_reliability")
        if raw is not None:
            try:
                return float(raw)
            except ValueError:
                pass
        # Fallback: heuristic from source type
        return _SOURCE_PROVENANCE.get(entry.source.value, 0.5)

def _sigmoid_scale(value: float, steepness: float = 10.0) -> float:
    """Map a small float [0, 1] to a sharper [0, 1] via a sigmoid."""
    if value <= 0.0:
        return 0.0
    # Standard sigmoid shifted so that value=0.5 maps near 0.99
    return 1.0 / (1.0 + math.exp(-steepness * (value - 0.5)))

    def __init__(self, half_life_hours: float = 168.0) -> None:
        if half_life_hours <= 0.0:
            raise ValueError(
                f"half_life_hours must be positive, got {half_life_hours}"
            )
        self._half_life_hours = half_life_hours

    def score(
        self,
        timestamp: datetime,
        reference: datetime | None = None,
    ) -> float:
        """Compute the recency score for a timestamp.

        Parameters
        ----------
        timestamp:
            When the item was created or last updated.
        reference:
            The reference point for age computation.  Defaults to the
            current UTC time when ``None``.

        Returns
        -------
        float
            Score in ``[0.0, 1.0]``.  Items in the future receive 1.0.
        """
        ref = reference if reference is not None else datetime.now(timezone.utc)
        ts_utc = _to_utc(timestamp)
        ref_utc = _to_utc(ref)

        age_seconds = (ref_utc - ts_utc).total_seconds()
        if age_seconds <= 0.0:
            # Future or same-time item — maximum freshness
            return 1.0

        age_hours = age_seconds / 3600.0
        raw = math.pow(2.0, -age_hours / self._half_life_hours)
        return round(min(1.0

    def rank(
        self,
        items: list[tuple[str, datetime]],
        reference: datetime | None = None,
    ) -> list[RetrievalResult]:
        """Score and rank a list of ``(content, timestamp)`` pairs.

        Parameters
        ----------
        items:
            Each element is ``(content, timestamp)`` where ``content``
            is the text of the memory item and ``timestamp`` is when it
            was created.
        reference:
            Reference time for age computation.  Defaults to now.

        Returns
        -------
        list[RetrievalResult]
            Results sorted by recency score descending.  Each entry has
            ``source="recency"``.
        """
        ref = reference if reference is not None else datetime.now(timezone.utc)
        results: list[RetrievalResult] = [
            RetrievalResult(
                content=content,
                source="recency",
                score=self.score(ts, reference=ref),
                metadata=

    def __init__(
        self,
        stores: Optional[dict[MemoryLayer, MemoryStore]] = None,
        ranker: Optional[object] = None,
        default_limit: int = 20,
    ) -> None:
        self._stores: dict[MemoryLayer, MemoryStore] = stores or {}
        self._ranker = ranker
        self._default_limit = default_limit

    def search(
        self,
        query: str,
        layer: Optional[MemoryLayer] = None,
        limit: Optional[int] = None,
        min_importance: float = 0.0,
        min_freshness: float = 0.0,
    ) -> list[SearchResult]:
        """Search for entries matching the query.

        Parameters
        ----------
        query:
            The free-text search query.
        layer:
            If provided, only the store for that layer is searched.
        limit:
            Maximum results to return; falls back to ``default_limit``.
        min_importance:
            Exclude entries with ``importance_score`` below this value.
        min_freshness:
            Exclude entries with ``freshness_score`` below this value.

        Returns
        -------
        list[SearchResult]
            Results sorted by score descending.
        """
        effective_limit = limit if limit is not None else self._default_limit
        raw_entries: dict[str, tuple[MemoryEntry, MemoryLayer]] 

    def search_all(
        self,
        query: str,
        limit: Optional[int] = None,
    ) -> list[SearchResult]:
        """Convenience method: search all layers without filters."""
        return self.search(query, layer=None, limit=limit)

    def get_by_layer(
        self,
        layer: MemoryLayer,
        limit: int = 100,
    ) -> list[MemoryEntry]:
        """Return all entries from a given layer, up to ``limit``."""
        store = self._stores.get(layer)
        if store is None:
            return []
        return list(store.all(layer=layer))[:limit]

    def _rank_results(
        self,
        entries: list[tuple[MemoryEntry, MemoryLayer]],
        query: str,
    ) -> list[SearchResult]:
        """Score and wrap entries as SearchResult objects."""
        from agent_memory.retrieval.ranking import ResultRanker

        ranker: ResultRanker = self._ranker or ResultRanker()  # type: ignore[assignment]
        results: list[SearchResult] = []
        for rank_idx, (entry, matched_layer) in enumerate(entries):
            score = ranker.score(entry, query)
            results.append(
                SearchResult(
                    entry=entry,
                    score=score,
                    rank=rank_idx + 1,
                    matched_layer=matched_layer,
                    query=query,
                )
            )
        return results

    def do_GET(self) -> None:
        """Handle all GET requests by routing on the URL path."""
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path.rstrip("/")
        query_params = urllib.parse.parse_qs(parsed.query)

        if path == "/health":
            status, data = routes.handle_health()
            self._send_json(status, data)

        elif path == "/search":
            query = self._first_param(query_params, "q") or ""
            layer = self._first_param(query_params, "layer")
            limit_str = self._first_param(query_params, "limit")
            limit = int(limit_str) if limit_str and limit_str.isdigit() else 20
            status, data = routes.handle_search_memories(
                query=query, layer=layer, limit=limit
            )
            self._send_json(status, data)

        else:
            match = _MEMORY_ID_PATTERN.match(path)
            if match:
                memory_id = match.group(1)
                status, data = r

    def do_POST(self) -> None:
        """Handle all POST requests by routing on the URL path."""
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path.rstrip("/")

        body = self._read_json_body()
        if body is None:
            return

        if path == "/memories":
            status, data = routes.handle_create_memory(body)
            self._send_json(status, data)
        else:
            self._send_json(
                404, {"error": "Not found", "detail": f"No route for POST {path}"}
            )

    def do_DELETE(self) -> None:
        """Handle all DELETE requests by routing on the URL path."""
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path.rstrip("/")

        match = _MEMORY_ID_PATTERN.match(path)
        if match:
            memory_id = match.group(1)
            status, data = routes.handle_delete_memory(memory_id)
            self._send_json(status, data)
        else:
            self._send_json(
                404,
                {"error": "Not found", "detail": f"No route for DELETE {path}"},
            )

    def _send_json(self, status: int, data: dict[str, object]) -> None:
        """Serialize *data* to JSON and send an HTTP response with *status*."""
        body = json.dumps(data, default=str).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def _read_json_body(self) -> dict[str, object] | None:
        """Read and parse the JSON request body.

        Returns None (and sends a 400 error response) if parsing fails.
        """
        content_length = int(self.headers.get("Content-Length", 0))
        if content_length == 0:
            return {}

        raw = self.rfile.read(content_length)
        try:
            parsed: dict[str, object] = json.loads(raw.decode("utf-8"))
            return parsed
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            self._send_json(400, {"error": "Invalid JSON", "detail": str(exc)})
            return None

    def _first_param(
        params: dict[str, list[str]], key: str
    ) -> str | None:
        """Return the first value for *key* from query parameters, or None."""
        values = params.get(key)
        return values[0] if values else None

def create_server(host: str = "0.0.0.0", port: int = 8080) -> HTTPServer:
    """Create (but do not start) the agent-memory HTTP server.

    Parameters
    ----------
    host:
        Bind address (default ``"0.0.0.0"`` — all interfaces).
    port:
        TCP port to listen on (default 8080).

    Returns
    -------
    HTTPServer
        A configured server instance ready to call ``serve_forever()`` on.
    """
    server = HTTPServer((host, port), AgentMemoryHandler)
    logger.info("agent-memory server created at http://%s:%d", host, port)
    return server

def run_server(host: str = "0.0.0.0", port: int = 8080) -> None:
    """Create and run the agent-memory HTTP server (blocking).

    Parameters
    ----------
    host:
        Bind address.
    port:
        TCP port.
    """
    server = create_server(host=host, port=port)
    logger.info("Serving agent-memory on http://%s:%d — press Ctrl-C to stop", host, port)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        logger.info("Shutting down agent-memory server.")
    finally:
        server.server_close()

def _build_arg_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="agent-memory HTTP server",
        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
    )
    parser.add_argument("--host", default="0.0.0.0", help="Bind address")
    parser.add_argument("--port", type=int, default=8080, help="TCP port")
    parser.add_argument(
        "--log-level",
        default="INFO",
        choices=["DEBUG", "INFO", "WARNING", "ERROR"],
        help="Logging level",
    )
    return parser

def _entry_to_response(entry: MemoryEntry) -> MemoryResponse:
    """Convert a MemoryEntry to a MemoryResponse."""
    return MemoryResponse(
        memory_id=entry.memory_id,
        content=entry.content,
        layer=entry.layer.value,
        importance_score=entry.importance_score,
        freshness_score=entry.freshness_score,
        source=entry.source.value,
        safety_critical=entry.safety_critical,
        composite_score=entry.composite_score,
        importance_level=entry.importance_level.value,
        created_at=entry.created_at.isoformat(),
        last_accessed=entry.last_accessed.isoformat(),
        access_count=entry.access_count,
        metadata=entry.metadata,
    )

def handle_create_memory(body: dict[str, object]) -> tuple[int, dict[str, object]]:
    """Handle POST /memories.

    Parameters
    ----------
    body:
        Parsed JSON request body.

    Returns
    -------
    tuple[int, dict[str, object]]
        HTTP status code and response dictionary.
    """
    try:
        request = CreateMemoryRequest.model_validate(body)
    except Exception as exc:
        return 422, ErrorResponse(error="Validation error", detail=str(exc)).model_dump()

    try:
        layer = MemoryLayer(request.layer)
    except ValueError:
        return 422, ErrorResponse(
            error="Invalid layer",
            detail=f"Layer {request.layer!r} is not valid. "
            f"Choose from: {[l.value for l in MemoryLayer]}",
        ).model_dump()

    try:
        source = MemorySource(request.source)
    except ValueError:
        source = MemorySource.AGENT_INFERENCE

    entry = MemoryEntry(
        content=request.content,
        layer=layer,
        im

def handle_search_memories(
    query: str,
    layer: str | None = None,
    limit: int = 20,
) -> tuple[int, dict[str, object]]:
    """Handle GET /search?q=...

    Performs case-insensitive substring matching against memory content.

    Parameters
    ----------
    query:
        Search term.
    layer:
        Optional layer filter.
    limit:
        Maximum results to return.

    Returns
    -------
    tuple[int, dict[str, object]]
        HTTP status code and response dictionary.
    """
    query_lower = query.lower()
    results: list[MemoryEntry] = []

    for entry in _memory_store.values():
        if query_lower and query_lower not in entry.content.lower():
            continue
        if layer and entry.layer.value != layer:
            continue
        results.append(entry)

    # Sort by composite_score descending
    results.sort(key=lambda e: e.composite_score, reverse=True)
    results = results[:limit]

    response = SearchResponse(
        query=query,
      

def handle_delete_memory(memory_id: str) -> tuple[int, dict[str, object]]:
    """Handle DELETE /memories/{id}.

    Parameters
    ----------
    memory_id:
        The memory identifier from the URL path.

    Returns
    -------
    tuple[int, dict[str, object]]
        HTTP status code and response dictionary.
    """
    if memory_id not in _memory_store:
        return 404, ErrorResponse(
            error="Not found",
            detail=f"Memory {memory_id!r} not found.",
        ).model_dump()

    del _memory_store[memory_id]
    response = DeleteMemoryResponse(memory_id=memory_id, deleted=True)
    return 200, response.model_dump()

def handle_get_memory(memory_id: str) -> tuple[int, dict[str, object]]:
    """Handle GET /memories/{id}.

    Parameters
    ----------
    memory_id:
        The memory identifier.

    Returns
    -------
    tuple[int, dict[str, object]]
        HTTP status code and response dictionary.
    """
    entry = _memory_store.get(memory_id)
    if entry is None:
        return 404, ErrorResponse(
            error="Not found",
            detail=f"Memory {memory_id!r} not found.",
        ).model_dump()

    # Touch access count
    entry = entry.touch()
    _memory_store[memory_id] = entry

    return 200, _entry_to_response(entry).model_dump()

def handle_health() -> tuple[int, dict[str, object]]:
    """Handle GET /health.

    Returns
    -------
    tuple[int, dict[str, object]]
        HTTP status code and response dictionary.
    """
    response = HealthResponse(memory_count=len(_memory_store))
    return 200, response.model_dump()

    def save(self, entry: MemoryEntry) -> None:
        """Persist a memory entry, replacing any existing entry with the same key.

        Parameters
        ----------
        entry:
            The entry to persist. ``entry.memory_id`` is used as the key.
        """

    def load(self, key: str) -> Optional[MemoryEntry]:
        """Load an entry by its key.

        Parameters
        ----------
        key:
            The ``memory_id`` of the entry to load.

        Returns
        -------
        MemoryEntry | None
            The entry, or None if no entry with that key exists.
        """

    def delete(self, key: str) -> bool:
        """Remove an entry by key.

        Parameters
        ----------
        key:
            The ``memory_id`` to remove.

        Returns
        -------
        bool
            True if the entry existed and was deleted, False otherwise.
        """

    def search(
        self,
        query: str,
        layer: Optional[MemoryLayer] = None,
        limit: int = 20,
    ) -> Sequence[MemoryEntry]:
        """Search stored entries by content.

        Parameters
        ----------
        query:
            The search query string.
        layer:
            Optional layer filter.
        limit:
            Maximum number of results to return.

        Returns
        -------
        Sequence[MemoryEntry]
            Matching entries, ordered by relevance (implementation-defined).
        """

    def list_keys(
        self,
        layer: Optional[MemoryLayer] = None,
        limit: int = 1000,
    ) -> list[str]:
        """List stored entry keys.

        Parameters
        ----------
        layer:
            Optional layer filter. If provided, only keys for that layer
            are returned.
        limit:
            Maximum number of keys to return.

        Returns
        -------
        list[str]
            Entry keys (``memory_id`` values), in an implementation-defined
            order.
        """

    def clear(self, layer: Optional[MemoryLayer] = None) -> int:
        """Remove all stored entries, optionally filtered by layer.

        Parameters
        ----------
        layer:
            If provided, only entries in this layer are removed.
            If None, all entries are removed.

        Returns
        -------
        int
            Number of entries deleted.
        """

    def count(self, layer: Optional[MemoryLayer] = None) -> int:
        """Return the number of stored entries.

        The default implementation uses ``list_keys``; backends with
        native count operations should override this for efficiency.
        """
        return len(self.list_keys(layer=layer))

    def load_all(
        self,
        layer: Optional[MemoryLayer] = None,
        limit: int = 1000,
    ) -> list[MemoryEntry]:
        """Load multiple entries at once.

        The default implementation calls ``load`` for each key from
        ``list_keys``. Backends may override this for batch efficiency.

        Parameters
        ----------
        layer:
            Optional layer filter.
        limit:
            Maximum number of entries to return.

        Returns
        -------
        list[MemoryEntry]
            Loaded entries (None results are omitted).
        """
        keys = self.list_keys(layer=layer, limit=limit)
        results: list[MemoryEntry] = []
        for key in keys:
            entry = self.load(key)
            if entry is not None:
                results.append(entry)
        return results