scripts.project_status

Reference › scripts › project_status

scripts/project_status.py produces the project status report from artifacts rather than from hand-maintained state (workflow-improvements-plan Phase 4). It is run by a human and by the session-prime step, which embeds the brief form; tests/scripts/test_project_status.py and tests/scripts/test_rtm_evidence_paths.py cover it. It has no internal imports.

build_report gathers one fact per source: get_latest_tag and get_session_number, get_test_verdict (the newest run_tests-shaped JSON report under reports/), get_rtm_summary (grade counts from reports/rtm-evidence.json, written by generate_rtm_evidence), get_coverage_floor, get_infra_status (runs start_test_services with --status --json), get_stale_tasks (not-done deferred work open longer than a number of days; read-only) and _read_context_json. Options: --root (default the repository), --brief (condensed one-fact-per-line output from render_brief) and --json (the full report as JSON). By default it prints the brief followed by the JSON. It writes nothing and exits 0.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/project_status.py

#!/usr/bin/env python3
"""Artifact-sourced project status report (workflow-improvements-plan.md Phase 4).

Generates ONE status report purely from artifacts already on disk / already
computable — never from hand-maintained prose in plan files. Session-CLI
convention (see ``AGENTS.md`` "Plan files vs. state" — Phase 4.2): plan files
and ``session task`` hold INTENT (what's next); this script is STATE (what's
done/verified).

Sections:
  1. Version         — latest git tag + current session number (context.json,
                        READ ONLY: parsed with a bare ``json.loads()``, never
                        through ``ContextFileManager``'s self-healing writer
                        path, so this script can never mutate context.json).
  2. Test verdict     — newest run_tests-shaped JSON report under ``reports/``,
                        if one exists. None is currently produced by any gate
                        in this repo — reported honestly as "not found" rather
                        than fabricated.
  3. RTM summary       — grade counts from ``reports/rtm-evidence.json`` (a
                        multi-MB artifact; parsed once in-process here, never
                        surfaced whole to a caller).
  4. Coverage floor   — ``[tool.coverage.report] fail_under`` from
                        ``pyproject.toml``, if set.
  5. Infra            — one-liner via ``scripts/start_test_services.py
                        --status --json``.
  6. Drift warnings    — ``session task`` items (``deferred_work``, not
                        ``done``) open for more than 30 days. Plan-checkbox
                        drift (an item marked done in a ``plans/*.md`` file
                        whose linked verification command actually fails) has
                        no machine-readable checkbox->command registry yet in
                        this repo — reported explicitly as "not yet
                        mappable", not silently skipped.

CLI: ``python scripts/project_status.py [--root PATH] [--brief] [--json]``.
Exits 0 always (this is a report, not a gate).
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import tomllib
from datetime import datetime, timedelta
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent

PLAN_DRIFT_NOTE = (
    "plan-drift check: not yet mappable (no checkbox->verification registry)"
)

_INFRA_TIMEOUT_S = 15.0
_GIT_TIMEOUT_S = 5.0
_STALE_TASK_DAYS = 30


def _read_context_json(root: Path) -> dict | None:
    """Read ``context.json`` READ-ONLY — never through the self-healing writer."""
    path = root / "context.json"
    if not path.is_file():
        return None
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        return None
    return data if isinstance(data, dict) else None


def get_latest_tag(root: Path) -> str | None:
    """Most recently created SESSION tag, or ``None`` if none exist / git fails.

    Scoped to ``session-*``: the status brief keys session identity off the
    session-tag family, and other tag families (plugin-*, adr-*) interleave
    with it by creation date (measured 2026-10-18: the newest tag was
    ``plugin-94a23ae16dad``, which the brief must not report as a session).
    """
    try:
        proc = subprocess.run(
            [
                "git",
                "for-each-ref",
                "--sort=-creatordate",
                "--format=%(refname:short)",
                "refs/tags/session-*",
            ],
            cwd=root,
            capture_output=True,
            text=True,
            timeout=_GIT_TIMEOUT_S,
        )
    except subprocess.TimeoutExpired, OSError:
        return None
    if proc.returncode != 0:
        return None
    lines = proc.stdout.strip().splitlines()
    return lines[0] if lines else None


def get_session_number(root: Path) -> int | None:
    """Current active session number from ``context.json`` (read-only)."""
    data = _read_context_json(root)
    if data is None:
        return None
    state = data.get("state")
    if not isinstance(state, dict):
        return None
    session_num = state.get("active_session")
    return session_num if isinstance(session_num, int) else None


def get_test_verdict(root: Path) -> dict:
    """Newest run_tests-shaped JSON report under ``reports/``, if any exist."""
    reports_dir = root / "reports"
    if not reports_dir.is_dir():
        return {"found": False, "note": "no reports/ directory"}
    patterns = (
        "run-tests*.json",
        "run_tests*.json",
        "test-report*.json",
        "pytest-report*.json",
    )
    candidates = [
        p for pattern in patterns for p in reports_dir.glob(pattern) if p.is_file()
    ]
    if not candidates:
        return {
            "found": False,
            "note": (
                "no run_tests JSON report found in reports/ (produce one with "
                "`python scripts/run_tests.py --json > reports/run-tests-latest.json`)"
            ),
        }
    newest = max(candidates, key=lambda p: p.stat().st_mtime)
    try:
        payload = json.loads(newest.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as e:
        return {"found": True, "path": str(newest.relative_to(root)), "error": str(e)}
    return {"found": True, "path": str(newest.relative_to(root)), "report": payload}


def get_rtm_summary(root: Path) -> dict:
    """Grade counts from ``reports/rtm-evidence.json``."""
    path = root / "reports" / "rtm-evidence.json"
    if not path.is_file():
        return {"found": False}
    try:
        with path.open(encoding="utf-8") as fh:
            data = json.load(fh)
    except (OSError, json.JSONDecodeError) as e:
        return {"found": True, "error": str(e)}
    rows = data.get("rows", []) if isinstance(data, dict) else []
    grades: dict[str, int] = {}
    for row in rows:
        grade = row.get("grade", "UNKNOWN") if isinstance(row, dict) else "UNKNOWN"
        grades[grade] = grades.get(grade, 0) + 1
    return {
        "found": True,
        "verified": data.get("verified") if isinstance(data, dict) else None,
        "requirements": data.get("requirements") if isinstance(data, dict) else None,
        "rows": len(rows),
        "grades": grades,
    }


def get_coverage_floor(root: Path) -> float | None:
    """``[tool.coverage.report] fail_under`` from pyproject.toml, if set."""
    pyproject = root / "pyproject.toml"
    if not pyproject.is_file():
        return None
    try:
        with pyproject.open("rb") as fh:
            data = tomllib.load(fh)
    except OSError, tomllib.TOMLDecodeError:
        return None
    report_cfg = data.get("tool", {}).get("coverage", {}).get("report", {})
    floor = report_cfg.get("fail_under")
    return floor if isinstance(floor, (int, float)) else None


def get_infra_status(root: Path, timeout: float = _INFRA_TIMEOUT_S) -> dict:
    """One-liner 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 {"error": f"timed out after {timeout:.0f}s"}
    except OSError as e:
        return {"error": str(e)}
    try:
        report = json.loads(proc.stdout)
    except json.JSONDecodeError, ValueError:
        return {"error": f"could not parse infra report (rc={proc.returncode})"}
    return report if isinstance(report, dict) else {"error": "unexpected infra shape"}


