#!/usr/bin/env python3 """kickoff.py - 프로젝트 킥오프 상태 관리 유틸리티""" import argparse import json import os import sys from datetime import datetime from pathlib import Path BASE_DIR = Path(os.environ.get("KICKOFF_BASE_DIR", "/home/jay/workspace")) KICKOFF_DIR = BASE_DIR / "memory" / "kickoff" APPROVAL_POINTS_TEMPLATE = { "phase0_review": {"status": "pending"}, "phase2_review": {"status": "pending"}, "phase3_approval": {"status": "pending", "by": "jay"}, } PHASE_TO_APPROVAL = { 0: "phase0_review", 2: "phase2_review", 3: "phase3_approval", } VALID_PHASES = [0, 1, 2, 3] def get_state_path(project: str) -> Path: return KICKOFF_DIR / project / "kickoff-state.json" def now_iso() -> str: return datetime.now().strftime("%Y-%m-%dT%H:%M:%S") def load_state(project: str) -> dict: state_path = get_state_path(project) if not state_path.exists(): print( f"[ERROR] kickoff-state.json not found for project '{project}': {state_path}", file=sys.stderr, ) sys.exit(1) with state_path.open("r", encoding="utf-8") as f: return json.load(f) def save_state(project: str, state: dict) -> None: state_path = get_state_path(project) with state_path.open("w", encoding="utf-8") as f: json.dump(state, f, ensure_ascii=False, indent=2) def cmd_init(args) -> None: project = args.project task_id = args.task state_path = get_state_path(project) if state_path.exists(): print( f"[ERROR] kickoff-state.json already exists for project '{project}': {state_path}", file=sys.stderr, ) sys.exit(1) state_path.parent.mkdir(parents=True, exist_ok=True) phases = {str(i): {"status": "pending"} for i in VALID_PHASES} import copy approval_points = copy.deepcopy(APPROVAL_POINTS_TEMPLATE) state = { "project": project, "task_id": task_id, "current_phase": 0, "created_at": now_iso(), "phases": phases, "approval_points": approval_points, } save_state(project, state) print(f"[OK] Initialized kickoff-state.json for project '{project}' (task: {task_id})") print(f" Path: {state_path}") def cmd_status(args) -> None: project = args.project state = load_state(project) print(f"=== Kickoff Status: {state['project']} ===") print(f" Task ID : {state.get('task_id', 'N/A')}") print(f" Current Phase: {state.get('current_phase', 'N/A')}") print(f" Created At : {state.get('created_at', 'N/A')}") print() print("[ Phases ]") phases = state.get("phases", {}) for phase_key in sorted(phases.keys(), key=lambda x: int(x)): phase = phases[phase_key] status = phase.get("status", "unknown") output = phase.get("output", None) completed_at = phase.get("completed_at", None) started_at = phase.get("started_at", None) line = f" Phase {phase_key}: {status}" if started_at: line += f" | started: {started_at}" if completed_at: line += f" | completed: {completed_at}" if output: line += f" | output: {output}" print(line) print() print("[ Approval Points ]") approval_points = state.get("approval_points", {}) for ap_key, ap_val in approval_points.items(): ap_status = ap_val.get("status", "unknown") by = ap_val.get("by", None) at = ap_val.get("at", None) line = f" {ap_key}: {ap_status}" if by: line += f" | by: {by}" if at: line += f" | at: {at}" print(line) def cmd_phase_done(args) -> None: project = args.project phase_num = args.phase output_path = args.output if phase_num not in VALID_PHASES: print( f"[ERROR] Invalid phase number: {phase_num}. Must be one of {VALID_PHASES}.", file=sys.stderr, ) sys.exit(1) state = load_state(project) phases = state.get("phases", {}) phase_key = str(phase_num) # 이미 완료된 Phase 중복 완료 방지 current_phase = phases.get(phase_key, {}) if current_phase.get("status") == "completed": print( f"[ERROR] Phase {phase_num} is already completed.", file=sys.stderr, ) sys.exit(1) # Phase 순서 검증: Phase 0이 아닌 경우 이전 Phase가 completed여야 함 if phase_num > 0: prev_phase_key = str(phase_num - 1) prev_phase = phases.get(prev_phase_key, {}) if prev_phase.get("status") != "completed": print( f"[ERROR] Phase {phase_num - 1} is not completed yet (status: '{prev_phase.get('status', 'unknown')}').", file=sys.stderr, ) sys.exit(1) # 현재 Phase 완료 처리 ts = now_iso() if phase_key not in phases: phases[phase_key] = {} phases[phase_key]["status"] = "completed" phases[phase_key]["completed_at"] = ts phases[phase_key]["output"] = output_path # approval_points 업데이트 (Phase 0, 2, 3 완료 시) ap_key = PHASE_TO_APPROVAL.get(phase_num) if ap_key and ap_key in state.get("approval_points", {}): state["approval_points"][ap_key]["status"] = "pending_review" # 다음 Phase 처리 next_phase_num = phase_num + 1 if next_phase_num in VALID_PHASES: next_phase_key = str(next_phase_num) if next_phase_key not in phases: phases[next_phase_key] = {} phases[next_phase_key]["status"] = "in_progress" phases[next_phase_key]["started_at"] = ts state["current_phase"] = next_phase_num else: # 마지막 Phase 완료 state["current_phase"] = phase_num state["phases"] = phases save_state(project, state) print(f"[OK] Phase {phase_num} marked as completed.") print(f" Output: {output_path}") if next_phase_num in VALID_PHASES: print(f" Phase {next_phase_num} is now in_progress.") else: print(" All phases completed.") if ap_key: print(f" Approval point '{ap_key}' set to pending_review.") def main(): parser = argparse.ArgumentParser( description="kickoff.py - 프로젝트 킥오프 상태 관리 유틸리티" ) subparsers = parser.add_subparsers(dest="command", required=True) # init 명령 init_parser = subparsers.add_parser("init", help="kickoff-state.json 초기화") init_parser.add_argument("--project", required=True, help="프로젝트 이름") init_parser.add_argument("--task", required=True, help="루트 태스크 ID") # status 명령 status_parser = subparsers.add_parser("status", help="현재 상태 조회") status_parser.add_argument("--project", required=True, help="프로젝트 이름") # phase-done 명령 phase_done_parser = subparsers.add_parser("phase-done", help="Phase 완료 마킹") phase_done_parser.add_argument("--project", required=True, help="프로젝트 이름") phase_done_parser.add_argument("--phase", required=True, type=int, help="완료할 Phase 번호 (0~3)") phase_done_parser.add_argument("--output", required=True, help="Phase 출력 파일 경로") args = parser.parse_args() if args.command == "init": cmd_init(args) elif args.command == "status": cmd_status(args) elif args.command == "phase-done": cmd_phase_done(args) else: print(f"[ERROR] Unknown command: {args.command}", file=sys.stderr) sys.exit(1) if __name__ == "__main__": main()