orchestrata_ultimate.session.cli.lifecycle

Reference › orchestrata_ultimate › session › cli › lifecycle

The session lifecycle commands of the CLI, built on the SessionController. begin starts a session with a required purpose, optional objectives and an AI-model label. work logs a work entry with details, artifact logs a created or modified file with an optional line count and summary, and choice logs a technical decision with its reason. handoff sets the handoff data (decisions, blockers, active work, key files, commands and alerts) for the next session. end closes the session with achievements and upcoming items; it takes repeatable --achievement and --upcoming options, --interactive (prompt for missing fields, also used when no achievement is given on a terminal), --smart (auto-detect artifacts, tests and git status), --commit (commit code changes first) and --validate/--no-validate (strict checklist validation, on by default). It exits with an error when there is no active session. resume reopens the latest session and summary shows the current one. These are the commands that the close checklist expects to have been run: work, artifacts, choices and a handoff.

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/lifecycle.py

# mypy: ignore-errors
"""Session lifecycle commands (begin, work, end, resume, summary) for the CLI."""

from __future__ import annotations

import sys
from collections.abc import Sequence
from pathlib import Path
from typing import TYPE_CHECKING

import click

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

if TYPE_CHECKING:
    from orchestrata_ultimate.session.models import SessionRecord
    from orchestrata_ultimate.session.session import SessionController


@click.command()
@click.argument("purpose")
@click.option(
    "--objective",
    "objectives",
    multiple=True,
    help="Session objective (can be repeated).",
)
@click.option("--model", default=default_ai_model, help="AI model name.")
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def begin(
    ctx: click.Context, purpose: str, objectives: tuple[str, ...], model: str
) -> None:
    """Begin a new session."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root, ai_model=model)
    session = ctrl.begin(purpose=purpose, objectives=list(objectives))
    click.echo(f"Session {session.seq:03d} started.")
    click.echo(f"Purpose: {session.purpose}")
    if session.objectives:
        click.echo("Objectives:")
        for obj in session.objectives:
            click.echo(f"  - {obj}")


@click.command()
@click.argument("desc")
@click.option(
    "--detail", "details", multiple=True, help="Detail bullet (can be repeated)."
)
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def work(ctx: click.Context, desc: str, details: tuple[str, ...]) -> None:
    """Log work in the current session."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = _get_active_session(ctrl)
    if session is None:
        click.echo("No active session. Start one with 'session begin'.", err=True)
        sys.exit(1)

    ctrl.log_work(desc=desc, details=list(details))
    click.echo(f"Logged: {desc}")


@click.command()
@click.argument("filepath")
@click.option("--lines", type=int, default=None, help="Line count.")
@click.option("--summary", default=None, help="Description of the file.")
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def artifact(
    ctx: click.Context,
    filepath: str,
    lines: int | None,
    summary: str | None,
) -> None:
    """Log a created or modified file."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = _get_active_session(ctrl)
    if session is None:
        click.echo("No active session. Start one with 'session begin'.", err=True)
        sys.exit(1)

    ctrl.log_artifact(filepath=filepath, line_count=lines, summary=summary)
    click.echo(f"Artifact logged: {filepath}")


@click.command()
@click.argument("choice")
@click.option("--reason", required=True, help="Rationale for the decision.")
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def choice(ctx: click.Context, choice: str, reason: str) -> None:
    """Log a technical decision."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = _get_active_session(ctrl)
    if session is None:
        click.echo("No active session. Start one with 'session begin'.", err=True)
        sys.exit(1)

    ctrl.log_choice(choice=choice, reason=reason)
    click.echo(f"Choice logged: {choice}")


