Pr Learning

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "PR Learning Candidate",
  "type": "object",
  "required": [
    "id",
    "type",
    "scope_suggestion",
    "topic",
    "severity",
    "proposed_text",
    "confidence",
    "support",
    "dedupe",
    "source_refs"
  ],
  "properties": {
    "id": {"type": "string"},
    "type": {"type": "string", "enum": ["rule", "learning"]},
    "scope_suggestion": {"type": "string", "enum": ["project", "user"]},
    "topic": {"type": "string"},
    "severity": {"type": "string", "enum": ["high", "medium", "low"]},
    "proposed_text": {"type": "string"},
    "rationale": {"type": "string"},
    "confidence": {
      "type": "object",
      "required": ["acceptance_average", "band", "score_total", "components"],
      "properties": {
        "acceptance_average": {"type": "number"},
        "band": {"type": "string", "enum": ["high", "medium", "low"]},
        "score_total": {"type": "number"},
        "components": {
          "type": "object",
          "required": ["support", "acceptance", "severity", "generality"],
          "properties": {
            "support": {"type": "number"},
            "acceptance": {"type": "number"},
            "severity": {"type": "number"},
            "generality": {"type": "number"}
          }
        }
      }
    },
    "support": {
      "type": "object",
      "required": ["observation_count", "distinct_prs", "review_signal_count", "disputed_present"],
      "properties": {
        "observation_count": {"type": "number"},
        "distinct_prs": {"type": "array", "items": {"type": "number"}},
        "review_signal_count": {"type": "number"},
        "disputed_present": {"type": "boolean"}
      }
    },
    "dedupe": {
      "type": "object",
      "required": ["semantic_key", "fuzzy_key"],
      "properties": {
        "semantic_key": {"type": "string"},
        "fuzzy_key": {"type": "string"}
      }
    },
    "source_refs": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["pr_number", "urls"],
        "properties": {
          "pr_number": {"type": "number"},
          "urls": {"type": "array", "items": {"type": "string"}}
        }
      }
    }
  }
}

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "PR Learning Store v1",
  "type": "object",
  "required": ["version", "codified"],
  "properties": {
    "version": {"type": "number", "const": 1},
    "repo": {"type": ["string", "null"]},
    "seen_observation_keys": {
      "type": "array",
      "items": {"type": "string"}
    },
    "codified": {
      "type": "array",
      "items": {
        "type": "object",
        "required": ["candidate_id", "dedupe", "inserted_at", "sources", "text_fingerprint", "scope", "file_path"],
        "properties": {
          "candidate_id": {"type": "string"},
          "dedupe": {
            "type": "object",
            "required": ["semantic_key", "fuzzy_key"],
            "properties": {
              "semantic_key": {"type": "string"},
              "fuzzy_key": {"type": "string"}
            }
          },
          "inserted_at": {"type": "string"},
          "sources": {"type": "array"},
          "text_fingerprint": {"type": "string"},
          "scope": {"type": "string", "enum": ["project", "user"]},
          "file_path": {"type": "string"}
        }
      }
    }
  }
}

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

import argparse
import re
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from common import (
    ACK_PHRASES,
    DISPUTE_PHRASES,
    POSITIVE_REVIEWER_PHRASES,
    canonicalize_text,
    contains_any,
    first_sentence,
    hamming_distance,
    informative_tokens,
    now_iso,
    parse_pr_learning_signatures,
    read_json,
    sha256_hex,
    simhash64,
    write_json,
)

TOPIC_KEYWORDS: list[tuple[str, tuple[str, ...]]] = [
    ("security", ("auth", "token", "secret", "xss", "csrf", "injection", "permission", "privacy", "secure")),
    ("correctness", ("bug", "incorrect", "wrong", "edge case", "null", "nil", "undefined", "exception", "crash")),
    ("performance", ("slow", "perf", "performance", "n+1", "cache", "latency", "memory", "cpu")),
    ("testing", ("test", "coverage", "regression", "flaky", "spec")),
    ("api-design", ("api", "contract", "breaking", "version", "public", "interface")),
    ("docs", ("docs", "readme", "documentation", "comment")),
    ("readability", ("readability", "clear", "clarity", "naming", "rename", "understand")),
    ("style", ("nit", "style", "format", "lint", "whitespace", "semicolon")),
]
SEVERITY_BY_TOPIC = {
    "security": "high",
    "correctness": "high",
    "performance": "medium",
    "api-design": "medium",
    "testing": "medium",
    "docs": "low",
    "readability": "low",
    "style": "low",
    "process": "medium",
    "other": "low",
}
ACCEPTANCE_THRESHOLD_HIGH = 3.0
ACCEPTANCE_THRESHOLD_MEDIUM = 1.5
FUZZY_DUPLICATE_HAMMING_THRESHOLD = 3
def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Build ranked rule/learning candidates from PR feedback.")
    parser.add_argument("--input", default=".pr-learning/raw/feedback.json", help="Feedback JSON from collect_feedback.py")
    parser.add_argument("--output-dir", default=".pr-learning/analysis", help="Output directory for analysis artifacts")
    parser.add_argument("--project-store", default=".pr-learning/store.v1.json", help="Project dedupe store path")
    parser.add_argument("--tool", choices=["codex", "claude"], default="codex", help="Choose user store root")
    parser.add_argument("--user-store", help="Optional custom user store path")
    parser.add_argument("--allow-truncated-input", action="store_true", help="Allow analysis of truncated feedback input")
    return parser.parse_args()

def resolve_user_store(tool: str, explicit_path: str | None) -> Path:
    if explicit_path:
        return Path(explicit_path).expanduser()
    base = Path.home() / (".codex" if tool == "codex" else ".claude")
    return base / "pr-learning" / "store.v1.json"

