FESADev/tests/test_execute.py

import json
import os
import subprocess
from types import SimpleNamespace

import pytest

from scripts import execute


@pytest.fixture(autouse=True)
def clear_sandbox_override(monkeypatch):
    monkeypatch.delenv("FEM_HARNESS_CODEX_SANDBOX", raising=False)


def write_top_index(phase_dir):
    (phase_dir.parent / "index.json").write_text(
        json.dumps({"phases": [{"dir": phase_dir.name, "status": "pending"}]}),
        encoding="utf-8",
    )


def write_implementation_agent_profile(
    root, *, model="gpt-5.6-terra", reasoning="xhigh"
):
    profile = root / ".codex" / "agents" / "implementation-agent.toml"
    profile.parent.mkdir(parents=True)
    profile.write_text(
        f'model = "{model}"\nmodel_reasoning_effort = "{reasoning}"\n'
        'developer_instructions = """Implementation Agent test role."""\n',
        encoding="utf-8",
    )


def test_invoke_codex_uses_utf8_for_unicode_prompt(tmp_path, monkeypatch):
    phase_dir = tmp_path / "phases" / "unicode-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "FESA Structural Solver",
                "phase": "unicode-phase",
                "steps": [{"step": 0, "name": "unicode", "status": "pending"}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("Euler–Bernoulli 보", encoding="utf-8")
    write_implementation_agent_profile(tmp_path)

    invocation = {}

    def capture_run(command, **kwargs):
        invocation["command"] = command
        invocation["kwargs"] = kwargs
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setattr(execute.subprocess, "run", capture_run)

    executor = execute.StepExecutor("unicode-phase")
    executor._invoke_codex({"step": 0, "name": "unicode"}, "승인된 설계 – ")

    assert "Euler–Bernoulli 보" in invocation["kwargs"]["input"]
    assert invocation["kwargs"]["encoding"] == "utf-8"
    sandbox_index = invocation["command"].index("--sandbox")
    assert invocation["command"][sandbox_index + 1] == "workspace-write"


def test_invoke_codex_uses_implementation_agent_command_contract(
    tmp_path, monkeypatch
):
    phase_dir = tmp_path / "phases" / "agent-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "FESA Structural Solver",
                "phase": "agent-phase",
                "steps": [{"step": 0, "name": "agent", "status": "pending"}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("agent routing", encoding="utf-8")
    write_implementation_agent_profile(tmp_path)

    invocation = {}

    def capture_run(command, **kwargs):
        invocation["command"] = command
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setattr(execute.subprocess, "run", capture_run)

    executor = execute.StepExecutor("agent-phase")
    executor._invoke_codex({"step": 0, "name": "agent"}, "")

    model_index = invocation["command"].index("--model")
    config_index = invocation["command"].index("--config")
    assert invocation["command"][model_index + 1] == "gpt-5.6-terra"
    assert invocation["command"][config_index + 1] == 'model_reasoning_effort="xhigh"'
    adjacent_arguments = list(zip(invocation["command"], invocation["command"][1:]))
    assert ("--disable", "unified_exec") in adjacent_arguments


def test_invoke_codex_accepts_explicit_danger_full_access_override(
    tmp_path, monkeypatch
):
    phase_dir = tmp_path / "phases" / "sandbox-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "FESA Structural Solver",
                "phase": "sandbox-phase",
                "steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")
    write_implementation_agent_profile(tmp_path)

    invocation = {}

    def capture_run(command, **kwargs):
        if command[0] == "git":
            responses = {
                ("rev-parse", "--absolute-git-dir"): str(tmp_path / ".git/worktrees/sandbox"),
                ("rev-parse", "--git-common-dir"): str(tmp_path / ".git"),
                ("status", "--porcelain=v1", "-z", "--untracked-files=all"): "",
            }
            return subprocess.CompletedProcess(
                command, 0, stdout=responses[tuple(command[1:])], stderr=""
            )
        invocation["command"] = command
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setattr(execute.subprocess, "run", capture_run)
    monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "danger-full-access")

    executor = execute.StepExecutor("sandbox-phase", danger_full_access_approved=True)
    executor._invoke_codex({"step": 0, "name": "sandbox"}, "")

    sandbox_index = invocation["command"].index("--sandbox")
    assert invocation["command"][sandbox_index + 1] == "danger-full-access"


