Merge branch 'master' into fix/issue-855-pr-scoped-merged-cleanup

tests/test_dirty_same_claimant_session_rebind.py

@@ -1,7 +1,10 @@
-"""Integration tests for dirty same-claimant author-session rebind (#864).
+"""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 worktree on disk.
+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
@@ -371,6 +374,7 @@ 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,
         {
@@ -379,6 +383,12 @@ def test_retry_after_journal_mid_state(dirty_repo, lock_dir):
             "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(
@@ -843,3 +853,494 @@ def test_content_fingerprint_stable(tmp_path):
     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)