nightly_learn.py

#!/usr/bin/env python3
"""Nightly compound learning — review closed issues, propose CLAUDE.md updates.

Runs as a cron/timer job. For each monitored repo with recently closed conductor
issues, creates a GitHub issue to update CLAUDE.md based on patterns from agent work.

Usage:
    python3 nightly_learn.py              # process last 24 hours
    python3 nightly_learn.py --hours 48   # process last 48 hours
    python3 nightly_learn.py --dry-run    # log only
"""

import argparse
import json
import logging
import os
import subprocess
import sys
from datetime import datetime, timezone, timedelta

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s",
    datefmt="%H:%M:%S",
)
log = logging.getLogger("nightly-learn")

REPOS_FILE = os.path.join(os.path.dirname(__file__), "repos.json")
CONDUCTOR_LABELS = {"conductor", "agent:opus", "agent:cursor", "agent:marketing"}


def load_repos() -> dict:
    with open(REPOS_FILE) as f:
        return json.load(f)


def get_closed_issues(repo: str, since_hours: int) -> list[dict]:
    """Fetch recently closed conductor issues for a repo."""
    since = (datetime.now(timezone.utc) - timedelta(hours=since_hours)).strftime("%Y-%m-%dT%H:%M:%SZ")
    try:
        result = subprocess.run(
            ["gh", "issue", "list", "--repo", repo, "--state", "closed",
             "--json", "number,title,body,labels,comments,closedAt",
             "--limit", "50"],
            capture_output=True, text=True, timeout=30,
        )
        if result.returncode != 0:
            return []
        issues = json.loads(result.stdout)
    except Exception:
        return []

    filtered = []
    for issue in issues:
        labels = {l.get("name", "") for l in issue.get("labels", [])}
        if not (CONDUCTOR_LABELS & labels):
            continue
        closed_at = issue.get("closedAt", "")
        if closed_at and closed_at >= since:
            filtered.append(issue)
    return filtered


COMPLETION_MARKERS = [
    # Protocol tool markers (from conductor complete())
    "completed", "summary", "done", "✅",
    # Common agent patterns
    "pr created", "pr opened", "merged", "pushed",
    "closes #", "fixes #",
]


def _find_completion_comment(comments: list[dict]) -> str:
    """Find the most likely completion comment from an issue's comments.

    Scans from newest to oldest. Prefers comments matching completion markers,
    falls back to the last non-bot-dispatch comment.
    """
    for c in reversed(comments):
        body = c.get("body", "")
        author = c.get("author", {}).get("login", "") if isinstance(c.get("author"), dict) else ""
        body_lower = body.lower()
        # Skip conductor dispatch comments
        if body.startswith("📡 Conductor dispatched"):
            continue
        if any(marker in body_lower for marker in COMPLETION_MARKERS):
            return body[:500]
    # Fallback: last substantive comment (skip dispatch comments)
    for c in reversed(comments):
        body = c.get("body", "")
        if body.startswith("📡 Conductor dispatched"):
            continue
        if len(body.strip()) > 20:
            return body[:500]
    return ""


def build_learning_body(repo: str, issues: list[dict]) -> str:
    """Build the issue body for a CLAUDE.md update dispatch."""
    issue_summaries = []
    for issue in issues:
        comments = issue.get("comments", [])
        completion = _find_completion_comment(comments)
        issue_summaries.append(
            f"### #{issue['number']}: {issue['title']}\n"
            f"Outcome: {completion or 'No summary found'}\n"
        )

    return (
        f"Review the following recently completed agent work in `{repo}` "
        f"and update CLAUDE.md with any useful lessons.\n\n"
        f"## Completed issues ({len(issues)})\n\n"
        + "".join(issue_summaries) +
        f"\n## Instructions\n\n"
        f"1. Read the completed issues above for patterns\n"
        f"2. Identify: common pitfalls, useful conventions, architecture notes\n"
        f"3. Update CLAUDE.md with concise, actionable additions\n"
        f"4. Do NOT remove existing content unless it's wrong\n"
        f"5. Keep additions brief — bullet points, not paragraphs\n"
        f"6. Skip this if there's nothing worth adding\n"
    )


def main():
    parser = argparse.ArgumentParser(description="Nightly compound learning")
    parser.add_argument("--hours", type=int, default=24)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    repos = load_repos()
    dispatched = 0

    for repo, config in repos.items():
        issues = get_closed_issues(repo, args.hours)
        if not issues:
            continue

        log.info("%s: %d closed conductor issues in last %dh", repo, len(issues), args.hours)

        body = build_learning_body(repo, issues)
        title = f"chore: Compound learning — update CLAUDE.md from {len(issues)} recent issue(s)"

        if args.dry_run:
            log.info("  [DRY RUN] Would dispatch: %s — %s", repo, title)
            continue

        try:
            result = subprocess.run(
                ["gh", "issue", "create",
                 "--repo", repo,
                 "--title", title,
                 "--body", body,
                 "--label", "conductor",
                 "--label", "documentation"],
                capture_output=True, text=True, timeout=30,
            )
            if result.returncode == 0:
                log.info("  Dispatched: %s", result.stdout.strip())
                dispatched += 1
            else:
                log.error("  Failed: %s", result.stderr.strip())
        except Exception as e:
            log.error("  Error: %s", e)

    log.info("Done. Dispatched %d learning issues.", dispatched)


if __name__ == "__main__":
    main()