Gh Trace Miner

interface:
  display_name: "GH Trace Miner"
  short_description: "Mine GH Actions agent traces."
  default_prompt: "Mine GitHub Actions agent traces for watched ontology repositories."

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import concurrent.futures
import json
import re
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any

import mine_traces


TRACE_JOB_RE = re.compile(r"(respond|claude|agent|goose)", re.I)


def main() -> int:
    args = parse_args()
    out_dir = Path(args.out_dir) / args.slug
    out_dir.mkdir(parents=True, exist_ok=True)

    runs = load_runs(args)
    original_run_count = len(runs)
    runs = filter_runs(runs, created_after=args.created_after, skip_run_skipped=not args.include_run_skipped)
    token = gh_token()
    job_results = {} if args.assume_trace_job else fetch_jobs_for_runs(args.repo, runs, token, args.workers)

    trace_summaries: list[dict[str, Any]] = []
    skipped_runs: list[dict[str, Any]] = []
    fetch_errors: list[dict[str, str]] = []
    trace_run_jobs = 0

    mine_inputs: list[tuple[dict[str, Any], dict[str, Any]]] = []
    for run in runs:
        run_id = str(run["databaseId"])
        trace_job = {"name": "assumed-trace-job"} if args.assume_trace_job else None
        if not trace_job:
            result = job_results.get(run_id, {})
            if result.get("error"):
                fetch_errors.append({"run_id": run_id, "error": result["error"]})
                continue
            trace_job = first_trace_job(result.get("jobs", []))
        if not trace_job:
            skipped_runs.append(skipped_summary(args.repo, run, "no non-skipped trace-like job"))
            continue
        trace_run_jobs += 1
        mine_inputs.append((run, trace_job))

    with concurrent.futures.ThreadPoolExecutor(max_workers=args.mine_workers) as executor:
        future_to_run = {
            executor.submit(
                mine_trace_run,
                args.repo,
                run,
                out_dir / str(run["databaseId"]),
                args.skip_artifact_check,
            ): str(run["databaseId"])
            for run, _trace_job in mine_inputs
        }
        for future in concurrent.futures.as_completed(future_to_run):
            run_id = future_to_run[future]
            try:
                summary = future.result()
            except Exception as exc:  # noqa: BLE001
                fetch_errors.append({"run_id": run_id, "error": str(exc)})
                continue
            if summary["trace_record_count"] or summary["artifact_trace_files"]:
                trace_summaries.append(summary)
            else:
                skipped_runs.append(summary)

    index = {
        "repo": args.repo,
        "slug": args.slug,
        "workflow": args.workflow,
        "original_run_count": original_run_count,
        "created_after": args.created_after,
        "candidate_run_count": len(runs),
        "job_inspected_count": len(job_results),
        "trace_job_run_count": trace_run_jobs,
        "trace_run_count": len(trace_summaries),
        "skipped_run_count": len(skipped_runs),
        "fetch_error_count": len(fetch_errors),
        "trace_summaries": trace_summaries,
        "skipped_runs": skipped_runs,
        "fetch_errors": fetch_errors,
    }
    mine_traces.write_json(out_dir / "index.json", index)
    print(
        f"{args.slug}: runs={len(runs)} trace_jobs={index['trace_job_run_count']} "
        f"traces={len(trace_summaries)} skipped={len(skipped_runs)} errors={len(fetch_errors)}"
    )
    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Concurrent full-history job scan for log-only Actions workflows.")
    parser.add_argument("--repo", required=True, help="Repository owner/name.")
    parser.add_argument("--slug", required=True, help="Output slug.")
    parser.add_argument("--workflow", required=True, help="Workflow file/name to scan.")
    parser.add_argument("--runs-json", default="", help="Optional paginated gh api workflow-runs JSON file.")
    parser.add_argument("--out-dir", default="build/full-agent-history/actions", help="Output parent directory.")
    parser.add_argument("--workers", type=int, default=12, help="Concurrent job metadata requests.")
    parser.add_argument("--mine-workers", type=int, default=4, help="Concurrent log/artifact mining requests.")
    parser.add_argument("--skip-artifact-check", action="store_true", help="Skip artifact lookup when mining matching runs.")
    parser.add_argument("--created-after", default="", help="Only inspect runs created at or after this ISO timestamp/date.")
    parser.add_argument("--include-run-skipped", action="store_true", help="Do not prefilter run-level skipped conclusions.")
    parser.add_argument("--assume-trace-job", action="store_true", help="Mine every filtered run without fetching job metadata.")
    return parser.parse_args()


def mine_trace_run(
    repo: str,
    run: dict[str, Any],
    run_dir: Path,
    skip_artifact_check: bool,
) -> dict[str, Any]:
    summary_path = run_dir / "summary.json"
    if summary_path.exists():
        return json.loads(summary_path.read_text(encoding="utf-8"))
    if skip_artifact_check:
        return mine_log_only_run(repo, run, run_dir)
    return mine_traces.mine_run(repo, run, run_dir, skip_artifact_check=skip_artifact_check)


def mine_log_only_run(repo: str, run: dict[str, Any], run_dir: Path) -> dict[str, Any]:
    mine_traces.reset_dir(run_dir)
    mine_traces.write_json(run_dir / "run.json", run)
    log_error = ""
    try:
        log_trace_count, session_ids, type_counts = mine_traces.mine_logs(repo, str(run["databaseId"]), run_dir)
    except Exception as exc:  # noqa: BLE001
        log_trace_count = 0
        session_ids = set()
        type_counts = {}
        log_error = str(exc)

    summary = {
        "repo": repo,
        "run_id": str(run["databaseId"]),
        "run_url": run.get("url"),
        "created_at": run.get("createdAt"),
        "title": run.get("displayTitle"),
        "event": run.get("event"),
        "conclusion": run.get("conclusion"),
        "trace_job": {"name": "assumed-trace-job"},
        "artifact_trace_files": [],
        "trace_record_count": log_trace_count,
        "session_ids": sorted(session_ids),
        "type_counts": dict(sorted(type_counts.items())),
        "log_error": log_error,
    }
    mine_traces.write_json(run_dir / "summary.json", summary)
    return summary


def load_runs(args: argparse.Namespace) -> list[dict[str, Any]]:
    if args.runs_json:
        payloads = load_json_stream(Path(args.runs_json).read_text(encoding="utf-8"))
        runs = [
            mine_traces.normalize_api_run(run, args.workflow)
            for payload in payloads
            for run in payload.get("workflow_runs", [])
        ]
        return mine_traces.dedupe_runs(runs)
    return mine_traces.discover_runs(args.repo, [args.workflow], 0, all_runs=True, max_pages=0)