def parse_iso(ts: str | None) -> datetime:
    if not ts:
        return datetime.min.replace(tzinfo=timezone.utc)
    return datetime.fromisoformat(ts.replace("Z", "+00:00"))


def classify_topic(text: str) -> str:
    lower = text.lower()
    for topic, words in TOPIC_KEYWORDS:
        if any(word in lower for word in words):
            return topic
    return "other"


def classify_intent(text: str) -> str:
    lower = text.lower()
    if "nit" in lower:
        return "nit"
    if "?" in lower and not any(w in lower for w in ("should", "must", "please", "prefer")):
        return "question"
    if any(w in lower for w in ("must", "need to", "should", "please", "avoid", "prefer", "use")):
        return "request-change"
    return "suggestion"


def infer_specificity(principle: str, path: str | None, files: list[str]) -> float:
    score = 0.0
    if path:
        score += 0.45
    if "/" in principle or re.search(r"\b[a-zA-Z0-9_\-]+\.[a-z]{2,5}\b", principle):
        score += 0.35
    basenames = {Path(f).name.lower() for f in files}
    if any(name and name in principle.lower() for name in basenames):
        score += 0.2
    return min(score, 1.0)


def proposed_text_from_seed(seed: str, candidate_type: str) -> str:
    cleaned = re.sub(r"\s+", " ", seed).strip().rstrip(".")
    cleaned = re.sub(r"^consider this pattern in similar changes:?\s*", "", cleaned, flags=re.IGNORECASE)
    cleaned = re.sub(r"^in similar changes,?\s*", "", cleaned, flags=re.IGNORECASE)

    prefer_match = re.search(r"prefer\s+(.+?)\s+over\s+(.+)$", cleaned, re.IGNORECASE)
    if prefer_match:
        return f"Prefer {prefer_match.group(1).strip()} over {prefer_match.group(2).strip()}."

    instead_match = re.search(r"use\s+(.+?)\s+instead of\s+(.+)$", cleaned, re.IGNORECASE)
    if instead_match:
        return f"Use {instead_match.group(1).strip()} instead of {instead_match.group(2).strip()}."

    lead = cleaned[:1].upper() + cleaned[1:] if cleaned else "Use accepted review guidance"
    return lead + "."


def load_existing_signatures(project_store_path: Path, user_store_path: Path) -> tuple[set[str], set[str]]:
    semantic_keys: set[str] = set()
    fuzzy_keys: set[str] = set()
    for store_path in (project_store_path, user_store_path):
        data = read_json(store_path, default={})
        for item in data.get("codified", []):
            dedupe = item.get("dedupe", {})
            semantic_key = dedupe.get("semantic_key")
            fuzzy_key = dedupe.get("fuzzy_key")
            if semantic_key:
                semantic_keys.add(semantic_key)
            if fuzzy_key:
                fuzzy_keys.add(fuzzy_key)
    for doc_name in ("AGENTS.md", "CLAUDE.md"):
        path = Path(doc_name)
        if path.exists():
            file_semantic, file_fuzzy = parse_pr_learning_signatures(path.read_text())
            semantic_keys.update(file_semantic)
            fuzzy_keys.update(file_fuzzy)
    return semantic_keys, fuzzy_keys


def acceptance_from_thread(
    thread: dict[str, Any],
    comments: list[dict[str, Any]],
    reviews: list[dict[str, Any]],
    commits: list[dict[str, Any]],
    pr_state: str,
    intent: str,
    pr_author: str,
) -> dict[str, Any]:
    first_time = parse_iso(comments[0].get("createdAt"))
    author_replies = [c for c in comments[1:] if ((c.get("author") or {}).get("login") or "") == pr_author]
    reviewer_replies = [c for c in comments[1:] if ((c.get("author") or {}).get("login") or "") not in {"", pr_author}]

    author_ack = any(contains_any(c.get("body", ""), ACK_PHRASES) for c in author_replies)
    author_dispute = any(contains_any(c.get("body", ""), DISPUTE_PHRASES) for c in author_replies)
    reviewer_positive = any(contains_any(c.get("body", ""), POSITIVE_REVIEWER_PHRASES) for c in reviewer_replies)
    reviewer_positive = reviewer_positive or any(
        (review.get("state") == "APPROVED") and parse_iso(review.get("submittedAt")) > first_time
        for review in reviews
    )
    commit_after = any(parse_iso(commit.get("committedDate")) > first_time for commit in commits)

    score = 0.0
    if reviewer_positive:
        score += 2.0
    if thread.get("isResolved"):
        score += 1.0
    if author_ack:
        score += 1.0
    if commit_after:
        score += 0.5
    if intent == "request-change" and pr_state == "MERGED" and not thread.get("isResolved"):
        score -= 1.0
    if author_dispute:
        score -= 2.0

    if author_dispute and not reviewer_positive:
        outcome = "disputed"
    elif score >= ACCEPTANCE_THRESHOLD_HIGH:
        outcome = "accepted"
    elif score >= ACCEPTANCE_THRESHOLD_MEDIUM:
        outcome = "partially"
    elif author_dispute:
        outcome = "wontfix"
    else:
        outcome = "unclear"

    band = "high" if score >= ACCEPTANCE_THRESHOLD_HIGH else "medium" if score >= ACCEPTANCE_THRESHOLD_MEDIUM else "low"
    return {
        "author_ack": author_ack,
        "author_dispute": author_dispute,
        "reviewer_positive": reviewer_positive,
        "commit_after": commit_after,
        "score": score,
        "band": band,
        "outcome": outcome,
    }


