427 lines
15 KiB
Python
427 lines
15 KiB
Python
"""Timmy Control Panel CLI — primary control surface for automations.
|
|
|
|
Usage:
|
|
timmyctl daily-run # Run the Daily Run orchestration
|
|
timmyctl log-run # Capture a Daily Run logbook entry
|
|
timmyctl inbox # Show what's "calling Timmy"
|
|
timmyctl config # Display key configuration
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import typer
|
|
import yaml
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
# Initialize Rich console for nice output
|
|
console = Console()
|
|
|
|
app = typer.Typer(
|
|
help="Timmy Control Panel — primary control surface for automations",
|
|
rich_markup_mode="rich",
|
|
)
|
|
|
|
# Default config paths
|
|
DEFAULT_CONFIG_DIR = Path("timmy_automations/config")
|
|
AUTOMATIONS_CONFIG = DEFAULT_CONFIG_DIR / "automations.json"
|
|
DAILY_RUN_CONFIG = DEFAULT_CONFIG_DIR / "daily_run.json"
|
|
TRIAGE_RULES_CONFIG = DEFAULT_CONFIG_DIR / "triage_rules.yaml"
|
|
|
|
GITEA_URL = os.environ.get("GITEA_URL", "http://143.198.27.163:3000")
|
|
GITEA_REPO = "rockachopa/Timmy-time-dashboard"
|
|
|
|
|
|
def _get_gitea_token() -> str | None:
|
|
"""Read the Gitea API token from env or config files."""
|
|
token = os.environ.get("GITEA_TOKEN")
|
|
if token:
|
|
return token.strip()
|
|
for candidate in [
|
|
Path("~/.hermes/gitea_token_vps").expanduser(),
|
|
Path("~/.hermes/gitea_token").expanduser(),
|
|
]:
|
|
try:
|
|
return candidate.read_text(encoding="utf-8").strip()
|
|
except FileNotFoundError:
|
|
continue
|
|
return None
|
|
|
|
|
|
def _gitea_api_get(endpoint: str) -> Any:
|
|
"""GET a Gitea API endpoint and return parsed JSON."""
|
|
url = f"{GITEA_URL}/api/v1{endpoint}"
|
|
token = _get_gitea_token()
|
|
req = urllib.request.Request(url)
|
|
if token:
|
|
req.add_header("Authorization", f"token {token}")
|
|
req.add_header("Accept", "application/json")
|
|
with urllib.request.urlopen(req, timeout=15) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
|
|
def _load_json_config(path: Path) -> dict[str, Any]:
|
|
"""Load a JSON config file, returning empty dict on error."""
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
return json.load(f)
|
|
except (FileNotFoundError, json.JSONDecodeError) as e:
|
|
console.print(f"[red]Error loading {path}: {e}[/red]")
|
|
return {}
|
|
|
|
|
|
def _load_yaml_config(path: Path) -> dict[str, Any]:
|
|
"""Load a YAML config file, returning empty dict on error."""
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
except (FileNotFoundError, yaml.YAMLError) as e:
|
|
console.print(f"[red]Error loading {path}: {e}[/red]")
|
|
return {}
|
|
|
|
|
|
def _get_config_dir() -> Path:
|
|
"""Return the config directory path."""
|
|
# Allow override via environment variable
|
|
env_dir = os.environ.get("TIMMY_CONFIG_DIR")
|
|
if env_dir:
|
|
return Path(env_dir)
|
|
return DEFAULT_CONFIG_DIR
|
|
|
|
|
|
@app.command()
|
|
def daily_run(
|
|
dry_run: bool = typer.Option(
|
|
False, "--dry-run", "-n", help="Show what would run without executing"
|
|
),
|
|
verbose: bool = typer.Option(False, "--verbose", "-v", help="Show detailed output"),
|
|
):
|
|
"""Run the Daily Run orchestration (agenda + summary).
|
|
|
|
Executes the daily run workflow including:
|
|
- Loop Guard checks
|
|
- Cycle Retrospective
|
|
- Triage scoring (if scheduled)
|
|
- Loop introspection (if scheduled)
|
|
"""
|
|
console.print("[bold green]Timmy Daily Run[/bold green]")
|
|
console.print()
|
|
|
|
config_path = _get_config_dir() / "daily_run.json"
|
|
config = _load_json_config(config_path)
|
|
|
|
if not config:
|
|
console.print("[yellow]No daily run configuration found.[/yellow]")
|
|
raise typer.Exit(1)
|
|
|
|
schedules = config.get("schedules", {})
|
|
triggers = config.get("triggers", {})
|
|
|
|
if verbose:
|
|
console.print(f"[dim]Config loaded from: {config_path}[/dim]")
|
|
console.print()
|
|
|
|
# Show the daily run schedule
|
|
table = Table(title="Daily Run Schedules")
|
|
table.add_column("Schedule", style="cyan")
|
|
table.add_column("Description", style="green")
|
|
table.add_column("Automations", style="yellow")
|
|
|
|
for schedule_name, schedule_data in schedules.items():
|
|
automations = schedule_data.get("automations", [])
|
|
table.add_row(
|
|
schedule_name,
|
|
schedule_data.get("description", ""),
|
|
", ".join(automations) if automations else "—",
|
|
)
|
|
|
|
console.print(table)
|
|
console.print()
|
|
|
|
# Show triggers
|
|
trigger_table = Table(title="Triggers")
|
|
trigger_table.add_column("Trigger", style="cyan")
|
|
trigger_table.add_column("Description", style="green")
|
|
trigger_table.add_column("Automations", style="yellow")
|
|
|
|
for trigger_name, trigger_data in triggers.items():
|
|
automations = trigger_data.get("automations", [])
|
|
trigger_table.add_row(
|
|
trigger_name,
|
|
trigger_data.get("description", ""),
|
|
", ".join(automations) if automations else "—",
|
|
)
|
|
|
|
console.print(trigger_table)
|
|
console.print()
|
|
|
|
if dry_run:
|
|
console.print("[yellow]Dry run mode — no actions executed.[/yellow]")
|
|
else:
|
|
console.print("[green]Executing daily run automations...[/green]")
|
|
auto_config_path = _get_config_dir() / "automations.json"
|
|
auto_config = _load_json_config(auto_config_path)
|
|
all_automations = auto_config.get("automations", [])
|
|
enabled = [a for a in all_automations if a.get("enabled", False)]
|
|
if not enabled:
|
|
console.print("[yellow]No enabled automations found.[/yellow]")
|
|
for auto in enabled:
|
|
cmd = auto.get("command")
|
|
name = auto.get("name", auto.get("id", "unnamed"))
|
|
if not cmd:
|
|
console.print(f"[yellow]Skipping {name} — no command defined.[/yellow]")
|
|
continue
|
|
console.print(f"[cyan]▶ Running: {name}[/cyan]")
|
|
if verbose:
|
|
console.print(f"[dim] $ {cmd}[/dim]")
|
|
try:
|
|
result = subprocess.run( # noqa: S602
|
|
cmd,
|
|
shell=True,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=120,
|
|
)
|
|
if result.stdout.strip():
|
|
console.print(result.stdout.strip())
|
|
if result.returncode != 0:
|
|
console.print(
|
|
f"[red] ✗ {name} exited with code {result.returncode}[/red]"
|
|
)
|
|
if result.stderr.strip():
|
|
console.print(f"[red]{result.stderr.strip()}[/red]")
|
|
else:
|
|
console.print(f"[green] ✓ {name} completed successfully[/green]")
|
|
except subprocess.TimeoutExpired:
|
|
console.print(f"[red] ✗ {name} timed out after 120s[/red]")
|
|
except Exception as exc:
|
|
console.print(f"[red] ✗ {name} failed: {exc}[/red]")
|
|
|
|
|
|
@app.command()
|
|
def log_run(
|
|
message: str = typer.Argument(..., help="Logbook entry message"),
|
|
category: str = typer.Option(
|
|
"general", "--category", "-c", help="Entry category (e.g., retro, todo, note)"
|
|
),
|
|
):
|
|
"""Capture a quick Daily Run logbook entry.
|
|
|
|
Logs a structured entry to the daily run logbook for later review.
|
|
Entries are timestamped and categorized automatically.
|
|
"""
|
|
from datetime import datetime
|
|
|
|
timestamp = datetime.now().isoformat()
|
|
|
|
console.print("[bold green]Daily Run Log Entry[/bold green]")
|
|
console.print()
|
|
console.print(f"[dim]Timestamp:[/dim] {timestamp}")
|
|
console.print(f"[dim]Category:[/dim] {category}")
|
|
console.print(f"[dim]Message:[/dim] {message}")
|
|
console.print()
|
|
|
|
logbook_path = Path(".loop/logbook.jsonl")
|
|
logbook_path.parent.mkdir(parents=True, exist_ok=True)
|
|
entry = json.dumps(
|
|
{"timestamp": timestamp, "category": category, "message": message}
|
|
)
|
|
with open(logbook_path, "a", encoding="utf-8") as f:
|
|
f.write(entry + "\n")
|
|
console.print(f"[green]✓[/green] Entry logged to {logbook_path}")
|
|
|
|
|
|
@app.command()
|
|
def inbox(
|
|
limit: int = typer.Option(10, "--limit", "-l", help="Maximum items to show"),
|
|
include_prs: bool = typer.Option(True, "--prs/--no-prs", help="Show open PRs"),
|
|
include_issues: bool = typer.Option(True, "--issues/--no-issues", help="Show relevant issues"),
|
|
):
|
|
"""Show what's "calling Timmy" — PRs, Daily Run items, alerts.
|
|
|
|
Displays a unified inbox of items requiring attention:
|
|
- Open pull requests awaiting review
|
|
- Daily run queue items
|
|
- Alerts and notifications
|
|
"""
|
|
console.print("[bold green]Timmy Inbox[/bold green]")
|
|
console.print()
|
|
|
|
# Load automations to show what's enabled
|
|
config_path = _get_config_dir() / "automations.json"
|
|
config = _load_json_config(config_path)
|
|
|
|
automations = config.get("automations", [])
|
|
enabled_automations = [a for a in automations if a.get("enabled", False)]
|
|
|
|
# Show automation status
|
|
auto_table = Table(title="Active Automations")
|
|
auto_table.add_column("ID", style="cyan")
|
|
auto_table.add_column("Name", style="green")
|
|
auto_table.add_column("Category", style="yellow")
|
|
auto_table.add_column("Trigger", style="magenta")
|
|
|
|
for auto in enabled_automations[:limit]:
|
|
auto_table.add_row(
|
|
auto.get("id", ""),
|
|
auto.get("name", ""),
|
|
"✓" if auto.get("enabled", False) else "✗",
|
|
auto.get("category", ""),
|
|
)
|
|
|
|
console.print(auto_table)
|
|
console.print()
|
|
|
|
if include_prs:
|
|
pr_table = Table(title="Open Pull Requests")
|
|
pr_table.add_column("#", style="cyan")
|
|
pr_table.add_column("Title", style="green")
|
|
pr_table.add_column("Author", style="yellow")
|
|
pr_table.add_column("Status", style="magenta")
|
|
try:
|
|
prs = _gitea_api_get(f"/repos/{GITEA_REPO}/pulls?state=open")
|
|
if prs:
|
|
for pr in prs[:limit]:
|
|
pr_table.add_row(
|
|
str(pr.get("number", "")),
|
|
pr.get("title", ""),
|
|
pr.get("user", {}).get("login", ""),
|
|
pr.get("state", ""),
|
|
)
|
|
else:
|
|
pr_table.add_row("—", "[dim]No open PRs[/dim]", "—", "—")
|
|
except Exception as exc:
|
|
pr_table.add_row(
|
|
"—", f"[red]Error fetching PRs: {exc}[/red]", "—", "—"
|
|
)
|
|
console.print(pr_table)
|
|
console.print()
|
|
|
|
if include_issues:
|
|
issue_table = Table(title="Issues Calling for Attention")
|
|
issue_table.add_column("#", style="cyan")
|
|
issue_table.add_column("Title", style="green")
|
|
issue_table.add_column("Type", style="yellow")
|
|
issue_table.add_column("Priority", style="magenta")
|
|
try:
|
|
issues = _gitea_api_get(
|
|
f"/repos/{GITEA_REPO}/issues?state=open&type=issues&limit={limit}"
|
|
)
|
|
if issues:
|
|
for issue in issues[:limit]:
|
|
labels = [lb.get("name", "") for lb in issue.get("labels", [])]
|
|
priority = next(
|
|
(lb for lb in labels if "priority" in lb.lower()),
|
|
"—",
|
|
)
|
|
issue_type = next(
|
|
(lb for lb in labels if lb.lower() in ("bug", "feature", "refactor", "enhancement")),
|
|
"—",
|
|
)
|
|
issue_table.add_row(
|
|
str(issue.get("number", "")),
|
|
issue.get("title", ""),
|
|
issue_type,
|
|
priority,
|
|
)
|
|
else:
|
|
issue_table.add_row("—", "[dim]No open issues[/dim]", "—", "—")
|
|
except Exception as exc:
|
|
issue_table.add_row(
|
|
"—", f"[red]Error fetching issues: {exc}[/red]", "—", "—"
|
|
)
|
|
console.print(issue_table)
|
|
console.print()
|
|
|
|
|
|
@app.command()
|
|
def config(
|
|
key: str | None = typer.Argument(None, help="Show specific config key (e.g., 'automations')"),
|
|
show_rules: bool = typer.Option(False, "--rules", "-r", help="Show triage rules overview"),
|
|
):
|
|
"""Display key configuration — labels, logbook issue ID, token rules overview.
|
|
|
|
Shows the current Timmy automation configuration including:
|
|
- Automation manifest
|
|
- Daily run schedules
|
|
- Triage scoring rules
|
|
"""
|
|
console.print("[bold green]Timmy Configuration[/bold green]")
|
|
console.print()
|
|
|
|
config_dir = _get_config_dir()
|
|
|
|
if key == "automations" or key is None:
|
|
auto_config = _load_json_config(config_dir / "automations.json")
|
|
automations = auto_config.get("automations", [])
|
|
|
|
table = Table(title="Automations Manifest")
|
|
table.add_column("ID", style="cyan")
|
|
table.add_column("Name", style="green")
|
|
table.add_column("Enabled", style="yellow")
|
|
table.add_column("Category", style="magenta")
|
|
|
|
for auto in automations:
|
|
enabled = "✓" if auto.get("enabled", False) else "✗"
|
|
table.add_row(
|
|
auto.get("id", ""),
|
|
auto.get("name", ""),
|
|
enabled,
|
|
auto.get("category", ""),
|
|
)
|
|
|
|
console.print(table)
|
|
console.print()
|
|
|
|
if key == "daily_run" or (key is None and not show_rules):
|
|
daily_config = _load_json_config(config_dir / "daily_run.json")
|
|
|
|
if daily_config:
|
|
console.print("[bold]Daily Run Configuration:[/bold]")
|
|
console.print(f"[dim]Version:[/dim] {daily_config.get('version', 'unknown')}")
|
|
console.print(f"[dim]Description:[/dim] {daily_config.get('description', '')}")
|
|
console.print()
|
|
|
|
if show_rules or key == "triage_rules":
|
|
rules_config = _load_yaml_config(config_dir / "triage_rules.yaml")
|
|
|
|
if rules_config:
|
|
thresholds = rules_config.get("thresholds", {})
|
|
console.print("[bold]Triage Scoring Rules:[/bold]")
|
|
console.print(f" Ready threshold: {thresholds.get('ready', 'N/A')}")
|
|
console.print(f" Excellent threshold: {thresholds.get('excellent', 'N/A')}")
|
|
console.print()
|
|
|
|
scope = rules_config.get("scope", {})
|
|
console.print("[bold]Scope Scoring:[/bold]")
|
|
console.print(f" Meta penalty: {scope.get('meta_penalty', 'N/A')}")
|
|
console.print()
|
|
|
|
alignment = rules_config.get("alignment", {})
|
|
console.print("[bold]Alignment Scoring:[/bold]")
|
|
console.print(f" Bug score: {alignment.get('bug_score', 'N/A')}")
|
|
console.print(f" Refactor score: {alignment.get('refactor_score', 'N/A')}")
|
|
console.print(f" Feature score: {alignment.get('feature_score', 'N/A')}")
|
|
console.print()
|
|
|
|
quarantine = rules_config.get("quarantine", {})
|
|
console.print("[bold]Quarantine Rules:[/bold]")
|
|
console.print(f" Failure threshold: {quarantine.get('failure_threshold', 'N/A')}")
|
|
console.print(f" Lookback cycles: {quarantine.get('lookback_cycles', 'N/A')}")
|
|
console.print()
|
|
|
|
|
|
def main():
|
|
"""Entry point for the timmyctl CLI."""
|
|
app()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|