def test_invoke_codex_timeout_writes_structured_failure(tmp_path, monkeypatch):
    phase_dir = tmp_path / "phases" / "timeout-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "FESA Structural Solver",
                "phase": "timeout-phase",
                "steps": [{"step": 0, "name": "timeout", "status": "pending"}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("timeout", encoding="utf-8")
    write_implementation_agent_profile(tmp_path)

    def timeout_run(command, **kwargs):
        raise subprocess.TimeoutExpired(
            command, kwargs["timeout"], output="partial stdout", stderr="partial stderr"
        )

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setattr(execute.subprocess, "run", timeout_run)

    executor = execute.StepExecutor("timeout-phase")
    output = executor._invoke_codex({"step": 0, "name": "timeout"}, "")

    assert output["exitCode"] == 124
    assert output["transportFailure"] == "timeout"
    assert output["stdout"] == "partial stdout"
    assert "Codex timed out after 14400 seconds." in output["stderr"]
    stored = json.loads((phase_dir / "step0-output.json").read_text(encoding="utf-8"))
    assert stored == output


def test_execute_single_step_retries_structured_codex_timeout(tmp_path, monkeypatch):
    phase_dir = tmp_path / "phases" / "retry-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "FESA Structural Solver",
                "phase": "retry-phase",
                "steps": [{"step": 0, "name": "retry", "status": "pending", "owned_paths": []}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("retry", encoding="utf-8")
    write_implementation_agent_profile(tmp_path)
    attempts = []

    def timeout_run(command, **kwargs):
        if command[0] == "git":
            responses = {
                ("symbolic-ref", "--quiet", "HEAD"): "refs/heads/feat-retry-phase",
                ("rev-parse", "--verify", "HEAD"): "a" * 40,
                ("status", "--porcelain=v1", "-z", "--untracked-files=all"): "",
            }
            return subprocess.CompletedProcess(
                command, 0, stdout=responses[tuple(command[1:])], stderr=""
            )
        attempts.append(command)
        raise subprocess.TimeoutExpired(command, kwargs["timeout"])

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setattr(execute.subprocess, "run", timeout_run)

    executor = execute.StepExecutor("retry-phase")

    with pytest.raises(SystemExit, match="1"):
        executor._execute_single_step(
            {"step": 0, "name": "retry", "status": "pending"}, ""
        )

    assert len(attempts) == 2
    index = json.loads((phase_dir / "index.json").read_text(encoding="utf-8"))
    assert index["steps"][0]["status"] == "error"
    assert "Codex timed out after 14400 seconds." in index["steps"][0]["error_message"]
    assert index["steps"][0]["attempt_history"][-1]["reason"] == index["steps"][0]["error_message"]
    top = json.loads((phase_dir.parent / "index.json").read_text(encoding="utf-8"))
    assert top["phases"][0]["status"] == "error"


def test_invoke_codex_rejects_unknown_sandbox_override(tmp_path, monkeypatch):
    phase_dir = tmp_path / "phases" / "sandbox-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "index.json").write_text(
        json.dumps(
            {
                "project": "FESA Structural Solver",
                "phase": "sandbox-phase",
                "steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
            }
        ),
        encoding="utf-8",
    )
    (phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "read-only")

    executor = execute.StepExecutor("sandbox-phase")
    with pytest.raises(execute.CodexEnvironmentError, match="FEM_HARNESS_CODEX_SANDBOX"):
        executor._invoke_codex({"step": 0, "name": "sandbox"}, "")


@pytest.mark.parametrize(
    ("approved", "isolated", "porcelain", "expected_error"),
    [
        (False, True, "", "--approve-danger-full-access"),
        (True, False, "", "isolated linked worktree"),
        (True, True, " M unrelated.md\0", "clean worktree"),
    ],
)
def test_sandbox_elevation_rejects_missing_safety_conditions(
    tmp_path, monkeypatch, approved, isolated, porcelain, expected_error
):
    phase_dir = tmp_path / "phases" / "sandbox-phase"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    (phase_dir / "step0.md").write_text("sandbox guard", encoding="utf-8")
    (phase_dir / "index.json").write_text(
        json.dumps({
            "project": "FESA Structural Solver",
            "phase": "sandbox-phase",
            "steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
        }),
        encoding="utf-8",
    )

    def git_probe(command, **kwargs):
        assert command[0] == "git", "Rejected elevation must not launch Codex"
        common_dir = tmp_path / ".git"
        git_dir = common_dir / "worktrees/sandbox" if isolated else common_dir
        responses = {
            ("rev-parse", "--absolute-git-dir"): str(git_dir),
            ("rev-parse", "--git-common-dir"): str(common_dir),
            ("status", "--porcelain=v1", "-z", "--untracked-files=all"): porcelain,
        }
        return subprocess.CompletedProcess(
            command, 0, stdout=responses[tuple(command[1:])], stderr=""
        )

    monkeypatch.setattr(execute, "ROOT", tmp_path)
    monkeypatch.setattr(execute.subprocess, "run", git_probe)
    monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "danger-full-access")
    executor = execute.StepExecutor("sandbox-phase", danger_full_access_approved=approved)

    with pytest.raises(execute.CodexEnvironmentError, match=expected_error):
        executor._codex_sandbox_mode()


