# “好 PR” 社区标准检索建议

目标：不是泛泛搜索“如何做代码审查”，而是提炼“什么样的 PR 才算高质量、值得 merge”。

建议检索方向：

- GitHub pull request best practices
- what makes a good pull request maintainer
- good PR size description tests reviewability
- GitHub pull request guidelines maintainers
- open source maintainer pull request quality

优先提炼的维度：

- 粒度是否合适（小而聚焦，还是范围漂移）
- 描述是否清楚（背景、动机、影响范围）
- 是否有测试/验证/回滚考虑
- 是否方便 reviewer 理解与审阅
- 是否与 issue / 文档 / CI 状态形成闭环

输出到最终报告时，尽量写成：

- 来源是什么
- 来源链接是什么
- 发布日期或访问日期是什么
- 它提出了什么标准
- 当前 PR 在该标准下表现如何
- 这会不会影响 merge 决策

# Good PR Standards

最后整理时间：2026-03-24

这份文档是 `git-pr-review` 内置的“好 PR”标准参考。
默认执行时优先使用本文件，不要求每次实时联网。

## 来源

1. GitHub Docs: Setting guidelines for repository contributors  
链接：https://docs.github.com/en/communities/setting-up-your-project-for-healthy-contributions/setting-guidelines-for-repository-contributors  
访问日期：2026-03-24

2. GitHub Docs: About issue and pull request templates  
链接：https://docs.github.com/enterprise-cloud@latest/communities/using-templates-to-encourage-useful-issues-and-pull-requests/manually-creating-a-single-issue-template-for-your-repository  
访问日期：2026-03-24

3. GitHub Docs: About code owners  
链接：https://docs.github.com/articles/about-code-owners  
访问日期：2026-03-24

4. GitHub Docs: About pull request reviews  
链接：https://docs.github.com/en/github/collaborating-with-issues-and-pull-requests/about-pull-request-reviews  
访问日期：2026-03-24

5. Balachandran, Sawant et al. What Makes a Code Change Easier to Review  
链接：https://www.mozillafoundation.org/en/research/library/what-makes-a-code-change-easier-to-review-an-empirical-investigation-on-code-change-reviewability/  
访问日期：2026-03-24

## 沉淀出的核心标准

### 1. 目标清楚

好 PR 应该让 reviewer 很快知道：
- 它要解决什么问题
- 为什么现在要改
- 改动影响什么范围

如果 reviewer 需要从大量 diff 猜意图，这通常不是好信号。

### 2. 描述充分且结构化

好 PR 通常会提供：
- 背景 / 动机
- 主要改动点
- 测试或验证方式
- 风险、限制或待确认点

GitHub 官方关于贡献指南和 PR 模板的实践，本质上都在推动这件事：让贡献者提交“格式良好、信息完整”的 PR。

### 3. 粒度合适、便于 review

研究和工程实践都支持一个结论：
- 更小、更聚焦、边界清楚的改动，通常更容易 review
- 把多个不相关改动揉在一起，会降低 review 质量
- tangled change（缠结改动）会让理解、回滚和追责都变难

所以评审时要看：
- 这个 PR 是单一问题还是混了多个目标
- 是否存在 scope drift（标题说 A，diff 里还夹了 B/C）

### 4. 可验证

好 PR 不只是“看起来有道理”，还要便于验证。
常见正向信号：
- 有测试
- 有复现步骤或验证步骤
- CI 信号清楚
- 对 reviewer 来说可重复检查

如果缺少测试、CI 失败、验证方式含糊，通常至少应降低 merge 信心。

### 5. 容易找到合适 reviewer

GitHub 官方关于 code owners 和 review request 的机制说明了一个事实：
- 好 PR 应该能较容易路由到正确 reviewer
- 改动范围清晰时，CODEOWNERS / reviewer assignment 才真正有效

如果改动横跨太多模块、责任边界模糊，就会降低 review 质量和 merge 可靠性。

### 6. review 反馈可阻断 merge 时，要严肃对待

GitHub 官方 review 机制强调：
- `Request changes` 不是装饰性意见
- 在启用了 required reviews 的仓库中，这类反馈本身就意味着 merge gate

因此在 `git-pr-review` 里：
- 如果 PR 存在阻断性问题，不应把它描述成“小瑕疵”
- 要明确区分“可 merge 但建议后补”与“当前不应 merge”

### 7. 需要显式说明风险与限制

好 PR 不应只讲优点。
更可靠的 PR 往往会主动说明：
- 已知限制
- 未覆盖场景
- 迁移 / 回滚风险
- 还需要谁确认

如果 PR 完全不提这些，而 diff 又涉及高风险路径，review 时要主动补这一层判断。

## 在 git-pr-review 里的使用方式

最终报告至少要回答：
- 目标是否清楚
- 描述是否充分
- 粒度是否合适
- 是否可验证
- reviewer / ownership 是否清晰
- 是否存在阻断 merge 的问题
- 风险和限制是否被诚实披露

## 使用提醒

这份参考是“默认基线”，不是死板 checklist。
如用户明确要求最新社区口径、特定组织规范、或当前 PR 场景明显超出这份基线，再联网补充即可。

# License / 合规审查清单

当 PR 涉及新依赖、vendored 代码、复制第三方代码、模板、字体、图标、数据集、模型权重、示例代码或许可证文件时，必须做 license 审查。

优先检查：

1. 新增依赖或升级依赖
- 是否引入新的第三方包、库、插件、模型或前端资源
- 其 license 是什么
- 新 license 是否与仓库现有 license 或分发方式冲突

2. 复制/粘贴或 vendored 代码
- 是否把外部仓库的代码、脚本、头文件、模板、字体、图标、数据直接拷入仓库
- 是否保留原作者声明、copyright、LICENSE 文件、NOTICE 文件
- 是否需要额外署名或保留许可证文本

3. copyleft / 强传染性条款
- 是否引入 GPL、AGPL、LGPL、SSPL 或其他可能影响整体分发义务的内容
- 该项目当前分发方式是否允许这种义务存在
- 如果是服务端部署场景，AGPL/SSPL 风险是否会触发

4. 专有或限制性条款
- 是否引入禁止商用、禁止再分发、禁止修改、要求付费授权或仅限内部使用的内容
- PR 描述是否掩盖了这些限制

5. 许可证不一致或缺失
- 新增文件是否带有互相冲突的 header/license notice
- 新依赖或复制来的代码是否根本没有明确 license
- 如果 license 不明确，默认不能视为可安全 merge

6. 仓库级影响
- 当前仓库已有主许可证是什么
- 该 PR 是否需要同时更新 `LICENSE`、`NOTICE`、`README`、依赖清单、第三方声明文档