def collect_observations(bundle: dict[str, Any]) -> list[dict[str, Any]]:
    pr = bundle["pr"]
    pr_author = pr.get("author") or ""
    files = bundle.get("files", [])
    commits = bundle.get("commits", [])
    reviews = bundle.get("reviews", [])
    observations: list[dict[str, Any]] = []

    for thread in bundle.get("review_threads", []):
        comments = thread.get("comments", {}).get("nodes", [])
        if not comments:
            continue
        first_body = comments[0].get("body") or ""
        if not first_body.strip():
            continue
        first_author = (comments[0].get("author") or {}).get("login") or ""
        if first_author == pr_author:
            continue

        topic = classify_topic(first_body)
        intent = classify_intent(first_body)
        action_summary = first_sentence(first_body, fallback="Review feedback")
        principle = first_sentence(first_body).rstrip(".")[:220]
        acceptance = acceptance_from_thread(thread, comments, reviews, commits, pr.get("state", ""), intent, pr_author)
        patch_excerpt = ((bundle.get("file_patches") or {}).get(thread.get("path") or "") or "")[:1200]

        tokens = informative_tokens(canonicalize_text(principle))
        if not tokens:
            continue
        semantic_key = sha256_hex(f"{topic}|{' '.join(tokens[:12])}")
        path = thread.get("path")
        specificity = infer_specificity(principle, path, files)

        observations.append(
            {
                "observation_id": sha256_hex(f"{pr['url']}|{thread.get('id')}|{semantic_key}"),
                "repo": pr.get("url", "").split("/pull/")[0].replace("https://github.com/", ""),
                "pr_number": pr["number"],
                "thread_id": thread.get("id"),
                "path": path,
                "line": thread.get("line"),
                "topic": topic,
                "severity": SEVERITY_BY_TOPIC.get(topic, "low"),
                "intent": intent,
                "action_summary": action_summary,
                "principle": principle,
                "evidence": {
                    "thread_resolved": bool(thread.get("isResolved")),
                    "author_acknowledged": acceptance["author_ack"],
                    "author_disputed": acceptance["author_dispute"],
                    "reviewer_positive": acceptance["reviewer_positive"],
                    "commit_after_comment": acceptance["commit_after"],
                    "merge_state": pr.get("state"),
                },
                "acceptance_score": acceptance["score"],
                "acceptance_band": acceptance["band"],
                "outcome": acceptance["outcome"],
                "specificity": round(specificity, 3),
                "generality": round(1.0 - specificity, 3),
                "dedupe": {
                    "source_ids": [c.get("id") for c in comments if c.get("id")],
                    "semantic_key": semantic_key,
                    "fuzzy_key": simhash64(tokens),
                },
                "sources": [{"url": c.get("url"), "comment_id": c.get("id"), "created_at": c.get("createdAt")} for c in comments],
                "thread_transcript": [
                    {
                        "author": (c.get("author") or {}).get("login"),
                        "body": c.get("body"),
                        "url": c.get("url"),
                        "created_at": c.get("createdAt"),
                        "reply_to": (c.get("replyTo") or {}).get("id") if c.get("replyTo") else None,
                    }
                    for c in comments
                ],
                "code_context": {
                    "path": thread.get("path"),
                    "line": thread.get("line"),
                    "patch_excerpt": patch_excerpt,
                },
                "created_at": now_iso(),
            }
        )

    return observations


def acceptance_component(avg_acceptance: float, any_disputed: bool, any_high: bool) -> int:
    if avg_acceptance >= ACCEPTANCE_THRESHOLD_HIGH:
        base = 3
    elif avg_acceptance >= ACCEPTANCE_THRESHOLD_MEDIUM:
        base = 2
    else:
        base = 1
    return min(base, 1) if any_disputed and not any_high else base


def build_candidate(cluster: list[dict[str, Any]], index: int) -> dict[str, Any] | None:
    if not cluster:
        return None

    support_prs = sorted({obs["pr_number"] for obs in cluster})
    review_signal_count = sum(1 for obs in cluster if obs["evidence"]["reviewer_positive"])
    avg_acceptance = sum(obs["acceptance_score"] for obs in cluster) / len(cluster)
    any_disputed = any(obs["outcome"] in {"disputed", "wontfix"} for obs in cluster)
    any_high = any(obs["acceptance_band"] == "high" for obs in cluster)

    acc_comp = acceptance_component(avg_acceptance, any_disputed, any_high)
    sev_level = "high" if any(obs["severity"] == "high" for obs in cluster) else "medium" if any(obs["severity"] == "medium" for obs in cluster) else "low"
    sev_comp = 2 if sev_level == "high" else 1 if sev_level == "medium" else 0
    avg_generality = sum(obs["generality"] for obs in cluster) / len(cluster)
    avg_specificity = sum(obs["specificity"] for obs in cluster) / len(cluster)
    gen_comp = 2 if avg_generality >= 0.75 else 1 if avg_generality >= 0.4 else 0
    support_comp = min(3, len(support_prs))
    total = support_comp + acc_comp + sev_comp + gen_comp

    # Heuristic hints only. Agent performs final selection.
    if acc_comp < 2:
        return None

    type_suggestion = "learning"
    if total >= 8 and acc_comp >= 2 and len(support_prs) >= 2 and not any_disputed and review_signal_count >= 2 and avg_acceptance >= ACCEPTANCE_THRESHOLD_HIGH:
        type_suggestion = "rule"

    scope = "project"
    if (
        avg_generality >= 0.75
        and avg_specificity < 0.35
        and acc_comp >= 2
        and (len(support_prs) >= 3 or (len(support_prs) >= 2 and review_signal_count >= 2))
    ):
        scope = "user"

    confidence_band = "high" if avg_acceptance >= ACCEPTANCE_THRESHOLD_HIGH else "medium" if avg_acceptance >= ACCEPTANCE_THRESHOLD_MEDIUM else "low"
    representative = sorted(cluster, key=lambda obs: obs["acceptance_score"], reverse=True)[0]

    return {
        "id": f"C{index:03d}",
        "type": type_suggestion,
        "scope_suggestion": scope,
        "topic": representative["topic"],
        "severity": sev_level,
        "proposed_text": proposed_text_from_seed(representative["action_summary"], type_suggestion),
        "rationale": representative["principle"],
        "confidence": {
            "acceptance_average": round(avg_acceptance, 3),
            "band": confidence_band,
            "score_total": total,
            "components": {"support": support_comp, "acceptance": acc_comp, "severity": sev_comp, "generality": gen_comp},
        },
        "heuristic_hints": {
            "minimum_acceptance_gate_passed": acc_comp >= 2,
            "disputed_present": any_disputed,
            "type_suggestion_confidence": confidence_band,
            "likely_business_logic_specific": avg_specificity >= 0.7 and len(support_prs) == 1,
            "selection_authority": "agent",
        },
        "support": {
            "observation_count": len(cluster),
            "distinct_prs": support_prs,
            "review_signal_count": review_signal_count,
            "disputed_present": any_disputed,
        },
        "dedupe": representative["dedupe"],
        "source_refs": [{"pr_number": obs["pr_number"], "urls": [src.get("url") for src in obs["sources"] if src.get("url")]} for obs in cluster],
        "examples": [
            {
                "pr_number": obs["pr_number"],
                "summary": obs["action_summary"],
                "outcome": obs["outcome"],
                "topic": obs["topic"],
                "code_context": obs.get("code_context"),
                "thread_transcript": obs.get("thread_transcript"),
            }
            for obs in cluster[:3]
        ],
    }


