"""Integration tests for dirty same-claimant author-session rebind (#864).

Uses real temp git repos/worktrees and a temp GITEA_ISSUE_LOCK_DIR. Does not
mutate any real #860/#864 worktree on disk.
"""

from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import dirty_same_claimant_session_rebind as rebind  # noqa: E402
import issue_lock_provenance  # noqa: E402
import issue_lock_store as ils  # noqa: E402
import issue_lock_worktree  # noqa: E402

ISSUE = 864
BRANCH = f"fix/issue-{ISSUE}-dirty-same-claimant-session-rebind"
REMOTE = "prgs"
ORG = "Scaled-Tech-Consulting"
REPO = "Gitea-Tools"
IDENTITY = "jcwalker3"
PROFILE = "prgs-author"


def _git(cwd: str, *args: str, check: bool = True) -> subprocess.CompletedProcess:
    return subprocess.run(
        ["git", "-C", cwd, *args],
        capture_output=True,
        text=True,
        check=check,
    )


def dead_pid() -> int:
    proc = subprocess.Popen([sys.executable, "-c", "pass"])
    proc.wait()
    return proc.pid


def future_ts(hours: int = 4) -> str:
    return (
        (datetime.now(timezone.utc) + timedelta(hours=hours))
        .isoformat()
        .replace("+00:00", "Z")
    )


@pytest.fixture
def lock_dir(tmp_path, monkeypatch):
    d = tmp_path / "issue-locks"
    d.mkdir()
    monkeypatch.setenv("GITEA_ISSUE_LOCK_DIR", str(d))
    return str(d)


@pytest.fixture
def dirty_repo(tmp_path):
    """Canonical repo root with branches/<name> worktree and dirty content."""
    root = tmp_path / "repo"
    root.mkdir()
    main = root / "main"
    main.mkdir()
    subprocess.run(["git", "init", "-q", str(main)], check=True, capture_output=True)
    _git(str(main), "config", "user.email", "t@t")
    _git(str(main), "config", "user.name", "t")
    (main / "README.md").write_text("base\n", encoding="utf-8")
    _git(str(main), "add", "README.md")
    _git(str(main), "commit", "-q", "-m", "base")
    _git(str(main), "branch", "-M", "master")

    # Bare remote + origin tracking so remote head is observable offline.
    bare = tmp_path / "remote.git"
    subprocess.run(
        ["git", "init", "--bare", "-q", str(bare)], check=True, capture_output=True
    )
    _git(str(main), "remote", "add", "origin", str(bare))
    _git(str(main), "push", "-q", "origin", "master:master")

    branches = root / "branches"
    branches.mkdir()
    wt_name = f"fix-issue-{ISSUE}-dirty-same-claimant-session-rebind"
    wt = branches / wt_name
    _git(str(main), "worktree", "add", "-q", "-b", BRANCH, str(wt))
    _git(str(wt), "push", "-q", "-u", "origin", BRANCH)

    # Seed committed files we will dirty.
    tracked = [
        "dirty_same_claimant_session_rebind.py",
        "issue_lock_provenance.py",
        "task_capability_map.py",
    ]
    for rel in tracked:
        p = wt / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(f"seed {rel}\n", encoding="utf-8")
    _git(str(wt), "add", *tracked)
    _git(str(wt), "commit", "-q", "-m", "seed tracked")
    _git(str(wt), "push", "-q", "origin", BRANCH)

    # Dirty tracked + untracked.
    for rel in tracked:
        (wt / rel).write_text(f"dirty {rel}\n", encoding="utf-8")
    untracked = [
        "tests/test_dirty_same_claimant_session_rebind.py",
        "docs/runbook-dirty-rebind.md",
        "scratch/notes-untracked.txt",
        "extra_untracked.txt",
    ]
    for rel in untracked:
        p = wt / rel
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_text(f"untracked {rel}\n", encoding="utf-8")

    inv = rebind.collect_dirty_inventory(str(wt))
    assert inv["ok"], inv.get("reasons")
    head = _git(str(wt), "rev-parse", "HEAD").stdout.strip()
    remote_head = _git(
        str(wt), "rev-parse", f"refs/remotes/origin/{BRANCH}"
    ).stdout.strip()
    assert head == remote_head

    return {
        "root": str(root),
        "main": str(main),
        "worktree": str(wt),
        "branch": BRANCH,
        "inventory": inv,
        "local_head": head,
        "remote_head": remote_head,
        "dirty_paths": list(inv["dirty_paths"]),
        "fingerprints": dict(inv["fingerprints"]),
    }