建议输出：
- 结论：无明显 license 风险 / 存在待确认风险 / 明显冲突，不建议 merge
- 风险来源：依赖 / vendored 代码 / 文件头 / 资源资产 / 文档缺失
- 建议动作：补许可证声明、补第三方归属、法务确认、替换依赖、不要 merge

# Independent Parallel Review RESULT Template

```markdown
# Independent PR Review

## Final Call
- Recommendation: Merge / Merge after fixes / Request changes / Do not merge / Escalate security review
- Risk Level: Low / Medium / High / Critical
- Confidence: High / Medium / Low

## Problem Understanding
- ...

## Strengths
- ...

## Limitations
- ...

## Security Review
- ...

## License Review
- ...

## Good PR Criteria Comparison
- ...

## Key Evidence
- ...

## Evidence Gaps
- ...

## Suggested Disposition
- ...
```

# parallel-vibe 集成说明

`git-pr-review` 的并行独立评审默认通过 `parallel-vibe` 落地。

推荐顺序：

1. `prepare_review_job.py`
2. `build_parallel_review_plan.py`
3. 优先执行 `parallel_review_job.json` 里的 `recommended_command`；若手动执行，则使用 `../parallel-vibe/scripts/parallel_vibe.py --plan-file ... --src-dir ... --out-dir ... --project-id ...`
4. `aggregate_parallel_reviews.py`
5. 宿主 AI 综合 `independent_review_summary.md` 与原始证据写最终报告

默认并行独立评审次数：5。

如果 `raw/`、`notes/`、`evidence/` 的证据材料发生变化，必须先重新运行 `build_parallel_review_plan.py`，让输入快照和 `project_id` 同步刷新。

关键产物：

- `parallel_review/parallel_review_job.json`
- `parallel_review/parallel_plan.json`
- `parallel_review/parallel_plan.md`
- `parallel_review/parallel_runs/.bensz-api/task-{yyyymmdd-hhmm}-{简短描述}/parallel-vibe/<project_id>/...`
- `parallel_review/independent_review_summary.md`
- `parallel_review/independent_review_summary.json`

最终报告应引用：

- 独立评审的 recommendation / risk 分布
- 主要共识
- 主要分歧
- 若涉及依赖/第三方代码，还应引用各 thread 对 license 风险的判断
- 这些分歧是否改变最终 merge 建议

# Git PR Review Report Template

```markdown
# Git PR Review: {owner/repo} #{pr_number}

## 结论摘要
- 建议：Merge / Merge after fixes / Request changes / Do not merge / Escalate security review
- 风险等级：Low / Medium / High / Critical
- 一句话判断：...

## 独立评审综合结果
- 独立评审次数：5（或用户指定值）
- recommendation 分布：...
- risk 分布：...
- 共识：...
- 分歧：...

## PR 在解决什么问题
- 背景：...
- 证据：...

## 方案分析
### 优势
- ...
### 局限
- ...
### 替代方案或待确认点
- ...

## 恶意/安全风险审查
- 结论：...
- 风险信号：...
- 证据：...

## License / 合规审查
- 结论：无明显风险 / 存在待确认风险 / 明显冲突
- 风险来源：依赖 / vendored 代码 / 资源资产 / 文件头 / 许可证文件
- 建议动作：补声明 / 替换依赖 / 法务确认 / 不建议 merge

## 与“好 PR”社区标准的对照
- 标准 + 链接 + 本 PR 表现

## 关键证据
- Diff：...
- 评论：...
- CI：...
- Issue / Docs：...
- 外部参考：...

## 证据不足与待确认点
- 缺失材料：...
- 对结论的影响：...

## 建议的处理方式
- ...
```

# 恶意 / 高风险 PR 审查清单

优先检查以下信号：

1. 凭证与敏感数据
- 是否新增读取环境变量、token、cookie、secret 的逻辑
- 是否把敏感信息发到外部域名、日志或第三方服务

2. 远程执行与下载
- 是否新增 `curl | sh`、动态下载脚本、远程模板注入、反序列化执行
- 是否把本来静态的流程改成了运行远端内容

3. 权限与发布链路
- 是否修改 CI/CD、发布脚本、权限文件、密钥注入路径、workflow 触发条件
- 是否增加了管理员、写权限、部署权限、包发布权限
- 是否替换 GitHub Actions / CI 里的第三方 action 版本、下载源或 runner 行为

4. 防护削弱
- 是否删掉测试、告警、审计日志、输入校验、签名验证、安全开关
- 是否把严格校验改成“默默放行”

5. 隐蔽性
- 是否引入混淆代码、难以解释的编码载荷、可疑依赖、异常大的二进制文件
- 是否把危险逻辑藏进“cleanup / refactor / docs / test helper”这类无害命名里

6. 供应链与依赖
- 是否升级/替换依赖到来路不明的镜像、fork、私有源或动态下载地址
- 是否修改包管理器锁文件却没有对应的合理说明

7. 社会工程
- 改动标题、描述、提交信息与真实 diff 是否不一致
- 改动范围是否显著偏离 PR 自称目标

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

import argparse
import json
import re
import sys
from collections import Counter
from pathlib import Path
from typing import Any

from common_config import get_skill_root, load_config


SCRIPT_PATH = Path(__file__)
SKILL_ROOT = get_skill_root(SCRIPT_PATH)
CONFIG = load_config(SKILL_ROOT)
PARALLEL_CFG = CONFIG["parallel_review"]
REVIEW_POLICY = CONFIG.get("review_policy") or {}
RESULT_FILENAME = str(PARALLEL_CFG["result_filename"])
REQUIRED_RESULT_SECTIONS = [str(item) for item in (PARALLEL_CFG.get("required_result_sections") or [])]
ALLOWED_RECOMMENDATIONS = {str(item) for item in (REVIEW_POLICY.get("final_recommendations") or [])}
ALLOWED_RISK_LEVELS = {str(item) for item in (REVIEW_POLICY.get("risk_levels") or [])}
ALLOWED_CONFIDENCE_LEVELS = {str(item) for item in (PARALLEL_CFG.get("confidence_levels") or ["High", "Medium", "Low"])}

REC_RE = re.compile(r"^- Recommendation:\s*(.+?)\s*$", re.MULTILINE)
RISK_RE = re.compile(r"^- Risk Level:\s*(.+?)\s*$", re.MULTILINE)
CONF_RE = re.compile(r"^- Confidence:\s*(.+?)\s*$", re.MULTILINE)


def fail(message: str) -> int:
    print(f"ERROR: {message}", file=sys.stderr)
    return 1


def _load_job(path: Path) -> dict[str, Any]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValueError(f"job file is not valid JSON: {path}") from exc
    if not isinstance(payload, dict):
        raise ValueError("job file must be a JSON object")
    return payload