@click.command()
@click.option(
    "--decision", "decisions", multiple=True, help="Key decision (repeatable)."
)
@click.option(
    "--blocker", "blockers", multiple=True, help="Current blocker (repeatable)."
)
@click.option(
    "--active", "active_work", multiple=True, help="In-progress work (repeatable)."
)
@click.option(
    "--file", "key_files", multiple=True, help="Important file path (repeatable)."
)
@click.option("--cmd", "cmds", multiple=True, help="Useful command (repeatable).")
@click.option("--alert", "alerts", multiple=True, help="Alert (repeatable).")
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def handoff(
    ctx: click.Context,
    decisions: tuple[str, ...],
    blockers: tuple[str, ...],
    active_work: tuple[str, ...],
    key_files: tuple[str, ...],
    cmds: tuple[str, ...],
    alerts: tuple[str, ...],
) -> None:
    """Set handoff data for the next session."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = _get_active_session(ctrl)
    if session is None:
        click.echo("No active session. Start one with 'session begin'.", err=True)
        sys.exit(1)

    ctrl.set_handoff(
        decisions=list(decisions) if decisions else None,
        blockers=list(blockers) if blockers else None,
        active_work=list(active_work) if active_work else None,
        key_files=list(key_files) if key_files else None,
        cmds=list(cmds) if cmds else None,
        alerts=list(alerts) if alerts else None,
    )
    click.echo("Handoff set.")


@click.command()
@click.option(
    "--achievement",
    "achievements",
    multiple=True,
    help="Achievement (can be repeated).",
)
@click.option(
    "--upcoming",
    "upcoming_items",
    multiple=True,
    help="Upcoming task (can be repeated).",
)
@click.option(
    "--interactive", "-i", is_flag=True, help="Prompt for missing fields interactively."
)
@click.option(
    "--smart", is_flag=True, help="Auto-detect artifacts, tests, and git status."
)
@click.option(
    "--commit", is_flag=True, help="Commit all code changes before ending session."
)
@click.option(
    "--validate/--no-validate",
    default=True,
    help="Run strict checklist validation before ending (default: on).",
)
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def end(
    ctx: click.Context,
    achievements: tuple[str, ...],
    upcoming_items: tuple[str, ...],
    interactive: bool,
    smart: bool,
    commit: bool,
    validate: bool,
) -> None:
    """End the current session."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = _get_active_session(ctrl)
    if session is None:
        click.echo("No active session. Start one with 'session begin'.", err=True)
        sys.exit(1)

    # Auto-detect if --smart
    if smart:
        _auto_detect_session_data(ctrl, root)

    # Interactive mode
    ach: Sequence[str] = achievements
    up: Sequence[str] = upcoming_items
    if interactive or (not achievements and sys.stdin.isatty()):
        ach, up, commit = _run_end_interactive(
            ctrl, achievements, upcoming_items, commit
        )

    if commit:
        _commit_code_changes(root)

    session = _finalize_session(ctrl, list(ach), list(up), validate)
    _report_ended_session(session)


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-001"])
def _run_end_interactive(
    ctrl: SessionController,
    achievements: Sequence[str],
    upcoming_items: Sequence[str],
    commit: bool,
) -> tuple[list[str], list[str], bool]:
    """Run interactive prompts for the ``end`` command, returning updated values."""
    achievements = _prompt_achievements(achievements)
    upcoming_items = _prompt_upcoming(upcoming_items)
    if click.confirm("Log handoff data?", default=False):
        _prompt_handoff(ctrl)
    if not commit:
        commit = click.confirm("Commit code changes?", default=False)
    return achievements, upcoming_items, commit


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-001", "REQ-006-FR6-002"])
def _finalize_session(
    ctrl: SessionController,
    achievements: list[str],
    upcoming: list[str],
    validate: bool,
) -> SessionRecord | None:
    """End the session, validating first when requested. Exits on validation failure."""
    if not validate:
        return ctrl.end(achievements=achievements, upcoming=upcoming)

    from orchestrata_ultimate.session.exceptions import ChecklistError

    try:
        return ctrl.validate_and_end(achievements=achievements, upcoming=upcoming)
    except ChecklistError as e:
        click.echo("Session validation failed:", err=True)
        for failure in e.failures:
            click.echo(f"  - {failure}", err=True)
        sys.exit(1)


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-001", "REQ-006-FR6-002"])
def _report_ended_session(session: SessionRecord | None) -> None:
    """Print the end-of-session summary line. Exits if the session failed to end."""
    if not session:
        click.echo("Failed to end session.", err=True)
        sys.exit(1)
    click.echo(f"Session {session.seq:03d} ended ({session.length}).")
    if session.achievements:
        click.echo(f"Achievements: {len(session.achievements)}")
    if session.upcoming:
        click.echo(f"Upcoming: {len(session.upcoming)}")