def test_codex_environment_failure_detects_missing_sandbox_helper_on_zero_exit():
    output = {
        "exitCode": 0,
        "stdout": "",
        "stderr": (
            "windows sandbox: orchestrator_helper_launch_failed: "
            "helper=codex-windows-sandbox-setup.exe, error=program not found"
        ),
    }

    assert execute.StepExecutor._codex_environment_failure(output) == (
        "Codex Windows sandbox helper를 실행할 수 없습니다."
    )


def test_run_git_decodes_output_as_utf8(tmp_path, monkeypatch):
    invocation = {}

    def capture_run(command, **kwargs):
        invocation["command"] = command
        invocation["kwargs"] = kwargs
        return subprocess.CompletedProcess(command, 0, stdout="", stderr="")

    monkeypatch.setattr(execute.subprocess, "run", capture_run)
    executor = object.__new__(execute.StepExecutor)
    executor._root = str(tmp_path)

    executor._run_git("status", "--short")

    assert invocation["kwargs"]["encoding"] == "utf-8"


def test_feature_commit_message_is_windows_console_safe():
    message = execute.StepExecutor.FEAT_MSG.format(
        phase="linear-static-3d-euler-beam", num=0, name="requirements-baseline"
    )

    message.encode("cp949")


def test_configure_utf8_stdio_reconfigures_both_streams(monkeypatch):
    class ReconfigurableStream:
        def __init__(self):
            self.encodings = []

        def reconfigure(self, *, encoding):
            self.encodings.append(encoding)

    stdout = ReconfigurableStream()
    stderr = ReconfigurableStream()
    monkeypatch.setattr(execute.sys, "stdout", stdout)
    monkeypatch.setattr(execute.sys, "stderr", stderr)

    execute._configure_utf8_stdio()

    assert stdout.encodings == ["utf-8"]
    assert stderr.encodings == ["utf-8"]


