orchestrata_ultimate.session.cli.goals

Reference › orchestrata_ultimate › session › cli › goals

The goal command group (SPEC-023). goal itself manages goals with exit criteria and tasks. The subcommands are set (a new active goal from a statement), status (status and progress), criterion and task (add an exit criterion or a task), progress (log progress), block and unblock (set or clear a blocker with a reason), evaluate (run every criterion's verify command and grade the goal; --no-alarm skips the sound), advance (mark the current task done and move on), step (set the current step index), and achieve and abandon (change the status). Each command calls the matching method of GoalMixin.

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

# mypy: ignore-errors
"""Goal management commands (goal group, SPEC-023) for the session CLI."""

from __future__ import annotations

import sys
from typing import TYPE_CHECKING

import click

from orchestrata_ultimate.session.cli._manager import _get_manager

if TYPE_CHECKING:
    from orchestrata_ultimate.session.models import Goal


@click.group()
def goal() -> None:
    """Manage SPEC-023 goals with exit criteria and tasks."""


@click.command(name="set")
@click.argument("statement")
@click.pass_context
def goal_set(ctx: click.Context, statement: str) -> None:
    """Set a new active goal."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    from orchestrata_ultimate.session.models import Goal

    ctx_data = mgr.read()
    new_goal = Goal(statement=statement)
    ctx_data.state.goal = new_goal
    mgr.write(ctx_data)
    click.echo(f"Goal set: {statement} ({new_goal.id})")


@click.command(name="status")
@click.pass_context
def goal_status(ctx: click.Context) -> None:
    """Show current goal status and progress."""
    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()
    g = ctx_data.state.goal

    if not g:
        click.echo("No active goal.")
        return

    click.echo(f"Goal: {g.statement}")
    click.echo(f"  ID:     {g.id}")
    click.echo(f"  Status: {g.status}")
    click.echo(f"  Set at: {g.set_at}")
    click.echo(f"  Current step: {g.current_step}")

    _echo_goal_criteria(g)
    _echo_goal_tasks(g)
    _echo_goal_progress(g)
    _echo_goal_blockers(g)


def _echo_goal_criteria(g: Goal) -> None:
    """Print the exit-criteria section of the goal status."""
    if not g.criteria:
        return
    click.echo(f"\n  Exit Criteria ({len(g.criteria)}):")
    for i, c in enumerate(g.criteria, 1):
        status_icon = "✓" if c.status == "met" else "○"
        click.echo(f"    {status_icon} {i}. {c.description} [{c.status}]")


def _echo_goal_tasks(g: Goal) -> None:
    """Print the tasks section of the goal status."""
    if not g.tasks:
        return
    click.echo(f"\n  Tasks ({len(g.tasks)}):")
    for i, t in enumerate(g.tasks, 1):
        status_icon = "✓" if t.status == "done" else "○"
        click.echo(f"    {status_icon} {i}. {t.description} [{t.status}]")


def _echo_goal_progress(g: Goal) -> None:
    """Print the recent-progress section of the goal status."""
    if not g.progress:
        return
    click.echo(f"\n  Progress ({len(g.progress)} entries):")
    for p in g.progress[-3:]:
        click.echo(f"    • {p.action}: {p.note}")


def _echo_goal_blockers(g: Goal) -> None:
    """Print the blockers section of the goal status."""
    if not g.blockers:
        return
    click.echo("\n  Blockers:")
    for b in g.blockers:
        click.echo(f"    ⚠ {b}")


@click.command(name="criterion")
@click.argument("description")
@click.pass_context
def goal_criterion(ctx: click.Context, description: str) -> None:
    """Add an exit criterion to the current goal."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    import uuid

    from orchestrata_ultimate.session.models import GoalCriterion

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal. Set one first with 'goal set'.", err=True)
        sys.exit(1)

    criterion = GoalCriterion(
        id=f"c-{uuid.uuid4().hex[:8]}",
        description=description,
    )
    g.criteria.append(criterion)
    mgr.write(ctx_data)
    click.echo(f"Added criterion: {description}")


@click.command(name="task")
@click.argument("description")
@click.pass_context
def goal_task(ctx: click.Context, description: str) -> None:
    """Add a task to the current goal."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    import uuid

    from orchestrata_ultimate.session.models import GoalTask

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal. Set one first with 'goal set'.", err=True)
        sys.exit(1)

    task = GoalTask(
        id=f"t-{uuid.uuid4().hex[:8]}",
        description=description,
    )
    g.tasks.append(task)
    mgr.write(ctx_data)
    click.echo(f"Added task: {description}")


@click.command(name="progress")
@click.argument("action")
@click.argument("note")
@click.pass_context
def goal_progress(ctx: click.Context, action: str, note: str) -> None:
    """Log progress on the current goal."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    from orchestrata_ultimate.session.models import GoalProgress

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal. Set one first with 'goal set'.", err=True)
        sys.exit(1)

    entry = GoalProgress(action=action, note=note)
    g.progress.append(entry)
    mgr.write(ctx_data)
    click.echo(f"Logged progress: {action} — {note}")


@click.command(name="abandon")
@click.pass_context
def goal_abandon(ctx: click.Context) -> None:
    """Abandon the current goal."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    from datetime import datetime

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal to abandon.", err=True)
        sys.exit(1)

    g.status = "abandoned"
    g.abandoned_at = datetime.now().isoformat()
    mgr.write(ctx_data)
    click.echo(f"Abandoned goal: {g.statement}")


@click.command(name="achieve")
@click.pass_context
def goal_achieve(ctx: click.Context) -> None:
    """Mark the current goal as achieved."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    from datetime import datetime

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal to mark as achieved.", err=True)
        sys.exit(1)

    g.status = "achieved"
    g.achieved_at = datetime.now().isoformat()
    mgr.write(ctx_data)
    click.echo(f"Achieved goal: {g.statement}")
    # REQ-023-034: the manual override triggers the success alarm.
    from orchestrata_ultimate.session.audio_alarm import AudioAlarm

    AudioAlarm().play_success()


@click.command(name="block")
@click.argument("reason")
@click.pass_context
def goal_block(ctx: click.Context, reason: str) -> None:
    """Mark the current goal blocked with a reason (REQ-023-036)."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal to block.", err=True)
        sys.exit(1)

    g.blockers.append(reason)
    g.status = "blocked"
    mgr.write(ctx_data)
    click.echo(f"Blocked goal: {g.statement} — {reason}")
    # REQ-023-036: blocking triggers the attention alarm.
    from orchestrata_ultimate.session.audio_alarm import AudioAlarm

    AudioAlarm().play_attention()


@click.command(name="unblock")
@click.pass_context
def goal_unblock(ctx: click.Context) -> None:
    """Clear the blocked status and resume work (REQ-023-037)."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal to unblock.", err=True)
        sys.exit(1)

    g.blockers = []
    g.status = "active"
    mgr.write(ctx_data)
    click.echo(f"Unblocked goal: {g.statement}")