def _make_lock(
    *,
    worktree: str,
    pid: int,
    lock_dir: str,
    identity: str = IDENTITY,
    profile: str = PROFILE,
    **overrides,
) -> dict:
    lease = {
        "operation_type": ils.AUTHOR_ISSUE_WORK_LEASE,
        "issue_number": ISSUE,
        "branch": BRANCH,
        "worktree_path": worktree,
        "claimant": {"username": identity, "profile": profile},
        "created_at": "2026-01-01T00:00:00Z",
        "expires_at": future_ts(),
        "last_heartbeat_at": "2026-01-01T00:00:00Z",
    }
    lock = {
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": worktree,
        "remote": REMOTE,
        "org": ORG,
        "repo": REPO,
        "session_pid": pid,
        "pid": pid,
        "work_lease": lease,
        "lock_generation": 1,
        "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
            tool="gitea_lock_issue",
            claimant={"username": identity, "profile": profile},
        ),
    }
    lock.update(overrides)
    path = ils.lock_file_path(
        remote=REMOTE, org=ORG, repo=REPO, issue_number=ISSUE, lock_dir=lock_dir
    )
    lock["lock_file_path"] = path
    ils.save_lock_file(path, lock)
    # Stale session pointer for the dead owner.
    ptr = {
        "pid": pid,
        "lock_file_path": path,
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "remote": REMOTE,
        "org": ORG,
        "repo": REPO,
    }
    ils.save_lock_file(os.path.join(lock_dir, f"session-{pid}.json"), ptr)
    return ils.read_lock_file(path) or lock


def _apply_kwargs(repo, lock, lock_dir, **overrides):
    kwargs = {
        "remote": REMOTE,
        "org": ORG,
        "repo": REPO,
        "issue_number": ISSUE,
        "branch_name": BRANCH,
        "worktree_path": repo["worktree"],
        "claimant_identity": IDENTITY,
        "claimant_profile": PROFILE,
        "old_pid": lock.get("session_pid") or lock.get("pid"),
        "expected_local_head": repo["local_head"],
        "expected_remote_head": repo["remote_head"],
        "expected_dirty_paths": repo["dirty_paths"],
        "expected_fingerprints": repo["fingerprints"],
        "existing_lock": lock,
        "current_identity": IDENTITY,
        "current_profile": PROFILE,
        "role_kind": "author",
        "current_pid": os.getpid(),
        "current_branch": BRANCH,
        "local_head": repo["local_head"],
        "remote_head": repo["remote_head"],
        "dirty_inventory": repo["inventory"],
        "competing_live_locks": [],
        "competing_sessions": [],
        "workflow_lease_active": False,
        "repo_root": repo["root"],
        "dry_run": False,
        "lock_dir": lock_dir,
    }
    kwargs.update(overrides)
    return kwargs


# ── 1. Successful dead-PID same-claimant dirty rebind ───────────────────────


def test_successful_dead_pid_same_claimant_dirty_rebind(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert result["success"], result
    assert result["outcome"] == rebind.REBIND_SANCTIONED
    assert result["old_pid"] == old
    assert result["new_pid"] == os.getpid()
    assert result["generation_after"] == result["generation_before"] + 1

    rebound = ils.read_lock_file(result["lock_path"])
    assert rebound is not None
    assert int(rebound["session_pid"]) == os.getpid()
    assert int(rebound["pid"]) == os.getpid()
    assert (
        rebound.get("lock_provenance", {}).get("source")
        == issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND
    )
    assert rebound.get("rebind_record", {}).get("old_pid") == old


# ── 2. Byte-for-byte preservation ───────────────────────────────────────────


def test_byte_for_byte_preservation(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    before = {
        rel: rebind.content_fingerprint(os.path.join(dirty_repo["worktree"], rel))
        for rel in dirty_repo["dirty_paths"]
    }
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert result["success"], result
    after = {
        rel: rebind.content_fingerprint(os.path.join(dirty_repo["worktree"], rel))
        for rel in dirty_repo["dirty_paths"]
    }
    assert before == after
    assert result["fingerprints"] == before


# ── 3. Exact dirty-path and fingerprint enforcement ─────────────────────────


def test_extra_dirty_path_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    pins = list(dirty_repo["dirty_paths"])[:-1]  # missing one observed path
    fps = {p: dirty_repo["fingerprints"][p] for p in pins}
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            expected_dirty_paths=pins,
            expected_fingerprints=fps,
        )
    )
    assert not result["success"]
    assert any("unexpected paths" in r for r in result["reasons"])