def write_report(path: Path, repo: str, candidates: list[dict[str, Any]], duplicates: list[dict[str, Any]], observation_count: int) -> None:
    lines = [
        "# PR Learning Candidate Report",
        "",
        f"- Repo: `{repo}`",
        f"- Observations: {observation_count}",
        f"- Candidates: {len(candidates)}",
        f"- Skipped as duplicates: {len(duplicates)}",
        "",
        "## Candidates",
        "",
    ]
    if not candidates:
        lines.append("No candidate rules/learnings generated.")
    else:
        for candidate in candidates:
            lines.extend(
                [
                    f"### {candidate['id']} — {candidate['type']} ({candidate['scope_suggestion']})",
                    f"- Text: {candidate['proposed_text']}",
                    f"- Topic: {candidate['topic']} | Severity: {candidate['severity']}",
                    f"- Confidence: {candidate['confidence']['band']} ({candidate['confidence']['acceptance_average']})",
                    f"- Support PRs: {', '.join(map(str, candidate['support']['distinct_prs']))}",
                    "",
                ]
            )
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines) + "\n")


def find_probable_fuzzy_duplicate(fuzzy_key: str, fuzzy_keys: set[str]) -> tuple[str | None, int | None]:
    for existing in fuzzy_keys:
        try:
            distance = hamming_distance(fuzzy_key, existing)
        except ValueError:
            continue
        if distance <= FUZZY_DUPLICATE_HAMMING_THRESHOLD:
            return existing, distance
    return None, None

