orchestrata_ultimate.session.cli.tasks

Reference › orchestrata_ultimate › session › cli › tasks

The task command group for deferred work items, which are the WorkItem records carried between sessions (see WorkCarryoverManager). Subcommands: list shows all items, add adds one with an optional priority, done marks an item done by index, reopen returns an item to TODO (the undo for done), and status shows statistics. The index-based operations use the index methods of the ContextFileManager.

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

# mypy: ignore-errors
"""Task management commands (task group) for the session CLI."""

import sys

import click

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


@click.group()
def task() -> None:
    """Manage deferred work items (tasks)."""


@click.command(name="list")
@click.pass_context
@traces(spec="006", story="FR-19", reqs=["REQ-006-FR19-003"])
def task_list(ctx: click.Context) -> None:
    """List all deferred work items."""
    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()
    items = ctx_data.deferred_work

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

    click.echo(f"{'#':>3}  {'State':<10}  {'Priority':<10}  {'Text'}")
    click.echo("-" * 70)
    for i, item in enumerate(items):
        prio = item.priority or "-"
        click.echo(f"{i:>3}  {item.state:<10}  {prio:<10}  {item.text}")
    click.echo(f"\nTotal: {len(items)} tasks")


@click.command(name="add")
@click.argument("text")
@click.option(
    "--priority", default=None, help="Task priority (e.g. high, medium, low)."
)
@click.pass_context
@traces(spec="006", story="FR-19", reqs=["REQ-006-FR19-001"])
def task_add(ctx: click.Context, text: str, priority: str | None) -> None:
    """Add a new deferred work item."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    from orchestrata_ultimate.session.models import WorkItem

    item = WorkItem(text=text, priority=priority)
    mgr.add_work_item(item)
    click.echo(f"Added task: {text}")


@click.command(name="done")
@click.argument("index", type=int)
@click.pass_context
@traces(spec="006", story="FR-19", reqs=["REQ-006-FR19-002"])
def task_done(ctx: click.Context, index: int) -> None:
    """Mark a task as done by index."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    # One transactional call, not a read here and a write from a fresh read:
    # the old form mutated a detached context and persisted a different one,
    # so it always echoed success and never changed anything.
    item = mgr.complete_work_item_at(index)
    if item is None:
        click.echo(
            f"Invalid index {index}. Use 'task list' to see valid indices.", err=True
        )
        sys.exit(1)

    click.echo(f"Marked as done: {item.text}")


@click.command(name="reopen")
@click.argument("index", type=int)
@click.pass_context
@traces(spec="006", story="FR-19", reqs=["REQ-006-FR19-002"])
def task_reopen(ctx: click.Context, index: int) -> None:
    """Return a task to TODO -- the undo for ``task done``."""
    root = ctx.obj["root"]
    mgr = _get_manager(root)

    item = mgr.reopen_work_item_at(index)
    if item is None:
        click.echo(
            f"Invalid index {index}. Use 'task list' to see valid indices.", err=True
        )
        sys.exit(1)

    click.echo(f"Reopened: {item.text}")


@click.command(name="status")
@click.pass_context
@traces(spec="006", story="FR-19", reqs=["REQ-006-FR19-004"])
def task_status(ctx: click.Context) -> None:
    """Show task statistics."""
    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()
    items = ctx_data.deferred_work

    total = len(items)
    done = sum(1 for i in items if i.state == "done")
    todo = sum(1 for i in items if i.state == "todo")
    active = sum(1 for i in items if i.state == "active")
    blocked = sum(1 for i in items if i.state == "blocked")
    deferred = sum(1 for i in items if i.state == "deferred")

    click.echo("Task Statistics:")
    click.echo(f"  Total:     {total}")
    click.echo(f"  Done:      {done}")
    click.echo(f"  Todo:      {todo}")
    click.echo(f"  Active:    {active}")
    click.echo(f"  Blocked:   {blocked}")
    click.echo(f"  Deferred:  {deferred}")