def get_stale_tasks(root: Path, days: int = _STALE_TASK_DAYS) -> list[dict]:
    """Not-done ``deferred_work`` items open for more than ``days``. READ-ONLY."""
    data = _read_context_json(root)
    if data is None:
        return []
    cutoff = datetime.now() - timedelta(days=days)
    stale: list[dict] = []
    for item in data.get("deferred_work", []):
        if not isinstance(item, dict) or item.get("state") == "done":
            continue
        ts = item.get("timestamp")
        if not ts:
            continue
        try:
            when = datetime.fromisoformat(ts).replace(tzinfo=None)
        except ValueError:
            continue
        if when < cutoff:
            stale.append(
                {
                    "text": item.get("text", ""),
                    "state": item.get("state"),
                    "age_days": (datetime.now() - when).days,
                }
            )
    stale.sort(key=lambda s: s["age_days"], reverse=True)
    return stale


def build_report(root: Path) -> dict:
    """Assemble the full artifact-sourced status report."""
    return {
        "root": str(root),
        "latest_tag": get_latest_tag(root),
        "session_number": get_session_number(root),
        "test_verdict": get_test_verdict(root),
        "rtm_summary": get_rtm_summary(root),
        "coverage_floor": get_coverage_floor(root),
        "infra": get_infra_status(root),
        "stale_tasks": get_stale_tasks(root),
        "plan_drift_note": PLAN_DRIFT_NOTE,
    }