def test_missing_expected_dirty_path_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    pins = list(dirty_repo["dirty_paths"]) + ["not_really_dirty.txt"]
    fps = dict(dirty_repo["fingerprints"])
    fps["not_really_dirty.txt"] = "0" * 64
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            expected_dirty_paths=pins,
            expected_fingerprints=fps,
        )
    )
    assert not result["success"]
    assert any("missing expected" in r for r in result["reasons"])


def test_modified_fingerprint_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    fps = dict(dirty_repo["fingerprints"])
    victim = dirty_repo["dirty_paths"][0]
    fps[victim] = "f" * 64
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            expected_fingerprints=fps,
        )
    )
    assert not result["success"]
    assert any("fingerprint disagreement" in r for r in result["reasons"])


# ── 4. Atomic session-pointer replacement ───────────────────────────────────


def test_session_pointer_points_to_lock(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert result["success"], result
    new_ptr_path = os.path.join(lock_dir, f"session-{os.getpid()}.json")
    assert os.path.exists(new_ptr_path)
    ptr = ils.read_lock_file(new_ptr_path)
    assert ptr is not None
    assert os.path.realpath(ptr["lock_file_path"]) == os.path.realpath(
        result["lock_path"]
    )
    # Old pointer removed when it targeted this lock.
    old_ptr = os.path.join(lock_dir, f"session-{old}.json")
    assert not os.path.exists(old_ptr)
    assert result.get("removed_old_session_pointer") is True


# ── 5. Retry after interruption (journal mid-state) ─────────────────────────


def test_retry_after_journal_mid_state(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    jpath = rebind.journal_path(lock_dir, ISSUE)
    rebind._atomic_write_json(
        jpath,
        {
            "phase": rebind.JOURNAL_PHASE_PRE_BIND,
            "issue_number": ISSUE,
            "old_pid": old,
            "new_pid": os.getpid(),
            "expected_generation": 1,
        },
    )
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert result["success"], result
    assert result["journal_phase"] == rebind.JOURNAL_PHASE_COMPLETE

    # Second apply is already_rebound (retry-safe).
    rebound_lock = ils.read_lock_file(result["lock_path"])
    result2 = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            rebound_lock,
            lock_dir,
            old_pid=old,
            existing_lock=rebound_lock,
        )
    )
    assert result2["success"], result2
    assert result2["already_rebound"] is True


# ── 6. Active-PID refusal ───────────────────────────────────────────────────


def test_active_pid_refused(dirty_repo, lock_dir):
    live = os.getpid()
    # Use a different "current" identity of session via fake current_pid...
    # Owner is live (this process). Rebind must refuse.
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=live, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=live,
            current_pid=live + 10_000_000,  # distinct "new" session id for pin check
        )
    )
    assert not result["success"]
    assert any("still alive" in r for r in result["reasons"])


# ── 7. Foreign claimant refusal ─────────────────────────────────────────────


def test_foreign_claimant_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            current_identity="someone-else",
            claimant_identity="someone-else",
        )
    )
    assert not result["success"]
    assert any("foreign claimant" in r or "does not match" in r for r in result["reasons"])


# ── 8. Profile mismatch refusal ─────────────────────────────────────────────


def test_profile_mismatch_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            current_profile="other-profile",
            claimant_profile="other-profile",
        )
    )
    assert not result["success"]
    assert any("profile" in r for r in result["reasons"])


# ── 9. Competing session/lock/lease refusal ─────────────────────────────────


def test_competing_live_lock_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    competing = [
        {
            "issue_number": ISSUE,
            "branch_name": BRANCH,
            "worktree_path": dirty_repo["worktree"] + "-other",
            "pid": os.getpid(),
        }
    ]
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            competing_live_locks=competing,
        )
    )
    assert not result["success"]
    assert any("competing live lock" in r for r in result["reasons"])


def test_competing_session_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            competing_sessions=[
                {"pid": os.getpid(), "lock_file_path": lock["lock_file_path"], "live": True}
            ],
        )
    )
    # current_pid is os.getpid(), so same session is skipped — use another live pid.
    # Spawn a long-lived process to act as competing live session.
    rival = subprocess.Popen([sys.executable, "-c", "import time; time.sleep(30)"])
    try:
        result = rebind.apply_dirty_same_claimant_session_rebind(
            **_apply_kwargs(
                dirty_repo,
                lock,
                lock_dir,
                old_pid=old,
                competing_sessions=[
                    {
                        "pid": rival.pid,
                        "lock_file_path": lock["lock_file_path"],
                        "live": True,
                    }
                ],
            )
        )
        assert not result["success"]
        assert any("competing live session" in r for r in result["reasons"])
    finally:
        rival.kill()
        rival.wait()


