Merge pull request 'fix(author): harden dirty-session rebind inventory and journal identity (#868)' (#869) from fix/issue-868-dirty-rebind-inventory-journal into master

Complete dirty-inventory revalidation immediately before and after
bind_session_lock so added, removed, or renamed paths fail closed.
Persist and validate full recovery-journal operation identity (remote,
org, repo, claimant identity, claimant profile) on execute, resume,
retry, and already_rebound. Add focused regression coverage and the
reconciler success-path integration test.

Closes #868.

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():
-        abs_path = os.path.join(worktree_for_fps, rel)
-        try:
-            actual = content_fingerprint(abs_path)
-        except OSError as exc:
-            notes.append(f"could not re-fingerprint '{rel}' for already_rebound: {exc}")
-            return False, notes
-        if actual != _text(expected):
+
+    # Durable lock repo binding must still match the caller's target.
+    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
+        observed = _text(existing_lock.get(field))
+        want = _text(expected)
+        if observed and want and observed != want:
             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.
-    pre_fps: dict[str, str] = {}
-    for rel in expected_dirty_paths or []:
-        abs_path = os.path.join(wt, rel)
-        try:
-            pre_fps[rel] = content_fingerprint(abs_path)
-        except OSError as exc:
-            return {
-                **base_result,
-                "success": False,
-                "reasons": [f"pre-bind fingerprint failed for '{rel}': {exc}"],
-                "journal_path": jpath,
-            }
-    for rel, expected in (expected_fingerprints or {}).items():
-        if pre_fps.get(rel) != _text(expected):
-            return {
-                **base_result,
-                "success": False,
-                "reasons": [
-                    f"pre-bind fingerprint drift for '{rel}': "
-                    f"observed {pre_fps.get(rel)}, expected {expected}"
-                ],
-                "journal_path": jpath,
-            }
+    # #868 F1 — complete dirty-inventory revalidation immediately before mutation.
+    # Fail closed with no bind so failures cannot leave a newly authoritative
+    # live session.
+    pre_inv = revalidate_complete_dirty_inventory(
+        wt,
+        expected_dirty_paths=expected_dirty_paths,
+        expected_fingerprints=expected_fingerprints,
+        phase="pre-bind",
+    )
+    if not pre_inv["ok"]:
+        journal["phase"] = "pre_bind_inventory_failed"
+        journal["pre_bind_inventory"] = {
+            "observed_dirty_paths": pre_inv.get("observed_dirty_paths"),
+            "reasons": pre_inv.get("reasons"),
+        }
+        _atomic_write_json(jpath, journal)
+        return {
+            **base_result,
+            "success": False,
+            "reasons": list(pre_inv["reasons"] or []),
+            "journal_path": jpath,
+            "journal_phase": "pre_bind_inventory_failed",
+        }
+
+    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.
-    post_fps: dict[str, str] = {}
-    for rel in expected_dirty_paths or []:
-        abs_path = os.path.join(wt, rel)
-        try:
-            post_fps[rel] = content_fingerprint(abs_path)
-        except OSError as exc:
-            return {
-                **base_result,
-                "success": False,
-                "reasons": [
-                    f"post-bind fingerprint failed for '{rel}': {exc}; "
-                    "lock may be rebound but content verification failed"
-                ],
-                "lock_path": lock_path,
-                "journal_path": jpath,
-                "generation_before": gen_before,
-            }
-    for rel, expected in (expected_fingerprints or {}).items():
-        if post_fps.get(rel) != _text(expected):
-            return {
-                **base_result,
-                "success": False,
-                "reasons": [
-                    f"post-bind fingerprint drift for '{rel}': "
-                    f"observed {post_fps.get(rel)}, expected {expected}"
-                ],
-                "lock_path": lock_path,
-                "journal_path": jpath,
-                "generation_before": gen_before,
-                "fingerprints_after": post_fps,
-            }
+    # #868 F1 — complete dirty-inventory revalidation immediately after mutation.
+    # Path set must remain exactly equal; fingerprints must be unchanged.
+    post_inv = revalidate_complete_dirty_inventory(
+        wt,
+        expected_dirty_paths=expected_dirty_paths,
+        expected_fingerprints=expected_fingerprints,
+        phase="post-bind",
+    )
+    post_fps = dict(post_inv.get("observed_fingerprints") or {})
+    if not post_inv["ok"]:
+        journal["phase"] = "post_bind_inventory_failed"
+        journal["post_bind_inventory"] = {
+            "observed_dirty_paths": post_inv.get("observed_dirty_paths"),
+            "reasons": post_inv.get("reasons"),
+        }
+        journal["post_bind_fingerprints"] = post_fps
+        _atomic_write_json(jpath, journal)
+        return {
+            **base_result,
+            "success": False,
+            "reasons": list(post_inv["reasons"] or []) + [
+                "post-bind complete inventory revalidation failed after "
+                "bind_session_lock; lock may be rebound but content/path "
+                "verification failed (fail closed)"
+            ],
+            "lock_path": lock_path,
+            "journal_path": jpath,
+            "journal_phase": "post_bind_inventory_failed",
+            "generation_before": gen_before,
+            "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 {

docs/mcp-restart-path-inventory.md

@@ -1,90 +0,0 @@
-# MCP restart / reload / kill path inventory (#657)
-
-Complete inventory of every code, script, and host path that can **restart,
-reload, reconnect, kill, or force-recreate** an MCP process in this project,
-with each path classified and linked to the guard that constrains it.
-
-This document is the human-readable companion to the machine-readable registry
-in [`mcp_restart_paths.py`](../mcp_restart_paths.py). The two are kept in
-lock-step by [`tests/test_mcp_restart_paths.py`](../tests/test_mcp_restart_paths.py):
-every `path_id` below must appear in this file, and the source guards are run
-against the live tree.
-
-Roadmap linkage: this inventory is the enumeration step of the restart
-governance work — parent **#655**, restart-governance ADR **#656**, vision
-**#652**, roadmap **#653**. Related detection/guard work: master-advance
-staleness **#591**/**#420**, side-effect-free resolver **#685**, transport flap
-**#584**, manual-kill contamination **#630**.
-
-## Classifications
-
-| Classification | Meaning |
-|---|---|
-| `sanctioned_narrow_recovery` | One-shot, safe-by-construction recovery that never targets the running daemon. |
-| `guarded_fail_closed` | Detects a restart-requiring condition, then fails mutations closed and emits reconnect guidance. Never self-restarts. |
-| `forbidden` | A workflow-safety violation; where an LLM tool could invoke it, it is marked contamination. |
-| `removed` | A previously-existing unguarded restart primitive that has been deleted; a regression guard keeps it absent. |
-| `host_residual` | Behavior owned by the host/IDE, outside this process's control. Documented, not code-guarded here. |
-
-## The rule
-
-**No component may perform an unguarded full restart of the MCP daemon.** The
-in-process daemon (`gitea_mcp_server.py`, `mcp_server.py`,
-`role_session_router.py`) must never replace or terminate its own process:
-replacing the process after the host has wired up the stdio pipes desyncs the
-JSON-RPC transport (observed with Antigravity/Cascade hosts). Recovery is owned
-by the host/operator via a client reconnect — the daemon only ever *detects*
-and *fails closed*.
-
-## Inventory
-
-| path_id | Classification | Mechanism | Guard | Refs |
-|---|---|---|---|---|
-| `cli_venv_bootstrap_execv` | sanctioned_narrow_recovery | CLI wrapper scripts re-exec into `venv/bin/python3` via `os.execv`, guarded by `sys.executable != venv_python`. | One-shot pre-import bootstrap; runs before any MCP transport exists and only when not already on the venv interpreter; idempotent guard prevents a re-exec loop. | #657 |
-| `daemon_self_replacement` | forbidden | The daemon replacing/terminating its own process (`os.execv`/`os.kill`/`os._exit`) to reload code. | Forbidden by design; enforced against the source tree by `assert_no_daemon_self_replacement()`. | #657, #584 |
-| `legacy_auto_restart_helper` | removed | A helper (`_trigger_mcp_auto_restart`) that actively restarted the server from the read-only resolver path. | Removed in #685; kept absent by `assert_auto_restart_helper_absent()`. | #685, #657 |
-| `config_touch_reload` | removed | Touching (utime) the MCP client config to make the host reload the server. | Removed from the resolver in #685: stale detection is report-only, never mutating config, spawning threads, or calling `os._exit`. | #685, #657 |
-| `master_advance_auto_restart` | guarded_fail_closed | On-disk master advancing past the running code. | `master_parity_gate` captures startup parity and blocks mutations while stale, emitting restart guidance; the process never self-restarts. | #420, #591, #657 |
-| `stale_runtime_resolver_reconnect` | guarded_fail_closed | The capability resolver detecting a stale serving process. | Report-only (#685): returns `restart_required`/`stop_required` and an exact reconnect action; no restart, thread, config touch, or `os._exit`. | #685, #657 |
-| `manual_daemon_kill` | forbidden | Shell kills of the daemon: `pkill -f mcp_server.py`, `killall`, broad `pkill -f python` sweeps, or `kill <pid>` of a daemon pid. | Forbidden (#630): `runtime_recovery_guard` classifies these as contamination and `gitea_record_daemon_process_kill_attempt` writes a durable marker that fails later mutations closed. Operator maintenance authorization is read only from the environment. | #630, #657 |
-| `conflict_marker_infra_stop` | guarded_fail_closed | The daemon entrypoint scans for unresolved merge-conflict markers at startup and stops (`sys.exit(1)`). | Fail-closed startup stop, not a restart: the process exits and waits for the operator to resolve conflicts and relaunch; never loops. | #657 |
-| `ide_client_reconnect` | host_residual | A manual `/mcp reconnect` (or equivalent host action) that recreates the MCP client connection. | Outside this process's control; the sanctioned recovery the gates point operators toward. No in-process code initiates it. | #584, #656, #657 |
-| `profile_switch_runtime` | sanctioned_narrow_recovery | Switching the active execution profile at runtime (dynamic-profile mode). | In-process and restart-free: `runtime_switching_supported` is true, so a switch rebinds capability without recreating the process. | #656, #657 |
-
-## Guards enforced in CI
-
-`tests/test_mcp_restart_paths.py` asserts, against the live source tree:
-
-1. **Registry well-formedness** — every path has a valid classification, a
-   non-empty guard description, references, and locations; ids are unique; all
-   five classifications are represented.
-2. **Unknown restart attempts fail closed** —
-   `assert_restart_attempt_registered()` raises `UnknownRestartPathError` for
-   any path id not in this inventory, so a novel/unnamed restart primitive
-   cannot slip through silently.
-3. **Daemon never self-replaces** — `assert_no_daemon_self_replacement()` scans
-   the daemon modules for `os.execv`/`os.kill`/`os._exit`/`os.abort` calls
-   (comment/docstring mentions are ignored) and finds none.
-4. **Legacy helper stays removed** — `assert_auto_restart_helper_absent()`
-   confirms `_trigger_mcp_auto_restart` has not returned.
-5. **pkill stays forbidden** — a daemon `pkill` command still classifies as
-   contamination via `runtime_recovery_guard`.
-
-## Residual host behaviors (outside process control)
-
-* `/mcp reconnect` in the IDE/host — the sanctioned recovery for stale-runtime,
-  transport-flap (#584), and worktree-binding conditions. The daemon can only
-  emit guidance toward it.
-* Host-level process management (the operator relaunching the daemon after a
-  fail-closed stop, or after resolving merge conflicts).
-
-These are documented rather than code-guarded because the process cannot
-observe or gate them from inside itself.
-
-## Rollout
-
-Per #657, guards are introduced flag-free as **regression assertions** (they
-codify invariants that already hold) before any hard runtime block is layered
-on. When the restart coordinator (#655/#656) lands, registered paths gain a
-coordinator token/capability check; unregistered attempts already fail closed
-today via `assert_restart_attempt_registered()`.

mcp_restart_paths.py

@@ -1,475 +0,0 @@
-"""Inventory and fail-closed guards for MCP restart/reload/kill paths (#657).
-
-Single source of truth enumerating every code/script/doc path that can
-restart, reload, reconnect, kill, or force-recreate an MCP process. Each path
-is classified and linked to the guard that constrains it. The companion
-human-readable inventory lives in ``docs/mcp-restart-path-inventory.md`` and is
-kept in lock-step with this module by ``tests/test_mcp_restart_paths.py``.
-
-Design intent (aligns with #655 restart-coordinator roadmap):
-
-* **No unguarded full restart.** The in-process MCP daemon
-  (``gitea_mcp_server.py`` / ``mcp_server.py`` / ``role_session_router.py``)
-  must never replace or kill its own process — replacing the process after the
-  host wired up the stdio pipes desyncs the JSON-RPC transport (observed with
-  Antigravity/Cascade hosts). ``assert_no_daemon_self_replacement`` enforces
-  this against the live source tree.
-* **No legacy auto-restart helper.** ``_trigger_mcp_auto_restart`` was removed
-  when the stale-runtime resolver became side-effect free (#685);
-  ``assert_auto_restart_helper_absent`` keeps it removed.
-* **Unknown restart attempts fail closed.** LLM tools must route any restart
-  intent through a *registered* path. ``assert_restart_attempt_registered``
-  raises ``UnknownRestartPathError`` for anything not in this inventory.
-* **pkill stays forbidden (#630).** Manual daemon kills are classified as
-  contamination by :mod:`runtime_recovery_guard`; this module records that path
-  and the test asserts the classification still holds.
-
-This module performs no restarts, spawns no threads, and touches no config or
-process state. It is pure inventory + read-only source assertions.
-"""
-
-from __future__ import annotations
-
-import os
-from dataclasses import dataclass
-from pathlib import Path
-from typing import Iterable
-
-# --- Classifications -------------------------------------------------------
-
-#: A narrow, one-shot recovery that is safe by construction (e.g. a CLI wrapper
-#: re-execing into the venv interpreter before importing anything, or an
-#: in-process profile switch). Never targets the running MCP daemon process.
-CLASS_SANCTIONED_NARROW = "sanctioned_narrow_recovery"
-
-#: The path detects a condition that would require a restart, then *fails
-#: closed* on mutations and emits restart/reconnect guidance. It never restarts
-#: the process itself (recovery is owned by the host/operator).
-CLASS_GUARDED_FAIL_CLOSED = "guarded_fail_closed"
-
-#: The path is forbidden. Attempting it is a workflow-safety violation and,
-#: where an LLM tool could invoke it, is marked as contamination.
-CLASS_FORBIDDEN = "forbidden"
-
-#: A previously-existing unguarded restart primitive that has been deleted. A
-#: regression guard keeps it absent.
-CLASS_REMOVED = "removed"
-
-#: Behavior that lives in the host/IDE and is outside this process's control
-#: (e.g. a manual ``/mcp reconnect``). Documented, not code-guarded here.
-CLASS_HOST_RESIDUAL = "host_residual"
-
-VALID_CLASSIFICATIONS = frozenset(
-    {
-        CLASS_SANCTIONED_NARROW,
-        CLASS_GUARDED_FAIL_CLOSED,
-        CLASS_FORBIDDEN,
-        CLASS_REMOVED,
-        CLASS_HOST_RESIDUAL,
-    }
-)
-
-#: The in-process MCP daemon modules. These must never self-replace/self-kill.
-DAEMON_MODULES = (
-    "gitea_mcp_server.py",
-    "mcp_server.py",
-    "role_session_router.py",
-)
-
-#: The legacy auto-restart helper removed in #685. Must stay removed.
-LEGACY_AUTO_RESTART_HELPER = "_trigger_mcp_auto_restart"
-
-#: Call patterns that would let the daemon replace or terminate its own
-#: process. Matched as calls (trailing ``(``) so prose/docstring mentions such
-#: as "we do NOT os.execv() here" or "never calls ``os._exit``" do not trip the
-#: scanner (comment lines are stripped first regardless).
-DAEMON_SELF_REPLACEMENT_PRIMITIVES = (
-    "os.execv(",
-    "os.execve(",
-    "os.execvp(",
-    "os.execvpe(",
-    "os.kill(",
-    "os.killpg(",
-    "os._exit(",
-    "os.abort(",
-)
-
-
-@dataclass(frozen=True)
-class RestartPath:
-    """One classified restart/reload/kill path in the inventory."""
-
-    path_id: str
-    title: str
-    mechanism: str
-    classification: str
-    guard: str
-    locations: tuple[str, ...]
-    references: tuple[str, ...]
-    residual_host: bool = False
-    notes: str = ""
-
-
-class UnknownRestartPathError(RuntimeError):
-    """Raised when a restart attempt is not a registered, classified path."""
-
-
-# --- The inventory ---------------------------------------------------------
-
-_RESTART_PATHS: tuple[RestartPath, ...] = (
-    RestartPath(
-        path_id="cli_venv_bootstrap_execv",
-        title="CLI wrapper venv re-exec",
-        mechanism=(
-            "Standalone CLI scripts re-exec into venv/bin/python3 via os.execv "
-            "at import top, guarded by `sys.executable != venv_python`."
-        ),
-        classification=CLASS_SANCTIONED_NARROW,
-        guard=(
-            "One-shot, pre-import bootstrap; runs before any MCP transport "
-            "exists and only when not already on the venv interpreter, so it "
-            "cannot desync a live daemon. Idempotent guard condition prevents "
-            "a re-exec loop."
-        ),
-        locations=(
-            "create_pr.py",
-            "create_issue.py",
-            "close_issue.py",
-            "merge_pr.py",
-            "review_pr.py",
-            "edit_pr.py",
-            "delete_branch.py",
-            "mark_issue.py",
-            "manage_labels.py",
-            "list_issues.py",
-            "list_prs.py",
-        ),
-        references=("#657",),
-    ),
-    RestartPath(
-        path_id="daemon_self_replacement",
-        title="MCP daemon self-replacement",
-        mechanism=(
-            "The in-process MCP daemon replacing/terminating its own process "
-            "(os.execv/os.kill/os._exit) to reload code."
-        ),
-        classification=CLASS_FORBIDDEN,
-        guard=(
-            "Forbidden by design: replacing the process after the host wired "
-            "up stdio desyncs JSON-RPC (Antigravity/Cascade). Enforced against "
-            "the source tree by assert_no_daemon_self_replacement()."
-        ),
-        locations=("gitea_mcp_server.py:~155 (decision comment)",) + DAEMON_MODULES,
-        references=("#657", "#584"),
-    ),
-    RestartPath(
-        path_id="legacy_auto_restart_helper",
-        title="Legacy _trigger_mcp_auto_restart helper",
-        mechanism=(
-            "A helper that actively restarted the MCP server from the "
-            "read-only resolver path."
-        ),
-        classification=CLASS_REMOVED,
-        guard=(
-            "Removed in #685 when the resolver became side-effect free. Kept "
-            "absent by assert_auto_restart_helper_absent()."
-        ),
-        locations=("gitea_mcp_server.py", "mcp_server.py"),
-        references=("#685", "#657"),
-    ),
-    RestartPath(
-        path_id="config_touch_reload",
-        title="MCP client config-touch reload",
-        mechanism=(
-            "Touching (utime) the MCP client config file to make the host "
-            "reload/recreate the server process."
-        ),
-        classification=CLASS_REMOVED,
-        guard=(
-            "Removed from the resolver in #685: stale-runtime detection is "
-            "report-only and never mutates client config, spawns threads, or "
-            "calls os._exit."
-        ),
-        locations=("gitea_mcp_server.py (resolve_task_capability)",),
-        references=("#685", "#657"),
-    ),
-    RestartPath(
-        path_id="master_advance_auto_restart",
-        title="Master-advance staleness gate",
-        mechanism=(
-            "On-disk master advancing past the running code. The master-parity "
-            "gate detects it and fails mutations closed with restart guidance."
-        ),
-        classification=CLASS_GUARDED_FAIL_CLOSED,
-        guard=(
-            "Detect + fail closed only; the process never self-restarts. "
-            "master_parity_gate captures startup parity and blocks mutations "
-            "while stale, emitting restart/reconnect guidance."
-        ),
-        locations=(
-            "master_parity_gate.py",
-            "gitea_mcp_server.py (gitea_assess_master_parity)",
-        ),
-        references=("#420", "#591", "#657"),
-    ),
-    RestartPath(
-        path_id="stale_runtime_resolver_reconnect",
-        title="Stale-runtime resolver reconnect guidance",
-        mechanism=(
-            "The capability resolver detecting a stale serving process and "
-            "reporting restart_required/stop_required for a client reconnect."
-        ),
-        classification=CLASS_GUARDED_FAIL_CLOSED,
-        guard=(
-            "Report-only (#685): returns restart_required/stop_required and an "
-            "exact_safe_next_action pointing at IDE/client reconnect; performs "
-            "no restart, thread spawn, config touch, or os._exit."
-        ),
-        locations=("gitea_mcp_server.py (gitea_resolve_task_capability)",),
-        references=("#685", "#657"),
-    ),
-    RestartPath(
-        path_id="manual_daemon_kill",
-        title="Manual daemon kill (pkill/killall/kill)",
-        mechanism=(
-            "Shell kills of the MCP daemon: `pkill -f mcp_server.py`, "
-            "`killall`, broad `pkill -f python` sweeps, or `kill <pid>` of a "
-            "daemon pid."
-        ),
-        classification=CLASS_FORBIDDEN,
-        guard=(
-            "Forbidden (#630): runtime_recovery_guard classifies these as "
-            "contamination and gitea_record_daemon_process_kill_attempt writes "
-            "a durable marker that fails subsequent mutations closed. Operator "
-            "maintenance authorization is read only from the environment, not "
-            "from a tool argument."
-        ),
-        locations=(
-            "runtime_recovery_guard.py",
-            "gitea_mcp_server.py (gitea_record_daemon_process_kill_attempt)",
-        ),
-        references=("#630", "#657"),
-    ),
-    RestartPath(
-        path_id="conflict_marker_infra_stop",
-        title="Startup conflict-marker infra stop",
-        mechanism=(
-            "The daemon entrypoint scans for unresolved merge-conflict markers "
-            "at startup and stops (sys.exit(1)) if found."
-        ),
-        classification=CLASS_GUARDED_FAIL_CLOSED,
-        guard=(
-            "Fail-closed startup stop, not a restart: the process exits and "
-            "waits for the operator to resolve conflicts and relaunch. Never "
-            "self-restarts or loops."
-        ),
-        locations=("mcp_server.py (check_conflict_markers)",),
-        references=("#657",),
-    ),
-    RestartPath(
-        path_id="ide_client_reconnect",
-        title="Host/IDE MCP reconnect",
-        mechanism=(
-            "A manual `/mcp reconnect` (or equivalent host action) that the "
-            "IDE performs to recreate the MCP client connection."
-        ),
-        classification=CLASS_HOST_RESIDUAL,
-        guard=(
-            "Outside this process's control. It is the sanctioned recovery the "
-            "gates point operators toward; documented as residual host "
-            "behavior. No in-process code initiates it."
-        ),
-        locations=("host/IDE",),
-        references=("#584", "#656", "#657"),
-        residual_host=True,
-    ),
-    RestartPath(
-        path_id="profile_switch_runtime",
-        title="Runtime profile switch",
-        mechanism=(
-            "Switching the active execution profile at runtime "
-            "(dynamic-profile mode)."
-        ),
-        classification=CLASS_SANCTIONED_NARROW,
-        guard=(
-            "In-process and restart-free: runtime_switching_supported is true, "
-            "so a profile switch rebinds capability without recreating the "
-            "process. No restart primitive is invoked."
-        ),
-        locations=("gitea_mcp_server.py (gitea_activate_profile)",),
-        references=("#656", "#657"),
-    ),
-)
-
-_BY_ID: dict[str, RestartPath] = {p.path_id: p for p in _RESTART_PATHS}
-
-
-# --- Read-only accessors ---------------------------------------------------
-
-
-def iter_restart_paths() -> tuple[RestartPath, ...]:
-    """Return the full inventory as an immutable tuple."""
-
-    return _RESTART_PATHS
-
-
-def restart_path_ids() -> frozenset[str]:
-    """Return the set of registered path ids."""
-
-    return frozenset(_BY_ID)
-
-
-def get_restart_path(path_id: str) -> RestartPath:
-    """Return the registered path, or raise :class:`UnknownRestartPathError`."""
-
-    try:
-        return _BY_ID[path_id]
-    except KeyError as exc:
-        raise UnknownRestartPathError(
-            f"unknown restart path id {path_id!r}; not in the #657 inventory"
-        ) from exc
-
-
-def paths_by_classification(classification: str) -> tuple[RestartPath, ...]:
-    """Return all registered paths with the given classification."""
-
-    if classification not in VALID_CLASSIFICATIONS:
-        raise ValueError(f"unknown classification {classification!r}")
-    return tuple(p for p in _RESTART_PATHS if p.classification == classification)
-
-
-def assert_restart_attempt_registered(path_id: str) -> RestartPath:
-    """Fail closed unless ``path_id`` is a registered, classified restart path.
-
-    LLM tools that intend to trigger any restart/reload/reconnect must name a
-    registered path so an unknown/novel restart primitive cannot slip through
-    silently. Forbidden and removed paths are registered too — this only
-    asserts the attempt is *known*, not that it is *permitted*; callers must
-    still honor the classification.
-    """
-
-    return get_restart_path(path_id)
-
-
-def assert_registry_wellformed() -> None:
-    """Validate the inventory's own invariants (fail closed on drift)."""
-
-    seen: set[str] = set()
-    for path in _RESTART_PATHS:
-        if path.path_id in seen:
-            raise ValueError(f"duplicate restart path id {path.path_id!r}")
-        seen.add(path.path_id)
-        if path.classification not in VALID_CLASSIFICATIONS:
-            raise ValueError(
-                f"{path.path_id!r} has invalid classification "
-                f"{path.classification!r}"
-            )
-        if not path.guard.strip():
-            raise ValueError(f"{path.path_id!r} is missing a guard description")
-        if not path.references:
-            raise ValueError(f"{path.path_id!r} is missing references")
-        if not path.locations:
-            raise ValueError(f"{path.path_id!r} is missing locations")
-        if path.classification == CLASS_HOST_RESIDUAL and not path.residual_host:
-            raise ValueError(
-                f"{path.path_id!r} is host_residual but residual_host is False"
-            )
-
-
-# --- Source-tree guards ----------------------------------------------------
-
-
-def _repo_root(root: str | os.PathLike[str] | None = None) -> Path:
-    if root is not None:
-        return Path(root)
-    return Path(__file__).resolve().parent
-
-
-def _iter_code_lines(text: str) -> Iterable[tuple[int, str]]:
-    """Yield (1-based lineno, line) for lines that are not full-line comments."""
-
-    for lineno, line in enumerate(text.splitlines(), start=1):
-        if line.lstrip().startswith("#"):
-            continue
-        yield lineno, line
-
-
-def scan_daemon_self_replacement(
-    root: str | os.PathLike[str] | None = None,
-) -> list[dict[str, object]]:
-    """Return violations where a daemon module could self-replace/self-kill.
-
-    Scans :data:`DAEMON_MODULES` for calls in
-    :data:`DAEMON_SELF_REPLACEMENT_PRIMITIVES`. Full-line comments are ignored,
-    and only call forms (with a trailing ``(``) match, so decision comments and
-    docstrings that merely mention the primitives do not produce false hits.
-    """
-
-    repo = _repo_root(root)
-    violations: list[dict[str, object]] = []
-    for module in DAEMON_MODULES:
-        path = repo / module
-        if not path.exists():
-            continue
-        text = path.read_text(encoding="utf-8", errors="replace")
-        for lineno, line in _iter_code_lines(text):
-            for primitive in DAEMON_SELF_REPLACEMENT_PRIMITIVES:
-                if primitive in line:
-                    violations.append(
-                        {
-                            "module": module,
-                            "line": lineno,
-                            "primitive": primitive,
-                            "text": line.strip(),
-                        }
-                    )
-    return violations
-
-
-def assert_no_daemon_self_replacement(
-    root: str | os.PathLike[str] | None = None,
-) -> None:
-    """Fail closed if any daemon module can restart/kill its own process."""
-
-    violations = scan_daemon_self_replacement(root)
-    if violations:
-        rendered = "; ".join(
-            f"{v['module']}:{v['line']} {v['primitive']}" for v in violations
-        )
-        raise AssertionError(
-            "MCP daemon must never self-replace/self-kill (#657); found: "
-            f"{rendered}"
-        )
-
-
-def scan_auto_restart_helper(
-    root: str | os.PathLike[str] | None = None,
-) -> list[dict[str, object]]:
-    """Return occurrences of a *definition* of the legacy auto-restart helper."""
-
-    repo = _repo_root(root)
-    needle = f"def {LEGACY_AUTO_RESTART_HELPER}"
-    hits: list[dict[str, object]] = []
-    for module in DAEMON_MODULES:
-        path = repo / module
-        if not path.exists():
-            continue
-        text = path.read_text(encoding="utf-8", errors="replace")
-        for lineno, line in _iter_code_lines(text):
-            if needle in line:
-                hits.append({"module": module, "line": lineno})
-    return hits
-
-
-def assert_auto_restart_helper_absent(
-    root: str | os.PathLike[str] | None = None,
-) -> None:
-    """Fail closed if the removed ``_trigger_mcp_auto_restart`` reappears."""
-
-    hits = scan_auto_restart_helper(root)
-    if hits:
-        rendered = "; ".join(f"{h['module']}:{h['line']}" for h in hits)
-        raise AssertionError(
-            f"{LEGACY_AUTO_RESTART_HELPER} was removed in #685 and must not "
-            f"return (#657); found definition at: {rendered}"
-        )

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)

tests/test_mcp_restart_paths.py

@@ -1,146 +0,0 @@
-"""Tests for the MCP restart-path inventory and guards (#657).
-
-Covers:
-* the registry is well-formed and every path is classified;
-* unknown restart attempts fail closed (AC "fail closed on unknown restart");
-* the previously-unguarded full-restart primitives stay guarded/absent
-  against the real source tree (AC "tests for at least one previously
-  unguarded path");
-* pkill of the daemon is still classified as contamination (#630, AC3);
-* the inventory doc and module stay in lock-step.
-"""
-
-import os
-import tempfile
-import unittest
-from pathlib import Path
-
-import mcp_restart_paths as rp
-import runtime_recovery_guard
-
-REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
-DOC_PATH = os.path.join(REPO_ROOT, "docs", "mcp-restart-path-inventory.md")
-
-
-class TestRegistryWellformed(unittest.TestCase):
-    def test_registry_is_wellformed(self):
-        # Must not raise.
-        rp.assert_registry_wellformed()
-
-    def test_every_path_has_valid_classification(self):
-        for path in rp.iter_restart_paths():
-            self.assertIn(path.classification, rp.VALID_CLASSIFICATIONS)
-            self.assertTrue(path.guard.strip(), path.path_id)
-            self.assertTrue(path.references, path.path_id)
-            self.assertTrue(path.locations, path.path_id)
-
-    def test_ids_are_unique(self):
-        ids = [p.path_id for p in rp.iter_restart_paths()]
-        self.assertEqual(len(ids), len(set(ids)))
-
-    def test_covers_every_classification(self):
-        present = {p.classification for p in rp.iter_restart_paths()}
-        self.assertEqual(present, set(rp.VALID_CLASSIFICATIONS))
-
-
-class TestUnknownAttemptFailsClosed(unittest.TestCase):
-    def test_unknown_path_raises(self):
-        with self.assertRaises(rp.UnknownRestartPathError):
-            rp.assert_restart_attempt_registered("totally_novel_restart_hack")
-
-    def test_get_unknown_raises(self):
-        with self.assertRaises(rp.UnknownRestartPathError):
-            rp.get_restart_path("nope")
-
-    def test_registered_attempt_returns_path(self):
-        path = rp.assert_restart_attempt_registered("manual_daemon_kill")
-        self.assertEqual(path.classification, rp.CLASS_FORBIDDEN)
-
-
-class TestDaemonNeverSelfReplaces(unittest.TestCase):
-    """Previously-unguarded full-restart primitive: daemon self-replacement."""
-
-    def test_no_self_replacement_in_source(self):
-        # The live daemon modules must contain no os.execv/os.kill/os._exit
-        # self-restart call. Must not raise.
-        rp.assert_no_daemon_self_replacement(REPO_ROOT)
-
-    def test_scanner_flags_injected_violation(self):
-        # Guard the guard: prove the scanner catches a real self-replace call.
-        with tempfile.TemporaryDirectory() as tmp:
-            bad = Path(tmp) / "gitea_mcp_server.py"
-            bad.write_text(
-                "import os\n"
-                "def restart():\n"
-                "    os.execv('/usr/bin/python', ['python'])\n",
-                encoding="utf-8",
-            )
-            found = rp.scan_daemon_self_replacement(tmp)
-            self.assertTrue(found)
-            with self.assertRaises(AssertionError):
-                rp.assert_no_daemon_self_replacement(tmp)
-
-    def test_scanner_ignores_comment_and_docstring_mentions(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            ok = Path(tmp) / "gitea_mcp_server.py"
-            ok.write_text(
-                "import os\n"
-                "# NOT os.execv() to re-point the interpreter here.\n"
-                '"""Never calls os._exit to restart."""\n'
-                "value = 1\n",
-                encoding="utf-8",
-            )
-            self.assertEqual(rp.scan_daemon_self_replacement(tmp), [])
-
-
-class TestLegacyAutoRestartHelperRemoved(unittest.TestCase):
-    """Previously-unguarded full-restart path: _trigger_mcp_auto_restart."""
-
-    def test_helper_absent_in_source(self):
-        # Must not raise: helper was removed in #685.
-        rp.assert_auto_restart_helper_absent(REPO_ROOT)
-
-    def test_scanner_flags_reintroduced_helper(self):
-        with tempfile.TemporaryDirectory() as tmp:
-            bad = Path(tmp) / "mcp_server.py"
-            bad.write_text(
-                "def _trigger_mcp_auto_restart():\n    return True\n",
-                encoding="utf-8",
-            )
-            with self.assertRaises(AssertionError):
-                rp.assert_auto_restart_helper_absent(tmp)
-
-
-class TestPkillStaysForbidden(unittest.TestCase):
-    """AC3: pkill of the daemon remains forbidden/contaminating (#630)."""
-
-    def test_manual_daemon_kill_registered_as_forbidden(self):
-        path = rp.get_restart_path("manual_daemon_kill")
-        self.assertEqual(path.classification, rp.CLASS_FORBIDDEN)
-
-    def test_pkill_classified_as_contamination(self):
-        assessment = runtime_recovery_guard.assess_recovery_command(
-            "pkill -f mcp_server.py"
-        )
-        self.assertTrue(assessment["contaminated"])
-
-    def test_read_only_probe_not_contamination(self):
-        assessment = runtime_recovery_guard.assess_recovery_command(
-            "ps aux | grep mcp_server"
-        )
-        self.assertFalse(assessment["contaminated"])
-
-
-class TestInventoryDocInSync(unittest.TestCase):
-    def test_doc_exists(self):
-        self.assertTrue(os.path.exists(DOC_PATH), DOC_PATH)
-
-    def test_doc_mentions_every_path_id(self):
-        with open(DOC_PATH, encoding="utf-8") as handle:
-            doc = handle.read()
-        for path in rp.iter_restart_paths():
-            self.assertIn(path.path_id, doc, f"doc missing {path.path_id}")
-
-
-if __name__ == "__main__":
-    unittest.main()