dirty_same_claimant_session_rebind.py

@@ -1,4 +1,4 @@
-"""Dirty-preserving same-claimant author-session rebind (#864).
+"""Dirty-preserving same-claimant author-session rebind (#864 / #868).
 
 A registered issue worktree can be dirty while its durable lock owner PID is
 provably dead. Ordinary ``gitea_lock_issue`` refuses dirty trees, and dead-session
@@ -14,6 +14,14 @@ This operation:
   heartbeat)
 * preserves every tracked/untracked byte
 * does NOT sync remote, create recovery worktrees, clean, reset, or change heads
+
+#868 hardens:
+
+* complete dirty-inventory revalidation (full path set + fingerprints)
+  immediately before and after ``bind_session_lock``
+* durable recovery-journal operation identity (remote, org, repo, claimant
+  identity, claimant profile) validated on execute / resume / retry /
+  already_rebound
 """
 
 from __future__ import annotations
@@ -66,6 +74,16 @@ REQUIRED_LOCK_FIELDS = (
     "repo",
 )
 
+# Durable journal operation identity (#868 F2). All five must be persisted on
+# JOURNAL_PHASE_ASSESSED and re-validated on resume / retry / already_rebound.
+REQUIRED_JOURNAL_IDENTITY_FIELDS = (
+    "remote",
+    "org",
+    "repo",
+    "claimant_identity",
+    "claimant_profile",
+)
+
 
 def _utc_now_iso() -> str:
     return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
@@ -196,6 +214,174 @@ def collect_dirty_inventory(worktree_path: str) -> dict[str, Any]:
     }
 
 
+def revalidate_complete_dirty_inventory(
+    worktree_path: str,
+    *,
+    expected_dirty_paths: Sequence[str] | None,
+    expected_fingerprints: Mapping[str, str] | None,
+    phase: str = "inventory",
+) -> dict[str, Any]:
+    """Collect the full dirty inventory and require exact pin equality (#868 F1).
+
+    Unlike fingerprint-only checks over the expected path list, this recollects
+    the authoritative tracked+untracked inventory and refuses added, removed,
+    or renamed paths as well as fingerprint movement.
+    """
+    reasons: list[str] = []
+    inv = collect_dirty_inventory(worktree_path)
+    if inv.get("ok") is False:
+        reasons.extend(list(inv.get("reasons") or []) or [f"{phase}: dirty inventory collection failed"])
+
+    observed_paths = sorted(
+        {_text(p) for p in (inv.get("dirty_paths") or []) if _text(p)}
+    )
+    pin_paths = sorted(
+        {_text(p) for p in (expected_dirty_paths or []) if _text(p)}
+    )
+    if not pin_paths:
+        reasons.append(
+            f"{phase}: expected_dirty_paths pin is empty; complete inventory "
+            "revalidation requires a non-empty pin (fail closed)"
+        )
+    if set(observed_paths) != set(pin_paths):
+        extra = sorted(set(observed_paths) - set(pin_paths))
+        missing = sorted(set(pin_paths) - set(observed_paths))
+        if extra:
+            reasons.append(
+                f"{phase}: complete dirty inventory path-set disagreement: "
+                f"unexpected paths {extra}"
+            )
+        if missing:
+            reasons.append(
+                f"{phase}: complete dirty inventory path-set disagreement: "
+                f"missing expected paths {missing}"
+            )
+
+    obs_fps = {
+        _text(k): _text(v)
+        for k, v in dict(inv.get("fingerprints") or {}).items()
+        if _text(k)
+    }
+    pin_fps = {
+        _text(k): _text(v)
+        for k, v in dict(expected_fingerprints or {}).items()
+        if _text(k)
+    }
+    if not pin_fps:
+        reasons.append(
+            f"{phase}: expected_fingerprints pin is empty; byte-level pins "
+            "are required (fail closed)"
+        )
+    else:
+        for rel, expected_hash in pin_fps.items():
+            if rel not in set(pin_paths):
+                reasons.append(
+                    f"{phase}: expected_fingerprints contains '{rel}' which is "
+                    "not in expected_dirty_paths"
+                )
+                continue
+            actual_hash = obs_fps.get(rel)
+            if not actual_hash:
+                reasons.append(
+                    f"{phase}: fingerprint missing for dirty path '{rel}'"
+                )
+            elif actual_hash != expected_hash:
+                reasons.append(
+                    f"{phase}: fingerprint disagreement for '{rel}': "
+                    f"observed {actual_hash}, expected {expected_hash}"
+                )
+        for rel in observed_paths:
+            if rel not in pin_fps:
+                reasons.append(
+                    f"{phase}: observed dirty path '{rel}' has no fingerprint pin"
+                )
+
+    return {
+        "ok": not reasons,
+        "reasons": reasons,
+        "inventory": inv,
+        "observed_dirty_paths": observed_paths,
+        "expected_dirty_paths": pin_paths,
+        "observed_fingerprints": obs_fps,
+        "expected_fingerprints": pin_fps,
+        "phase": phase,
+    }
+
+
+def build_journal_operation_identity(
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    claimant_identity: str | None,
+    claimant_profile: str | None,
+) -> dict[str, str]:
+    """Return the five-field durable operation identity for the recovery journal."""
+    return {
+        "remote": _text(remote),
+        "org": _text(org),
+        "repo": _text(repo),
+        "claimant_identity": _text(claimant_identity),
+        "claimant_profile": _text(claimant_profile),
+    }
+
+
+def validate_journal_operation_identity(
+    journal: Mapping[str, Any] | None,
+    *,
+    remote: str,
+    org: str,
+    repo: str,
+    claimant_identity: str | None,
+    claimant_profile: str | None,
+    require_present: bool = True,
+) -> list[str]:
+    """Validate durable journal identity fields (#868 F2).
+
+    Rejects missing, mismatched, stale, cross-repository, or cross-claimant
+    journal state. When *require_present* is True, incomplete legacy journals
+    (any of the five fields absent/empty) fail closed.
+    """
+    reasons: list[str] = []
+    if not isinstance(journal, Mapping):
+        if require_present:
+            reasons.append(
+                "recovery journal is missing or unreadable; complete operation "
+                "identity cannot be proven (fail closed)"
+            )
+        return reasons
+
+    expected = build_journal_operation_identity(
+        remote=remote,
+        org=org,
+        repo=repo,
+        claimant_identity=claimant_identity,
+        claimant_profile=claimant_profile,
+    )
+    for field in REQUIRED_JOURNAL_IDENTITY_FIELDS:
+        observed = _text(journal.get(field))
+        want = expected[field]
+        if not observed:
+            reasons.append(
+                f"recovery journal omits operation identity field '{field}' "
+                "(incomplete legacy or malformed journal identity; fail closed)"
+            )
+            continue
+        if not want:
+            reasons.append(
+                f"caller pin for journal identity field '{field}' is empty "
+                "(fail closed)"
+            )
+            continue
+        if observed != want:
+            reasons.append(
+                f"recovery journal identity mismatch for '{field}': "
+                f"journal={observed!r}, expected={want!r} "
+                "(cross-repository / cross-claimant / replay refused)"
+            )
+    return reasons
+
+
 def journal_path(lock_dir: str, issue_number: int) -> str:
     root = (lock_dir or "").strip()
     return os.path.join(root, f".rebind-journal-{int(issue_number)}.json")
@@ -789,10 +975,22 @@ def _already_rebound(
     existing_lock: Mapping[str, Any],
     current_pid: int,
     worktree_path: str,
+    expected_dirty_paths: Sequence[str] | None,
     expected_fingerprints: Mapping[str, str],
     worktree_for_fps: str,
+    remote: str,
+    org: str,
+    repo: str,
+    claimant_identity: str | None,
+    claimant_profile: str | None,
+    journal: Mapping[str, Any] | None = None,
 ) -> tuple[bool, list[str]]:
-    """Return (True, notes) when lock is already rebound to this session."""
+    """Return (True, notes) when lock is already rebound to this session.
+
+    #868: require complete matching operation identity (remote/org/repo/
+    claimant) and complete dirty-inventory revalidation, not fingerprint-only
+    checks. Incomplete or mismatched journal identity fails closed.
+    """
     notes: list[str] = []
     pid = _recorded_pid(existing_lock)
     try:
@@ -803,19 +1001,61 @@ def _already_rebound(
         return False, []
     if not _same_realpath(_text(existing_lock.get("worktree_path")), worktree_path):
         return False, []
-    # Fingerprints must still match pins (byte preservation).
+
-    for rel, expected in (expected_fingerprints or {}).items():
+    # Durable lock repo binding must still match the caller's target.
-        abs_path = os.path.join(worktree_for_fps, rel)
+    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
-        try:
+        observed = _text(existing_lock.get(field))
-            actual = content_fingerprint(abs_path)
+        want = _text(expected)
-        except OSError as exc:
+        if observed and want and observed != want:
-            notes.append(f"could not re-fingerprint '{rel}' for already_rebound: {exc}")
-            return False, notes
-        if actual != _text(expected):
             notes.append(
-                f"fingerprint drift on already-rebound check for '{rel}'"
+                f"already_rebound refused: lock {field}={observed!r} does not "
+                f"match expected {want!r} (cross-repository replay)"
             )
             return False, notes
+
+    lock_claimant = _lock_claimant(existing_lock)
+    locked_identity = _text(lock_claimant.get("username"))
+    locked_profile = _text(lock_claimant.get("profile"))
+    pin_identity = _text(claimant_identity)
+    pin_profile = _text(claimant_profile)
+    if pin_identity and locked_identity and pin_identity != locked_identity:
+        notes.append(
+            f"already_rebound refused: lock claimant '{locked_identity}' does "
+            f"not match pin '{pin_identity}' (cross-claimant replay)"
+        )
+        return False, notes
+    if pin_profile and locked_profile and pin_profile != locked_profile:
+        notes.append(
+            f"already_rebound refused: lock profile '{locked_profile}' does "
+            f"not match pin '{pin_profile}' (cross-claimant replay)"
+        )
+        return False, notes
+
+    # When a durable journal is present, require complete matching identity.
+    if isinstance(journal, Mapping) and journal:
+        id_reasons = validate_journal_operation_identity(
+            journal,
+            remote=remote,
+            org=org,
+            repo=repo,
+            claimant_identity=claimant_identity,
+            claimant_profile=claimant_profile,
+            require_present=True,
+        )
+        if id_reasons:
+            notes.extend(id_reasons)
+            return False, notes
+
+    inv_check = revalidate_complete_dirty_inventory(
+        worktree_for_fps,
+        expected_dirty_paths=expected_dirty_paths,
+        expected_fingerprints=expected_fingerprints,
+        phase="already_rebound",
+    )
+    if not inv_check["ok"]:
+        notes.extend(list(inv_check["reasons"] or []))
+        return False, notes
+
     gen = lock_generation(existing_lock)
     if gen < 1:
         # A never-written generation is suspicious for a completed rebind, but
@@ -917,6 +1157,53 @@ def apply_dirty_same_claimant_session_rebind(
         "remote_head": remote_head,
     }
 
+    root_for_journal = (lock_dir or "").strip() or None
+    if root_for_journal is None and isinstance(existing_lock, Mapping):
+        root_for_journal = os.path.dirname(
+            _text(existing_lock.get("lock_file_path"))
+            or lock_file_path(
+                remote=remote, org=org, repo=repo, issue_number=issue_number
+            )
+        )
+    jpath_probe = (
+        journal_path(root_for_journal, issue_number) if root_for_journal else ""
+    )
+    existing_journal = _read_json(jpath_probe) if jpath_probe else None
+
+    # Resume / retry: reject incomplete, mismatched, or cross-repo journal
+    # identity before treating any prior journal as authoritative (#868 F2).
+    if isinstance(existing_journal, Mapping) and existing_journal:
+        journal_id_reasons = validate_journal_operation_identity(
+            existing_journal,
+            remote=remote,
+            org=org,
+            repo=repo,
+            claimant_identity=claimant_identity,
+            claimant_profile=claimant_profile,
+            require_present=True,
+        )
+        # Incomplete legacy journals from pre-#868 apply paths must fail closed
+        # when any identity field is missing — even if the rest of the payload
+        # looks familiar. Only a complete matching identity may proceed.
+        phase = _text(existing_journal.get("phase"))
+        if journal_id_reasons and phase not in ("", JOURNAL_PHASE_COMPLETE):
+            # Allow a completed journal with missing legacy identity only when
+            # already_rebound path will re-validate lock + inventory; for
+            # mid-flight incomplete journals, refuse.
+            if phase in (
+                JOURNAL_PHASE_ASSESSED,
+                JOURNAL_PHASE_PRE_BIND,
+                JOURNAL_PHASE_BOUND,
+                "bind_failed",
+            ):
+                return {
+                    **base_result,
+                    "success": False,
+                    "reasons": journal_id_reasons,
+                    "journal_path": jpath_probe,
+                    "journal_phase": phase or None,
+                }
+
     # Retry-safe: if already rebound to this session, succeed even when assess
     # refuses because old_pid no longer matches the (updated) lock.
     if (
@@ -928,8 +1215,15 @@ def apply_dirty_same_claimant_session_rebind(
             existing_lock=existing_lock,
             current_pid=pid_now,
             worktree_path=worktree_path,
+            expected_dirty_paths=expected_dirty_paths,
             expected_fingerprints=expected_fingerprints,
             worktree_for_fps=wt,
+            remote=remote,
+            org=org,
+            repo=repo,
+            claimant_identity=claimant_identity,
+            claimant_profile=claimant_profile,
+            journal=existing_journal,
         )
         if done:
             lock_path = _text(existing_lock.get("lock_file_path")) or lock_file_path(
@@ -956,6 +1250,20 @@ def apply_dirty_same_claimant_session_rebind(
                 "generation_after": lock_generation(existing_lock),
                 "journal_phase": JOURNAL_PHASE_ALREADY_REBOUND,
             }
+        # Same-pid candidate that failed complete identity/inventory checks
+        # must not fall through into a fresh bind that would re-mint authority.
+        if _recorded_pid(existing_lock) is not None:
+            try:
+                if int(_recorded_pid(existing_lock)) == int(pid_now) and notes:
+                    return {
+                        **base_result,
+                        "success": False,
+                        "already_rebound": False,
+                        "reasons": notes,
+                        "journal_path": jpath_probe or None,
+                    }
+            except (TypeError, ValueError):
+                pass
 
     if not assessment["rebind_sanctioned"]:
         return base_result
@@ -982,6 +1290,25 @@ def apply_dirty_same_claimant_session_rebind(
         issue_number,
     )
 
+    op_identity = build_journal_operation_identity(
+        remote=remote,
+        org=org,
+        repo=repo,
+        claimant_identity=claimant_identity,
+        claimant_profile=claimant_profile,
+    )
+    # Refuse incomplete caller identity before any durable write.
+    for field, value in op_identity.items():
+        if not value:
+            return {
+                **base_result,
+                "success": False,
+                "reasons": [
+                    f"cannot write recovery journal: operation identity field "
+                    f"'{field}' is empty (fail closed)"
+                ],
+            }
+
     journal = {
         "phase": JOURNAL_PHASE_ASSESSED,
         "issue_number": issue_number,
@@ -996,36 +1323,40 @@ def apply_dirty_same_claimant_session_rebind(
         "remote_head": remote_head,
         "started_at": _utc_now_iso(),
         "source": SOURCE,
+        # #868 F2 — complete durable operation identity
+        **op_identity,
     }
     _atomic_write_json(jpath, journal)
 
-    # Immediate pre-bind fingerprint re-verification.
+    # #868 F1 — complete dirty-inventory revalidation immediately before mutation.
-    pre_fps: dict[str, str] = {}
+    # Fail closed with no bind so failures cannot leave a newly authoritative
-    for rel in expected_dirty_paths or []:
+    # live session.
-        abs_path = os.path.join(wt, rel)
+    pre_inv = revalidate_complete_dirty_inventory(
-        try:
+        wt,
-            pre_fps[rel] = content_fingerprint(abs_path)
+        expected_dirty_paths=expected_dirty_paths,
-        except OSError as exc:
+        expected_fingerprints=expected_fingerprints,
-            return {
+        phase="pre-bind",
-                **base_result,
+    )
-                "success": False,
+    if not pre_inv["ok"]:
-                "reasons": [f"pre-bind fingerprint failed for '{rel}': {exc}"],
+        journal["phase"] = "pre_bind_inventory_failed"
-                "journal_path": jpath,
+        journal["pre_bind_inventory"] = {
-            }
+            "observed_dirty_paths": pre_inv.get("observed_dirty_paths"),
-    for rel, expected in (expected_fingerprints or {}).items():
+            "reasons": pre_inv.get("reasons"),
-        if pre_fps.get(rel) != _text(expected):
+        }
-            return {
+        _atomic_write_json(jpath, journal)
-                **base_result,
+        return {
-                "success": False,
+            **base_result,
-                "reasons": [
+            "success": False,
-                    f"pre-bind fingerprint drift for '{rel}': "
+            "reasons": list(pre_inv["reasons"] or []),
-                    f"observed {pre_fps.get(rel)}, expected {expected}"
+            "journal_path": jpath,
-                ],
+            "journal_phase": "pre_bind_inventory_failed",
-                "journal_path": jpath,
+        }
-            }
+
+    pre_fps = dict(pre_inv.get("observed_fingerprints") or {})
 
     journal["phase"] = JOURNAL_PHASE_PRE_BIND
     journal["pre_bind_fingerprints"] = pre_fps
+    journal["pre_bind_dirty_paths"] = list(pre_inv.get("observed_dirty_paths") or [])
     _atomic_write_json(jpath, journal)
 
     now = _utc_now_iso()
@@ -1095,38 +1426,37 @@ def apply_dirty_same_claimant_session_rebind(
     journal["lock_path"] = lock_path
     _atomic_write_json(jpath, journal)
 
-    # Post-bind fingerprint verification — every byte unchanged.
+    # #868 F1 — complete dirty-inventory revalidation immediately after mutation.
-    post_fps: dict[str, str] = {}
+    # Path set must remain exactly equal; fingerprints must be unchanged.
-    for rel in expected_dirty_paths or []:
+    post_inv = revalidate_complete_dirty_inventory(
-        abs_path = os.path.join(wt, rel)
+        wt,
-        try:
+        expected_dirty_paths=expected_dirty_paths,
-            post_fps[rel] = content_fingerprint(abs_path)
+        expected_fingerprints=expected_fingerprints,
-        except OSError as exc:
+        phase="post-bind",
-            return {
+    )
-                **base_result,
+    post_fps = dict(post_inv.get("observed_fingerprints") or {})
-                "success": False,
+    if not post_inv["ok"]:
-                "reasons": [
+        journal["phase"] = "post_bind_inventory_failed"
-                    f"post-bind fingerprint failed for '{rel}': {exc}; "
+        journal["post_bind_inventory"] = {
-                    "lock may be rebound but content verification failed"
+            "observed_dirty_paths": post_inv.get("observed_dirty_paths"),
-                ],
+            "reasons": post_inv.get("reasons"),
-                "lock_path": lock_path,
+        }
-                "journal_path": jpath,
+        journal["post_bind_fingerprints"] = post_fps
-                "generation_before": gen_before,
+        _atomic_write_json(jpath, journal)
-            }
+        return {
-    for rel, expected in (expected_fingerprints or {}).items():
+            **base_result,
-        if post_fps.get(rel) != _text(expected):
+            "success": False,
-            return {
+            "reasons": list(post_inv["reasons"] or []) + [
-                **base_result,
+                "post-bind complete inventory revalidation failed after "
-                "success": False,
+                "bind_session_lock; lock may be rebound but content/path "
-                "reasons": [
+                "verification failed (fail closed)"
-                    f"post-bind fingerprint drift for '{rel}': "
+            ],
-                    f"observed {post_fps.get(rel)}, expected {expected}"
+            "lock_path": lock_path,
-                ],
+            "journal_path": jpath,
-                "lock_path": lock_path,
+            "journal_phase": "post_bind_inventory_failed",
-                "journal_path": jpath,
+            "generation_before": gen_before,
-                "generation_before": gen_before,
+            "fingerprints_after": post_fps,
-                "fingerprints_after": post_fps,
+        }
-            }
 
     # Remove stale session pointer for old_pid when it points at this lock.
     removed_old_pointer = False
@@ -1154,6 +1484,9 @@ def apply_dirty_same_claimant_session_rebind(
     journal["generation_after"] = gen_after
     journal["removed_old_session_pointer"] = removed_old_pointer
     journal["post_bind_fingerprints"] = post_fps
+    journal["post_bind_dirty_paths"] = list(
+        post_inv.get("observed_dirty_paths") or []
+    )
     _atomic_write_json(jpath, journal)
 
     return {

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)