def test_workflow_lease_active_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            workflow_lease_active=True,
        )
    )
    assert not result["success"]
    assert any("workflow lease" in r for r in result["reasons"])


# ── 10. Local- and remote-head movement refusal ─────────────────────────────


def test_local_head_movement_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            expected_local_head="b" * 40,
        )
    )
    assert not result["success"]
    assert any("local head" in r for r in result["reasons"])


def test_remote_head_movement_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            expected_remote_head="c" * 40,
        )
    )
    assert not result["success"]
    assert any("remote head" in r for r in result["reasons"])


# ── 11. Path/symlink/registration mismatches ────────────────────────────────


def test_worktree_not_under_branches_refused(dirty_repo, lock_dir, tmp_path):
    old = dead_pid()
    # Use a path outside branches/ as the declared worktree (still real dir).
    outside = tmp_path / "outside-wt"
    outside.mkdir()
    lock = _make_lock(worktree=str(outside), pid=old, lock_dir=lock_dir)
    # Inventory empty for outside path; use empty pins to hit path gate first
    # by providing matching empty-ish inventory after we force path checks.
    inv = {
        "dirty_paths": dirty_repo["dirty_paths"],
        "fingerprints": dirty_repo["fingerprints"],
        "ok": True,
        "reasons": [],
    }
    result = rebind.assess_dirty_same_claimant_session_rebind(
        remote=REMOTE,
        org=ORG,
        repo=REPO,
        issue_number=ISSUE,
        branch_name=BRANCH,
        worktree_path=str(outside),
        claimant_identity=IDENTITY,
        claimant_profile=PROFILE,
        old_pid=old,
        expected_local_head=dirty_repo["local_head"],
        expected_remote_head=dirty_repo["remote_head"],
        expected_dirty_paths=dirty_repo["dirty_paths"],
        expected_fingerprints=dirty_repo["fingerprints"],
        existing_lock=lock,
        current_identity=IDENTITY,
        current_profile=PROFILE,
        role_kind="author",
        current_pid=os.getpid(),
        current_branch=BRANCH,
        local_head=dirty_repo["local_head"],
        remote_head=dirty_repo["remote_head"],
        dirty_inventory=inv,
        repo_root=dirty_repo["root"],
    )
    assert not result["rebind_sanctioned"]
    assert any("branches/" in r for r in result["reasons"])


def test_lock_worktree_mismatch_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(
        worktree=dirty_repo["worktree"] + "-elsewhere",
        pid=old,
        lock_dir=lock_dir,
    )
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert not result["success"]
    assert any("does not match declared" in r for r in result["reasons"])


# ── 12. Malformed lock/session records ──────────────────────────────────────


def test_malformed_lock_missing_pid_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    lock.pop("session_pid", None)
    lock.pop("pid", None)
    ils.save_lock_file(lock["lock_file_path"], lock)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert not result["success"]
    assert any("incomplete" in r or "session_pid" in r for r in result["reasons"])


def test_empty_old_pid_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=None)
    )
    assert not result["success"]
    assert any("old_pid" in r for r in result["reasons"])


def test_reviewer_role_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, role_kind="reviewer")
    )
    assert not result["success"]
    assert any("reviewer" in r for r in result["reasons"])


def test_reconciler_without_authorize_refused(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(
            dirty_repo,
            lock,
            lock_dir,
            old_pid=old,
            role_kind="reconciler",
            authorize_reconciler_execute=False,
        )
    )
    assert not result["success"]
    assert any("authorize_reconciler_execute" in r for r in result["reasons"])


# ── 13. No duplicate ownership after success or retry ───────────────────────


def test_no_duplicate_ownership_after_success_or_retry(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    r1 = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old)
    )
    assert r1["success"], r1
    rebound = ils.read_lock_file(r1["lock_path"])
    r2 = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, rebound, lock_dir, old_pid=old, existing_lock=rebound)
    )
    assert r2["success"], r2
    assert r2["already_rebound"] is True

    # Only one durable lock file for this issue; session pointer is current pid.
    # Skip session pointers and rebind journals (dotfiles / non-lock records).
    matching = []
    for p in ils.iter_lock_files(lock_dir):
        name = os.path.basename(p)
        if name.startswith(".") or name.startswith("session-"):
            continue
        rec = ils.read_lock_file(p)
        if not rec:
            continue
        if (
            rec.get("issue_number") == ISSUE
            and rec.get("remote") == REMOTE
            and rec.get("branch_name") == BRANCH
            and rec.get("session_pid") is not None
        ):
            matching.append(rec)
    assert len(matching) == 1
    assert int(matching[0]["session_pid"]) == os.getpid()
    # No live session pointer for the dead old pid.
    assert not os.path.exists(os.path.join(lock_dir, f"session-{old}.json"))