def _read_text(path: Path) -> str:
    return path.read_text(encoding="utf-8", errors="replace")


def _extract(pattern: re.Pattern[str], text: str) -> str:
    m = pattern.search(text)
    return m.group(1).strip() if m else "Unknown"


def _expected_thread_ids(review_count: int) -> list[str]:
    width = int(PARALLEL_CFG["thread_id_width"])
    return [str(index).zfill(width) for index in range(1, review_count + 1)]


def _validate_result_text(text: str) -> list[str]:
    issues: list[str] = []
    for section in REQUIRED_RESULT_SECTIONS:
        if section not in text:
            issues.append(f"missing required section {section!r}")

    recommendation = _extract(REC_RE, text)
    risk_level = _extract(RISK_RE, text)
    confidence = _extract(CONF_RE, text)
    if recommendation == "Unknown":
        issues.append("missing Recommendation field")
    elif ALLOWED_RECOMMENDATIONS and recommendation not in ALLOWED_RECOMMENDATIONS:
        issues.append(f"invalid Recommendation: {recommendation}")
    if risk_level == "Unknown":
        issues.append("missing Risk Level field")
    elif ALLOWED_RISK_LEVELS and risk_level not in ALLOWED_RISK_LEVELS:
        issues.append(f"invalid Risk Level: {risk_level}")
    if confidence == "Unknown":
        issues.append("missing Confidence field")
    elif ALLOWED_CONFIDENCE_LEVELS and confidence not in ALLOWED_CONFIDENCE_LEVELS:
        issues.append(f"invalid Confidence: {confidence}")
    return issues


def _thread_entries(project_root: Path, expected_thread_ids: list[str]) -> tuple[list[dict[str, Any]], list[str], list[str]]:
    entries: list[dict[str, Any]] = []
    missing_threads: list[str] = []
    invalid_threads: list[str] = []
    for thread_id in expected_thread_ids:
        child = project_root / thread_id
        if not child.is_dir():
            missing_threads.append(f"{thread_id} (missing thread directory)")
            continue
        result_path = child / RESULT_FILENAME
        if not result_path.exists():
            missing_threads.append(f"{thread_id} (missing {RESULT_FILENAME})")
            continue
        text = _read_text(result_path)
        thread_json = child / "thread.json"
        title = child.name
        if thread_json.exists():
            try:
                title = json.loads(thread_json.read_text(encoding="utf-8")).get("title") or title
            except Exception:
                pass
        issues = _validate_result_text(text)
        if issues:
            invalid_threads.append(f"{thread_id} ({'; '.join(issues)})")
            continue
        entries.append(
            {
                "thread_id": child.name,
                "title": title,
                "result_path": str(result_path),
                "recommendation": _extract(REC_RE, text),
                "risk_level": _extract(RISK_RE, text),
                "confidence": _extract(CONF_RE, text),
                "excerpt": "\n".join(line for line in text.splitlines()[:20]).strip(),
            }
        )
    return entries, missing_threads, invalid_threads


def _mode_or_mixed(values: list[str]) -> str:
    filtered = [v for v in values if v and v != "Unknown"]
    if not filtered:
        return "Unknown"
    counts = Counter(filtered).most_common()
    if len(counts) > 1 and counts[0][1] == counts[1][1]:
        return "Mixed"
    return counts[0][0]


def main() -> int:
    parser = argparse.ArgumentParser(description="Aggregate parallel-vibe independent PR review results.")
    parser.add_argument("--job-file", required=True, help="Path to parallel_review_job.json")
    args = parser.parse_args()

    job_path = Path(args.job_file).expanduser().resolve()
    if not job_path.exists():
        return fail(f"job file not found: {job_path}")
    try:
        job = _load_job(job_path)
    except ValueError as exc:
        return fail(str(exc))
    for key in ("project_root", "aggregate_markdown", "aggregate_json", "review_count"):
        if key not in job:
            return fail(f"job file missing required field: {key}")
    project_root = Path(job["project_root"]).resolve()
    aggregate_md = Path(job["aggregate_markdown"]).resolve()
    aggregate_json = Path(job["aggregate_json"]).resolve()
    review_count = int(job["review_count"])
    if review_count < 1:
        return fail(f"review_count must be >= 1: {review_count}")
    if not project_root.exists() or not project_root.is_dir():
        recommended = job.get("recommended_command")
        hint = f"; run `{recommended}` first" if recommended else ""
        return fail(f"parallel review project root not found: {project_root}{hint}")

    entries, missing_threads, invalid_threads = _thread_entries(project_root, _expected_thread_ids(review_count))
    if missing_threads:
        return fail("parallel review outputs are incomplete: " + ", ".join(missing_threads))
    if invalid_threads:
        return fail("parallel review outputs are invalid: " + ", ".join(invalid_threads))
    if not entries:
        return fail(f"no {RESULT_FILENAME} files found under {project_root}")
    rec_counts = Counter(entry["recommendation"] for entry in entries)
    risk_counts = Counter(entry["risk_level"] for entry in entries)
    consensus = {
        "recommendation": _mode_or_mixed([entry["recommendation"] for entry in entries]),
        "risk_level": _mode_or_mixed([entry["risk_level"] for entry in entries]),
    }
    payload = {
        "expected_review_count": review_count,
        "review_count": len(entries),
        "project_root": str(project_root),
        "consensus": consensus,
        "recommendation_counts": dict(rec_counts),
        "risk_counts": dict(risk_counts),
        "threads": entries,
    }
    aggregate_json.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

    md_lines = [
        "# Independent Review Summary",
        "",
        f"- Expected Review Count: {review_count}",
        f"- Review Count: {len(entries)}",
        f"- Project Root: `{project_root}`",
        f"- Consensus Recommendation: {consensus['recommendation']}",
        f"- Consensus Risk Level: {consensus['risk_level']}",
        "",
        "## Recommendation Distribution",
    ]
    for key, value in sorted(rec_counts.items()):
        md_lines.append(f"- {key}: {value}")
    md_lines.extend(["", "## Risk Distribution"])
    for key, value in sorted(risk_counts.items()):
        md_lines.append(f"- {key}: {value}")
    md_lines.extend(["", "## Review Matrix", "| Thread | Title | Recommendation | Risk | Confidence |", "|------|------|------|------|------|"])
    for entry in entries:
        md_lines.append(
            f"| {entry['thread_id']} | {entry['title']} | {entry['recommendation']} | {entry['risk_level']} | {entry['confidence']} |"
        )
    md_lines.extend(["", "## Per-Thread Excerpts"])
    for entry in entries:
        md_lines.extend([
            f"### Thread {entry['thread_id']} - {entry['title']}",
            f"- Result Path: `{entry['result_path']}`",
            "",
            "```markdown",
            entry["excerpt"],
            "```",
            "",
        ])
    md_lines.extend([
        "## Guidance For Final Report",
        "- 在最终 `Git-PR-Review_*.md` 中新增 `## 独立评审综合结果` 章节。",
        "- 先报告 recommendation / risk 的分布，再写共识与分歧。",
        "- 若共识为 `Mixed`，必须明确说明主要分歧点。",
    ])
    aggregate_md.write_text("\n".join(md_lines).rstrip() + "\n", encoding="utf-8")
    print(json.dumps({"aggregate_markdown": str(aggregate_md), "aggregate_json": str(aggregate_json)}, ensure_ascii=False, indent=2))
    return 0


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

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

