#!/usr/bin/env python3 """ task_scope.py — task 책임 범위 계산 (Guard MVP Phase 1) 4 집합 분리 수집: head_diff : git diff --name-only ..HEAD index_staged : git diff --name-only --cached working_tree_modified : git diff --name-only untracked : git ls-files --others --exclude-standard 산출물: changed_paths.json, scope_matrix.json → --output-dir """ import argparse import json import os import subprocess import sys from pathlib import Path from typing import Optional # ── scope classification ───────────────────────────────────────────────────── def classify_scopes(paths: list[str]) -> list[str]: """변경 파일 목록으로 scope list(sorted unique) 반환.""" scope_set: set[str] = set() for path in paths: if path.startswith("extension/"): scope_set.add("extension") elif path.startswith("server/"): scope_set.add("server") elif path.startswith("dashboard/"): scope_set.add("dashboard") elif path.startswith("config/"): scope_set.add("config") elif path.startswith("skills/"): scope_set.add("skills") elif path.startswith("scripts/"): scope_set.add("scripts") elif path.startswith("docs/") or path.endswith(".md"): scope_set.add("docs") else: scope_set.add("unscoped") return sorted(scope_set) def scope_test_matrix(scopes: list[str]) -> list[str]: """scope list → required test paths list.""" _MAP = { "extension": "extension/__tests__/", "server": "server/tests/", "dashboard": "dashboard/tests/", "config": "tests/config/", "skills": "tests/skills/", "scripts": "tests/scripts/", "docs": None, # smoke only "unscoped": None, # smoke only } result: list[str] = [] for scope in scopes: val = _MAP.get(scope) if val: result.append(val) return result # ── git 호출 ───────────────────────────────────────────────────────────────── def _run_git(args: list[str], cwd: str) -> tuple[int, list[str]]: """git 명령 실행 → (returncode, lines). shell=False.""" cmd = ["git", "-C", cwd] + args result = subprocess.run(cmd, capture_output=True, text=True, check=False) if result.returncode != 0: return result.returncode, [] lines = [line for line in result.stdout.splitlines() if line.strip()] return 0, lines def get_diff_sets(base_sha: str, cwd: str) -> tuple[dict[str, list[str]], Optional[str]]: """4 집합 수집. 실패 시 (부분 결과, 에러메시지).""" errors: list[str] = [] # head_sha rc, head_lines = _run_git(["rev-parse", "HEAD"], cwd) head_sha = head_lines[0] if (rc == 0 and head_lines) else "UNKNOWN" # head_diff rc, head_diff = _run_git(["diff", "--name-only", f"{base_sha}..HEAD"], cwd) if rc != 0: errors.append(f"git diff {base_sha}..HEAD 실패 (rc={rc})") # index_staged rc, index_staged = _run_git(["diff", "--name-only", "--cached"], cwd) if rc != 0: errors.append(f"git diff --cached 실패 (rc={rc})") # working_tree_modified rc, working_tree_modified = _run_git(["diff", "--name-only"], cwd) if rc != 0: errors.append(f"git diff (working tree) 실패 (rc={rc})") # untracked rc, untracked = _run_git(["ls-files", "--others", "--exclude-standard"], cwd) if rc != 0: errors.append(f"git ls-files --others 실패 (rc={rc})") diff_sets = { "base_sha": base_sha, "head_sha": head_sha, "head_diff": head_diff, "index_staged": index_staged, "working_tree_modified": working_tree_modified, "untracked": untracked, } err = "; ".join(errors) if errors else None return diff_sets, err # ── 출력 ────────────────────────────────────────────────────────────────────── def write_outputs(diff_sets: dict, scopes: list[str], test_paths: list[str], output_dir: str) -> None: """output_dir에 changed_paths.json, scope_matrix.json 저장.""" os.makedirs(output_dir, exist_ok=True) changed_paths_obj = { "base_sha": diff_sets["base_sha"], "head_sha": diff_sets["head_sha"], "head_diff": diff_sets["head_diff"], "index_staged": diff_sets["index_staged"], "working_tree_modified": diff_sets["working_tree_modified"], "untracked": diff_sets["untracked"], } scope_matrix_obj = { "scopes": scopes, "test_paths": test_paths, } with open(os.path.join(output_dir, "changed_paths.json"), "w", encoding="utf-8") as f: json.dump(changed_paths_obj, f, indent=2, ensure_ascii=False) with open(os.path.join(output_dir, "scope_matrix.json"), "w", encoding="utf-8") as f: json.dump(scope_matrix_obj, f, indent=2, ensure_ascii=False) # ── task-2569 AD-1~3: lock_sha 기반 diff base 자동 분기 ────────────────────── def _resolve_diff_base(task_id: str, workspace: Path) -> str: """lock_sha..HEAD 기준 diff base 결정 (PR sub-task worktree 오판 방지). worktree 컨텍스트 + lock 파일에 lock_sha 존재 시 lock_sha 반환, 그 외에는 origin/main fallback. """ lock_file = workspace / ".tasks" / "locks" / f"{task_id}.lock" lock_sha = "" if lock_file.exists(): try: lock_sha = json.loads(lock_file.read_text()).get("lock_sha", "") except Exception: lock_sha = "" # worktree 컨텍스트 감지 cwd = os.getcwd() is_worktree = "/.worktrees/" in cwd or "/.worktrees/" in os.environ.get("GIT_DIR", "") if is_worktree and lock_sha: return lock_sha return "origin/main" # ── CLI ─────────────────────────────────────────────────────────────────────── def main() -> None: parser = argparse.ArgumentParser( description="task_scope.py — task 책임 범위 계산 (Guard MVP Phase 1)" ) parser.add_argument("--base-sha", required=True, help="비교 기준 SHA (예: origin/main, 또는 auto/AUTO로 lock_sha 자동 분기)") parser.add_argument("--output-dir", default=None, help="JSON 저장 디렉토리 (--no-output 시 무시)") parser.add_argument("--cwd", default="/home/jay/workspace", help="git 저장소 루트 (기본: /home/jay/workspace)") parser.add_argument("--task-id", default=None, help="task ID (auto/AUTO base-sha 시 lock_sha 조회에 사용)") parser.add_argument("--workspace", default="/home/jay/workspace", help="워크스페이스 루트 (lock 파일 검색용, 기본: /home/jay/workspace)") parser.add_argument("--no-output", action="store_true", help="stdout에만 출력 (파일 저장 안 함, 테스트용)") args = parser.parse_args() if not args.no_output and not args.output_dir: print("오류: --output-dir 또는 --no-output 중 하나가 필요합니다.", file=sys.stderr) sys.exit(1) # task-2569 AD-1~3: lock_sha 자동 분기 base_sha = args.base_sha if base_sha in ("auto", "AUTO"): task_id = args.task_id or "" base_sha = _resolve_diff_base(task_id, Path(args.workspace)) print(f"[task-scope] base_sha auto → {base_sha}", file=sys.stderr) diff_sets, err = get_diff_sets(base_sha, args.cwd) if err: print(f"[task-scope] ERROR: {err}", file=sys.stderr) sys.exit(1) # scope classification: head_diff 기준 (push될 커밋) all_paths = diff_sets["head_diff"] scopes = classify_scopes(all_paths) test_paths = scope_test_matrix(scopes) if args.no_output: out = { "changed_paths": { "base_sha": diff_sets["base_sha"], "head_sha": diff_sets["head_sha"], "head_diff": diff_sets["head_diff"], "index_staged": diff_sets["index_staged"], "working_tree_modified": diff_sets["working_tree_modified"], "untracked": diff_sets["untracked"], }, "scope_matrix": { "scopes": scopes, "test_paths": test_paths, }, } print(json.dumps(out, ensure_ascii=False)) else: write_outputs(diff_sets, scopes, test_paths, args.output_dir) print(f"[task-scope] 저장 완료: {args.output_dir}", file=sys.stderr) print(f"[task-scope] scopes={scopes}", file=sys.stderr) print(f"[task-scope] test_paths={test_paths}", file=sys.stderr) sys.exit(0) if __name__ == "__main__": main()