"""Model worker path, evidence, retention, and readiness contracts."""

from __future__ import annotations

import io
import json
import os
import subprocess
import sys
import urllib.error
from pathlib import Path

import pytest


ROOT = Path(__file__).parents[2]
SCRIPTS = ROOT / "services/hermes/scripts"
sys.path.insert(0, str(SCRIPTS))

import execution_pool_protocol as protocol  # noqa: E402
import execution_pool_worker as worker  # noqa: E402


def assignment(**changes):
    value = {
        "board": "metis", "task_id": "t_deadbeef", "run_id": "23",
        "worker_ordinal": 0, "attempt": 1, "protocol_version": 2,
        "workspace": "/workspace/runs/metis/t_deadbeef/23",
        "baseline_sha": "a" * 40,
        "payload": {"context": "safe objective", "assignee": "cli-auto"},
    }
    value.update(changes)
    return value


class Response:
    def __init__(self, value):
        self.value = value

    def __enter__(self):
        return self

    def __exit__(self, *_args):
        return False

    def read(self, _size=-1):
        return self.value


def test_post_client_and_poll_validate_every_response_boundary(monkeypatch):
    monkeypatch.setattr(
        worker.urllib.request, "urlopen",
        lambda *_a, **_k: Response(json.dumps({"safe": True}).encode()),
    )
    assert worker._post("http://mediator", {}) == {"safe": True}
    monkeypatch.setattr(
        worker.urllib.request, "urlopen",
        lambda *_a, **_k: Response(b"x" * (64 * 1024 + 1)),
    )
    with pytest.raises(protocol.ProtocolError, match="wire"):
        worker._post("http://mediator", {})
    monkeypatch.setattr(
        worker.urllib.request, "urlopen", lambda *_a, **_k: Response(b"[]")
    )
    with pytest.raises(protocol.ProtocolError, match="object"):
        worker._post("http://mediator", {})
    conflict = urllib.error.HTTPError(
        "http://mediator", 409, "Conflict", {}, io.BytesIO(b'{"error":"untrusted detail"}')
    )
    monkeypatch.setattr(
        worker.urllib.request, "urlopen", lambda *_a, **_k: (_ for _ in ()).throw(conflict)
    )
    with pytest.raises(protocol.ProtocolError, match="local mediator rejected") as rejected:
        worker._post("http://mediator", {})
    assert "untrusted detail" not in str(rejected.value)

    monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"error": "denied"})
    with pytest.raises(protocol.ProtocolError, match="local mediator rejected"):
        worker._client("poll")
    monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"safe": True})
    assert worker._client("poll") == {"safe": True}
    monkeypatch.setattr(worker, "ORDINAL", 0)
    monkeypatch.setattr(worker, "_client", lambda *_a, **_k: {"assignment": None})
    assert worker._poll() is None
    for value in ([], assignment(worker_ordinal=1), assignment(protocol_version=1)):
        monkeypatch.setattr(worker, "_client", lambda *_a, value=value, **_k: {"assignment": value})
        with pytest.raises(protocol.ProtocolError, match="foreign"):
            worker._poll()
    monkeypatch.setattr(worker, "_client", lambda *_a, **_k: {"assignment": assignment()})
    assert worker._poll()["task_id"] == "t_deadbeef"
    assert worker._binding(assignment()) == {
        "board": "metis", "task_id": "t_deadbeef", "run_id": "23",
        "worker_ordinal": 0, "attempt": 1,
    }


def test_state_path_rejects_traversal_root_and_leaf_symlinks(tmp_path, monkeypatch):
    monkeypatch.setattr(worker, "ROOT", tmp_path)
    path = worker._state_path(assignment())
    assert path == tmp_path / "session-state/metis/t_deadbeef/23.json"
    with pytest.raises(protocol.ProtocolError, match="invalid"):
        worker._state_path(assignment(task_id="../bad"))

    state_root = tmp_path / "session-state"
    outside = tmp_path / "outside"
    path.parent.rmdir()
    path.parent.parent.rmdir()
    state_root.rmdir()
    outside.mkdir()
    state_root.symlink_to(outside, target_is_directory=True)
    with pytest.raises(protocol.ProtocolError, match="root"):
        worker._state_path(assignment())
    state_root.unlink()
    path = worker._state_path(assignment())
    path.symlink_to("/etc/passwd")
    with pytest.raises(protocol.ProtocolError, match="symlink"):
        worker._state_path(assignment())


