Function bodies 227 total

def _task_state_from_dict(data: dict[str, object]) -> USFTaskState:
    return USFTaskState(
        task_id=data["task_id"],
        status=data["status"],
        progress=float(data["progress"]),
        result=data.get("result"),
    )

    def _canonical_dict(self) -> dict[str, object]:
        """Return a stable, JSON-serialisable dict for checksum computation.

        The ``checksum`` field is excluded to avoid circularity.
        """
        return {
            "version": self.version,
            "session_id": self.session_id,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
            "framework_source": self.framework_source,
            "messages": [_message_to_dict(m) for m in self.messages],
            "working_memory": self.working_memory,
            "entities": [_entity_to_dict(e) for e in self.entities],
            "task_state": [_task_state_to_dict(t) for t in self.task_state],
            "metadata": self.metadata,
        }

    def compute_checksum(self) -> str:
        """Compute and store a SHA-256 checksum of this session's content.

        The checksum is deterministic: given the same field values in the same
        order, the same digest is always produced.

        Returns
        -------
        str
            64-character lowercase hex SHA-256 digest.
        """
        canonical_json = json.dumps(self._canonical_dict(), sort_keys=True)
        digest = hashlib.sha256(canonical_json.encode()).hexdigest()
        self.checksum = digest
        return digest

    def verify_checksum(self) -> bool:
        """Return True when the stored checksum matches the recomputed one.

        Returns
        -------
        bool
            True if the session has not been modified since the last
            call to :meth:`compute_checksum`.
        """
        stored = self.checksum
        recomputed = self.compute_checksum()
        # Restore original so calling verify_checksum does not mutate state
        self.checksum = stored
        return stored == recomputed

    def to_json(self) -> str:
        """Serialise the session to a JSON string.

        The checksum is refreshed before serialisation to ensure it reflects
        the current content.

        Returns
        -------
        str
            A valid JSON string representing the full session.
        """
        self.compute_checksum()
        payload: dict[str, object] = self._canonical_dict()
        payload["checksum"] = self.checksum
        return json.dumps(payload, sort_keys=True, default=str)

    def from_json(cls, json_str: str) -> "UniversalSession":
        """Deserialise a session from a JSON string produced by :meth:`to_json`.

        Parameters
        ----------
        json_str:
            JSON string previously produced by :meth:`to_json`.

        Returns
        -------
        UniversalSession
            The reconstructed session.

        Raises
        ------
        ValueError
            If ``json_str`` is not valid JSON or is missing required fields.
        """
        try:
            data: dict[str, object] = json.loads(json_str)
        except json.JSONDecodeError as exc:
            raise ValueError(f"Invalid JSON: {exc}") from exc

        stored_checksum = data.pop("checksum", "")

        messages = [_message_from_dict(m) for m in data.get("messages", [])]
        entities = [_entity_from_dict(e) for e in data.get("entities", [])]
        task_state = [_task_state_from_dict(t) for t in data.get("task_state", [])]

        created_raw = data.get("

    def to_dict(self) -> dict[str, object]:
        """Serialise to a plain dict."""
        return {
            "segment_id": self.segment_id,
            "segment_type": self.segment_type.value,
            "importance_score": self.importance_score,
            "token_count": self.token_count,
            "content_preview": self.content_preview,
            "metadata": dict(self.metadata),
        }

    def score_segment(
        self,
        segment_id: str,
        segment_type: SegmentType,
        content: str,
        token_count: int,
        *,
        recency_rank: float = 0.0,
        metadata: dict[str, object] | None = None,
    ) -> ScoredSegment:
        """Compute the importance score for a single segment.

        Parameters
        ----------
        segment_id:
            Unique identifier for the segment.
        segment_type:
            The classified type of this segment.
        content:
            Full text content (used for keyword detection).
        token_count:
            Estimated token count.
        recency_rank:
            A value in [0.0, 1.0] where 1.0 means the most recent segment.
            Used to apply the recency boost.
        metadata:
            Optional extra data to carry through.

        Returns
        -------
        ScoredSegment
            Scored segment.
        """
        # Base score from type prior
        prior = _TYP

    def score_segments(
        self,
        segments: list[dict[str, object]],
    ) -> list[ScoredSegment]:
        """Score a list of raw segment dicts.

        Each dict must have the following keys:
        - ``segment_id`` (str)
        - ``segment_type`` (str — a SegmentType value)
        - ``content`` (str)
        - ``token_count`` (int)

        Optional keys:
        - ``metadata`` (dict)

        Segments are assumed to be in chronological order (oldest first).
        The most recent segment receives recency_rank=1.0.

        Parameters
        ----------
        segments:
            List of raw segment dicts in chronological order.

        Returns
        -------
        list[ScoredSegment]
            Scored segments in the same order as input.
        """
        if not segments:
            return []

        total = len(segments)
        scored: list[ScoredSegment] = []

        for index, seg in enumerate(segments):
            segment_id = str(seg.get("segment

    def rank_by_importance(
        self,
        segments: list[dict[str, object]],
    ) -> list[ScoredSegment]:
        """Score segments and return them sorted by descending importance.

        Parameters
        ----------
        segments:
            List of raw segment dicts.

        Returns
        -------
        list[ScoredSegment]
            Scored segments sorted highest importance first.
        """
        scored = self.score_segments(segments)
        return sorted(scored, key=lambda s: s.importance_score, reverse=True)

    def type_prior(self, segment_type: SegmentType) -> float:
        """Return the base importance prior for a given segment type.

        Parameters
        ----------
        segment_type:
            The segment type to look up.

        Returns
        -------
        float
            Base importance prior in [0.0, 1.0].
        """
        return _TYPE_PRIORS.get(segment_type, 0.40)

    def __init__(self, config: SegmentClassifierConfig | None = None) -> None:
        self._config = config if config is not None else SegmentClassifierConfig()
        rules = (
            self._config.rules if self._config.rules else list(_DEFAULT_RULES)
        )
        self._rules: list[ClassificationRule] = sorted(
            rules, key=lambda r: r.priority
        )
        self._compiled: dict[str, re.Pattern[str]] = {}
        for rule in self._rules:
            if rule.content_pattern and rule.content_pattern not in self._compiled:
                self._compiled[rule.content_pattern] = re.compile(
                    rule.content_pattern, re.IGNORECASE
                )

    def classify(
        self,
        content: str,
        metadata: dict[str, object] | None = None,
    ) -> SegmentType:
        """Classify a single segment into a SegmentType.

        Parameters
        ----------
        content:
            The text content of the segment.
        metadata:
            Optional segment metadata dict.  Examined for field-based rules.

        Returns
        -------
        SegmentType
            The classified segment type.
        """
        meta = metadata or {}

        # Trust explicit type if present and valid
        if self._config.trust_existing_type:
            existing = str(meta.get("segment_type", "")).strip()
            if existing:
                try:
                    return SegmentType(existing)
                except ValueError:
                    pass  # fall through to rules

        for rule in self._rules:
            # When trust_existing_type is False, skip field-based rules that
            # examine the "segm

    def classify_batch(
        self,
        segments: list[dict[str, object]],
    ) -> list[SegmentType]:
        """Classify a list of raw segment dicts.

        Each dict should have a ``content`` key.  A ``metadata`` key is
        optional.

        Parameters
        ----------
        segments:
            List of raw segment dicts.

        Returns
        -------
        list[SegmentType]
            Classification for each segment in input order.
        """
        results: list[SegmentType] = []
        for seg in segments:
            content = str(seg.get("content", ""))
            metadata = dict(seg.get("metadata", {}))
            # If segment_type is a top-level key, pass it via metadata
            for key in ("segment_type", "role"):
                if key in seg and key not in metadata:
                    metadata[key] = str(seg[key])
            results.append(self.classify(content, metadata))
        return results

    def annotate(
        self,
        segments: list[dict[str, object]],
    ) -> list[dict[str, object]]:
        """Classify segments and annotate each with a ``segment_type`` field.

        The original dict is not mutated; a new dict is returned for each
        segment with the ``segment_type`` key set.

        Parameters
        ----------
        segments:
            List of raw segment dicts.

        Returns
        -------
        list[dict[str, Any]]
            Copies of input dicts with ``segment_type`` filled in.
        """
        annotated: list[dict[str, object]] = []
        for seg in segments:
            content = str(seg.get("content", ""))
            metadata = dict(seg.get("metadata", {}))
            for key in ("segment_type", "role"):
                if key in seg and key not in metadata:
                    metadata[key] = str(seg[key])
            classified = self.classify(content, metadata)
            new_seg = dict(seg)
            new_seg["segme

    def _matches_rule(
        self,
        rule: ClassificationRule,
        content: str,
        metadata: dict[str, object],
    ) -> bool:
        """Evaluate whether a rule matches the given segment.

        A rule matches when at least one of its conditions is defined and
        satisfied.  If a rule has both ``field_name`` and ``content_pattern``,
        either matching is sufficient (OR semantics).

        Parameters
        ----------
        rule:
            The rule to evaluate.
        content:
            Segment content text.
        metadata:
            Segment metadata dict.

        Returns
        -------
        bool
            True when the rule matches.
        """
        has_condition = False

        if rule.field_name and rule.field_value:
            has_condition = True
            field_val = str(metadata.get(rule.field_name, "")).lower()
            if rule.field_value.lower() in field_val:
                return True

        if rule.content_patte

    def to_dict(self) -> dict[str, object]:
        """Serialise the result summary (without full segment content)."""
        return {
            "selected_count": len(self.selected_segments),
            "total_tokens_loaded": self.total_tokens_loaded,
            "total_tokens_available": self.total_tokens_available,
            "segments_considered": self.segments_considered,
            "segments_skipped": self.segments_skipped,
            "budget_used_pct": self.budget_used_pct,
        }

    def __init__(
        self,
        config: SelectiveLoaderConfig | None = None,
        scorer_config: ImportanceScorerConfig | None = None,
    ) -> None:
        self._config = config if config is not None else SelectiveLoaderConfig()
        self._scorer = ImportanceScorer(scorer_config)

    def load(self, segments: list[dict[str, object]]) -> LoadResult:
        """Select segments within the configured token budget.

        Parameters
        ----------
        segments:
            Raw segment dicts in chronological order.  Each dict requires:
            ``segment_id``, ``segment_type``, ``content``, ``token_count``.

        Returns
        -------
        LoadResult
            Selection result with metadata.
        """
        if not segments:
            return LoadResult(
                selected_segments=[],
                total_tokens_loaded=0,
                total_tokens_available=0,
                segments_considered=0,
                segments_skipped=0,
                budget_used_pct=0.0,
            )

        # Score all segments, preserving original index for order restoration.
        scored_with_index: list[tuple[int, ScoredSegment]] = []
        for original_index, raw in enumerate(segments):
            scored_list = self._scorer.score_segmen

    def load_scored(self, scored_segments: list[ScoredSegment]) -> LoadResult:
        """Load from pre-scored segments (skips the scoring step).

        Parameters
        ----------
        scored_segments:
            Already-scored segments in chronological order.

        Returns
        -------
        LoadResult
            Selection result.
        """
        raw = [
            {
                "segment_id": s.segment_id,
                "segment_type": s.segment_type.value,
                "content": s.content_preview,
                "token_count": s.token_count,
                "metadata": s.metadata,
            }
            for s in scored_segments
        ]
        return self.load(raw)

    def __init__(
        self,
        backend: StorageBackend,
        serializer: SessionSerializer | None = None,
        default_agent_id: str = "default",
    ) -> None:
        self._backend = backend
        self._serializer = serializer or SessionSerializer()
        self._default_agent_id = default_agent_id

    def create_session(
        self,
        agent_id: str | None = None,
        *,
        parent_session_id: str | None = None,
        preferences: dict[str, str] | None = None,
    ) -> SessionState:
        """Create and return a new in-memory session (not yet persisted).

        Call ``save_session`` to persist the returned state.

        Parameters
        ----------
        agent_id:
            Override the manager's ``default_agent_id`` for this session.
        parent_session_id:
            Link this session to a prior session for chain navigation.
        preferences:
            Initial preference key-value pairs.

        Returns
        -------
        SessionState
            A freshly initialised, unpersisted session.
        """
        state = SessionState(
            agent_id=agent_id or self._default_agent_id,
            parent_session_id=parent_session_id,
            preferences=preferences or {},
        )
        return state

    def save_session(self, state: SessionState) -> str:
        """Persist a session to the storage backend.

        ``updated_at`` is refreshed and a fresh checksum is embedded before
        serialisation.

        Parameters
        ----------
        state:
            The session state to persist.

        Returns
        -------
        str
            The ``session_id`` under which the data was saved.
        """
        state.updated_at = datetime.now(timezone.utc)
        raw = self._serializer.to_json(state)
        self._backend.save(state.session_id, raw)
        return state.session_id

    def load_session(self, session_id: str) -> SessionState:
        """Load and return a session from the storage backend.

        Parameters
        ----------
        session_id:
            The session to retrieve.

        Returns
        -------
        SessionState
            The deserialised and checksum-verified session.

        Raises
        ------
        SessionNotFoundError
            If no session with ``session_id`` exists in the backend.
        """
        if not self._backend.exists(session_id):
            raise SessionNotFoundError(session_id)
        raw = self._backend.load(session_id)
        return self._serializer.from_json(raw)

    def delete_session(self, session_id: str) -> None:
        """Remove a session from the storage backend.

        Parameters
        ----------
        session_id:
            The session to delete.

        Raises
        ------
        SessionNotFoundError
            If no session with ``session_id`` exists.
        """
        if not self._backend.exists(session_id):
            raise SessionNotFoundError(session_id)
        self._backend.delete(session_id)

    def session_exists(self, session_id: str) -> bool:
        """Return True if ``session_id`` is present in the backend.

        Parameters
        ----------
        session_id:
            The session identifier to check.

        Returns
        -------
        bool
        """
        return self._backend.exists(session_id)

    def list_sessions(self) -> list[str]:
        """Return the IDs of all sessions in the backend.

        Returns
        -------
        list[str]
            Sorted list of session IDs.
        """
        return sorted(self._backend.list())

    def list_sessions_for_agent(self, agent_id: str) -> list[str]:
        """Return session IDs belonging to ``agent_id``.

        This performs a full scan: each session is loaded and filtered.
        For large collections prefer a backend that supports indexed queries.

        Parameters
        ----------
        agent_id:
            The agent identifier to filter by.

        Returns
        -------
        list[str]
            Session IDs (sorted) whose ``agent_id`` matches.
        """
        matching: list[str] = []
        for session_id in self._backend.list():
            try:
                state = self.load_session(session_id)
                if state.agent_id == agent_id:
                    matching.append(session_id)
            except Exception:  # noqa: BLE001 — skip corrupt / unreadable entries
                continue
        return sorted(matching)

    def continue_session(self, parent_session_id: str) -> SessionState:
        """Load ``parent_session_id`` and create a child continuation session.

        The child session has ``parent_session_id`` set and inherits the
        parent's ``agent_id``, ``preferences``, active tasks, and entities.

        Parameters
        ----------
        parent_session_id:
            The session to continue from.

        Returns
        -------
        SessionState
            A new unpersisted session linked to the parent.
        """
        parent = self.load_session(parent_session_id)
        child = SessionState(
            agent_id=parent.agent_id,
            parent_session_id=parent_session_id,
            preferences=dict(parent.preferences),
            entities=list(parent.entities),
            tasks=[t for t in parent.tasks if t.status.value in ("pending", "in_progress")],
        )
        return child

    def get_stats(self) -> dict[str, object]:
        """Return aggregate statistics across all sessions in the backend.

        The returned dict includes:
        - ``total_sessions`` — count of stored sessions
        - ``total_segments`` — sum of segment counts
        - ``total_tokens`` — sum of token counts
        - ``total_tasks`` — sum of task counts
        - ``total_entities`` — sum of entity counts
        - ``total_cost_usd`` — summed cost across all sessions
        - ``agents`` — list of unique agent IDs

        Returns
        -------
        dict[str, object]
            Aggregate statistics dictionary.
        """
        total_sessions = 0
        total_segments = 0
        total_tokens = 0
        total_tasks = 0
        total_entities = 0
        total_cost: float = 0.0
        agent_ids: set[str] = set()

        for session_id in self._backend.list():
            try:
                state = self.load_session(session_id)
                total_sessions += 1
    

    def __init__(self, version: str) -> None:
        self.version = version
        supported = ", ".join(sorted(_SUPPORTED_SCHEMA_VERSIONS))
        super().__init__(
            f"Unsupported schema version {version!r}. "
            f"Supported versions: {supported}"
        )

    def to_json(self, state: SessionState, *, indent: int = 2) -> str:
        """Serialise a ``SessionState`` to a JSON string.

        A fresh checksum is computed and embedded before serialisation.

        Parameters
        ----------
        state:
            The session to serialise.
        indent:
            JSON indentation level (default 2).

        Returns
        -------
        str
            JSON-encoded document with schema_version and checksum fields.
        """
        state.compute_checksum()
        data = state.model_dump(mode="json")
        return json.dumps(data, indent=indent, default=str)

    def from_json(self, raw: str) -> SessionState:
        """Deserialize a ``SessionState`` from a JSON string.

        Parameters
        ----------
        raw:
            JSON string previously produced by ``to_json``.

        Returns
        -------
        SessionState
            The reconstructed session.

        Raises
        ------
        SchemaVersionError
            If the ``schema_version`` field is not in the supported set.
        ValueError
            If ``validate_checksum`` is True and the checksum does not match.
        json.JSONDecodeError
            If ``raw`` is not valid JSON.
        """
        data: dict[str, object] = json.loads(raw)
        return self._deserialize(data)

    def to_yaml(self, state: SessionState) -> str:
        """Serialise a ``SessionState`` to a YAML string.

        Parameters
        ----------
        state:
            The session to serialise.

        Returns
        -------
        str
            YAML-encoded document.
        """
        state.compute_checksum()
        data = state.model_dump(mode="json")
        return yaml.dump(data, default_flow_style=False, allow_unicode=True, sort_keys=True)

    def from_yaml(self, raw: str) -> SessionState:
        """Deserialize a ``SessionState`` from a YAML string.

        Parameters
        ----------
        raw:
            YAML string previously produced by ``to_yaml``.

        Returns
        -------
        SessionState
            The reconstructed session.

        Raises
        ------
        SchemaVersionError
            If the ``schema_version`` field is not in the supported set.
        ValueError
            If ``validate_checksum`` is True and the checksum does not match.
        """
        data: dict[str, object] = yaml.safe_load(raw)
        return self._deserialize(data)

    def serialize(
        self, state: SessionState, format: Literal["json", "yaml"] = "json"
    ) -> str:
        """Serialize using the named format.

        Parameters
        ----------
        state:
            Session to serialize.
        format:
            Either ``"json"`` (default) or ``"yaml"``.

        Returns
        -------
        str
            Serialized document string.
        """
        if format == "yaml":
            return self.to_yaml(state)
        return self.to_json(state)

    def deserialize(
        self, raw: str, format: Literal["json", "yaml"] = "json"
    ) -> SessionState:
        """Deserialize using the named format.

        Parameters
        ----------
        raw:
            Serialized document string.
        format:
            Either ``"json"`` (default) or ``"yaml"``.

        Returns
        -------
        SessionState
            Reconstructed session.
        """
        if format == "yaml":
            return self.from_yaml(raw)
        return self.from_json(raw)

    def _deserialize(self, data: dict[str, object]) -> SessionState:
        """Common deserialization path for both JSON and YAML.

        Parameters
        ----------
        data:
            Dictionary decoded from the serialized document.

        Returns
        -------
        SessionState
            Validated and (optionally checksum-verified) session state.
        """
        version = str(data.get("schema_version", ""))
        if version not in _SUPPORTED_SCHEMA_VERSIONS:
            raise SchemaVersionError(version)

        state = SessionState.model_validate(data)

        if self.validate_checksum and state.checksum:
            stored = state.checksum
            computed = state.compute_checksum()
            if stored != computed:
                raise ValueError(
                    f"Checksum mismatch for session {state.session_id!r}: "
                    f"stored={stored!r} computed={computed!r}"
                )

        return state

    def _canonical_dict(self) -> dict[str, object]:
        """Return a stable dict suitable for checksum computation.

        The ``checksum`` field itself is excluded to avoid circularity.
        """
        data = self.model_dump(mode="json")
        data.pop("checksum", None)
        return data  # type: ignore[return-value]

    def compute_checksum(self) -> str:
        """Compute and return the SHA-256 checksum of this session's content.

        The result is stored in ``self.checksum`` as a side-effect and also
        returned as a hex-encoded string.

        Returns
        -------
        str
            64-character lowercase hex SHA-256 digest.
        """
        canonical_json = json.dumps(self._canonical_dict(), sort_keys=True)
        digest = hashlib.sha256(canonical_json.encode()).hexdigest()
        self.checksum = digest
        return digest

    def verify_checksum(self) -> bool:
        """Return True if the stored checksum matches the computed one.

        Returns
        -------
        bool
            True when the session has not been tampered with after the
            last call to ``compute_checksum``.
        """
        return self.checksum == self.compute_checksum()

    def add_segment(
        self,
        role: str,
        content: str,
        *,
        token_count: int = 0,
        segment_type: str = "conversation",
        metadata: dict[str, str] | None = None,
    ) -> ContextSegment:
        """Append a new context segment and return it.

        Parameters
        ----------
        role:
            Message role: "user", "assistant", "system", or "tool".
        content:
            The text content of the segment.
        token_count:
            Estimated tokens consumed by this segment.
        segment_type:
            Categorical label (e.g. "conversation", "code", "plan").
        metadata:
            Optional additional key-value data.

        Returns
        -------
        ContextSegment
            The newly created and appended segment.
        """
        turn_index = len(self.segments)
        segment = ContextSegment(
            role=role,
            content=content,
            token_count=token_count,
            

    def track_entity(
        self,
        canonical_name: str,
        entity_type: str = "concept",
        *,
        aliases: list[str] | None = None,
        attributes: dict[str, str] | None = None,
        confidence: float = 1.0,
    ) -> EntityReference:
        """Add or update an entity reference in this session.

        If an entity with the same ``canonical_name`` already exists it is
        returned unchanged.  Otherwise a new ``EntityReference`` is created.

        Parameters
        ----------
        canonical_name:
            Primary name used for deduplication and matching.
        entity_type:
            Categorical label.
        aliases:
            Alternative names for this entity.
        attributes:
            Key-value descriptive attributes.
        confidence:
            Extraction confidence in [0.0, 1.0].

        Returns
        -------
        EntityReference
            The existing or newly created entity reference.
        """
        for exi

    def update_task(
        self,
        task_id: str,
        *,
        status: TaskStatus | None = None,
        notes: str | None = None,
        priority: int | None = None,
    ) -> TaskState:
        """Update an existing task by ID.

        Parameters
        ----------
        task_id:
            The ``task_id`` of the task to update.
        status:
            New status value, if changing.
        notes:
            Appended to existing notes separated by a newline.
        priority:
            New priority value.

        Returns
        -------
        TaskState
            The updated task.

        Raises
        ------
        KeyError
            If no task with ``task_id`` exists in this session.
        """
        for task in self.tasks:
            if task.task_id == task_id:
                if status is not None:
                    task.status = status
                if notes is not None:
                    task.notes = f"{task.notes}\n{notes}".strip()
  

    def add_task(
        self,
        title: str,
        *,
        description: str = "",
        priority: int = 5,
        tags: list[str] | None = None,
        parent_task_id: str | None = None,
    ) -> TaskState:
        """Create and append a new task to this session.

        Parameters
        ----------
        title:
            Short human-readable title.
        description:
            Detailed description.
        priority:
            Priority level 1 (highest) to 10 (lowest).
        tags:
            Optional categorisation labels.
        parent_task_id:
            Optional parent task for hierarchy.

        Returns
        -------
        TaskState
            The newly created task.
        """
        task = TaskState(
            title=title,
            description=description,
            priority=priority,
            tags=tags or [],
            parent_task_id=parent_task_id,
        )
        self.tasks.append(task)
        self.updated_at = datetime.n

    def total_tokens(self) -> int:
        """Return the sum of token_count across all segments.

        Returns
        -------
        int
            Total estimated token count for the session.
        """
        return sum(segment.token_count for segment in self.segments)

    def save(self, session_id: str, payload: str) -> None:
        """Persist ``payload`` under ``session_id``.

        If a document already exists for ``session_id`` it is overwritten.

        Parameters
        ----------
        session_id:
            Unique session identifier used as the storage key.
        payload:
            UTF-8 string to persist (typically JSON).
        """

    def load(self, session_id: str) -> str:
        """Return the raw payload stored under ``session_id``.

        Parameters
        ----------
        session_id:
            The session to retrieve.

        Returns
        -------
        str
            The previously saved payload string.

        Raises
        ------
        KeyError
            If no entry exists for ``session_id``.
        """

    def list(self) -> list[str]:
        """Return a list of all stored session IDs.

        Returns
        -------
        list[str]
            All session IDs currently in this backend.  Order is
            implementation-defined.
        """

    def delete(self, session_id: str) -> None:
        """Remove the entry for ``session_id``.

        Parameters
        ----------
        session_id:
            The session to remove.

        Raises
        ------
        KeyError
            If no entry exists for ``session_id``.
        """