def render_brief(report: dict) -> str:
    """Condensed one-fact-per-line rendering, suitable for embedding (session prime)."""
    lines: list[str] = []

    session_num = report["session_number"]
    lines.append(
        f"tag: {report['latest_tag'] or 'none'} | "
        f"session: {session_num if session_num is not None else 'unknown'}"
    )

    tv = report["test_verdict"]
    if tv.get("found"):
        r = tv.get("report", {})
        if isinstance(r, dict) and ("passed" in r or "failed" in r):
            lines.append(
                f"tests: {tv['path']} — passed={r.get('passed', '?')} "
                f"failed={r.get('failed', '?')}"
            )
        else:
            lines.append(f"tests: {tv['path']} (unrecognized report schema)")
    else:
        lines.append(f"tests: {tv.get('note', 'unavailable')}")

    rtm = report["rtm_summary"]
    if rtm.get("found") and "error" not in rtm:
        grades = rtm.get("grades", {})
        grade_str = ", ".join(f"{k}={v}" for k, v in sorted(grades.items()))
        lines.append(f"rtm: {rtm.get('rows', 0)} rows ({grade_str})")
    elif rtm.get("found"):
        lines.append(
            f"rtm: error reading reports/rtm-evidence.json: {rtm.get('error')}"
        )
    else:
        lines.append("rtm: reports/rtm-evidence.json not found")

    floor = report["coverage_floor"]
    lines.append(f"coverage floor: {floor if floor is not None else 'not configured'}")

    infra = report["infra"]
    if "error" in infra:
        lines.append(f"infra: error: {infra['error']}")
    else:
        lines.append(
            f"infra: {'all ready' if infra.get('all_required_ok') else 'NOT all ready'}"
        )

    stale = report["stale_tasks"]
    lines.append(f"stale tasks (>{_STALE_TASK_DAYS}d open): {len(stale)}")
    lines.extend(f"  - [{s['age_days']}d] {s['text']}" for s in stale[:5])
    if len(stale) > 5:
        lines.append(f"  ... and {len(stale) - 5} more")

    lines.append(report["plan_drift_note"])
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Artifact-sourced project status report (never hand-maintained state)"
    )
    parser.add_argument(
        "--root",
        type=Path,
        default=REPO_ROOT,
        help="Project root (default: this script's repo).",
    )
    parser.add_argument(
        "--brief", action="store_true", help="Condensed one-fact-per-line output."
    )
    parser.add_argument(
        "--json", action="store_true", help="Full report as JSON (implies not --brief)."
    )
    args = parser.parse_args(argv)

    report = build_report(args.root)

    if args.json:
        print(json.dumps(report, indent=2, default=str))
    else:
        print(render_brief(report))
        if not args.brief:
            print()
            print(json.dumps(report, indent=2, default=str))

    return 0


if __name__ == "__main__":
    sys.exit(main())