"""CLI entry point with argparse subcommands.""" from __future__ import annotations import argparse import logging import sys from pathlib import Path from cross_eval import __version__ from cross_eval.config import REASONING_EFFORT_CHOICES, resolve_agent_shorthand logger = logging.getLogger(__name__) # --------------------------------------------------------------------------- # Scaffolding templates for `cross-eval init` # --------------------------------------------------------------------------- DEFAULT_CONFIG_YAML = """\ # ─── cross-eval 설정 ─────────────────────────────────────────── # # 기본 제공 에이전트 (별도 정의 없이 바로 사용 가능): # claude-coder, claude-reviewer (Claude, opus 모델) # claude-senior (Claude, opus 모델) # codex-coder, codex-reviewer (Codex, gpt-5.4 모델) # codex-senior (Codex, gpt-5.4 모델) # # CLI에서 --coder claude --reviewer codex --senior codex 같이 축약해서 지정 가능 # ──────────────────────────────────────────────────────────────── # 입력 파일 (이 파일 기준 상대경로) inputs: plan: plan.md checklist: checklist.md # 에이전트 역할 지정 coders: [claude-coder] reviewers: [claude-reviewer] # seniors: [codex-senior] # 파이프라인 종류: simple | cross-review | plan-review | review-only | review-fix | coding-review-fix pipeline: preset:{preset} # 반복 설정 max_iterations: 3 # min_iterations: 1 # PASS여도 최소 이만큼 반복 # 프롬프트 언어 language: {language} # 결과 저장 경로 output_dir: .cross-eval/output # ─── 커스텀 에이전트 (선택) ──────────────────────────────────── # 기본 제공 에이전트를 덮어쓰거나 새 에이전트를 정의할 수 있습니다. # # agents: # my-reviewer: # command: my-tool # args: ["--flag"] # system_prompt: "..." # ──────────────────────────────────────────────────────────────── """ PLAN_SAMPLE_EN = """\ # Project Plan ## Objective [Describe what you want to build] ## Requirements 1. [Requirement 1] 2. [Requirement 2] ## Constraints - [Constraint 1] - [Constraint 2] ## Out of Scope - [Explicitly list what should NOT be implemented] """ PLAN_SAMPLE_KO = """\ # 프로젝트 기획서 ## 목표 [구현할 내용을 설명하세요] ## 요구사항 1. [요구사항 1] 2. [요구사항 2] ## 제약조건 - [제약조건 1] - [제약조건 2] ## 범위 밖 (구현하지 않을 것) - [명시적으로 구현하지 않을 항목 나열] """ CHECKLIST_SAMPLE_EN = """\ # Implementation Checklist ## Functional Requirements - [ ] [Item 1] - [ ] [Item 2] ## Code Quality - [ ] No unused imports or dead code - [ ] Error handling for edge cases - [ ] Follows project coding conventions ## Constraints - [ ] Does NOT add features beyond the plan - [ ] Does NOT introduce unnecessary abstractions """ CHECKLIST_SAMPLE_KO = """\ # 구현 체크리스트 ## 기능 요구사항 - [ ] [항목 1] - [ ] [항목 2] ## 코드 품질 - [ ] 사용하지 않는 import나 죽은 코드 없음 - [ ] 엣지 케이스에 대한 에러 처리 - [ ] 프로젝트 코딩 컨벤션 준수 ## 제약 - [ ] 기획서 범위를 넘는 기능을 추가하지 않음 - [ ] 불필요한 추상화를 도입하지 않음 """ # --------------------------------------------------------------------------- # CLI entry point # --------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> int: """Main CLI entry point.""" parser = argparse.ArgumentParser( prog="cross-eval", description=( "AI 코딩 에이전트의 결과물을 자동으로 검증하는 CLI 도구.\n" "\n" "동작 방식:\n" " 1. 기획서(plan)를 바탕으로 Coder 에이전트가 코드를 작성\n" " 2. Reviewer 에이전트가 기획서 대비 코드를 검토하고 PASS/FAIL 판정\n" " 3. FAIL이면 피드백을 반영해서 1~2를 반복 (최대 N회)\n" "\n" "빠른 시작:\n" " cross-eval init 설정 파일 생성\n" " cross-eval run --plan plan.md 기획서로 바로 실행\n" " cross-eval run .cross-eval/config.yaml 기반 실행\n" "\n" "자세한 사용법: cross-eval --help" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "-v", "--version", action="version", version=f"%(prog)s {__version__}", ) parser.add_argument( "--verbose", action="store_true", help="상세 로그 출력", ) subparsers = parser.add_subparsers(dest="command") # --- init --- init_parser = subparsers.add_parser( "init", help="설정 파일 생성 (config.yaml, plan.md, checklist.md)", description=( "현재 디렉토리에 .cross-eval/ 폴더를 만들고 템플릿을 생성합니다.\n" "이미 있는 파일은 건드리지 않습니다.\n" "\n" "생성되는 파일:\n" " .cross-eval/config.yaml 에이전트, 파이프라인 설정\n" " .cross-eval/plan.md 기획서 템플릿\n" " .cross-eval/checklist.md 체크리스트 템플릿" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) init_parser.add_argument( "--dir", type=Path, default=Path("."), help="초기화할 디렉토리 (기본: 현재 디렉토리)", ) init_parser.add_argument( "--preset", default="simple", choices=[ "simple", "cross-review", "plan-review", "review-only", "review-fix", "coding-review-fix", ], help=( "파이프라인 종류 (기본: simple). " "simple=코딩+리뷰, cross-review=교차리뷰, plan-review=문서기획검토, " "review-only=리뷰만, review-fix=리뷰수렴+자동수정, " "coding-review-fix=초기코딩후리뷰수렴" ), ) init_parser.add_argument( "--lang", default="ko", choices=["en", "ko"], help="프롬프트 언어 (기본: ko)", ) init_parser.add_argument( "--guided", action="store_true", help="대화형 설정 마법사 실행", ) # --- doctor --- doctor_parser = subparsers.add_parser( "doctor", help="실행 환경 점검 (CLI 설치, 인증, 설정 파일 검증)", description="cross-eval 실행에 필요한 환경을 점검합니다.", ) doctor_parser.add_argument( "--dir", type=Path, default=Path("."), help="점검할 디렉토리 (기본: 현재 디렉토리)", ) # --- demo --- demo_parser = subparsers.add_parser( "demo", help="내장 데모 실행 (파이프라인 동작 체험)", description=( "내장된 간단한 기획서로 cross-eval 파이프라인의 전체 동작을 체험합니다.\n" "기본값은 mock 모드(시뮬레이션)이며, --live로 실제 에이전트를 호출할 수 있습니다." ), formatter_class=argparse.RawDescriptionHelpFormatter, ) demo_parser.add_argument( "--live", action="store_true", help="실제 에이전트를 호출하여 데모 실행 (API 비용 발생)", ) demo_parser.add_argument( "--preset", default="simple", choices=["simple", "review-fix", "coding-review-fix"], help="데모할 파이프라인 종류 (기본: simple)", ) demo_parser.add_argument( "--escalate", action="store_true", help="ESCALATE 시나리오 데모 (mock 모드 전용)", ) demo_parser.add_argument( "--timeout", type=int, default=None, metavar="SEC", help="에이전트 1회 호출 제한 시간(초). 0=무제한 (기본: 무제한, --live 전용)", ) # --- run --- run_parser = subparsers.add_parser( "run", help="검증 파이프라인 실행", description=( "기획서(plan)를 기반으로 AI 에이전트가 코딩과 리뷰를 반복합니다.\n" "\n" "설정 파일 없이 바로 실행할 수 있고, config.yaml로도 실행할 수 있습니다.\n" "CLI 옵션이 config.yaml보다 우선합니다." ), epilog=( "파이프라인 종류 (--preset):\n" " ┌──────────────┬─────────────────────────────────────────────────────┐\n" " │ simple │ Coder가 코드 작성 → Reviewer가 리뷰 │\n" " │ (기본값) │ FAIL이면 피드백 반영해서 재코딩, PASS까지 반복 │\n" " ├──────────────┼─────────────────────────────────────────────────────┤\n" " │ review-fix │ 2단계 파이프라인: │\n" " │ │ Reviewer N명 병렬 리뷰 → 취합 → 수정 → 재검증 │\n" " ├──────────────┼─────────────────────────────────────────────────────┤\n" " │ coding- │ 3단계 파이프라인: │\n" " │ review-fix │ 초기 코딩 1회 → 리뷰 취합 → 수정 → 재검증 반복 │\n" " ├──────────────┼─────────────────────────────────────────────────────┤\n" " │ plan-review │ 구현 전 기획서/체크리스트/문서를 검토 │\n" " │ │ 필요하면 현재 코드베이스와의 정합성도 점검 │\n" " ├──────────────┼─────────────────────────────────────────────────────┤\n" " │ review-only │ 코드 작성 없이 Reviewer N명이 기존 코드만 검토 │\n" " │ │ (이미 작성된 코드의 품질 감사용) │\n" " ├──────────────┼─────────────────────────────────────────────────────┤\n" " │ cross-review │ Coder 2명이 각각 구현 → 상대방 코드를 교차 리뷰 │\n" " │ │ (서로 다른 에이전트의 구현 비교용) │\n" " └──────────────┴─────────────────────────────────────────────────────┘\n" "\n" "기본 제공 에이전트:\n" " ┌──────────────────┬─────────┬───────────┬──────────────────────────┐\n" " │ 이름 │ CLI │ 기본 모델 │ 역할 │\n" " ├──────────────────┼─────────┼───────────┼──────────────────────────┤\n" " │ claude-coder │ claude │ opus │ 코드 작성 │\n" " │ claude-reviewer │ claude │ opus │ 코드 리뷰 │\n" " │ claude-senior │ claude │ opus │ 리뷰 취합/판정 │\n" " │ codex-coder │ codex │ gpt-5.4 │ 코드 작성 │\n" " │ codex-reviewer │ codex │ gpt-5.4 │ 코드 리뷰 │\n" " │ codex-senior │ codex │ gpt-5.4 │ 리뷰 취합/판정 │\n" " └──────────────────┴─────────┴───────────┴──────────────────────────┘\n" " --coder, --reviewer, --senior에서 축약 가능: claude → claude-\n" "\n" "사용 예시:\n" "\n" " 기본 실행 (Claude가 코딩하고 Claude가 리뷰):\n" " cross-eval run --plan plan.md\n" "\n" " Codex가 코딩, Claude가 리뷰:\n" " cross-eval run --plan plan.md --coder codex --reviewer claude\n" "\n" " 리뷰어 2명 (Claude + Codex):\n" " cross-eval run --plan plan.md --reviewer claude --reviewer codex\n" "\n" " 리뷰 취합용 Senior 추가:\n" " cross-eval run --plan plan.md --preset review-fix \\\n" " --reviewer claude --reviewer codex --senior codex\n" "\n" " 리뷰 수렴 후 자동 수정 (review-fix):\n" " cross-eval run --plan plan.md --preset review-fix \\\n" " --reviewer claude --reviewer codex\n" "\n" " 초기 코딩 후 리뷰 수렴 + 자동 수정 (coding-review-fix):\n" " cross-eval run --plan plan.md --preset coding-review-fix \\\n" " --reviewer claude --reviewer codex\n" "\n" " 기존 코드 리뷰만 (review-only):\n" " cross-eval run --plan plan.md --preset review-only \\\n" " --reviewer claude --reviewer codex\n" "\n" " 구현 전 문서/기획 검토 (plan-review):\n" " cross-eval run --plan plan.md --preset plan-review \\\n" " --reviewer claude --reviewer codex\n" "\n" " 모델 변경:\n" " cross-eval run --plan plan.md --model sonnet\n" "\n" " config.yaml 기반 실행:\n" " cross-eval run\n" " cross-eval run -c my-config.yaml" ), formatter_class=argparse.RawDescriptionHelpFormatter, ) # -- 입력 파일 -- input_group = run_parser.add_argument_group("입력 파일") input_group.add_argument( "--plan", type=Path, default=None, help="기획서 파일 경로 (필수)", ) input_group.add_argument( "--checklist", type=Path, default=None, help="체크리스트 파일 경로 (선택)", ) input_group.add_argument( "--docs", type=Path, default=None, help="참고 문서 폴더. 폴더 안 모든 파일을 에이전트에게 전달", ) input_group.add_argument( "--input", action="append", dest="inputs", metavar="KEY=PATH", help="추가 입력 파일 (예: --input spec=./api-spec.md)", ) input_group.add_argument( "--env-file", action="append", dest="env_files", type=Path, default=None, help="에이전트 subprocess에 주입할 추가 .env 파일 (여러 개 가능)", ) input_group.add_argument( "--target", action="append", dest="execution_targets", default=None, help="에이전트에게 강조할 실행 대상 힌트 (예: clickhouse, postgres)", ) # -- 에이전트 설정 -- agent_group = run_parser.add_argument_group( "에이전트 설정", "축약 가능: claude → claude-, codex → codex-", ) agent_group.add_argument( "--coder", action="append", dest="coders", metavar="NAME", help="코드를 생성할 에이전트 (여러 개 가능, 기본: claude)", ) agent_group.add_argument( "--reviewer", action="append", dest="reviewers", metavar="NAME", help="코드를 리뷰할 에이전트 (여러 개 가능, 기본: claude)", ) agent_group.add_argument( "--senior", action="append", dest="seniors", metavar="NAME", help="리뷰를 취합하고 최종 판정할 시니어 에이전트 (선택)", ) agent_group.add_argument( "--reasoning-effort", default=None, metavar="LEVEL", choices=REASONING_EFFORT_CHOICES + ("extra-high", "extra_high", "x-high"), help="모든 역할의 reasoning effort (minimal|low|medium|high|xhigh)", ) agent_group.add_argument( "--coder-effort", default=None, metavar="LEVEL", choices=REASONING_EFFORT_CHOICES + ("extra-high", "extra_high", "x-high"), help="Coder용 reasoning effort", ) agent_group.add_argument( "--reviewer-effort", default=None, metavar="LEVEL", choices=REASONING_EFFORT_CHOICES + ("extra-high", "extra_high", "x-high"), help="Reviewer용 reasoning effort", ) agent_group.add_argument( "--senior-effort", default=None, metavar="LEVEL", choices=REASONING_EFFORT_CHOICES + ("extra-high", "extra_high", "x-high"), help="Senior용 reasoning effort", ) agent_group.add_argument( "--agentic", action="store_true", default=False, help="Coder를 agentic 모드로 실행 (worktree에서 파일 직접 수정, git diff로 결과 캡처)", ) agent_group.add_argument( "--model", default=None, metavar="MODEL", help="모든 에이전트의 모델을 한번에 변경 (예: sonnet, opus)", ) agent_group.add_argument( "--coder-model", default=None, metavar="MODEL", help="Coder 에이전트 모델만 변경", ) agent_group.add_argument( "--reviewer-model", default=None, metavar="MODEL", help="Reviewer 에이전트 모델만 변경", ) agent_group.add_argument( "--senior-model", default=None, metavar="MODEL", help="Senior 에이전트 모델만 변경", ) # -- 파이프라인 -- pipe_group = run_parser.add_argument_group("파이프라인") pipe_group.add_argument( "--preset", default=None, choices=[ "simple", "cross-review", "plan-review", "review-only", "review-fix", "coding-review-fix", ], help="파이프라인 종류 (기본: simple). 각 종류 설명은 아래 참조", ) pipe_group.add_argument( "--max-iter", type=int, default=None, help="최대 반복 횟수 (기본: 3)", ) pipe_group.add_argument( "--min-iter", type=int, default=None, help="최소 반복 횟수. PASS여도 이 횟수까지 반복 (기본: 1)", ) pipe_group.add_argument( "--timeout", type=int, default=None, metavar="SEC", help="에이전트 1회 호출 제한 시간(초). 0=무제한 (기본: 무제한)", ) pipe_group.add_argument( "--lang", default=None, choices=["en", "ko"], help="프롬프트 언어 (기본: ko)", ) # -- 기타 -- etc_group = run_parser.add_argument_group("기타") etc_group.add_argument( "-c", "--config", type=Path, default=None, help="설정 파일 경로 (기본: .cross-eval/config.yaml)", ) etc_group.add_argument( "--output-dir", type=Path, default=None, help="결과 저장 디렉토리 (기본: .cross-eval/output/)", ) etc_group.add_argument( "--dry-run", action="store_true", help="실제 실행 없이 에이전트에게 보낼 프롬프트만 미리보기", ) args = parser.parse_args(argv) # Setup logging level = logging.DEBUG if args.verbose else logging.INFO logging.basicConfig( level=level, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) if args.command == "init": return cmd_init(args) elif args.command == "doctor": return cmd_doctor(args) elif args.command == "demo": return cmd_demo(args) elif args.command == "run": return cmd_run(args) else: parser.print_help() return 0 def cmd_doctor(args: argparse.Namespace) -> int: """Run environment health checks.""" from cross_eval.doctor import format_doctor_results, run_doctor checks = run_doctor(args.dir.resolve()) print(format_doctor_results(checks)) has_critical = any(not c.passed and c.critical for c in checks) return 1 if has_critical else 0 def cmd_demo(args: argparse.Namespace) -> int: """Run a built-in demo to show the pipeline lifecycle.""" from cross_eval.demo import run_live_demo, run_mock_demo if args.live: print("\n⚠ --live 모드: 실제 AI 에이전트를 호출합니다 (API 비용 발생).") print(" 내장 피보나치 함수 기획서를 사용합니다.\n") try: answer = input("계속하시겠습니까? [y/N] ").strip().lower() except (EOFError, KeyboardInterrupt): print("\n취소됨.") return 0 if answer not in ("y", "yes"): print("취소됨.") return 0 try: raw_timeout = args.timeout if args.timeout is not None else 0 agent_timeout = None if raw_timeout == 0 else raw_timeout result = run_live_demo(preset=args.preset, timeout=agent_timeout) print(f"\nResult: {result.final_verdict}") print(f"Iterations: {len(result.iterations)}") if result.run_dir: print(f"Output: {result.run_dir}/") return 0 except (RuntimeError, KeyboardInterrupt) as e: if isinstance(e, KeyboardInterrupt): print("\nInterrupted.") return 130 print(f"Demo error: {e}", file=sys.stderr) return 1 else: run_mock_demo(preset=args.preset, show_escalate=args.escalate) return 0 # --------------------------------------------------------------------------- # Guided init wizard # --------------------------------------------------------------------------- _PRESET_DESCRIPTIONS = { "simple": "코딩 + 리뷰 (가장 기본)", "review-fix": "리뷰 → 취합 → 수정 → 재검증 반복", "coding-review-fix": "초기 코딩 + 리뷰 수렴 반복", "plan-review": "구현 전 기획서/문서 검토", "review-only": "기존 코드만 리뷰 (코딩 없음)", "cross-review": "2명이 각각 구현 후 교차 리뷰", } _PRESET_ORDER = [ "simple", "review-fix", "coding-review-fix", "plan-review", "review-only", "cross-review", ] def _prompt_choice( message: str, choices: list[str], descriptions: dict[str, str] | None = None, default: int = 1, ) -> str: """Prompt user to pick from a numbered list.""" print(f"\n{message}") for i, choice in enumerate(choices, 1): desc = f" — {descriptions[choice]}" if descriptions and choice in descriptions else "" marker = " (기본)" if i == default else "" print(f" {i}. {choice}{desc}{marker}") while True: try: raw = input(f"선택 [{default}]: ").strip() except (EOFError, KeyboardInterrupt): print() return choices[default - 1] if not raw: return choices[default - 1] try: idx = int(raw) if 1 <= idx <= len(choices): return choices[idx - 1] except ValueError: if raw in choices: return raw print(f" 1-{len(choices)} 사이 숫자를 입력하세요.") def _prompt_text(message: str, default: str = "") -> str: """Prompt for text input with default.""" suffix = f" [{default}]" if default else "" try: raw = input(f"{message}{suffix}: ").strip() except (EOFError, KeyboardInterrupt): print() return default return raw or default def _run_guided_init(target: Path) -> dict: """Interactive setup wizard. Returns settings dict.""" print("\n━━━ cross-eval 설정 마법사 ━━━\n") lang = _prompt_choice( "언어 / Language:", ["ko", "en"], {"ko": "한국어", "en": "English"}, default=1, ) preset = _prompt_choice( "파이프라인 종류:", _PRESET_ORDER, _PRESET_DESCRIPTIONS, default=1, ) print("\n--- 에이전트 설정 ---") print(" 사용 가능: claude, codex (또는 claude-coder, codex-reviewer 등)") coder = _prompt_text(" Coder 에이전트", default="claude") reviewer = _prompt_text(" Reviewer 에이전트", default="claude") needs_senior = preset in ("review-fix", "coding-review-fix") senior = "" if needs_senior: senior = _prompt_text(" Senior 에이전트", default=reviewer) else: senior = _prompt_text(" Senior 에이전트 (선택, Enter로 건너뛰기)", default="") max_iter = _prompt_text("최대 반복 횟수", default="3") try: max_iter_int = int(max_iter) except ValueError: max_iter_int = 3 create_templates = _prompt_text( "\n템플릿 파일(plan.md, checklist.md) 생성?", default="Y", ).lower() in ("y", "yes", "") return { "lang": lang, "preset": preset, "coder": coder, "reviewer": reviewer, "senior": senior, "max_iter": max_iter_int, "create_templates": create_templates, } def cmd_init(args: argparse.Namespace) -> int: """Scaffold a new cross-eval project.""" target = args.dir.resolve() if args.guided: settings = _run_guided_init(target) args.lang = settings["lang"] args.preset = settings["preset"] # We'll use guided settings for enhanced config generation return _write_init_files(target, args, guided_settings=settings) return _write_init_files(target, args) def _write_init_files( target: Path, args: argparse.Namespace, guided_settings: dict | None = None, ) -> int: """Write config and template files to target directory.""" ce_dir = target / ".cross-eval" ce_dir.mkdir(parents=True, exist_ok=True) lang = args.lang plan_sample = PLAN_SAMPLE_KO if lang == "ko" else PLAN_SAMPLE_EN checklist_sample = CHECKLIST_SAMPLE_KO if lang == "ko" else CHECKLIST_SAMPLE_EN # Generate config content if guided_settings: config_content = _generate_guided_config(args.preset, lang, guided_settings) else: config_content = DEFAULT_CONFIG_YAML.format( preset=args.preset, language=lang, ) files: dict[str, str] = { ".cross-eval/config.yaml": config_content, } # Add templates unless guided mode opted out if not guided_settings or guided_settings.get("create_templates", True): files[".cross-eval/plan.md"] = plan_sample files[".cross-eval/checklist.md"] = checklist_sample created = [] skipped = [] for name, content in files.items(): path = target / name if path.exists(): skipped.append(name) else: path.write_text(content, encoding="utf-8") created.append(name) if created: print(f"\n 생성: {', '.join(created)}") if skipped: print(f" 이미 존재 (건너뜀): {', '.join(skipped)}") print(f"\n 파이프라인: {args.preset}") print(f" 언어: {lang}") if guided_settings: print(f" Coder: {guided_settings['coder']}") print(f" Reviewer: {guided_settings['reviewer']}") if guided_settings.get("senior"): print(f" Senior: {guided_settings['senior']}") print(f" 최대 반복: {guided_settings['max_iter']}") print("") print("다음 단계:") print(" 1. .cross-eval/plan.md 에 기획서 작성") print(" 2. .cross-eval/checklist.md 에 체크리스트 작성 (선택)") print(" 3. cross-eval run 으로 실행") print("") print("팁: cross-eval doctor 로 환경 점검을 먼저 하세요.") print(" cross-eval demo 로 동작 방식을 미리 볼 수 있습니다.") return 0 def _generate_guided_config( preset: str, lang: str, settings: dict, ) -> str: """Generate config.yaml content from guided init settings.""" coder_name = resolve_agent_shorthand(settings["coder"], "coder") reviewer_name = resolve_agent_shorthand(settings["reviewer"], "reviewer") lines = [ "# cross-eval 설정 (guided init으로 생성됨)", "", "inputs:", " plan: plan.md", " checklist: checklist.md", "", f"coders: [{coder_name}]", f"reviewers: [{reviewer_name}]", ] senior = settings.get("senior", "") if senior: senior_name = resolve_agent_shorthand(senior, "senior") lines.append(f"seniors: [{senior_name}]") lines.extend([ "", f"pipeline: preset:{preset}", "", f"max_iterations: {settings['max_iter']}", f"language: {lang}", "output_dir: .cross-eval/output", "", ]) return "\n".join(lines) + "\n" def _read_docs_dir(docs_dir: Path) -> str: """Read all files in a directory and concatenate with filename headers.""" parts: list[str] = [] for f in sorted(docs_dir.iterdir()): if f.is_file() and not f.name.startswith("."): try: content = f.read_text(encoding="utf-8") parts.append(f"### {f.name}\n{content}") except (UnicodeDecodeError, OSError): continue # skip binary or unreadable files return "\n\n".join(parts) def _apply_model_override(config, agent_name: str, model: str) -> None: """Replace --model in agent args.""" agent = config.agents.get(agent_name) if agent is None: return new_args = list(agent.args) for i, arg in enumerate(new_args): if arg == "--model" and i + 1 < len(new_args): new_args[i + 1] = model agent.args = new_args return # --model not found, append it new_args.extend(["--model", model]) agent.args = new_args def _apply_phased_iteration_override(config, max_iter: int | None) -> None: """Apply CLI max-iter to converging phases while preserving setup phases.""" from cross_eval.config import sync_phased_iterations sync_phased_iterations(config, max_iter) def cmd_run(args: argparse.Namespace) -> int: """Load config, validate, and execute the pipeline.""" from cross_eval.config import ( ensure_fix_preset_agentic, apply_input_overrides, default_config, load_config, sync_phased_iterations, validate_config, ) from cross_eval.prompts import PIPELINE_PRESETS from cross_eval.pipeline import run_pipeline # 1. Load config: YAML if exists, otherwise defaults config_path = args.config if config_path is not None: config_path = config_path.resolve() if not config_path.exists(): print(f"Config file not found: {config_path}", file=sys.stderr) return 1 try: config = load_config(config_path) except (ValueError, FileNotFoundError) as e: print(f"Config error: {e}", file=sys.stderr) return 1 config_source = config_path.name else: # Try default location, fall back to built-in defaults default_path = Path(".cross-eval/config.yaml").resolve() if default_path.exists(): try: config = load_config(default_path) config_source = default_path.name except (ValueError, FileNotFoundError) as e: print(f"Config error: {e}", file=sys.stderr) return 1 else: config = default_config() config_source = "defaults" # 2. Apply CLI overrides if args.max_iter is not None: config.max_iterations = args.max_iter if args.min_iter is not None: config.min_iterations = args.min_iter if args.output_dir is not None: config.output_dir = args.output_dir if args.lang is not None: config.language = args.lang # --coder / --reviewer: resolve shorthands and override roles from cross_eval.config import ( _default_seniors_for_preset, _infer_roles, _resolve_agents, apply_reasoning_effort_settings, resolve_agent_shorthand, ) if args.coders or args.reviewers or args.seniors: coders = [resolve_agent_shorthand(c, "coder") for c in (args.coders or [])] reviewers = [resolve_agent_shorthand(r, "reviewer") for r in (args.reviewers or [])] seniors = [resolve_agent_shorthand(s, "senior") for s in (args.seniors or [])] # Fill defaults if only one side specified if not coders: coders = config.coders or ["claude-coder"] if not reviewers: reviewers = config.reviewers or ["claude-reviewer"] if not seniors: seniors = config.seniors config.coders = coders config.reviewers = reviewers config.seniors = seniors # Auto-merge built-in agents config.agents = _resolve_agents(config.agents, coders, reviewers, seniors) # --preset: rebuild pipeline from preset need_rebuild = args.preset is not None or args.coders or args.reviewers or args.seniors if need_rebuild: from cross_eval.prompts import PHASED_PRESETS preset = args.preset or "simple" # Determine which preset was configured (from YAML or defaults) if args.preset is None and config.phases: preset = config.preset_name if config.preset_name != "custom" else "review-fix" elif args.preset is None and not args.coders and not args.reviewers and not args.seniors: pass # no changes needed inferred_coders, inferred_reviewers, inferred_seniors = _infer_roles( list(config.agents.keys()) ) coders = config.coders or inferred_coders reviewers = config.reviewers or inferred_reviewers seniors = config.seniors or [] if not seniors: seniors = _default_seniors_for_preset( f"preset:{preset}", reviewers, config.agents, ) config.agents = _resolve_agents(config.agents, coders, reviewers, seniors) config.coders = coders config.reviewers = reviewers config.seniors = seniors config.preset_name = preset if preset in PHASED_PRESETS: config.phases = PHASED_PRESETS[preset](coders, reviewers, seniors) _apply_phased_iteration_override(config, args.max_iter) config.pipeline = [] elif preset in PIPELINE_PRESETS: config.pipeline = PIPELINE_PRESETS[preset](coders, reviewers, seniors) config.phases = [] if preset in {"plan-review", "review-only"} and args.max_iter is None and args.min_iter is None: config.max_iterations = 1 sync_phased_iterations(config) if args.max_iter is not None: sync_phased_iterations(config, args.max_iter) apply_reasoning_effort_settings( config, reasoning_effort=args.reasoning_effort, coder_effort=args.coder_effort, reviewer_effort=args.reviewer_effort, senior_effort=args.senior_effort, ) # --agentic: convert coder agents to agentic mode if args.agentic: from cross_eval.config import _make_agentic for coder_name in config.coders: if coder_name in config.agents: _make_agentic(config.agents[coder_name]) ensure_fix_preset_agentic(config) # --model: apply to ALL agents if args.model is not None: for agent_name in config.agents: _apply_model_override(config, agent_name, args.model) # --coder-model / --reviewer-model / --senior-model: apply by role if args.coder_model is not None: for coder_name in config.coders: _apply_model_override(config, coder_name, args.coder_model) if args.reviewer_model is not None: for reviewer_name in config.reviewers: _apply_model_override(config, reviewer_name, args.reviewer_model) if args.senior_model is not None: for senior_name in config.seniors: _apply_model_override(config, senior_name, args.senior_model) # --plan / --checklist shortcuts for key, val in [("plan", args.plan), ("checklist", args.checklist)]: if val is not None: p = val.resolve() if not p.exists(): print(f"File not found: {p}", file=sys.stderr) return 1 config.inputs[key] = p # --docs: read all files in directory, inject as {docs} if args.docs is not None: docs_dir = args.docs.resolve() if not docs_dir.is_dir(): print(f"Not a directory: {docs_dir}", file=sys.stderr) return 1 docs_content = _read_docs_dir(docs_dir) if not docs_content: print(f"No files found in: {docs_dir}", file=sys.stderr) return 1 config.inputs["docs"] = docs_content config.inputs["docs_ref"] = str(docs_dir) if args.env_files: for env_file in args.env_files: resolved = env_file.resolve() if not resolved.exists(): print(f"Env file not found: {resolved}", file=sys.stderr) return 1 config.execution.env_files.append(str(resolved)) if args.execution_targets: config.execution.auto_context_targets = list(args.execution_targets) if args.inputs: overrides = {} for item in args.inputs: if "=" not in item: print( f"Invalid --input format: '{item}'. Use KEY=PATH.", file=sys.stderr, ) return 1 key, path = item.split("=", 1) overrides[key] = path apply_input_overrides(config, overrides) # 3. Validate after all overrides errors = validate_config(config) if errors: print("Config error:\n " + "\n ".join(errors), file=sys.stderr) return 1 # 4. Run pipeline logger.info("Config: %s", config_source) logger.info( "Agents: %s", ", ".join(f"{n} ({a.command})" for n, a in config.agents.items()), ) if config.coders or config.reviewers or config.seniors: logger.info("Coders: %s", config.coders) logger.info("Reviewers: %s", config.reviewers) logger.info("Seniors: %s", config.seniors) if config.phases: phase_desc = " → ".join( f"{p.name}(max {p.max_iterations}, {p.consecutive_pass}xPASS)" for p in config.phases ) logger.info("Pipeline: phased [%s], lang=%s", phase_desc, config.language) else: iter_info = f"max {config.max_iterations}" if config.min_iterations > 1: iter_info = f"min {config.min_iterations}, max {config.max_iterations}" logger.info( "Pipeline: %d steps, %s iterations, lang=%s", len(config.pipeline), iter_info, config.language, ) try: raw_timeout = args.timeout if args.timeout is not None else 0 agent_timeout = None if raw_timeout == 0 else raw_timeout result = run_pipeline(config, dry_run=args.dry_run, timeout=agent_timeout) except (RuntimeError, KeyboardInterrupt) as e: if isinstance(e, KeyboardInterrupt): print("\nInterrupted by user.", file=sys.stderr) return 130 print(f"Pipeline error: {e}", file=sys.stderr) return 1 # 4. Print summary print(f"\nResult: {result.final_verdict}") print(f"Iterations: {len(result.iterations)}") if not args.dry_run and result.run_dir: print(f"Output: {result.run_dir}/") if result.final_verdict == "ESCALATE": from cross_eval.report import print_escalation_report print_escalation_report(config, result) return 2 return 0 if result.final_verdict == "PASS" else 1 if __name__ == "__main__": sys.exit(main())