def prepare_provider_roots(tmp_path, monkeypatch):
    worker_root = tmp_path / "worker"
    data_root = tmp_path / "data"
    codex = tmp_path / "runtime/codex"
    claude = tmp_path / "runtime/claude"
    (worker_root / "provider-state").mkdir(parents=True)
    data_root.mkdir()
    codex.mkdir(parents=True)
    claude.mkdir(parents=True)
    monkeypatch.setattr(worker, "ROOT", worker_root)
    monkeypatch.setattr(worker.cli_lane_runner, "DATA_ROOT", data_root)
    monkeypatch.setenv("CODEX_HOME", str(codex))
    monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(claude))
    return worker_root, data_root, codex, claude


def test_provider_session_binding_replaces_only_owned_symlinks(tmp_path, monkeypatch):
    worker_root, data_root, codex, claude = prepare_provider_roots(tmp_path, monkeypatch)
    old = tmp_path / "old"
    old.mkdir()
    (data_root / "home").symlink_to(old, target_is_directory=True)
    (codex / "sessions").symlink_to(old, target_is_directory=True)
    worker._bind_provider_sessions(assignment())
    assert "metis/t_deadbeef/23" in str((data_root / "home").resolve())
    assert "metis/t_deadbeef/23" in str((codex / "sessions").resolve())
    settings = worker_root / "provider-state/metis/t_deadbeef/23/home/.claude/settings.json"
    assert json.loads(settings.read_text()) == {}
    worker._bind_provider_sessions(assignment())

    with pytest.raises(protocol.ProtocolError, match="binding"):
        worker._bind_provider_sessions(assignment(run_id="../bad"))
    monkeypatch.setattr(worker, "ROOT", tmp_path / "missing")
    with pytest.raises(protocol.ProtocolError, match="unavailable"):
        worker._bind_provider_sessions(assignment())


def test_provider_session_binding_preserves_legacy_runtime_directories(tmp_path, monkeypatch):
    _worker_root, _data_root, _codex, claude = prepare_provider_roots(tmp_path, monkeypatch)
    legacy = claude / "projects"
    (legacy / "prior-session").mkdir(parents=True)
    (legacy / "prior-session" / "record").write_text("preserved")

    worker._bind_provider_sessions(assignment())

    assert legacy.is_symlink()
    preserved = claude / ".hermes-legacy-projects/prior-session/record"
    assert preserved.read_text() == "preserved"
    worker._bind_provider_sessions(assignment())
    legacy.unlink()
    legacy.mkdir()
    with pytest.raises(protocol.ProtocolError, match="migration conflicts"):
        worker._bind_provider_sessions(assignment())
    assert preserved.read_text() == "preserved"


def test_provider_session_binding_rejects_durable_and_runtime_tampering(
    tmp_path, monkeypatch
):
    worker_root, data_root, codex, _claude = prepare_provider_roots(tmp_path, monkeypatch)
    run_parent = worker_root / "provider-state/metis"
    run_parent.symlink_to(tmp_path, target_is_directory=True)
    with pytest.raises(protocol.ProtocolError, match="session path"):
        worker._bind_provider_sessions(assignment())
    run_parent.unlink()
    home = worker_root / "provider-state/metis/t_deadbeef/23/home"
    home.parent.mkdir(parents=True)
    home.symlink_to(tmp_path, target_is_directory=True)
    with pytest.raises(protocol.ProtocolError, match="HOME"):
        worker._bind_provider_sessions(assignment())
    home.unlink()
    (data_root / "home").write_text("not-owned")
    with pytest.raises(protocol.ProtocolError, match="task-bound"):
        worker._bind_provider_sessions(assignment())
    (data_root / "home").unlink()
    (codex / "sessions").write_text("not-owned")
    with pytest.raises(protocol.ProtocolError, match="not a symlink"):
        worker._bind_provider_sessions(assignment())
    (codex / "sessions").unlink()
    durable_sessions = (
        worker_root / "provider-state/metis/t_deadbeef/23/codex/sessions"
    )
    durable_sessions.rmdir()
    durable_sessions.symlink_to(tmp_path, target_is_directory=True)
    with pytest.raises(protocol.ProtocolError, match="session path"):
        worker._bind_provider_sessions(assignment())


