Local SE Agent Masterclass

# PR Synthesizer - Operational Playbook

## 1. Context & Role
You are an Enterprise Release Engineer and Technical Lead. Your job is to transform raw code diffs and verification outputs into concise, risk-aware Pull Request documentation.

## 2. Impact Risk Taxonomy (Blast Radius Matrix)
When evaluating changed methods via the Call Graph, categorize risk:
- **LOW RISK**: Changes contained entirely within private helper methods or isolated unit tests.
- **MEDIUM RISK**: Changes to public methods called by 1-3 downstream controllers or internal services.
- **HIGH RISK**: Changes to Core Domain Interfaces, Entities, or methods called by >3 distinct modules or payment pipelines.

## 3. Mandatory Markdown Structure
Every generated PR Description must follow this schema:
1. **Executive Summary**: 2-3 sentences explaining the "Why" and "What".
2. **Impact Assessment**: Risk Level + Downstream Affected Components list.
3. **Verification Summary**: Test execution metrics (Total / Passed / Duration).
4. **Architectural Diff Notes**: Brevity bullet points of key structural modifications.
5. **Reviewer Checklist**: Clear verification criteria for peer engineers.

import subprocess
from typing import Dict, List, Any
from pathlib import Path

def extract_semantic_diff(worktree_dir: str, base_branch: str = "main") -> Dict[str, Any]:
    """Extracts git diff and list of changed files from the isolated worktree."""
    diff_cmd = ["git", "-C", worktree_dir, "diff", f"{base_branch}...HEAD"]
    name_cmd = ["git", "-C", worktree_dir, "diff", "--name-only", f"{base_branch}...HEAD"]

    diff_res = subprocess.run(diff_cmd, capture_output=True, text=True)
    name_res = subprocess.run(name_cmd, capture_output=True, text=True)

    files = [f.strip() for f in name_res.stdout.splitlines() if f.strip()]
    return {
        "raw_diff": diff_res.stdout,
        "modified_files": files,
        "files_count": len(files)
    }

def calculate_blast_radius(modified_files: List[str], call_graph) -> Dict[str, Any]:
    """Computes upstream callers and calculates the risk rating."""
    affected_nodes = []
    for f in modified_files:
        symbol_name = Path(f).stem
        if hasattr(call_graph, "get_downstream_dependents"):
            deps = call_graph.get_downstream_dependents(symbol_name)
            affected_nodes.extend(deps)

    unique_affected = sorted(list(set(affected_nodes)))
    risk_level = "LOW"
    if len(unique_affected) > 4:
        risk_level = "HIGH"
    elif len(unique_affected) >= 1:
        risk_level = "MEDIUM"

    return {
        "risk_level": risk_level,
        "affected_consumers": unique_affected
    }

def format_pr_markdown(title: str, summary: str, blast_data: Dict[str, Any], test_results: str) -> str:
    """Assembles the final GitHub-flavored markdown description."""
    consumers_list = "\n".join([f"- `{c}`" for c in blast_data.get("affected_consumers", [])]) or "- None (Isolated changes)"
    return f"""# {title}

### Executive Summary
{summary}

### Impact Assessment & Blast Radius
- **Assessed Risk**: `{blast_data['risk_level']}`
- **Downstream Consumers Impacted**:
{consumers_list}

### Automated Verification Suite
{test_results}

### Peer Review Checklist
* [ ] Business logic matches domain constraints
* [ ] No regression on downstream consumers listed above
* [ ] Verified locally on Apple Silicon / macOS M3
"""

from skills.pr_synthesizer.tools import calculate_blast_radius, format_pr_markdown


class MockCallGraph:
    """Mocking the Day 19 Call Hierarchy Graph."""
    def get_downstream_dependents(self, class_name: str):
        mapping = {
            "PaymentService": ["CheckoutController", "SubscriptionScheduler", "OrderAuditConsumer"],
            "StringUtils": []
        }
        return mapping.get(class_name, [])


def test_pr_synthesis_flow():
    print("=== Testing Day 25: PR Synthesis & Impact Assessment ===")

    # 1. Simulate modified files and call graph lookup
    modified = ["src/main/java/com/service/PaymentService.java"]
    graph = MockCallGraph()
    blast = calculate_blast_radius(modified, graph)

    print(f"[+] Assessed Risk: {blast['risk_level']}")
    print(f"[+] Impacted Consumers: {blast['affected_consumers']}")

    # 2. Format the PR description
    summary = "Resolved NullPointerException when token expires during async gateway dispatch."
    test_run = "Results: 14 passed, 0 failed, 0 errors (Elapsed: 3.42s)"

    pr_doc = format_pr_markdown(
        title="fix(payment): prevent NPE on expired gateway token",
        summary=summary,
        blast_data=blast,
        test_results=test_run
    )

    print("\n--- Synthesized PR Body Preview ---")
    print(pr_doc)
    assert blast["risk_level"] == "MEDIUM"
    print("\nDay 25 Skill Verification Passed Successfully!")


if __name__ == "__main__":
    test_pr_synthesis_flow()