#!/usr/bin/env python3 """ Harness Step Executor — phase 내 Step을 순차 실행하고 결과와 파일 소유권을 검증한다. Usage: python scripts/execute.py [--push] """ import argparse import contextlib import json import os import re import subprocess import sys import tempfile import threading import time import types from datetime import datetime, timezone, timedelta from pathlib import Path, PureWindowsPath from typing import Optional ROOT = Path(__file__).resolve().parent.parent class HarnessError(RuntimeError): """Executor가 안전하게 계속할 수 없는 검증 또는 실행 오류.""" class CodexEnvironmentError(HarnessError): """재시도로 해결할 수 없는 Codex CLI 환경 오류.""" class StepOwnershipError(HarnessError): """승인된 Step 파일 범위 밖의 변경; 사용자 데이터를 되돌리지 않는다.""" def _configure_utf8_stdio(): """Windows 콘솔에서도 Harness 진행 문자를 안정적으로 출력한다.""" for stream in (sys.stdout, sys.stderr): reconfigure = getattr(stream, "reconfigure", None) if reconfigure is not None: reconfigure(encoding="utf-8") @contextlib.contextmanager def progress_indicator(label: str): """터미널 진행 표시기. with 문으로 사용하며 .elapsed 로 경과 시간을 읽는다.""" frames = "◐◓◑◒" stop = threading.Event() t0 = time.monotonic() def _animate(): idx = 0 while not stop.wait(0.12): sec = int(time.monotonic() - t0) sys.stderr.write(f"\r{frames[idx % len(frames)]} {label} [{sec}s]") sys.stderr.flush() idx += 1 sys.stderr.write("\r" + " " * (len(label) + 20) + "\r") sys.stderr.flush() th = threading.Thread(target=_animate, daemon=True) th.start() info = types.SimpleNamespace(elapsed=0.0) try: yield info finally: stop.set() th.join() info.elapsed = time.monotonic() - t0 class StepExecutor: """Phase 디렉토리 안의 step들을 순차 실행하는 하네스.""" MAX_TRANSPORT_ATTEMPTS = 2 CODEX_TIMEOUT_SECONDS = 14400 CODEX_SANDBOX_ENV = "FEM_HARNESS_CODEX_SANDBOX" CODEX_SANDBOX_MODES = frozenset({"workspace-write", "danger-full-access"}) PHASE_NAME = re.compile(r"^[a-z0-9]+(?:[._-][a-z0-9]+)*$") STEP_NAME = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") STEP_STATUSES = frozenset({"pending", "completed", "error", "blocked"}) AGENT_STEP_FIELDS = frozenset( {"status", "summary", "error_message", "blocked_reason"} ) IMPLEMENTATION_AGENT_PROFILE = Path( ".codex/agents/implementation-agent.toml" ) FEAT_MSG = "feat({phase}): step {num} - {name}" CHORE_MSG = "chore({phase}): step {num} output" TZ = timezone(timedelta(hours=9)) def __init__( self, phase_dir_name: str, *, auto_push: bool = False, danger_full_access_approved: bool = False, ): if not isinstance(phase_dir_name, str) or not self.PHASE_NAME.fullmatch( phase_dir_name ): raise HarnessError( "phase directory must be one safe repository-relative name" ) root_path = ROOT.resolve() self._root = str(root_path) self._phases_dir = (root_path / "phases").resolve() try: self._phases_dir.relative_to(root_path) except ValueError as exc: raise HarnessError( "phases directory must resolve inside the Harness repository root" ) from exc self._phase_dir = (self._phases_dir / phase_dir_name).resolve() try: self._phase_dir.relative_to(self._phases_dir.resolve()) except ValueError as exc: raise HarnessError( "phase directory must stay inside the repository phases directory" ) from exc self._phase_dir_name = phase_dir_name self._top_index_file = self._phases_dir / "index.json" self._auto_push = auto_push self._danger_full_access_approved = danger_full_access_approved self._validated_sandbox_mode: str | None = None self._implementation_profile_cache: tuple[str, str, str] | None = None if not self._phase_dir.is_dir(): raise HarnessError(f"phase directory not found: {self._phase_dir}") self._index_file = self._phase_dir / "index.json" if not self._index_file.exists(): raise HarnessError(f"phase index not found: {self._index_file}") if not self._top_index_file.is_file(): raise HarnessError( f"top-level phase index not found: {self._top_index_file}" ) self._read_top_index() idx = self._read_phase_index() self._project = idx["project"] self._phase_name = idx["phase"] self._total = len(idx["steps"]) def run(self): self._print_header() self._check_blockers() self._preflight_repository() self._checkout_branch() guardrails = self._load_guardrails() self._ensure_created_at() self._execute_all_steps(guardrails) self._finalize() # --- timestamps --- def _stamp(self) -> str: return datetime.now(self.TZ).strftime("%Y-%m-%dT%H:%M:%S%z") # --- JSON I/O --- def _control_scope(self, path: Path) -> Path: if path == self._top_index_file: return self._phases_dir try: path.relative_to(self._phase_dir) except ValueError as exc: raise HarnessError( f"control path is outside the selected phase: {path}" ) from exc return self._phase_dir def _validate_control_parent(self, path: Path, scope: Path) -> None: root = Path(self._root) try: resolved_scope = scope.resolve(strict=True) resolved_scope.relative_to(root) path.parent.resolve(strict=True).relative_to(resolved_scope) except (OSError, ValueError) as exc: raise HarnessError( f"control path parent escapes its trusted scope: {path}" ) from exc def _read_control_bytes(self, path: Path) -> bytes: scope = self._control_scope(path) self._validate_control_parent(path, scope) try: metadata = path.lstat() resolved = path.resolve(strict=True) resolved.relative_to(scope.resolve(strict=True)) except (OSError, ValueError) as exc: raise HarnessError( f"protected control file is missing or escapes its scope: {path}" ) from exc if path.is_symlink() or not path.is_file() or metadata.st_nlink != 1: raise HarnessError( f"protected control file must be a non-link regular file: {path}" ) try: return path.read_bytes() except OSError as exc: raise HarnessError( f"protected control file could not be read: {path}: {exc}" ) from exc def _atomic_write_control_bytes(self, path: Path, contents: bytes) -> None: scope = self._control_scope(path) self._validate_control_parent(path, scope) if path.exists() and not path.is_symlink() and not path.is_file(): raise HarnessError( f"control output path is not a regular file: {path}" ) temporary_path: Path | None = None try: with tempfile.NamedTemporaryFile( mode="wb", dir=path.parent, prefix=f".{path.name}.", suffix=".tmp", delete=False, ) as handle: temporary_path = Path(handle.name) handle.write(contents) handle.flush() os.fsync(handle.fileno()) os.replace(temporary_path, path) except OSError as exc: raise HarnessError( f"control output could not be written atomically: {path}: {exc}" ) from exc finally: if temporary_path is not None and temporary_path.exists(): temporary_path.unlink() def _read_json(self, p: Path) -> dict: try: data = json.loads(self._read_control_bytes(p).decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise HarnessError(f"invalid JSON file {p}: {exc}") from exc if not isinstance(data, dict): raise HarnessError(f"JSON root must be an object: {p}") return data def _write_json(self, p: Path, data: dict): contents = json.dumps(data, indent=2, ensure_ascii=False).encode("utf-8") self._atomic_write_control_bytes(p, contents) def _read_phase_index(self) -> dict: index = self._read_json(self._index_file) self._validate_phase_index(index) return index def _read_top_index(self) -> dict: index = self._read_json(self._top_index_file) phases = index.get("phases") if not isinstance(phases, list) or not phases: raise HarnessError("top-level phase index requires a non-empty phases array") selected_count = 0 seen: set[str] = set() for position, phase in enumerate(phases): if not isinstance(phase, dict): raise HarnessError( f"top-level phase entry {position} must be an object" ) directory = phase.get("dir") if not isinstance(directory, str) or not self.PHASE_NAME.fullmatch( directory ): raise HarnessError( f"top-level phase entry {position} has an invalid directory" ) if directory in seen: raise HarnessError( f"top-level phase '{directory}' must appear exactly once" ) seen.add(directory) if directory == self._phase_dir_name: selected_count += 1 if phase.get("status") not in self.STEP_STATUSES: raise HarnessError( f"top-level phase entry {position} has an invalid status" ) if selected_count != 1: raise HarnessError( f"selected phase '{self._phase_dir_name}' must appear exactly once " "in the top-level phase index" ) return index def _validate_phase_index(self, index: dict) -> None: project = index.get("project") if not isinstance(project, str) or not project.strip(): raise HarnessError("phase index project must be a non-empty string") phase = index.get("phase") if phase != self._phase_dir_name: raise HarnessError( "phase index phase must match its phase directory name" ) steps = index.get("steps") if not isinstance(steps, list) or not steps: raise HarnessError("phase index steps must be a non-empty array") expected_steps = list(range(len(steps))) actual_steps = [step.get("step") if isinstance(step, dict) else None for step in steps] if actual_steps != expected_steps: raise HarnessError("phase step numbers must be contiguous from zero") state = "completed-prefix" for expected, step in enumerate(steps): if not isinstance(step, dict): raise HarnessError(f"phase step {expected} must be an object") name = step.get("name") if not isinstance(name, str) or not self.STEP_NAME.fullmatch(name): raise HarnessError( f"phase step {expected} name must be a kebab-case slug" ) status = step.get("status") if status not in self.STEP_STATUSES: raise HarnessError(f"phase step {expected} has invalid status") if "owned_paths" in step: self._step_owned_paths(step) self._validate_attempt_history(step) if status == "completed": if state != "completed-prefix": raise HarnessError("phase step status order is invalid") if not isinstance(step.get("summary"), str) or not step["summary"].strip(): raise HarnessError( f"completed phase step {expected} requires summary" ) elif status == "pending": if state == "completed-prefix": state = "pending-suffix" else: if state != "completed-prefix": raise HarnessError("phase step status order is invalid") detail_key = "error_message" if status == "error" else "blocked_reason" if not isinstance(step.get(detail_key), str) or not step[detail_key].strip(): raise HarnessError( f"{status} phase step {expected} requires {detail_key}" ) state = "terminal-then-pending" if state == "terminal-then-pending" and status == "pending": state = "pending-suffix" step_file = self._phase_dir / f"step{expected}.md" if not step_file.is_file(): raise HarnessError(f"phase step file not found: {step_file}") @staticmethod def _path_key(path: str) -> str: return os.path.normcase(path).replace("\\", "/") def _step_owned_paths(self, step: dict) -> tuple[str, ...]: paths = step.get("owned_paths") if not isinstance(paths, list): raise HarnessError(f"Step {step['step']} requires an owned_paths file array before execution") normalized = [] seen = set() for path in paths: if not isinstance(path, str): raise HarnessError("owned_paths entries must be repository-relative file paths") path = path.replace("\\", "/") parts = path.split("/") if ( not path or PureWindowsPath(path).drive or path.startswith("/") or any(part in {"", ".", ".."} or part.endswith((".", " ")) for part in parts) or any(ord(char) < 32 or char in ':*?[]<>"|' for char in path) or any(part.casefold() == ".git" or PureWindowsPath(part).is_reserved() for part in parts) or parts[0].casefold() == "phases" ): raise HarnessError(f"owned_paths requires an exact non-control relative file path: {path!r}") key = self._path_key(path) if key in seen: raise HarnessError(f"owned_paths contains duplicate/aliased paths: {path}") seen.add(key) normalized.append(path) return tuple(normalized) def _validate_owned_file(self, path: str) -> None: root = Path(self._root) candidate = root / path try: resolved = candidate.resolve() resolved.relative_to(root) if resolved != candidate or candidate.is_symlink() or candidate.is_dir(): raise ValueError("directory or linked path") if candidate.exists() and candidate.lstat().st_nlink != 1: raise ValueError("hard-linked file") except (OSError, RuntimeError, ValueError) as exc: raise HarnessError(f"owned_paths must name non-linked files inside the repository: {path}") from exc def _validate_attempt_history(self, step: dict) -> None: if "attempt_history" not in step: return history = step["attempt_history"] if not isinstance(history, list): raise HarnessError("attempt_history must be an Executor-owned array") for number, record in enumerate(history, 1): if ( not isinstance(record, dict) or type(record.get("attempt")) is not int or record["attempt"] != number or "exit_code" not in record or (record["exit_code"] is not None and type(record["exit_code"]) is not int) or not isinstance(record.get("reported_status"), str) or record["reported_status"] not in self.STEP_STATUSES or not isinstance(record.get("recorded_at"), str) or not record["recorded_at"] or not isinstance(record.get("reported_reason"), str) or not isinstance(record.get("reason"), str) or not isinstance(record.get("diagnostic"), str) or not isinstance(record.get("classification"), str) or record["classification"] not in { "environment", "reported-blocked", "reported-error", "completed", "inconsistent-result", "transport-timeout", "process-error", "missing-result", "ownership-violation", "commit-error", } ): raise HarnessError(f"attempt_history contains an invalid record at attempt {number}") def _snapshot_phase_control_files( self, index: dict ) -> tuple[bytes, dict[str, bytes]]: expected_names = {f"step{step['step']}.md" for step in index["steps"]} current_names = { path.name for path in self._phase_dir.glob("step*.md") if path.is_file() } if current_names != expected_names: raise HarnessError( "protected phase control file inventory does not match the phase index" ) try: top_index = self._read_control_bytes(self._top_index_file) step_files = { f"step{step['step']}.md": ( self._read_control_bytes( self._phase_dir / f"step{step['step']}.md" ) ) for step in index["steps"] } except HarnessError as exc: raise HarnessError( f"protected phase control file could not be read: {exc}" ) from exc return top_index, step_files def _validate_phase_control_files( self, snapshot: tuple[bytes, dict[str, bytes]] ) -> None: expected_top, expected_steps = snapshot try: current_top = self._read_control_bytes(self._top_index_file) current_names = { path.name for path in self._phase_dir.glob("step*.md") if path.is_file() } current_steps = { name: self._read_control_bytes(self._phase_dir / name) for name in current_names } except HarnessError as exc: raise HarnessError( f"protected phase control file changed or disappeared: {exc}" ) from exc if current_top != expected_top or current_steps != expected_steps: raise HarnessError( "Codex modified a protected phase control file; only the current " "Step outcome in the phase index may change" ) def _record_untrusted_step_result( self, index_before: dict, current_step: int, classification: str, reason: str, control_snapshot: tuple[bytes, dict[str, bytes]], ) -> None: expected_top, expected_steps = control_snapshot self._top_index_file.parent.mkdir(parents=True, exist_ok=True) self._atomic_write_control_bytes(self._top_index_file, expected_top) self._phase_dir.mkdir(parents=True, exist_ok=True) for name, contents in expected_steps.items(): self._atomic_write_control_bytes(self._phase_dir / name, contents) recovered = json.loads(json.dumps(index_before)) current = recovered["steps"][current_step] self._clear_step_outcome(current) current["status"] = "error" current["error_message"] = f"[{classification}] {reason}" current["failed_at"] = self._stamp() self._write_json(self._index_file, recovered) self._update_top_index("error") def _validate_step_index_mutation( self, before: dict, after: dict, current_step: int ) -> None: before_metadata = {key: value for key, value in before.items() if key != "steps"} after_metadata = {key: value for key, value in after.items() if key != "steps"} if before_metadata != after_metadata: raise HarnessError( "Codex modified protected phase index metadata" ) before_steps = before["steps"] after_steps = after["steps"] if len(before_steps) != len(after_steps): raise HarnessError("Codex modified the protected Step inventory") for position, (old_step, new_step) in enumerate( zip(before_steps, after_steps) ): if position != current_step: if old_step != new_step: raise HarnessError( f"Codex modified non-current Step {position}" ) continue old_fixed = { key: value for key, value in old_step.items() if key not in self.AGENT_STEP_FIELDS } new_fixed = { key: value for key, value in new_step.items() if key not in self.AGENT_STEP_FIELDS } if old_fixed != new_fixed: raise HarnessError( "Codex modified immutable metadata on the current Step" ) # --- git --- def _run_git(self, *args) -> subprocess.CompletedProcess: cmd = ["git"] + list(args) return subprocess.run( cmd, cwd=self._root, capture_output=True, text=True, encoding="utf-8" ) def _run_git_checked(self, *args) -> subprocess.CompletedProcess: result = self._run_git(*args) if result.returncode != 0: diagnostic = (result.stderr or result.stdout).strip() message = f"git {args[0]} failed" if diagnostic: message = f"{message}: {diagnostic}" raise HarnessError(message) return result def _git_execution_identity(self) -> tuple[str, str]: branch = self._run_git_checked( "symbolic-ref", "--quiet", "HEAD" ).stdout.strip() head = self._run_git_checked( "rev-parse", "--verify", "HEAD" ).stdout.strip() if not branch or not head: raise HarnessError("Git branch and HEAD identity must be non-empty") return branch, head def _worktree_paths(self) -> tuple[str, ...]: result = self._run_git_checked( "status", "--porcelain=v1", "-z", "--untracked-files=all" ) paths: list[str] = [] records = result.stdout.split("\0") position = 0 while position < len(records): record = records[position] position += 1 if not record: continue if len(record) < 4 or record[2] != " ": raise HarnessError("git status returned an invalid porcelain record") status = record[:2] paths.append(record[3:].replace("\\", "/")) if "R" in status or "C" in status: if position >= len(records) or not records[position]: raise HarnessError( "git status returned an incomplete rename/copy record" ) paths.append(records[position].replace("\\", "/")) position += 1 return tuple(dict.fromkeys(paths)) def _is_phase_owned_path(self, path: str) -> bool: phase_prefix = f"phases/{self._phase_dir_name}/" declared = { "phases/index.json", f"{phase_prefix}index.json", *( f"{phase_prefix}step{step_num}.md" for step_num in range(self._total) ), } return path in declared or self._is_step_output_path(path) def _is_step_output_path(self, path: str) -> bool: phase_prefix = f"phases/{self._phase_dir_name}/" if not path.startswith(phase_prefix): return False name = path.removeprefix(phase_prefix) match = re.fullmatch(r"step(\d+)-output\.json", name) return match is not None and int(match.group(1)) < self._total def _preflight_repository(self) -> None: for step in self._read_phase_index()["steps"]: if step["status"] == "pending": for path in self._step_owned_paths(step): self._validate_owned_file(path) self._implementation_agent_profile() top_level_result = self._run_git_checked("rev-parse", "--show-toplevel") raw_top_level = top_level_result.stdout.strip() if not raw_top_level or Path(raw_top_level).resolve() != Path(self._root): raise HarnessError( "Git top-level must exactly match the Harness repository root" ) mode = self._codex_sandbox_mode() paths = self._worktree_paths() if mode == "danger-full-access": if paths: raise CodexEnvironmentError( "danger-full-access requires a clean worktree" ) return unrelated = [path for path in paths if not self._is_phase_owned_path(path)] if unrelated: preview = ", ".join(unrelated[:5]) raise HarnessError( "Harness requires a clean worktree outside the selected phase; " f"unrelated changes: {preview}" ) def _is_isolated_worktree(self) -> bool: git_dir = Path( self._run_git_checked("rev-parse", "--absolute-git-dir").stdout.strip() ).resolve() common_raw = self._run_git_checked("rev-parse", "--git-common-dir").stdout.strip() common_dir = Path(common_raw) if not common_dir.is_absolute(): common_dir = Path(self._root) / common_dir return git_dir != common_dir.resolve() def _checkout_branch(self): branch = f"feat-{self._phase_name}" branch_ref = f"refs/heads/{branch}" current = self._run_git("symbolic-ref", "--quiet", "HEAD") if current.returncode == 0 and current.stdout.strip() == branch_ref: return exists = self._run_git( "show-ref", "--verify", "--quiet", branch_ref ) if exists.returncode == 0: switched = self._run_git("switch", branch) elif exists.returncode == 1: switched = self._run_git("switch", "-c", branch) else: raise HarnessError( f"git show-ref failed for branch '{branch}': " f"{(exists.stderr or exists.stdout).strip()}" ) if switched.returncode != 0: raise HarnessError( f"git switch failed for branch '{branch}': " f"{(switched.stderr or switched.stdout).strip()}" ) verified = self._run_git_checked( "symbolic-ref", "--quiet", "HEAD" ) if verified.stdout.strip() != branch_ref: raise HarnessError( f"Git did not enter the expected local branch '{branch}'; " f"actual={verified.stdout.strip()!r}" ) print(f" Branch: {branch}") def _assert_step_scope(self, owned_paths: tuple[str, ...]) -> tuple[str, ...]: allowed = {self._path_key(path) for path in owned_paths} changed = self._worktree_paths() unexpected = [ path for path in changed if not self._is_phase_owned_path(path) and self._path_key(path) not in allowed ] if unexpected: raise StepOwnershipError("Step ownership violation: " + ", ".join(unexpected)) try: for path in owned_paths: self._validate_owned_file(path) except HarnessError as exc: raise StepOwnershipError(str(exc)) from exc return changed def _commit_paths( self, paths: list[str], message: str, *, owned_paths: tuple[str, ...] = () ) -> bool: validated_paths = self._assert_step_scope(owned_paths) if not paths: return False # Codex may stage files even though the prompt tells it not to. Reset only # the index so the following path-scoped add controls the commit content; # working-tree data is preserved. This also expands staged renames back to # a deletion/addition pair, allowing both porcelain pathnames to be staged. if validated_paths: self._run_git_checked( "--literal-pathspecs", "reset", "--quiet", "HEAD", "--", *sorted(validated_paths) ) self._run_git_checked("--literal-pathspecs", "add", "-A", "--", *sorted(paths)) staged = self._run_git_checked( "diff", "--cached", "--name-only", "--no-renames", "-z" ).stdout selected = {self._path_key(path) for path in paths} unexpected = [ path for path in staged.split("\0") if path and self._path_key(path) not in selected ] if unexpected: raise StepOwnershipError("Unexpected staged paths before commit: " + ", ".join(unexpected)) self._assert_step_scope(owned_paths) if not staged: return False self._run_git_checked("commit", "-m", message) return True def _commit_step(self, step_num: int, step_name: str, owned_paths: tuple[str, ...]): changed = self._assert_step_scope(owned_paths) code_paths = [ path for path in changed if not self._is_phase_owned_path(path) ] message = self.FEAT_MSG.format( phase=self._phase_name, num=step_num, name=step_name ) if self._commit_paths(code_paths, message, owned_paths=owned_paths): print(f" Commit: {message}") metadata_paths = [ path for path in self._assert_step_scope(owned_paths) if self._is_phase_owned_path(path) and not self._is_step_output_path(path) ] message = self.CHORE_MSG.format( phase=self._phase_name, num=step_num ) self._commit_paths(metadata_paths, message, owned_paths=owned_paths) # --- top-level index --- def _update_top_index(self, status: str): if status not in {"pending", "completed", "error", "blocked"}: raise HarnessError(f"invalid top-level phase status: {status}") top = self._read_top_index() phases = top["phases"] ts = self._stamp() matched = False for phase in phases: if not isinstance(phase, dict): raise HarnessError("top-level phase entry must be an object") if phase.get("dir") == self._phase_dir_name: matched = True phase["status"] = status ts_key = {"completed": "completed_at", "error": "failed_at", "blocked": "blocked_at"}.get(status) for key in ("completed_at", "failed_at", "blocked_at"): if key != ts_key: phase.pop(key, None) if ts_key: phase[ts_key] = ts break if not matched: # Defensive: _read_top_index already enforces this invariant. raise HarnessError("selected phase is missing from the top-level phase index") self._write_json(self._top_index_file, top) # --- guardrails & context --- def _load_guardrails(self) -> str: sections = [] agents_md = ROOT / "AGENTS.md" if agents_md.exists(): sections.append( "## 프로젝트 규칙 (AGENTS.md)\n\n" f"{agents_md.read_text(encoding='utf-8')}" ) docs_dir = ROOT / "docs" if docs_dir.is_dir(): for doc in sorted(docs_dir.glob("*.md")): sections.append( f"## {doc.stem}\n\n{doc.read_text(encoding='utf-8')}" ) return "\n\n---\n\n".join(sections) if sections else "" @staticmethod def _build_step_context(index: dict) -> str: lines = [ f"- Step {s['step']} ({s['name']}): {s['summary']}" for s in index["steps"] if s["status"] == "completed" and s.get("summary") ] if not lines: return "" return "## 이전 Step 산출물\n\n" + "\n".join(lines) + "\n\n" def _build_preamble(self, guardrails: str, step_context: str, prev_error: Optional[str] = None) -> str: retry_section = "" if prev_error: retry_section = ( f"\n## ⚠ 이전 시도 실패 — 아래 에러를 반드시 참고하여 수정하라\n\n" f"{prev_error}\n\n---\n\n" ) _, _, role_instructions = self._implementation_agent_profile() return ( f"당신은 {self._project} 프로젝트의 개발자입니다. 아래 step을 수행하세요.\n\n" f"## Implementation Agent role\n\n{role_instructions}\n\n---\n\n" f"{guardrails}\n\n---\n\n" f"{step_context}{retry_section}" f"## 작업 규칙\n\n" f"1. 이전 step에서 작성된 코드를 확인하고 일관성을 유지하라.\n" f"2. 이 step에 명시된 작업만 수행하라. 추가 기능이나 파일을 만들지 마라.\n" f"3. 기존 테스트를 깨뜨리지 마라.\n" f"4. AC(Acceptance Criteria) 검증을 직접 실행하라.\n" f"5. /phases/{self._phase_dir_name}/index.json의 해당 step status를 업데이트하라:\n" f" - AC 통과 → \"completed\" + \"summary\" 필드에 이 step의 산출물을 한 줄로 요약\n" f" - 구현 또는 검증 실패 → \"error\" + \"error_message\" 기록\n" f" - 사용자 개입이 필요한 경우 (API 키, 인증, 수동 설정 등) → \"blocked\" + \"blocked_reason\" 기록 후 즉시 중단\n" f"6. 변경사항을 직접 커밋하지 마라. Git 커밋과 timestamp는 실행기가 처리한다.\n\n" f"---\n\n" ) # --- Codex 호출 --- def _codex_sandbox_mode(self) -> str: if self._validated_sandbox_mode is not None: return self._validated_sandbox_mode mode = os.environ.get(self.CODEX_SANDBOX_ENV, "workspace-write") if mode not in self.CODEX_SANDBOX_MODES: allowed = ", ".join(sorted(self.CODEX_SANDBOX_MODES)) raise CodexEnvironmentError( f"{self.CODEX_SANDBOX_ENV} must be one of: {allowed}" ) if mode == "danger-full-access": if not self._danger_full_access_approved: raise CodexEnvironmentError( "danger-full-access requires explicit authorization through " "--approve-danger-full-access" ) if not self._is_isolated_worktree(): raise CodexEnvironmentError( "danger-full-access requires an isolated linked worktree" ) if self._worktree_paths(): raise CodexEnvironmentError( "danger-full-access requires a clean worktree" ) self._validated_sandbox_mode = mode return mode def _implementation_agent_profile(self) -> tuple[str, str, str]: if self._implementation_profile_cache is not None: return self._implementation_profile_cache profile_path = Path(self._root) / self.IMPLEMENTATION_AGENT_PROFILE try: profile_text = profile_path.read_text(encoding="utf-8") except OSError as exc: raise CodexEnvironmentError( f"Implementation Agent profile을 읽을 수 없습니다: {profile_path}" ) from exc settings = {} for line in profile_text.splitlines(): key, separator, value = line.partition("=") key = key.strip() if separator and key in {"model", "model_reasoning_effort"}: try: settings[key] = json.loads(value.strip()) except json.JSONDecodeError as exc: raise CodexEnvironmentError( f"Implementation Agent profile의 {key} 값이 올바른 문자열이 아닙니다." ) from exc model = settings.get("model") reasoning_effort = settings.get("model_reasoning_effort") if not isinstance(model, str) or not isinstance(reasoning_effort, str): raise CodexEnvironmentError( "Implementation Agent profile에 model과 model_reasoning_effort 문자열이 필요합니다." ) match = re.search( r'^developer_instructions\s*=\s*"""(.*?)"""\s*$', profile_text, re.MULTILINE | re.DOTALL, ) if match is None or not match.group(1).strip(): raise CodexEnvironmentError( "Implementation Agent profile에 developer_instructions가 필요합니다." ) instructions = match.group(1).strip() self._implementation_profile_cache = ( model, reasoning_effort, instructions, ) return self._implementation_profile_cache def _implementation_agent_settings(self) -> tuple[str, str]: model, reasoning_effort, _ = self._implementation_agent_profile() return model, reasoning_effort @staticmethod def _process_text(value: str | bytes | None) -> str: if value is None: return "" if isinstance(value, bytes): return value.decode("utf-8", errors="replace") return value def _invoke_codex(self, step: dict, preamble: str) -> dict: step_num, step_name = step["step"], step["name"] step_file = self._phase_dir / f"step{step_num}.md" if not step_file.exists(): print(f" ERROR: {step_file} not found") sys.exit(1) try: step_text = self._read_control_bytes(step_file).decode("utf-8") except UnicodeDecodeError as exc: raise HarnessError(f"Step instruction must be valid UTF-8: {step_file}") from exc prompt = preamble + step_text sandbox_mode = self._codex_sandbox_mode() model, reasoning_effort = self._implementation_agent_settings() command = [ "codex", "exec", "--disable", "unified_exec", "--json", "--model", model, "--config", f"model_reasoning_effort={json.dumps(reasoning_effort)}", "--sandbox", sandbox_mode, "--dangerously-bypass-hook-trust", "--cd", self._root, "-", ] transport_failure = None try: result = subprocess.run( command, cwd=self._root, input=prompt, capture_output=True, text=True, encoding="utf-8", timeout=self.CODEX_TIMEOUT_SECONDS, ) except FileNotFoundError as exc: raise CodexEnvironmentError( "Codex CLI를 찾을 수 없습니다. codex를 설치하고 PATH를 확인하세요." ) from exc except subprocess.TimeoutExpired as exc: transport_failure = "timeout" timeout_message = f"Codex timed out after {exc.timeout} seconds." stdout = self._process_text(exc.stdout) stderr = self._process_text(exc.stderr) if stderr: stderr = f"{timeout_message}\n{stderr}" else: stderr = timeout_message result = subprocess.CompletedProcess( command, 124, stdout=stdout, stderr=stderr ) if result.returncode != 0: print(f"\n WARN: Codex가 비정상 종료됨 (code {result.returncode})") if result.stderr: print(f" stderr: {result.stderr[:500]}") output = { "step": step_num, "name": step_name, "exitCode": result.returncode, "stdout": result.stdout, "stderr": result.stderr, "transportFailure": transport_failure, } out_path = self._phase_dir / f"step{step_num}-output.json" self._write_json(out_path, output) return output @staticmethod def _stdout_has_helper_failure(stdout: str) -> bool: """JSONL source/tool payloads are not runtime diagnostics.""" marker = "orchestrator_helper_launch_failed" for line in stdout.splitlines(): if marker not in line.lower(): continue try: event = json.loads(line) except json.JSONDecodeError: pass else: if not isinstance(event, dict): continue if event.get("type") in {"error", "turn.failed"}: error = event.get("message", "") if event["type"] == "error" else event.get("error", "") if marker in json.dumps(error).lower(): return True continue item = event.get("item") if ( not isinstance(item, dict) or item.get("type") != "command_execution" or item.get("exit_code") in (None, 0) ): continue line = item.get("aggregated_output", "") # Plain output must have a diagnostic prefix, including inside a failed command. if re.search( r"(?im)^\s*(?:(?:windows sandbox:\s*)?orchestrator_helper_launch_failed\b|" r"\S+\s+ERROR\b[^\r\n]*\borchestrator_helper_launch_failed\b)", line, ): return True return False @staticmethod def _codex_environment_failure(output: dict) -> Optional[str]: diagnostic = ( f"{output.get('stderr', '')}\n{output.get('stdout', '')}" ).lower() if ( "orchestrator_helper_launch_failed" in output.get("stderr", "").lower() or StepExecutor._stdout_has_helper_failure(output.get("stdout", "")) ): return "Codex Windows sandbox helper를 실행할 수 없습니다." if output.get("exitCode", 0) == 0: return None markers = { "not logged in": "Codex 인증이 필요합니다.", "authentication": "Codex 인증에 실패했습니다.", "unexpected argument": "현재 Codex CLI가 필요한 옵션을 지원하지 않습니다.", "unrecognized option": "현재 Codex CLI가 필요한 옵션을 지원하지 않습니다.", } for marker, message in markers.items(): if marker in diagnostic: return message return None # --- 헤더 & 검증 --- def _print_header(self): print(f"\n{'='*60}") print(f" Harness Step Executor") print(f" Phase: {self._phase_name} | Steps: {self._total}") if self._auto_push: print(f" Auto-push: enabled") print(f"{'='*60}") def _check_blockers(self): index = self._read_phase_index() current = next( (step for step in index["steps"] if step["status"] != "completed"), None, ) if current is None or current["status"] == "pending": return if current["status"] == "error": print(f"\n ✗ Step {current['step']} ({current['name']}) failed.") print(f" Error: {current['error_message']}") print(" Coordinator must classify the workflow correction before retry.") sys.exit(1) print(f"\n ⏸ Step {current['step']} ({current['name']}) blocked.") print(f" Reason: {current['blocked_reason']}") print(" Resolve the blocker and reset the Step to pending before retry.") sys.exit(2) @staticmethod def _clear_step_outcome(step: dict, keep: frozenset[str] = frozenset()) -> None: for key in ( "summary", "completed_at", "error_message", "failed_at", "blocked_reason", "blocked_at", ): if key not in keep: step.pop(key, None) def _ensure_created_at(self): index = self._read_phase_index() changed = False has_pending = any(step["status"] == "pending" for step in index["steps"]) if "created_at" not in index: index["created_at"] = self._stamp() changed = True if has_pending: if "completed_at" in index: index.pop("completed_at") changed = True for step in index["steps"]: if step["status"] != "pending": continue before = dict(step) self._clear_step_outcome(step) if step != before: changed = True if changed: self._write_json(self._index_file, index) if has_pending: top = self._read_top_index() selected = next( phase for phase in top["phases"] if phase["dir"] == self._phase_dir_name ) if selected["status"] != "pending" or any( key in selected for key in ("completed_at", "failed_at", "blocked_at") ): self._update_top_index("pending") # --- 실행 루프 --- def _classify_attempt( self, current: dict, output: dict, environment_failure: Optional[str] ) -> tuple[str, str, str]: """Return local classification, action and reason; never infer retry from exit alone.""" environment_failure = environment_failure or self._codex_environment_failure(output) if environment_failure: return "environment", "blocked", environment_failure status = current["status"] if status == "blocked": return "reported-blocked", "blocked", current["blocked_reason"] if status == "error": return "reported-error", "error", current["error_message"] exit_code = output["exitCode"] if status == "completed": if exit_code == 0: return "completed", "completed", current["summary"] return ( "inconsistent-result", "error", f"Step reported completed but Codex exited with code {exit_code}", ) diagnostic = (output.get("stderr", "").strip() or output.get("stdout", "").strip())[-4000:] if output.get("transportFailure") == "timeout": return "transport-timeout", "retry", diagnostic if exit_code != 0: return "process-error", "error", diagnostic or f"Codex exited with code {exit_code}" return "missing-result", "error", "Step did not update its workflow status" def _record_attempt(self, current: dict, output: dict, classification: str, reason: str) -> None: history = current.setdefault("attempt_history", []) field = {"completed": "summary", "error": "error_message", "blocked": "blocked_reason"}.get(current["status"]) history.append({ "attempt": len(history) + 1, "recorded_at": self._stamp(), "exit_code": output["exitCode"], "reported_status": current["status"], "reported_reason": current.get(field, "") if field else "", "classification": classification, "reason": reason, "diagnostic": (output.get("stderr", "").strip() or output.get("stdout", "").strip())[-4000:], }) def _stop_step(self, index: dict, step_num: int, status: str, reason: str) -> None: current = index["steps"][step_num] current["attempt_history"][-1]["reason"] = reason self._clear_step_outcome(current) current["status"] = status field, timestamp, exit_code = ( ("blocked_reason", "blocked_at", 2) if status == "blocked" else ("error_message", "failed_at", 1) ) current[field] = reason current[timestamp] = self._stamp() self._write_json(self._index_file, index) self._update_top_index(status) print(f" Step {step_num}: {status} — {reason}") sys.exit(exit_code) def _execute_single_step(self, step: dict, guardrails: str) -> bool: """Step을 실행한다. 자동 재시도는 Codex transport 실패에만 적용한다.""" step_num, step_name = step["step"], step["name"] owned_paths = self._step_owned_paths(self._read_phase_index()["steps"][step_num]) for path in owned_paths: self._validate_owned_file(path) done = sum( 1 for item in self._read_phase_index()["steps"] if item["status"] == "completed" ) prev_error = None for attempt in range(1, self.MAX_TRANSPORT_ATTEMPTS + 1): index = self._read_phase_index() step_context = self._build_step_context(index) preamble = self._build_preamble(guardrails, step_context, prev_error) control_snapshot = self._snapshot_phase_control_files(index) index_before_invocation = index git_identity = self._git_execution_identity() tag = f"Step {step_num}/{self._total - 1} ({done} done): {step_name}" if attempt > 1: tag += ( f" [transport retry {attempt}/" f"{self.MAX_TRANSPORT_ATTEMPTS}]" ) environment_failure = None with progress_indicator(tag) as pi: try: output = self._invoke_codex(step, preamble) except CodexEnvironmentError as exc: environment_failure = str(exc) output = {"exitCode": None, "stdout": "", "stderr": str(exc)} elapsed = int(pi.elapsed) try: if self._git_execution_identity() != git_identity: raise HarnessError( "Codex changed the Git branch or HEAD; Git history is " "Executor-owned" ) self._validate_phase_control_files(control_snapshot) index = self._read_phase_index() self._validate_step_index_mutation( index_before_invocation, index, step_num ) except HarnessError as exc: self._record_untrusted_step_result( index_before_invocation, step_num, "control violation", str(exc), control_snapshot, ) raise exit_code = output.get("exitCode") if type(exit_code) is not int and environment_failure is None: reason = "Codex output requires an integer exitCode" self._record_untrusted_step_result( index_before_invocation, step_num, "protocol violation", reason, control_snapshot, ) raise HarnessError(reason) current = index["steps"][step_num] ts = self._stamp() try: self._assert_step_scope(owned_paths) except StepOwnershipError as exc: self._record_attempt(current, output, "ownership-violation", str(exc)) self._stop_step(index, step_num, "error", str(exc)) classification, action, reason = self._classify_attempt( current, output, environment_failure ) self._record_attempt(current, output, classification, reason) if action == "retry": if attempt < self.MAX_TRANSPORT_ATTEMPTS: self._clear_step_outcome(current) self._write_json(self._index_file, index) prev_error = reason print( f" ↻ Step {step_num}: transport retry " f"{attempt}/{self.MAX_TRANSPORT_ATTEMPTS} — {reason}" ) continue reason = ( f"[transport failure after {self.MAX_TRANSPORT_ATTEMPTS} " f"attempts] {reason}" ) action = "error" if action == "completed": self._clear_step_outcome( current, frozenset({"summary", "completed_at"}) ) current["completed_at"] = ts self._write_json(self._index_file, index) try: self._commit_step(step_num, step_name, owned_paths) except HarnessError as exc: current["attempt_history"][-1]["classification"] = ( "ownership-violation" if isinstance(exc, StepOwnershipError) else "commit-error" ) self._stop_step(index, step_num, "error", str(exc)) print(f" ✓ Step {step_num}: {step_name} [{elapsed}s]") return True self._stop_step(index, step_num, action, reason) return False # unreachable def _execute_all_steps(self, guardrails: str): while True: index = self._read_phase_index() pending = next((s for s in index["steps"] if s["status"] == "pending"), None) if pending is None: print("\n All steps completed!") return step_num = pending["step"] for s in index["steps"]: if s["step"] == step_num and "started_at" not in s: s["started_at"] = self._stamp() self._write_json(self._index_file, index) break self._execute_single_step(pending, guardrails) def _finalize(self): index = self._read_phase_index() if any(step["status"] != "completed" for step in index["steps"]): raise HarnessError("phase cannot finalize before every Step is completed") changed = list(self._worktree_paths()) unexpected = [ path for path in changed if not self._is_phase_owned_path(path) ] if unexpected: raise HarnessError( "uncommitted implementation paths remain before finalization: " + ", ".join(unexpected[:5]) ) index["completed_at"] = self._stamp() self._write_json(self._index_file, index) self._update_top_index("completed") changed = list(self._worktree_paths()) metadata_paths = [ path for path in changed if not self._is_step_output_path(path) ] msg = f"chore({self._phase_name}): mark phase completed" if self._commit_paths(metadata_paths, msg): print(f" ✓ {msg}") if self._auto_push: branch = f"feat-{self._phase_name}" self._run_git_checked("push", "-u", "origin", branch) print(f" ✓ Pushed to origin/{branch}") print(f"\n{'='*60}") print(f" Phase '{self._phase_name}' completed!") print(f"{'='*60}") def main(): _configure_utf8_stdio() parser = argparse.ArgumentParser(description="Harness Step Executor") parser.add_argument("phase_dir", help="Phase directory name (e.g. 0-mvp)") parser.add_argument("--push", action="store_true", help="Push branch after completion") parser.add_argument( "--approve-danger-full-access", action="store_true", help=( "Explicitly authorize danger-full-access when the matching sandbox " "environment setting is used in a clean linked worktree" ), ) args = parser.parse_args() try: StepExecutor( args.phase_dir, auto_push=args.push, danger_full_access_approved=args.approve_danger_full_access, ).run() except HarnessError as exc: print(f"ERROR: {exc}") sys.exit(1) if __name__ == "__main__": main()