def test_prompt_activity_git_and_result_bounding(tmp_path, monkeypatch):
    text = worker._prompt("objective", tmp_path, worker._binding(assignment()))
    assert "no Kubernetes identity" in text and "objective" in text
    missing = tmp_path / "missing.log"
    assert worker._read_activity(missing, 7) == ("", 7)
    log = tmp_path / "worker.log"
    log.write_text("abc")
    assert worker._read_activity(log, 0) == ("abc", 3)
    assert worker._read_activity(log, 99) == ("abc", 3)
    fifo = tmp_path / "fifo"
    os.mkfifo(fifo)
    with pytest.raises(protocol.ProtocolError, match="regular"):
        worker._read_activity(fifo, 0)

    repo = tmp_path / "repo"
    subprocess.run(["git", "init", "-q", str(repo)], check=True)
    assert worker._git(repo, "status", "--porcelain") == ""
    with pytest.raises(RuntimeError, match="ambiguous"):
        worker._git(repo, "rev-parse", "missing")

    value = {
        "status": "completed", "summary": "s" * 20_000,
        "changed_files": ["x" * 3000] * 100,
        "tests_run": "not-list", "artifacts": [], "findings": [], "blockers": [],
    }
    bounded = worker._bounded_result(value)
    assert len(protocol.canonical_json(bounded)) <= 32 * 1024
    assert bounded["tests_run"] == [] and len(bounded["summary"]) <= 8000


def test_refresh_assignment_requires_exact_binding(monkeypatch):
    exact = worker._binding(assignment())
    monkeypatch.setattr(worker, "_poll", lambda: assignment())
    assert worker._refresh_assignment(exact)["run_id"] == "23"
    for value in (None, assignment(attempt=2)):
        monkeypatch.setattr(worker, "_poll", lambda value=value: value)
        with pytest.raises(protocol.ProtocolError, match="changed"):
            worker._refresh_assignment(exact)


def write_gc_state(root, task_id, workspace, terminal_at):
    path = root / f"session-state/metis/{task_id}/23.json"
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps({"terminal_at": terminal_at, "workspace": str(workspace)}))
    return path


def init_clean_repo(path):
    subprocess.run(["git", "init", "-q", str(path)], check=True)
    subprocess.run(["git", "-C", str(path), "config", "user.email", "a@b.c"], check=True)
    subprocess.run(["git", "-C", str(path), "config", "user.name", "Test"], check=True)
    (path / "tracked").write_text("safe")
    subprocess.run(["git", "-C", str(path), "add", "tracked"], check=True)
    subprocess.run(["git", "-C", str(path), "commit", "-qm", "initial"], check=True)
    branch = subprocess.run(
        ["git", "-C", str(path), "symbolic-ref", "--short", "HEAD"], text=True,
        capture_output=True, check=True,
    ).stdout.strip()
    subprocess.run(["git", "-C", str(path), "update-ref", f"refs/remotes/origin/{branch}", "HEAD"], check=True)