import argparse
import hashlib
import json
import shlex
import shutil
import sys
from pathlib import Path
from textwrap import dedent
from typing import Any

from common_config import get_skill_root, load_config


SCRIPT_PATH = Path(__file__)
SKILL_ROOT = get_skill_root(SCRIPT_PATH)
CONFIG = load_config(SKILL_ROOT)
DIRECTORIES_CFG = CONFIG["directories"]
PARALLEL_CFG = CONFIG["parallel_review"]
FILES_CFG = CONFIG["files"]
DEPENDENCIES_CFG = CONFIG.get("dependencies") or {}
REVIEW_POLICY = CONFIG.get("review_policy") or {}
SNAPSHOT_SOURCES = (
    ("raw_dir", str(DIRECTORIES_CFG["raw"])),
    ("notes_dir", str(DIRECTORIES_CFG["notes"])),
    ("evidence_dir", str(DIRECTORIES_CFG["evidence"])),
)


def fail(message: str) -> int:
    print(f"ERROR: {message}", file=sys.stderr)
    return 1


def _ensure_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)


def _copy_if_exists(src: Path, dst: Path) -> None:
    if not src.exists():
        return
    if src.is_dir():
        if dst.exists():
            shutil.rmtree(dst)
        shutil.copytree(src, dst)
    else:
        _ensure_dir(dst.parent)
        shutil.copy2(src, dst)


def _digest_tree(root: Path) -> str:
    digest = hashlib.md5()
    for path in sorted(p for p in root.rglob("*") if p.is_file()):
        rel = path.relative_to(root).as_posix().encode("utf-8")
        digest.update(rel)
        digest.update(b"\0")
        digest.update(path.read_bytes())
        digest.update(b"\0")
    return digest.hexdigest()


def _resolve_parallel_vibe_script() -> Path:
    configured = str(DEPENDENCIES_CFG.get("parallel_vibe_script") or "").strip()
    if not configured:
        raise ValueError("config.dependencies.parallel_vibe_script is required")
    candidate = Path(configured).expanduser()
    if not candidate.is_absolute():
        candidate = (SKILL_ROOT / candidate).resolve()
    else:
        candidate = candidate.resolve()
    if not candidate.exists() or not candidate.is_file():
        raise ValueError(f"parallel-vibe runner script not found: {candidate}")
    return candidate


def _result_template_text() -> str:
    sections = [str(item) for item in (PARALLEL_CFG.get("required_result_sections") or [])]
    if not sections:
        sections = [
            "# Independent PR Review",
            "## Final Call",
            "## Problem Understanding",
            "## Strengths",
            "## Limitations",
            "## Security Review",
            "## License Review",
            "## Good PR Criteria Comparison",
            "## Key Evidence",
            "## Evidence Gaps",
            "## Suggested Disposition",
        ]
    recommendations = " / ".join(str(item) for item in (REVIEW_POLICY.get("final_recommendations") or []))
    risks = " / ".join(str(item) for item in (REVIEW_POLICY.get("risk_levels") or []))
    confidences = " / ".join(str(item) for item in (PARALLEL_CFG.get("confidence_levels") or ["High", "Medium", "Low"]))

    lines: list[str] = [sections[0], ""]
    for section in sections[1:]:
        lines.append(section)
        if section == "## Final Call":
            lines.extend(
                [
                    f"- Recommendation: {recommendations}",
                    f"- Risk Level: {risks}",
                    f"- Confidence: {confidences}",
                ]
            )
        else:
            lines.append("- ...")
        lines.append("")
    return "\n".join(lines).rstrip()


def _load_manifest(path: Path) -> dict[str, Any]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValueError(f"manifest is not valid JSON: {path}") from exc
    if not isinstance(payload, dict):
        raise ValueError("manifest must be a JSON object")
    return payload


def _require_mapping(payload: dict[str, Any], key: str) -> dict[str, Any]:
    value = payload.get(key)
    if not isinstance(value, dict):
        raise ValueError(f"manifest missing object: {key}")
    return value


def _resolve_path(value: Any, *, field_name: str) -> Path:
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"manifest missing path: {field_name}")
    return Path(value).expanduser().resolve()


def _validate_manifest(manifest: dict[str, Any]) -> tuple[Path, dict[str, Path]]:
    repo = _require_mapping(manifest, "repo")
    pull_request = _require_mapping(manifest, "pull_request")
    if not str(repo.get("owner") or "").strip() or not str(repo.get("name") or "").strip():
        raise ValueError("manifest.repo must include owner and name")
    if not str(pull_request.get("input") or pull_request.get("pr_slug") or "").strip():
        raise ValueError("manifest.pull_request must include input or pr_slug")

    paths = _require_mapping(manifest, "paths")
    run_dir = _resolve_path(paths.get("run_dir"), field_name="paths.run_dir")
    if not run_dir.exists() or not run_dir.is_dir():
        raise ValueError(f"manifest paths.run_dir does not exist: {run_dir}")

    resolved: dict[str, Path] = {"run_dir": run_dir}
    for key, _ in SNAPSHOT_SOURCES:
        current = _resolve_path(paths.get(key), field_name=f"paths.{key}")
        if not current.exists() or not current.is_dir():
            raise ValueError(f"manifest {key} does not exist: {current}")
        try:
            current.relative_to(run_dir)
        except ValueError as exc:
            raise ValueError(f"manifest {key} must stay inside run_dir: {current}") from exc
        resolved[key] = current
    return run_dir, resolved