def main() -> None:
    args = parse_args()
    payload = read_json(Path(args.input), default={})
    if not payload:
        raise SystemExit(f"Input file not found or empty: {args.input}")
    input_truncated = bool(payload.get("truncated"))
    if input_truncated and not args.allow_truncated_input:
        raise SystemExit("Input feedback is truncated. Re-run collect_feedback.py without truncation or pass --allow-truncated-input.")

    output_dir = Path(args.output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    project_store = Path(args.project_store)
    user_store = resolve_user_store(args.tool, args.user_store)
    existing_semantic, existing_fuzzy = load_existing_signatures(project_store, user_store)

    all_observations: list[dict[str, Any]] = []
    for bundle in payload.get("prs", []):
        all_observations.extend(collect_observations(bundle))

    clusters: dict[str, list[dict[str, Any]]] = defaultdict(list)
    duplicates: list[dict[str, Any]] = []
    for observation in all_observations:
        semantic_key = observation["dedupe"]["semantic_key"]
        fuzzy_key = observation["dedupe"]["fuzzy_key"]
        if semantic_key in existing_semantic:
            duplicates.append({"observation_id": observation["observation_id"], "semantic_key": semantic_key, "reason": "already_codified", "pr_number": observation["pr_number"]})
            continue
        fuzzy_match, distance = find_probable_fuzzy_duplicate(fuzzy_key, existing_fuzzy)
        if fuzzy_match:
            duplicates.append(
                {
                    "observation_id": observation["observation_id"],
                    "semantic_key": semantic_key,
                    "fuzzy_key": fuzzy_key,
                    "matched_fuzzy_key": fuzzy_match,
                    "hamming_distance": distance,
                    "reason": "probable_duplicate",
                    "pr_number": observation["pr_number"],
                }
            )
            continue
        clusters[semantic_key].append(observation)

    candidates: list[dict[str, Any]] = []
    for idx, cluster in enumerate(sorted(clusters.values(), key=len, reverse=True), start=1):
        candidate = build_candidate(cluster, idx)
        if not candidate:
            continue
        if input_truncated:
            if candidate["type"] == "rule":
                candidate["type"] = "learning"
            candidate["scope_suggestion"] = "project"
        candidates.append(candidate)

    write_json(output_dir / "observations.json", {"generated_at": now_iso(), "observations": all_observations})
    write_json(
        output_dir / "candidates.json",
        {
            "generated_at": now_iso(),
            "repo": payload.get("repo"),
            "query": payload.get("query"),
            "tool": args.tool,
            "project_store": str(project_store),
            "user_store": str(user_store),
            "candidate_count": len(candidates),
            "candidates": candidates,
        },
    )
    write_json(output_dir / "duplicates.json", {"generated_at": now_iso(), "duplicates": duplicates})
    write_report(output_dir / "report.md", payload.get("repo", "unknown"), candidates, duplicates, len(all_observations))

    print(f"Observations: {len(all_observations)}")
    print(f"Candidates: {len(candidates)}")
    print(f"Duplicates skipped: {len(duplicates)}")
    print(f"Artifacts written to: {output_dir}")


if __name__ == "__main__":
    main()

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

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

from common import now_iso, parse_pr_learning_keys, read_json, sha256_hex, write_json


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Codify selected PR learning candidates into AGENTS/CLAUDE docs.")
    parser.add_argument("--candidates", default=".pr-learning/analysis/candidates.json", help="candidates.json path")
    parser.add_argument(
        "--select",
        required=True,
        help="Selection: all | none | comma-separated candidate IDs (e.g. C001,C004)",
    )
    parser.add_argument("--write", action="store_true", help="Apply file updates. Without this flag, only preview.")
    parser.add_argument("--yes", action="store_true", help="Required with --write to confirm target writes")
    parser.add_argument("--project-root", default=".", help="Project root path")
    parser.add_argument("--tool", choices=["codex", "claude"], default="codex", help="User-level target home")
    parser.add_argument("--project-store", default=".pr-learning/store.v1.json", help="Project store path")
    parser.add_argument("--user-store", help="Optional custom user store path")
    return parser.parse_args()


def resolve_project_target(root: Path) -> Path:
    agents = root / "AGENTS.md"
    claude = root / "CLAUDE.md"
    if agents.exists():
        return agents
    if claude.exists():
        return claude
    return agents


def resolve_user_target(tool: str) -> Path:
    base = Path.home() / (".codex" if tool == "codex" else ".claude")
    agents = base / "AGENTS.md"
    claude = base / "CLAUDE.md"
    if agents.exists():
        return agents
    if claude.exists():
        return claude
    return agents


def resolve_user_store(tool: str, explicit_path: str | None) -> Path:
    if explicit_path:
        return Path(explicit_path).expanduser()
    base = Path.home() / (".codex" if tool == "codex" else ".claude")
    return base / "pr-learning" / "store.v1.json"


def parse_selection(select_value: str, available_ids: set[str]) -> set[str]:
    if select_value == "none":
        return set()
    if select_value == "all":
        return set(available_ids)

    selected = {part.strip() for part in select_value.split(",") if part.strip()}
    unknown = selected - available_ids
    if unknown:
        raise ValueError(f"Unknown candidate IDs: {', '.join(sorted(unknown))}")
    return selected


def ensure_pr_learning_sections(content: str) -> str:
    if "## PR Learnings" not in content:
        if content and not content.endswith("\n"):
            content += "\n"
        content += "\n## PR Learnings\n\n### Rules\n\n### Learnings\n"

    section_match = re.search(r"## PR Learnings[\s\S]*?(?=\n## |\Z)", content)
    if not section_match:
        return content

    section = section_match.group(0)
    if "### Rules" not in section:
        section += "\n### Rules\n"
    if "### Learnings" not in section:
        section += "\n### Learnings\n"

    return content[: section_match.start()] + section + content[section_match.end() :]


def append_item(content: str, subsection: str, bullet_text: str, metadata_comment: str) -> str:
    content = ensure_pr_learning_sections(content)

    section_match = re.search(r"## PR Learnings[\s\S]*?(?=\n## |\Z)", content)
    if not section_match:
        return content

    section = section_match.group(0)
    sub_match = re.search(rf"### {subsection}[\s\S]*?(?=\n### |\Z)", section)

    if not sub_match:
        section += f"\n### {subsection}\n"
        sub_match = re.search(rf"### {subsection}[\s\S]*?(?=\n### |\Z)", section)
    if not sub_match:
        return content

    block = sub_match.group(0)
    insertion = f"\n- {bullet_text}\n  <!-- {metadata_comment} -->\n"
    new_block = block.rstrip() + insertion

    section = section[: sub_match.start()] + new_block + section[sub_match.end() :]
    return content[: section_match.start()] + section + content[section_match.end() :]


def load_store(path: Path, repo: str | None = None) -> dict[str, Any]:
    return read_json(
        path,
        default={
            "version": 1,
            "repo": repo,
            "seen_observation_keys": [],
            "codified": [],
        },
    )


def store_existing_signatures(store: dict[str, Any]) -> set[tuple[str, str]]:
    signatures: set[tuple[str, str]] = set()
    for record in store.get("codified", []):
        scope = record.get("scope") or ""
        semantic = (record.get("dedupe") or {}).get("semantic_key")
        if semantic and scope:
            signatures.add((scope, semantic))
    return signatures


def source_summary(candidate: dict[str, Any]) -> str:
    chunks: list[str] = []
    for ref in candidate.get("source_refs", []):
        pr_number = ref.get("pr_number")
        urls = ref.get("urls") or []
        first_url = urls[0] if urls else ""
        chunk = f"PR#{pr_number}:{first_url}" if first_url else f"PR#{pr_number}"
        chunks.append(chunk)
    return "|".join(chunks)


def metadata_comment(candidate: dict[str, Any], scope: str) -> str:
    return (
        f"pr-learning:v=1 type={candidate['type']} scope={scope} "
        f"key={candidate['dedupe']['semantic_key']} sim={candidate['dedupe']['fuzzy_key']} "
        f"sources={source_summary(candidate)} "
        f"confidence={candidate['confidence']['acceptance_average']}"
    )


def append_store_records(store: dict[str, Any], inserted: list[dict[str, Any]], target_path: Path, scope: str) -> None:
    signatures = store_existing_signatures(store)
    for candidate in inserted:
        signature = (scope, candidate["dedupe"]["semantic_key"])
        if signature in signatures:
            continue
        store.setdefault("codified", []).append(
            {
                "candidate_id": candidate["id"],
                "dedupe": candidate["dedupe"],
                "inserted_at": now_iso(),
                "sources": candidate["source_refs"],
                "text_fingerprint": f"sha256:{sha256_hex(candidate['proposed_text'])}",
                "scope": scope,
                "file_path": str(target_path),
            }
        )
        signatures.add(signature)


def main() -> None:
    args = parse_args()

    payload = read_json(Path(args.candidates), default={})
    candidates = payload.get("candidates", [])
    if not candidates:
        raise SystemExit("No candidates found. Run build_candidates.py first.")

    candidate_map = {candidate["id"]: candidate for candidate in candidates}
    selected_ids = parse_selection(args.select, set(candidate_map))
    selected = [candidate_map[candidate_id] for candidate_id in sorted(selected_ids)]

    project_selected = [c for c in selected if c.get("scope_suggestion") == "project"]
    user_selected = [c for c in selected if c.get("scope_suggestion") == "user"]

    project_target = resolve_project_target(Path(args.project_root).resolve())
    user_target = resolve_user_target(args.tool)
    user_store = resolve_user_store(args.tool, args.user_store)

    preview = {
        "total_candidates": len(candidates),
        "selected": [c["id"] for c in selected],
        "project_target": str(project_target),
        "user_target": str(user_target),
        "project_store": str(Path(args.project_store)),
        "user_store": str(user_store),
        "write": args.write,
    }
    print("Selection summary")
    print(json.dumps(preview, indent=2))

    if not args.write:
        return
    if not args.yes:
        raise SystemExit("Refusing to write without explicit confirmation. Re-run with --write --yes.")
    if not selected:
        raise SystemExit("Nothing selected. Refusing to write.")

    project_content = project_target.read_text() if project_target.exists() else ""
    user_content = user_target.read_text() if user_target.exists() else ""

    existing_project_keys = parse_pr_learning_keys(project_content)
    existing_user_keys = parse_pr_learning_keys(user_content)

    inserted_project: list[dict[str, Any]] = []
    inserted_user: list[dict[str, Any]] = []

    for candidate in project_selected:
        semantic_key = candidate["dedupe"]["semantic_key"]
        if semantic_key in existing_project_keys:
            continue

        subsection = "Rules" if candidate["type"] == "rule" else "Learnings"
        project_content = append_item(project_content, subsection, candidate["proposed_text"], metadata_comment(candidate, "project"))
        existing_project_keys.add(semantic_key)
        inserted_project.append(candidate)

    for candidate in user_selected:
        semantic_key = candidate["dedupe"]["semantic_key"]
        if semantic_key in existing_user_keys:
            continue

        subsection = "Rules" if candidate["type"] == "rule" else "Learnings"
        user_content = append_item(user_content, subsection, candidate["proposed_text"], metadata_comment(candidate, "user"))
        existing_user_keys.add(semantic_key)
        inserted_user.append(candidate)

    project_changed = bool(inserted_project)
    user_changed = bool(inserted_user)

    if project_changed:
        project_target.parent.mkdir(parents=True, exist_ok=True)
        project_target.write_text(project_content)
    if user_changed:
        user_target.parent.mkdir(parents=True, exist_ok=True)
        user_target.write_text(user_content)

    if not project_changed and not user_changed:
        print("No new insertions were required (all selected candidates already codified).")
        return

    project_store_path = Path(args.project_store)
    project_data = load_store(project_store_path, repo=payload.get("repo"))
    user_data = load_store(user_store, repo=None)

    append_store_records(project_data, inserted_project, project_target, "project")
    append_store_records(user_data, inserted_user, user_target, "user")

    write_json(project_store_path, project_data)
    write_json(user_store, user_data)

    written = [
        *[{"id": c["id"], "scope": "project", "target": str(project_target)} for c in inserted_project],
        *[{"id": c["id"], "scope": "user", "target": str(user_target)} for c in inserted_user],
    ]

    print("Write complete")
    print(
        json.dumps(
            {
                "written": written,
                "project_store": str(project_store_path),
                "user_store": str(user_store),
            },
            indent=2,
        )
    )


if __name__ == "__main__":
    main()

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

import argparse
from datetime import datetime, timedelta, timezone
from pathlib import Path

from common import now_iso, run_command, run_gh_graphql, run_gh_json, write_json

PR_DETAILS_QUERY = """
query PullRequestDetails($owner: String!, $name: String!, $number: Int!) {
  repository(owner: $owner, name: $name) {
    pullRequest(number: $number) {
      number
      title
      url
      state
      mergedAt
      updatedAt
      baseRefName
      headRefName
      author { login }
      comments(first: 100) {
        nodes {
          id
          url
          body
          createdAt
          author { login }
        }
        pageInfo { hasNextPage endCursor }
      }
      reviews(first: 100) {
        nodes {
          id
          url
          body
          state
          submittedAt
          author { login }
        }
        pageInfo { hasNextPage endCursor }
      }
      reviewThreads(first: 100) {
        nodes {
          id
          isResolved
          isOutdated
          path
          line
          originalLine
          resolvedBy { login }
          comments(first: 100) {
            nodes {
              id
              url
              body
              createdAt
              author { login }
              replyTo { id }
            }
            pageInfo { hasNextPage endCursor }
          }
        }
        pageInfo { hasNextPage endCursor }
      }
      commits(first: 100) {
        nodes {
          commit {
            oid
            committedDate
            messageHeadline
            url
          }
        }
        pageInfo { hasNextPage endCursor }
      }
      files(first: 100) {
        nodes { path }
        pageInfo { hasNextPage endCursor }
      }
    }
  }
}
"""


def resolve_repo(explicit_repo: str | None) -> str:
    if explicit_repo:
        return explicit_repo
    return run_command(["gh", "repo", "view", "--json", "nameWithOwner", "-q", ".nameWithOwner"]).strip()


def split_repo(repo: str) -> tuple[str, str]:
    if "/" not in repo:
        raise ValueError(f"Expected owner/repo but got: {repo}")
    owner, name = repo.split("/", 1)
    return owner, name


def resolve_actor_login() -> str:
    return run_command(["gh", "api", "user", "-q", ".login"]).strip()


def build_search_query(login: str, since_days: int) -> str:
    parts = [f"is:pr involves:{login}", "sort:updated-desc"]
    if since_days > 0:
        since_date = (datetime.now(timezone.utc) - timedelta(days=since_days)).date().isoformat()
        parts.insert(1, f"updated:>={since_date}")
    return " ".join(parts)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Collect PR feedback artifacts for pr-learning skill.")
    parser.add_argument("--repo", help="owner/repo. Defaults to current repository.")
    parser.add_argument(
        "--since-days",
        type=int,
        default=0,
        help="How many days back to search PRs. Use 0 to disable date filtering (default).",
    )
    parser.add_argument("--limit", type=int, default=200, help="Max PRs to fetch.")
    parser.add_argument("--out", default=".pr-learning/raw/feedback.json", help="Output JSON file path.")
    parser.add_argument(
        "--allow-truncated",
        action="store_true",
        help="Allow partial results when GraphQL pagination indicates more data exists.",
    )
    return parser.parse_args()


def fetch_file_patches(owner: str, name: str, pr_number: int) -> dict[str, str]:
    patches: dict[str, str] = {}
    page = 1
    while page <= 20:
        files_page = run_gh_json(["api", f"repos/{owner}/{name}/pulls/{pr_number}/files?per_page=100&page={page}"])
        if not files_page:
            break
        for file_entry in files_page:
            filename = file_entry.get("filename")
            patch = file_entry.get("patch")
            if filename and patch:
                patches[filename] = patch
        if len(files_page) < 100:
            break
        page += 1
    return patches


def main() -> None:
    args = parse_args()
    repo = resolve_repo(args.repo)
    owner, name = split_repo(repo)
    actor_login = resolve_actor_login()
    query = build_search_query(actor_login, args.since_days)

    prs = run_gh_json(
        [
            "pr",
            "list",
            "-R",
            repo,
            "--state",
            "all",
            "--search",
            query,
            "--limit",
            str(args.limit),
            "--json",
            "number,url,title,state,mergedAt,updatedAt,author,baseRefName,headRefName",
        ]
    )

    bundles = []
    for pr in prs:
        payload = run_gh_graphql(
            PR_DETAILS_QUERY,
            {"owner": owner, "name": name, "number": int(pr["number"])},
        )
        node = payload["data"]["repository"]["pullRequest"]
        file_patches = fetch_file_patches(owner, name, int(pr["number"]))
        bundles.append(
            {
                "pr": {
                    "number": node["number"],
                    "url": node["url"],
                    "title": node["title"],
                    "state": node["state"],
                    "mergedAt": node["mergedAt"],
                    "updatedAt": node["updatedAt"],
                    "baseRefName": node["baseRefName"],
                    "headRefName": node["headRefName"],
                    "author": (node.get("author") or {}).get("login"),
                },
                "issue_comments": node["comments"]["nodes"],
                "reviews": node["reviews"]["nodes"],
                "review_threads": node["reviewThreads"]["nodes"],
                "commits": [n["commit"] for n in node["commits"]["nodes"]],
                "files": [n["path"] for n in node["files"]["nodes"]],
                "file_patches": file_patches,
                "page_info": {
                    "comments_has_next": node["comments"]["pageInfo"]["hasNextPage"],
                    "reviews_has_next": node["reviews"]["pageInfo"]["hasNextPage"],
                    "threads_has_next": node["reviewThreads"]["pageInfo"]["hasNextPage"],
                    "thread_comments_has_next": any(
                        t.get("comments", {}).get("pageInfo", {}).get("hasNextPage")
                        for t in node["reviewThreads"]["nodes"]
                    ),
                    "commits_has_next": node["commits"]["pageInfo"]["hasNextPage"],
                    "files_has_next": node["files"]["pageInfo"]["hasNextPage"],
                },
            }
        )

    truncation_hits = []
    for bundle in bundles:
        for key, value in bundle.get("page_info", {}).items():
            if value:
                truncation_hits.append({"pr_number": bundle["pr"]["number"], "signal": key})

    if truncation_hits and not args.allow_truncated:
        first = truncation_hits[0]
        raise SystemExit(
            "Collection is truncated (GraphQL page limit reached). "
            f"First hit: PR #{first['pr_number']} ({first['signal']}). "
            "Re-run with --allow-truncated to proceed with partial data."
        )

    output = {
        "version": 1,
        "generated_at": now_iso(),
        "repo": repo,
        "query": query,
        "params": {
            "since_days": args.since_days,
            "limit": args.limit,
            "allow_truncated": args.allow_truncated,
        },
        "stats": {
            "pr_count": len(bundles),
            "thread_count": sum(len(bundle["review_threads"]) for bundle in bundles),
            "issue_comment_count": sum(len(bundle["issue_comments"]) for bundle in bundles),
            "truncation_hits": len(truncation_hits),
        },
        "truncated": bool(truncation_hits),
        "truncation_details": truncation_hits,
        "prs": bundles,
    }

    out_path = Path(args.out)
    write_json(out_path, output)
    print(f"Wrote feedback bundle: {out_path} ({len(bundles)} PRs)")
    if truncation_hits:
        print(f"Warning: output is truncated ({len(truncation_hits)} pagination signals).")


if __name__ == "__main__":
    main()

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

import hashlib
import json
import re
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

STOP_WORDS = {
    "a", "an", "and", "are", "as", "at", "be", "by", "for", "from", "how",
    "i", "if", "in", "into", "is", "it", "its", "of", "on", "or", "that",
    "the", "their", "then", "there", "these", "this", "to", "was", "we", "were",
    "will", "with", "you", "your", "our", "can", "should", "could", "would",
}

ACK_PHRASES = (
    "fixed", "addressed", "done", "updated", "good catch", "agree", "makes sense",
)
DISPUTE_PHRASES = (
    "disagree",
    "won't fix",
    "wont fix",
    "not necessary",
    "by design",
    "prefer not to",
    "already supported",
    "already handled",
    "already does",
    "out of date",
    "stale feedback",
)
POSITIVE_REVIEWER_PHRASES = (
    "lgtm", "looks good", "approved", "ship it", "good to merge",
)


def run_command(args: list[str], stdin_text: str | None = None) -> str:
    result = subprocess.run(
        args,
        input=stdin_text,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        check=False,
    )
    if result.returncode != 0:
        raise RuntimeError(
            f"Command failed ({result.returncode}): {' '.join(args)}\n{result.stderr.strip()}"
        )
    return result.stdout


def run_gh_json(args: list[str], stdin_text: str | None = None) -> Any:
    raw = run_command(["gh", *args], stdin_text=stdin_text)
    try:
        return json.loads(raw)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"Failed to parse gh JSON output: {exc}\nOutput snippet: {raw[:400]}") from exc


