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"