feat: align claim/lock gates with branches-only worktrees (Closes #275) #280
+111
-8
@@ -171,6 +171,9 @@ _preflight_whoami_violation_files: list[str] = []
|
|||||||
_preflight_capability_violation_files: list[str] = []
|
_preflight_capability_violation_files: list[str] = []
|
||||||
_preflight_reviewer_violation_files: list[str] = []
|
_preflight_reviewer_violation_files: list[str] = []
|
||||||
|
|
||||||
|
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
|
||||||
|
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
||||||
|
|
||||||
|
|
||||||
def _preflight_in_test_mode() -> bool:
|
def _preflight_in_test_mode() -> bool:
|
||||||
return "pytest" in sys.modules or "unittest" in sys.modules
|
return "pytest" in sys.modules or "unittest" in sys.modules
|
||||||
@@ -184,19 +187,47 @@ def _ensure_process_start_porcelain() -> str:
|
|||||||
return _process_start_porcelain
|
return _process_start_porcelain
|
||||||
|
|
||||||
|
|
||||||
def _get_workspace_porcelain() -> str:
|
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
||||||
|
"""Resolve the workspace root inspected by pre-flight guards."""
|
||||||
|
path = (worktree_path or "").strip()
|
||||||
|
if not path:
|
||||||
|
path = (os.environ.get(ACTIVE_WORKTREE_ENV) or "").strip()
|
||||||
|
if not path:
|
||||||
|
path = (os.environ.get(AUTHOR_WORKTREE_ENV) or "").strip()
|
||||||
|
if not path:
|
||||||
|
path = PROJECT_ROOT
|
||||||
|
return os.path.realpath(os.path.abspath(path))
|
||||||
|
|
||||||
|
|
||||||
|
def _get_git_root(path: str) -> str | None:
|
||||||
|
try:
|
||||||
|
res = subprocess.run(
|
||||||
|
["git", "-C", path, "rev-parse", "--show-toplevel"],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
except Exception:
|
||||||
|
return None
|
||||||
|
if res.returncode != 0:
|
||||||
|
return None
|
||||||
|
return (res.stdout or "").strip() or None
|
||||||
|
|
||||||
|
|
||||||
|
def _get_workspace_porcelain(worktree_path: str | None = None) -> str:
|
||||||
"""Return tracked-workspace porcelain for pre-flight attribution."""
|
"""Return tracked-workspace porcelain for pre-flight attribution."""
|
||||||
if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
|
if os.environ.get("GITEA_TEST_FORCE_DIRTY"):
|
||||||
return " M __gitea_test_force_dirty__.py\n"
|
return " M __gitea_test_force_dirty__.py\n"
|
||||||
override = os.environ.get("GITEA_TEST_PORCELAIN")
|
override = os.environ.get("GITEA_TEST_PORCELAIN")
|
||||||
if override is not None:
|
if override is not None:
|
||||||
return override
|
return override
|
||||||
|
workspace = _resolve_preflight_workspace_path(worktree_path)
|
||||||
try:
|
try:
|
||||||
res = subprocess.run(
|
res = subprocess.run(
|
||||||
["git", "status", "--porcelain"],
|
["git", "status", "--porcelain"],
|
||||||
capture_output=True,
|
capture_output=True,
|
||||||
text=True,
|
text=True,
|
||||||
cwd=PROJECT_ROOT,
|
cwd=workspace,
|
||||||
)
|
)
|
||||||
return res.stdout or ""
|
return res.stdout or ""
|
||||||
except Exception:
|
except Exception:
|
||||||
@@ -234,7 +265,35 @@ def _format_preflight_files(files: list[str]) -> str:
|
|||||||
return ", ".join(files)
|
return ", ".join(files)
|
||||||
|
|
||||||
|
|
||||||
def assess_preflight_status() -> dict:
|
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
|
||||||
|
workspace = _resolve_preflight_workspace_path(worktree_path)
|
||||||
|
inspected_root = _get_git_root(workspace)
|
||||||
|
control_root = os.path.realpath(PROJECT_ROOT)
|
||||||
|
active_root = os.path.realpath(inspected_root or workspace)
|
||||||
|
if active_root == control_root:
|
||||||
|
dirty_scope = "control checkout"
|
||||||
|
else:
|
||||||
|
dirty_scope = "active task workspace"
|
||||||
|
return {
|
||||||
|
"mcp_server_process_root": control_root,
|
||||||
|
"active_task_workspace_root": active_root,
|
||||||
|
"inspected_git_root": inspected_root,
|
||||||
|
"dirty_files": list(dirty_files),
|
||||||
|
"dirty_scope": dirty_scope,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _format_preflight_workspace_details(details: dict) -> str:
|
||||||
|
return (
|
||||||
|
f"MCP server process root: {details.get('mcp_server_process_root')}; "
|
||||||
|
f"active task workspace root: {details.get('active_task_workspace_root')}; "
|
||||||
|
f"inspected git root: {details.get('inspected_git_root')}; "
|
||||||
|
f"dirty files: {_format_preflight_files(details.get('dirty_files') or [])}; "
|
||||||
|
f"dirty scope: {details.get('dirty_scope')}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def assess_preflight_status(worktree_path: str | None = None) -> dict:
|
||||||
"""Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
|
"""Non-throwing pre-flight readiness for runtime-context alignment (#252)."""
|
||||||
reasons: list[str] = []
|
reasons: list[str] = []
|
||||||
if not _preflight_whoami_called:
|
if not _preflight_whoami_called:
|
||||||
@@ -245,6 +304,25 @@ def assess_preflight_status() -> dict:
|
|||||||
reasons.append(
|
reasons.append(
|
||||||
"Task capability (gitea_resolve_task_capability) has not been resolved"
|
"Task capability (gitea_resolve_task_capability) has not been resolved"
|
||||||
)
|
)
|
||||||
|
workspace_details = None
|
||||||
|
if worktree_path:
|
||||||
|
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
|
||||||
|
workspace_details = _preflight_workspace_details(worktree_path, dirty_files)
|
||||||
|
if dirty_files:
|
||||||
|
reasons.append(
|
||||||
|
"Active task workspace has tracked file edits before mutation "
|
||||||
|
f"({_format_preflight_workspace_details(workspace_details)})"
|
||||||
|
)
|
||||||
|
return {
|
||||||
|
"preflight_ready": not reasons,
|
||||||
|
"preflight_block_reasons": reasons,
|
||||||
|
"preflight_whoami_verified": _preflight_whoami_called,
|
||||||
|
"preflight_capability_resolved": _preflight_capability_called,
|
||||||
|
"preflight_whoami_violation_files": [],
|
||||||
|
"preflight_capability_violation_files": [],
|
||||||
|
"preflight_reviewer_violation_files": [],
|
||||||
|
"preflight_workspace": workspace_details,
|
||||||
|
}
|
||||||
if _preflight_whoami_violation:
|
if _preflight_whoami_violation:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
"Workspace file edits occurred before gitea_whoami verification "
|
"Workspace file edits occurred before gitea_whoami verification "
|
||||||
@@ -278,6 +356,7 @@ def assess_preflight_status() -> dict:
|
|||||||
"preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
|
"preflight_whoami_violation_files": list(_preflight_whoami_violation_files),
|
||||||
"preflight_capability_violation_files": list(_preflight_capability_violation_files),
|
"preflight_capability_violation_files": list(_preflight_capability_violation_files),
|
||||||
"preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
|
"preflight_reviewer_violation_files": list(_preflight_reviewer_violation_files),
|
||||||
|
"preflight_workspace": _preflight_workspace_details(None, []),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -318,7 +397,7 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
|
|||||||
_preflight_resolved_role = resolved_role
|
_preflight_resolved_role = resolved_role
|
||||||
|
|
||||||
|
|
||||||
def verify_preflight_purity(remote: str | None = None):
|
def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
|
||||||
"""Verify that identity and capability were verified prior to session edits."""
|
"""Verify that identity and capability were verified prior to session edits."""
|
||||||
global _preflight_reviewer_violation_files
|
global _preflight_reviewer_violation_files
|
||||||
|
|
||||||
@@ -338,6 +417,17 @@ def verify_preflight_purity(remote: str | None = None):
|
|||||||
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
|
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if worktree_path:
|
||||||
|
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(worktree_path)))
|
||||||
|
if dirty_files:
|
||||||
|
details = _preflight_workspace_details(worktree_path, dirty_files)
|
||||||
|
raise RuntimeError(
|
||||||
|
"Pre-flight order violation: Active task workspace has tracked "
|
||||||
|
"file edits before mutation (fail closed). "
|
||||||
|
f"{_format_preflight_workspace_details(details)}"
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
if _preflight_whoami_violation:
|
if _preflight_whoami_violation:
|
||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
"Pre-flight order violation: Workspace file edits occurred before "
|
"Pre-flight order violation: Workspace file edits occurred before "
|
||||||
@@ -739,6 +829,8 @@ def gitea_create_issue(
|
|||||||
Returns:
|
Returns:
|
||||||
dict with 'number' of the created issue ('url' only with the reveal opt-in).
|
dict with 'number' of the created issue ('url' only with the reveal opt-in).
|
||||||
"""
|
"""
|
||||||
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
|
auth = _auth(h)
|
||||||
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
|
ok, block_reasons = role_session_router.check_author_mutation_after_reviewer_stop(
|
||||||
"create_issue"
|
"create_issue"
|
||||||
)
|
)
|
||||||
@@ -760,8 +852,6 @@ def gitea_create_issue(
|
|||||||
if blocked:
|
if blocked:
|
||||||
return blocked
|
return blocked
|
||||||
verify_preflight_purity(remote)
|
verify_preflight_purity(remote)
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
|
||||||
auth = _auth(h)
|
|
||||||
base = repo_api_url(h, o, r)
|
base = repo_api_url(h, o, r)
|
||||||
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
|
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
|
||||||
closed_issues = api_get_all(
|
closed_issues = api_get_all(
|
||||||
@@ -829,14 +919,23 @@ def gitea_lock_issue(
|
|||||||
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
|
f"Branch name '{branch_name}' must contain locked issue pattern '{expected_pattern}' (fail closed)"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
blocked = _profile_permission_block(
|
||||||
|
task_capability_map.required_permission("lock_issue"))
|
||||||
|
if blocked:
|
||||||
|
return blocked
|
||||||
|
|
||||||
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
|
resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
|
||||||
worktree_path, PROJECT_ROOT
|
worktree_path, PROJECT_ROOT
|
||||||
)
|
)
|
||||||
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
|
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
|
||||||
|
verify_preflight_purity(remote, worktree_path=resolved_worktree)
|
||||||
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
|
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
|
||||||
worktree_path=resolved_worktree,
|
worktree_path=resolved_worktree,
|
||||||
current_branch=git_state.get("current_branch"),
|
current_branch=git_state.get("current_branch"),
|
||||||
porcelain_status=git_state.get("porcelain_status") or "",
|
porcelain_status=git_state.get("porcelain_status") or "",
|
||||||
|
base_equivalent=git_state.get("base_equivalent"),
|
||||||
|
inspected_git_root=git_state.get("inspected_git_root"),
|
||||||
|
base_branch=git_state.get("base_branch"),
|
||||||
)
|
)
|
||||||
if lock_assessment["block"]:
|
if lock_assessment["block"]:
|
||||||
raise RuntimeError(
|
raise RuntimeError(
|
||||||
@@ -4337,6 +4436,7 @@ def _build_runtime_task_capabilities(
|
|||||||
def gitea_get_runtime_context(
|
def gitea_get_runtime_context(
|
||||||
remote: str = "dadeschools",
|
remote: str = "dadeschools",
|
||||||
host: str | None = None,
|
host: str | None = None,
|
||||||
|
worktree_path: str | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Read-only: explicit visibility into active profile, configuration model, and eligibility.
|
"""Read-only: explicit visibility into active profile, configuration model, and eligibility.
|
||||||
|
|
||||||
@@ -4424,7 +4524,7 @@ def gitea_get_runtime_context(
|
|||||||
allowed, forbidden, config
|
allowed, forbidden, config
|
||||||
)
|
)
|
||||||
|
|
||||||
preflight = assess_preflight_status()
|
preflight = assess_preflight_status(worktree_path)
|
||||||
if not preflight["preflight_ready"]:
|
if not preflight["preflight_ready"]:
|
||||||
safe_next_action = (
|
safe_next_action = (
|
||||||
"Complete pre-flight verification before mutating: call gitea_whoami, then "
|
"Complete pre-flight verification before mutating: call gitea_whoami, then "
|
||||||
@@ -4448,6 +4548,7 @@ def gitea_get_runtime_context(
|
|||||||
"safe_next_action": safe_next_action,
|
"safe_next_action": safe_next_action,
|
||||||
"preflight_ready": preflight["preflight_ready"],
|
"preflight_ready": preflight["preflight_ready"],
|
||||||
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
"preflight_block_reasons": preflight["preflight_block_reasons"],
|
||||||
|
"preflight_workspace": preflight.get("preflight_workspace"),
|
||||||
"session_capabilities": session_capabilities,
|
"session_capabilities": session_capabilities,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -4699,6 +4800,7 @@ def gitea_mark_issue(
|
|||||||
host: str | None = None,
|
host: str | None = None,
|
||||||
org: str | None = None,
|
org: str | None = None,
|
||||||
repo: str | None = None,
|
repo: str | None = None,
|
||||||
|
worktree_path: str | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Claim or release an issue via the status:in-progress label.
|
"""Claim or release an issue via the status:in-progress label.
|
||||||
|
|
||||||
@@ -4711,6 +4813,7 @@ def gitea_mark_issue(
|
|||||||
host: Override the Gitea host.
|
host: Override the Gitea host.
|
||||||
org: Override the owner/organization.
|
org: Override the owner/organization.
|
||||||
repo: Override the repository name.
|
repo: Override the repository name.
|
||||||
|
worktree_path: Active task worktree to inspect for pre-flight purity.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
dict with 'success' boolean and 'message'.
|
dict with 'success' boolean and 'message'.
|
||||||
@@ -4722,7 +4825,7 @@ def gitea_mark_issue(
|
|||||||
task_capability_map.required_permission("mark_issue"))
|
task_capability_map.required_permission("mark_issue"))
|
||||||
if blocked:
|
if blocked:
|
||||||
return blocked
|
return blocked
|
||||||
verify_preflight_purity(remote)
|
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||||
h, o, r = _resolve(remote, host, org, repo)
|
h, o, r = _resolve(remote, host, org, repo)
|
||||||
auth = _auth(h)
|
auth = _auth(h)
|
||||||
base = repo_api_url(h, o, r)
|
base = repo_api_url(h, o, r)
|
||||||
|
|||||||
+81
-14
@@ -14,7 +14,7 @@ import subprocess
|
|||||||
from reviewer_worktree import parse_dirty_tracked_files
|
from reviewer_worktree import parse_dirty_tracked_files
|
||||||
|
|
||||||
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
||||||
BASE_BRANCHES = frozenset({"master", "main"})
|
BASE_BRANCHES = frozenset({"master", "main", "dev"})
|
||||||
|
|
||||||
|
|
||||||
def resolve_author_worktree_path(
|
def resolve_author_worktree_path(
|
||||||
@@ -50,9 +50,28 @@ def read_worktree_git_state(worktree_path: str) -> dict:
|
|||||||
text=True,
|
text=True,
|
||||||
check=False,
|
check=False,
|
||||||
)
|
)
|
||||||
|
root_res = subprocess.run(
|
||||||
|
["git", "-C", path, "rev-parse", "--show-toplevel"],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
head_res = subprocess.run(
|
||||||
|
["git", "-C", path, "rev-parse", "HEAD"],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None
|
||||||
|
base_branch, base_sha = _find_matching_base_ref(path, head_sha)
|
||||||
return {
|
return {
|
||||||
"current_branch": current_branch,
|
"current_branch": current_branch,
|
||||||
"porcelain_status": status_res.stdout or "",
|
"porcelain_status": status_res.stdout or "",
|
||||||
|
"inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None,
|
||||||
|
"head_sha": head_sha,
|
||||||
|
"base_branch": base_branch,
|
||||||
|
"base_sha": base_sha,
|
||||||
|
"base_equivalent": bool(head_sha and base_sha and head_sha == base_sha),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -61,6 +80,9 @@ def assess_issue_lock_worktree(
|
|||||||
worktree_path: str,
|
worktree_path: str,
|
||||||
current_branch: str | None,
|
current_branch: str | None,
|
||||||
porcelain_status: str,
|
porcelain_status: str,
|
||||||
|
base_equivalent: bool | None = None,
|
||||||
|
inspected_git_root: str | None = None,
|
||||||
|
base_branch: str | None = None,
|
||||||
base_branches: frozenset[str] | None = None,
|
base_branches: frozenset[str] | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
"""Fail closed when lock preconditions are not met on the declared worktree."""
|
"""Fail closed when lock preconditions are not met on the declared worktree."""
|
||||||
@@ -77,21 +99,39 @@ def assess_issue_lock_worktree(
|
|||||||
if dirty_files:
|
if dirty_files:
|
||||||
reasons.append(
|
reasons.append(
|
||||||
"tracked file edits exist before issue lock; "
|
"tracked file edits exist before issue lock; "
|
||||||
"lock must precede implementation work"
|
f"lock must precede implementation work in '{path}' "
|
||||||
)
|
f"(dirty files: {', '.join(dirty_files)})"
|
||||||
if not branch:
|
|
||||||
reasons.append(
|
|
||||||
"current branch unknown (detached HEAD?); issue lock must be taken "
|
|
||||||
f"from base branch ({_base_list(bases)})"
|
|
||||||
)
|
|
||||||
elif branch not in bases:
|
|
||||||
reasons.append(
|
|
||||||
f"issue lock must be taken from base branch ({_base_list(bases)}), "
|
|
||||||
f"not '{branch}'"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
if base_equivalent is False:
|
||||||
|
reasons.append(
|
||||||
|
"issue lock worktree must be base-equivalent to one of "
|
||||||
|
f"{_base_list(bases)} before implementation work; inspected "
|
||||||
|
f"branch '{branch or '(detached)'}' at '{path}'"
|
||||||
|
)
|
||||||
|
elif base_equivalent is None:
|
||||||
|
if not branch:
|
||||||
|
reasons.append(
|
||||||
|
"current branch unknown (detached HEAD?); issue lock base-equivalence "
|
||||||
|
f"to {_base_list(bases)} could not be proven"
|
||||||
|
)
|
||||||
|
elif branch not in bases:
|
||||||
|
reasons.append(
|
||||||
|
"issue lock worktree base-equivalence could not be proven; "
|
||||||
|
f"branch '{branch}' is not {_base_list(bases)}"
|
||||||
|
)
|
||||||
|
|
||||||
proven = not reasons
|
proven = not reasons
|
||||||
return _assessment(proven, reasons, path, branch or None, dirty_files)
|
return _assessment(
|
||||||
|
proven,
|
||||||
|
reasons,
|
||||||
|
path,
|
||||||
|
branch or None,
|
||||||
|
dirty_files,
|
||||||
|
inspected_git_root=inspected_git_root,
|
||||||
|
base_branch=base_branch,
|
||||||
|
base_equivalent=base_equivalent,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
def format_issue_lock_worktree_error(assessment: dict) -> str:
|
def format_issue_lock_worktree_error(assessment: dict) -> str:
|
||||||
@@ -145,12 +185,39 @@ def _assessment(
|
|||||||
worktree_path: str,
|
worktree_path: str,
|
||||||
current_branch: str | None,
|
current_branch: str | None,
|
||||||
dirty_files: list[str],
|
dirty_files: list[str],
|
||||||
|
*,
|
||||||
|
inspected_git_root: str | None = None,
|
||||||
|
base_branch: str | None = None,
|
||||||
|
base_equivalent: bool | None = None,
|
||||||
) -> dict:
|
) -> dict:
|
||||||
return {
|
return {
|
||||||
"proven": proven,
|
"proven": proven,
|
||||||
"block": not proven,
|
"block": not proven,
|
||||||
"reasons": reasons,
|
"reasons": reasons,
|
||||||
"worktree_path": worktree_path or None,
|
"worktree_path": worktree_path or None,
|
||||||
|
"inspected_git_root": inspected_git_root,
|
||||||
"current_branch": current_branch,
|
"current_branch": current_branch,
|
||||||
"dirty_files": dirty_files,
|
"dirty_files": dirty_files,
|
||||||
}
|
"base_branch": base_branch,
|
||||||
|
"base_equivalent": base_equivalent,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def _find_matching_base_ref(path: str, head_sha: str | None) -> tuple[str | None, str | None]:
|
||||||
|
"""Return the stable branch ref whose commit matches HEAD, if any."""
|
||||||
|
if not head_sha:
|
||||||
|
return None, None
|
||||||
|
candidates: list[str] = []
|
||||||
|
for branch in sorted(BASE_BRANCHES):
|
||||||
|
candidates.extend((f"origin/{branch}", branch))
|
||||||
|
for ref in candidates:
|
||||||
|
res = subprocess.run(
|
||||||
|
["git", "-C", path, "rev-parse", "--verify", ref],
|
||||||
|
capture_output=True,
|
||||||
|
text=True,
|
||||||
|
check=False,
|
||||||
|
)
|
||||||
|
sha = (res.stdout or "").strip()
|
||||||
|
if res.returncode == 0 and sha == head_sha:
|
||||||
|
return ref, sha
|
||||||
|
return None, None
|
||||||
|
|||||||
@@ -28,6 +28,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
|||||||
"permission": "gitea.issue.comment",
|
"permission": "gitea.issue.comment",
|
||||||
"role": "author",
|
"role": "author",
|
||||||
},
|
},
|
||||||
|
"lock_issue": {
|
||||||
|
"permission": "gitea.issue.comment",
|
||||||
|
"role": "author",
|
||||||
|
},
|
||||||
"set_issue_labels": {
|
"set_issue_labels": {
|
||||||
"permission": "gitea.issue.comment",
|
"permission": "gitea.issue.comment",
|
||||||
"role": "author",
|
"role": "author",
|
||||||
@@ -119,4 +123,4 @@ def required_role(task: str) -> str:
|
|||||||
|
|
||||||
def tool_required_permission(tool_name: str) -> str:
|
def tool_required_permission(tool_name: str) -> str:
|
||||||
"""Return the operation an issue-mutating tool must gate on."""
|
"""Return the operation an issue-mutating tool must gate on."""
|
||||||
return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])
|
return required_permission(ISSUE_MUTATION_TOOL_TASKS[tool_name])
|
||||||
|
|||||||
@@ -36,7 +36,32 @@ class TestIssueLockWorktreeAssessment(unittest.TestCase):
|
|||||||
porcelain_status="",
|
porcelain_status="",
|
||||||
)
|
)
|
||||||
self.assertFalse(result["proven"])
|
self.assertFalse(result["proven"])
|
||||||
self.assertIn("issue lock must be taken from base branch", result["reasons"][0])
|
self.assertIn("base-equivalence could not be proven", result["reasons"][0])
|
||||||
|
|
||||||
|
def test_base_equivalent_feature_branch_passes(self):
|
||||||
|
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||||
|
worktree_path="/repo/branches/issue-275",
|
||||||
|
current_branch="feat/issue-275-claim-lock-branches-worktree",
|
||||||
|
porcelain_status="",
|
||||||
|
base_equivalent=True,
|
||||||
|
inspected_git_root="/repo/branches/issue-275",
|
||||||
|
base_branch="origin/master",
|
||||||
|
)
|
||||||
|
self.assertTrue(result["proven"])
|
||||||
|
self.assertFalse(result["block"])
|
||||||
|
self.assertEqual(result["base_branch"], "origin/master")
|
||||||
|
|
||||||
|
def test_non_base_equivalent_branch_fails(self):
|
||||||
|
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||||
|
worktree_path="/repo/branches/issue-275",
|
||||||
|
current_branch="feat/issue-275-claim-lock-branches-worktree",
|
||||||
|
porcelain_status="",
|
||||||
|
base_equivalent=False,
|
||||||
|
inspected_git_root="/repo/branches/issue-275",
|
||||||
|
base_branch=None,
|
||||||
|
)
|
||||||
|
self.assertFalse(result["proven"])
|
||||||
|
self.assertIn("must be base-equivalent", result["reasons"][0])
|
||||||
|
|
||||||
def test_untracked_files_do_not_block(self):
|
def test_untracked_files_do_not_block(self):
|
||||||
result = issue_lock_worktree.assess_issue_lock_worktree(
|
result = issue_lock_worktree.assess_issue_lock_worktree(
|
||||||
@@ -96,4 +121,4 @@ class TestPrWorktreeMatch(unittest.TestCase):
|
|||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main()
|
unittest.main()
|
||||||
|
|||||||
@@ -2937,20 +2937,31 @@ class TestVerifyMutationAuthority(unittest.TestCase):
|
|||||||
mcp_server.verify_mutation_authority("prgs")
|
mcp_server.verify_mutation_authority("prgs")
|
||||||
|
|
||||||
|
|
||||||
|
def _clean_master_git_state_for_lock():
|
||||||
|
return {
|
||||||
|
"current_branch": "master",
|
||||||
|
"porcelain_status": "",
|
||||||
|
"base_equivalent": True,
|
||||||
|
"inspected_git_root": "/scratch/wt",
|
||||||
|
"base_branch": "origin/master",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
class TestIssueLocking(unittest.TestCase):
|
class TestIssueLocking(unittest.TestCase):
|
||||||
"""Test issue locking and PR gating constraints."""
|
"""Test issue locking and PR gating constraints."""
|
||||||
|
|
||||||
@staticmethod
|
def setUp(self):
|
||||||
def _clean_master_git_state():
|
self._env_patcher = patch.dict(os.environ, ISSUE_WRITE_ENV, clear=True)
|
||||||
return {"current_branch": "master", "porcelain_status": ""}
|
self._env_patcher.start()
|
||||||
|
|
||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
|
self._env_patcher.stop()
|
||||||
if os.path.exists(ISSUE_LOCK_FILE):
|
if os.path.exists(ISSUE_LOCK_FILE):
|
||||||
os.remove(ISSUE_LOCK_FILE)
|
os.remove(ISSUE_LOCK_FILE)
|
||||||
|
|
||||||
@patch(
|
@patch(
|
||||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
return_value=_clean_master_git_state_for_lock(),
|
||||||
)
|
)
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_get_all")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@@ -2970,7 +2981,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
|
|
||||||
@patch(
|
@patch(
|
||||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
return_value=_clean_master_git_state_for_lock(),
|
||||||
)
|
)
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_get_all")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@@ -2987,7 +2998,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
|
|
||||||
@patch(
|
@patch(
|
||||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
return_value=_clean_master_git_state_for_lock(),
|
||||||
)
|
)
|
||||||
@patch("mcp_server.api_get_all")
|
@patch("mcp_server.api_get_all")
|
||||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||||
@@ -3008,7 +3019,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
|
scratch = "/tmp/gitea-tools-author-scratch/issue-249-clean"
|
||||||
with patch(
|
with patch(
|
||||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||||
return_value={"current_branch": "master", "porcelain_status": ""},
|
return_value=_clean_master_git_state_for_lock(),
|
||||||
) as mock_git:
|
) as mock_git:
|
||||||
res = gitea_lock_issue(
|
res = gitea_lock_issue(
|
||||||
issue_number=249,
|
issue_number=249,
|
||||||
@@ -3028,6 +3039,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
return_value={
|
return_value={
|
||||||
"current_branch": "master",
|
"current_branch": "master",
|
||||||
"porcelain_status": " M gitea_mcp_server.py\n",
|
"porcelain_status": " M gitea_mcp_server.py\n",
|
||||||
|
"base_equivalent": True,
|
||||||
},
|
},
|
||||||
):
|
):
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
with self.assertRaises(RuntimeError) as ctx:
|
||||||
@@ -3047,6 +3059,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
return_value={
|
return_value={
|
||||||
"current_branch": "feat/issue-243-forbidden-git-gaps",
|
"current_branch": "feat/issue-243-forbidden-git-gaps",
|
||||||
"porcelain_status": "",
|
"porcelain_status": "",
|
||||||
|
"base_equivalent": False,
|
||||||
},
|
},
|
||||||
):
|
):
|
||||||
with self.assertRaises(RuntimeError) as ctx:
|
with self.assertRaises(RuntimeError) as ctx:
|
||||||
@@ -3056,7 +3069,7 @@ class TestIssueLocking(unittest.TestCase):
|
|||||||
remote="prgs",
|
remote="prgs",
|
||||||
worktree_path="/tmp/scratch/wt",
|
worktree_path="/tmp/scratch/wt",
|
||||||
)
|
)
|
||||||
self.assertIn("issue lock must be taken from base branch", str(ctx.exception))
|
self.assertIn("base-equivalent", str(ctx.exception))
|
||||||
|
|
||||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||||
return_value=(True, []))
|
return_value=(True, []))
|
||||||
@@ -3199,7 +3212,12 @@ class TestPreflightVerification(unittest.TestCase):
|
|||||||
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
|
mcp_server._preflight_whoami_violation_files = self.orig_whoami_files
|
||||||
mcp_server._preflight_capability_violation_files = self.orig_capability_files
|
mcp_server._preflight_capability_violation_files = self.orig_capability_files
|
||||||
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
|
mcp_server._preflight_reviewer_violation_files = self.orig_reviewer_files
|
||||||
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
|
for key in (
|
||||||
|
"GITEA_TEST_FORCE_DIRTY",
|
||||||
|
"GITEA_TEST_PORCELAIN",
|
||||||
|
"GITEA_ACTIVE_WORKTREE",
|
||||||
|
"GITEA_AUTHOR_WORKTREE",
|
||||||
|
):
|
||||||
if key in os.environ:
|
if key in os.environ:
|
||||||
del os.environ[key]
|
del os.environ[key]
|
||||||
|
|
||||||
@@ -3289,3 +3307,35 @@ class TestPreflightVerification(unittest.TestCase):
|
|||||||
status = mcp_server.assess_preflight_status()
|
status = mcp_server.assess_preflight_status()
|
||||||
self.assertFalse(status["preflight_ready"])
|
self.assertFalse(status["preflight_ready"])
|
||||||
self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])
|
self.assertIn("gitea_whoami", status["preflight_block_reasons"][0])
|
||||||
|
|
||||||
|
def test_declared_clean_task_worktree_ignores_control_checkout_violation(self):
|
||||||
|
"""#275: active branches/ worktree is the inspected mutation workspace."""
|
||||||
|
import mcp_server
|
||||||
|
mcp_server._preflight_whoami_called = True
|
||||||
|
mcp_server._preflight_capability_called = True
|
||||||
|
mcp_server._preflight_whoami_violation = True
|
||||||
|
mcp_server._preflight_whoami_violation_files = ["mcp_server.py"]
|
||||||
|
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||||
|
|
||||||
|
worktree = "/repo/branches/issue-275-clean"
|
||||||
|
status = mcp_server.assess_preflight_status(worktree_path=worktree)
|
||||||
|
self.assertTrue(status["preflight_ready"])
|
||||||
|
self.assertEqual(status["preflight_block_reasons"], [])
|
||||||
|
self.assertIn("preflight_workspace", status)
|
||||||
|
mcp_server.verify_preflight_purity(worktree_path=worktree)
|
||||||
|
|
||||||
|
def test_declared_dirty_task_worktree_reports_workspace_diagnostics(self):
|
||||||
|
"""#275: dirty failures name the inspected task workspace, not just files."""
|
||||||
|
import mcp_server
|
||||||
|
mcp_server._preflight_whoami_called = True
|
||||||
|
mcp_server._preflight_capability_called = True
|
||||||
|
os.environ["GITEA_TEST_PORCELAIN"] = " M task_file.py\n"
|
||||||
|
|
||||||
|
worktree = "/repo/branches/issue-275-dirty"
|
||||||
|
with self.assertRaises(RuntimeError) as ctx:
|
||||||
|
mcp_server.verify_preflight_purity(worktree_path=worktree)
|
||||||
|
msg = str(ctx.exception)
|
||||||
|
self.assertIn("active task workspace root", msg)
|
||||||
|
self.assertIn("inspected git root", msg)
|
||||||
|
self.assertIn("dirty files: task_file.py", msg)
|
||||||
|
self.assertIn("dirty scope:", msg)
|
||||||
|
|||||||
@@ -141,6 +141,17 @@ class TestResolveTaskCapability(unittest.TestCase):
|
|||||||
self.assertTrue(res["allowed_in_current_session"])
|
self.assertTrue(res["allowed_in_current_session"])
|
||||||
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
||||||
|
|
||||||
|
@patch("mcp_server.api_request", return_value={"login": "author-user"})
|
||||||
|
@patch("mcp_server.get_auth_header", return_value="token author-pass")
|
||||||
|
def test_resolve_lock_issue_author_profile_allowed(self, _auth, _api):
|
||||||
|
# #275: lock_issue must be an exact resolver task for claim/lock gates.
|
||||||
|
with patch.dict(os.environ, self._env("author-profile")):
|
||||||
|
res = mcp_server.gitea_resolve_task_capability(task="lock_issue", remote="prgs")
|
||||||
|
self.assertEqual(res["requested_task"], "lock_issue")
|
||||||
|
self.assertEqual(res["required_operation_permission"], "gitea.issue.comment")
|
||||||
|
self.assertTrue(res["allowed_in_current_session"])
|
||||||
|
self.assertIn("gitea.issue.comment", res["active_profile_allowed_operations"])
|
||||||
|
|
||||||
def test_resolve_unknown_task_fails_closed(self):
|
def test_resolve_unknown_task_fails_closed(self):
|
||||||
with patch.dict(os.environ, self._env("author-profile")):
|
with patch.dict(os.environ, self._env("author-profile")):
|
||||||
with self.assertRaises(ValueError):
|
with self.assertRaises(ValueError):
|
||||||
|
|||||||
Reference in New Issue
Block a user