def run_gh_graphql(query: str, variables: dict[str, Any]) -> Any:
    args = ["api", "graphql", "-f", f"query={query}"]
    for key, value in variables.items():
        if isinstance(value, bool):
            args.extend(["-F", f"{key}={'true' if value else 'false'}"])
        elif isinstance(value, int):
            args.extend(["-F", f"{key}={value}"])
        else:
            args.extend(["-f", f"{key}={value}"])
    payload = run_gh_json(args)
    if "errors" in payload:
        raise RuntimeError(f"GraphQL errors: {payload['errors']}")
    return payload


def now_iso() -> str:
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat()


def read_json(path: Path, default: Any) -> Any:
    if not path.exists():
        return default
    return json.loads(path.read_text())


def write_json(path: Path, payload: Any) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")


def strip_markdown(text: str) -> str:
    text = re.sub(r"```[\s\S]*?```", " ", text)
    text = re.sub(r"`[^`]+`", " <code> ", text)
    text = re.sub(r"\[(.*?)\]\((.*?)\)", r"\1", text)
    text = re.sub(r"https?://\S+", " ", text)
    text = text.replace("\r", " ")
    return text


def normalize_whitespace(text: str) -> str:
    return re.sub(r"\s+", " ", text).strip()


def canonicalize_text(text: str) -> str:
    text = strip_markdown(text).lower()
    text = re.sub(r"\b\d+(?:\.\d+)*\b", " <num> ", text)
    text = re.sub(r"[^a-z0-9_\-/<code>\s]", " ", text)
    return normalize_whitespace(text)


