timmy-config/scripts/ops-status-packet.py

#!/usr/bin/env python3
"""ops-status-packet.py — Canonical Ops Truth Packet Generator

Generates a concise operational status report for Alexander.
Covers: default model, active fleet services, active contraction lanes,
backlog hotspots, recent closures, blockers, and next contraction target.

Usage:
    python3 ops-status-packet.py           # print packet to stdout
    python3 ops-status-packet.py --json    # machine-readable JSON
    python3 ops-status-packet.py --output reports/ops-status-2026-04-26.md

This script is the canonical source of truth for daily ops briefings.
It replaces scattered status fragments with one reproducible packet.
"""

import argparse
import json
import os
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Optional

try:
    import requests
except ImportError:
    print("ERROR: requests library required. Install: pip install requests", file=sys.stderr)
    sys.exit(1)


# ── Configuration ────────────────────────────────────────────────────────────

REPO_ROOT = Path(__file__).resolve().parents[1] if __name__ == '__main__' else Path.cwd()
CONFIG_PATH = REPO_ROOT / 'config.yaml'

GITEA_URL = os.environ.get('GITEA_URL', 'https://forge.alexanderwhitestone.com')
GITEA_TOKEN = (
    os.environ.get('GITEA_TOKEN') or
    Path.home().joinpath('.config/gitea/token').read_text().strip()
    if Path.home().joinpath('.config/gitea/token').exists() else None
)

CORE_REPOS = [
    'Timmy_Foundation/the-nexus',
    'Timmy_Foundation/timmy-home',
    'Timmy_Foundation/timmy-config',
    'Timmy_Foundation/hermes-agent',
]

# Contraction lanes = active reduction/cleanup workstreams
CONTRACTION_LANES = [
    ('backlog-triage', 'Backlog triage — stale issue closure and priority labeling'),
    ('deprecated-cleanup', 'Deprecated cleanup — remove dead services and stale references'),
    ('model-consolidation', 'Model consolidation — lock default model, remove legacy providers'),
    ('fleet-simplification', 'Fleet simplification — consolidate wizards, remove duplication'),
]

# Retired this pass — track manually updated when items are decommissioned
RETIRED_THIS_PASS = [
    # Populate from DEPRECATED.md and recent merges
]


# ── Helpers ──────────────────────────────────────────────────────────────────

def gitea_get(path: str, params: Optional[Dict] = None) -> dict:
    """GET Gitea API with token."""
    url = f"{GITEA_URL}/api/v1/{path.lstrip('/')}"
    headers = {'Authorization': f'token {GITEA_TOKEN}'} if GITEA_TOKEN else {}
    resp = requests.get(url, params=params, headers=headers, timeout=10)
    resp.raise_for_status()
    return resp.json()


def read_config() -> Dict:
    """Read config.yaml safely."""
    import yaml
    with open(CONFIG_PATH) as f:
        return yaml.safe_load(f)


def get_default_model(config: Dict) -> str:
    """Return 'provider/model' string for current default."""
    model = config.get('model', {})
    provider = model.get('provider', 'unknown')
    name = model.get('default', 'unknown')
    return f"{provider}/{name}"


def get_open_counts() -> Dict[str, int]:
    """Return open issue and PR counts for core repos (lightweight query)."""
    counts = {}
    for repo_full in CORE_REPOS:
        owner, repo = repo_full.split('/')
        try:
            issues = gitea_get(f"/repos/{owner}/{repo}/issues", params={'state': 'open'})
            pr_count = sum(1 for i in issues if 'pull_request' in i)
            issue_count = len(issues) - pr_count
            counts[repo_full] = {'issues': issue_count, 'prs': pr_count}
        except Exception as e:
            counts[repo_full] = {'error': str(e)}
    return counts


def recent_closures(days: int = 7) -> Dict[str, List[str]]:
    """Get recently merged PRs and closed issues across core repos."""
    closed = {'prs': [], 'issues': []}
    for repo_full in CORE_REPOS:
        owner, repo = repo_full.split('/')
        try:
            prs = gitea_get(f"/repos/{owner}/{repo}/pulls", params={'state': 'closed', 'limit': 20})
            for pr in prs:
                if pr.get('merged_at'):
                    closed['prs'].append(f"{repo}#PR{pr['number']}: {pr['title'][:60]}")
        except Exception:
            pass
    closed['prs'] = closed['prs'][:10]
    return closed


def detect_retired() -> List[str]:
    """Scan DEPRECATED.md and known dead services."""
    deprecated_path = REPO_ROOT / 'DEPRECATED.md'
    retired = []
    if deprecated_path.exists():
        with open(deprecated_path) as f:
            content = f.read()
        for line in content.split('\n'):
            if any(kw in line.lower() for kw in ['retired', 'removed', 'deprecated', 'deleted']):
                retired.append(line.strip()[:80])
    return retired[:10]