def test_retention_skips_young_outside_dirty_and_symlink_workspaces(tmp_path, monkeypatch):
    monkeypatch.setattr(worker, "ROOT", tmp_path)
    monkeypatch.setattr(worker, "RETENTION_SECONDS", 3600)
    run_root = tmp_path / "runs/metis"
    clean = run_root / "clean/23"
    dirty = run_root / "dirty/23"
    clean.mkdir(parents=True)
    dirty.mkdir(parents=True)
    init_clean_repo(clean)
    init_clean_repo(dirty)
    (dirty / "untracked").write_text("dirty")
    now = 10_000
    write_gc_state(tmp_path, "clean", clean, 1)
    dirty_state = write_gc_state(tmp_path, "dirty", dirty, 1)
    young = write_gc_state(tmp_path, "young", clean, now)
    outside = write_gc_state(tmp_path, "outside", tmp_path / "missing", 1)
    assert worker.garbage_collect(now=now) == 1
    assert not clean.exists() and dirty.exists()
    assert dirty_state.exists() and young.exists() and outside.exists()

    link = run_root / "link/23"
    link.parent.mkdir()
    link.symlink_to(dirty, target_is_directory=True)
    link_state = write_gc_state(tmp_path, "link", link, 1)
    assert worker.garbage_collect(now=now) == 0
    assert link_state.exists()


def test_retention_keeps_a_clean_unpublished_commit(tmp_path, monkeypatch):
    """A terminal source checkout remains available for mediator publication retry."""
    monkeypatch.setattr(worker, "ROOT", tmp_path)
    monkeypatch.setattr(worker, "RETENTION_SECONDS", 3600)
    workspace = tmp_path / "runs/metis/retry/23"
    workspace.mkdir(parents=True)
    init_clean_repo(workspace)
    branch = subprocess.run(
        ["git", "-C", str(workspace), "symbolic-ref", "--short", "HEAD"], text=True,
        capture_output=True, check=True,
    ).stdout.strip()
    subprocess.run(["git", "-C", str(workspace), "update-ref", "-d", f"refs/remotes/origin/{branch}"], check=True)
    state = write_gc_state(tmp_path, "retry", workspace, 1)
    assert worker.garbage_collect(now=10_000) == 0
    assert workspace.exists() and state.exists()


def test_readiness_checks_ordinal_paths_credentials_and_mediator(tmp_path, monkeypatch):
    worker_root = tmp_path / "worker"
    data_root = tmp_path / "data"
    claude_token = tmp_path / "claude-oauth/token"
    for path in (worker_root, worker_root / "provider-state", data_root):
        path.mkdir(parents=True, exist_ok=True)
    claude_token.parent.mkdir()
    claude_token.write_text("setup-token")
    schema = tmp_path / "schema/result.json"
    claude_bin = tmp_path / "tools/claude"
    native_claude = tmp_path / "tools/claude-native"
    codex_bin = tmp_path / "tools/codex"
    claude_bin.parent.mkdir()
    for binary in (claude_bin, native_claude):
        binary.write_text("#!/bin/sh\nexit 0\n")
        binary.chmod(0o755)
    monkeypatch.setattr(worker, "ORDINAL", 0)
    monkeypatch.setattr(worker, "ROOT", worker_root)
    monkeypatch.setattr(worker.cli_lane_runner, "DATA_ROOT", data_root)
    monkeypatch.setattr(worker.cli_lane_runner, "RESULT_SCHEMA_PATH", schema)
    monkeypatch.setattr(worker.cli_lane_runner, "CLAUDE_BIN", claude_bin)
    monkeypatch.setattr(worker.cli_lane_runner, "CODEX_BIN", codex_bin)
    monkeypatch.setattr(worker, "DISABLED_PROVIDER", "codex")
    monkeypatch.setenv("HERMES_CLAUDE_NATIVE_BIN", str(native_claude))
    monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN_FILE", str(claude_token))
    monkeypatch.setattr(
        worker, "_poll", lambda: pytest.fail("readiness must not poll the mediator")
    )
    worker.readiness()
    assert json.loads(schema.read_text()) == worker.cli_lane_runner.RESULT_SCHEMA
    monkeypatch.setattr(worker, "DISABLED_PROVIDER", "")
    with pytest.raises(protocol.ProtocolError, match="provider executable.*codex"):
        worker.readiness()
    codex_bin.write_text("#!/bin/sh\nexit 0\n")
    codex_bin.chmod(0o755)
    worker.readiness()
    monkeypatch.setattr(worker, "DISABLED_PROVIDER", "codex")

    monkeypatch.setattr(worker, "ORDINAL", 3)
    with pytest.raises(protocol.ProtocolError, match="ordinal"):
        worker.readiness()
    monkeypatch.setattr(worker, "ORDINAL", 0)
    monkeypatch.setattr(worker, "ROOT", tmp_path / "missing")
    with pytest.raises(protocol.ProtocolError, match="path"):
        worker.readiness()
    monkeypatch.setattr(worker, "ROOT", worker_root)
    native_claude.unlink()
    with pytest.raises(protocol.ProtocolError, match="provider executable.*claude-native"):
        worker.readiness()
    native_claude.write_text("#!/bin/sh\nexit 0\n")
    native_claude.chmod(0o755)
    claude_token.unlink()
    with pytest.raises(protocol.ProtocolError, match="credential"):
        worker.readiness()