def filter_runs(
    runs: list[dict[str, Any]],
    *,
    created_after: str,
    skip_run_skipped: bool,
) -> list[dict[str, Any]]:
    filtered: list[dict[str, Any]] = []
    cutoff = normalize_cutoff(created_after)
    for run in runs:
        if skip_run_skipped and run.get("conclusion") == "skipped":
            continue
        created_at = run.get("createdAt") or ""
        if cutoff and created_at < cutoff:
            continue
        filtered.append(run)
    return filtered


def normalize_cutoff(value: str) -> str:
    if not value:
        return ""
    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        return f"{value}T00:00:00Z"
    return value


def load_json_stream(text: str) -> list[dict[str, Any]]:
    decoder = json.JSONDecoder()
    index = 0
    payloads: list[dict[str, Any]] = []
    while index < len(text):
        while index < len(text) and text[index].isspace():
            index += 1
        if index >= len(text):
            break
        payload, index = decoder.raw_decode(text, index)
        if isinstance(payload, dict):
            payloads.append(payload)
    return payloads


def gh_token() -> str:
    proc = subprocess.run(["gh", "auth", "token"], text=True, capture_output=True, check=True)
    return proc.stdout.strip()


def fetch_jobs_for_runs(
    repo: str,
    runs: list[dict[str, Any]],
    token: str,
    workers: int,
) -> dict[str, dict[str, Any]]:
    results: dict[str, dict[str, Any]] = {}
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        future_to_run = {
            executor.submit(fetch_jobs, repo, str(run["databaseId"]), token): str(run["databaseId"])
            for run in runs
        }
        for future in concurrent.futures.as_completed(future_to_run):
            run_id = future_to_run[future]
            try:
                results[run_id] = {"jobs": future.result()}
            except Exception as exc:  # noqa: BLE001
                results[run_id] = {"jobs": [], "error": str(exc)}
    return results


def fetch_jobs(repo: str, run_id: str, token: str) -> list[dict[str, Any]]:
    jobs: list[dict[str, Any]] = []
    page = 1
    while True:
        payload = fetch_json(f"https://api.github.com/repos/{repo}/actions/runs/{run_id}/jobs", token, page)
        jobs.extend(payload.get("jobs", []))
        if len(jobs) >= payload.get("total_count", len(jobs)):
            return jobs
        page += 1


