jcwalker3
0b60fd6557
- Fix F1: Prepare recovery worktree detached at remote_head without git checkout -B to avoid exit 128 when branch is held by source worktree - Fix F2: Pass recovery_sanctioned=True in bind_session_lock and assess_same_issue_lease_conflict - Fix F3: Add SOURCE_RECOVER_DIRTY_ORPHANED to SANCTIONED_LOCK_SOURCES - Fix F4: Stop after Phase 4 dirty apply when conflicts exist; do not finalize session binding - Fix F5: Dynamically query competing live locks and workflow leases in MCP server - Fix F6: Fail closed on recovery worktree resume when HEAD does not match expected remote_head - Fix F7: Fail closed on remote HEAD observation failure rather than copying expected_remote_head pin - Fix F8: Enforce foreign overwrite protection requiring same claimant or sanctioned reclaim - Fix F9: Add real multi-worktree integration tests for prepare_recovery_worktree and lock rebind - Fix TestWorktreeStart: Bypass session lock check for dry-run and review/pr-* branches in scripts/worktree-start
1081 lines
38 KiB
Python
1081 lines
38 KiB
Python
"""Dirty orphaned author-issue worktree recovery (#860).
|
|
|
|
Self-hosting deadlock class observed against #850 / PR #853 and #855:
|
|
|
|
* same-claimant durable issue lock (``jcwalker3`` / ``prgs-author``)
|
|
* registered dirty worktree under ``branches/``
|
|
* lock missing PID / session PID / expiry / heartbeat (malformed)
|
|
* existing recovery (#753/#768/#772) and renewal require a *clean* worktree
|
|
and/or a determinable owner PID
|
|
* no sanctioned dirty-preserving rebind + sync to a newer remote PR head
|
|
|
|
This module is the pure evidence assessor and crash-safe recovery orchestrator
|
|
for that one class. It is an **explicit** recovery operation — it does not
|
|
silently widen ``gitea_lock_issue``.
|
|
|
|
Safety model
|
|
------------
|
|
Eligibility requires *all* of:
|
|
|
|
* same claimant identity and profile as the durable lock
|
|
* exact issue / repository / branch / registered source worktree agreement
|
|
* registered worktree under the canonical branches root (resolved-path ancestry)
|
|
* no active original process (when PID is present)
|
|
* no active competing author session or workflow lease
|
|
* sufficient corroborating evidence when PID fields are absent (caller pins)
|
|
* explicit caller-provided local head, remote/PR head, and dirty fingerprints
|
|
* no foreign, duplicate, or ambiguous ownership
|
|
|
|
A PID-less lock is **never** considered live merely because expiration fields
|
|
are absent (see also ``issue_lock_store.assess_lock_freshness``).
|
|
|
|
Dirty preservation
|
|
------------------
|
|
The source worktree is frozen: never cleaned, reset, overwritten, or deleted.
|
|
Recovery prefers a separately prepared recovery worktree checked out at the
|
|
pinned remote PR head. Dirty bytes are re-applied with path-level conflict
|
|
detection when upstream also changed a dirty path.
|
|
|
|
Crash safety
|
|
------------
|
|
A durable journal is written *before* filesystem or ownership mutation.
|
|
Retries resume or fail closed without stealing ownership, duplicating
|
|
worktrees, or losing dirty bytes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
from issue_lock_store import is_process_alive
|
|
from reviewer_worktree import parse_dirty_tracked_files
|
|
|
|
# Outcomes
|
|
ELIGIBLE = "ELIGIBLE"
|
|
REFUSED = "REFUSED"
|
|
NO_CANDIDATE = "NO_CANDIDATE"
|
|
RECOVERY_COMPLETED = "RECOVERY_COMPLETED"
|
|
RECOVERY_RESUMED = "RECOVERY_RESUMED"
|
|
CONFLICTS_PRESENT = "CONFLICTS_PRESENT"
|
|
|
|
# Journal phases (ordered)
|
|
PHASE_ELIGIBILITY = "1_eligibility_proven"
|
|
PHASE_JOURNAL_PERSISTED = "2_journal_persisted"
|
|
PHASE_RECOVERY_WORKTREE = "3_recovery_worktree_prepared"
|
|
PHASE_DIRTY_APPLIED = "4_dirty_applied"
|
|
PHASE_BINDING = "5_session_bound"
|
|
PHASE_COMPLETE = "6_complete"
|
|
|
|
JOURNAL_DIR_NAME = "dirty-orphan-recovery-journals"
|
|
REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
|
|
|
|
# Marker directory written into recovery worktrees for governed conflicts.
|
|
CONFLICT_STATE_DIR = ".gitea-recovery"
|
|
CONFLICT_STATE_FILE = "conflicts.json"
|
|
|
|
|
|
def _text(value: Any) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _same_realpath(left: str | None, right: str | None) -> bool:
|
|
if not left or not right:
|
|
return False
|
|
try:
|
|
return os.path.realpath(left) == os.path.realpath(right)
|
|
except OSError:
|
|
return left == right
|
|
|
|
|
|
def sha256_bytes(data: bytes) -> str:
|
|
return hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def sha256_file(path: str) -> str:
|
|
with open(path, "rb") as fh:
|
|
return sha256_bytes(fh.read())
|
|
|
|
|
|
def get_journal_dir(override: str | None = None) -> str:
|
|
if override:
|
|
path = override
|
|
elif os.environ.get("GITEA_DIRTY_ORPHAN_RECOVERY_JOURNAL_DIR"):
|
|
path = os.environ["GITEA_DIRTY_ORPHAN_RECOVERY_JOURNAL_DIR"]
|
|
else:
|
|
path = os.path.join(
|
|
os.path.expanduser("~/.cache/gitea-tools"), JOURNAL_DIR_NAME
|
|
)
|
|
os.makedirs(path, mode=0o700, exist_ok=True)
|
|
return path
|
|
|
|
|
|
def _journal_path(idempotency_key: str, journal_dir: str | None = None) -> str:
|
|
safe = "".join(
|
|
c if c.isalnum() or c in ("-", "_", ".") else "_"
|
|
for c in idempotency_key
|
|
)
|
|
return os.path.join(get_journal_dir(journal_dir), f"{safe}.json")
|
|
|
|
|
|
def load_journal(
|
|
idempotency_key: str, journal_dir: str | None = None
|
|
) -> dict[str, Any] | None:
|
|
path = _journal_path(idempotency_key, journal_dir=journal_dir)
|
|
if not os.path.isfile(path):
|
|
return None
|
|
if os.path.islink(path):
|
|
raise ValueError(f"refusing journal path that is a symlink: {path}")
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
data = json.load(fh)
|
|
if not isinstance(data, dict):
|
|
raise ValueError("corrupt recovery journal: not an object")
|
|
return data
|
|
|
|
|
|
def save_journal(
|
|
journal: Mapping[str, Any], journal_dir: str | None = None
|
|
) -> str:
|
|
key = _text(journal.get("idempotency_key"))
|
|
if not key:
|
|
raise ValueError("journal requires idempotency_key")
|
|
path = _journal_path(key, journal_dir=journal_dir)
|
|
if os.path.islink(path):
|
|
raise ValueError(f"refusing to write journal through symlink: {path}")
|
|
tmp = f"{path}.tmp.{os.getpid()}"
|
|
with open(tmp, "w", encoding="utf-8") as fh:
|
|
json.dump(dict(journal), fh, indent=2, sort_keys=True)
|
|
fh.flush()
|
|
os.fsync(fh.fileno())
|
|
os.replace(tmp, path)
|
|
return path
|
|
|
|
|
|
def derive_idempotency_key(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
source_worktree: str,
|
|
expected_local_head: str,
|
|
expected_remote_head: str,
|
|
) -> str:
|
|
raw = "|".join(
|
|
[
|
|
"dirty-orphan-recovery",
|
|
_text(remote),
|
|
_text(org),
|
|
_text(repo),
|
|
f"issue-{int(issue_number)}",
|
|
os.path.realpath(_text(source_worktree)),
|
|
_text(expected_local_head)[:40],
|
|
_text(expected_remote_head)[:40],
|
|
]
|
|
)
|
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()[:32]
|
|
|
|
|
|
def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, str]:
|
|
claimant = lock.get("claimant")
|
|
if not isinstance(claimant, Mapping):
|
|
lease = lock.get("work_lease")
|
|
claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
|
|
if not isinstance(claimant, Mapping):
|
|
return {}
|
|
return {
|
|
"username": _text(claimant.get("username")),
|
|
"profile": _text(claimant.get("profile")),
|
|
}
|
|
|
|
|
|
def _recorded_pid(lock: Mapping[str, Any]) -> Any:
|
|
pid = lock.get("session_pid")
|
|
if pid is None:
|
|
pid = lock.get("pid")
|
|
return pid
|
|
|
|
|
|
def _malformed_pid_fields(lock: Mapping[str, Any]) -> list[str]:
|
|
"""Return reasons the lock's PID fields are unusable (not automatically live)."""
|
|
missing: list[str] = []
|
|
pid = _recorded_pid(lock)
|
|
if pid is None or _text(pid) == "":
|
|
missing.append("session_pid/pid")
|
|
return missing
|
|
try:
|
|
if int(pid) <= 0:
|
|
missing.append("session_pid/pid")
|
|
except (TypeError, ValueError):
|
|
missing.append("session_pid/pid")
|
|
return missing
|
|
|
|
|
|
def is_path_under_canonical_branches(
|
|
path: str,
|
|
*,
|
|
canonical_repo_root: str,
|
|
branches_dirname: str = "branches",
|
|
) -> tuple[bool, list[str]]:
|
|
"""Resolved-path ancestry under ``{canonical_repo_root}/branches``.
|
|
|
|
Rejects string-substring tricks, traversal, and symlink escapes outside
|
|
the canonical branches root.
|
|
"""
|
|
reasons: list[str] = []
|
|
if not path or not canonical_repo_root:
|
|
return False, ["path and canonical_repo_root are required"]
|
|
try:
|
|
repo_root = os.path.realpath(canonical_repo_root)
|
|
branches_root = os.path.realpath(os.path.join(repo_root, branches_dirname))
|
|
# realpath on a non-existent path still normalizes; prefer it so
|
|
# assessment can run without the directory existing yet.
|
|
target = os.path.realpath(path)
|
|
except OSError as exc:
|
|
return False, [f"path resolution failed: {exc}"]
|
|
|
|
if not target.startswith(branches_root + os.sep) and target != branches_root:
|
|
reasons.append(
|
|
f"path '{path}' is not under canonical branches root '{branches_root}'"
|
|
)
|
|
return False, reasons
|
|
try:
|
|
rel = os.path.relpath(target, branches_root)
|
|
except ValueError:
|
|
return False, ["path not relative to branches root"]
|
|
if rel.startswith(".."):
|
|
return False, ["path escapes branches root via relative traversal"]
|
|
return True, []
|
|
|
|
|
|
def _result(
|
|
outcome: str,
|
|
*,
|
|
eligible: bool,
|
|
reasons: list[str],
|
|
evidence: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"outcome": outcome,
|
|
"eligible": eligible,
|
|
"recovery_eligible": eligible,
|
|
"reasons": list(reasons),
|
|
"evidence": evidence,
|
|
}
|
|
|
|
|
|
def assess_dirty_orphan_recovery(
|
|
existing_lock: Mapping[str, Any] | None,
|
|
*,
|
|
issue_number: int,
|
|
branch_name: str,
|
|
source_worktree_path: str,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
identity: str,
|
|
profile: str,
|
|
expected_local_head: str,
|
|
expected_remote_head: str,
|
|
expected_dirty_fingerprints: Mapping[str, str],
|
|
current_branch: str | None,
|
|
porcelain_status: str,
|
|
observed_local_head: str | None,
|
|
observed_remote_head: str | None,
|
|
observed_dirty_fingerprints: Mapping[str, str] | None,
|
|
competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
|
|
competing_live_sessions: Sequence[Mapping[str, Any]] | None = None,
|
|
workflow_lease_active: bool | None = None,
|
|
workflow_lease_expired: bool | None = None,
|
|
canonical_repo_root: str,
|
|
worktree_registered: bool,
|
|
current_pid: int | None = None,
|
|
owner_process_alive_override: bool | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Assess whether dirty-orphan recovery is eligible. Pure; no I/O mutation."""
|
|
evidence: dict[str, Any] = {
|
|
"issue_number": issue_number,
|
|
"branch_name": branch_name,
|
|
"source_worktree_path": source_worktree_path,
|
|
"remote": remote,
|
|
"org": org,
|
|
"repo": repo,
|
|
"identity": identity,
|
|
"profile": profile,
|
|
"expected_local_head": _text(expected_local_head),
|
|
"expected_remote_head": _text(expected_remote_head),
|
|
"expected_dirty_paths": sorted(expected_dirty_fingerprints or {}),
|
|
}
|
|
reasons: list[str] = []
|
|
|
|
if not existing_lock:
|
|
return _result(
|
|
NO_CANDIDATE,
|
|
eligible=False,
|
|
reasons=["no existing durable lock for this issue"],
|
|
evidence=evidence,
|
|
)
|
|
|
|
lock = dict(existing_lock)
|
|
if lock.get("issue_number") != issue_number:
|
|
return _result(
|
|
NO_CANDIDATE,
|
|
eligible=False,
|
|
reasons=[
|
|
f"existing lock targets issue #{lock.get('issue_number')}, "
|
|
f"not #{issue_number}"
|
|
],
|
|
evidence=evidence,
|
|
)
|
|
|
|
for field in REQUIRED_LOCK_FIELDS:
|
|
if not _text(lock.get(field)):
|
|
reasons.append(f"durable lock missing required field '{field}'")
|
|
|
|
# Repository / branch / worktree agreement
|
|
for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
|
|
actual = _text(lock.get(field))
|
|
if actual and actual != _text(expected):
|
|
reasons.append(
|
|
f"lock {field} '{actual}' does not match requested '{_text(expected)}'"
|
|
)
|
|
elif not actual:
|
|
# Some legacy locks omit remote/org/repo; require explicit pin match
|
|
# via caller still supplying them and branch/worktree agreement.
|
|
evidence[f"lock_{field}_absent"] = True
|
|
|
|
locked_branch = _text(lock.get("branch_name"))
|
|
if locked_branch != _text(branch_name):
|
|
reasons.append(
|
|
f"lock branch '{locked_branch}' does not match requested "
|
|
f"'{_text(branch_name)}'"
|
|
)
|
|
evidence["locked_branch"] = locked_branch
|
|
|
|
locked_wt = _text(lock.get("worktree_path"))
|
|
if not _same_realpath(locked_wt, source_worktree_path):
|
|
reasons.append(
|
|
f"lock worktree '{locked_wt}' does not match declared source "
|
|
f"'{_text(source_worktree_path)}'"
|
|
)
|
|
evidence["locked_worktree_path"] = locked_wt
|
|
|
|
under, under_reasons = is_path_under_canonical_branches(
|
|
source_worktree_path, canonical_repo_root=canonical_repo_root
|
|
)
|
|
if not under:
|
|
reasons.extend(under_reasons)
|
|
if not worktree_registered:
|
|
reasons.append("source worktree is not registered in git worktree list")
|
|
|
|
# Claimant identity
|
|
claimant = _lock_claimant(lock)
|
|
evidence["lock_claimant"] = claimant
|
|
if claimant.get("username") != _text(identity):
|
|
reasons.append(
|
|
f"foreign claimant identity '{claimant.get('username')}' "
|
|
f"(caller '{_text(identity)}')"
|
|
)
|
|
if claimant.get("profile") != _text(profile):
|
|
reasons.append(
|
|
f"foreign claimant profile '{claimant.get('profile')}' "
|
|
f"(caller '{_text(profile)}')"
|
|
)
|
|
|
|
# PID / liveness
|
|
pid_missing = _malformed_pid_fields(lock)
|
|
recorded_pid = _recorded_pid(lock)
|
|
evidence["recorded_pid"] = recorded_pid
|
|
evidence["pid_fields_missing"] = pid_missing
|
|
if pid_missing:
|
|
evidence["pid_less_malformed"] = True
|
|
# PID-less is never live by missing expiry alone. Eligibility continues
|
|
# only with full corroborating pins (already required below).
|
|
else:
|
|
alive = (
|
|
owner_process_alive_override
|
|
if owner_process_alive_override is not None
|
|
else is_process_alive(int(recorded_pid))
|
|
)
|
|
evidence["owner_process_alive"] = alive
|
|
if alive:
|
|
reasons.append(
|
|
f"original owner process pid {recorded_pid} is still alive; "
|
|
"recovery refused"
|
|
)
|
|
|
|
# Competing ownership
|
|
for entry in competing_live_locks or ():
|
|
if not isinstance(entry, Mapping):
|
|
continue
|
|
if entry.get("issue_number") == issue_number:
|
|
reasons.append(
|
|
"competing live lock observed for the same issue; recovery refused"
|
|
)
|
|
for entry in competing_live_sessions or ():
|
|
if not isinstance(entry, Mapping):
|
|
continue
|
|
sess_issue = entry.get("issue_number")
|
|
sess_identity = _text(entry.get("identity") or entry.get("username"))
|
|
if sess_issue == issue_number and sess_identity and sess_identity != _text(identity):
|
|
reasons.append(
|
|
f"competing live author session by '{sess_identity}' on issue "
|
|
f"#{issue_number}"
|
|
)
|
|
elif sess_issue == issue_number and entry.get("active"):
|
|
# same claimant active elsewhere still blocks ambiguous ownership
|
|
if entry.get("session_pid") not in (None, current_pid, os.getpid()):
|
|
reasons.append(
|
|
"ambiguous competing same-issue author session still active"
|
|
)
|
|
|
|
if workflow_lease_active is True and workflow_lease_expired is not True:
|
|
reasons.append(
|
|
"active competing workflow lease still live; recovery refused"
|
|
)
|
|
evidence["workflow_lease_active"] = workflow_lease_active
|
|
evidence["workflow_lease_expired"] = workflow_lease_expired
|
|
|
|
# Dirty requirement (this recovery class is *for* dirty trees)
|
|
dirty_files = parse_dirty_tracked_files(porcelain_status)
|
|
if not dirty_files and not expected_dirty_fingerprints:
|
|
reasons.append(
|
|
"worktree is clean and no dirty fingerprints were provided; "
|
|
"use clean-worktree recovery (#753/#772) instead"
|
|
)
|
|
evidence["observed_dirty_files"] = dirty_files
|
|
|
|
# Explicit pins
|
|
if not _text(expected_local_head) or len(_text(expected_local_head)) < 40:
|
|
reasons.append("expected_local_head pin missing or not a full SHA")
|
|
if not _text(expected_remote_head) or len(_text(expected_remote_head)) < 40:
|
|
reasons.append("expected_remote_head pin missing or not a full SHA")
|
|
if not expected_dirty_fingerprints:
|
|
reasons.append("expected_dirty_fingerprints pin is required")
|
|
|
|
if _text(observed_local_head) and _text(observed_local_head) != _text(
|
|
expected_local_head
|
|
):
|
|
reasons.append(
|
|
f"local head mismatch: observed {_text(observed_local_head)} != "
|
|
f"pinned {_text(expected_local_head)}"
|
|
)
|
|
if _text(observed_remote_head) and _text(observed_remote_head) != _text(
|
|
expected_remote_head
|
|
):
|
|
reasons.append(
|
|
f"remote/PR head mismatch: observed {_text(observed_remote_head)} != "
|
|
f"pinned {_text(expected_remote_head)}"
|
|
)
|
|
if _text(expected_local_head) == _text(expected_remote_head):
|
|
# Divergence is the motivating case; equal heads are allowed only when
|
|
# dirty files still need rebinding, so do not refuse equality.
|
|
evidence["heads_equal"] = True
|
|
else:
|
|
evidence["heads_diverged"] = True
|
|
|
|
if current_branch and _text(current_branch) != locked_branch:
|
|
reasons.append(
|
|
f"source worktree is on branch '{_text(current_branch)}', not "
|
|
f"locked branch '{locked_branch}'"
|
|
)
|
|
|
|
# Fingerprint verification
|
|
observed_fps = dict(observed_dirty_fingerprints or {})
|
|
for path, expected_fp in (expected_dirty_fingerprints or {}).items():
|
|
rel = _text(path)
|
|
if not rel or rel.startswith("/") or ".." in rel.split("/"):
|
|
reasons.append(f"unsafe dirty path pin refused: {path!r}")
|
|
continue
|
|
obs = _text(observed_fps.get(rel))
|
|
if not obs:
|
|
reasons.append(f"missing observed fingerprint for dirty path '{rel}'")
|
|
elif obs != _text(expected_fp):
|
|
reasons.append(
|
|
f"dirty fingerprint mismatch for '{rel}': "
|
|
f"observed {obs} != pinned {_text(expected_fp)}"
|
|
)
|
|
|
|
# PID-less corroboration: all explicit pins must already have passed.
|
|
if pid_missing and reasons:
|
|
reasons.append(
|
|
"PID-less malformed lock additionally requires full pin corroboration; "
|
|
"one or more corroborating checks failed"
|
|
)
|
|
|
|
if reasons:
|
|
return _result(REFUSED, eligible=False, reasons=reasons, evidence=evidence)
|
|
|
|
evidence["eligibility"] = ELIGIBLE
|
|
return _result(ELIGIBLE, eligible=True, reasons=[], evidence=evidence)
|
|
|
|
|
|
def detect_path_conflicts(
|
|
*,
|
|
dirty_paths: Sequence[str],
|
|
local_head_contents: Mapping[str, bytes | None],
|
|
remote_head_contents: Mapping[str, bytes | None],
|
|
dirty_contents: Mapping[str, bytes],
|
|
) -> list[dict[str, Any]]:
|
|
"""Path-level conflicts: upstream and dirty patch both changed the path."""
|
|
conflicts: list[dict[str, Any]] = []
|
|
for path in dirty_paths:
|
|
local_b = local_head_contents.get(path)
|
|
remote_b = remote_head_contents.get(path)
|
|
dirty_b = dirty_contents.get(path)
|
|
if dirty_b is None:
|
|
continue
|
|
# Upstream changed relative to the local head version of the path.
|
|
upstream_changed = (local_b or b"") != (remote_b or b"")
|
|
dirty_differs_from_remote = dirty_b != (remote_b or b"")
|
|
if upstream_changed and dirty_differs_from_remote:
|
|
conflicts.append(
|
|
{
|
|
"path": path,
|
|
"local_head_sha256": sha256_bytes(local_b) if local_b is not None else None,
|
|
"remote_head_sha256": sha256_bytes(remote_b) if remote_b is not None else None,
|
|
"dirty_sha256": sha256_bytes(dirty_b),
|
|
"reason": (
|
|
"upstream and preserved dirty patch both changed this path"
|
|
),
|
|
}
|
|
)
|
|
return conflicts
|
|
|
|
|
|
def _safe_open_lockfile(path: str):
|
|
"""Open a lock file refusing symlinks (O_NOFOLLOW when available)."""
|
|
if os.path.islink(path):
|
|
raise ValueError(f"refusing lock file that is a symlink: {path}")
|
|
flags = os.O_RDWR | os.O_CREAT
|
|
if hasattr(os, "O_NOFOLLOW"):
|
|
flags |= os.O_NOFOLLOW
|
|
fd = os.open(path, flags, 0o600)
|
|
try:
|
|
st = os.fstat(fd)
|
|
if stat.S_ISLNK(st.st_mode):
|
|
os.close(fd)
|
|
raise ValueError(f"refusing lock file that is a symlink: {path}")
|
|
except Exception:
|
|
try:
|
|
os.close(fd)
|
|
except OSError:
|
|
pass
|
|
raise
|
|
return fd
|
|
|
|
|
|
def build_recovery_lock_record(
|
|
*,
|
|
existing_lock: Mapping[str, Any],
|
|
issue_number: int,
|
|
branch_name: str,
|
|
recovery_worktree_path: str,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
identity: str,
|
|
profile: str,
|
|
expected_remote_head: str,
|
|
source_worktree_path: str,
|
|
conflicts: Sequence[Mapping[str, Any]],
|
|
session_pid: int,
|
|
) -> dict[str, Any]:
|
|
"""Construct a new durable lock bound to the recovery worktree + live PID."""
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
now = datetime.now(timezone.utc)
|
|
expires = now + timedelta(hours=4)
|
|
record = {
|
|
"remote": remote,
|
|
"org": org,
|
|
"repo": repo,
|
|
"issue_number": issue_number,
|
|
"branch_name": branch_name,
|
|
"worktree_path": recovery_worktree_path,
|
|
"pid": session_pid,
|
|
"session_pid": session_pid,
|
|
"claimant": {"username": identity, "profile": profile},
|
|
"work_lease": {
|
|
"operation_type": "author_issue_work",
|
|
"issue_number": issue_number,
|
|
"pr_number": existing_lock.get("work_lease", {}).get("pr_number")
|
|
if isinstance(existing_lock.get("work_lease"), Mapping)
|
|
else existing_lock.get("pr_number"),
|
|
"branch": branch_name,
|
|
"worktree_path": recovery_worktree_path,
|
|
"claimant": {"username": identity, "profile": profile},
|
|
"created_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"expires_at": expires.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"last_heartbeat_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
},
|
|
"dirty_orphan_recovery": {
|
|
"recovered": True,
|
|
"source_worktree_path": source_worktree_path,
|
|
"recovery_worktree_path": recovery_worktree_path,
|
|
"accepted_head": expected_remote_head,
|
|
"conflicts": list(conflicts),
|
|
"source_frozen": True,
|
|
},
|
|
"lock_provenance": {
|
|
"source": "gitea_recover_dirty_orphaned_issue_worktree",
|
|
"written_by_tool": "gitea_recover_dirty_orphaned_issue_worktree",
|
|
"written_at": now.strftime("%Y-%m-%dT%H:%M:%SZ"),
|
|
"claimant": {"username": identity, "profile": profile},
|
|
},
|
|
}
|
|
# Preserve prior assignment/lease ids when present (non-authoritative).
|
|
for key in ("assignment_id", "lease_id", "owner_session", "expected_base_sha"):
|
|
if key in existing_lock:
|
|
record[key] = existing_lock[key]
|
|
return record
|
|
|
|
|
|
def write_conflict_state(
|
|
recovery_worktree: str, conflicts: Sequence[Mapping[str, Any]]
|
|
) -> str:
|
|
state_dir = os.path.join(recovery_worktree, CONFLICT_STATE_DIR)
|
|
os.makedirs(state_dir, mode=0o700, exist_ok=True)
|
|
path = os.path.join(state_dir, CONFLICT_STATE_FILE)
|
|
payload = {
|
|
"conflicts": list(conflicts),
|
|
"resolution": "author_edit_required" if conflicts else "none",
|
|
}
|
|
with open(path, "w", encoding="utf-8") as fh:
|
|
json.dump(payload, fh, indent=2, sort_keys=True)
|
|
return path
|
|
|
|
|
|
def apply_dirty_bytes(
|
|
*,
|
|
recovery_worktree: str,
|
|
dirty_contents: Mapping[str, bytes],
|
|
conflict_paths: set[str],
|
|
) -> list[str]:
|
|
"""Write non-conflicting dirty bytes into the recovery worktree.
|
|
|
|
Conflicting paths are written as ``*.recovered-dirty`` siblings so the
|
|
original dirty bytes remain recoverable without overwriting upstream.
|
|
"""
|
|
written: list[str] = []
|
|
for rel, data in dirty_contents.items():
|
|
rel = _text(rel)
|
|
if not rel or rel.startswith("/") or ".." in rel.split("/"):
|
|
raise ValueError(f"unsafe relative path: {rel!r}")
|
|
dest = os.path.join(recovery_worktree, rel)
|
|
parent = os.path.dirname(dest)
|
|
if parent:
|
|
os.makedirs(parent, exist_ok=True)
|
|
if rel in conflict_paths:
|
|
sidecar = dest + ".recovered-dirty"
|
|
with open(sidecar, "wb") as fh:
|
|
fh.write(data)
|
|
written.append(rel + ".recovered-dirty")
|
|
else:
|
|
with open(dest, "wb") as fh:
|
|
fh.write(data)
|
|
written.append(rel)
|
|
return written
|
|
|
|
|
|
@dataclass
|
|
class GitOps:
|
|
"""Injectable git operations for tests."""
|
|
|
|
def run(self, args: list[str], *, cwd: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
args,
|
|
cwd=cwd,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def prepare_recovery_worktree(
|
|
*,
|
|
canonical_repo_root: str,
|
|
recovery_worktree_path: str,
|
|
branch_name: str,
|
|
remote_head: str,
|
|
git_ops: GitOps | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create or resume a recovery worktree at the pinned remote head.
|
|
|
|
Leaves the source worktree untouched. Uses ``git worktree add`` only when
|
|
the recovery path does not already exist (idempotent resume).
|
|
"""
|
|
git = git_ops or GitOps()
|
|
under, reasons = is_path_under_canonical_branches(
|
|
recovery_worktree_path, canonical_repo_root=canonical_repo_root
|
|
)
|
|
if not under:
|
|
return {"success": False, "reasons": reasons, "created": False}
|
|
|
|
if os.path.isdir(recovery_worktree_path):
|
|
# Resume: verify HEAD matches pin.
|
|
probe = git.run(
|
|
["git", "rev-parse", "HEAD"], cwd=recovery_worktree_path
|
|
)
|
|
head = (probe.stdout or "").strip()
|
|
if probe.returncode != 0 or head != remote_head:
|
|
return {
|
|
"success": False,
|
|
"created": False,
|
|
"resumed": True,
|
|
"head": head,
|
|
"reasons": [
|
|
f"resume recovery worktree HEAD '{head}' does not match expected remote HEAD '{remote_head}'"
|
|
],
|
|
}
|
|
return {
|
|
"success": True,
|
|
"created": False,
|
|
"resumed": True,
|
|
"head": head,
|
|
"reasons": [],
|
|
}
|
|
|
|
# Create detached-at-head worktree. Do not run git checkout -B because
|
|
# the source worktree holds the branch name (Git exit 128).
|
|
add = git.run(
|
|
[
|
|
"git",
|
|
"worktree",
|
|
"add",
|
|
"--detach",
|
|
recovery_worktree_path,
|
|
remote_head,
|
|
],
|
|
cwd=canonical_repo_root,
|
|
)
|
|
if add.returncode != 0:
|
|
return {
|
|
"success": False,
|
|
"created": False,
|
|
"reasons": [
|
|
f"git worktree add failed: {(add.stderr or add.stdout or '').strip()}"
|
|
],
|
|
}
|
|
return {
|
|
"success": True,
|
|
"created": True,
|
|
"resumed": False,
|
|
"head": remote_head,
|
|
"reasons": [],
|
|
}
|
|
|
|
|
|
def run_dirty_orphan_recovery(
|
|
*,
|
|
assessment: Mapping[str, Any],
|
|
existing_lock: Mapping[str, Any],
|
|
issue_number: int,
|
|
branch_name: str,
|
|
source_worktree_path: str,
|
|
recovery_worktree_path: str,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
identity: str,
|
|
profile: str,
|
|
expected_local_head: str,
|
|
expected_remote_head: str,
|
|
expected_dirty_fingerprints: Mapping[str, str],
|
|
dirty_contents: Mapping[str, bytes],
|
|
local_head_contents: Mapping[str, bytes | None],
|
|
remote_head_contents: Mapping[str, bytes | None],
|
|
canonical_repo_root: str,
|
|
bind_lock: bool,
|
|
lock_writer: Any | None = None,
|
|
git_ops: GitOps | None = None,
|
|
journal_dir: str | None = None,
|
|
session_pid: int | None = None,
|
|
interrupt_after_phase: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Execute recovery with crash-journal phases. Idempotent on retry."""
|
|
if not assessment.get("eligible"):
|
|
return {
|
|
"success": False,
|
|
"performed": False,
|
|
"outcome": assessment.get("outcome") or REFUSED,
|
|
"reasons": list(assessment.get("reasons") or ["not eligible"]),
|
|
"evidence": dict(assessment.get("evidence") or {}),
|
|
}
|
|
|
|
# Verify dirty bytes match pins before any mutation.
|
|
for path, expected_fp in expected_dirty_fingerprints.items():
|
|
data = dirty_contents.get(path)
|
|
if data is None:
|
|
return {
|
|
"success": False,
|
|
"performed": False,
|
|
"outcome": REFUSED,
|
|
"reasons": [f"dirty content missing for pinned path '{path}'"],
|
|
"evidence": {},
|
|
}
|
|
if sha256_bytes(data) != _text(expected_fp):
|
|
return {
|
|
"success": False,
|
|
"performed": False,
|
|
"outcome": REFUSED,
|
|
"reasons": [f"dirty content fingerprint drift for '{path}'"],
|
|
"evidence": {},
|
|
}
|
|
|
|
idem = derive_idempotency_key(
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
issue_number=issue_number,
|
|
source_worktree=source_worktree_path,
|
|
expected_local_head=expected_local_head,
|
|
expected_remote_head=expected_remote_head,
|
|
)
|
|
journal = load_journal(idem, journal_dir=journal_dir) or {
|
|
"idempotency_key": idem,
|
|
"issue_number": issue_number,
|
|
"branch_name": branch_name,
|
|
"source_worktree_path": os.path.realpath(source_worktree_path),
|
|
"recovery_worktree_path": recovery_worktree_path,
|
|
"expected_local_head": expected_local_head,
|
|
"expected_remote_head": expected_remote_head,
|
|
"expected_dirty_fingerprints": dict(expected_dirty_fingerprints),
|
|
"phase": None,
|
|
"artifacts_created": {
|
|
"journal": False,
|
|
"recovery_worktree": False,
|
|
"dirty_applied": False,
|
|
"binding": False,
|
|
},
|
|
"conflicts": [],
|
|
"complete": False,
|
|
}
|
|
|
|
if journal.get("complete"):
|
|
return {
|
|
"success": True,
|
|
"performed": False,
|
|
"outcome": RECOVERY_RESUMED,
|
|
"reasons": ["recovery already complete; idempotent no-op"],
|
|
"evidence": {
|
|
"journal": journal,
|
|
"recovery_worktree_path": journal.get("recovery_worktree_path"),
|
|
},
|
|
"journal": journal,
|
|
}
|
|
|
|
# Phase 1: eligibility already proven by caller assessment.
|
|
journal["phase"] = PHASE_ELIGIBILITY
|
|
if interrupt_after_phase == PHASE_ELIGIBILITY:
|
|
return {
|
|
"success": False,
|
|
"performed": False,
|
|
"outcome": "INTERRUPTED",
|
|
"reasons": ["interrupted after eligibility (test harness)"],
|
|
"journal": journal,
|
|
"evidence": {"phase": PHASE_ELIGIBILITY},
|
|
}
|
|
|
|
# Phase 2: persist journal BEFORE filesystem/ownership mutation.
|
|
journal["phase"] = PHASE_JOURNAL_PERSISTED
|
|
journal["artifacts_created"]["journal"] = True
|
|
save_journal(journal, journal_dir=journal_dir)
|
|
if interrupt_after_phase == PHASE_JOURNAL_PERSISTED:
|
|
return {
|
|
"success": False,
|
|
"performed": True,
|
|
"outcome": "INTERRUPTED",
|
|
"reasons": ["interrupted after journal persistence (test harness)"],
|
|
"journal": journal,
|
|
"evidence": {"phase": PHASE_JOURNAL_PERSISTED},
|
|
}
|
|
|
|
# Phase 3: recovery worktree at remote head.
|
|
prep = prepare_recovery_worktree(
|
|
canonical_repo_root=canonical_repo_root,
|
|
recovery_worktree_path=recovery_worktree_path,
|
|
branch_name=branch_name,
|
|
remote_head=expected_remote_head,
|
|
git_ops=git_ops,
|
|
)
|
|
if not prep.get("success"):
|
|
journal["phase"] = PHASE_RECOVERY_WORKTREE
|
|
journal["last_error"] = prep.get("reasons")
|
|
save_journal(journal, journal_dir=journal_dir)
|
|
return {
|
|
"success": False,
|
|
"performed": True,
|
|
"outcome": REFUSED,
|
|
"reasons": list(prep.get("reasons") or ["recovery worktree failed"]),
|
|
"journal": journal,
|
|
"evidence": prep,
|
|
}
|
|
if prep.get("created"):
|
|
journal["artifacts_created"]["recovery_worktree"] = True
|
|
journal["phase"] = PHASE_RECOVERY_WORKTREE
|
|
save_journal(journal, journal_dir=journal_dir)
|
|
if interrupt_after_phase == PHASE_RECOVERY_WORKTREE:
|
|
return {
|
|
"success": False,
|
|
"performed": True,
|
|
"outcome": "INTERRUPTED",
|
|
"reasons": ["interrupted after recovery worktree creation (test harness)"],
|
|
"journal": journal,
|
|
"evidence": prep,
|
|
}
|
|
|
|
# Phase 4: conflict detection + dirty apply (source frozen).
|
|
conflicts = detect_path_conflicts(
|
|
dirty_paths=list(expected_dirty_fingerprints.keys()),
|
|
local_head_contents=local_head_contents,
|
|
remote_head_contents=remote_head_contents,
|
|
dirty_contents=dirty_contents,
|
|
)
|
|
conflict_paths = {c["path"] for c in conflicts}
|
|
written = apply_dirty_bytes(
|
|
recovery_worktree=recovery_worktree_path,
|
|
dirty_contents=dirty_contents,
|
|
conflict_paths=conflict_paths,
|
|
)
|
|
conflict_state_path = write_conflict_state(recovery_worktree_path, conflicts)
|
|
journal["conflicts"] = list(conflicts)
|
|
journal["written_paths"] = written
|
|
journal["conflict_state_path"] = conflict_state_path
|
|
journal["artifacts_created"]["dirty_applied"] = True
|
|
journal["phase"] = PHASE_DIRTY_APPLIED
|
|
# Prove source worktree still exists and was not deleted.
|
|
journal["source_still_present"] = os.path.isdir(source_worktree_path)
|
|
save_journal(journal, journal_dir=journal_dir)
|
|
|
|
# #860 F4: Do NOT finalize session binding if conflicts remain.
|
|
if conflicts:
|
|
return {
|
|
"success": False,
|
|
"performed": True,
|
|
"outcome": CONFLICTS_PRESENT,
|
|
"reasons": [
|
|
"conflicts present: manual resolution required before session binding (fail closed)"
|
|
],
|
|
"conflicts": list(conflicts),
|
|
"recovery_worktree_path": recovery_worktree_path,
|
|
"source_worktree_path": source_worktree_path,
|
|
"source_frozen": True,
|
|
"journal": journal,
|
|
"evidence": {
|
|
"written_paths": written,
|
|
"conflict_state_path": conflict_state_path,
|
|
"accepted_head": expected_remote_head,
|
|
},
|
|
}
|
|
|
|
# Phase 5: bind session (optional for pure assessor tests).
|
|
pid = session_pid if session_pid is not None else os.getpid()
|
|
lock_record = build_recovery_lock_record(
|
|
existing_lock=existing_lock,
|
|
issue_number=issue_number,
|
|
branch_name=branch_name,
|
|
recovery_worktree_path=recovery_worktree_path,
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
identity=identity,
|
|
profile=profile,
|
|
expected_remote_head=expected_remote_head,
|
|
source_worktree_path=source_worktree_path,
|
|
conflicts=conflicts,
|
|
session_pid=pid,
|
|
)
|
|
if bind_lock:
|
|
if lock_writer is None:
|
|
import issue_lock_store as _ils
|
|
|
|
prior_gen = None
|
|
try:
|
|
prior_gen = _ils.lock_generation(dict(existing_lock))
|
|
except Exception:
|
|
prior_gen = None
|
|
_ils.bind_session_lock(
|
|
lock_record,
|
|
expected_generation=prior_gen,
|
|
recovery_sanctioned=True,
|
|
)
|
|
else:
|
|
lock_writer(lock_record)
|
|
journal["artifacts_created"]["binding"] = True
|
|
journal["phase"] = PHASE_BINDING
|
|
save_journal(journal, journal_dir=journal_dir)
|
|
if interrupt_after_phase == PHASE_BINDING:
|
|
return {
|
|
"success": False,
|
|
"performed": True,
|
|
"outcome": "INTERRUPTED",
|
|
"reasons": ["interrupted after binding (test harness)"],
|
|
"journal": journal,
|
|
"evidence": {"lock_record": lock_record},
|
|
}
|
|
|
|
journal["phase"] = PHASE_COMPLETE
|
|
journal["complete"] = True
|
|
save_journal(journal, journal_dir=journal_dir)
|
|
|
|
return {
|
|
"success": True,
|
|
"performed": True,
|
|
"outcome": RECOVERY_COMPLETED,
|
|
"reasons": [],
|
|
"conflicts": [],
|
|
"recovery_worktree_path": recovery_worktree_path,
|
|
"source_worktree_path": source_worktree_path,
|
|
"source_frozen": True,
|
|
"lock_record": lock_record,
|
|
"journal": journal,
|
|
"evidence": {
|
|
"written_paths": written,
|
|
"conflict_state_path": conflict_state_path,
|
|
"accepted_head": expected_remote_head,
|
|
"recovery_provenance": "dirty_orphan_recovery",
|
|
},
|
|
}
|
|
|
|
|
|
def preflight_recognizes_recovered_provenance(
|
|
lock: Mapping[str, Any] | None,
|
|
) -> dict[str, Any]:
|
|
"""Whether commit/publication preflights should accept recovered provenance."""
|
|
if not lock:
|
|
return {"recognized": False, "reasons": ["no lock"]}
|
|
rec = lock.get("dirty_orphan_recovery")
|
|
if not isinstance(rec, Mapping) or not rec.get("recovered"):
|
|
return {"recognized": False, "reasons": ["no dirty_orphan_recovery record"]}
|
|
conflicts = rec.get("conflicts") or []
|
|
if conflicts:
|
|
return {
|
|
"recognized": False,
|
|
"reasons": [
|
|
"recovery conflicts remain; author must resolve before "
|
|
"commit/publication preflight"
|
|
],
|
|
"conflicts": list(conflicts),
|
|
}
|
|
if not _text(lock.get("worktree_path")):
|
|
return {"recognized": False, "reasons": ["recovered lock missing worktree"]}
|
|
if _recorded_pid(lock) is None:
|
|
return {
|
|
"recognized": False,
|
|
"reasons": ["recovered lock still PID-less; binding incomplete"],
|
|
}
|
|
return {
|
|
"recognized": True,
|
|
"reasons": [],
|
|
"recovery_worktree_path": rec.get("recovery_worktree_path"),
|
|
"source_worktree_path": rec.get("source_worktree_path"),
|
|
"accepted_head": rec.get("accepted_head"),
|
|
}
|