def test_main_defers_rejected_or_malformed_poll_before_valid_assignment(monkeypatch):
    monkeypatch.setattr(worker, "readiness", lambda: None)
    monkeypatch.setattr(worker, "garbage_collect", lambda: 0)
    expected = assignment()
    polls = iter((
        urllib.error.HTTPError("http://mediator", 409, "Conflict", {}, None),
        protocol.ProtocolError("bad wire"),
        expected,
    ))
    executed = []

    def poll():
        value = next(polls)
        if isinstance(value, Exception):
            raise value
        return value

    def execute(value):
        executed.append(value)
        raise StopIteration("stop")

    monkeypatch.setattr(worker, "_poll", poll)
    monkeypatch.setattr(worker, "execute", execute)
    monkeypatch.setattr(worker.time, "sleep", lambda _delay: None)
    with pytest.raises(StopIteration, match="stop"):
        worker.main()
    assert executed == [expected]


def test_publication_retry_calls_only_mediator_and_never_starts_a_model(monkeypatch):
    """A fresh run republishes retained evidence without a checkout or provider session."""
    structured = {
        "status": "completed", "summary": "Replace cache literals.",
        "changed_files": ["internal/k8s/job_manifests.go"], "tests_run": ["go test ./..."],
        "artifacts": [], "findings": [], "blockers": [],
    }
    resume = {"structured": structured, "title": "Repair", "body": "evidence", "head": "b" * 40}
    item = assignment(payload={"scm_resume": resume})
    calls = []

    def client(operation, **values):
        calls.append((operation, values))
        if operation == "resume":
            return {"scm_submission": {"branch": "wt/t_deadbeef", "pull_request": "https://scm/pulls/3", "head": "b" * 40}}
        return {"ack": {"accepted": True}}

    monkeypatch.setattr(worker, "_client", client)
    monkeypatch.setattr(worker, "_bind_provider_sessions", lambda *_args: pytest.fail("must not bind provider state"))
    monkeypatch.setattr(worker.cli_lane_runner, "run_provider", lambda *_args: pytest.fail("must not invoke a model"))
    worker.execute(item)
    assert [name for name, _values in calls] == ["resume", "finish"]
    terminal = calls[1][1]["payload"]
    assert terminal["structured"]["status"] == "completed"
    assert "preserved-head:" + "b" * 40 in terminal["structured"]["artifacts"]


def test_publication_retry_policy_error_is_a_visible_nontransient_block(monkeypatch):
    """Only the mediator's explicit transient response can release a receipt."""
    structured = {"status": "completed", "summary": "Repair", "changed_files": [], "tests_run": [], "artifacts": [], "findings": [], "blockers": []}
    item = assignment(payload={"scm_resume": {"structured": structured, "title": "Repair", "body": "evidence", "head": "b" * 40}})
    calls = []

    def client(operation, **values):
        calls.append((operation, values))
        if operation == "resume":
            raise protocol.ProtocolError("policy denied")
        return {"ack": {"accepted": True}}

    monkeypatch.setattr(worker, "_client", client)
    monkeypatch.setattr(worker.cli_lane_runner, "run_provider", lambda *_args: pytest.fail("must not invoke a model"))
    worker.execute(item)
    terminal = calls[-1][1]["payload"]
    assert [name for name, _ in calls] == ["resume", "finish"]
    assert terminal["capacity_failure"] is False
    assert terminal["structured"]["status"] == "blocked"