def fetch_json(url: str, token: str, page: int) -> dict[str, Any]:
    query = urllib.parse.urlencode({"per_page": 100, "page": page})
    request = urllib.request.Request(
        f"{url}?{query}",
        headers={
            "Accept": "application/vnd.github+json",
            "Authorization": f"Bearer {token}",
            "X-GitHub-Api-Version": "2022-11-28",
            "User-Agent": "gh-trace-miner",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code}: {body[:300]}") from exc


def first_trace_job(jobs: list[dict[str, Any]]) -> dict[str, Any] | None:
    for job in jobs:
        if job.get("conclusion") == "skipped":
            continue
        if TRACE_JOB_RE.search(job.get("name", "")):
            return job
    return None


def skipped_summary(repo: str, run: dict[str, Any], reason: str) -> dict[str, Any]:
    return {
        "repo": repo,
        "run_id": str(run["databaseId"]),
        "run_url": run.get("url"),
        "created_at": run.get("createdAt"),
        "title": run.get("displayTitle"),
        "event": run.get("event"),
        "conclusion": run.get("conclusion"),
        "artifact_trace_files": [],
        "trace_record_count": 0,
        "session_ids": [],
        "type_counts": {},
        "skipped_reason": reason,
    }


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any

import mine_traces


SESSION_RE = re.compile(
    r"(?:/sessions/|session_id=)([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})",
    re.I,
)
COPILOT_AUTHOR_LOGINS = {"app/copilot-swe-agent", "Copilot", "copilot-swe-agent[bot]"}
AGENT_TASK_JSON_FIELDS = (
    "completedAt,createdAt,id,name,pullRequestNumber,pullRequestState,"
    "pullRequestTitle,pullRequestUrl,repository,state,updatedAt,user"
)


def main() -> int:
    args = parse_args()
    targets = load_targets(args)
    if not targets:
        print("No targets selected.", file=sys.stderr)
        return 1

    gh_agent = find_agent_task_gh(args.gh)
    if not gh_agent:
        print(
            "No GitHub CLI with agent-task support found. Install gh >= 2.80.0 "
            "or pass --gh /path/to/new/gh.",
            file=sys.stderr,
        )
        return 1

    out_dir = Path(args.out_dir)
    for target in targets:
        mine_repo(
            repo=target["repo"],
            slug=target["slug"],
            out_dir=out_dir,
            gh_agent=gh_agent,
            limit_prs=args.limit_prs,
            limit_agent_tasks=args.limit_agent_tasks,
            max_sessions=args.max_sessions,
            skip_logs=args.skip_logs,
        )
    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Mine GitHub Copilot coding-agent session logs for PRs.")
    parser.add_argument("--config", default="", help="agent-watcher config/targets.json path.")
    parser.add_argument("--all-targets", action="store_true", help="Mine every target in --config.")
    parser.add_argument("--target", action="append", default=[], help="Configured repo to include from --config.")
    parser.add_argument("--repo", action="append", default=[], help="Repository owner/name. Repeatable.")
    parser.add_argument("--slug", default="", help="Ontology output slug for a single --repo run.")
    parser.add_argument("--out-dir", default="build/copilot-traces", help="Output directory.")
    parser.add_argument("--limit-prs", type=int, default=300, help="Recent PRs to inspect per repo.")
    parser.add_argument("--limit-agent-tasks", type=int, default=200, help="Visible agent tasks to list.")
    parser.add_argument("--max-sessions", type=int, default=0, help="Maximum sessions to download per repo; 0 means no limit.")
    parser.add_argument("--skip-logs", action="store_true", help="Write session metadata without downloading agent.log.")
    parser.add_argument("--gh", default="", help="Path to a GitHub CLI binary with gh agent-task support.")
    return parser.parse_args()


def load_targets(args: argparse.Namespace) -> list[dict[str, str]]:
    targets: list[dict[str, str]] = []
    if args.config:
        config = json.loads(Path(args.config).read_text())
        selected = set(args.target)
        for item in config.get("targets", []):
            repo = item["repo"]
            if not args.all_targets and repo not in selected:
                continue
            targets.append({"repo": repo, "slug": item.get("short_name") or repo.split("/")[-1]})

    for repo in args.repo:
        slug = args.slug if len(args.repo) == 1 and args.slug else repo.split("/")[-1]
        targets.append({"repo": repo, "slug": slug})

    seen: set[str] = set()
    deduped: list[dict[str, str]] = []
    for target in targets:
        if target["repo"] in seen:
            continue
        seen.add(target["repo"])
        deduped.append(target)
    return deduped


def find_agent_task_gh(explicit: str) -> str:
    candidates: list[str] = []
    if explicit:
        candidates.append(explicit)
    candidates.append("gh")

    script_path = Path(__file__).resolve()
    search_roots = [Path.cwd(), *script_path.parents]
    for root in search_roots:
        candidates.extend(
            str(path)
            for path in sorted((root / "build" / "tools").glob("gh-*/gh_*/bin/gh"), reverse=True)
        )

    seen: set[str] = set()
    for candidate in candidates:
        if candidate in seen:
            continue
        seen.add(candidate)
        if shutil.which(candidate) or Path(candidate).exists():
            proc = subprocess.run(
                [candidate, "agent-task", "view", "--help"],
                text=True,
                capture_output=True,
                check=False,
            )
            if proc.returncode == 0:
                return candidate
    return ""


def mine_repo(
    *,
    repo: str,
    slug: str,
    out_dir: Path,
    gh_agent: str,
    limit_prs: int,
    limit_agent_tasks: int,
    max_sessions: int,
    skip_logs: bool,
) -> None:
    repo_dir = out_dir / slug
    repo_dir.mkdir(parents=True, exist_ok=True)
    prs = list_copilot_prs(repo, limit_prs)
    visible_sessions = list_visible_agent_tasks(gh_agent, repo, limit_agent_tasks)
    prs = add_visible_task_prs(repo, prs, visible_sessions)
    sessions_by_pr = sessions_by_pull_request(visible_sessions)

    results: list[dict[str, Any]] = []
    downloaded = 0
    session_cache: dict[str, dict[str, Any]] = {}

    for pr in prs:
        if max_sessions and downloaded >= max_sessions:
            results.append(skipped_pr_result(pr, "max sessions reached"))
            continue

        context = get_pr_context(repo, int(pr["number"]))
        candidate_sessions = extract_sessions_from_pr(pr, context)
        for session_id in sessions_by_pr.get(int(pr["number"]), []):
            candidate_sessions.setdefault(session_id, []).append("agent-task-list")

        result, count = mine_pr_sessions(
            gh_agent=gh_agent,
            repo=repo,
            pr=pr,
            context=context,
            repo_dir=repo_dir,
            candidate_sessions=candidate_sessions,
            session_cache=session_cache,
            remaining_sessions=max_sessions - downloaded if max_sessions else 0,
            skip_logs=skip_logs,
        )
        downloaded += count
        results.append(result)

    index = {
        "repo": repo,
        "slug": slug,
        "copilot_pr_count": len(prs),
        "visible_agent_task_count": len(visible_sessions),
        "session_pr_count": sum(1 for result in results if result["sessions"]),
        "session_count": sum(len(result["sessions"]) for result in results),
        "log_count": sum(1 for result in results for session in result["sessions"] if session.get("log_path")),
        "prs": results,
    }
    mine_traces.write_json(repo_dir / "agent-tasks-visible.json", visible_sessions)
    mine_traces.write_json(repo_dir / "index.json", index)
    print(
        f"{slug}: copilot_prs={len(prs)} sessions={index['session_count']} "
        f"logs={index['log_count']} visible_tasks={len(visible_sessions)}"
    )


def list_copilot_prs(repo: str, limit: int) -> list[dict[str, Any]]:
    proc = subprocess.run(
        [
            "gh",
            "pr",
            "list",
            "--repo",
            repo,
            "--state",
            "all",
            "--limit",
            str(limit),
            "--json",
            "number,title,state,createdAt,updatedAt,url,headRefName,author,body",
        ],
        text=True,
        capture_output=True,
        check=True,
    )
    prs = json.loads(proc.stdout or "[]")
    return [pr for pr in prs if is_copilot_pr(pr)]


def add_visible_task_prs(
    repo: str,
    prs: list[dict[str, Any]],
    visible_sessions: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    prs_by_number = {int(pr["number"]): pr for pr in prs}
    for task in visible_sessions:
        number = task.get("pullRequestNumber")
        if not isinstance(number, int) or number in prs_by_number:
            continue
        pr = fetch_pr_summary(repo, number)
        if pr:
            prs_by_number[number] = pr
    return sorted(prs_by_number.values(), key=lambda pr: int(pr["number"]), reverse=True)


def fetch_pr_summary(repo: str, number: int) -> dict[str, Any] | None:
    proc = subprocess.run(
        [
            "gh",
            "pr",
            "view",
            str(number),
            "--repo",
            repo,
            "--json",
            "number,title,state,createdAt,updatedAt,url,headRefName,author,body",
        ],
        text=True,
        capture_output=True,
        check=False,
    )
    if proc.returncode != 0:
        return None
    return json.loads(proc.stdout or "{}")


def is_copilot_pr(pr: dict[str, Any]) -> bool:
    author_login = ((pr.get("author") or {}).get("login") or "").lower()
    head_ref = (pr.get("headRefName") or "").lower()
    return author_login in {login.lower() for login in COPILOT_AUTHOR_LOGINS} or head_ref.startswith("copilot/")


def get_pr_context(repo: str, number: int) -> dict[str, Any]:
    proc = subprocess.run(
        ["gh", "pr", "view", str(number), "--repo", repo, "--json", "body,comments,reviews,commits"],
        text=True,
        capture_output=True,
        check=True,
    )
    return json.loads(proc.stdout or "{}")


def extract_sessions_from_pr(pr: dict[str, Any], context: dict[str, Any]) -> dict[str, list[str]]:
    sources: dict[str, list[str]] = {}
    add_sessions(sources, pr.get("body") or "", "pr-body")
    add_sessions(sources, context.get("body") or "", "pr-view-body")

    for index, comment in enumerate(context.get("comments", []), start=1):
        add_sessions(sources, comment.get("body") or "", f"comment-{index}")
    for index, review in enumerate(context.get("reviews", []), start=1):
        add_sessions(sources, review.get("body") or "", f"review-{index}")
    for index, commit in enumerate(context.get("commits", []), start=1):
        add_sessions(sources, commit.get("messageBody") or "", f"commit-{index}")
        add_sessions(sources, commit.get("messageHeadline") or "", f"commit-{index}-headline")
    return sources


def add_sessions(sources: dict[str, list[str]], text: str, source: str) -> None:
    for session_id in SESSION_RE.findall(text):
        sources.setdefault(session_id.lower(), []).append(source)


def list_visible_agent_tasks(gh_agent: str, repo: str, limit: int) -> list[dict[str, Any]]:
    proc = subprocess.run(
        [gh_agent, "agent-task", "list", "--limit", str(limit), "--json", AGENT_TASK_JSON_FIELDS],
        text=True,
        capture_output=True,
        check=False,
    )
    if proc.returncode != 0:
        return []
    tasks = json.loads(proc.stdout or "[]")
    return [task for task in tasks if task.get("repository") == repo]


def sessions_by_pull_request(tasks: list[dict[str, Any]]) -> dict[int, list[str]]:
    grouped: dict[int, list[str]] = {}
    for task in tasks:
        number = task.get("pullRequestNumber")
        session_id = task.get("id")
        if isinstance(number, int) and isinstance(session_id, str):
            grouped.setdefault(number, []).append(session_id.lower())
    return grouped


def mine_pr_sessions(
    *,
    gh_agent: str,
    repo: str,
    pr: dict[str, Any],
    context: dict[str, Any],
    repo_dir: Path,
    candidate_sessions: dict[str, list[str]],
    session_cache: dict[str, dict[str, Any]],
    remaining_sessions: int,
    skip_logs: bool,
) -> tuple[dict[str, Any], int]:
    pr_dir = repo_dir / f"pr-{pr['number']}"
    pr_dir.mkdir(parents=True, exist_ok=True)
    pr_record = pr_record_from_context(pr, context, candidate_sessions)
    mine_traces.write_json(pr_dir / "pr.json", pr_record)

    sessions: list[dict[str, Any]] = []
    external_sessions: list[dict[str, Any]] = []
    errors: list[str] = []
    downloaded = 0

    for session_id, sources in sorted(candidate_sessions.items()):
        if remaining_sessions and downloaded >= remaining_sessions:
            break
        try:
            session_meta = view_session(gh_agent, repo, session_id, session_cache)
        except subprocess.CalledProcessError as exc:
            errors.append(f"{session_id}: {exc.stderr.strip() or exc.stdout.strip() or exc}")
            continue

        if session_meta.get("pullRequestNumber") != pr["number"]:
            external_sessions.append(session_summary(session_meta, sources, "", "session belongs to another PR"))
            continue

        session_dir = pr_dir / f"session-{session_id}"
        session_dir.mkdir(parents=True, exist_ok=True)
        mine_traces.write_json(session_dir / "session.json", session_meta)
        log_path = ""
        log_bytes = 0
        log_lines = 0
        if not skip_logs:
            try:
                log_path, log_bytes, log_lines = write_session_log(gh_agent, repo, session_id, session_dir)
            except subprocess.CalledProcessError as exc:
                errors.append(f"{session_id} log: {exc.stderr.strip() or exc.stdout.strip() or exc}")
        summary = session_summary(session_meta, sources, str(Path(f"session-{session_id}") / "agent.log") if log_path else "", "")
        summary["log_bytes"] = log_bytes
        summary["log_lines"] = log_lines
        sessions.append(summary)
        downloaded += 1

    result = {
        **pr_record,
        "sessions": sessions,
        "external_sessions": external_sessions,
        "errors": errors,
        "missing_reason": "" if sessions else missing_reason(candidate_sessions, external_sessions, errors),
    }
    mine_traces.write_json(pr_dir / "summary.json", result)
    return result, downloaded


def pr_record_from_context(
    pr: dict[str, Any],
    context: dict[str, Any],
    candidate_sessions: dict[str, list[str]],
) -> dict[str, Any]:
    return {
        "number": pr["number"],
        "title": pr["title"],
        "state": pr["state"],
        "created_at": pr["createdAt"],
        "updated_at": pr["updatedAt"],
        "url": pr["url"],
        "head_ref": pr["headRefName"],
        "author": (pr.get("author") or {}).get("login"),
        "candidate_sessions": [{"id": key, "sources": value} for key, value in sorted(candidate_sessions.items())],
        "comment_count": len(context.get("comments", [])),
        "review_count": len(context.get("reviews", [])),
        "commit_count": len(context.get("commits", [])),
    }


def view_session(
    gh_agent: str,
    repo: str,
    session_id: str,
    cache: dict[str, dict[str, Any]],
) -> dict[str, Any]:
    if session_id in cache:
        return cache[session_id]
    proc = subprocess.run(
        [gh_agent, "agent-task", "view", "--repo", repo, session_id, "--json", AGENT_TASK_JSON_FIELDS],
        text=True,
        capture_output=True,
        check=True,
    )
    payload = json.loads(proc.stdout or "{}")
    cache[session_id] = payload
    return payload


def write_session_log(gh_agent: str, repo: str, session_id: str, session_dir: Path) -> tuple[str, int, int]:
    proc = subprocess.run(
        [gh_agent, "agent-task", "view", "--repo", repo, session_id, "--log"],
        text=True,
        capture_output=True,
        check=True,
    )
    path = session_dir / "agent.log"
    path.write_text(proc.stdout, encoding="utf-8")
    return str(path), len(proc.stdout.encode("utf-8")), proc.stdout.count("\n")


def session_summary(session_meta: dict[str, Any], sources: list[str], log_path: str, note: str) -> dict[str, Any]:
    return {
        "id": session_meta.get("id"),
        "name": session_meta.get("name"),
        "state": session_meta.get("state"),
        "repository": session_meta.get("repository"),
        "pull_request_number": session_meta.get("pullRequestNumber"),
        "pull_request_url": session_meta.get("pullRequestUrl"),
        "user": session_meta.get("user"),
        "created_at": session_meta.get("createdAt"),
        "updated_at": session_meta.get("updatedAt"),
        "sources": sorted(set(sources)),
        "log_path": log_path,
        "note": note,
    }


def skipped_pr_result(pr: dict[str, Any], reason: str) -> dict[str, Any]:
    return {
        "number": pr["number"],
        "title": pr["title"],
        "state": pr["state"],
        "created_at": pr["createdAt"],
        "updated_at": pr["updatedAt"],
        "url": pr["url"],
        "head_ref": pr["headRefName"],
        "author": (pr.get("author") or {}).get("login"),
        "candidate_sessions": [],
        "comment_count": 0,
        "review_count": 0,
        "commit_count": 0,
        "sessions": [],
        "external_sessions": [],
        "errors": [],
        "missing_reason": reason,
    }


def missing_reason(
    candidate_sessions: dict[str, list[str]],
    external_sessions: list[dict[str, Any]],
    errors: list[str],
) -> str:
    if external_sessions and not candidate_sessions:
        return "only external sessions found"
    if external_sessions and len(external_sessions) == len(candidate_sessions):
        return "candidate sessions belong to other PRs"
    if candidate_sessions and errors:
        return "candidate sessions found but retrieval failed"
    if candidate_sessions:
        return "candidate sessions found but no matching session for this PR"
    return "no session id found in PR metadata or visible agent-task list"


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import json
import re
import subprocess
from pathlib import Path
from typing import Any

import mine_traces


RUN_URL_RE = re.compile(r"github\.com/[^/\s]+/[^/\s]+/actions/runs/(\d+)")
RUN_NUMBER_RE = re.compile(r"(?:^|[-_/])run(\d+)(?:$|[-_/])", re.I)
ISSUE_NUMBER_RE = re.compile(r"(?:^|[-_/])issue[-_/](\d+)(?:$|[-_/])", re.I)
DEFAULT_WORKFLOWS = ("ai-agent.yml",)


def main() -> int:
    args = parse_args()
    targets = load_targets(args)
    if not targets:
        print("No targets selected.")
        return 1

    out_dir = Path(args.out_dir)
    for target in targets:
        mine_repo_prs(
            repo=target["repo"],
            slug=target["slug"],
            out_dir=out_dir,
            author=args.author,
            workflows=args.workflow or list(DEFAULT_WORKFLOWS),
            limit_prs=args.limit_prs,
            max_workflow_pages=args.max_workflow_pages,
        )
    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Mine traces for PRs authored by dragon-ai-agent.")
    parser.add_argument("--config", default="", help="agent-watcher config/targets.json path.")
    parser.add_argument("--all-targets", action="store_true", help="Mine every target in --config.")
    parser.add_argument("--target", action="append", default=[], help="Configured repo to include from --config.")
    parser.add_argument("--repo", action="append", default=[], help="Repository owner/name. Repeatable.")
    parser.add_argument("--slug", default="", help="Ontology output slug for a single --repo run.")
    parser.add_argument("--author", default="dragon-ai-agent", help="GitHub PR author login.")
    parser.add_argument("--workflow", action="append", default=[], help="Workflow file/name to map run numbers.")
    parser.add_argument("--out-dir", default="build/dragon-pr-traces", help="Output directory.")
    parser.add_argument("--limit-prs", type=int, default=1000, help="PRs to inspect per repo.")
    parser.add_argument("--max-workflow-pages", type=int, default=100, help="Workflow run pages to scan for run numbers.")
    return parser.parse_args()


def load_targets(args: argparse.Namespace) -> list[dict[str, str]]:
    targets: list[dict[str, str]] = []
    if args.config:
        config = json.loads(Path(args.config).read_text())
        selected = set(args.target)
        for item in config.get("targets", []):
            repo = item["repo"]
            if not args.all_targets and repo not in selected:
                continue
            targets.append({"repo": repo, "slug": item.get("short_name") or repo.split("/")[-1]})

    for repo in args.repo:
        slug = args.slug if len(args.repo) == 1 and args.slug else repo.split("/")[-1]
        targets.append({"repo": repo, "slug": slug})

    seen: set[str] = set()
    deduped: list[dict[str, str]] = []
    for target in targets:
        if target["repo"] in seen:
            continue
        seen.add(target["repo"])
        deduped.append(target)
    return deduped


def mine_repo_prs(
    *,
    repo: str,
    slug: str,
    out_dir: Path,
    author: str,
    workflows: list[str],
    limit_prs: int,
    max_workflow_pages: int,
) -> None:
    repo_dir = out_dir / slug
    repo_dir.mkdir(parents=True, exist_ok=True)
    prs = list_prs(repo, author, limit_prs)
    wanted_run_numbers = sorted({run_number_from_pr(pr) for pr in prs if run_number_from_pr(pr)})
    run_number_map = map_workflow_run_numbers(repo, workflows, wanted_run_numbers, max_workflow_pages)

    results: list[dict[str, Any]] = []
    for pr in prs:
        result = mine_pr(repo, pr, repo_dir, run_number_map)
        results.append(result)

    index = {
        "repo": repo,
        "slug": slug,
        "author": author,
        "pr_count": len(prs),
        "trace_pr_count": sum(1 for result in results if result["trace_summaries"]),
        "missing_trace_count": sum(1 for result in results if not result["trace_summaries"]),
        "prs": results,
    }
    mine_traces.write_json(repo_dir / "index.json", index)
    print(f"{slug}: prs={len(prs)} traced={index['trace_pr_count']} missing={index['missing_trace_count']}")


def list_prs(repo: str, author: str, limit: int) -> list[dict[str, Any]]:
    proc = subprocess.run(
        [
            "gh",
            "pr",
            "list",
            "--repo",
            repo,
            "--author",
            author,
            "--state",
            "all",
            "--limit",
            str(limit),
            "--json",
            "number,title,state,createdAt,updatedAt,url,headRefName,author,body",
        ],
        text=True,
        capture_output=True,
        check=True,
    )
    return json.loads(proc.stdout or "[]")


def mine_pr(
    repo: str,
    pr: dict[str, Any],
    repo_dir: Path,
    run_number_map: dict[int, dict[str, Any]],
) -> dict[str, Any]:
    pr_dir = repo_dir / f"pr-{pr['number']}"
    pr_dir.mkdir(parents=True, exist_ok=True)
    comments_payload = get_pr_context(repo, int(pr["number"]))
    run_ids = sorted(extract_run_ids(pr, comments_payload))
    run_number = run_number_from_pr(pr)
    mapped_run = run_number_map.get(run_number) if run_number else None
    if mapped_run:
        run_ids.append(str(mapped_run["id"]))
    run_ids = sorted(set(run_ids), key=int)

    pr_record = {
        "number": pr["number"],
        "title": pr["title"],
        "state": pr["state"],
        "created_at": pr["createdAt"],
        "updated_at": pr["updatedAt"],
        "url": pr["url"],
        "head_ref": pr["headRefName"],
        "issue_number": issue_number_from_pr(pr),
        "run_number": run_number,
        "run_ids": run_ids,
    }
    mine_traces.write_json(pr_dir / "pr.json", pr_record)

    trace_summaries: list[dict[str, Any]] = []
    errors: list[str] = []
    for run_id in run_ids:
        try:
            run = mine_traces.explicit_runs(repo, [run_id])[0]
            summary = mine_traces.mine_run(repo, run, pr_dir / f"run-{run_id}")
        except Exception as exc:  # noqa: BLE001
            errors.append(f"{run_id}: {exc}")
            continue
        if summary["trace_record_count"] or summary["artifact_trace_files"]:
            trace_summaries.append(summary)

    result = {
        **pr_record,
        "trace_summaries": trace_summaries,
        "errors": errors,
        "missing_reason": "" if trace_summaries else missing_reason(run_ids, errors),
    }
    mine_traces.write_json(pr_dir / "summary.json", result)
    return result


def get_pr_context(repo: str, number: int) -> dict[str, Any]:
    proc = subprocess.run(
        ["gh", "pr", "view", str(number), "--repo", repo, "--json", "body,comments,reviews"],
        text=True,
        capture_output=True,
        check=True,
    )
    return json.loads(proc.stdout or "{}")


def extract_run_ids(pr: dict[str, Any], context: dict[str, Any]) -> set[str]:
    texts = [pr.get("body") or "", context.get("body") or ""]
    texts.extend(comment.get("body") or "" for comment in context.get("comments", []))
    texts.extend(review.get("body") or "" for review in context.get("reviews", []))
    run_ids: set[str] = set()
    for text in texts:
        run_ids.update(RUN_URL_RE.findall(text))
    return run_ids


def run_number_from_pr(pr: dict[str, Any]) -> int | None:
    match = RUN_NUMBER_RE.search(pr.get("headRefName") or "")
    return int(match.group(1)) if match else None


def issue_number_from_pr(pr: dict[str, Any]) -> int | None:
    match = ISSUE_NUMBER_RE.search(pr.get("headRefName") or "")
    return int(match.group(1)) if match else None


def map_workflow_run_numbers(
    repo: str,
    workflows: list[str],
    wanted_run_numbers: list[int],
    max_pages: int,
) -> dict[int, dict[str, Any]]:
    if not wanted_run_numbers:
        return {}
    wanted = set(wanted_run_numbers)
    found: dict[int, dict[str, Any]] = {}
    for workflow in workflows:
        for page in range(1, max_pages + 1):
            proc = subprocess.run(
                [
                    "gh",
                    "api",
                    "--method",
                    "GET",
                    f"repos/{repo}/actions/workflows/{workflow}/runs",
                    "-f",
                    "per_page=100",
                    "-f",
                    f"page={page}",
                ],
                text=True,
                capture_output=True,
                check=False,
            )
            if proc.returncode != 0:
                break
            payload = json.loads(proc.stdout or "{}")
            runs = payload.get("workflow_runs", [])
            if not runs:
                break
            for run in runs:
                run_number = run.get("run_number")
                if run_number in wanted:
                    found[run_number] = run
            if wanted.issubset(found):
                return found
    return found


def missing_reason(run_ids: list[str], errors: list[str]) -> str:
    if not run_ids:
        return "no run id found from PR text/comments or branch run number"
    if errors:
        return "run id found but trace retrieval failed or logs/artifacts are no longer retained"
    return "run id found but no trace-like records detected"


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import collections
import json
import re
import shutil
import subprocess
import sys
import zipfile
from pathlib import Path
from typing import Any


TRACE_ARTIFACT_RE = re.compile(r"(claude-response|claude-execution|trace|execution-output)", re.I)
TRACE_MARKER_RE = re.compile(r'"(session_id|tool_use_result|total_cost_usd|type)"|Log saved to .*claude-execution-output', re.I)
TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\s*")
DEFAULT_WORKFLOWS = ("ai-agent.yml",)


def main() -> int:
    args = parse_args()
    out_dir = Path(args.out_dir)
    targets = load_targets(args)

    if not targets:
        print("No targets selected.", file=sys.stderr)
        return 1

    for target in targets:
        mine_target(
            repo=target["repo"],
            slug=target["slug"],
            out_dir=out_dir,
            workflows=args.workflow or list(DEFAULT_WORKFLOWS),
            run_ids=args.run_id,
            limit_runs=args.limit_runs,
            max_samples=args.max_samples,
            all_runs=args.all_runs,
            max_pages=args.max_pages,
            from_artifacts=args.from_artifacts,
            skip_artifact_check=args.skip_artifact_check,
        )

    return 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Mine AI-agent traces from GitHub Actions artifacts or logs.")
    parser.add_argument("--repo", action="append", default=[], help="Repository owner/name. Repeatable.")
    parser.add_argument("--slug", default="", help="Ontology output slug for a single --repo run.")
    parser.add_argument("--config", default="", help="agent-watcher config/targets.json path.")
    parser.add_argument("--all-targets", action="store_true", help="Mine every target in --config.")
    parser.add_argument("--target", action="append", default=[], help="Configured repo to include from --config. Repeatable.")
    parser.add_argument("--workflow", action="append", default=[], help="Workflow file/name to scan. Repeatable.")
    parser.add_argument("--run-id", action="append", default=[], help="Known run id to mine. Repeatable.")
    parser.add_argument("--out-dir", default="build/trace-samples", help="Output directory.")
    parser.add_argument("--limit-runs", type=int, default=40, help="Recent runs to inspect per workflow.")
    parser.add_argument("--all-runs", action="store_true", help="Inspect every available run for each workflow.")
    parser.add_argument("--from-artifacts", action="store_true", help="Inspect runs that have retained trace-like artifacts.")
    parser.add_argument("--skip-artifact-check", action="store_true", help="Skip per-run artifact lookup for log-only workflows.")
    parser.add_argument("--max-pages", type=int, default=0, help="Workflow run pages to scan with --all-runs; 0 means until empty.")
    parser.add_argument("--max-samples", type=int, default=2, help="Runs with traces to keep per repo; 0 means no limit.")
    return parser.parse_args()


def load_targets(args: argparse.Namespace) -> list[dict[str, str]]:
    targets: list[dict[str, str]] = []

    if args.config:
        config = json.loads(Path(args.config).read_text())
        selected = set(args.target)
        for item in config.get("targets", []):
            repo = item["repo"]
            if not args.all_targets and repo not in selected:
                continue
            targets.append({"repo": repo, "slug": item.get("short_name") or repo.split("/")[-1]})

    for repo in args.repo:
        slug = args.slug if len(args.repo) == 1 and args.slug else repo.split("/")[-1]
        targets.append({"repo": repo, "slug": slug})

    seen: set[str] = set()
    deduped: list[dict[str, str]] = []
    for target in targets:
        key = target["repo"]
        if key in seen:
            continue
        seen.add(key)
        deduped.append(target)
    return deduped


def mine_target(
    *,
    repo: str,
    slug: str,
    out_dir: Path,
    workflows: list[str],
    run_ids: list[str],
    limit_runs: int,
    max_samples: int,
    all_runs: bool,
    max_pages: int,
    from_artifacts: bool,
    skip_artifact_check: bool,
) -> None:
    target_dir = out_dir / slug
    target_dir.mkdir(parents=True, exist_ok=True)
    candidate_runs = (
        explicit_runs(repo, run_ids)
        if run_ids
        else discover_artifact_runs(repo)
        if from_artifacts
        else discover_runs(repo, workflows, limit_runs, all_runs=all_runs, max_pages=max_pages)
    )
    trace_summaries: list[dict[str, Any]] = []
    skipped_runs: list[dict[str, Any]] = []
    inspected = 0
    errors: list[str] = []

    for run in candidate_runs:
        if max_samples and len(trace_summaries) >= max_samples:
            break
        inspected += 1
        run_id = str(run["databaseId"])
        run_dir = target_dir / run_id
        try:
            summary = mine_run(repo, run, run_dir, skip_artifact_check=skip_artifact_check)
        except Exception as exc:  # noqa: BLE001
            errors.append(f"{run_id}: {exc}")
            continue
        if summary["trace_record_count"] or summary["artifact_trace_files"]:
            trace_summaries.append(summary)
        else:
            skipped_runs.append(summary)

    index = {
        "repo": repo,
        "slug": slug,
        "workflows": workflows,
        "all_runs": all_runs,
        "from_artifacts": from_artifacts,
        "skip_artifact_check": skip_artifact_check,
        "candidate_run_count": len(candidate_runs),
        "inspected_runs": inspected,
        "trace_run_count": len(trace_summaries),
        "sample_count": len(trace_summaries),
        "skipped_run_count": len(skipped_runs),
        "samples": trace_summaries,
        "trace_summaries": trace_summaries,
        "skipped_runs": skipped_runs,
        "errors": errors,
    }
    write_json(target_dir / "index.json", index)
    print(f"{slug}: inspected={inspected} traces={len(trace_summaries)} skipped={len(skipped_runs)}")


def discover_runs(
    repo: str,
    workflows: list[str],
    limit_runs: int,
    *,
    all_runs: bool = False,
    max_pages: int = 0,
) -> list[dict[str, Any]]:
    runs: list[dict[str, Any]] = []
    for workflow in workflows:
        if all_runs:
            runs.extend(discover_runs_paginated(repo, workflow, max_pages))
            continue
        cmd = [
            "gh",
            "run",
            "list",
            "--repo",
            repo,
            "--workflow",
            workflow,
            "--limit",
            str(limit_runs),
            "--json",
            "databaseId,createdAt,displayTitle,url,event,conclusion,status,workflowName",
        ]
        proc = subprocess.run(cmd, text=True, capture_output=True, check=False)
        if proc.returncode != 0:
            continue
        for run in json.loads(proc.stdout or "[]"):
            run["workflowQuery"] = workflow
            runs.append(run)
    return dedupe_runs(runs)


def discover_artifact_runs(repo: str) -> list[dict[str, Any]]:
    runs_by_id: dict[str, dict[str, Any]] = {}
    page = 1
    while True:
        proc = subprocess.run(
            [
                "gh",
                "api",
                "--method",
                "GET",
                f"repos/{repo}/actions/artifacts",
                "-f",
                "per_page=100",
                "-f",
                f"page={page}",
            ],
            text=True,
            capture_output=True,
            check=False,
        )
        if proc.returncode != 0:
            break
        payload = json.loads(proc.stdout or "{}")
        artifacts = payload.get("artifacts", [])
        if not artifacts:
            break
        for artifact in artifacts:
            if artifact.get("expired") or not TRACE_ARTIFACT_RE.search(artifact.get("name", "")):
                continue
            workflow_run = artifact.get("workflow_run") or {}
            run_id = workflow_run.get("id")
            if not run_id:
                continue
            key = str(run_id)
            run = runs_by_id.setdefault(
                key,
                {
                    "databaseId": run_id,
                    "createdAt": artifact.get("created_at"),
                    "displayTitle": workflow_run.get("head_branch") or artifact.get("name"),
                    "url": workflow_run.get("html_url"),
                    "event": None,
                    "conclusion": None,
                    "status": None,
                    "workflowName": None,
                    "workflowQuery": "repo-artifacts",
                    "artifactCandidates": [],
                },
            )
            run["artifactCandidates"].append(
                {
                    "id": artifact.get("id"),
                    "name": artifact.get("name"),
                    "size_in_bytes": artifact.get("size_in_bytes"),
                    "created_at": artifact.get("created_at"),
                    "expires_at": artifact.get("expires_at"),
                }
            )
        page += 1
    return sorted(runs_by_id.values(), key=lambda run: run.get("createdAt") or "", reverse=True)


def discover_runs_paginated(repo: str, workflow: str, max_pages: int) -> list[dict[str, Any]]:
    runs: list[dict[str, Any]] = []
    page = 1
    while True:
        if max_pages and page > max_pages:
            break
        proc = subprocess.run(
            [
                "gh",
                "api",
                "--method",
                "GET",
                f"repos/{repo}/actions/workflows/{workflow}/runs",
                "-f",
                "per_page=100",
                "-f",
                f"page={page}",
            ],
            text=True,
            capture_output=True,
            check=False,
        )
        if proc.returncode != 0:
            break
        payload = json.loads(proc.stdout or "{}")
        page_runs = payload.get("workflow_runs", [])
        if not page_runs:
            break
        for run in page_runs:
            runs.append(normalize_api_run(run, workflow))
        page += 1
    return runs


def normalize_api_run(run: dict[str, Any], workflow: str) -> dict[str, Any]:
    return {
        "databaseId": run.get("id"),
        "createdAt": run.get("created_at"),
        "displayTitle": run.get("display_title") or run.get("name"),
        "url": run.get("html_url"),
        "event": run.get("event"),
        "conclusion": run.get("conclusion"),
        "status": run.get("status"),
        "workflowName": run.get("name"),
        "workflowQuery": workflow,
        "runNumber": run.get("run_number"),
    }


def explicit_runs(repo: str, run_ids: list[str]) -> list[dict[str, Any]]:
    runs: list[dict[str, Any]] = []
    for run_id in run_ids:
        payload = gh_json(["api", f"repos/{repo}/actions/runs/{run_id}"])
        runs.append(
            {
                "databaseId": payload["id"],
                "createdAt": payload.get("created_at"),
                "displayTitle": payload.get("display_title") or payload.get("name"),
                "url": payload.get("html_url"),
                "event": payload.get("event"),
                "conclusion": payload.get("conclusion"),
                "status": payload.get("status"),
                "workflowName": payload.get("name"),
            }
        )
    return runs


def dedupe_runs(runs: list[dict[str, Any]]) -> list[dict[str, Any]]:
    seen: set[str] = set()
    deduped: list[dict[str, Any]] = []
    for run in runs:
        run_id = str(run["databaseId"])
        if run_id in seen:
            continue
        seen.add(run_id)
        deduped.append(run)
    return deduped


def mine_run(
    repo: str,
    run: dict[str, Any],
    run_dir: Path,
    *,
    skip_artifact_check: bool = False,
) -> dict[str, Any]:
    run_id = str(run["databaseId"])
    trace_artifacts = []
    if not skip_artifact_check:
        artifact_payload = gh_json(["api", f"repos/{repo}/actions/runs/{run_id}/artifacts"])
        trace_artifacts = matching_trace_artifacts(artifact_payload)
    trace_job = find_trace_job(repo, run_id)

    if not trace_artifacts and not trace_job:
        return {
            "repo": repo,
            "run_id": run_id,
            "run_url": run.get("url"),
            "created_at": run.get("createdAt"),
            "title": run.get("displayTitle"),
            "event": run.get("event"),
            "conclusion": run.get("conclusion"),
            "artifact_trace_files": [],
            "trace_record_count": 0,
            "session_ids": [],
            "type_counts": {},
            "skipped_reason": "no trace artifact and no trace-like job",
        }

    reset_dir(run_dir)
    write_json(run_dir / "run.json", run)
    artifact_files = download_trace_artifacts(repo, run_id, run_dir / "artifact", trace_artifacts)
    log_error = ""
    try:
        log_trace_count, session_ids, type_counts = mine_logs(repo, run_id, run_dir)
    except Exception as exc:  # noqa: BLE001
        log_trace_count = 0
        session_ids = set()
        type_counts = collections.Counter()
        log_error = str(exc)

    summary = {
        "repo": repo,
        "run_id": run_id,
        "run_url": run.get("url"),
        "created_at": run.get("createdAt"),
        "title": run.get("displayTitle"),
        "event": run.get("event"),
        "conclusion": run.get("conclusion"),
        "trace_job": trace_job,
        "artifact_trace_files": [str(path.relative_to(run_dir)) for path in artifact_files],
        "trace_record_count": log_trace_count,
        "session_ids": sorted(session_ids),
        "type_counts": dict(sorted(type_counts.items())),
        "log_error": log_error,
    }
    write_json(run_dir / "summary.json", summary)
    return summary

def matching_trace_artifacts(payload: dict[str, Any]) -> list[dict[str, Any]]:
    return [
        artifact
        for artifact in payload.get("artifacts", [])
        if not artifact.get("expired") and TRACE_ARTIFACT_RE.search(artifact.get("name", ""))
    ]


def find_trace_job(repo: str, run_id: str) -> dict[str, Any] | None:
    payload = gh_json(["api", f"repos/{repo}/actions/runs/{run_id}/jobs"])
    for job in payload.get("jobs", []):
        name = job.get("name", "")
        if job.get("conclusion") == "skipped":
            continue
        if re.search(r"(respond|claude|agent|goose)", name, re.I):
            return {
                "id": job.get("id"),
                "name": name,
                "conclusion": job.get("conclusion"),
                "status": job.get("status"),
                "html_url": job.get("html_url"),
                "started_at": job.get("started_at"),
                "completed_at": job.get("completed_at"),
            }
    return None


def download_trace_artifacts(
    repo: str,
    run_id: str,
    artifact_dir: Path,
    matches: list[dict[str, Any]],
) -> list[Path]:
    if not matches:
        return []

    artifact_dir.mkdir(parents=True, exist_ok=True)
    for artifact in matches:
        subprocess.run(
            [
                "gh",
                "run",
                "download",
                "--repo",
                repo,
                run_id,
                "-n",
                artifact["name"],
                "-D",
                str(artifact_dir / artifact["name"]),
            ],
            text=True,
            capture_output=True,
            check=True,
        )
    return [path for path in artifact_dir.rglob("*") if path.is_file()]


def mine_logs(repo: str, run_id: str, run_dir: Path) -> tuple[int, set[str], collections.Counter[str]]:
    logs_dir = run_dir / "logs"
    logs_dir.mkdir(parents=True, exist_ok=True)
    zip_path = run_dir / "logs.zip"
    with zip_path.open("wb") as handle:
        subprocess.run(["gh", "api", f"repos/{repo}/actions/runs/{run_id}/logs"], stdout=handle, check=True)
    with zipfile.ZipFile(zip_path) as archive:
        archive.extractall(logs_dir)

    trace_path = run_dir / "log-trace.jsonl"
    session_ids: set[str] = set()
    type_counts: collections.Counter[str] = collections.Counter()
    count = 0

    with trace_path.open("w", encoding="utf-8") as output:
        for log_file in logs_dir.rglob("*.txt"):
            text = log_file.read_text(errors="replace")
            if not TRACE_MARKER_RE.search(text):
                continue
            for record in parse_json_records(text):
                count += 1
                record_type = record.get("type")
                if isinstance(record_type, str):
                    type_counts[record_type] += 1
                session_id = record.get("session_id")
                if isinstance(session_id, str):
                    session_ids.add(session_id)
                output.write(json.dumps(record, sort_keys=True) + "\n")

    if count == 0:
        trace_path.unlink(missing_ok=True)
    return count, session_ids, type_counts


def parse_json_records(text: str) -> list[dict[str, Any]]:
    records: list[dict[str, Any]] = []
    buffer: list[str] = []
    for raw_line in text.splitlines():
        line = TIMESTAMP_RE.sub("", raw_line)
        if not buffer:
            if line.startswith("{"):
                buffer.append(line)
            continue
        buffer.append(line)
        try:
            parsed = json.loads("\n".join(buffer))
        except json.JSONDecodeError:
            if len(buffer) > 20000:
                buffer = []
            continue
        if isinstance(parsed, dict):
            records.append(parsed)
        buffer = []
    return records


def gh_json(args: list[str]) -> dict[str, Any]:
    proc = subprocess.run(["gh", *args], text=True, capture_output=True, check=True)
    return json.loads(proc.stdout)


def reset_dir(path: Path) -> None:
    if path.exists():
        shutil.rmtree(path)
    path.mkdir(parents=True, exist_ok=True)


def write_json(path: Path, payload: Any) -> None:
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")


if __name__ == "__main__":
    raise SystemExit(main())