orchestrata_ultimate.session.cli.overview

Reference › orchestrata_ultimate › session › cli › overview

session overview prints a one-shot startup digest of seven sections, run in parallel on a thread pool so that no single slow source blocks the command. Each section is built by a private function with its own time limit: infrastructure status (_section_infra, from the test-services status script), the current session and open deferred tasks (_section_session), the last test verdict (_section_last_test_verdict), rulings (_section_rulings, read from the directory named by RULINGS_DIR, or reported as unavailable), stale tracking (_section_stale_tracking, comparing the age of the plan and context files with the newest git tag), git state (_section_git) and project status (_section_project_status). Helpers format ages, tail the last line of a file and run git with a timeout. The command is a digest and not a gate: it always exits 0, and a section that fails is shown as an error line for that section only. It uses only repo-relative or environment-driven inputs.

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

# mypy: ignore-errors
"""``session overview`` — one-shot startup digest (workflow-improvements-plan.md Phase 1).

**Boundary rule:** this module lives in the product package
(``src/orchestrata_ultimate/``), so every input it touches must be
repo-relative (``ctx.obj["root"]``, i.e. the project root Click already
resolves for every session command) or env-driven (e.g. ``RULINGS_DIR``),
degrading gracefully to "unavailable" when unset. Host-specific paths
(``~/.claude/...``, IRC watchdogs, etc.) never belong here — they live in
the hooks that CALL ``overview`` (``.kimi/hooks/session-start.sh``,
``.claude/hooks/session-overview.sh``).

Prints one digest across seven sections, run in parallel where independent
(``ThreadPoolExecutor`` — the command must never sequential-block on an
infra timeout):

  1. INFRA              — ``scripts/start_test_services.py --status --json``
  2. SESSION             — session status + open deferred tasks, in-process
  3. LAST TEST VERDICT   — newest of ``.flake_recoveries.ndjson`` / ``reports/*review*``
  4. RULINGS             — ``$RULINGS_DIR/index.md`` id+title lines, or "unavailable"
  5. STALE TRACKING      — plan/context.md age vs. newest git tag
  6. GIT                 — branch, ahead/behind, dirty count
  7. PROJECT STATUS      — embeds ``scripts/project_status.py --brief`` (Phase 4)

This is a DIGEST, not a gate: it always exits 0. Any internal failure inside
a section is caught and rendered as an ``error: <one line>`` body for that
section only — one broken subsystem never hides the rest of the picture.
Non-zero exit is reserved for a catastrophic failure of the command itself
(e.g. Click argument parsing), which Click already handles before this
function body ever runs.

Spec: workflow-improvements-plan.md Phase 1 (no ratified SPEC-NNN yet — this
is process tooling, matching the untraced convention already used by sibling
tooling such as ``scripts/start_test_services.py`` / ``scripts/health_check.py``).
"""

from __future__ import annotations

import json
import os
import re
import subprocess
import sys
import time
from collections.abc import Mapping
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import TimeoutError as FutureTimeoutError
from datetime import datetime, timedelta
from pathlib import Path

import click

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

# Fallback default for RULINGS_DIR when the env var is unset. This IS a
# host-specific absolute path, which looks like it violates the boundary
# rule above — it doesn't: it is only the *default value* of an env-driven
# input (``RULINGS_DIR`` always wins when set), and every call site still
# degrades to "rulings: unavailable" when the directory doesn't exist. The
# concrete path is named explicitly per the build brief (the knowledge-wiki
# rulings store another builder is standing up concurrently).
_DEFAULT_RULINGS_DIR = str(Path.home() / "workspace" / "knowledge-wiki" / "rulings")

_INFRA_TIMEOUT_S = 20.0
_STATUS_TIMEOUT_S = 15.0
_GIT_TIMEOUT_S = 5.0
_FAST_TIMEOUT_S = 10.0

_SECTION_ORDER = [
    "INFRA",
    "SESSION",
    "LAST TEST VERDICT",
    "RULINGS",
    "STALE TRACKING",
    "GIT",
    "PROJECT STATUS",
]

