orchestrata_ultimate.session.cli.archive

Reference › orchestrata_ultimate › session › cli › archive

Archive, status and listing commands of the session CLI. archive has options to verify history integrity (--verify), recover lost sessions from git (--recover) and show statistics (--stats); otherwise it archives old sessions. These call verify_history_integrity, recover_from_git and archive_old_sessions of the ContextFileManager. status shows the current session state and list_sessions lists all sessions across the context file and the history file. See also SessionArchiver.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

# mypy: ignore-errors
"""Archive, status, and session-listing commands for the session CLI."""

import json
import logging
import sys

import click

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.session.cli._manager import _get_manager

logger = logging.getLogger(__name__)


@click.command()
@click.option("--verify", is_flag=True, help="Run integrity verification.")
@click.option("--recover", is_flag=True, help="Recover lost sessions from git.")
@click.option("--stats", is_flag=True, help="Show archive statistics.")
@click.pass_context
@traces(
    spec="006",
    story="FR-018",
    reqs=[
        "REQ-006-FR18-001",
        "REQ-006-FR18-002",
        "REQ-006-FR18-003",
        "REQ-006-FR18-004",
    ],
)
def archive(ctx: click.Context, verify: bool, recover: bool, stats: bool) -> None:
    """Archive sessions, verify integrity, or recover from git."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    if not mgr.file_exists():
        click.echo(f"No context files found in {root}", err=True)
        sys.exit(1)

    if verify:
        result = mgr.verify_history_integrity()
        click.echo(json.dumps(result, indent=2))
        if not result["valid"]:
            sys.exit(1)
        return

    if recover:
        result = mgr.recover_from_git()
        click.echo(json.dumps(result, indent=2))
        if result["errors"]:
            sys.exit(1)
        return

    if stats:
        from orchestrata_ultimate.session.archive import SessionArchiver

        archiver = SessionArchiver(root)
        result = archiver.stats()
        click.echo(json.dumps(result, indent=2))
        return

    # Default: run archive now
    mgr.archive_old_sessions()
    click.echo("Archive check complete.")


@click.command()
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def status(ctx: click.Context) -> None:
    """Show current session state."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    if not mgr.file_exists():
        click.echo(f"No context files found in {root}", err=True)
        sys.exit(1)

    ctx_data = mgr.read()
    state = ctx_data.state

    click.echo(f"Project root: {root}")
    click.echo(f"Active session: {state.active_session}")
    click.echo(f"Version: {state.version}")
    click.echo(f"Branch: {state.git_branch}")
    if state.test_state:
        click.echo(f"Test state: {state.test_state}")
    click.echo(f"Status: {state.status_line}")
    click.echo(f"Sessions in context: {len(ctx_data.session_history)}")
    click.echo(f"Pending work items: {len(ctx_data.pending_work())}")


@click.command(name="list")
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-006"])
def list_sessions(ctx: click.Context) -> None:
    """List all sessions across context and history."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    sessions: list[dict] = []

    # Sessions from history
    if mgr.history_path.exists():
        import re

        content = mgr.history_path.read_text()
        sessions.extend(
            {
                "seq": int(match.group(1)),
                "date": match.group(2),
                "model": match.group(3).strip(),
                "source": "history",
            }
            for match in re.finditer(
                r"### Session (\d+) - ([\d-]+ [\d:]+) - (.+)", content
            )
        )

    # Sessions from context.json
    if mgr.json_path.exists():
        try:
            ctx_data = mgr._read_json()
            sessions.extend(
                {
                    "seq": s.seq,
                    "date": s.when.strftime("%Y-%m-%d %H:%M"),
                    "model": s.ai_model,
                    "source": "context",
                }
                for s in ctx_data.session_history
            )
        except (OSError, ValueError, KeyError) as e:
            logger.debug("Failed to read context.json for session list: %s", e)

    sessions.sort(key=lambda s: s["seq"])

    if not sessions:
        click.echo("No sessions found.")
        return

    click.echo(f"{'Seq':>5}  {'Date':>16}  {'Model':<25}  {'Source':<10}")
    click.echo("-" * 62)
    for s in sessions:
        click.echo(
            f"{s['seq']:>5}  {s['date']:>16}  {s['model']:<25}  {s['source']:<10}"
        )
    click.echo(f"\nTotal: {len(sessions)} sessions")