@click.command()
@click.pass_context
@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def resume(ctx: click.Context) -> None:
    """Resume the latest session."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = ctrl.resume()
    if session:
        click.echo(f"Resumed session {session.seq:03d}.")
        click.echo(f"Purpose: {session.purpose}")
    else:
        click.echo("No session to resume.", err=True)
        sys.exit(1)


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

    from orchestrata_ultimate.session.session import SessionController

    ctrl = SessionController(root)
    session = _get_active_session(ctrl)

    if session is None:
        click.echo("No active session. Start one with 'session begin'.", err=True)
        sys.exit(1)

    click.echo(f"Session {session.seq:03d} — {session.purpose}")
    click.echo(f"Started: {session.when.strftime('%Y-%m-%d %H:%M')}")
    click.echo("")

    _echo_summary_objectives(session)
    _echo_summary_work_log(session)
    _echo_summary_artifacts(session)
    _echo_summary_choices(session)
    _echo_summary_handoff(session)
    _echo_summary_achievements(session)


# ── Summary section renderers ───────────────────


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_summary_objectives(session: SessionRecord) -> None:
    """Print the objectives section of the session summary."""
    if not session.objectives:
        return
    click.echo("Objectives:")
    for obj in session.objectives:
        click.echo(f"  - {obj}")
    click.echo("")


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_summary_work_log(session: SessionRecord) -> None:
    """Print the work log section of the session summary."""
    if not session.log:
        return
    click.echo(f"Work log ({len(session.log)} entries):")
    for entry in session.log:
        click.echo(f"  • {entry.desc}")
        for detail in entry.details:
            click.echo(f"    - {detail}")
    click.echo("")


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_summary_artifacts(session: SessionRecord) -> None:
    """Print the artifacts section of the session summary."""
    if not session.artifacts:
        return
    click.echo(f"Artifacts ({len(session.artifacts)}):")
    for art in session.artifacts:
        lc = f" ({art.line_count} lines)" if art.line_count else ""
        click.echo(f"  • {art.filepath}{lc}")
    click.echo("")


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_summary_choices(session: SessionRecord) -> None:
    """Print the choices section of the session summary."""
    if not session.choices:
        return
    click.echo(f"Choices ({len(session.choices)}):")
    for choice in session.choices:
        click.echo(f"  • {choice.choice}")
        click.echo(f"    {choice.reason}")
    click.echo("")


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_summary_handoff(session: SessionRecord) -> None:
    """Print the handoff section of the session summary."""
    if not session.handoff:
        return
    h = session.handoff
    click.echo("Handoff:")
    _echo_handoff_block("Decisions:", "    - ", h.decisions)
    _echo_handoff_block("Blockers:", "    ⚠ ", h.blockers)
    _echo_handoff_block("Active work:", "    - ", h.active_work)
    _echo_handoff_block("Key files:", "    - ", h.key_files)
    _echo_handoff_block("Commands:", "    $ ", h.cmds)
    _echo_handoff_block("Alerts:", "    ⚠ ", h.alerts)
    click.echo("")


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_handoff_block(label: str, prefix: str, items: list[str]) -> None:
    """Print a single labelled bullet block within the handoff section."""
    if not items:
        return
    click.echo(f"  {label}")
    for item in items:
        click.echo(f"{prefix}{item}")


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-005"])
def _echo_summary_achievements(session: SessionRecord) -> None:
    """Print the achievements section of the session summary."""
    if not session.achievements:
        return
    click.echo(f"Achievements so far ({len(session.achievements)}):")
    for ach in session.achievements:
        click.echo(f"  ✓ {ach}")


# ── Helper Functions ────────────────────────────


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-001", "REQ-006-FR14-006"])
def _prompt_achievements(existing: Sequence[str]) -> list[str]:
    """Prompt for achievements interactively."""
    items = list(existing)
    if items:
        click.echo("Current achievements:")
        for item in items:
            click.echo(f"  - {item}")
    click.echo("Achievements (empty line to finish):")
    while True:
        value = click.prompt("> ", default="", show_default=False).strip()
        if not value:
            break
        items.append(value)
    return items


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-001", "REQ-006-FR14-006"])
def _prompt_upcoming(existing: Sequence[str]) -> list[str]:
    """Prompt for upcoming tasks interactively."""
    items = list(existing)
    if items:
        click.echo("Current upcoming tasks:")
        for item in items:
            click.echo(f"  - {item}")
    click.echo("Upcoming tasks (empty line to finish):")
    while True:
        value = click.prompt("> ", default="", show_default=False).strip()
        if not value:
            break
        items.append(value)
    return items


@traces(spec="006", story="FR-14", reqs=["REQ-006-FR14-006"])
def _prompt_lines(label: str) -> list[str]:
    """Collect a list of lines for one handoff field (empty line ends input)."""
    click.echo(f"{label} (empty line to finish):")
    items: list[str] = []
    while True:
        value = click.prompt("> ", default="", show_default=False).strip()
        if not value:
            break
        items.append(value)
    return items


@traces(spec="006", story="FR-14", reqs=["REQ-006-FR14-006"])
def _prompt_handoff(ctrl: SessionController) -> None:
    """Prompt for handoff data interactively."""
    ctrl.set_handoff(
        decisions=_prompt_lines("Decisions"),
        blockers=_prompt_lines("Blockers"),
        active_work=_prompt_lines("Active work"),
        key_files=_prompt_lines("Key files"),
        cmds=_prompt_lines("Commands"),
        alerts=_prompt_lines("Alerts"),
    )


@traces(
    spec="006",
    story="US-002",
    reqs=["REQ-006-US-002-007", "REQ-006-US-002-008", "REQ-006-US-002-009"],
)
def _auto_detect_session_data(ctrl: SessionController, root: Path) -> None:
    """Auto-detect artifacts, test state, and git status for the session."""
    _detect_modified_artifacts(ctrl, root)
    _detect_test_state(ctrl, root)
    _detect_recent_commits(ctrl, root)


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-007"])
def _detect_modified_artifacts(ctrl: SessionController, root: Path) -> None:
    """Log modified (non-untracked) files from git status as session artifacts."""
    import subprocess

    try:
        result = subprocess.run(
            ["git", "status", "--porcelain"],
            cwd=root,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError, subprocess.SubprocessError:
        return
    if result.returncode != 0:
        return

    for line in result.stdout.splitlines():
        if not line.strip():
            continue
        status = line[:2]
        filepath = line[3:].strip()
        # Only track modified files, not untracked
        if not status.strip() or line.startswith("?"):
            continue
        ctrl.log_artifact(
            filepath=filepath,
            line_count=_count_lines(root / filepath),
            summary=f"Git status: {status.strip()}",
        )


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-007"])
def _count_lines(fp: Path) -> int | None:
    """Return the line count of a file, or None if it can't be read."""
    try:
        return len(fp.read_text().splitlines()) if fp.exists() else None
    except OSError, UnicodeDecodeError:
        return None


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-009"])
def _detect_test_state(ctrl: SessionController, root: Path) -> None:
    """Detect the collected test count and record it on the session state."""
    import subprocess

    try:
        result = subprocess.run(
            ["uv", "run", "pytest", "tests/", "--co", "-q"],
            cwd=root,
            capture_output=True,
            text=True,
            timeout=120,
        )
    except FileNotFoundError, subprocess.TimeoutExpired, subprocess.SubprocessError:
        return
    if result.returncode != 0:
        return

    import re as _re

    for line in reversed(result.stdout.strip().splitlines()):
        line = line.strip()
        if "test" in line and "collected" in line:
            m = _re.search(r"(\d+)\s+test", line)
            if m:
                ctrl.ctx.update_state(test_state=f"{m.group(1)} collected")
            break


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-008"])
def _detect_recent_commits(ctrl: SessionController, root: Path) -> None:
    """Log commits made since session start as work entries."""
    import subprocess

    session = ctrl.active_session
    if not session:
        return

    since = session.when.strftime("%Y-%m-%d %H:%M:%S")
    try:
        result = subprocess.run(
            ["git", "log", f"--since={since}", "--format=%s"],
            cwd=root,
            capture_output=True,
            text=True,
        )
    except FileNotFoundError, subprocess.SubprocessError:
        return

    if result.returncode == 0 and result.stdout.strip():
        for commit_msg in result.stdout.strip().splitlines()[:5]:
            if commit_msg.strip():
                ctrl.log_work(desc=f"Committed: {commit_msg.strip()}")