@pytest.fixture
def step_runtime(tmp_path, monkeypatch):
    """Run the real Step loop and Git against a disposable repo; replace only Codex."""
    monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull)
    monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
    monkeypatch.delenv("FEM_HARNESS_CODEX_SANDBOX", raising=False)
    real_run = subprocess.run

    def git(*args):
        return real_run(
            ["git", *args], cwd=tmp_path, capture_output=True,
            text=True, encoding="utf-8", check=True,
        ).stdout

    git("init", "--quiet")
    git("config", "user.name", "Harness Tests")
    git("config", "user.email", "harness@example.invalid")
    git("config", "commit.gpgsign", "false")
    empty_hooks = tmp_path / ".git" / "empty-hooks"
    empty_hooks.mkdir()
    git("config", "core.hooksPath", str(empty_hooks))
    phase_dir = tmp_path / "phases" / "runtime"
    phase_dir.mkdir(parents=True)
    write_top_index(phase_dir)
    index_file = phase_dir / "index.json"
    index_file.write_text(json.dumps({
        "project": "Harness Tests", "phase": "runtime",
        "steps": [{"step": 0, "name": "runtime", "status": "pending", "owned_paths": []}],
    }), encoding="utf-8")
    (phase_dir / "step0.md").write_text("Run the assigned test Step.", encoding="utf-8")
    (tmp_path / ".gitignore").write_text("phases/**/step*-output.json\n", encoding="utf-8")
    write_implementation_agent_profile(tmp_path)
    git("add", "--", ".")
    git("commit", "--quiet", "-m", "test: initialize disposable repo")
    monkeypatch.setattr(execute, "ROOT", tmp_path)
    runtime = SimpleNamespace(
        executor=execute.StepExecutor("runtime"), root=tmp_path,
        index_file=index_file, git=git, real_run=real_run, responses=[], calls=[],
    )

    def invoke(command, **kwargs):
        if command[0] == "git":
            return real_run(command, **kwargs)
        assert command[0] == "codex", "Only the simulated Codex process is allowed"
        runtime.calls.append(kwargs["input"])
        response = runtime.responses.pop(0)
        if callable(response):
            response = response(runtime)
        if isinstance(response, Exception):
            raise response
        index = json.loads(index_file.read_text(encoding="utf-8"))
        current = index["steps"][0]
        current["status"] = response.get("status", "pending")
        if current["status"] != "pending":
            field = {"completed": "summary", "error": "error_message", "blocked": "blocked_reason"}[current["status"]]
            current[field] = response.get("reason", "reported reason")
        index_file.write_text(json.dumps(index), encoding="utf-8")
        return subprocess.CompletedProcess(
            command, response.get("exit", 0),
            stdout=response.get("stdout", ""), stderr=response.get("stderr", ""),
        )

    monkeypatch.setattr(execute.subprocess, "run", invoke)
    return runtime


def run_step(runtime):
    index = runtime.executor._read_phase_index()
    return runtime.executor._execute_single_step(index["steps"][0], "")


@pytest.mark.parametrize("status,expected_exit", [("error", 1), ("blocked", 2)])
@pytest.mark.parametrize("process_exit", [0, 1, 124])
def test_explicit_failure_preserves_reason_without_retry(step_runtime, status, expected_exit, process_exit):
    runtime = step_runtime
    runtime.responses = [
        {"status": status, "reason": "original failure", "exit": process_exit, "stderr": "process diagnostic"},
        {"status": "completed"},
    ]
    with pytest.raises(SystemExit) as stopped:
        run_step(runtime)
    assert stopped.value.code == expected_exit
    assert len(runtime.calls) == 1
    current = runtime.executor._read_phase_index()["steps"][0]
    assert current["status"] == status
    assert current["error_message" if status == "error" else "blocked_reason"] == "original failure"
    assert current["attempt_history"][0]["reported_reason"] == "original failure"
    assert current["attempt_history"][0]["reason"] == "original failure"
    assert current["attempt_history"][0]["exit_code"] == process_exit
    assert current["attempt_history"][0]["diagnostic"] == "process diagnostic"
    assert runtime.executor._read_top_index()["phases"][0]["status"] == status
    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"


@pytest.mark.parametrize("process_exit", [0, 1])
@pytest.mark.parametrize("status", ["pending", "completed", "error"])
def test_environment_failure_blocks_independent_of_exit_and_report(step_runtime, process_exit, status):
    runtime = step_runtime
    diagnostic = "windows sandbox: orchestrator_helper_launch_failed: helper not found"
    runtime.responses = [{"status": status, "exit": process_exit, "stderr": diagnostic}]
    with pytest.raises(SystemExit) as stopped:
        run_step(runtime)
    assert stopped.value.code == 2
    assert len(runtime.calls) == 1
    current = runtime.executor._read_phase_index()["steps"][0]
    assert current["status"] == "blocked"
    assert "sandbox helper" in current["blocked_reason"]
    record = current["attempt_history"][0]
    assert record["reported_status"] == status
    assert record["classification"] == "environment"
    assert record["diagnostic"] == diagnostic
    assert runtime.executor._read_top_index()["phases"][0]["status"] == "blocked"
    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"