def informative_tokens(text: str) -> list[str]:
    tokens = re.findall(r"[a-z0-9_\-]+", text)
    return [t for t in tokens if len(t) > 2 and t not in STOP_WORDS]


def sha256_hex(value: str) -> str:
    return hashlib.sha256(value.encode("utf-8")).hexdigest()


def simhash64(tokens: list[str]) -> str:
    if not tokens:
        return "0x0"
    bits = [0] * 64
    for token in tokens:
        h = int(hashlib.sha256(token.encode("utf-8")).hexdigest()[:16], 16)
        for idx in range(64):
            bits[idx] += 1 if ((h >> idx) & 1) else -1
    out = 0
    for idx, value in enumerate(bits):
        if value >= 0:
            out |= 1 << idx
    return f"0x{out:016x}"


def hamming_distance(a_hex: str, b_hex: str) -> int:
    a = int(a_hex, 16)
    b = int(b_hex, 16)
    return (a ^ b).bit_count()


def first_sentence(text: str, fallback: str = "") -> str:
    cleaned = normalize_whitespace(strip_markdown(text))
    if not cleaned:
        return fallback
    match = re.split(r"(?<=[.!?])\s+", cleaned, maxsplit=1)
    return match[0][:240]


def contains_any(text: str, phrases: tuple[str, ...]) -> bool:
    hay = text.lower()
    return any(phrase in hay for phrase in phrases)


def parse_pr_learning_signatures(markdown_text: str) -> tuple[set[str], set[str]]:
    keys: set[str] = set()
    sims: set[str] = set()

    for match in re.finditer(r"pr-learning:v=1[^>]*key=([^\s;]+)", markdown_text):
        keys.add(match.group(1))
    for match in re.finditer(r"pr-learning:v=1[^>]*sim=([^\s;]+)", markdown_text):
        sims.add(match.group(1))

    return keys, sims


def parse_pr_learning_keys(markdown_text: str) -> set[str]:
    keys, _ = parse_pr_learning_signatures(markdown_text)
    return keys