orchestrata_ultimate.session.cli.main

Reference › orchestrata_ultimate › session › cli › main

The root of the session command line (spec 006, FR-018). main is a Click group with a --root option that names the project directory. Without it, _find_project_root walks up from the current directory and picks the first parent holding a context.json, else one holding a .venv, else one holding a pyproject.toml, and falls back to the current directory. The group stores the environment mapping (the process environment by default) in the Click context for the subcommands, and the other modules of this package register their commands on it: lifecycle, archive, goals, tasks, achievements, kg, overview and pipeline. It runs as python -m orchestrata_ultimate.session.cli through main.

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

# mypy: ignore-errors
"""Session management CLI.

Provides commands for archive management, integrity verification,
git-based recovery, and session status queries.

Spec: 006 - Session Management
"""

import logging
import os
from pathlib import Path

import click

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@traces(spec="006", story="FR-14", reqs=["REQ-006-FR14-001", "REQ-006-FR18-005"])
def _find_project_root() -> Path:
    """Walk up from cwd to find the project root.

    Looks for markers in order of preference:
    1. ``context.json`` — existing session data
    2. ``.venv`` — Python virtual environment (project root)
    3. ``pyproject.toml`` — Python project config
    Falls back to cwd if nothing found.
    """
    current = Path.cwd()
    for parent in [current, *current.parents]:
        if (parent / "context.json").exists():
            return parent
    for parent in [current, *current.parents]:
        if (parent / ".venv").is_dir():
            return parent
    for parent in [current, *current.parents]:
        if (parent / "pyproject.toml").exists():
            return parent
    return current


@traces(
    spec="006",
    story="FR-018",
    reqs=[
        "REQ-006-FR18-001",
        "REQ-006-FR18-002",
        "REQ-006-FR18-003",
        "REQ-006-FR18-004",
        "REQ-006-FR18-005",
        "REQ-006-FR18-006",
    ],
)
@click.group()
@click.option(
    "--root",
    type=click.Path(exists=True, file_okay=False, path_type=Path),
    default=None,
    help="Project root directory. Auto-detected if not specified.",
)
@click.pass_context
def main(ctx: click.Context, root: Path | None) -> None:
    """Orchestrata session management CLI."""
    ctx.ensure_object(dict)
    if "env" not in ctx.obj:
        ctx.obj["env"] = os.environ
    env = ctx.obj["env"]
    if root is None:
        root = _find_project_root()
        # SPEC-023 live-repo protection: a `session` subprocess spawned from
        # inside a pytest run inherits PYTEST_CURRENT_TEST and, with no
        # explicit --root, auto-detects whatever tracked project the test
        # process happens to run from — the LIVE repository during a suite
        # run. That is exactly how gate runs appended test fixtures into the
        # real context.json. Refuse the auto-detected root in that situation;
        # an explicit --root (a deliberately chosen target) stays allowed, as
        # does SESSION_CLI_ALLOW_UNDER_PYTEST=1 for tests that intentionally
        # exercise auto-detection against their own tmp project.
        if (
            (root / "context.json").exists()
            and env.get("PYTEST_CURRENT_TEST")
            and not env.get("SESSION_CLI_ALLOW_UNDER_PYTEST")
        ):
            raise click.UsageError(
                f"refusing auto-detected project root '{root}' inside a "
                "pytest run: session writes would mutate a live tracked "
                "project (SPEC-023). Pass --root <dir> explicitly or set "
                "SESSION_CLI_ALLOW_UNDER_PYTEST=1."
            )
    ctx.obj["root"] = root


# ── Command registration ────────────────────────────────────
# Imported after `main` is defined so each command module can register
# against it. The command modules get their shared `ContextFileManager`
# factory from `session.cli._manager` (not from this module), which avoids
# the import cycle that used to exist between `main` and the command modules.

from orchestrata_ultimate.session.cli.achievements import (  # noqa: E402
    achievement,
    achievement_add,
    achievement_list,
)
from orchestrata_ultimate.session.cli.archive import (  # noqa: E402
    archive,
    list_sessions,
    status,
)
from orchestrata_ultimate.session.cli.goals import (  # noqa: E402
    goal,
    goal_abandon,
    goal_achieve,
    goal_advance,
    goal_block,
    goal_criterion,
    goal_evaluate_cmd,
    goal_progress,
    goal_set,
    goal_status,
    goal_step,
    goal_task,
    goal_unblock,
)
from orchestrata_ultimate.session.cli.kg import (  # noqa: E402
    query,
    query_file,
    query_search,
    query_spec,
    sync,
    unchecked,
    undocumented,
    untraced,
)
from orchestrata_ultimate.session.cli.lifecycle import (  # noqa: E402
    artifact,
    begin,
    choice,
    end,
    handoff,
    resume,
    summary,
    work,
)
from orchestrata_ultimate.session.cli.overview import overview  # noqa: E402
from orchestrata_ultimate.session.cli.pipeline import (  # noqa: E402
    pipeline_approve,
    pipeline_list,
    pipeline_reject,
    pipeline_respond,
    pipeline_resume,
    pipeline_run,
    pipeline_status,
)
from orchestrata_ultimate.session.cli.tasks import (  # noqa: E402
    task,
    task_add,
    task_done,
    task_list,
    task_reopen,
    task_status,
)

# Root-level commands and groups
main.add_command(archive)
main.add_command(status)
main.add_command(list_sessions)
main.add_command(task)
main.add_command(achievement)
main.add_command(goal)
main.add_command(sync)
main.add_command(query)
main.add_command(artifact)
main.add_command(begin)
main.add_command(choice)
main.add_command(end)
main.add_command(handoff)
main.add_command(resume)
main.add_command(summary)
main.add_command(work)
main.add_command(overview)

# Task subcommands
task.add_command(task_list)
task.add_command(task_add)
task.add_command(task_done)
task.add_command(task_reopen)
task.add_command(task_status)

# Achievement subcommands
achievement.add_command(achievement_list)
achievement.add_command(achievement_add)

# Goal subcommands
goal.add_command(goal_set)
goal.add_command(goal_status)
goal.add_command(goal_criterion)
goal.add_command(goal_task)
goal.add_command(goal_progress)
goal.add_command(goal_abandon)
goal.add_command(goal_achieve)
goal.add_command(goal_advance)
goal.add_command(goal_step)
goal.add_command(goal_block)
goal.add_command(goal_unblock)
goal.add_command(goal_evaluate_cmd)

# Knowledge-graph query subcommands
query.add_command(query_file)
query.add_command(query_spec)
query.add_command(undocumented)
query.add_command(untraced)
query.add_command(unchecked)
query.add_command(query_search)

# Pipeline subcommands
pipeline = click.Group("pipeline", help="Run, monitor, and resume agent pipelines")
pipeline.add_command(pipeline_run)
pipeline.add_command(pipeline_status)
pipeline.add_command(pipeline_resume)
pipeline.add_command(pipeline_respond)
pipeline.add_command(pipeline_approve)
pipeline.add_command(pipeline_reject)
pipeline.add_command(pipeline_list)
main.add_command(pipeline)