Merge pull request 'feat: prove reviewer worktree ownership before reset (Closes #312)' (#362) from feat/issue-312-worktree-ownership into master
This commit was merged in pull request #362.
This commit is contained in:
@@ -4249,3 +4249,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs):
|
||||
from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_worktree_ownership_report(report_text, **kwargs):
|
||||
"""#312: prove session-owned or safe-reuse worktree before reset/validation."""
|
||||
from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
@@ -0,0 +1,207 @@
|
||||
"""Reviewer worktree ownership and safe-reuse proof verifier (#312)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_SESSION_OWNED_RE = re.compile(
|
||||
r"branches/(?:review-pr\d+[\w/-]*|review-[\w-]+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_WORKTREE_PATH_RE = re.compile(
|
||||
r"(?:review worktree path|worktree path|session-owned worktree)\s*:\s*(\S+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BRANCHES_PATH_RE = re.compile(r"/branches/|\bbranches/", re.IGNORECASE)
|
||||
_MAIN_CHECKOUT_RE = re.compile(
|
||||
r"main checkout|not (?:the )?main checkout|outside branches/",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_SAFE_REUSE_RE = re.compile(r"safe[- ]reuse proof", re.IGNORECASE)
|
||||
_REUSE_POLICY_RE = re.compile(
|
||||
r"(?:project policy|policy) (?:allows|allowing) (?:reuse|reset)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLEAN_TRACKED_RE = re.compile(
|
||||
r"(?:clean tracked state|tracked state\s*:\s*clean|no uncommitted tracked)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLEAN_UNTRACKED_RE = re.compile(
|
||||
r"(?:clean untracked state|untracked state\s*:\s*clean|no untracked files)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_NOT_OTHER_SESSION_RE = re.compile(
|
||||
r"not owned by (?:another|other) (?:active )?(?:task|session)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BRANCH_BEFORE_RESET_RE = re.compile(
|
||||
r"(?:branch/head before reset|head before reset|branch before reset)\s*:\s*(\S+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_RESET_TARGET_RE = re.compile(
|
||||
r"(?:reset target sha|reset target)\s*:\s*([0-9a-f]{7,40})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DESTRUCTIVE_CMD_RE = re.compile(
|
||||
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
|
||||
r"(?:reset\s+--hard|clean(?:\s+-[A-Za-z]+)*|checkout\s+(?!HEAD\s+--)|switch\s+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_WORKSPACE_NONE_RE = re.compile(r"workspace mutations\s*:\s*none", re.IGNORECASE)
|
||||
_WORKTREE_MUTATION_RE = re.compile(
|
||||
r"(?:worktree(?:/index)? mutations|destructive reset)\s*:\s*(?!none\b).+",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_RESET_IN_REPORT_RE = re.compile(r"reset\s+--hard", re.IGNORECASE)
|
||||
|
||||
|
||||
def _command_text(entry) -> str:
|
||||
if isinstance(entry, dict):
|
||||
return str(entry.get("command") or "").strip()
|
||||
return str(entry or "").strip()
|
||||
|
||||
|
||||
def _destructive_commands(command_log: list | None) -> list[str]:
|
||||
return [
|
||||
cmd
|
||||
for cmd in (_command_text(entry) for entry in (command_log or []))
|
||||
if cmd and _DESTRUCTIVE_CMD_RE.search(cmd)
|
||||
]
|
||||
|
||||
|
||||
def _extract_worktree_path(text: str, session: dict) -> str:
|
||||
path = (session.get("worktree_path") or "").strip()
|
||||
if path:
|
||||
return path
|
||||
match = _WORKTREE_PATH_RE.search(text or "")
|
||||
return match.group(1).strip() if match else ""
|
||||
|
||||
|
||||
def _is_session_owned_path(path: str) -> bool:
|
||||
normalized = (path or "").replace("\\", "/")
|
||||
return bool(_SESSION_OWNED_RE.search(normalized))
|
||||
|
||||
|
||||
def _safe_reuse_fields_present(text: str) -> list[str]:
|
||||
missing: list[str] = []
|
||||
if not _WORKTREE_PATH_RE.search(text):
|
||||
missing.append("exact worktree path")
|
||||
if not _BRANCHES_PATH_RE.search(text):
|
||||
missing.append("worktree inside branches/")
|
||||
if not _MAIN_CHECKOUT_RE.search(text):
|
||||
missing.append("worktree is not the main checkout")
|
||||
if not _NOT_OTHER_SESSION_RE.search(text):
|
||||
missing.append("worktree not owned by another active session")
|
||||
if not _CLEAN_TRACKED_RE.search(text):
|
||||
missing.append("clean tracked state")
|
||||
if not _CLEAN_UNTRACKED_RE.search(text):
|
||||
missing.append("clean untracked state")
|
||||
if not _BRANCH_BEFORE_RESET_RE.search(text):
|
||||
missing.append("branch/head before reset")
|
||||
if not _RESET_TARGET_RE.search(text):
|
||||
missing.append("reset target SHA")
|
||||
if not _REUSE_POLICY_RE.search(text):
|
||||
missing.append("explicit project policy allowing reuse/reset")
|
||||
return missing
|
||||
|
||||
|
||||
def assess_worktree_ownership_report(
|
||||
report_text: str,
|
||||
*,
|
||||
ownership_session: dict | None = None,
|
||||
command_log: list | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Prove reviewer worktree ownership before reset or validation (#312)."""
|
||||
text = report_text or ""
|
||||
session = dict(ownership_session or {})
|
||||
reasons: list[str] = []
|
||||
|
||||
worktree_path = _extract_worktree_path(text, session)
|
||||
destructive = _destructive_commands(command_log or session.get("command_log"))
|
||||
if not destructive and session.get("destructive_reset"):
|
||||
destructive = ["git reset --hard (session)"]
|
||||
|
||||
session_owned = bool(
|
||||
session.get("session_owned")
|
||||
or (worktree_path and _is_session_owned_path(worktree_path))
|
||||
)
|
||||
safe_reuse = bool(session.get("safe_reuse") or _SAFE_REUSE_RE.search(text))
|
||||
main_checkout = bool(session.get("main_checkout"))
|
||||
dirty_tracked = session.get("dirty_tracked")
|
||||
dirty_untracked = session.get("dirty_untracked")
|
||||
other_session_owned = bool(session.get("other_session_owned"))
|
||||
|
||||
if worktree_path or destructive or session.get("validation_ran"):
|
||||
if not worktree_path:
|
||||
reasons.append("report missing exact review worktree path")
|
||||
elif main_checkout or "branches/" not in worktree_path.replace("\\", "/"):
|
||||
reasons.append("review worktree must be inside branches/, not the main checkout")
|
||||
elif not _BRANCHES_PATH_RE.search(worktree_path.replace("\\", "/")):
|
||||
reasons.append("review worktree path must be under branches/")
|
||||
|
||||
if other_session_owned and not safe_reuse:
|
||||
reasons.append(
|
||||
"reused worktree appears owned by another active task/session; "
|
||||
"safe-reuse proof required"
|
||||
)
|
||||
|
||||
if dirty_tracked:
|
||||
reasons.append(
|
||||
"worktree has uncommitted tracked changes before reset or validation"
|
||||
)
|
||||
if dirty_untracked and safe_reuse:
|
||||
reasons.append("safe-reuse proof requires clean untracked state")
|
||||
|
||||
if safe_reuse and not session_owned:
|
||||
reasons.extend(
|
||||
f"safe-reuse proof missing {field}"
|
||||
for field in _safe_reuse_fields_present(text)
|
||||
)
|
||||
|
||||
if destructive:
|
||||
if not session_owned and not safe_reuse:
|
||||
reasons.append(
|
||||
"destructive worktree commands forbidden without session-owned "
|
||||
"worktree or safe-reuse proof"
|
||||
)
|
||||
if _WORKSPACE_NONE_RE.search(text) and (
|
||||
_RESET_IN_REPORT_RE.search(text) or any("reset" in c.lower() for c in destructive)
|
||||
):
|
||||
reasons.append(
|
||||
"final report must not claim 'Workspace mutations: none' when "
|
||||
"git reset --hard occurred"
|
||||
)
|
||||
if not _WORKTREE_MUTATION_RE.search(text) and not _RESET_IN_REPORT_RE.search(text):
|
||||
reasons.append(
|
||||
"destructive reset must be reported under Worktree/index mutations "
|
||||
"or destructive reset operations"
|
||||
)
|
||||
|
||||
if (
|
||||
worktree_path
|
||||
and not session_owned
|
||||
and not safe_reuse
|
||||
and (destructive or session.get("validation_ran"))
|
||||
and not _is_session_owned_path(worktree_path)
|
||||
):
|
||||
reasons.append(
|
||||
"reused branch-named worktree requires safe-reuse proof before reset or validation"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": reasons,
|
||||
"worktree_path": worktree_path or None,
|
||||
"session_owned": session_owned,
|
||||
"safe_reuse": safe_reuse,
|
||||
"destructive_commands": destructive,
|
||||
"safe_next_action": (
|
||||
"use a fresh session-owned review worktree under branches/review-pr<N>-* "
|
||||
"or document full safe-reuse proof before destructive reset"
|
||||
if reasons
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
@@ -146,3 +146,9 @@ def test_non_mergeable_skip_verifier_exported():
|
||||
from review_proofs import assess_non_mergeable_skip_proof
|
||||
|
||||
assert callable(assess_non_mergeable_skip_proof)
|
||||
|
||||
|
||||
def test_worktree_ownership_verifier_exported():
|
||||
from review_proofs import assess_worktree_ownership_report
|
||||
|
||||
assert callable(assess_worktree_ownership_report)
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Tests for reviewer worktree ownership verifier (#312)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from reviewer_worktree_ownership import assess_worktree_ownership_report # noqa: E402
|
||||
|
||||
|
||||
def _fresh_review_report() -> str:
|
||||
return "\n".join([
|
||||
"Review worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Worktree is inside branches/ and not the main checkout.",
|
||||
"Tracked state: clean; untracked state: clean.",
|
||||
"Official PR-head validation on session-owned worktree.",
|
||||
"Worktree mutations: none",
|
||||
])
|
||||
|
||||
|
||||
def _safe_reuse_report() -> str:
|
||||
return "\n".join([
|
||||
"Safe-reuse proof for existing review worktree.",
|
||||
"Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
"Worktree inside branches/; not the main checkout.",
|
||||
"Not owned by another active task/session.",
|
||||
"Clean tracked state; clean untracked state.",
|
||||
"Branch/head before reset: feat/issue-260-example",
|
||||
"Reset target SHA: abc123def4567890abcdef1234567890abcdef12",
|
||||
"Project policy allows reuse/reset for this clean review worktree.",
|
||||
"Worktree mutations: git reset --hard FETCH_HEAD",
|
||||
"Destructive reset operations: git reset --hard FETCH_HEAD",
|
||||
])
|
||||
|
||||
|
||||
class TestWorktreeOwnership(unittest.TestCase):
|
||||
def test_fresh_session_owned_passes(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_fresh_review_report(),
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
},
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
self.assertTrue(result["session_owned"])
|
||||
|
||||
def test_safe_reuse_with_reset_passes(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_safe_reuse_report(),
|
||||
ownership_session={
|
||||
"safe_reuse": True,
|
||||
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_destructive_reset_without_proof_blocks(self):
|
||||
report = "\n".join([
|
||||
"Review worktree path: branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
"Worktree mutations: git reset --hard FETCH_HEAD",
|
||||
])
|
||||
result = assess_worktree_ownership_report(
|
||||
report,
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_reused_dirty_tracked_blocks(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_fresh_review_report(),
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
"dirty_tracked": True,
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_safe_reuse_dirty_untracked_blocks(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
_safe_reuse_report(),
|
||||
ownership_session={
|
||||
"safe_reuse": True,
|
||||
"dirty_untracked": True,
|
||||
"worktree_path": "branches/docs-issue-260-forbid-commit-fallbacks",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("untracked" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_workspace_none_with_reset_blocks(self):
|
||||
report = "\n".join([
|
||||
"Review worktree path: branches/review-pr280-feat-issue-275-claim",
|
||||
"Workspace mutations: none",
|
||||
"Worktree mutations: git reset --hard FETCH_HEAD",
|
||||
])
|
||||
result = assess_worktree_ownership_report(
|
||||
report,
|
||||
ownership_session={
|
||||
"session_owned": True,
|
||||
"worktree_path": "branches/review-pr280-feat-issue-275-claim",
|
||||
},
|
||||
command_log=["git reset --hard FETCH_HEAD"],
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_main_checkout_blocks(self):
|
||||
result = assess_worktree_ownership_report(
|
||||
"Review worktree path: /Users/dev/Gitea-Tools",
|
||||
ownership_session={
|
||||
"validation_ran": True,
|
||||
"main_checkout": True,
|
||||
"worktree_path": "/Users/dev/Gitea-Tools",
|
||||
},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
|
||||
class TestExport(unittest.TestCase):
|
||||
def test_review_proofs_reexport(self):
|
||||
from review_proofs import assess_worktree_ownership_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user