def test_missing_cli_is_persisted_as_environment_block(step_runtime):
    runtime = step_runtime
    runtime.responses = [FileNotFoundError("codex")]
    with pytest.raises(SystemExit) as stopped:
        run_step(runtime)
    assert stopped.value.code == 2
    current = runtime.executor._read_phase_index()["steps"][0]
    assert current["status"] == "blocked"
    assert current["attempt_history"][0]["exit_code"] is None
    assert runtime.executor._read_top_index()["phases"][0]["status"] == "blocked"
    assert len(runtime.calls) == 1


@pytest.mark.parametrize("status,process_exit", [("pending", 0), ("pending", 1), ("pending", 124), ("completed", 1)])
def test_unclassified_or_inconsistent_result_never_retries(step_runtime, status, process_exit):
    runtime = step_runtime
    runtime.responses = [{"status": status, "exit": process_exit}, {"status": "completed"}]
    with pytest.raises(SystemExit) as stopped:
        run_step(runtime)
    assert stopped.value.code == 1
    assert len(runtime.calls) == 1
    assert runtime.executor._read_phase_index()["steps"][0]["status"] == "error"
    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"


def test_timeout_retry_preserves_history_and_can_complete(step_runtime):
    runtime = step_runtime
    runtime.responses = [subprocess.TimeoutExpired("codex", 1800), {"status": "completed"}]
    assert run_step(runtime)
    current = runtime.executor._read_phase_index()["steps"][0]
    assert current["status"] == "completed"
    assert [record["classification"] for record in current["attempt_history"]] == ["transport-timeout", "completed"]
    assert [record["attempt"] for record in current["attempt_history"]] == [1, 2]
    assert "timed out" in runtime.calls[1]
    assert runtime.git("status", "--porcelain").strip() == ""


def test_history_survives_assigned_pending_reset(step_runtime):
    runtime = step_runtime
    runtime.responses = [{"status": "error", "reason": "original failure"}]
    with pytest.raises(SystemExit):
        run_step(runtime)
    index = runtime.executor._read_phase_index()
    history = index["steps"][0]["attempt_history"]
    index["steps"][0]["status"] = "pending"
    runtime.executor._write_json(runtime.index_file, index)
    runtime.executor._ensure_created_at()
    current = runtime.executor._read_phase_index()["steps"][0]
    assert "error_message" not in current
    assert current["attempt_history"] == history
    runtime.responses = [{"status": "completed"}]
    assert run_step(runtime)
    assert [record["attempt"] for record in runtime.executor._read_phase_index()["steps"][0]["attempt_history"]] == [1, 2]


def test_full_runner_completes_metadata_only_phase(step_runtime):
    runtime = step_runtime
    runtime.responses = [{"status": "completed"}]
    runtime.executor.run()
    assert len(runtime.calls) == 1
    assert runtime.executor._read_top_index()["phases"][0]["status"] == "completed"
    assert "completed_at" in runtime.executor._read_phase_index()
    assert runtime.git("symbolic-ref", "HEAD").strip() == "refs/heads/feat-runtime"
    assert runtime.git("status", "--porcelain").strip() == ""


@pytest.mark.parametrize("status", ["error", "blocked", "completed"])
def test_terminal_report_prevents_retry_even_when_wrapper_times_out(step_runtime, status):
    runtime = step_runtime

    def report_then_timeout(runtime):
        index = runtime.executor._read_phase_index()
        current = index["steps"][0]
        current["status"] = status
        field = {"error": "error_message", "blocked": "blocked_reason", "completed": "summary"}[status]
        current[field] = "original terminal report"
        runtime.executor._write_json(runtime.index_file, index)
        return subprocess.TimeoutExpired("codex", 1800)

    runtime.responses = [report_then_timeout]
    with pytest.raises(SystemExit) as stopped:
        run_step(runtime)
    assert stopped.value.code == (2 if status == "blocked" else 1)
    assert len(runtime.calls) == 1
    assert runtime.executor._read_phase_index()["steps"][0]["attempt_history"][0]["reported_reason"] == "original terminal report"