"""Integration tests for dirty same-claimant author-session rebind (#864 / #868). Uses real temp git repos/worktrees and a temp GITEA_ISSUE_LOCK_DIR. Does not mutate any real #860/#864/#868 worktree on disk. #868 adds complete dirty-inventory revalidation around bind_session_lock and complete recovery-journal operation identity (remote/org/repo/claimant). """ 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/ 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) # Complete operation identity required for mid-flight resume (#868 F2). rebind._atomic_write_json( jpath, { "phase": rebind.JOURNAL_PHASE_PRE_BIND, "issue_number": ISSUE, "old_pid": old, "new_pid": os.getpid(), "expected_generation": 1, "remote": REMOTE, "org": ORG, "repo": REPO, "claimant_identity": IDENTITY, "claimant_profile": PROFILE, "source": rebind.SOURCE, }, ) 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 # ── #868 F1 — Complete dirty-inventory revalidation ───────────────────────── def test_extra_tracked_dirty_path_before_bind_refused(dirty_repo, lock_dir): """Extra tracked dirty path appearing immediately before binding fails closed.""" old = dead_pid() lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) wt = dirty_repo["worktree"] # Seed a tracked file, then dirty it without including it in the pin set. tracked_extra = "tracked_extra_before_bind.txt" path = Path(wt) / tracked_extra path.write_text("seed tracked extra\n", encoding="utf-8") _git(wt, "add", tracked_extra) _git(wt, "commit", "-q", "-m", "seed extra tracked") # Heads moved — re-pin heads so only inventory disagreement is tested. head = _git(wt, "rev-parse", "HEAD").stdout.strip() _git(wt, "push", "-q", "origin", BRANCH) remote_head = _git(wt, "rev-parse", f"refs/remotes/origin/{BRANCH}").stdout.strip() path.write_text("dirty tracked extra\n", encoding="utf-8") # Pins still describe the original inventory (without tracked_extra). result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs( dirty_repo, lock, lock_dir, old_pid=old, expected_local_head=head, expected_remote_head=remote_head, local_head=head, remote_head=remote_head, dirty_inventory=None, # force live recollect in apply ) ) assert not result["success"] joined = " ".join(result["reasons"]) assert "unexpected paths" in joined or "path-set disagreement" in joined # Must not leave a newly authoritative live session for this pid. rebound = ils.read_lock_file(lock["lock_file_path"]) assert rebound is not None assert int(rebound.get("session_pid") or 0) == old def test_extra_untracked_path_before_bind_refused(dirty_repo, lock_dir): """Extra untracked path appearing immediately before binding fails closed.""" old = dead_pid() lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) wt = dirty_repo["worktree"] extra = Path(wt) / "surprise_untracked_before_bind.txt" extra.write_text("sneaky\n", encoding="utf-8") result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs( dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None, ) ) assert not result["success"] joined = " ".join(result["reasons"]) assert "unexpected paths" in joined or "path-set disagreement" in joined rebound = ils.read_lock_file(lock["lock_file_path"]) assert int(rebound.get("session_pid") or 0) == old def test_path_added_during_mutation_window_refused(dirty_repo, lock_dir, monkeypatch): """Path added during the mutation window is detected by post-bind inventory.""" old = dead_pid() lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) wt = dirty_repo["worktree"] real_bind = ils.bind_session_lock def _bind_then_add_path(lock_payload, **kwargs): path = real_bind(lock_payload, **kwargs) surprise = Path(wt) / "added_during_bind.txt" surprise.write_text("during bind\n", encoding="utf-8") return path monkeypatch.setattr(rebind, "bind_session_lock", _bind_then_add_path) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None) ) assert not result["success"] joined = " ".join(result["reasons"]) assert "post-bind" in joined assert "unexpected paths" in joined or "path-set disagreement" in joined assert result.get("journal_phase") == "post_bind_inventory_failed" def test_path_removed_during_mutation_window_refused(dirty_repo, lock_dir, monkeypatch): """Path removed during the mutation window is detected by post-bind inventory.""" old = dead_pid() lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) wt = dirty_repo["worktree"] victim = dirty_repo["dirty_paths"][-1] # prefer untracked for easy remove real_bind = ils.bind_session_lock def _bind_then_remove_path(lock_payload, **kwargs): path = real_bind(lock_payload, **kwargs) target = Path(wt) / victim if target.exists(): target.unlink() return path monkeypatch.setattr(rebind, "bind_session_lock", _bind_then_remove_path) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None) ) assert not result["success"] joined = " ".join(result["reasons"]) assert "post-bind" in joined assert "missing expected" in joined or "path-set disagreement" in joined def test_fingerprint_movement_unchanged_path_set_refused(dirty_repo, lock_dir, monkeypatch): """Fingerprint movement with unchanged path set fails pre- or post-bind check.""" old = dead_pid() lock = _make_lock(worktree=dirty_repo["worktree"], pid=old, lock_dir=lock_dir) wt = dirty_repo["worktree"] victim = dirty_repo["dirty_paths"][0] real_bind = ils.bind_session_lock def _bind_then_mutate_bytes(lock_payload, **kwargs): path = real_bind(lock_payload, **kwargs) target = Path(wt) / victim target.write_text(target.read_text(encoding="utf-8") + "mutated\n", encoding="utf-8") return path monkeypatch.setattr(rebind, "bind_session_lock", _bind_then_mutate_bytes) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old, dirty_inventory=None) ) assert not result["success"] joined = " ".join(result["reasons"]) assert "fingerprint" in joined assert "post-bind" in joined # ── #868 F2 — Complete recovery-journal identity ──────────────────────────── def _complete_journal(**overrides): base = { "phase": rebind.JOURNAL_PHASE_ASSESSED, "issue_number": ISSUE, "branch_name": BRANCH, "worktree_path": "/tmp/wt", "old_pid": 1, "new_pid": os.getpid(), "remote": REMOTE, "org": ORG, "repo": REPO, "claimant_identity": IDENTITY, "claimant_profile": PROFILE, "source": rebind.SOURCE, } base.update(overrides) return base def test_journal_remote_mismatch_refused(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, _complete_journal( phase=rebind.JOURNAL_PHASE_PRE_BIND, old_pid=old, remote="dadeschools", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("remote" in r and "mismatch" in r for r in result["reasons"]) def test_journal_org_mismatch_refused(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, _complete_journal( phase=rebind.JOURNAL_PHASE_PRE_BIND, old_pid=old, org="Other-Org", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("org" in r and "mismatch" in r for r in result["reasons"]) def test_journal_repo_mismatch_refused(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, _complete_journal( phase=rebind.JOURNAL_PHASE_PRE_BIND, old_pid=old, repo="Other-Repo", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("repo" in r and "mismatch" in r for r in result["reasons"]) def test_journal_claimant_identity_mismatch_refused(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, _complete_journal( phase=rebind.JOURNAL_PHASE_PRE_BIND, old_pid=old, claimant_identity="intruder", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("claimant_identity" in r for r in result["reasons"]) def test_journal_claimant_profile_mismatch_refused(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, _complete_journal( phase=rebind.JOURNAL_PHASE_PRE_BIND, old_pid=old, claimant_profile="prgs-reviewer", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("claimant_profile" in r for r in result["reasons"]) def test_incomplete_legacy_journal_identity_refused(dirty_repo, lock_dir): """Pre-#868 journals missing the five identity fields fail closed mid-flight.""" 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(), # deliberately omit remote/org/repo/claimant_* }, ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("omits operation identity" in r for r in result["reasons"]) def test_journal_replay_cross_repository_refused(dirty_repo, lock_dir): """Replaying a journal from another repository is refused.""" 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, _complete_journal( phase=rebind.JOURNAL_PHASE_ASSESSED, old_pid=old, remote="dadeschools", org="Other-Org", repo="Other-Repo", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("replay" in r or "mismatch" in r for r in result["reasons"]) def test_journal_replay_cross_claimant_refused(dirty_repo, lock_dir): """Replaying a journal from another claimant is refused.""" 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, _complete_journal( phase=rebind.JOURNAL_PHASE_ASSESSED, old_pid=old, claimant_identity="other-user", claimant_profile="other-profile", ), ) result = rebind.apply_dirty_same_claimant_session_rebind( **_apply_kwargs(dirty_repo, lock, lock_dir, old_pid=old) ) assert not result["success"] assert any("claimant" in r for r in result["reasons"]) def test_successful_exact_retry_with_complete_identity(dirty_repo, lock_dir): """Exact retry after success is already_rebound with complete matching identity.""" 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 # Journal must persist the five identity fields. jpath = rebind.journal_path(lock_dir, ISSUE) journal = rebind._read_json(jpath) assert journal is not None assert journal["phase"] == rebind.JOURNAL_PHASE_COMPLETE for field in rebind.REQUIRED_JOURNAL_IDENTITY_FIELDS: assert journal.get(field), field assert journal["remote"] == REMOTE assert journal["org"] == ORG assert journal["repo"] == REPO assert journal["claimant_identity"] == IDENTITY assert journal["claimant_profile"] == PROFILE 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 def test_already_rebound_requires_complete_matching_identity(dirty_repo, lock_dir): """already_rebound with mismatched journal identity fails closed.""" 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 # Corrupt journal identity after success. jpath = rebind.journal_path(lock_dir, ISSUE) journal = rebind._read_json(jpath) assert journal is not None journal["claimant_identity"] = "not-the-owner" rebind._atomic_write_json(jpath, journal) 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, ) ) # Same pid on lock would look like already_rebound, but identity must match. assert not r2["success"] assert any("claimant_identity" in r or "mismatch" in r for r in r2["reasons"]) def test_ordinary_dirty_worktree_refusal_preserved(dirty_repo): """#868 must not weaken ordinary dirty-worktree refusal on lock_issue path.""" 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 # ── #868 — Reconciler success path ────────────────────────────────────────── def test_reconciler_success_path_tightly_pinned(dirty_repo, lock_dir): """Reconciler with authorize_reconciler_execute=True may execute rebind. Reconciler execution grants no commit/push/publication/review/merge capability — only the tightly pinned session rebind. """ 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=True, # Reconciler may act for the recorded claimant without being that # identity in the active session (still pin-checked against lock). current_identity="sysadmin", current_profile="prgs-reconciler", ) ) assert result["success"], result assert result["outcome"] == rebind.REBIND_SANCTIONED rebound = ils.read_lock_file(result["lock_path"]) assert int(rebound["session_pid"]) == os.getpid() # Provenance records the rebind tool; no publication authority is granted. assert ( rebound.get("lock_provenance", {}).get("source") == issue_lock_provenance.SOURCE_DIRTY_SAME_CLAIMANT_REBIND ) # Reconciler rebind does not stamp commit/push/review/merge capabilities. prov = rebound.get("lock_provenance") or {} blob = json.dumps(prov) for forbidden in ( "gitea.repo.commit", "gitea.branch.push", "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.create", ): assert forbidden not in blob def test_revalidate_complete_dirty_inventory_helper(dirty_repo): ok = rebind.revalidate_complete_dirty_inventory( dirty_repo["worktree"], expected_dirty_paths=dirty_repo["dirty_paths"], expected_fingerprints=dirty_repo["fingerprints"], phase="unit", ) assert ok["ok"] is True bad = rebind.revalidate_complete_dirty_inventory( dirty_repo["worktree"], expected_dirty_paths=dirty_repo["dirty_paths"][:-1], expected_fingerprints={ p: dirty_repo["fingerprints"][p] for p in dirty_repo["dirty_paths"][:-1] }, phase="unit", ) assert bad["ok"] is False assert any("unexpected paths" in r for r in bad["reasons"]) def test_validate_journal_operation_identity_helper(): complete = _complete_journal() assert ( rebind.validate_journal_operation_identity( complete, remote=REMOTE, org=ORG, repo=REPO, claimant_identity=IDENTITY, claimant_profile=PROFILE, ) == [] ) incomplete = {"phase": "assessed", "remote": REMOTE} reasons = rebind.validate_journal_operation_identity( incomplete, remote=REMOTE, org=ORG, repo=REPO, claimant_identity=IDENTITY, claimant_profile=PROFILE, ) assert any("omits operation identity" in r for r in reasons) assert any("org" in r for r in reasons)