# Backstop timeouts for collecting each future's result — a bit above the
# section's own internal subprocess/IO timeout, so the section's own
# "error: timed out after Ns" message wins over the pool's generic one.
_FUTURE_TIMEOUTS = {
    "INFRA": _INFRA_TIMEOUT_S + 5,
    "SESSION": _FAST_TIMEOUT_S,
    "LAST TEST VERDICT": _FAST_TIMEOUT_S,
    "RULINGS": _FAST_TIMEOUT_S,
    "STALE TRACKING": _FAST_TIMEOUT_S,
    "GIT": _FAST_TIMEOUT_S,
    "PROJECT STATUS": _STATUS_TIMEOUT_S + 5,
}

_RULING_ID_RE = re.compile(r"RUL-\d+")


def _fmt_age(dt: datetime) -> str:
    """Human-readable age like ``3h ago`` / ``2d ago`` for a naive local datetime."""
    seconds = (datetime.now() - dt).total_seconds()
    if seconds < 0:
        return "just now"
    if seconds < 3600:
        return f"{int(seconds // 60)}m ago"
    if seconds < 86400:
        return f"{int(seconds // 3600)}h ago"
    return f"{int(seconds // 86400)}d ago"


def _tail_last_line(path: Path, max_bytes: int = 65536) -> str | None:
    """Best-effort last non-empty line of a (possibly large) ndjson file.

    Only reads the tail (``max_bytes``) rather than the whole file — this
    file grows unboundedly across the project's lifetime.
    """
    try:
        size = path.stat().st_size
        with path.open("rb") as fh:
            fh.seek(-min(max_bytes, size), os.SEEK_END)
            data = fh.read()
    except OSError:
        return None
    lines = [
        ln for ln in data.decode("utf-8", errors="replace").splitlines() if ln.strip()
    ]
    return lines[-1] if lines else None