@click.command(name="evaluate")
@click.option(
    "--no-alarm",
    is_flag=True,
    help="Skip the success alarm on achievement (e.g. in scripts).",
)
@click.pass_context
def goal_evaluate_cmd(ctx: click.Context, no_alarm: bool) -> None:
    """Run every criterion's verify_command and grade the goal (US-002)."""
    root = ctx.obj["root"]

    from orchestrata_ultimate.session.controller._core import SessionController

    controller = SessionController(root)
    controller.goal_evaluate(play_alarm=not no_alarm)
    goal_after = controller.get_goal()
    if goal_after is None:
        click.echo("No active goal to evaluate.", err=True)
        sys.exit(1)
    click.echo(f"Goal status after evaluation: {goal_after.status}")
    for criterion in goal_after.criteria:
        click.echo(f"  {criterion.status:>7}  {criterion.description}")


@click.command(name="advance")
@click.pass_context
def goal_advance(ctx: click.Context) -> None:
    """Mark the current task done and advance to the next step."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal. Set one first with 'goal set'.", err=True)
        sys.exit(1)

    previous = g.current_step
    from orchestrata_ultimate.session.controller._core import SessionController

    SessionController(root).advance_goal_step()
    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("Goal was cleared during advance.", err=True)
        sys.exit(1)
    click.echo(
        f"Advanced from step {previous} to step {g.current_step} "
        f"({len(g.tasks)} tasks)."
    )


@click.command(name="step")
@click.argument("index", type=int)
@click.pass_context
def goal_step(ctx: click.Context, index: int) -> None:
    """Set the current step index explicitly."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("No active goal. Set one first with 'goal set'.", err=True)
        sys.exit(1)

    from orchestrata_ultimate.session.controller._core import SessionController

    SessionController(root).set_goal_step(index)
    ctx_data = mgr.read()
    g = ctx_data.state.goal
    if not g:
        click.echo("Goal was cleared during step update.", err=True)
        sys.exit(1)
    click.echo(f"Current step set to {g.current_step}.")