orchestrata_ultimate.session.archive

Reference › orchestrata_ultimate › session › archive

SessionArchiver manages the session history file, the archive that holds sessions older than the display window of context.md. It works on SessionRecord objects from models and complements the archiving that ContextFileManager does when it writes. The CLI archive command in cli/archive uses this kind of operation.

The class can check whether the history file exists and read it, archive a list of sessions to it, count the archived sessions and list their numbers, search the history for a query string, fetch one archived session by number, report statistics, clear the file, and decide with should_archive whether a context file is ready for archiving. See the session narrative for the wider design.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionArchiver

Decorators: traces(spec='006', story='FR-014', reqs=['REQ-006-FR14-004', 'REQ-006-FR14-005', 'REQ-006-FR15-001', 'REQ-006-FR15-002'… · Line 36

Manages session history archiving.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/archive.py

"""Fresh implementation of archive management.

Handles session history archiving.
"""

import logging
import re
from pathlib import Path

from jinja2 import Environment, FileSystemLoader

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.models import SessionRecord

# Spec: 006 - Session Management


logger = logging.getLogger(__name__)


@traces(
    spec="006",
    story="FR-014",
    reqs=[
        "REQ-006-FR14-004",
        "REQ-006-FR14-005",
        "REQ-006-FR15-001",
        "REQ-006-FR15-002",
        "REQ-006-FR15-003",
        "REQ-006-FR16-004",
        "REQ-006-FR17-004",
        "REQ-006-FR18-004",
        "REQ-006-NFR9-001",
    ],
)
class SessionArchiver:
    """Manages session history archiving."""

    HISTORY_FILE = "context.history.md"
    MAX_SESSIONS = 3
    MAX_LINES = 500

    def __init__(self, root: str | Path) -> None:
        """Initialize archiver.

        Args:
            root: Project root directory.
        """
        self.root = Path(root)
        self.history_path = self.root / self.HISTORY_FILE

        templates_dir = Path(__file__).parent / "templates"
        self.jinja = Environment(
            loader=FileSystemLoader(templates_dir),
            trim_blocks=True,
            lstrip_blocks=True,
        )

    def history_exists(self) -> bool:
        """Check if history file exists."""
        return self.history_path.exists()

    def read_history(self) -> str:
        """Read history file content.

        Returns:
            History content or empty string.
        """
        if not self.history_exists():
            return ""
        return self.history_path.read_text()

    @traces(
        spec="006",
        story="FR-015",
        reqs=[
            "REQ-006-FR15-001",
            "REQ-006-FR15-002",
            "REQ-006-FR15-003",
            "REQ-006-FR17-004",
            "REQ-006-NFR9-001",
        ],
    )
    def archive(self, sessions: list[SessionRecord]) -> None:
        """Archive sessions to history.

        Args:
            sessions: Sessions to archive.
        """
        if not sessions:
            return

        tpl = self.jinja.get_template("session.md.j2")
        new_content = ""

        for session in sessions:
            new_content += tpl.render(session=session) + "\n\n"

        if self.history_exists():
            existing = self.history_path.read_text()
            header = "# Session History Archive\n\n"
            header += "This file contains archived sessions.\n"
            header += "Sessions are archived when context.md exceeds 500 lines.\n\n"

            if existing.startswith("# Session History Archive"):
                full_content = existing + "\n" + new_content
            else:
                full_content = header + existing + "\n" + new_content
        else:
            header = "# Session History Archive\n\n"
            header += "This file contains archived sessions.\n"
            header += "Sessions are archived when context.md exceeds 500 lines.\n\n"
            full_content = header + new_content

        self.history_path.write_text(full_content)

    @traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-004"])
    def count_archived(self) -> int:
        """Count archived sessions.

        Returns:
            Number of archived sessions.
        """
        if not self.history_exists():
            return 0

        content = self.read_history()
        pattern = r"### Session \d+"
        matches = re.findall(pattern, content)
        return len(matches)

    @traces(spec="006", story="FR-015", reqs=["REQ-006-FR15-003", "REQ-006-FR16-004"])
    def archived_session_nums(self) -> list[int]:
        """Get archived session numbers.

        Returns:
            List of session numbers.
        """
        if not self.history_exists():
            return []

        content = self.read_history()
        pattern = r"### Session (\d+)"
        matches = re.findall(pattern, content)
        return [int(m) for m in matches]

    def search(self, query: str) -> list[dict]:
        """Search history for query.

        Args:
            query: Search query.

        Returns:
            List of matches with context.
        """
        if not self.history_exists():
            return []

        content = self.read_history()
        lines = content.splitlines()
        results = []
        query_lower = query.lower()

        for i, line in enumerate(lines):
            if query_lower in line.lower():
                start = max(0, i - 2)
                end = min(len(lines), i + 3)
                context = "\n".join(lines[start:end])

                results.append(
                    {
                        "line_number": i + 1,
                        "line": line,
                        "context": context,
                    }
                )

        return results

    def get_archived_session(self, seq: int) -> dict | None:
        """Get archived session by number.

        Args:
            seq: Session number.

        Returns:
            Session data or None.
        """
        if not self.history_exists():
            return None

        content = self.read_history()
        pattern = rf"### Session {seq:03d} - ([\d-]+ [\d:]+) - (.+?)\n"
        match = re.search(pattern, content)

        if not match:
            pattern = rf"### Session {seq} - ([\d-]+ [\d:]+) - (.+?)\n"
            match = re.search(pattern, content)

        if not match:
            return None

        return {
            "seq": seq,
            "date": match.group(1),
            "model": match.group(2).strip(),
        }

    @traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-004"])
    def stats(self) -> dict:
        """Get history statistics.

        Returns:
            Stats dict.
        """
        if not self.history_exists():
            return {
                "exists": False,
                "sessions_count": 0,
                "lines": 0,
                "size_bytes": 0,
            }

        content = self.read_history()
        lines = len(content.splitlines())
        size = self.history_path.stat().st_size

        return {
            "exists": True,
            "sessions_count": self.count_archived(),
            "lines": lines,
            "size_bytes": size,
        }

    def clear(self) -> bool:
        """Clear history file.

        Returns:
            True if cleared.
        """
        if not self.history_exists():
            return False

        self.history_path.unlink()
        return True

    @traces(spec="006", story="FR-014", reqs=["REQ-006-FR14-005"])
    def should_archive(self, ctx_path: Path) -> bool:
        """Check if context is ready for archiving.

        Args:
            ctx_path: Path to context.md.

        Returns:
            True if archiving needed.
        """
        if not ctx_path.exists():
            return False

        content = ctx_path.read_text()
        lines = len(content.splitlines())
        return lines > self.MAX_LINES