diff --git a/docs/HARNESS.md b/docs/HARNESS.md index feff168..b94bfac 100644 --- a/docs/HARNESS.md +++ b/docs/HARNESS.md @@ -20,6 +20,20 @@ python scripts/execute.py python scripts/execute.py --push ``` +Codex의 `workspace-write` sandbox 밖에 설치된 실행 도구나 runtime이 필요한 경우에만 +`--codex-add-dir`를 반복 지정한다. 경로는 존재하는 절대 디렉터리여야 하며, Codex +CLI의 추가 writable directory와 child PATH 끝에 함께 전달된다. + +```powershell +python scripts/execute.py ` + --codex-add-dir "C:\path\to\tool-bin" ` + --codex-add-dir "C:\path\to\runtime" +``` + +이 옵션은 지정한 디렉터리에 child agent의 쓰기 권한도 부여하므로, 사용자 프로필이나 +드라이브 루트처럼 넓은 경로를 지정하지 말고 실행에 필요한 최소 설치 디렉터리만 +허용한다. + ## Harness Python 검증 이 저장소의 테스트와 최종 acceptance 검증은 pytest를 시스템 Python에 설치하지 않고 diff --git a/scripts/execute.py b/scripts/execute.py index 3ca984b..31d15a1 100644 --- a/scripts/execute.py +++ b/scripts/execute.py @@ -3,13 +3,14 @@ Harness Step Executor — phase 내 step을 순차 실행하고 자가 교정한다. Usage: - python scripts/execute.py [--push] + python scripts/execute.py [--push] [--codex-add-dir DIR] """ import argparse import contextlib import json import os +import shutil import subprocess import sys import threading @@ -26,6 +27,78 @@ class CodexEnvironmentError(RuntimeError): """재시도로 해결할 수 없는 Codex CLI 환경 오류.""" +def normalize_codex_add_dirs(raw_dirs) -> tuple[Path, ...]: + """검증된 Codex 추가 writable directory를 반환한다.""" + normalized = [] + for raw in raw_dirs: + candidate = Path(raw) + if not candidate.is_absolute(): + raise CodexEnvironmentError( + f"Codex 추가 경로는 절대 경로여야 합니다: {raw}" + ) + resolved = candidate.resolve() + if not resolved.is_dir(): + raise CodexEnvironmentError( + f"Codex 추가 경로가 존재하는 디렉터리가 아닙니다: {resolved}" + ) + normalized.append(resolved) + return tuple(normalized) + + +def build_codex_command( + codex_executable: Path, + root: Path, + add_dirs: tuple[Path, ...], +) -> list[str]: + """workspace-write Codex command를 안전한 argv로 구성한다.""" + command = [ + str(codex_executable.resolve()), + "exec", + "--json", + "--sandbox", + "workspace-write", + "--dangerously-bypass-hook-trust", + ] + for path in add_dirs: + command.extend(("--add-dir", str(path))) + command.extend(("--cd", str(root.resolve()), "-")) + return command + + +def build_codex_environment( + base_env, + add_dirs: tuple[Path, ...], +) -> dict[str, str]: + """명시적 tool directory를 기존 PATH 뒤에 추가한다.""" + environment = dict(base_env) + path_entries = [environment.get("PATH", "")] + path_entries.extend(str(path) for path in add_dirs) + environment["PATH"] = os.pathsep.join(entry for entry in path_entries if entry) + return environment + + +def run_utf8_process(command, *, cwd, prompt, env, timeout): + """UTF-8 prompt/output 계약으로 subprocess를 실행한다.""" + return subprocess.run( + list(command), + cwd=cwd, + input=prompt, + capture_output=True, + encoding="utf-8", + errors="replace", + env=env, + timeout=timeout, + ) + + +def configure_standard_streams(*streams) -> None: + """Windows legacy console에서 status 출력이 중단되지 않게 한다.""" + for stream in streams: + reconfigure = getattr(stream, "reconfigure", None) + if reconfigure is not None: + reconfigure(errors="replace") + + @contextlib.contextmanager def progress_indicator(label: str): """터미널 진행 표시기. with 문으로 사용하며 .elapsed 로 경과 시간을 읽는다.""" @@ -62,13 +135,20 @@ class StepExecutor: CHORE_MSG = "chore({phase}): step {num} output" TZ = timezone(timedelta(hours=9)) - def __init__(self, phase_dir_name: str, *, auto_push: bool = False): + def __init__( + self, + phase_dir_name: str, + *, + auto_push: bool = False, + codex_add_dirs=(), + ): self._root = str(ROOT) self._phases_dir = ROOT / "phases" self._phase_dir = self._phases_dir / phase_dir_name self._phase_dir_name = phase_dir_name self._top_index_file = self._phases_dir / "index.json" self._auto_push = auto_push + self._codex_add_dirs = normalize_codex_add_dirs(codex_add_dirs) if not self._phase_dir.is_dir(): print(f"ERROR: {self._phase_dir} not found") @@ -241,24 +321,26 @@ class StepExecutor: sys.exit(1) prompt = preamble + step_file.read_text(encoding="utf-8") - command = [ - "codex", - "exec", - "--json", - "--sandbox", - "workspace-write", - "--dangerously-bypass-hook-trust", - "--cd", - self._root, - "-", - ] + codex_executable = shutil.which("codex") + if codex_executable is None: + raise CodexEnvironmentError( + "Codex CLI를 찾을 수 없습니다. codex를 설치하고 PATH를 확인하세요." + ) + command = build_codex_command( + Path(codex_executable), + Path(self._root), + self._codex_add_dirs, + ) + environment = build_codex_environment( + os.environ, + self._codex_add_dirs, + ) try: - result = subprocess.run( + result = run_utf8_process( command, cwd=self._root, - input=prompt, - capture_output=True, - text=True, + prompt=prompt, + env=environment, timeout=1800, ) except FileNotFoundError as exc: @@ -465,13 +547,28 @@ class StepExecutor: def main(): + configure_standard_streams(sys.stdout, sys.stderr) 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( + "--codex-add-dir", + action="append", + default=[], + metavar="DIR", + help=( + "Codex workspace-write sandbox에 추가할 절대 tool/runtime directory. " + "여러 번 지정할 수 있습니다." + ), + ) args = parser.parse_args() try: - StepExecutor(args.phase_dir, auto_push=args.push).run() + StepExecutor( + args.phase_dir, + auto_push=args.push, + codex_add_dirs=args.codex_add_dir, + ).run() except CodexEnvironmentError as exc: print(f"ERROR: {exc}") sys.exit(1) diff --git a/tests/harness/test_execute.py b/tests/harness/test_execute.py new file mode 100644 index 0000000..ed4d775 --- /dev/null +++ b/tests/harness/test_execute.py @@ -0,0 +1,122 @@ +import os +import sys +from pathlib import Path + +import pytest + +from scripts.execute import ( + CodexEnvironmentError, + build_codex_command, + build_codex_environment, + configure_standard_streams, + normalize_codex_add_dirs, + run_utf8_process, +) + + +def test_normalize_codex_add_dirs_accepts_existing_absolute_directories(tmp_path): + first = tmp_path / "tool" + second = tmp_path / "runtime" + first.mkdir() + second.mkdir() + + result = normalize_codex_add_dirs([str(first), str(second)]) + + assert result == (first.resolve(), second.resolve()) + + +@pytest.mark.parametrize("raw", ["relative/tool", "missing"]) +def test_normalize_codex_add_dirs_rejects_unsafe_paths(tmp_path, raw): + candidate = raw if raw.startswith("relative") else str(tmp_path / raw) + + with pytest.raises(CodexEnvironmentError): + normalize_codex_add_dirs([candidate]) + + +def test_build_codex_command_grants_only_explicit_directories(tmp_path): + root = tmp_path / "repo" + tool = tmp_path / "tool" + runtime = tmp_path / "runtime" + root.mkdir() + tool.mkdir() + runtime.mkdir() + codex = tmp_path / "codex.exe" + + command = build_codex_command( + codex, + root, + (tool.resolve(), runtime.resolve()), + ) + + assert command == [ + str(codex), + "exec", + "--json", + "--sandbox", + "workspace-write", + "--dangerously-bypass-hook-trust", + "--add-dir", + str(tool.resolve()), + "--add-dir", + str(runtime.resolve()), + "--cd", + str(root.resolve()), + "-", + ] + + +def test_build_codex_environment_appends_tool_directories(tmp_path): + tool = (tmp_path / "tool").resolve() + runtime = (tmp_path / "runtime").resolve() + base = {"PATH": os.pathsep.join(("existing-one", "existing-two")), "KEEP": "value"} + + result = build_codex_environment(base, (tool, runtime)) + + assert result["PATH"].split(os.pathsep) == [ + "existing-one", + "existing-two", + str(tool), + str(runtime), + ] + assert result["KEEP"] == "value" + assert result is not base + + +def test_run_utf8_process_sends_prompt_as_utf8(tmp_path): + verifier = tmp_path / "verify_utf8.py" + verifier.write_text( + "import sys\n" + "payload = sys.stdin.buffer.read()\n" + "text = payload.decode('utf-8')\n" + "sys.stdout.buffer.write(text.encode('utf-8'))\n", + encoding="utf-8", + ) + prompt = "FESA 한글 prompt" + + result = run_utf8_process( + [sys.executable, str(verifier)], + cwd=tmp_path, + prompt=prompt, + env=os.environ.copy(), + timeout=10, + ) + + assert result.returncode == 0 + assert result.stdout == prompt + + +def test_configure_standard_streams_replaces_unencodable_status_characters(): + class RecordingStream: + def __init__(self): + self.calls = [] + + def reconfigure(self, **kwargs): + self.calls.append(kwargs) + + stdout = RecordingStream() + stderr = RecordingStream() + + configure_standard_streams(stdout, stderr) + + assert stdout.calls == [{"errors": "replace"}] + assert stderr.calls == [{"errors": "replace"}]