def _review_prompt(*, thread_id: str, idx: int, total: int, repo: str, pr_ref: str, focus: str, extra: str) -> str:
    extra_block = f"\n补充指示：\n{extra.strip()}\n" if extra.strip() else ""
    result_template = _result_template_text()
    return dedent(
        f"""
        你正在执行一次 GitHub Pull Request 的独立评审。

        这是第 {idx}/{total} 个独立评审 thread，thread_id={thread_id}。
        你的任务不是参考其他 thread，而是基于当前 workspace 中提供的材料，独立完成一次完整 PR 审查。

        目标仓库：{repo}
        目标 PR：{pr_ref}
        本 thread 的次级关注焦点：{focus}
        {extra_block}
        你必须遵守以下要求：
        - 只能基于当前 workspace 中已有的 `manifest.json`、`raw/`、`notes/`、`evidence/` 材料进行判断。
        - 不要假设你能看到其他 thread 的结果。
        - 不要修改原始证据文件；只在当前工作区根目录写出 `RESULT.md`。
        - 如果材料不足，要明确写出“证据不足”与它对结论的影响。
        - 如果 PR 涉及依赖、复制的第三方代码、字体、图标、模板、数据或许可证文件，必须显式给出 license / 合规判断。
        - 优先使用 workspace 中 `notes/community_good_pr.md` 的内置“好 PR”标准。
        - 只有用户明确要求最新口径、指定特殊社区规范，或你判断当前内置标准不足时，才额外联网补充来源；如补充了，必须在 RESULT.md 里说明。

        `RESULT.md` 必须使用以下结构：

        {result_template}
        """
    ).strip() + "\n"