@traces(spec="006", story="US-002", reqs=["REQ-006-US-002-010"])
def _commit_code_changes(root: Path) -> None:
    """Commit all code changes with an auto-generated or user-provided message."""
    import subprocess

    # Check if there are changes to commit
    result = subprocess.run(
        ["git", "status", "--porcelain"],
        cwd=root,
        capture_output=True,
        text=True,
    )
    if result.returncode != 0 or not result.stdout.strip():
        click.echo("No code changes to commit.")
        return

    # Generate default message from recent work or use generic
    default_msg = "chore: session checkpoint"
    try:
        log_result = subprocess.run(
            ["git", "log", "-1", "--format=%s"],
            cwd=root,
            capture_output=True,
            text=True,
        )
        if log_result.returncode == 0 and log_result.stdout.strip():
            default_msg = log_result.stdout.strip()
    except FileNotFoundError, subprocess.SubprocessError:
        pass

    msg = click.prompt("Commit message", default=default_msg)

    subprocess.run(["git", "add", "-A"], cwd=root, capture_output=True)
    commit_result = subprocess.run(
        ["git", "commit", "-m", msg],
        cwd=root,
        capture_output=True,
        text=True,
    )
    if commit_result.returncode == 0:
        click.echo(f"Committed: {msg}")
        push = click.confirm("Push to origin?", default=False)
        if push:
            subprocess.run(["git", "push", "origin"], cwd=root, capture_output=True)
            click.echo("Pushed.")
    else:
        click.echo(f"Commit failed: {commit_result.stderr}", err=True)


@traces(spec="006", story="US-003", reqs=["REQ-006-US-003-001", "REQ-006-FR2-005"])
def _get_active_session(ctrl: SessionController) -> SessionRecord | None:
    """Get active session, auto-resuming latest unclosed if needed.

    Since each CLI command runs in a separate process, the in-memory
    _active flag is lost. We check context.json for the latest session
    and resume it if it was never properly closed.
    """
    if ctrl.active_session is not None:
        return ctrl.active_session
    # Try to resume latest session if it was never closed
    try:
        ctx = ctrl.ctx.read()
        latest = ctx.latest_session()
        if latest and latest.length is None:
            return ctrl.resume()
    except OSError, ValueError, FileNotFoundError:
        pass
    return None