modify documents

tests/test_agent_skill_workflow_contract.py

@@ -3,6 +3,10 @@ from pathlib import Path
 import re
 import tomllib
 
+import pytest
+
+from test_execute import step_runtime
+
 
 ROOT = Path(__file__).resolve().parents[1]
 AGENT_DIR = ROOT / ".codex" / "agents"
@@ -24,14 +28,14 @@ EXPECTED_AGENTS = {
 
 EXPECTED_SKILLS = {
     "fem-theory-query",
-    "fesa-cpp-msvc-tdd",
-    "fesa-formulation-spec",
-    "fesa-io-contract",
-    "fesa-numerical-review",
-    "fesa-physics-sanity",
-    "fesa-release-readiness",
-    "fesa-requirements-baseline",
-    "fesa-research-evidence",
+    "fem-implementation-tdd",
+    "fem-formulation-spec",
+    "fem-io-contract",
+    "fem-numerical-review",
+    "fem-physics-review",
+    "fem-release-readiness",
+    "fem-requirements-baseline",
+    "fem-research-evidence",
 }
 
 LEGACY_OUTPUT_DIRS = {
@@ -94,22 +98,27 @@ AGENT_STAGE_GUIDANCE_FILES = (
 )
 
 EXPECTED_AGENT_STAGE_MAPPING = (
-    ("1. 요구조건", "requirement-agent", "requirements.md"),
-    ("2. 연구", "research-agent", "research.md"),
-    ("3. 정식화", "formulation-agent", "formulation.md"),
+    ("requirements", "requirement-agent", ("requirements.md",)),
+    ("research", "research-agent", ("research.md",)),
+    ("formulation", "formulation-agent", ("formulation.md",)),
     (
-        "4. 수치 검토 + reference model 계약",
+        "numerical-review",
         "numerical-review-agent",
-        "numerical-review.md, reference-model.md",
+        ("numerical-review.md", "reference-model.md"),
     ),
-    ("5. I/O 정의", "io-definition-agent", "io.md"),
+    ("io-definition", "io-definition-agent", ("io.md",)),
     (
-        "6. 구현 계획 + C++ 구현 + build/test + reference comparison",
-        "implementation-planning-agent, implementation-agent",
-        "implementation-plan.md, implementation-report.md, build-test.md, reference-comparison.md",
+        "implementation-planning",
+        "implementation-planning-agent",
+        ("implementation-plan.md",),
     ),
-    ("7. 물리 검토", "physics-evaluation-agent", "physics-evaluation.md"),
-    ("8. 배포 준비", "release-agent", "release.md"),
+    (
+        "implementation",
+        "implementation-agent",
+        ("implementation-report.md", "build-test.md", "reference-comparison.md"),
+    ),
+    ("physics-evaluation", "physics-evaluation-agent", ("physics-evaluation.md",)),
+    ("release-readiness", "release-agent", ("release.md",)),
 )
 
 CANONICAL_FEATURE_DIRS = (
@@ -173,12 +182,24 @@ def live_contract_text() -> str:
     return "\n".join(read(path) for path in paths)
 
 
-def markdown_stage_mapping(path: Path) -> tuple[tuple[str, str, str], ...]:
+def markdown_stage_mapping(path: Path) -> tuple[tuple[str, str, tuple[str, ...]], ...]:
+    heading = (
+        "## Agent/Skill Workflow"
+        if path.name == "AGENTS.md"
+        else "## 4. 단계와 소유권"
+    )
+    section = read(path).split(heading, maxsplit=1)[1].split("\n## ", maxsplit=1)[0]
     rows = []
-    for line in read(path).splitlines():
+    for line in section.splitlines():
+        if not line.lstrip().startswith("|"):
+            continue
         columns = [column.strip().replace("`", "") for column in line.strip().split("|")[1:-1]]
-        if len(columns) >= 4 and re.fullmatch(r"[1-8]\. .+", columns[0]):
-            rows.append((columns[0], columns[1], columns[3]))
+        if len(columns) < 3:
+            continue
+        stage = re.sub(r"^\d+\.\s+", "", columns[0])
+        if re.fullmatch(r"[a-z]+(?:-[a-z]+)*", stage):
+            reports = tuple(re.findall(r"\b[\w-]+\.md\b", columns[-1]))
+            rows.append((stage, columns[1], reports))
     return tuple(rows)
 
 
@@ -254,84 +275,68 @@ def test_existing_feature_artifacts_are_bundled():
     for feature_id, expected_files in FEATURE_FILES.items():
         feature_dir = docs_dir / feature_id
         actual_files = {path.name for path in feature_dir.glob("*.md")}
-        assert actual_files == expected_files
+        # Preserve historical artifacts while allowing Feature management and linked details.
+        assert expected_files <= actual_files
 
 
 def test_numerical_review_absorbs_reference_model_contract():
     assert not (AGENT_DIR / "reference-model-agent.toml").exists()
     assert not (SKILL_DIR / "fesa-reference-models").exists()
-    text = read(SKILL_DIR / "fesa-numerical-review" / "SKILL.md").lower()
-    for marker in (
-        "numerical-review.md",
-        "reference-model.md",
-        "reference case inventory",
-        "source identity",
-        "row precheck",
-        "tolerance",
-        "pass-for-io-definition",
-    ):
-        assert marker in text
+    text = read(SKILL_DIR / "fem-numerical-review" / "SKILL.md")
+    outputs = set(re.findall(r"docs/<feature-id>/([\w-]+\.md)", text))
+    assert {"numerical-review.md", "reference-model.md"} <= outputs
 
 
 def test_implementation_absorbs_build_test_and_reference_comparison():
     assert not (AGENT_DIR / "build-test-executor-agent.toml").exists()
     assert not (AGENT_DIR / "reference-verification-agent.toml").exists()
     assert not (SKILL_DIR / "fesa-reference-comparison").exists()
-    text = read(SKILL_DIR / "fesa-cpp-msvc-tdd" / "SKILL.md").lower()
-    for marker in (
+    text = read(SKILL_DIR / "fem-implementation-tdd" / "SKILL.md")
+    outputs = set(re.findall(r"docs/<feature-id>/([\w-]+\.md)", text))
+    assert {
+        "implementation-plan.md",
         "implementation-report.md",
         "build-test.md",
         "reference-comparison.md",
-        "artifact check -> compare -> classify -> report",
-        "pass-for-physics-evaluation",
-    ):
-        assert marker in text
+        "corrections.md",
+    } <= outputs
 
 
-def test_implementation_steps_keep_final_comparison_in_a_final_gate():
-    paths = (
-        AGENT_DIR / "implementation-agent.toml",
-        SKILL_DIR / "fesa-cpp-msvc-tdd" / "SKILL.md",
-    )
-    for path in paths:
-        text = read(path).lower()
-        assert "final implementation-owned verification step/gate" in text, path
-        assert "non-final step" in text, path
-        assert "artifact check -> compare -> classify -> report" in text, path
-        assert "results.h5" in text, path
-        assert "reference-comparison.md" in text, path
+def test_implementation_verification_reference_is_resolvable():
+    skill = SKILL_DIR / "fem-implementation-tdd" / "SKILL.md"
+    links = re.findall(r"\[[^\]]+\]\(([^)]+)\)", read(skill))
+    assert "references/toolchain-and-verification.md" in links
+    for link in links:
+        assert (skill.parent / link).is_file(), link
 
 
-def test_p_agent_001_implementation_agent_reads_coding_style_before_cpp_steps():
+def test_p_agent_001_implementation_guidance_resolves_coding_style_contract():
     parsed = tomllib.loads(read(AGENT_DIR / "implementation-agent.toml"))
     instructions = parsed["developer_instructions"]
-    section_match = re.search(
-        r"Mandatory global input:\n(?P<section>.*?)(?=\n[A-Z][^\n]+:\n)",
-        instructions,
-        re.DOTALL,
-    )
-
-    assert section_match is not None
-    section = section_match.group("section")
-    assert "Before every C++ implementation Step" in section
-    assert "read docs/CODINGSTYLE.md" in section
-    assert "apply it to production and test code" in section
-    assert "Doxygen coverage applies only to production code" in section
+    assert "AGENTS.md" in instructions
+    project_links = re.findall(r"\[[^\]]+\]\(([^)]+)\)", read(ROOT / "AGENTS.md"))
+    assert "docs/CODINGSTYLE.md" in project_links
+    assert (ROOT / "docs/CODINGSTYLE.md").is_file()
 
 
-def test_agent_hierarchy_is_explicit():
-    coordinator = read(AGENT_DIR / "coordinator-agent.toml").lower()
-    assert "main agent" in coordinator
-    assert "docs/<feature-id>/coordination.md" in coordinator
-    assert "worklist" in coordinator
-    assert "sub-agent dispatch" in coordinator
-
+def test_agent_profiles_reference_shared_authority_and_feature_context():
     for path in sorted(AGENT_DIR.glob("*.toml")):
-        if path.name == "coordinator-agent.toml":
-            continue
-        text = read(path).lower()
-        assert "sub-agent" in text, path.name
-        assert "coordinator agent" in text, path.name
+        instructions = tomllib.loads(read(path))["developer_instructions"]
+        for document in (
+            "AGENTS.md",
+            "docs/SOLVER_AGENT_DESIGN.md",
+            "docs/FEATURE_DOCUMENTATION.md",
+            "FEATURED_AGENTS.md",
+            "PRD.md",
+            "ADR.md",
+        ):
+            assert document in instructions, (path.name, document)
+        for skill_name in re.findall(r"\$([a-z0-9-]+)", instructions):
+            candidates = (
+                SKILL_DIR / skill_name / "SKILL.md",
+                ROOT / ".agents/skills" / skill_name / "SKILL.md",
+            )
+            assert any(candidate.is_file() for candidate in candidates), (path, skill_name)
 
 
 def test_live_contracts_have_no_retired_names_or_paths():
@@ -341,10 +346,9 @@ def test_live_contracts_have_no_retired_names_or_paths():
         assert token not in text, token
 
 
-def test_live_guidance_declares_exact_ordered_eight_stage_workflow():
+def test_live_guidance_declares_consistent_ordered_nine_stage_workflow():
     for path in LIVE_CONTRACT_FILES:
         text = read(path)
-        assert "8단계" in text, path
         assert "docs/<feature-id>/" in text, path
 
     for path in AGENT_STAGE_GUIDANCE_FILES:
@@ -389,7 +393,9 @@ def test_agent_toml_and_limited_skill_yaml_schemas_are_parseable():
     for path in sorted(AGENT_DIR.glob("*.toml")):
         parsed = tomllib.loads(read(path))
         assert parsed["name"] == path.stem
-        assert parsed["model_reasoning_effort"] == "extra high"
+        assert parsed["model_reasoning_effort"] == "xhigh"
+        assert parsed["sandbox_mode"] == "workspace-write"
+        assert parsed["developer_instructions"].strip(), path
 
     for skill_name in EXPECTED_SKILLS:
         skill_dir = SKILL_DIR / skill_name
@@ -397,3 +403,20 @@ def test_agent_toml_and_limited_skill_yaml_schemas_are_parseable():
         interface = parse_skill_interface(skill_dir / "agents" / "openai.yaml")
         assert frontmatter["name"] == skill_name
         assert f"${skill_name}" in interface["default_prompt"]
+
+
+@pytest.mark.parametrize("document", [".agents/skills/harness/SKILL.md", "docs/HARNESS_WORKFLOW.md"])
+def test_documented_phase_example_passes_executor_preflight(step_runtime, document):
+    """Exercise the documented schema with its real consumer, not prose matching."""
+    from scripts import execute
+
+    examples = [json.loads(block) for block in re.findall(r"```json\s*\n(.*?)\n```", read(ROOT / document), re.DOTALL)]
+    phase = next(example for example in examples if "steps" in example)
+    phase["project"] = "Harness Tests"
+    phase["phase"] = "runtime"
+    phase["steps"][0]["name"] = "runtime"
+    runtime = step_runtime
+    runtime.executor._write_json(runtime.index_file, phase)
+    executor = execute.StepExecutor("runtime")
+    executor._preflight_repository()
+    assert runtime.calls == []

tests/test_execute.py

@@ -1,18 +1,33 @@
 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',
+        f'model = "{model}"\nmodel_reasoning_effort = "{reasoning}"\n'
+        'developer_instructions = """Implementation Agent test role."""\n',
         encoding="utf-8",
     )
 
@@ -20,6 +35,7 @@ def write_implementation_agent_profile(
 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(
             {
@@ -57,6 +73,7 @@ def test_invoke_codex_uses_implementation_agent_command_contract(
 ):
     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(
             {
@@ -95,6 +112,7 @@ def test_invoke_codex_accepts_explicit_danger_full_access_override(
 ):
     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(
             {
@@ -111,14 +129,23 @@ def test_invoke_codex_accepts_explicit_danger_full_access_override(
     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("FESA_HARNESS_CODEX_SANDBOX", "danger-full-access")
+    monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "danger-full-access")
 
-    executor = execute.StepExecutor("sandbox-phase")
+    executor = execute.StepExecutor("sandbox-phase", danger_full_access_approved=True)
     executor._invoke_codex({"step": 0, "name": "sandbox"}, "")
 
     sandbox_index = invocation["command"].index("--sandbox")
@@ -128,6 +155,7 @@ def test_invoke_codex_accepts_explicit_danger_full_access_override(
 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(
             {
@@ -153,6 +181,7 @@ def test_invoke_codex_timeout_writes_structured_failure(tmp_path, monkeypatch):
     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 1800 seconds." in output["stderr"]
     stored = json.loads((phase_dir / "step0-output.json").read_text(encoding="utf-8"))
@@ -162,12 +191,13 @@ def test_invoke_codex_timeout_writes_structured_failure(tmp_path, monkeypatch):
 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"}],
+                "steps": [{"step": 0, "name": "retry", "status": "pending", "owned_paths": []}],
             }
         ),
         encoding="utf-8",
@@ -177,6 +207,15 @@ def test_execute_single_step_retries_structured_codex_timeout(tmp_path, monkeypa
     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"])
 
@@ -184,9 +223,6 @@ def test_execute_single_step_retries_structured_codex_timeout(tmp_path, monkeypa
     monkeypatch.setattr(execute.subprocess, "run", timeout_run)
 
     executor = execute.StepExecutor("retry-phase")
-    executor.MAX_RETRIES = 2
-    monkeypatch.setattr(executor, "_commit_step", lambda *args: None)
-    monkeypatch.setattr(executor, "_update_top_index", lambda *args: None)
 
     with pytest.raises(SystemExit, match="1"):
         executor._execute_single_step(
@@ -197,11 +233,15 @@ def test_execute_single_step_retries_structured_codex_timeout(tmp_path, monkeypa
     index = json.loads((phase_dir / "index.json").read_text(encoding="utf-8"))
     assert index["steps"][0]["status"] == "error"
     assert "Codex timed out after 1800 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(
             {
@@ -215,13 +255,59 @@ def test_invoke_codex_rejects_unknown_sandbox_override(tmp_path, monkeypatch):
     (phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")
 
     monkeypatch.setattr(execute, "ROOT", tmp_path)
-    monkeypatch.setenv("FESA_HARNESS_CODEX_SANDBOX", "read-only")
+    monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "read-only")
 
     executor = execute.StepExecutor("sandbox-phase")
-    with pytest.raises(execute.CodexEnvironmentError, match="FESA_HARNESS_CODEX_SANDBOX"):
+    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,
@@ -279,3 +365,205 @@ def test_configure_utf8_stdio_reconfigures_both_streams(monkeypatch):
 
     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"

tests/test_execute_ownership.py

@@ -0,0 +1,348 @@
+import os
+import subprocess
+
+import pytest
+
+from scripts import execute
+from test_execute import run_step, step_runtime
+
+
+def set_owned_paths(runtime, paths):
+    index = runtime.executor._read_phase_index()
+    index["steps"][0]["owned_paths"] = paths
+    runtime.executor._write_json(runtime.index_file, index)
+
+
+def link_directory(runtime, link, target):
+    if os.name == "nt":
+        # Junctions exercise Windows link resolution without symlink privileges.
+        runtime.real_run(
+            ["cmd", "/c", "mklink", "/J", str(link), str(target)],
+            cwd=runtime.root, capture_output=True, check=True,
+        )
+    else:
+        link.symlink_to(target, target_is_directory=True)
+
+
+@pytest.mark.parametrize("paths", [
+    None, "src/a.py", [1], [""], ["."], ["src/"], ["src/**"],
+    ["../a.py"], ["src/../a.py"], ["/a.py"], ["C:/a.py"], ["C:a.py"],
+    ["\\\\server\\share\\a.py"], ["src/a.py:stream"], [".git/config"],
+    ["phases/runtime/index.json"], ["phases/other/step0.md"],
+    ["src/a.py", "src\\a.py"], ["src/a.py\x00"], ["src/a.py."],
+])
+def test_owned_paths_rejects_ambiguous_or_unsafe_entries(step_runtime, paths):
+    runtime = step_runtime
+    set_owned_paths(runtime, paths)
+    with pytest.raises(execute.HarnessError, match="owned_paths"):
+        runtime.executor._read_phase_index()
+
+
+@pytest.mark.skipif(os.name != "nt", reason="Windows path identity")
+def test_owned_paths_rejects_case_alias_duplicates(step_runtime):
+    set_owned_paths(step_runtime, ["src/a.py", "SRC/A.PY"])
+    with pytest.raises(execute.HarnessError, match="owned_paths"):
+        step_runtime.executor._read_phase_index()
+
+
+def test_missing_pending_ownership_stops_before_branch_or_state_changes(step_runtime):
+    runtime = step_runtime
+    index = runtime.executor._read_phase_index()
+    del index["steps"][0]["owned_paths"]
+    runtime.executor._write_json(runtime.index_file, index)
+    original = runtime.index_file.read_bytes()
+    branch = runtime.git("symbolic-ref", "HEAD")
+    with pytest.raises(execute.HarnessError, match="owned_paths"):
+        runtime.executor.run()
+    assert runtime.calls == []
+    assert runtime.index_file.read_bytes() == original
+    assert runtime.git("symbolic-ref", "HEAD") == branch
+
+
+def test_completed_history_without_ownership_remains_readable(step_runtime):
+    runtime = step_runtime
+    index = runtime.executor._read_phase_index()
+    index["steps"][0] = {"step": 0, "name": "runtime", "status": "completed", "summary": "historical"}
+    runtime.executor._write_json(runtime.index_file, index)
+    assert execute.StepExecutor("runtime")._read_phase_index()["steps"][0]["summary"] == "historical"
+
+
+@pytest.mark.parametrize("path", ["src", "escape/file.py"])
+def test_pending_ownership_rejects_directory_or_link_escape(step_runtime, path):
+    runtime = step_runtime
+    if path == "src":
+        (runtime.root / "src").mkdir()
+    else:
+        outside = runtime.root.parent / (runtime.root.name + "-outside")
+        outside.mkdir()
+        link_directory(runtime, runtime.root / "escape", outside)
+    set_owned_paths(runtime, [path])
+    with pytest.raises(execute.HarnessError, match="owned_paths"):
+        runtime.executor._preflight_repository()
+    assert runtime.calls == []
+
+
+@pytest.mark.parametrize("history", [None, {}, ["bad"], [{"attempt": 1}]])
+def test_malformed_attempt_history_is_rejected_without_erasure(step_runtime, history):
+    runtime = step_runtime
+    index = runtime.executor._read_phase_index()
+    index["steps"][0]["attempt_history"] = history
+    runtime.executor._write_json(runtime.index_file, index)
+    original = runtime.index_file.read_bytes()
+    with pytest.raises(execute.HarnessError, match="attempt_history"):
+        runtime.executor._read_phase_index()
+    assert runtime.index_file.read_bytes() == original
+
+
+@pytest.mark.parametrize("field,value", [("owned_paths", ["unassigned.md"]), ("attempt_history", [])])
+def test_agent_cannot_change_executor_or_planning_metadata(step_runtime, field, value):
+    runtime = step_runtime
+
+    def tamper(runtime):
+        index = runtime.executor._read_phase_index()
+        index["steps"][0][field] = value
+        runtime.executor._write_json(runtime.index_file, index)
+        return {"status": "completed"}
+
+    runtime.responses = [tamper]
+    with pytest.raises(execute.HarnessError, match="immutable metadata"):
+        run_step(runtime)
+    current = runtime.executor._read_phase_index()["steps"][0]
+    assert current["owned_paths"] == []
+    assert "attempt_history" not in current
+    assert current["status"] == "error"
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
+
+
+@pytest.mark.parametrize("field,value", [("reported_status", []), ("classification", []), ("exit_code", True), ("attempt", True)])
+def test_history_record_types_fail_closed(step_runtime, field, value):
+    runtime = step_runtime
+    runtime.responses = [{"status": "error"}]
+    with pytest.raises(SystemExit):
+        run_step(runtime)
+    index = runtime.executor._read_phase_index()
+    index["steps"][0]["attempt_history"][0][field] = value
+    runtime.executor._write_json(runtime.index_file, index)
+    with pytest.raises(execute.HarnessError, match="attempt_history"):
+        runtime.executor._read_phase_index()
+
+
+def test_all_pending_steps_need_ownership_before_execution(step_runtime):
+    runtime = step_runtime
+    index = runtime.executor._read_phase_index()
+    index["steps"].append({"step": 1, "name": "later", "status": "pending"})
+    (runtime.index_file.parent / "step1.md").write_text("Later Step", encoding="utf-8")
+    runtime.executor._write_json(runtime.index_file, index)
+    executor = execute.StepExecutor("runtime")
+    with pytest.raises(execute.HarnessError, match="Step 1 requires an owned_paths"):
+        executor.run()
+    assert runtime.calls == []
+    assert "created_at" not in executor._read_phase_index()
+
+
+def test_metadata_commit_failure_preserves_implementation_commit_and_reason(step_runtime, monkeypatch):
+    runtime = step_runtime
+    set_owned_paths(runtime, ["approved.py"])
+
+    def change(runtime):
+        (runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
+        return {"status": "completed"}
+
+    original_run_git = runtime.executor._run_git
+
+    def fail_metadata_commit(*args):
+        if args[:2] == ("commit", "-m") and args[-1].startswith("chore("):
+            return subprocess.CompletedProcess(args, 1, stdout="", stderr="metadata commit failed")
+        return original_run_git(*args)
+
+    monkeypatch.setattr(runtime.executor, "_run_git", fail_metadata_commit)
+    runtime.responses = [change]
+    with pytest.raises(SystemExit) as stopped:
+        run_step(runtime)
+    assert stopped.value.code == 1
+    current = runtime.executor._read_phase_index()["steps"][0]
+    assert current["status"] == "error"
+    assert current["attempt_history"][-1]["reason"] == current["error_message"]
+    assert current["attempt_history"][-1]["classification"] == "commit-error"
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "2"
+    assert runtime.git("show", "HEAD:approved.py") == "approved\n"
+    assert runtime.executor._read_top_index()["phases"][0]["status"] == "error"
+
+
+@pytest.mark.parametrize("staged", [False, True])
+@pytest.mark.parametrize("result", ["completed", "timeout"])
+def test_unassigned_changes_stop_commit_and_retry_without_reverting_files(step_runtime, staged, result):
+    runtime = step_runtime
+    set_owned_paths(runtime, ["approved.py"])
+
+    def change(runtime):
+        (runtime.root / "approved.py").write_text("approved change\n", encoding="utf-8")
+        (runtime.root / "unassigned.md").write_text("user change\n", encoding="utf-8")
+        if staged:
+            runtime.git("add", "--", "unassigned.md")
+        if result == "timeout":
+            return subprocess.TimeoutExpired("codex", 1800)
+        return {"status": "completed"}
+
+    runtime.responses = [change, {"status": "completed"}]
+    with pytest.raises(SystemExit) as stopped:
+        run_step(runtime)
+    assert stopped.value.code == 1
+    assert len(runtime.calls) == 1
+    current = runtime.executor._read_phase_index()["steps"][0]
+    assert current["status"] == "error"
+    assert "unassigned.md" in current["error_message"]
+    assert current["attempt_history"][0]["classification"] == "ownership-violation"
+    assert "unassigned.md" in current["attempt_history"][0]["reason"]
+    assert (runtime.root / "unassigned.md").read_text(encoding="utf-8") == "user change\n"
+    assert (runtime.root / "approved.py").read_text(encoding="utf-8") == "approved change\n"
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
+    assert runtime.git("diff", "--cached", "--name-only").strip() == ("unassigned.md" if staged else "")
+
+
+def test_empty_ownership_allows_only_metadata_commit(step_runtime):
+    runtime = step_runtime
+    runtime.responses = [{"status": "completed"}]
+    assert run_step(runtime)
+    assert runtime.git("log", "-1", "--format=%s").strip() == "chore(runtime): step 0 output"
+    assert runtime.git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD").strip() == "phases/runtime/index.json"
+
+
+def test_owned_creation_modification_deletion_and_rename_commit_separately(step_runtime):
+    runtime = step_runtime
+    for name in ("modify.py", "delete.py", "old.py"):
+        (runtime.root / name).write_text("original content\n", encoding="utf-8")
+    runtime.git("add", "--", "modify.py", "delete.py", "old.py")
+    runtime.git("commit", "--quiet", "-m", "test: seed tracked files")
+    set_owned_paths(runtime, ["created.py", "modify.py", "delete.py", "old.py", "new.py"])
+
+    def change(runtime):
+        (runtime.root / "created.py").write_text("created\n", encoding="utf-8")
+        (runtime.root / "modify.py").write_text("modified\n", encoding="utf-8")
+        (runtime.root / "delete.py").unlink()
+        runtime.git("mv", "--", "old.py", "new.py")
+        return {"status": "completed"}
+
+    runtime.responses = [change]
+    assert run_step(runtime)
+    assert set(runtime.git("diff-tree", "--no-commit-id", "--no-renames", "--name-only", "-r", "HEAD~1").splitlines()) == {"created.py", "modify.py", "delete.py", "old.py", "new.py"}
+    assert runtime.git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD").strip() == "phases/runtime/index.json"
+    assert runtime.git("status", "--porcelain").strip() == ""
+
+
+@pytest.mark.parametrize("owned", [["old.py"], ["new.py"]])
+def test_rename_requires_both_endpoints(step_runtime, owned):
+    runtime = step_runtime
+    (runtime.root / "old.py").write_text("content\n", encoding="utf-8")
+    runtime.git("add", "--", "old.py")
+    runtime.git("commit", "--quiet", "-m", "test: seed rename")
+    set_owned_paths(runtime, owned)
+
+    def rename(runtime):
+        runtime.git("mv", "--", "old.py", "new.py")
+        return {"status": "completed"}
+
+    runtime.responses = [rename]
+    with pytest.raises(SystemExit) as stopped:
+        run_step(runtime)
+    assert stopped.value.code == 1
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "2"
+    assert (runtime.root / "new.py").is_file()
+
+
+@pytest.mark.parametrize("injected_path", ["unassigned.md", "phases/runtime/index.json"])
+def test_new_unassigned_staged_path_is_rejected_before_commit(step_runtime, monkeypatch, injected_path):
+    runtime = step_runtime
+    set_owned_paths(runtime, ["approved.py"])
+
+    def change(runtime):
+        (runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
+        return {"status": "completed"}
+
+    original_run_git = runtime.executor._run_git
+
+    def inject_during_add(*args):
+        result = original_run_git(*args)
+        if "add" in args:
+            if injected_path == "unassigned.md":
+                (runtime.root / injected_path).write_text("late change\n", encoding="utf-8")
+            runtime.git("add", "--", injected_path)
+        return result
+
+    monkeypatch.setattr(runtime.executor, "_run_git", inject_during_add)
+    runtime.responses = [change]
+    with pytest.raises(SystemExit) as stopped:
+        run_step(runtime)
+    assert stopped.value.code == 1
+    assert injected_path in runtime.executor._read_phase_index()["steps"][0]["error_message"]
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
+    assert (runtime.root / injected_path).is_file()
+
+
+def test_owned_path_replaced_by_external_junction_is_rejected(step_runtime):
+    runtime = step_runtime
+    set_owned_paths(runtime, ["src/approved.py"])
+    outside = runtime.root.parent / (runtime.root.name + "-outside")
+    outside.mkdir()
+    (outside / "approved.py").write_text("outside data\n", encoding="utf-8")
+
+    def substitute(runtime):
+        link_directory(runtime, runtime.root / "src", outside)
+        return {"status": "completed"}
+
+    runtime.responses = [substitute]
+    with pytest.raises(SystemExit) as stopped:
+        run_step(runtime)
+    assert stopped.value.code == 1
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
+    assert (outside / "approved.py").read_text(encoding="utf-8") == "outside data\n"
+
+
+def test_late_unassigned_staging_survives_index_reset(step_runtime, monkeypatch):
+    runtime = step_runtime
+    set_owned_paths(runtime, ["approved.py"])
+
+    def change(runtime):
+        (runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
+        return {"status": "completed"}
+
+    original_run_git = runtime.executor._run_git
+
+    def inject_before_reset(*args):
+        if "reset" in args:
+            (runtime.root / "unassigned.md").write_text("staged user change\n", encoding="utf-8")
+            runtime.git("add", "--", "unassigned.md")
+        return original_run_git(*args)
+
+    monkeypatch.setattr(runtime.executor, "_run_git", inject_before_reset)
+    runtime.responses = [change]
+    with pytest.raises(SystemExit) as stopped:
+        run_step(runtime)
+    assert stopped.value.code == 1
+    assert "unassigned.md" in runtime.git("diff", "--cached", "--name-only").splitlines()
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
+
+
+@pytest.mark.parametrize("when", ["preflight", "after-attempt"])
+def test_owned_hard_link_is_rejected(step_runtime, when):
+    runtime = step_runtime
+    outside = runtime.root.parent / (runtime.root.name + "-outside.py")
+    outside.write_text("external data\n", encoding="utf-8")
+    set_owned_paths(runtime, ["approved.py"])
+
+    def substitute(runtime):
+        (runtime.root / "approved.py").hardlink_to(outside)
+        return {"status": "completed"}
+
+    if when == "preflight":
+        substitute(runtime)
+        with pytest.raises(execute.HarnessError, match="owned_paths"):
+            runtime.executor._preflight_repository()
+        assert runtime.calls == []
+    else:
+        runtime.responses = [substitute]
+        with pytest.raises(SystemExit) as stopped:
+            run_step(runtime)
+        assert stopped.value.code == 1
+    assert outside.read_text(encoding="utf-8") == "external data\n"
+    assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"