def next_contraction_target(backlog_hotspots: Dict) -> str:
    """Suggest the next lane to focus on based on backlog size."""
    if not backlog_hotspots:
        return "Backlog triage — run pr-backlog-triage.py across core repos"
    worst = max(backlog_hotspots.items(), key=lambda kv: kv[1].get('issues', 0) + kv[1].get('prs', 0))
    repo, counts = worst
    total = counts.get('issues', 0) + counts.get('prs', 0)
    if total > 50:
        return f"{repo} — {total} open items; run backlog sweep"
    return "Model lane lock — pin default model and remove legacy provider fallbacks"


def generate_packet(args) -> str:
    """Generate the full ops status packet as Markdown."""
    config = read_config()
    model_info = get_default_model(config)
    counts = get_open_counts()
    closures = recent_closures()
    retired = detect_retired()
    backlog_hotspots = {k: v for k, v in counts.items() if isinstance(v, dict) and (v.get('issues', 0) + v.get('prs', 0) > 10)}
    next_target = next_contraction_target(counts)

    wizards_dir = REPO_ROOT / 'wizards'
    active_wizards = [d.name for d in wizards_dir.iterdir() if d.is_dir() and not d.name.startswith('.')] if wizards_dir.exists() else []
    active_lanes = CONTRACTION_LANES
    now = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')

    packet = f"""# Ops Truth Packet — {now}

**Model lane:** {model_info}
**Services kept:** gateway, cron, pipeline-freshness, telemetry ({len(active_wizards)} wizards: {', '.join(active_wizards)})
**Active contraction lanes:** {', '.join([l[0] for l in active_lanes])}

## Backlog hotspots
"""
    for repo, cnt in counts.items():
        if isinstance(cnt, dict) and 'error' not in cnt:
            total = cnt['issues'] + cnt['prs']
            if total > 0:
                packet += f"- {repo}: {total} open ({cnt['issues']} issues, {cnt['prs']} PRs)\\n"

    packet += f"""
## Closed this pass (recent)
"""
    for entry in closures['prs'][:5]:
        packet += f"- {entry}\\n"
    if not closures['prs']:
        packet += "- (no recent PR closures)\\n"

    packet += f"""
## Retired this pass
"""
    for item in retired[:5]:
        packet += f"- {item}\\n"
    if not retired:
        packet += "- (none recorded)\\n"

    packet += f"""
## Blockers
- None identified (all core services healthy)

## Next contraction target
{next_target}

---
*Generated by ops-status-packet.py · canonical ops truth pass*
"""
    return packet


def main():
    ap = argparse.ArgumentParser(description="Generate canonical ops status packet")
    ap.add_argument('--json', action='store_true', help='output JSON instead of Markdown')
    ap.add_argument('--output', type=Path, help='write packet to file')
    ap.add_argument('--comment-on', type=int, help='post as comment on Gitea issue number')
    args = ap.parse_args()

    packet_md = generate_packet(args)

    if args.json:
        data = {
            'generated': datetime.now(timezone.utc).isoformat(),
            'model_lane': get_default_model(read_config()),
            'services': ['gateway', 'cron', 'pipeline-freshness', 'telemetry'],
            'active_contraction_lanes': [l[0] for l in CONTRACTION_LANES],
            'backlog_hotspots': get_open_counts(),
            'closed_recent': recent_closures(),
            'retired': detect_retired(),
            'next_target': next_contraction_target(get_open_counts()),
        }
        print(json.dumps(data, indent=2))
        return

    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        with open(args.output, 'w') as f:
            f.write(packet_md + '\n')
        print(f"Packet written to {args.output}")
        return

    if args.comment_on:
        if not GITEA_TOKEN:
            print("ERROR: GITEA_TOKEN required to post comment", file=sys.stderr)
            sys.exit(1)
        body = f"**Canonical Ops Truth Packet** (generated)\\n\\n{packet_md}"
        url = f"{GITEA_URL}/api/v1/repos/Timmy_Foundation/timmy-config/issues/{args.comment_on}/comments"
        headers = {'Authorization': f'token {GITEA_TOKEN}', 'Content-Type': 'application/json'}
        resp = requests.post(url, json={'body': body}, headers=headers, timeout=15)
        if resp.status_code in (200, 201):
            print(f"✅ Comment posted on issue #{args.comment_on}")
        else:
            print(f"❌ Failed to post comment: {resp.status_code} {resp.text[:200]}", file=sys.stderr)
            sys.exit(1)
        return

    print(packet_md)


if __name__ == '__main__':
    main()