fix: support harness codex model override
This commit is contained in:
+16
-1
@@ -25,6 +25,12 @@ from typing import Optional
|
||||
ROOT = Path(__file__).resolve().parent.parent
|
||||
|
||||
|
||||
def configure_output_encoding():
|
||||
for stream in (sys.stdout, sys.stderr):
|
||||
if hasattr(stream, "reconfigure"):
|
||||
stream.reconfigure(encoding="utf-8", errors="replace")
|
||||
|
||||
|
||||
@contextlib.contextmanager
|
||||
def progress_indicator(label: str):
|
||||
"""터미널 진행 표시기. with 문으로 사용하며 .elapsed 로 경과 시간을 읽는다."""
|
||||
@@ -374,7 +380,7 @@ class StepExecutor:
|
||||
|
||||
prompt = preamble + step_file.read_text(encoding="utf-8")
|
||||
result = subprocess.run(
|
||||
[self._codex_command(), "exec", "--dangerously-bypass-approvals-and-sandbox", "--json", "-"],
|
||||
self._codex_exec_command(),
|
||||
cwd=self._root, capture_output=True, text=True, input=prompt,
|
||||
encoding="utf-8", errors="replace", timeout=1800,
|
||||
)
|
||||
@@ -406,6 +412,14 @@ class StepExecutor:
|
||||
or "codex"
|
||||
)
|
||||
|
||||
def _codex_exec_command(self) -> list[str]:
|
||||
cmd = [self._codex_command(), "exec"]
|
||||
model = os.environ.get("HARNESS_CODEX_MODEL", "").strip()
|
||||
if model:
|
||||
cmd.extend(["-m", model])
|
||||
cmd.extend(["--dangerously-bypass-approvals-and-sandbox", "--json", "-"])
|
||||
return cmd
|
||||
|
||||
def _print_header(self):
|
||||
print(f"\n{'='*60}")
|
||||
print(f" Harness Step Executor")
|
||||
@@ -561,6 +575,7 @@ class StepExecutor:
|
||||
|
||||
|
||||
def main():
|
||||
configure_output_encoding()
|
||||
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")
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
@@ -321,6 +322,47 @@ class ExecuteRunnerSafetyTests(unittest.TestCase):
|
||||
with patch.object(execute.shutil, "which", side_effect=fake_which):
|
||||
self.assertEqual(execute.StepExecutor._codex_command(), "C:/tools/codex.cmd")
|
||||
|
||||
def test_codex_exec_command_uses_model_env_override(self):
|
||||
execute = load_execute()
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
root = Path(tmp)
|
||||
write_phase(root)
|
||||
executor = make_executor(execute, root)
|
||||
|
||||
with patch.object(executor, "_codex_command", return_value="codex.cmd"):
|
||||
with patch.dict(os.environ, {"HARNESS_CODEX_MODEL": "gpt-5"}):
|
||||
self.assertEqual(
|
||||
executor._codex_exec_command(),
|
||||
[
|
||||
"codex.cmd",
|
||||
"exec",
|
||||
"-m",
|
||||
"gpt-5",
|
||||
"--dangerously-bypass-approvals-and-sandbox",
|
||||
"--json",
|
||||
"-",
|
||||
],
|
||||
)
|
||||
|
||||
def test_configure_output_encoding_sets_utf8_replace(self):
|
||||
execute = load_execute()
|
||||
|
||||
class Stream:
|
||||
def __init__(self):
|
||||
self.calls = []
|
||||
|
||||
def reconfigure(self, **kwargs):
|
||||
self.calls.append(kwargs)
|
||||
|
||||
stdout = Stream()
|
||||
stderr = Stream()
|
||||
with patch.object(execute.sys, "stdout", stdout):
|
||||
with patch.object(execute.sys, "stderr", stderr):
|
||||
execute.configure_output_encoding()
|
||||
|
||||
self.assertEqual(stdout.calls, [{"encoding": "utf-8", "errors": "replace"}])
|
||||
self.assertEqual(stderr.calls, [{"encoding": "utf-8", "errors": "replace"}])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user