def _build_plan(
    *,
    manifest: dict[str, Any],
    review_count: int,
    extra_instructions: str,
    review_runner_type: str,
    review_runner_profile: str,
) -> dict[str, Any]:
    thread_width = int(PARALLEL_CFG["thread_id_width"])
    repo = manifest["repo"]
    pr = manifest["pull_request"]
    repo_ref = f"{repo['owner']}/{repo['name']}"
    pr_ref = str(pr.get("input") or pr.get("pr_slug") or pr.get("number") or "")
    focuses = list(PARALLEL_CFG.get("review_focuses") or [])
    threads: list[dict[str, Any]] = []

    for idx in range(1, review_count + 1):
        thread_id = str(idx).zfill(thread_width)
        focus = focuses[(idx - 1) % len(focuses)] if focuses else "完整独立评审"
        threads.append(
            {
                "thread_id": thread_id,
                "title": f"Independent Review {idx}",
                "runner": {
                    "type": review_runner_type,
                    "profile": review_runner_profile,
                    "model": "",
                    "args": [],
                },
                "prompt": _review_prompt(
                    thread_id=thread_id,
                    idx=idx,
                    total=review_count,
                    repo=repo_ref,
                    pr_ref=pr_ref,
                    focus=focus,
                    extra=extra_instructions,
                ),
            }
        )

    synthesis_cfg = PARALLEL_CFG.get("synthesis") or {}
    synthesis_runner = synthesis_cfg.get("runner") or {"type": "claude", "profile": "deep"}
    return {
        "plan_version": 1,
        "prompt": f"Independent PR review for {repo_ref} {pr_ref}",
        "threads": threads,
        "synthesis": {
            "enabled": bool(synthesis_cfg.get("enabled", False)),
            "runner": {
                "type": str(synthesis_runner.get("type") or "claude"),
                "profile": str(synthesis_runner.get("profile") or "deep"),
                "model": "",
                "args": [],
            },
            "prompt": "请综合多份独立 PR 审查结果，生成一份最终决策建议。",
        },
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Build a parallel-vibe plan for independent git-pr-review runs.")
    parser.add_argument("--manifest", required=True, help="Path to git-pr-review manifest.json")
    parser.add_argument("--n", type=int, default=int(PARALLEL_CFG["default_review_count"]), help="Independent review count")
    parser.add_argument("--extra-instructions", default="", help="Extra instructions passed to all independent reviewers")
    parser.add_argument(
        "--review-runner-type",
        default=str(PARALLEL_CFG["review_runner"]["type"]),
        help="Runner type for independent reviewers (default from config.yaml)",
    )
    parser.add_argument(
        "--review-runner-profile",
        default=str(PARALLEL_CFG["review_runner"]["profile"]),
        help="Runner profile for independent reviewers (default from config.yaml)",
    )
    args = parser.parse_args()

    manifest_path = Path(args.manifest).expanduser().resolve()
    if not manifest_path.exists():
        return fail(f"manifest not found: {manifest_path}")

    review_count = int(args.n)
    if review_count < int(PARALLEL_CFG["min_review_count"]) or review_count > int(PARALLEL_CFG["max_review_count"]):
        return fail(
            f"--n must be within [{PARALLEL_CFG['min_review_count']}, {PARALLEL_CFG['max_review_count']}]: {review_count}"
        )

    try:
        manifest = _load_manifest(manifest_path)
        run_dir, manifest_paths = _validate_manifest(manifest)
    except ValueError as exc:
        return fail(str(exc))

    parallel_root = run_dir / str(PARALLEL_CFG["workspace_dir"])
    input_snapshot = parallel_root / str(PARALLEL_CFG["input_snapshot_dir"])
    parallel_out_dir = parallel_root / str(PARALLEL_CFG["parallel_out_dir"])
    plan_path = parallel_root / str(PARALLEL_CFG["plan_filename"])
    plan_md_path = parallel_root / str(PARALLEL_CFG["plan_markdown"])
    job_path = parallel_root / str(PARALLEL_CFG["job_manifest_name"])
    aggregate_md_path = parallel_root / str(PARALLEL_CFG["aggregate_markdown"])
    aggregate_json_path = parallel_root / str(PARALLEL_CFG["aggregate_json"])

    _ensure_dir(parallel_root)
    _ensure_dir(parallel_out_dir)
    if input_snapshot.exists():
        shutil.rmtree(input_snapshot)
    _ensure_dir(input_snapshot)

    _copy_if_exists(manifest_path, input_snapshot / str(FILES_CFG["manifest_name"]))
    for path_key, dest_dirname in SNAPSHOT_SOURCES:
        _copy_if_exists(manifest_paths[path_key], input_snapshot / dest_dirname)

    snapshot_fingerprint = _digest_tree(input_snapshot)
    plan = _build_plan(
        manifest=manifest,
        review_count=review_count,
        extra_instructions=str(args.extra_instructions or ""),
        review_runner_type=str(args.review_runner_type or PARALLEL_CFG["review_runner"]["type"]),
        review_runner_profile=str(args.review_runner_profile or PARALLEL_CFG["review_runner"]["profile"]),
    )
    plan_bytes = json.dumps(plan, ensure_ascii=False, indent=2).encode("utf-8") + b"\n"
    plan_path.write_bytes(plan_bytes)
    project_id = hashlib.md5(plan_bytes + snapshot_fingerprint.encode("utf-8")).hexdigest()
    project_root = parallel_out_dir / ".bensz-api" / "skills" / "parallel-vibe" / project_id
    try:
        parallel_vibe_script = _resolve_parallel_vibe_script()
    except ValueError as exc:
        return fail(str(exc))

    command = " ".join(
        [
            "python3",
            shlex.quote(str(parallel_vibe_script)),
            "--plan-file",
            shlex.quote(str(plan_path)),
            "--src-dir",
            shlex.quote(str(input_snapshot)),
            "--out-dir",
            shlex.quote(str(parallel_out_dir)),
            "--project-id",
            shlex.quote(project_id),
        ]
    )
    plan_md_path.write_text(
        dedent(
            f"""
            # Parallel Independent Review Plan

            - Review count: {review_count}
            - Manifest: `{manifest_path}`
            - Input snapshot: `{input_snapshot}`
            - Input snapshot fingerprint: `{snapshot_fingerprint}`
            - Parallel out dir: `{parallel_out_dir}`
            - Parallel runner script: `{parallel_vibe_script}`
            - Expected project root: `{project_root}`
            - Aggregate markdown: `{aggregate_md_path}`
            - Aggregate json: `{aggregate_json_path}`
            - Recommended command: `{command}`
            """
        ).strip()
        + "\n",
        encoding="utf-8",
    )
    job = {
        "review_count": review_count,
        "manifest": str(manifest_path),
        "run_dir": str(run_dir),
        "parallel_root": str(parallel_root),
        "input_snapshot": str(input_snapshot),
        "input_snapshot_fingerprint": snapshot_fingerprint,
        "parallel_out_dir": str(parallel_out_dir),
        "parallel_vibe_script": str(parallel_vibe_script),
        "plan_file": str(plan_path),
        "plan_markdown": str(plan_md_path),
        "project_id": project_id,
        "project_root": str(project_root),
        "aggregate_markdown": str(aggregate_md_path),
        "aggregate_json": str(aggregate_json_path),
        "recommended_command": command,
    }
    job_path.write_text(json.dumps(job, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps(job, ensure_ascii=False, indent=2))
    return 0


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

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

from pathlib import Path
from typing import Any


def _strip_comment(text: str) -> str:
    if "#" not in text:
        return text.rstrip()
    in_quote = False
    quote_char = ""
    out: list[str] = []
    for ch in text:
        if ch in {'"', "'"}:
            if in_quote and ch == quote_char:
                in_quote = False
                quote_char = ""
            elif not in_quote:
                in_quote = True
                quote_char = ch
        if ch == "#" and not in_quote:
            break
        out.append(ch)
    return "".join(out).rstrip()


def _parse_scalar(value: str) -> Any:
    value = _strip_comment(value).strip()
    if value == "":
        return ""
    if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")):
        return value[1:-1]
    if value.lower() == "true":
        return True
    if value.lower() == "false":
        return False
    if value.isdigit():
        return int(value)
    return value


def _indent_of(line: str) -> int:
    return len(line) - len(line.lstrip(" "))


def _next_meaningful(lines: list[str], index: int) -> int:
    while index < len(lines):
        stripped = _strip_comment(lines[index]).strip()
        if stripped:
            return index
        index += 1
    return index


def _parse_block(lines: list[str], index: int, indent: int) -> tuple[Any, int]:
    index = _next_meaningful(lines, index)
    if index >= len(lines):
        return {}, index

    stripped = _strip_comment(lines[index]).strip()
    if stripped.startswith("- "):
        result: list[Any] = []
        while index < len(lines):
            index = _next_meaningful(lines, index)
            if index >= len(lines):
                break
            line = lines[index]
            current_indent = _indent_of(line)
            stripped = _strip_comment(line).strip()
            if current_indent < indent or not stripped.startswith("- "):
                break
            item_text = stripped[2:].strip()
            if item_text:
                result.append(_parse_scalar(item_text))
                index += 1
                continue
            child, index = _parse_block(lines, index + 1, current_indent + 2)
            result.append(child)
        return result, index

    result_dict: dict[str, Any] = {}
    while index < len(lines):
        index = _next_meaningful(lines, index)
        if index >= len(lines):
            break
        line = lines[index]
        current_indent = _indent_of(line)
        if current_indent < indent:
            break
        if current_indent > indent:
            raise ValueError(f"Unexpected indentation at line {index + 1}: {line!r}")
        stripped = _strip_comment(line).strip()
        if stripped.startswith("- "):
            raise ValueError(f"Unexpected list item at line {index + 1}: {line!r}")
        key, sep, rest = stripped.partition(":")
        if not sep:
            raise ValueError(f"Invalid line {index + 1}: {line!r}")
        key = key.strip()
        rest = rest.strip()
        if rest:
            result_dict[key] = _parse_scalar(rest)
            index += 1
            continue
        next_index = _next_meaningful(lines, index + 1)
        if next_index >= len(lines) or _indent_of(lines[next_index]) <= current_indent:
            result_dict[key] = {}
            index = next_index
            continue
        child, index = _parse_block(lines, next_index, _indent_of(lines[next_index]))
        result_dict[key] = child
    return result_dict, index


def load_config(skill_root: Path) -> dict[str, Any]:
    config_path = skill_root / "config.yaml"
    text = config_path.read_text(encoding="utf-8")
    try:
        import yaml  # type: ignore
    except Exception:
        yaml = None  # type: ignore
    if yaml is not None:
        try:
            data = yaml.safe_load(text)
            if isinstance(data, dict):
                return data
        except Exception:
            pass
    parsed, _ = _parse_block(text.splitlines(), 0, 0)
    if not isinstance(parsed, dict):
        raise ValueError("config.yaml must parse to a mapping")
    return parsed


def get_skill_root(script_path: Path) -> Path:
    return script_path.resolve().parent.parent

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

import argparse
import json
import re
import sys
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse

from common_config import get_skill_root, load_config


SCRIPT_PATH = Path(__file__)
SKILL_ROOT = get_skill_root(SCRIPT_PATH)
CONFIG = load_config(SKILL_ROOT)
DIRECTORIES = CONFIG["directories"]
FILES = CONFIG["files"]
OUTPUT = CONFIG["output"]
GOOD_PR_REFERENCE = SKILL_ROOT / "references" / "good-pr-standards.md"
PR_NUMBER_RE = re.compile(r"^(?:#|pr[-\s]?)?(\d+)$", re.IGNORECASE)


def sanitize(text: str) -> str:
    text = re.sub(r"[^A-Za-z0-9._-]+", "_", text.strip())
    text = re.sub(r"_+", "_", text).strip("._-")
    return text or "unknown"


def validate_timestamp(timestamp: str) -> str:
    fmt = str(OUTPUT["timestamp_format"])
    try:
        datetime.strptime(timestamp, fmt)
    except ValueError as exc:
        raise ValueError(f"timestamp must match {fmt!r}: {timestamp}") from exc
    return timestamp


def parse_repo(repo: str) -> tuple[str, str, str]:
    repo = repo.strip()
    if repo.startswith(("http://", "https://")):
        parsed = urlparse(repo)
        if parsed.netloc not in {"github.com", "www.github.com"}:
            raise ValueError(f"Only github.com repositories are supported: {repo}")
        parts = [p for p in parsed.path.split("/") if p]
        if len(parts) != 2:
            raise ValueError(f"Repository URL must point to the repo root: {repo}")
        owner, name = parts[0], parts[1]
    else:
        parts = [p for p in repo.split("/") if p]
        if len(parts) != 2:
            raise ValueError("Repository must be a GitHub URL or owner/repo slug.")
        owner, name = parts[0], parts[1]
    repo_name = name.removesuffix(".git")
    if not owner or not repo_name:
        raise ValueError("Repository must include both owner and repo name.")
    repo_slug = sanitize(f"{owner}_{repo_name}")
    return owner, repo_name, repo_slug


def parse_pr(pr: str) -> tuple[str, int, tuple[str, str] | None]:
    pr = pr.strip()
    if pr.startswith(("http://", "https://")):
        parsed = urlparse(pr)
        if parsed.netloc not in {"github.com", "www.github.com"}:
            raise ValueError(f"Only github.com PR URLs are supported: {pr}")
        parts = [p for p in parsed.path.split("/") if p]
        if len(parts) >= 4 and parts[2] == "pull" and parts[3].isdigit():
            number = int(parts[3])
            return f"pr-{number}", number, (parts[0], parts[1])
        raise ValueError(f"Could not parse PR number from URL: {pr}")

    match = PR_NUMBER_RE.fullmatch(pr)
    if not match:
        raise ValueError("PR must be a PR URL, '#123', '123', or 'pr-123'.")
    number = int(match.group(1))
    return f"pr-{number}", number, None


def ensure_dir(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)


def allocate_run_dir(workspace_root: Path, timestamp: str) -> tuple[Path, str]:
    candidate = workspace_root / timestamp
    if not candidate.exists():
        return candidate, timestamp
    for idx in range(2, 100):
        run_id = f"{timestamp}-{idx:02d}"
        candidate = workspace_root / run_id
        if not candidate.exists():
            return candidate, run_id
    raise RuntimeError(f"failed to allocate unique run directory under {workspace_root}")


def write_placeholder(path: Path, content: str) -> None:
    if not path.exists():
        path.write_text(content, encoding="utf-8")


def build_good_pr_note() -> str:
    if not GOOD_PR_REFERENCE.exists():
        return "# Community Good PR Notes\n\n- 未找到内置参考 `references/good-pr-standards.md`，请手动补充。\n"
    source = GOOD_PR_REFERENCE.read_text(encoding="utf-8").strip()
    return (
        "# Community Good PR Notes\n\n"
        "以下内容由 skill 内置参考 `references/good-pr-standards.md` 预填充；"
        "默认优先使用这些标准，不必每次实时联网。\n\n"
        + source
        + "\n"
    )


def main() -> int:
    parser = argparse.ArgumentParser(description="Prepare an isolated workspace for git-pr-review.")
    parser.add_argument("--repo", required=True, help="GitHub repo URL or owner/repo")
    parser.add_argument("--pr", required=True, help="PR URL or PR number")
    parser.add_argument("--workspace-dir", default=str(DIRECTORIES["default_workspace"]), help="Workspace root")
    parser.add_argument("--report-dir", default=str(OUTPUT["default_report_dir"]), help="Final report directory")
    parser.add_argument("--timestamp", default="", help="Override timestamp")
    args = parser.parse_args()

    try:
        owner, repo_name, repo_slug = parse_repo(args.repo)
        pr_slug, pr_number, pr_repo = parse_pr(args.pr)
        if pr_repo is not None and tuple(pr_repo) != (owner, repo_name):
            raise ValueError(
                f"PR URL points to {pr_repo[0]}/{pr_repo[1]}, but --repo points to {owner}/{repo_name}."
            )
        fmt = str(OUTPUT["timestamp_format"])
        timestamp = validate_timestamp(args.timestamp) if args.timestamp else datetime.now().strftime(fmt)
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2

    workspace_root = Path(args.workspace_dir).expanduser()
    report_root = Path(args.report_dir).expanduser()
    if not workspace_root.is_absolute():
        workspace_root = (Path.cwd() / workspace_root).resolve()
    else:
        workspace_root = workspace_root.resolve()
    if not report_root.is_absolute():
        report_root = (Path.cwd() / report_root).resolve()
    else:
        report_root = report_root.resolve()

    try:
        run_dir, run_id = allocate_run_dir(workspace_root, f"{DIRECTORIES['run_prefix']}{timestamp}")
    except RuntimeError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2
    raw_dir = run_dir / str(DIRECTORIES["raw"])
    notes_dir = run_dir / str(DIRECTORIES["notes"])
    evidence_dir = run_dir / str(DIRECTORIES["evidence"])
    logs_dir = run_dir / str(DIRECTORIES["logs"])

    for path in (workspace_root, run_dir, raw_dir, notes_dir, evidence_dir, logs_dir, report_root):
        ensure_dir(path)

    write_placeholder(
        raw_dir / str(FILES["raw_readme"]),
        "# Raw Inputs\n\n把 PR 元数据、diff、评论、CI 状态与关联 issue 保存到这个目录。\n",
    )
    write_placeholder(
        notes_dir / str(FILES["user_context_note"]),
        "# User Context\n\n- 用户已有判断：\n- 关注点：\n- 禁区或限制：\n",
    )
    write_placeholder(
        notes_dir / str(FILES["community_note"]),
        build_good_pr_note(),
    )
    write_placeholder(
        notes_dir / str(FILES["license_review_note"]),
        "# License Review\n\n- 是否涉及新依赖、复制代码、字体、图标、模板、数据或模型资源：\n- 发现的 license / notice：\n- 是否存在兼容性风险：\n- 建议动作：\n",
    )
    write_placeholder(
        evidence_dir / str(FILES["key_findings_note"]),
        "# Key Findings\n\n- 关键发现 1：\n- 关键发现 2：\n",
    )
    write_placeholder(
        evidence_dir / str(FILES["missing_items_note"]),
        "# Missing Items\n\n- 未获取材料：\n- 原因：\n- 对结论的影响：\n",
    )

    report_name = f"{OUTPUT['report_prefix']}_{repo_slug}_{pr_slug}_{run_id}{OUTPUT['report_extension']}"
    manifest_path = run_dir / str(FILES["manifest_name"])
    manifest = {
        "generated_at": timestamp,
        "run_id": run_id,
        "repo": {
            "input": args.repo,
            "owner": owner,
            "name": repo_name,
            "repo_slug": repo_slug,
            "skill_version": str(CONFIG["skill_info"]["version"]),
        },
        "pull_request": {
            "input": args.pr,
            "pr_slug": pr_slug,
            "number": pr_number,
            "url_repo": list(pr_repo) if pr_repo is not None else None,
        },
        "paths": {
            "workspace_root": str(workspace_root),
            "run_dir": str(run_dir),
            "raw_dir": str(raw_dir),
            "notes_dir": str(notes_dir),
            "evidence_dir": str(evidence_dir),
            "logs_dir": str(logs_dir),
            "report_dir": str(report_root),
            "report_path": str(report_root / report_name),
        },
        "files": {
            "manifest_name": str(FILES["manifest_name"]),
            "manifest_path": str(manifest_path),
            "raw_readme": str(raw_dir / str(FILES["raw_readme"])),
            "user_context_note": str(notes_dir / str(FILES["user_context_note"])),
            "community_note": str(notes_dir / str(FILES["community_note"])),
            "license_review_note": str(notes_dir / str(FILES["license_review_note"])),
            "key_findings_note": str(evidence_dir / str(FILES["key_findings_note"])),
            "missing_items_note": str(evidence_dir / str(FILES["missing_items_note"])),
        },
        "policy": {
            "read_only": True,
            "default_hidden_workspace": args.workspace_dir == str(DIRECTORIES["default_workspace"]),
            "must_keep_intermediate_inside_workspace": True,
            "required_report_sections": list(OUTPUT["required_sections"]),
        },
    }
    manifest_path.write_text(json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")

    print(json.dumps({
        "manifest": str(manifest_path),
        "run_dir": str(run_dir),
        "report_path": str(report_root / report_name),
    }, ensure_ascii=False, indent=2))
    return 0


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

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

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

from common_config import get_skill_root, load_config


SCRIPT_PATH = Path(__file__)
SKILL_ROOT = get_skill_root(SCRIPT_PATH)
CONFIG = load_config(SKILL_ROOT)
OUTPUT = CONFIG["output"]

REQUIRED_PATH_KEYS = [
    "workspace_root",
    "run_dir",
    "raw_dir",
    "notes_dir",
    "evidence_dir",
    "logs_dir",
    "report_dir",
    "report_path",
]
REQUIRED_FILE_KEYS = [
    "manifest_path",
    "raw_readme",
    "user_context_note",
    "community_note",
    "license_review_note",
    "key_findings_note",
    "missing_items_note",
]


def fail(message: str) -> int:
    print(f"ERROR: {message}", file=sys.stderr)
    return 1


def build_report_regex() -> re.Pattern[str]:
    prefix = re.escape(str(OUTPUT["report_prefix"]))
    ext = re.escape(str(OUTPUT["report_extension"]))
    return re.compile(
        rf"^{prefix}_[A-Za-z0-9._-]+_pr-\d+_\d{{4}}-\d{{2}}-\d{{2}}-\d{{2}}-\d{{2}}(?:-\d{{2}})?{ext}$"
    )


def _load_manifest(path: Path) -> dict[str, Any]:
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise ValueError(f"manifest is not valid JSON: {path}") from exc
    if not isinstance(payload, dict):
        raise ValueError("manifest must be a JSON object")
    return payload


def _require_mapping(payload: dict[str, Any], key: str) -> dict[str, Any]:
    value = payload.get(key)
    if not isinstance(value, dict):
        raise ValueError(f"manifest missing object: {key}")
    return value


def _resolve_manifest_paths(paths: dict[str, Any]) -> dict[str, Path]:
    resolved: dict[str, Path] = {}
    for key in REQUIRED_PATH_KEYS:
        value = paths.get(key)
        if not isinstance(value, str) or not value.strip():
            raise ValueError(f"manifest missing paths.{key}")
        resolved[key] = Path(value).expanduser().resolve()
    return resolved


def main() -> int:
    parser = argparse.ArgumentParser(description="Validate git-pr-review workspace and report artifacts.")
    parser.add_argument("--manifest", required=True, help="Path to manifest.json")
    parser.add_argument("--report", required=True, help="Path to final Markdown report")
    args = parser.parse_args()

    manifest_path = Path(args.manifest).expanduser().resolve()
    report_path = Path(args.report).expanduser().resolve()
    if not manifest_path.exists():
        return fail(f"manifest not found: {manifest_path}")
    if not report_path.exists():
        return fail(f"report not found: {report_path}")

    try:
        manifest = _load_manifest(manifest_path)
        paths = _require_mapping(manifest, "paths")
        files = _require_mapping(manifest, "files")
        policy = manifest.get("policy", {})
        if not isinstance(policy, dict):
            return fail("manifest.policy must be a JSON object")
        resolved = _resolve_manifest_paths(paths)
    except ValueError as exc:
        return fail(str(exc))
    for key in REQUIRED_FILE_KEYS:
        if key not in files:
            return fail(f"manifest missing files.{key}")

    run_dir = resolved["run_dir"]
    workspace_root = resolved["workspace_root"]
    for key in ("raw_dir", "notes_dir", "evidence_dir", "logs_dir"):
        path = resolved[key]
        if not path.exists() or not path.is_dir():
            return fail(f"required directory missing: {path}")
        try:
            path.relative_to(run_dir)
        except ValueError:
            return fail(f"{key} must be inside run_dir: {path}")

    try:
        run_dir.relative_to(workspace_root)
    except ValueError:
        return fail("run_dir must be inside workspace_root")

    if OUTPUT.get("enforce_hidden_workspace_when_default") and policy.get("default_hidden_workspace"):
        if ".bensz-api" not in workspace_root.parts and not workspace_root.name.startswith("."):
            return fail("default workspace must be inside a hidden directory")

    for key in REQUIRED_FILE_KEYS:
        value = files[key]
        path = Path(value).expanduser().resolve()
        if not path.exists() or not path.is_file():
            return fail(f"required scaffold file missing: {key} -> {path}")
        try:
            path.relative_to(run_dir)
        except ValueError:
            return fail(f"scaffold file must stay inside run_dir: {key} -> {path}")

    if report_path != resolved["report_path"]:
        return fail("report path does not match manifest suggestion")
    if not build_report_regex().fullmatch(report_path.name):
        return fail(f"report filename is invalid: {report_path.name}")
    if report_path.suffix.lower() != str(OUTPUT["report_extension"]):
        return fail(f"report must use {OUTPUT['report_extension']} extension")

    required_sections = policy.get("required_report_sections") or OUTPUT.get("required_sections") or []
    report_text = report_path.read_text(encoding="utf-8")
    for section in required_sections:
        if section not in report_text:
            return fail(f"report missing required section: {section}")

    print("OK")
    return 0


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