def _run_git(root: Path, *args: str, timeout: float = _GIT_TIMEOUT_S) -> str | None:
    """Run a git subcommand against ``root``; ``None`` on any failure."""
    try:
        proc = subprocess.run(
            ["git", *args],
            cwd=root,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired, OSError:
        return None
    if proc.returncode != 0:
        return None
    return proc.stdout.strip()


def _section_infra(root: Path, timeout: float = _INFRA_TIMEOUT_S) -> str:
    """Section 1: infra readiness via ``start_test_services.py --status --json``."""
    script = root / "scripts" / "start_test_services.py"
    if not script.exists():
        return "error: scripts/start_test_services.py not found"
    try:
        proc = subprocess.run(
            [sys.executable, str(script), "--status", "--json"],
            cwd=root,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return f"error: timed out after {timeout:.0f}s"
    except OSError as e:
        return f"error: {e}"
    try:
        report = json.loads(proc.stdout)
    except json.JSONDecodeError, ValueError:
        return f"error: could not parse infra report (rc={proc.returncode})"
    services = report.get("services", [])
    lines = [
        f"  {svc.get('name', '?'):<10} {svc.get('status', '?')}" for svc in services
    ]
    overall = "all ready" if report.get("all_required_ok") else "NOT all ready"
    body = "\n".join(lines) if lines else "  (no services reported)"
    return f"{overall} ({report.get('total_ms', '?')}ms)\n{body}"


def _section_session(root: Path) -> str:
    """Section 2: session status + open deferred tasks, in-process (no subprocess)."""
    mgr = _get_manager(root)
    if not mgr.file_exists():
        return "no context.json found"
    try:
        ctx_data = mgr.read()
    except (OSError, ValueError) as e:
        return f"error: {e}"
    state = ctx_data.state
    pending = ctx_data.pending_work()
    lines = [
        f"session {state.active_session} | branch {state.git_branch} | {state.status_line}"
    ]
    lines.append(f"open tasks: {len(pending)}")
    for item in pending[:5]:
        lines.append(f"  - [{item.state}] {item.text}")
    if len(pending) > 5:
        lines.append(f"  ... and {len(pending) - 5} more")
    return "\n".join(lines)


def _section_last_test_verdict(root: Path) -> str:
    """Section 3: newest of ``.flake_recoveries.ndjson`` / ``reports/*review*``."""
    candidates: list[tuple[Path, float]] = []
    flake_path = root / ".flake_recoveries.ndjson"
    if flake_path.is_file():
        candidates.append((flake_path, flake_path.stat().st_mtime))
    reports_dir = root / "reports"
    if reports_dir.is_dir():
        for p in reports_dir.glob("*review*"):
            if p.is_file():
                candidates.append((p, p.stat().st_mtime))
    if not candidates:
        return "no test-verdict artifacts found (.flake_recoveries.ndjson, reports/*review*)"
    newest_path, newest_mtime = max(candidates, key=lambda c: c[1])
    age = _fmt_age(datetime.fromtimestamp(newest_mtime))
    detail = ""
    if newest_path.name == ".flake_recoveries.ndjson":
        last_line = _tail_last_line(newest_path)
        if last_line:
            try:
                event = json.loads(last_line)
                detail = f" — last recovery: {event.get('module', '?')}"
            except json.JSONDecodeError, ValueError:
                pass
    return f"{newest_path.relative_to(root)} ({age}){detail}"


def _section_rulings(env: Mapping[str, str]) -> str:
    """Section 4: id+title lines from ``$RULINGS_DIR/index.md``, filesystem only.

    Never hits a live Wiki.js server (project rule: gates/hooks read the
    git-backed markdown store, never ``wiki get`` — a wiki outage must never
    break session start).
    """
    rulings_dir = Path(env.get("RULINGS_DIR", _DEFAULT_RULINGS_DIR))
    index_path = rulings_dir / "index.md"
    if not index_path.is_file():
        return "rulings: unavailable"
    try:
        text = index_path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return "rulings: unavailable"

    lines: list[str] = []
    for line in text.splitlines():
        stripped = line.strip()
        if not _RULING_ID_RE.search(stripped):
            continue
        # Markdown table row: "| RUL-001 | date | title | enforcement | status |"
        if stripped.startswith("|"):
            cells = [c.strip() for c in stripped.strip("|").split("|")]
            rid = next((c for c in cells if _RULING_ID_RE.fullmatch(c)), cells[0])
            title = cells[2] if len(cells) > 2 else (cells[1] if len(cells) > 1 else "")
            lines.append(f"{rid}: {title}" if title else rid)
        else:
            lines.append(stripped)
    if not lines:
        return "rulings: index.md present but no RUL-* rows found"
    return "\n".join(lines)


def _newest_tag_date(root: Path) -> datetime | None:
    out = _run_git(
        root,
        "for-each-ref",
        "--sort=-creatordate",
        "--format=%(creatordate:iso-strict)",
        "refs/tags",
    )
    if not out:
        return None
    first = out.splitlines()[0].strip()
    try:
        return datetime.fromisoformat(first).replace(tzinfo=None)
    except ValueError:
        return None


#: Phase 5.2 (plans/workflow-improvements-plan.md): the quarterly
#: ``orchestrata-history-mining.timer`` should keep dropping a fresh
#: ``reports/ruling-drift-YYYY-MM-DD.md``. A report older than this many
#: days (or none at all) means the timer likely died silently -- the timer
#: does the mining work, this section only surfaces its failure.
_RULING_DRIFT_MAX_AGE_DAYS = 100


def _ruling_drift_staleness(
    root: Path, *, max_age_days: int = _RULING_DRIFT_MAX_AGE_DAYS
) -> str | None:
    """``None`` when a fresh ``reports/ruling-drift-*.md`` exists, else a
    ``STALE: ...`` line. Independent of git tags (the miner runs on a
    calendar cadence, not a release cadence) -- checked unconditionally,
    even when :func:`_newest_tag_date` finds no tags at all.
    """
    reports_dir = root / "reports"
    candidates = (
        sorted(reports_dir.glob("ruling-drift-*.md")) if reports_dir.is_dir() else []
    )
    if not candidates:
        return (
            "  STALE: ruling-drift report stale — miner timer broken? "
            "(no reports/ruling-drift-*.md found)"
        )
    newest = max(candidates, key=lambda p: p.stat().st_mtime)
    age_days = (time.time() - newest.stat().st_mtime) / 86400
    if age_days > max_age_days:
        return (
            f"  STALE: ruling-drift report stale — miner timer broken? "
            f"(newest is {newest.relative_to(root)}, {age_days:.0f}d old)"
        )
    return None


def _section_stale_tracking(root: Path) -> str:
    """Section 5: flag if the ACTIVE plan file (``plans/master-execution-plan.md``)
    or ``context.md`` are >7d older than the newest git tag, AND (independently,
    Phase 5.2) if the quarterly ruling-drift report has gone stale -- a dead
    miner timer signal.

    The legacy ``plans/execution-plan.md`` is correctly marked STATUS: DONE /
    superseded by stage2_port_plan.md (2026-04-19) and is NOT a tracking signal —
    this function tracks the active plan only. ``master-execution-plan.md`` is
    the live, ratified-decisions workstream registry; an old execution-plan.md
    is by design and never refreshed.
    """
    lines: list[str] = []
    tag_date = _newest_tag_date(root)
    if tag_date is None:
        lines.append("no git tags found — cannot assess tracking-file staleness")
    else:
        for rel in ("plans/master-execution-plan.md", "context.md"):
            p = root / rel
            if not p.is_file():
                continue
            mtime = datetime.fromtimestamp(p.stat().st_mtime)
            if mtime < tag_date - timedelta(days=7):
                age_days = (tag_date - mtime).days
                lines.append(f"  STALE: {rel} is {age_days}d older than newest tag")
    drift_line = _ruling_drift_staleness(root)
    if drift_line:
        lines.append(drift_line)
    if not lines:
        return "OK — tracking files within 7 days of newest tag; ruling-drift report is fresh"
    return "\n".join(lines)


def _section_git(root: Path) -> str:
    """Section 6: branch, ahead/behind origin, dirty file count."""
    branch = _run_git(root, "rev-parse", "--abbrev-ref", "HEAD") or "unknown"
    dirty_out = _run_git(root, "status", "--porcelain")
    dirty_count = len(dirty_out.splitlines()) if dirty_out else 0

    ahead_behind = "ahead/behind: unavailable (no upstream)"
    upstream = _run_git(
        root, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}"
    )
    if upstream:
        counts = _run_git(
            root, "rev-list", "--left-right", "--count", f"{upstream}...HEAD"
        )
        if counts:
            parts = counts.split()
            if len(parts) == 2:
                behind, ahead = parts
                ahead_behind = f"ahead {ahead}, behind {behind} ({upstream})"

    return f"branch {branch} | {ahead_behind} | dirty files: {dirty_count}"


def _section_project_status(root: Path, timeout: float = _STATUS_TIMEOUT_S) -> str:
    """Section 7 (Phase 4 integration): embeds ``scripts/project_status.py --brief``."""
    script = root / "scripts" / "project_status.py"
    if not script.exists():
        return "error: scripts/project_status.py not found"
    try:
        proc = subprocess.run(
            [sys.executable, str(script), "--root", str(root), "--brief"],
            cwd=root,
            capture_output=True,
            text=True,
            timeout=timeout,
        )
    except subprocess.TimeoutExpired:
        return f"error: timed out after {timeout:.0f}s"
    except OSError as e:
        return f"error: {e}"
    out = proc.stdout.strip()
    if not out:
        return f"error: empty output (rc={proc.returncode}); stderr: {proc.stderr.strip()[:300]}"
    return out


@traces(spec="006", story="FR-018", reqs=["REQ-006-FR18-007"])
@click.command()
@click.pass_context
def overview(ctx: click.Context) -> None:
    """One-shot startup digest: infra, session, tests, rulings, staleness, git, status.

    Prints seven ``## SECTION`` blocks. Always exits 0 — this is a digest,
    not a gate; a broken subsystem shows up as an ``error: ...`` line inside
    its own section rather than aborting the command.
    """
    root: Path = ctx.obj["root"]
    env: Mapping[str, str] = ctx.obj.get("env", os.environ) if ctx.obj else os.environ

    results: dict[str, str] = {}
    try:
        with ThreadPoolExecutor(max_workers=len(_SECTION_ORDER)) as pool:
            futures = {
                "INFRA": pool.submit(_section_infra, root),
                "SESSION": pool.submit(_section_session, root),
                "LAST TEST VERDICT": pool.submit(_section_last_test_verdict, root),
                "RULINGS": pool.submit(_section_rulings, env),
                "STALE TRACKING": pool.submit(_section_stale_tracking, root),
                "GIT": pool.submit(_section_git, root),
                "PROJECT STATUS": pool.submit(_section_project_status, root),
            }
            for title, future in futures.items():
                try:
                    results[title] = future.result(timeout=_FUTURE_TIMEOUTS[title])
                except FutureTimeoutError:
                    results[title] = "error: section timed out"
                except Exception as e:  # noqa: BLE001 — digest must never crash
                    results[title] = f"error: {e}"
    except Exception as e:  # noqa: BLE001 — catastrophic pool failure only
        click.echo(f"error: {e}")
        return

    for title in _SECTION_ORDER:
        click.echo(f"## {title}")
        click.echo(results.get(title, "error: section did not run"))
        click.echo()