# ── 14. Ordinary dirty-worktree locking remains fail-closed ─────────────────


def test_ordinary_dirty_lock_worktree_assessment_blocks(dirty_repo):
    porcelain = dirty_repo["inventory"]["porcelain_status"]
    assessment = issue_lock_worktree.assess_issue_lock_worktree(
        worktree_path=dirty_repo["worktree"],
        current_branch=BRANCH,
        porcelain_status=porcelain,
        base_equivalent=False,
    )
    assert assessment["block"] is True
    assert any(
        "tracked file edits exist before issue lock" in r
        for r in assessment["reasons"]
    )


# ── 15. Fixture matching #860 class with 7 fingerprint-pinned dirty paths ───


def test_issue_860_regression_fixture_spec():
    spec = rebind.build_issue_860_regression_fixture_spec()
    assert spec["claimant_identity"] == "jcwalker3"
    assert spec["claimant_profile"] == "prgs-author"
    assert spec["old_pid_alive"] is False
    assert spec["live_session_pointer"] is None
    assert spec["dirty_path_count"] == 7
    assert len(spec["expected_dirty_paths"]) == 7
    assert len(spec["expected_fingerprints"]) == 7
    assert spec["expected_local_head"] == spec["expected_remote_head"]
    for path in spec["expected_dirty_paths"]:
        assert path in spec["expected_fingerprints"]
        assert len(spec["expected_fingerprints"][path]) == 64


def test_dry_run_does_not_write(dirty_repo, lock_dir):
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    before = ils.read_lock_file(lock["lock_file_path"])
    result = rebind.apply_dirty_same_claimant_session_rebind(
        **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dry_run=True)
    )
    assert result["success"], result
    assert result["dry_run"] is True
    after = ils.read_lock_file(lock["lock_file_path"])
    assert after["session_pid"] == before["session_pid"]
    assert not os.path.exists(os.path.join(lock_dir, f"session-{os.getpid()}.json"))


def test_provenance_source_is_sanctioned():
    assert (
        issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND
        in issue_lock_provenance.SANCTIONED_LOCK_SOURCES
    )
    assessment = issue_lock_provenance.assess_lock_file_for_create_pr(
        {
            "work_lease": {"operation_type": "author_issue_work"},
            "lock_provenance": {
                "source": issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND,
                "written_by_tool": issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND,
                "written_at": "2026-01-01T00:00:00Z",
            },
        }
    )
    assert assessment["proven"] is True


def test_permission_allowed_is_not_ownership_proof(dirty_repo, lock_dir):
    """permission_allowed=True must not bypass foreign claimant refusal."""
    old = dead_pid()
    lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir)
    result = rebind.assess_dirty_same_claimant_session_rebind(
        remote=REMOTE,
        org=ORG,
        repo=REPO,
        issue_number=ISSUE,
        branch_name=BRANCH,
        worktree_path=dirty_repo["worktree"],
        claimant_identity=IDENTITY,
        claimant_profile=PROFILE,
        old_pid=old,
        expected_local_head=dirty_repo["local_head"],
        expected_remote_head=dirty_repo["remote_head"],
        expected_dirty_paths=dirty_repo["dirty_paths"],
        expected_fingerprints=dirty_repo["fingerprints"],
        existing_lock=lock,
        current_identity="intruder",
        current_profile=PROFILE,
        role_kind="author",
        current_pid=os.getpid(),
        current_branch=BRANCH,
        local_head=dirty_repo["local_head"],
        remote_head=dirty_repo["remote_head"],
        dirty_inventory=dirty_repo["inventory"],
        permission_allowed=True,
        repo_root=dirty_repo["root"],
    )
    assert not result["rebind_sanctioned"]
    assert any("does not match active identity" in r for r in result["reasons"])


def test_content_fingerprint_stable(tmp_path):
    p = tmp_path / "f.txt"
    p.write_bytes(b"abc123")
    a = rebind.content_fingerprint(str(p))
    b = rebind.content_fingerprint(str(p))
    assert a == b
    assert len(a) == 64