fix: resolve conflicts for PR #516
Merge prgs/master into feat/issue-514-branch-delete-guard and preserve PR intent alongside compatible master guardrails.
This commit is contained in:
@@ -46,3 +46,12 @@ GITEA_TOKEN_SOURCE=GITEA_TOKEN
|
||||
# profile's values. Leave unset for pure env-based configuration.
|
||||
GITEA_MCP_CONFIG=/Users/jasonwalker/.config/gitea-tools/profiles.json
|
||||
GITEA_MCP_PROFILE=prgs
|
||||
|
||||
# Namespace-scoped active task workspaces (#510). Each MCP namespace uses only
|
||||
# its own role env var; foreign bindings (e.g. GITEA_AUTHOR_WORKTREE in a
|
||||
# merger process) are ignored.
|
||||
# GITEA_AUTHOR_WORKTREE=/path/to/repo/branches/issue-123-work
|
||||
# GITEA_REVIEWER_WORKTREE=/path/to/repo/branches/review-pr456
|
||||
# GITEA_MERGER_WORKTREE=/path/to/repo/branches/merge-pr456
|
||||
# GITEA_RECONCILER_WORKTREE=/path/to/repo/branches/reconcile-pr456
|
||||
# GITEA_ACTIVE_WORKTREE=/path/to/repo/branches/session-override
|
||||
|
||||
@@ -0,0 +1,344 @@
|
||||
"""Audit vs cleanup phase gates for reconciliation workflows (#419).
|
||||
|
||||
Audit/reconciliation tasks are read-only unless a separate cleanup phase is
|
||||
explicitly authorized with exact capability proof, safety proof, and
|
||||
before/after snapshots. Cleanup mutations must be classified in final reports;
|
||||
audit reports must not claim ``no mutations`` when cleanup occurred.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md"
|
||||
|
||||
PHASE_AUDIT = "audit"
|
||||
PHASE_CLEANUP = "cleanup"
|
||||
|
||||
# Tasks that enter audit phase on capability resolution (read-only default).
|
||||
AUDIT_PHASE_TASKS = frozenset({
|
||||
"reconcile-landed-pr",
|
||||
"reconcile_landed_pr",
|
||||
"reconcile_issue_claims",
|
||||
"reconcile_merged_cleanups",
|
||||
})
|
||||
|
||||
# Mutation tasks forbidden during audit phase (fail closed).
|
||||
AUDIT_FORBIDDEN_TASKS = frozenset({
|
||||
"delete_branch",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"commit_files",
|
||||
"gitea_commit_files",
|
||||
"mark_issue",
|
||||
"lock_issue",
|
||||
"claim_issue",
|
||||
"close_pr",
|
||||
"close_issue",
|
||||
"create_issue",
|
||||
"merge_pr",
|
||||
"review_pr",
|
||||
"submit_pr_review",
|
||||
"comment_pr",
|
||||
"comment_issue",
|
||||
"set_issue_labels",
|
||||
})
|
||||
|
||||
# Shell/git commands audit phase must not run.
|
||||
AUDIT_FORBIDDEN_COMMAND_RE = re.compile(
|
||||
r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|"
|
||||
r"gitea_delete_branch|delete_remote_branch)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
_NO_MUTATIONS_RE = re.compile(
|
||||
r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLEANUP_OCCURRED_RE = re.compile(
|
||||
r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|"
|
||||
r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|"
|
||||
r"cleanup\s+phase\s*:\s*(?!none\b)\S)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_EXTERNAL_STATE_RE = re.compile(
|
||||
r"^\s*[-*]?\s*external[- ]state mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_GIT_REF_RE = re.compile(
|
||||
r"^\s*[-*]?\s*git ref mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_CLEANUP_MUTATIONS_RE = re.compile(
|
||||
r"^\s*[-*]?\s*cleanup mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_CLEANUP_PHASE_AUTH_RE = re.compile(
|
||||
r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_DELETE_CAPABILITY_RE = re.compile(
|
||||
r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_BEFORE_AFTER_RE = re.compile(
|
||||
r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_SAFETY_PROOF_RE = re.compile(
|
||||
r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
|
||||
_session: dict[str, Any] | None = None
|
||||
|
||||
|
||||
def _blank_session() -> dict[str, Any]:
|
||||
return {
|
||||
"phase": PHASE_AUDIT,
|
||||
"entered_from_task": None,
|
||||
"cleanup_authorized": False,
|
||||
"cleanup_authorization": {},
|
||||
}
|
||||
|
||||
|
||||
def current_phase() -> str | None:
|
||||
"""Return active reconciliation phase or None when unset."""
|
||||
if not _session:
|
||||
return None
|
||||
return _session.get("phase")
|
||||
|
||||
|
||||
def active_record() -> dict[str, Any] | None:
|
||||
"""Return a copy of the session record, if any."""
|
||||
return dict(_session) if _session else None
|
||||
|
||||
|
||||
def clear_phase() -> None:
|
||||
"""Clear reconciliation phase state."""
|
||||
global _session
|
||||
_session = None
|
||||
|
||||
|
||||
def enter_audit_phase(task: str) -> dict[str, Any]:
|
||||
"""Enter read-only audit phase for a reconciliation task."""
|
||||
global _session
|
||||
normalized = (task or "").strip().lower()
|
||||
_session = _blank_session()
|
||||
_session["entered_from_task"] = normalized
|
||||
return dict(_session)
|
||||
|
||||
|
||||
def authorize_cleanup_phase(
|
||||
*,
|
||||
operator_approved: bool = False,
|
||||
workflow_authorized: bool = False,
|
||||
delete_capability_proven: bool = False,
|
||||
safety_proof: dict[str, Any] | None = None,
|
||||
before_after_snapshot: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Authorize cleanup phase after explicit approval and safety proofs."""
|
||||
reasons: list[str] = []
|
||||
if not (operator_approved or workflow_authorized):
|
||||
reasons.append(
|
||||
"cleanup phase requires operator approval or explicit workflow "
|
||||
"authorization"
|
||||
)
|
||||
if not delete_capability_proven:
|
||||
reasons.append(
|
||||
"cleanup phase requires exact delete_branch capability proof "
|
||||
"(gitea.branch.delete)"
|
||||
)
|
||||
safety = dict(safety_proof or {})
|
||||
if not safety.get("safe_to_delete_remote") and not safety.get(
|
||||
"safe_to_remove_worktree"
|
||||
):
|
||||
reasons.append(
|
||||
"cleanup phase requires proof that branch/worktree is safe to remove"
|
||||
)
|
||||
snapshot = dict(before_after_snapshot or {})
|
||||
if not snapshot.get("before") or not snapshot.get("after"):
|
||||
reasons.append(
|
||||
"cleanup phase requires before/after state snapshot"
|
||||
)
|
||||
|
||||
if reasons:
|
||||
return {
|
||||
"authorized": False,
|
||||
"phase": current_phase() or PHASE_AUDIT,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"remain in audit-only mode or supply operator approval, "
|
||||
"delete_branch capability proof, safety proof, and "
|
||||
"before/after snapshot before cleanup"
|
||||
),
|
||||
}
|
||||
|
||||
global _session
|
||||
if _session is None:
|
||||
_session = _blank_session()
|
||||
_session["phase"] = PHASE_CLEANUP
|
||||
_session["cleanup_authorized"] = True
|
||||
_session["cleanup_authorization"] = {
|
||||
"operator_approved": operator_approved,
|
||||
"workflow_authorized": workflow_authorized,
|
||||
"delete_capability_proven": delete_capability_proven,
|
||||
"safety_proof": safety,
|
||||
"before_after_snapshot": snapshot,
|
||||
}
|
||||
return {
|
||||
"authorized": True,
|
||||
"phase": PHASE_CLEANUP,
|
||||
"reasons": [],
|
||||
"cleanup_authorization": dict(_session["cleanup_authorization"]),
|
||||
"safe_next_action": "proceed with authorized cleanup mutations only",
|
||||
}
|
||||
|
||||
|
||||
def check_audit_task_enters_phase(task: str) -> bool:
|
||||
"""Return whether resolving *task* should enter audit phase."""
|
||||
return (task or "").strip().lower() in AUDIT_PHASE_TASKS
|
||||
|
||||
|
||||
def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]:
|
||||
"""Fail closed when a mutation task runs during audit phase."""
|
||||
normalized = (task or "").strip().lower()
|
||||
phase = current_phase()
|
||||
if phase != PHASE_AUDIT:
|
||||
return True, []
|
||||
if normalized in AUDIT_FORBIDDEN_TASKS:
|
||||
return False, [
|
||||
f"task '{normalized}' is forbidden in audit-only reconciliation "
|
||||
"mode: switch to an explicit cleanup phase with operator approval "
|
||||
"and exact delete_branch capability proof before cleanup mutations"
|
||||
]
|
||||
return True, []
|
||||
|
||||
|
||||
def check_cleanup_execution_allowed() -> tuple[bool, list[str]]:
|
||||
"""Fail closed when cleanup execution is attempted without authorization."""
|
||||
phase = current_phase()
|
||||
if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"):
|
||||
return True, []
|
||||
if phase is None:
|
||||
return False, [
|
||||
"cleanup execution requires an active reconciliation session; "
|
||||
"resolve a reconciliation audit task first"
|
||||
]
|
||||
return False, [
|
||||
"cleanup execution forbidden in audit-only reconciliation mode; "
|
||||
"call gitea_authorize_reconciliation_cleanup_phase with operator "
|
||||
"approval, delete_branch capability proof, safety proof, and "
|
||||
"before/after snapshot"
|
||||
]
|
||||
|
||||
|
||||
def classify_cleanup_mutation(action: str) -> str:
|
||||
"""Map a cleanup action to the required mutation ledger category (#419)."""
|
||||
normalized = (action or "").strip().lower()
|
||||
if "delete_remote" in normalized or normalized in {
|
||||
"delete_branch",
|
||||
"gitea_delete_branch",
|
||||
}:
|
||||
return "external-state"
|
||||
if "branch" in normalized and "delete" in normalized:
|
||||
return "git-ref"
|
||||
if "worktree" in normalized or "remove_local" in normalized:
|
||||
return "cleanup"
|
||||
return "cleanup"
|
||||
|
||||
|
||||
def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]:
|
||||
"""Validate audit/cleanup reconciliation reports (fail closed)."""
|
||||
text = report_text or ""
|
||||
reasons: list[str] = []
|
||||
|
||||
cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text))
|
||||
claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text))
|
||||
|
||||
if cleanup_occurred and claims_no_mutations:
|
||||
reasons.append(
|
||||
"report claims no mutations but documents cleanup mutations; "
|
||||
"audit-only reports must not perform cleanup and cleanup reports "
|
||||
"must not claim no mutations"
|
||||
)
|
||||
|
||||
if cleanup_occurred:
|
||||
if not _CLEANUP_PHASE_AUTH_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without "
|
||||
"'Cleanup phase authorized: true'"
|
||||
)
|
||||
if not _DELETE_CAPABILITY_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without delete_branch capability "
|
||||
"proof"
|
||||
)
|
||||
if not _BEFORE_AFTER_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without before/after state snapshot"
|
||||
)
|
||||
if not _SAFETY_PROOF_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without branch/worktree safety proof"
|
||||
)
|
||||
|
||||
if re.search(r"delete_remote|remote branch.*delet", text, re.I):
|
||||
if not _EXTERNAL_STATE_RE.search(text):
|
||||
reasons.append(
|
||||
"remote branch deletion must be classified under "
|
||||
"External-state mutations"
|
||||
)
|
||||
if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I):
|
||||
if not _GIT_REF_RE.search(text):
|
||||
reasons.append(
|
||||
"local branch deletion must be classified under "
|
||||
"Git ref mutations"
|
||||
)
|
||||
if re.search(r"worktree.*remov|remove_local_worktree", text, re.I):
|
||||
if not _CLEANUP_MUTATIONS_RE.search(text):
|
||||
reasons.append(
|
||||
"worktree removal must be classified under Cleanup mutations"
|
||||
)
|
||||
|
||||
if (
|
||||
RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text
|
||||
or "reconcile-landed-pr" in text.lower()
|
||||
):
|
||||
if cleanup_occurred and "audit phase" in text.lower():
|
||||
if "cleanup phase" not in text.lower():
|
||||
reasons.append(
|
||||
"report mixes audit phase with cleanup mutations without "
|
||||
"documenting cleanup phase transition"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": reasons,
|
||||
"cleanup_occurred": cleanup_occurred,
|
||||
"claims_no_mutations": claims_no_mutations,
|
||||
"safe_next_action": (
|
||||
"proceed"
|
||||
if proven
|
||||
else "fix audit/cleanup report: separate audit from cleanup phase, "
|
||||
"classify mutations, and do not claim no mutations after cleanup"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]:
|
||||
"""Block shell commands that perform cleanup during audit phase."""
|
||||
phase = current_phase()
|
||||
if phase != PHASE_AUDIT:
|
||||
return True, []
|
||||
cmd = (command or "").strip()
|
||||
if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd):
|
||||
return False, [
|
||||
f"command forbidden in audit-only reconciliation mode: {cmd!r}; "
|
||||
"authorize cleanup phase before branch/worktree deletion or push"
|
||||
]
|
||||
return True, []
|
||||
@@ -12,6 +12,8 @@ import subprocess
|
||||
BASE_BRANCHES = frozenset({"master", "main", "dev"})
|
||||
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
|
||||
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
||||
# Author-only: reviewer/merger/reconciler namespaces use role-specific env vars
|
||||
# via namespace_workspace_binding (#510).
|
||||
|
||||
|
||||
def _normalize_path(path: str) -> str:
|
||||
|
||||
@@ -823,6 +823,45 @@ scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md
|
||||
scripts/release-tag v0.4.0 --notes-file /tmp/release-notes.md --push
|
||||
```
|
||||
|
||||
## Namespace workspace binding (#510)
|
||||
|
||||
Each MCP namespace resolves its **own** active task workspace. Foreign role
|
||||
worktree environment variables must not poison another namespace's purity
|
||||
checks.
|
||||
|
||||
| Namespace | Workspace env vars (in priority under `GITEA_ACTIVE_WORKTREE`) | Allowed roots |
|
||||
|-----------|------------------------------------------------------------------|---------------|
|
||||
| author | `GITEA_AUTHOR_WORKTREE` | `branches/<task>` worktree only (#274) |
|
||||
| reviewer | `GITEA_REVIEWER_WORKTREE` | clean `branches/<review>` worktree |
|
||||
| merger | `GITEA_MERGER_WORKTREE` | clean `branches/<merge>` worktree **or** clean control checkout |
|
||||
| reconciler | `GITEA_RECONCILER_WORKTREE` | clean `branches/<reconcile>` worktree **or** clean control checkout |
|
||||
|
||||
`GITEA_AUTHOR_WORKTREE` is **author-only**. Reviewer, merger, and reconciler
|
||||
MCP processes ignore it even when it points at a dirty author WIP tree.
|
||||
|
||||
### Safe reconnect / rebind procedure
|
||||
|
||||
When a mutation blocks on workspace binding:
|
||||
|
||||
1. Read the error — it names the **resolved workspace path**, **role
|
||||
namespace**, and **binding source** (tool arg, env var, or process root).
|
||||
2. Reconnect or relaunch the correct namespace MCP server from the intended
|
||||
workspace (or set the role-specific env var before launch).
|
||||
3. Pass `worktree_path` on reviewer/merger mutation tools when the active
|
||||
branches/ worktree differs from the MCP process root.
|
||||
4. **Do not** clean, reset, or discard foreign role worktrees to unblock your
|
||||
own namespace — that destroys another agent's WIP.
|
||||
|
||||
### CTH guidance for workspace binding blockers
|
||||
|
||||
When posting a Canonical Thread Handoff after a binding blocker:
|
||||
|
||||
- State which namespace was active (author / reviewer / merger / reconciler).
|
||||
- Quote the resolved workspace path and binding source from the error.
|
||||
- Name the safe reconnect action (relaunch MCP from `branches/...`, set
|
||||
`GITEA_*_WORKTREE`, or pass `worktree_path`).
|
||||
- Explicitly note that foreign worktrees must not be cleaned to unblock.
|
||||
|
||||
## Safety notes
|
||||
|
||||
- Never place raw tokens or passwords in any LLM MCP config; reference secrets
|
||||
|
||||
@@ -1038,6 +1038,22 @@ def _rule_shared_raw_branch_delete_bypass(report_text: str) -> list[dict[str, st
|
||||
)
|
||||
|
||||
|
||||
def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
|
||||
from audit_reconciliation_mode import assess_audit_reconciliation_report
|
||||
|
||||
result = assess_audit_reconciliation_report(report_text)
|
||||
if result.get("proven"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"reconcile.audit_cleanup_boundary",
|
||||
result.get("reasons") or [],
|
||||
field="Audit/cleanup phase",
|
||||
severity="block",
|
||||
safe_next_action=result.get("safe_next_action")
|
||||
or "separate audit from authorized cleanup and classify mutations",
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_review_mutation(
|
||||
report_text: str,
|
||||
*,
|
||||
@@ -1115,6 +1131,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_git_fetch_readonly,
|
||||
_rule_reviewer_legacy_workspace_mutations,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
_rule_audit_reconciliation_boundary,
|
||||
],
|
||||
"author_issue": [
|
||||
_rule_shared_controller_handoff,
|
||||
|
||||
+457
-45
@@ -175,12 +175,76 @@ _preflight_reviewer_violation_files: list[str] = []
|
||||
|
||||
ACTIVE_WORKTREE_ENV = "GITEA_ACTIVE_WORKTREE"
|
||||
AUTHOR_WORKTREE_ENV = "GITEA_AUTHOR_WORKTREE"
|
||||
REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
|
||||
MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
|
||||
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
|
||||
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
|
||||
|
||||
def _preflight_in_test_mode() -> bool:
|
||||
return "pytest" in sys.modules or "unittest" in sys.modules
|
||||
|
||||
|
||||
def _reviewer_session_worktree() -> str | None:
|
||||
import reviewer_pr_lease as _reviewer_pr_lease
|
||||
|
||||
session = _reviewer_pr_lease.get_session_lease()
|
||||
if not session:
|
||||
return None
|
||||
worktree = (session.get("worktree") or "").strip()
|
||||
return worktree or None
|
||||
|
||||
|
||||
def _effective_workspace_role() -> str:
|
||||
"""Resolve the namespace key used for workspace binding (#510)."""
|
||||
profile = get_profile()
|
||||
role = _preflight_resolved_role
|
||||
if not role:
|
||||
role = _role_kind(
|
||||
profile.get("allowed_operations") or [],
|
||||
profile.get("forbidden_operations") or [],
|
||||
)
|
||||
return nwb.normalize_role_kind(
|
||||
role,
|
||||
profile_name=profile.get("profile_name"),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
||||
"""Resolve the namespace-scoped workspace root inspected by pre-flight guards."""
|
||||
role = _effective_workspace_role()
|
||||
workspace, _source = nwb.resolve_namespace_workspace(
|
||||
role_kind=role,
|
||||
worktree_path=worktree_path,
|
||||
process_project_root=PROJECT_ROOT,
|
||||
session_lease_worktree=(
|
||||
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
|
||||
),
|
||||
profile_name=get_profile().get("profile_name"),
|
||||
)
|
||||
return workspace
|
||||
|
||||
|
||||
def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
|
||||
"""Canonical namespace workspace + repository root for guards (#460/#510)."""
|
||||
role = _effective_workspace_role()
|
||||
return nwb.resolve_namespace_mutation_context(
|
||||
role_kind=role,
|
||||
worktree_path=worktree_path,
|
||||
process_project_root=PROJECT_ROOT,
|
||||
session_lease_worktree=(
|
||||
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
|
||||
),
|
||||
profile_name=get_profile().get("profile_name"),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
|
||||
"""Backward-compatible alias for namespace workspace context."""
|
||||
return _resolve_namespace_mutation_context(worktree_path)
|
||||
|
||||
|
||||
def _ensure_process_start_porcelain() -> str:
|
||||
"""Capture the shared-worktree baseline once per MCP process (#252)."""
|
||||
global _process_start_porcelain
|
||||
@@ -189,26 +253,6 @@ def _ensure_process_start_porcelain() -> str:
|
||||
return _process_start_porcelain
|
||||
|
||||
|
||||
def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
|
||||
"""Resolve the workspace root inspected by pre-flight guards."""
|
||||
return author_mutation_worktree.resolve_mutation_workspace(
|
||||
worktree_path,
|
||||
PROJECT_ROOT,
|
||||
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
|
||||
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
|
||||
)
|
||||
|
||||
|
||||
def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
|
||||
"""Canonical workspace + repository root for runtime_context and guards (#460)."""
|
||||
return author_mutation_worktree.resolve_author_mutation_context(
|
||||
worktree_path,
|
||||
PROJECT_ROOT,
|
||||
active_worktree_env=os.environ.get(ACTIVE_WORKTREE_ENV),
|
||||
author_worktree_env=os.environ.get(AUTHOR_WORKTREE_ENV),
|
||||
)
|
||||
|
||||
|
||||
def _get_git_root(path: str) -> str | None:
|
||||
try:
|
||||
res = subprocess.run(
|
||||
@@ -276,7 +320,7 @@ def _format_preflight_files(files: list[str]) -> str:
|
||||
|
||||
|
||||
def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[str]) -> dict:
|
||||
ctx = _resolve_author_mutation_context(worktree_path)
|
||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
inspected_root = _get_git_root(workspace)
|
||||
process_root = ctx["process_project_root"]
|
||||
@@ -294,6 +338,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
|
||||
"dirty_files": list(dirty_files),
|
||||
"dirty_scope": dirty_scope,
|
||||
"workspace_roots_aligned": ctx["roots_aligned"],
|
||||
"workspace_role_kind": ctx.get("workspace_role_kind"),
|
||||
"workspace_binding_source": ctx.get("workspace_binding_source"),
|
||||
"ignored_bindings": list(ctx.get("ignored_bindings") or []),
|
||||
}
|
||||
if not ctx["roots_aligned"]:
|
||||
details["workspace_root_mismatch"] = (
|
||||
@@ -304,13 +351,19 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
|
||||
|
||||
|
||||
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')}"
|
||||
)
|
||||
parts = [
|
||||
f"MCP server process root: {details.get('mcp_server_process_root')}",
|
||||
f"active task workspace root: {details.get('active_task_workspace_root')}",
|
||||
f"workspace role: {details.get('workspace_role_kind')}",
|
||||
f"binding source: {details.get('workspace_binding_source')}",
|
||||
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')}",
|
||||
]
|
||||
ignored = details.get("ignored_bindings") or []
|
||||
if ignored:
|
||||
parts.append(f"ignored foreign bindings: {'; '.join(ignored)}")
|
||||
return "; ".join(parts)
|
||||
|
||||
|
||||
def assess_preflight_status(worktree_path: str | None = None) -> dict:
|
||||
@@ -333,6 +386,25 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
|
||||
"Active task workspace has tracked file edits before mutation "
|
||||
f"({_format_preflight_workspace_details(workspace_details)})"
|
||||
)
|
||||
role = _effective_workspace_role()
|
||||
if role in nwb.NON_AUTHOR_ROLES:
|
||||
binding = nwb.assess_metadata_only_worktree_binding(
|
||||
role_kind=role,
|
||||
declared_worktree_path=worktree_path,
|
||||
mutation_workspace=_resolve_preflight_workspace_path(worktree_path),
|
||||
process_project_root=PROJECT_ROOT,
|
||||
profile_name=get_profile().get("profile_name"),
|
||||
)
|
||||
if binding.get("block"):
|
||||
reasons.append(binding["reasons"][0])
|
||||
reasons.append(
|
||||
nwb.format_namespace_workspace_binding_error(
|
||||
role_kind=role,
|
||||
workspace_path=binding["mutation_workspace"],
|
||||
binding_source="MCP server process root (default)",
|
||||
reasons=binding.get("reasons"),
|
||||
)
|
||||
)
|
||||
return {
|
||||
"preflight_ready": not reasons,
|
||||
"preflight_block_reasons": reasons,
|
||||
@@ -466,13 +538,14 @@ def record_preflight_check(
|
||||
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
|
||||
"""#274: author file/branch mutations must run from a branches/ worktree.
|
||||
|
||||
Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
|
||||
Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
|
||||
(#468).
|
||||
Reviewer, merger, and reconciler roles are exempt: reconciler ``close_pr``
|
||||
is a Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
|
||||
(#468). Non-author namespaces use dedicated workspace env vars (#510).
|
||||
"""
|
||||
if _preflight_resolved_role in ("reviewer", "reconciler"):
|
||||
role = _effective_workspace_role()
|
||||
if role in nwb.NON_AUTHOR_ROLES:
|
||||
return
|
||||
ctx = _resolve_author_mutation_context(worktree_path)
|
||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(workspace)
|
||||
assessment = author_mutation_worktree.assess_author_mutation_worktree(
|
||||
@@ -520,11 +593,12 @@ def verify_preflight_purity(
|
||||
f"'{task}' (fail closed)"
|
||||
)
|
||||
|
||||
ctx = _resolve_author_mutation_context(worktree_path)
|
||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
canonical_root = ctx["canonical_repo_root"]
|
||||
process_root = ctx["process_project_root"]
|
||||
real_workspace = os.path.realpath(workspace)
|
||||
role = ctx.get("workspace_role_kind") or _effective_workspace_role()
|
||||
|
||||
if real_workspace != process_root:
|
||||
if not _preflight_in_test_mode():
|
||||
@@ -541,11 +615,15 @@ def verify_preflight_purity(
|
||||
|
||||
dirty_files = sorted(_parse_porcelain_entries(_get_workspace_porcelain(workspace)))
|
||||
if dirty_files:
|
||||
details = _preflight_workspace_details(workspace, 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)}"
|
||||
nwb.format_namespace_workspace_binding_error(
|
||||
role_kind=role,
|
||||
workspace_path=workspace,
|
||||
binding_source=ctx.get("workspace_binding_source")
|
||||
or "unknown binding source",
|
||||
dirty_files=dirty_files,
|
||||
ignored_bindings=ctx.get("ignored_bindings"),
|
||||
)
|
||||
)
|
||||
else:
|
||||
if _preflight_whoami_violation:
|
||||
@@ -561,14 +639,14 @@ def verify_preflight_purity(
|
||||
f"{_format_preflight_files(_preflight_capability_violation_files)}"
|
||||
)
|
||||
|
||||
if _preflight_resolved_role == "reviewer":
|
||||
if role in {"reviewer", "merger"}:
|
||||
current = _get_workspace_porcelain()
|
||||
baseline = _preflight_capability_baseline_porcelain or ""
|
||||
reviewer_delta = _new_tracked_changes_since(baseline, current)
|
||||
_preflight_reviewer_violation_files = reviewer_delta
|
||||
if reviewer_delta:
|
||||
raise RuntimeError(
|
||||
"Reviewer role violation: Reviewer profile is forbidden from modifying "
|
||||
f"{role.title()} role violation: profile is forbidden from modifying "
|
||||
"tracked workspace files (fail closed). Offending files: "
|
||||
f"{_format_preflight_files(reviewer_delta)}"
|
||||
)
|
||||
@@ -576,6 +654,45 @@ def verify_preflight_purity(
|
||||
_enforce_branches_only_author_mutation(worktree_path)
|
||||
_clear_preflight_capability_state()
|
||||
|
||||
|
||||
def _verify_role_mutation_workspace(
|
||||
remote: str | None = None,
|
||||
*,
|
||||
worktree_path: str | None = None,
|
||||
worktree: str | None = None,
|
||||
task: str | None = None,
|
||||
) -> str:
|
||||
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
|
||||
role = _effective_workspace_role()
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(
|
||||
_resolve_preflight_workspace_path(worktree_path)
|
||||
)
|
||||
assessment = nwb.assess_namespace_mutation_workspace(
|
||||
role_kind=role,
|
||||
worktree_path=worktree_path,
|
||||
worktree=worktree,
|
||||
process_project_root=PROJECT_ROOT,
|
||||
session_lease_worktree=(
|
||||
_reviewer_session_worktree() if role in {"reviewer", "merger"} else None
|
||||
),
|
||||
profile_name=get_profile().get("profile_name"),
|
||||
current_branch=git_state.get("current_branch"),
|
||||
)
|
||||
if assessment["block"]:
|
||||
raise RuntimeError(
|
||||
nwb.format_namespace_workspace_binding_error(
|
||||
role_kind=role,
|
||||
workspace_path=assessment["mutation_workspace"],
|
||||
binding_source=assessment.get("workspace_binding_source")
|
||||
or "unknown binding source",
|
||||
reasons=assessment.get("reasons"),
|
||||
ignored_bindings=assessment.get("ignored_bindings"),
|
||||
)
|
||||
)
|
||||
resolved = assessment["mutation_workspace"]
|
||||
verify_preflight_purity(remote, worktree_path=resolved, task=task)
|
||||
return resolved
|
||||
|
||||
from mcp.server.fastmcp import FastMCP # noqa: E402
|
||||
|
||||
from gitea_auth import ( # noqa: E402
|
||||
@@ -597,6 +714,7 @@ import role_session_router # noqa: E402
|
||||
import role_namespace_gate # noqa: E402
|
||||
import task_capability_map # noqa: E402
|
||||
import review_proofs # noqa: E402
|
||||
import review_workflow_load # noqa: E402
|
||||
import agent_temp_artifacts
|
||||
import issue_lock_worktree # noqa: E402
|
||||
import issue_lock_provenance # noqa: E402
|
||||
@@ -612,6 +730,7 @@ import reviewer_pr_lease # noqa: E402
|
||||
import merged_cleanup_reconcile # noqa: E402
|
||||
import reconciler_profile # noqa: E402
|
||||
import reconciliation_workflow # noqa: E402
|
||||
import audit_reconciliation_mode # noqa: E402
|
||||
import review_merge_state_machine # noqa: E402
|
||||
import pr_work_lease # noqa: E402
|
||||
import native_mcp_preference # noqa: E402
|
||||
@@ -1330,7 +1449,6 @@ def gitea_create_issue(
|
||||
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
|
||||
"""Fetch all OPEN pull requests for a repo (used for stacked-base proof, #484)."""
|
||||
try:
|
||||
@@ -1341,6 +1459,7 @@ def _list_open_pulls(h: str, o: str, r: str, auth: str) -> list[dict]:
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_lock_issue(
|
||||
issue_number: int,
|
||||
branch_name: str,
|
||||
@@ -2268,6 +2387,7 @@ def init_review_decision_lock(remote: str | None, task: str | None):
|
||||
"""Seed read-only-until-ready state for reviewer PR review tasks."""
|
||||
if task != "review_pr":
|
||||
return
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
profile = get_profile()
|
||||
profile_name = (profile.get("profile_name") or "").strip()
|
||||
session_lock = (
|
||||
@@ -2294,6 +2414,11 @@ def init_review_decision_lock(remote: str | None, task: str | None):
|
||||
})
|
||||
|
||||
|
||||
def _review_workflow_load_gate_reasons() -> list[str]:
|
||||
"""Fail closed when canonical review workflow was not loaded (#389)."""
|
||||
return review_workflow_load.review_workflow_load_blockers(PROJECT_ROOT)
|
||||
|
||||
|
||||
def check_review_decision_gate(
|
||||
pr_number: int,
|
||||
action: str,
|
||||
@@ -2304,7 +2429,10 @@ def check_review_decision_gate(
|
||||
repo: str | None = None,
|
||||
) -> list[str]:
|
||||
"""Fail closed unless validation completed and the final decision is ready."""
|
||||
reasons = []
|
||||
reasons = list(_review_workflow_load_gate_reasons())
|
||||
if reasons:
|
||||
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
|
||||
return reasons
|
||||
lock = _load_review_decision_lock()
|
||||
if lock is None:
|
||||
reasons.append(
|
||||
@@ -2712,10 +2840,14 @@ def _evaluate_pr_review_submission(
|
||||
*,
|
||||
live: bool,
|
||||
final_review_decision_ready: bool = False,
|
||||
worktree_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Shared gate chain for live submit and dry-run review tools."""
|
||||
verify_preflight_purity(remote, task="review_pr")
|
||||
_verify_role_mutation_workspace(
|
||||
remote, worktree_path=worktree_path, task="review_pr"
|
||||
)
|
||||
action = (action or "").strip().lower()
|
||||
workflow_blockers = _review_workflow_load_gate_reasons() if live else []
|
||||
result = {
|
||||
"requested_action": action,
|
||||
"performed": False,
|
||||
@@ -2731,6 +2863,10 @@ def _evaluate_pr_review_submission(
|
||||
"reasons": [],
|
||||
}
|
||||
reasons = result["reasons"]
|
||||
if workflow_blockers:
|
||||
reasons.extend(workflow_blockers)
|
||||
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
|
||||
return result
|
||||
|
||||
if action not in _REVIEW_ACTIONS:
|
||||
reasons.append(
|
||||
@@ -2949,6 +3085,13 @@ def gitea_mark_final_review_decision(
|
||||
}
|
||||
org = resolved_org
|
||||
repo = resolved_repo
|
||||
workflow_blockers = _review_workflow_load_gate_reasons()
|
||||
if workflow_blockers:
|
||||
return {
|
||||
"marked_ready": False,
|
||||
"reasons": workflow_blockers + (
|
||||
review_workflow_load.recovery_handoff_without_replay()),
|
||||
}
|
||||
hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
|
||||
if hard_stop:
|
||||
return {"marked_ready": False, "reasons": hard_stop}
|
||||
@@ -3114,6 +3257,7 @@ def gitea_dry_run_pr_review(
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Validate review submission mechanics without a live PR mutation."""
|
||||
return _evaluate_pr_review_submission(
|
||||
@@ -3126,6 +3270,7 @@ def gitea_dry_run_pr_review(
|
||||
org=org,
|
||||
repo=repo,
|
||||
live=False,
|
||||
worktree_path=worktree_path,
|
||||
)
|
||||
|
||||
|
||||
@@ -3141,6 +3286,7 @@ def gitea_submit_pr_review(
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
final_review_decision_ready: bool = False,
|
||||
worktree_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Gated PR review mutation: comment findings, request changes, or approve.
|
||||
|
||||
@@ -3159,6 +3305,7 @@ def gitea_submit_pr_review(
|
||||
repo=repo,
|
||||
live=True,
|
||||
final_review_decision_ready=final_review_decision_ready,
|
||||
worktree_path=worktree_path,
|
||||
)
|
||||
|
||||
|
||||
@@ -3519,6 +3666,7 @@ def gitea_merge_pr(
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Gated merge of a Gitea pull request (#16).
|
||||
|
||||
@@ -3565,6 +3713,10 @@ def gitea_merge_pr(
|
||||
host: Override the Gitea host.
|
||||
org: Override the owner/organization.
|
||||
repo: Override the repository name.
|
||||
worktree_path: Merger workspace path under ``branches/`` or clean
|
||||
control checkout; defaults to ``GITEA_MERGER_WORKTREE``,
|
||||
``GITEA_ACTIVE_WORKTREE``, or the MCP server process root. Ignores
|
||||
foreign ``GITEA_AUTHOR_WORKTREE`` bindings (#510).
|
||||
|
||||
Returns:
|
||||
dict describing the attempt: performed, authenticated user, profile
|
||||
@@ -3572,7 +3724,10 @@ def gitea_merge_pr(
|
||||
reasons/gates passed or blocked, and merge result / merge commit if
|
||||
available. Never secrets.
|
||||
"""
|
||||
verify_preflight_purity(remote, task="merge_pr")
|
||||
_verify_role_mutation_workspace(
|
||||
remote, worktree_path=worktree_path, task="merge_pr"
|
||||
)
|
||||
workflow_blockers = _review_workflow_load_gate_reasons()
|
||||
do = (do or "").strip().lower()
|
||||
result = {
|
||||
"performed": False,
|
||||
@@ -3590,6 +3745,10 @@ def gitea_merge_pr(
|
||||
"reasons": [],
|
||||
}
|
||||
reasons = result["reasons"]
|
||||
if workflow_blockers:
|
||||
reasons.extend(workflow_blockers)
|
||||
reasons.extend(review_workflow_load.recovery_handoff_without_replay())
|
||||
return result
|
||||
|
||||
# Gate 1 — valid merge method (no API call on a bad method).
|
||||
if do not in _MERGE_METHODS:
|
||||
@@ -4101,6 +4260,18 @@ def gitea_delete_branch(
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
audit_allowed, audit_reasons = (
|
||||
audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
|
||||
)
|
||||
if not audit_allowed:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": audit_reasons,
|
||||
"audit_phase": audit_reconciliation_mode.current_phase(),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote, task="delete_branch")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
@@ -4313,6 +4484,18 @@ def gitea_reconcile_merged_cleanups(
|
||||
"execute_confirmed must be True when dry_run=False (fail closed)"
|
||||
)
|
||||
|
||||
if not dry_run:
|
||||
exec_allowed, exec_reasons = (
|
||||
audit_reconciliation_mode.check_cleanup_execution_allowed()
|
||||
)
|
||||
if not exec_allowed:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"reasons": exec_reasons,
|
||||
"audit_phase": audit_reconciliation_mode.current_phase(),
|
||||
}
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -4395,6 +4578,53 @@ def gitea_reconcile_merged_cleanups(
|
||||
return {"success": True, "performed": True, **report}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved: bool = False,
|
||||
workflow_authorized: bool = False,
|
||||
delete_capability_proven: bool = False,
|
||||
safe_to_delete_remote: bool = False,
|
||||
safe_to_remove_worktree: bool = False,
|
||||
before_state: str = "",
|
||||
after_state: str = "",
|
||||
) -> dict:
|
||||
"""Authorize cleanup phase after audit-only reconciliation (#419).
|
||||
|
||||
Requires operator or workflow approval, exact delete_branch capability proof,
|
||||
branch/worktree safety proof, and before/after state snapshots. Audit phase
|
||||
forbids branch deletion, worktree removal, pushes, and issue/PR mutations.
|
||||
"""
|
||||
delete_gate = _profile_operation_gate("gitea.branch.delete")
|
||||
capability_ok = not bool(delete_gate)
|
||||
if delete_capability_proven and delete_gate:
|
||||
return {
|
||||
"authorized": False,
|
||||
"performed": False,
|
||||
"delete_capability_verified": False,
|
||||
"reasons": [
|
||||
"delete_capability_proven=true but active profile lacks "
|
||||
"gitea.branch.delete",
|
||||
] + delete_gate,
|
||||
"audit_phase": audit_reconciliation_mode.current_phase(),
|
||||
}
|
||||
result = audit_reconciliation_mode.authorize_cleanup_phase(
|
||||
operator_approved=operator_approved,
|
||||
workflow_authorized=workflow_authorized,
|
||||
delete_capability_proven=delete_capability_proven and capability_ok,
|
||||
safety_proof={
|
||||
"safe_to_delete_remote": safe_to_delete_remote,
|
||||
"safe_to_remove_worktree": safe_to_remove_worktree,
|
||||
},
|
||||
before_after_snapshot={
|
||||
"before": (before_state or "").strip(),
|
||||
"after": (after_state or "").strip(),
|
||||
},
|
||||
)
|
||||
result["performed"] = bool(result.get("authorized"))
|
||||
result["delete_capability_verified"] = capability_ok
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_assess_already_landed_reconciliation(
|
||||
pr_number: int,
|
||||
@@ -5198,7 +5428,7 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
"permission_report": _permission_block_report("gitea.pr.comment"),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote, task="review_pr")
|
||||
_verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
profile = get_profile()
|
||||
@@ -5208,6 +5438,13 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
|
||||
comments = _fetch_pr_comments(
|
||||
pr_number, remote=remote, host=host, org=org, repo=repo)
|
||||
# Refuse merge-oriented lease acquisition/adoption on an already-merged or
|
||||
# closed PR: the lease is moot and adopting it for merge work is unsafe (#515).
|
||||
pr_live = api_request(
|
||||
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
|
||||
pr_merged_or_closed = bool(
|
||||
pr_live.get("merged") or pr_live.get("merged_at")
|
||||
) or (str(pr_live.get("state") or "").strip().lower() == "closed")
|
||||
assessment = reviewer_pr_lease.assess_acquire_lease(
|
||||
comments,
|
||||
pr_number=pr_number,
|
||||
@@ -5220,6 +5457,7 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
candidate_head=candidate_head,
|
||||
target_branch=target_branch,
|
||||
target_branch_sha=target_branch_sha,
|
||||
pr_merged_or_closed=pr_merged_or_closed,
|
||||
)
|
||||
if not assessment.get("acquire_allowed"):
|
||||
return {
|
||||
@@ -5227,6 +5465,7 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
"acquired": False,
|
||||
"reasons": assessment.get("reasons") or [],
|
||||
"existing_lease": assessment.get("existing_lease"),
|
||||
"post_merge_moot": assessment.get("post_merge_moot", False),
|
||||
}
|
||||
|
||||
body = assessment["lease_body"]
|
||||
@@ -5375,6 +5614,126 @@ def gitea_assess_reviewer_pr_lease(
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_cleanup_post_merge_moot_lease(
|
||||
pr_number: int,
|
||||
apply: bool = False,
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
) -> dict:
|
||||
"""Safely resolve a moot reviewer lease left on an ALREADY-MERGED/closed PR (#515).
|
||||
|
||||
Read-first and fail-safe. This tool never merges and never adopts a lease.
|
||||
It only acts when the live PR state is merged/closed, and it never steals or
|
||||
force-cleans an active *foreign* lease while the PR is still open. When
|
||||
``apply`` is true and a lease is still active on a merged/closed PR, it posts
|
||||
a terminal ``phase: released`` lease marker (``blocker: post-merge-moot``) —
|
||||
an append-only comment that neutralises the moot lease without deleting any
|
||||
other session's comment.
|
||||
|
||||
Args:
|
||||
pr_number: The PR whose lingering lease to assess/clean.
|
||||
apply: When false (default) report only (read-only). When true, post the
|
||||
terminal released marker if — and only if — cleanup is allowed.
|
||||
remote: Known instance — 'dadeschools' or 'prgs'.
|
||||
host: Override the Gitea host.
|
||||
org: Override the owner/organization.
|
||||
repo: Override the repository name.
|
||||
|
||||
Returns:
|
||||
dict reporting PR merged/closed state, merge_commit_sha, linked-issue
|
||||
closure state, whether the lease is moot, whether cleanup was performed
|
||||
or skipped (and why), and ``no_merge_or_adoption`` True — this path never
|
||||
merges or adopts.
|
||||
"""
|
||||
read_block = _profile_operation_gate("gitea.read")
|
||||
if read_block:
|
||||
return {
|
||||
"success": False,
|
||||
"cleanup_performed": False,
|
||||
"no_merge_or_adoption": True,
|
||||
"reasons": read_block,
|
||||
"permission_report": _permission_block_report("gitea.read"),
|
||||
}
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
pr_live = api_request(
|
||||
"GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
|
||||
comments = _fetch_pr_comments(
|
||||
pr_number, remote=remote, host=host, org=org, repo=repo)
|
||||
assessment = reviewer_pr_lease.assess_post_merge_moot_lease(
|
||||
comments,
|
||||
pr_number=pr_number,
|
||||
pr_merged=bool(pr_live.get("merged") or pr_live.get("merged_at")),
|
||||
pr_state=pr_live.get("state"),
|
||||
merge_commit_sha=pr_live.get("merge_commit_sha"),
|
||||
)
|
||||
|
||||
# Best-effort linked-issue closure state for the report.
|
||||
active = assessment.get("active_lease") or {}
|
||||
issue_no = active.get("issue_number")
|
||||
linked_issue_state = None
|
||||
if issue_no:
|
||||
try:
|
||||
issue = api_request(
|
||||
"GET", f"{repo_api_url(h, o, r)}/issues/{issue_no}", auth) or {}
|
||||
linked_issue_state = issue.get("state")
|
||||
except Exception:
|
||||
linked_issue_state = None
|
||||
|
||||
report = {
|
||||
"success": True,
|
||||
"pr_number": pr_number,
|
||||
"pr_state": assessment.get("pr_state"),
|
||||
"pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
|
||||
"merge_commit_sha": assessment.get("merge_commit_sha"),
|
||||
"linked_issue_number": issue_no,
|
||||
"linked_issue_state": linked_issue_state,
|
||||
"lease_moot": assessment.get("is_moot"),
|
||||
"active_lease": assessment.get("active_lease"),
|
||||
"cleanup_allowed": assessment.get("cleanup_allowed"),
|
||||
"cleanup_performed": False,
|
||||
"no_merge_or_adoption": True,
|
||||
"mode": "apply" if apply else "read_only",
|
||||
"reasons": assessment.get("reasons") or [],
|
||||
}
|
||||
|
||||
if not apply:
|
||||
return report
|
||||
if not assessment.get("cleanup_allowed"):
|
||||
report["cleanup_skipped_reason"] = (
|
||||
assessment.get("reasons") or ["cleanup not allowed"]
|
||||
)
|
||||
return report
|
||||
|
||||
comment_block = _profile_operation_gate("gitea.pr.comment")
|
||||
if comment_block:
|
||||
report["success"] = False
|
||||
report["reasons"] = comment_block
|
||||
report["permission_report"] = _permission_block_report("gitea.pr.comment")
|
||||
return report
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
body = assessment["release_body"]
|
||||
comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
|
||||
with _audited(
|
||||
"comment_pr",
|
||||
host=h,
|
||||
remote=remote,
|
||||
org=o,
|
||||
repo=r,
|
||||
pr_number=pr_number,
|
||||
request_metadata={"source": "cleanup_post_merge_moot_lease"},
|
||||
):
|
||||
posted = api_request("POST", comment_url, auth, {"body": body})
|
||||
|
||||
report["cleanup_performed"] = True
|
||||
report["released_comment_id"] = posted.get("id")
|
||||
return report
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_list_issue_comments(
|
||||
issue_number: int,
|
||||
@@ -5774,6 +6133,8 @@ _PROJECT_SKILLS = {
|
||||
"steps": [
|
||||
"Resolve task first: gitea_resolve_task_capability(task='review_pr') "
|
||||
"to confirm reviewer namespace and avoid author-profile blocks.",
|
||||
"Load canonical workflow proof with gitea_load_review_workflow "
|
||||
"before any review/merge mutation (#389).",
|
||||
"Verify reviewer identity with gitea_whoami; the PR author "
|
||||
"must be a different user.",
|
||||
"Reconcile live queue state FIRST (do not trust prior handoffs): "
|
||||
@@ -6697,6 +7058,8 @@ def gitea_get_runtime_context(
|
||||
),
|
||||
"role_kind": _role_kind(allowed, forbidden),
|
||||
"shell_health": native_mcp_preference.shell_health_status(),
|
||||
"workflow_load_proof": review_workflow_load.workflow_load_status(
|
||||
PROJECT_ROOT),
|
||||
}
|
||||
|
||||
if reveal and h:
|
||||
@@ -6705,6 +7068,42 @@ def gitea_get_runtime_context(
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_load_review_workflow(
|
||||
prompt_text: str | None = None,
|
||||
) -> dict:
|
||||
"""Load and record canonical review-merge workflow proof for this session (#389).
|
||||
|
||||
Read-only with respect to Gitea API; records in-process workflow source/hash
|
||||
proof required before reviewer review or merge mutations.
|
||||
"""
|
||||
try:
|
||||
recorded = review_workflow_load.record_review_workflow_load(
|
||||
PROJECT_ROOT, prompt_text=prompt_text)
|
||||
except OSError as exc:
|
||||
return {
|
||||
"success": False,
|
||||
"loaded": False,
|
||||
"reasons": [str(exc)],
|
||||
"recovery_handoff": review_workflow_load.recovery_handoff_without_replay(),
|
||||
}
|
||||
return {
|
||||
"success": True,
|
||||
"loaded": True,
|
||||
"workflow_source": recorded["workflow_source"],
|
||||
"task_mode": recorded["task_mode"],
|
||||
"workflow_hash": recorded["workflow_hash"],
|
||||
"workflow_version": recorded["workflow_version"],
|
||||
"final_report_schema_path": recorded["final_report_schema_path"],
|
||||
"final_report_schema_hash": recorded["final_report_schema_hash"],
|
||||
"prompt_conflicts_with_workflow": recorded[
|
||||
"prompt_conflicts_with_workflow"],
|
||||
"prompt_conflict_reasons": recorded.get("prompt_conflict_reasons") or [],
|
||||
"workflow_load_proof_present": True,
|
||||
"reasons": [],
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_list_profiles() -> dict:
|
||||
"""Read-only: list all Gitea MCP profiles with redacted metadata.
|
||||
@@ -7873,7 +8272,20 @@ def gitea_resolve_task_capability(
|
||||
}
|
||||
if reason_msg:
|
||||
result["reason"] = reason_msg
|
||||
if task in ("review_pr", "merge_pr"):
|
||||
result["workflow_load_proof"] = review_workflow_load.workflow_load_status(
|
||||
PROJECT_ROOT)
|
||||
if not result["workflow_load_proof"].get("workflow_load_valid"):
|
||||
guidance = (
|
||||
"Call gitea_load_review_workflow before any reviewer review "
|
||||
"or merge mutation."
|
||||
)
|
||||
if guidance not in task_role_guidance:
|
||||
task_role_guidance.append(guidance)
|
||||
role_session_router.sync_route_from_capability(result)
|
||||
if audit_reconciliation_mode.check_audit_task_enters_phase(task):
|
||||
phase_record = audit_reconciliation_mode.enter_audit_phase(task)
|
||||
result["reconciliation_phase"] = phase_record.get("phase")
|
||||
was_terminal = capability_stop_terminal.is_active()
|
||||
terminal = capability_stop_terminal.sync_from_capability_result(result)
|
||||
if terminal:
|
||||
|
||||
@@ -0,0 +1,307 @@
|
||||
"""Namespace-scoped MCP workspace binding (#510).
|
||||
|
||||
Each role namespace (author, reviewer, merger, reconciler) resolves its own
|
||||
active task workspace. Foreign role worktree environment variables must not
|
||||
poison workspace purity checks in another namespace.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
import author_mutation_worktree as amw
|
||||
|
||||
ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
|
||||
AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV
|
||||
REVIEWER_WORKTREE_ENV = "GITEA_REVIEWER_WORKTREE"
|
||||
MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
|
||||
RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
|
||||
|
||||
ROLE_WORKTREE_ENVS: dict[str, str] = {
|
||||
"author": AUTHOR_WORKTREE_ENV,
|
||||
"reviewer": REVIEWER_WORKTREE_ENV,
|
||||
"merger": MERGER_WORKTREE_ENV,
|
||||
"reconciler": RECONCILER_WORKTREE_ENV,
|
||||
}
|
||||
|
||||
NON_AUTHOR_ROLES = frozenset({"reviewer", "merger", "reconciler"})
|
||||
|
||||
|
||||
def normalize_role_kind(
|
||||
role_kind: str | None,
|
||||
*,
|
||||
profile_name: str | None = None,
|
||||
) -> str:
|
||||
"""Map profile/task role to a workspace namespace key."""
|
||||
role = (role_kind or "author").strip().lower()
|
||||
profile = (profile_name or "").strip().lower()
|
||||
if role == "reviewer" and "merger" in profile:
|
||||
return "merger"
|
||||
if role in ROLE_WORKTREE_ENVS:
|
||||
return role
|
||||
return "author"
|
||||
|
||||
|
||||
def _env_value(env: dict[str, str] | os._Environ, key: str) -> str | None:
|
||||
text = (env.get(key) or "").strip()
|
||||
return text or None
|
||||
|
||||
|
||||
def resolve_namespace_workspace(
|
||||
*,
|
||||
role_kind: str,
|
||||
worktree_path: str | None = None,
|
||||
worktree: str | None = None,
|
||||
process_project_root: str,
|
||||
env: dict[str, str] | os._Environ | None = None,
|
||||
session_lease_worktree: str | None = None,
|
||||
profile_name: str | None = None,
|
||||
) -> tuple[str, str]:
|
||||
"""Return ``(resolved_path, binding_source)`` for *role_kind*."""
|
||||
env_map = env if env is not None else os.environ
|
||||
role = normalize_role_kind(role_kind, profile_name=profile_name)
|
||||
role_env_key = ROLE_WORKTREE_ENVS[role]
|
||||
|
||||
for candidate, source in (
|
||||
(worktree_path, "worktree_path argument"),
|
||||
(worktree, "worktree argument"),
|
||||
(_env_value(env_map, ACTIVE_WORKTREE_ENV), f"{ACTIVE_WORKTREE_ENV} environment variable"),
|
||||
(_env_value(env_map, role_env_key), f"{role_env_key} environment variable"),
|
||||
(session_lease_worktree if role in {"reviewer", "merger"} else None,
|
||||
"reviewer PR lease worktree"),
|
||||
):
|
||||
text = (candidate or "").strip()
|
||||
if text:
|
||||
return os.path.realpath(os.path.abspath(text)), source
|
||||
|
||||
return os.path.realpath(process_project_root), "MCP server process root (default)"
|
||||
|
||||
|
||||
def resolve_namespace_mutation_context(
|
||||
*,
|
||||
role_kind: str,
|
||||
worktree_path: str | None,
|
||||
process_project_root: str,
|
||||
env: dict[str, str] | os._Environ | None = None,
|
||||
session_lease_worktree: str | None = None,
|
||||
worktree: str | None = None,
|
||||
profile_name: str | None = None,
|
||||
) -> dict:
|
||||
"""Shared workspace resolution for runtime_context and mutation guards."""
|
||||
workspace, binding_source = resolve_namespace_workspace(
|
||||
role_kind=role_kind,
|
||||
worktree_path=worktree_path,
|
||||
worktree=worktree,
|
||||
process_project_root=process_project_root,
|
||||
env=env,
|
||||
session_lease_worktree=session_lease_worktree,
|
||||
profile_name=profile_name,
|
||||
)
|
||||
process_root = os.path.realpath(process_project_root)
|
||||
role = normalize_role_kind(role_kind, profile_name=profile_name)
|
||||
pollution = assess_foreign_role_worktree_pollution(
|
||||
role_kind=role,
|
||||
resolved_workspace=workspace,
|
||||
binding_source=binding_source,
|
||||
env=env,
|
||||
profile_name=profile_name,
|
||||
)
|
||||
canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
|
||||
return {
|
||||
"workspace_path": workspace,
|
||||
"workspace_binding_source": binding_source,
|
||||
"workspace_role_kind": role,
|
||||
"ignored_bindings": pollution.get("ignored_bindings") or [],
|
||||
"process_project_root": process_root,
|
||||
"canonical_repo_root": canonical_root,
|
||||
"roots_aligned": canonical_root == process_root,
|
||||
}
|
||||
|
||||
|
||||
def assess_foreign_role_worktree_pollution(
|
||||
*,
|
||||
role_kind: str,
|
||||
resolved_workspace: str,
|
||||
binding_source: str,
|
||||
env: dict[str, str] | os._Environ | None = None,
|
||||
profile_name: str | None = None,
|
||||
) -> dict:
|
||||
"""Detect when a foreign role env would have hijacked workspace binding."""
|
||||
env_map = env if env is not None else os.environ
|
||||
role = normalize_role_kind(role_kind, profile_name=profile_name)
|
||||
if role == "author":
|
||||
return {"would_pollute": False, "ignored_bindings": []}
|
||||
|
||||
ignored: list[str] = []
|
||||
author_path = _env_value(env_map, AUTHOR_WORKTREE_ENV)
|
||||
if author_path:
|
||||
author_real = os.path.realpath(os.path.abspath(author_path))
|
||||
resolved_real = os.path.realpath(resolved_workspace)
|
||||
if author_real != resolved_real and binding_source != f"{AUTHOR_WORKTREE_ENV} environment variable":
|
||||
ignored.append(
|
||||
f"{AUTHOR_WORKTREE_ENV}={author_real} (ignored for {role} namespace)"
|
||||
)
|
||||
return {
|
||||
"would_pollute": bool(ignored),
|
||||
"ignored_bindings": ignored,
|
||||
}
|
||||
|
||||
|
||||
def assess_metadata_only_worktree_binding(
|
||||
*,
|
||||
role_kind: str,
|
||||
declared_worktree_path: str | None,
|
||||
mutation_workspace: str,
|
||||
process_project_root: str,
|
||||
profile_name: str | None = None,
|
||||
) -> dict:
|
||||
"""Fail closed when declared worktree_path would not redirect mutations."""
|
||||
declared = (declared_worktree_path or "").strip()
|
||||
process_root = os.path.realpath(process_project_root)
|
||||
mutation_root = os.path.realpath(mutation_workspace)
|
||||
role = normalize_role_kind(role_kind, profile_name=profile_name)
|
||||
if not declared:
|
||||
return {"block": False, "reasons": [], "metadata_only": False}
|
||||
|
||||
declared_root = os.path.realpath(os.path.abspath(declared))
|
||||
if declared_root == mutation_root:
|
||||
return {"block": False, "reasons": [], "metadata_only": False}
|
||||
|
||||
if declared_root != process_root and mutation_root == process_root:
|
||||
return {
|
||||
"block": True,
|
||||
"metadata_only": True,
|
||||
"reasons": [
|
||||
f"worktree_path is metadata-only for {role} mutations: preflight "
|
||||
f"inspected '{declared_root}' but mutation tools would still "
|
||||
f"validate MCP server process root '{process_root}'"
|
||||
],
|
||||
"declared_worktree_path": declared_root,
|
||||
"mutation_workspace": mutation_root,
|
||||
"process_project_root": process_root,
|
||||
}
|
||||
|
||||
return {"block": False, "reasons": [], "metadata_only": False}
|
||||
|
||||
|
||||
def format_namespace_workspace_binding_error(
|
||||
*,
|
||||
role_kind: str,
|
||||
workspace_path: str,
|
||||
binding_source: str,
|
||||
reasons: list[str] | None = None,
|
||||
ignored_bindings: list[str] | None = None,
|
||||
dirty_files: list[str] | None = None,
|
||||
) -> str:
|
||||
"""Canonical error when namespace workspace binding blocks mutations."""
|
||||
role = normalize_role_kind(role_kind)
|
||||
workspace = os.path.realpath(workspace_path)
|
||||
parts = [
|
||||
f"Namespace workspace binding blocked ({role} namespace, #510): "
|
||||
f"resolved workspace '{workspace}' via {binding_source}."
|
||||
]
|
||||
if ignored_bindings:
|
||||
parts.append(
|
||||
"Foreign role bindings ignored: " + "; ".join(ignored_bindings) + "."
|
||||
)
|
||||
if dirty_files:
|
||||
parts.append(
|
||||
"Dirty tracked files in active task workspace: "
|
||||
+ ", ".join(dirty_files)
|
||||
+ "."
|
||||
)
|
||||
if reasons:
|
||||
parts.append("Details: " + "; ".join(reasons) + ".")
|
||||
parts.append(
|
||||
"Remediation: reconnect or relaunch the MCP server from a clean dedicated "
|
||||
f"branches/ {role} worktree, set "
|
||||
f"{ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} or {ACTIVE_WORKTREE_ENV} "
|
||||
"to that path, or pass worktree_path on mutation tools. Do not clean or "
|
||||
"reset foreign role worktrees to unblock this namespace."
|
||||
)
|
||||
return " ".join(parts)
|
||||
|
||||
|
||||
def assess_namespace_mutation_workspace(
|
||||
*,
|
||||
role_kind: str,
|
||||
worktree_path: str | None,
|
||||
worktree: str | None,
|
||||
process_project_root: str,
|
||||
env: dict[str, str] | os._Environ | None = None,
|
||||
session_lease_worktree: str | None = None,
|
||||
profile_name: str | None = None,
|
||||
current_branch: str | None = None,
|
||||
) -> dict:
|
||||
"""Evaluate namespace workspace binding before preflight/mutation."""
|
||||
ctx = resolve_namespace_mutation_context(
|
||||
role_kind=role_kind,
|
||||
worktree_path=worktree_path,
|
||||
worktree=worktree,
|
||||
process_project_root=process_project_root,
|
||||
env=env,
|
||||
session_lease_worktree=session_lease_worktree,
|
||||
profile_name=profile_name,
|
||||
)
|
||||
mutation_workspace = ctx["workspace_path"]
|
||||
binding_source = ctx["workspace_binding_source"]
|
||||
role = ctx["workspace_role_kind"]
|
||||
process_root = ctx["process_project_root"]
|
||||
|
||||
metadata = assess_metadata_only_worktree_binding(
|
||||
role_kind=role,
|
||||
declared_worktree_path=worktree_path,
|
||||
mutation_workspace=mutation_workspace,
|
||||
process_project_root=process_root,
|
||||
profile_name=profile_name,
|
||||
)
|
||||
pollution = assess_foreign_role_worktree_pollution(
|
||||
role_kind=role,
|
||||
resolved_workspace=mutation_workspace,
|
||||
binding_source=binding_source,
|
||||
env=env,
|
||||
profile_name=profile_name,
|
||||
)
|
||||
|
||||
reasons = list(metadata.get("reasons") or [])
|
||||
if role == "author":
|
||||
branches = amw.assess_author_mutation_worktree(
|
||||
workspace_path=mutation_workspace,
|
||||
project_root=ctx["canonical_repo_root"],
|
||||
current_branch=current_branch,
|
||||
)
|
||||
if branches["block"]:
|
||||
reasons.extend(branches["reasons"])
|
||||
elif (
|
||||
role == "reviewer"
|
||||
and mutation_workspace == process_root
|
||||
and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
|
||||
):
|
||||
reasons.append(
|
||||
f"{role} mutation blocked: workspace is the stable control checkout; "
|
||||
f"create or reconnect to a session-owned worktree under branches/ "
|
||||
f"or set {ROLE_WORKTREE_ENVS[role]} / {ACTIVE_WORKTREE_ENV}"
|
||||
)
|
||||
elif (
|
||||
role in {"reviewer", "merger"}
|
||||
and mutation_workspace != process_root
|
||||
and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
|
||||
):
|
||||
reasons.append(
|
||||
f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
|
||||
f"'{ctx['canonical_repo_root']}/branches/'"
|
||||
)
|
||||
|
||||
block = bool(reasons)
|
||||
return {
|
||||
"block": block,
|
||||
"reasons": reasons,
|
||||
"mutation_workspace": mutation_workspace,
|
||||
"workspace_binding_source": binding_source,
|
||||
"workspace_role_kind": role,
|
||||
"process_project_root": process_root,
|
||||
"canonical_repo_root": ctx["canonical_repo_root"],
|
||||
"metadata_only": metadata.get("metadata_only", False),
|
||||
"declared_worktree_path": metadata.get("declared_worktree_path"),
|
||||
"ignored_bindings": pollution.get("ignored_bindings") or [],
|
||||
}
|
||||
@@ -5115,6 +5115,15 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
|
||||
return _assess(report_text or "")
|
||||
|
||||
|
||||
def assess_audit_reconciliation_report(report_text: str | None) -> dict:
|
||||
"""#419: validate audit vs cleanup reconciliation report boundaries."""
|
||||
from audit_reconciliation_mode import (
|
||||
assess_audit_reconciliation_report as _assess,
|
||||
)
|
||||
|
||||
return _assess(report_text or "")
|
||||
|
||||
|
||||
_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
|
||||
|
||||
_NOT_APPLICABLE_VALUE = re.compile(
|
||||
|
||||
@@ -0,0 +1,198 @@
|
||||
"""Canonical review-merge workflow load proof for reviewer mutations (#389)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
WORKFLOW_REL_PATH = (
|
||||
"skills/llm-project-workflow/workflows/review-merge-pr.md"
|
||||
)
|
||||
SCHEMA_REL_PATH = (
|
||||
"skills/llm-project-workflow/schemas/review-merge-final-report.md"
|
||||
)
|
||||
TASK_MODE = "review-merge-pr"
|
||||
LOAD_TOOL_NAME = "gitea_load_review_workflow"
|
||||
|
||||
_REVIEW_WORKFLOW_LOAD: dict | None = None
|
||||
|
||||
|
||||
def compute_content_hash(text: str) -> str:
|
||||
"""Short deterministic hash for workflow/schema version proof."""
|
||||
return hashlib.sha256((text or "").encode("utf-8")).hexdigest()[:12]
|
||||
|
||||
|
||||
def _read_text(path: Path) -> str:
|
||||
return path.read_text(encoding="utf-8")
|
||||
|
||||
|
||||
def _canonical_paths(project_root: str) -> tuple[Path, Path]:
|
||||
root = Path(project_root)
|
||||
workflow = root / WORKFLOW_REL_PATH
|
||||
schema = root / SCHEMA_REL_PATH
|
||||
if not workflow.is_file():
|
||||
raise FileNotFoundError(f"canonical workflow missing: {workflow}")
|
||||
if not schema.is_file():
|
||||
raise FileNotFoundError(f"final report schema missing: {schema}")
|
||||
return workflow, schema
|
||||
|
||||
|
||||
def build_canonical_workflow_metadata(
|
||||
project_root: str,
|
||||
*,
|
||||
prompt_text: str | None = None,
|
||||
) -> dict:
|
||||
"""Load workflow + schema from disk and compute proof metadata."""
|
||||
workflow_path, schema_path = _canonical_paths(project_root)
|
||||
workflow_text = _read_text(workflow_path)
|
||||
schema_text = _read_text(schema_path)
|
||||
workflow_hash = compute_content_hash(workflow_text)
|
||||
schema_hash = compute_content_hash(schema_text)
|
||||
conflict, conflict_reasons = assess_prompt_conflict(prompt_text)
|
||||
return {
|
||||
"workflow_source": WORKFLOW_REL_PATH,
|
||||
"workflow_path": str(workflow_path),
|
||||
"task_mode": TASK_MODE,
|
||||
"workflow_hash": workflow_hash,
|
||||
"workflow_version": workflow_hash,
|
||||
"final_report_schema_path": SCHEMA_REL_PATH,
|
||||
"final_report_schema_hash": schema_hash,
|
||||
"prompt_conflicts_with_workflow": conflict,
|
||||
"prompt_conflict_reasons": conflict_reasons,
|
||||
"load_tool": LOAD_TOOL_NAME,
|
||||
}
|
||||
|
||||
|
||||
def assess_prompt_conflict(prompt_text: str | None) -> tuple[bool, list[str]]:
|
||||
"""Detect obvious task-mode conflicts between prompt and review workflow."""
|
||||
if not (prompt_text or "").strip():
|
||||
return False, []
|
||||
text = prompt_text.lower()
|
||||
reasons: list[str] = []
|
||||
conflicting = (
|
||||
(r"\bwork[- ]issue\b", "work-issue author mode"),
|
||||
(r"\bcreate[- ]issue\b", "create-issue mode"),
|
||||
(r"\bauthor/coder\b", "author/coder mode"),
|
||||
(r"\breconcile[- ]landed\b", "reconcile-landed mode"),
|
||||
)
|
||||
for pattern, label in conflicting:
|
||||
if re.search(pattern, text):
|
||||
reasons.append(
|
||||
f"active prompt appears to request {label} while loading "
|
||||
f"{TASK_MODE} workflow"
|
||||
)
|
||||
return bool(reasons), reasons
|
||||
|
||||
|
||||
def record_review_workflow_load(
|
||||
project_root: str,
|
||||
*,
|
||||
prompt_text: str | None = None,
|
||||
) -> dict:
|
||||
"""Record in-process workflow load proof for the current MCP session."""
|
||||
global _REVIEW_WORKFLOW_LOAD
|
||||
meta = build_canonical_workflow_metadata(
|
||||
project_root, prompt_text=prompt_text)
|
||||
_REVIEW_WORKFLOW_LOAD = {
|
||||
**meta,
|
||||
"session_pid": os.getpid(),
|
||||
"loaded": True,
|
||||
}
|
||||
return dict(_REVIEW_WORKFLOW_LOAD)
|
||||
|
||||
|
||||
def clear_review_workflow_load() -> None:
|
||||
"""Test helper and review_pr session reset."""
|
||||
global _REVIEW_WORKFLOW_LOAD
|
||||
_REVIEW_WORKFLOW_LOAD = None
|
||||
|
||||
|
||||
def workflow_load_status(project_root: str | None = None) -> dict:
|
||||
"""Non-throwing status for capability/runtime reports."""
|
||||
load = _REVIEW_WORKFLOW_LOAD
|
||||
if load is None:
|
||||
return {
|
||||
"workflow_load_proof_present": False,
|
||||
"workflow_load_valid": False,
|
||||
"workflow_source": None,
|
||||
"workflow_hash": None,
|
||||
"final_report_schema_path": SCHEMA_REL_PATH,
|
||||
"reasons": [
|
||||
f"{LOAD_TOOL_NAME} has not been called in this session "
|
||||
"(fail closed for reviewer mutations)"
|
||||
],
|
||||
}
|
||||
reasons = _session_validation_reasons(load, project_root)
|
||||
return {
|
||||
"workflow_load_proof_present": True,
|
||||
"workflow_load_valid": not reasons,
|
||||
"workflow_source": load.get("workflow_source"),
|
||||
"workflow_hash": load.get("workflow_hash"),
|
||||
"task_mode": load.get("task_mode"),
|
||||
"final_report_schema_path": load.get("final_report_schema_path"),
|
||||
"final_report_schema_hash": load.get("final_report_schema_hash"),
|
||||
"prompt_conflicts_with_workflow": load.get(
|
||||
"prompt_conflicts_with_workflow"),
|
||||
"session_pid": load.get("session_pid"),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def _session_validation_reasons(
|
||||
load: dict,
|
||||
project_root: str | None,
|
||||
) -> list[str]:
|
||||
reasons: list[str] = []
|
||||
if load.get("session_pid") != os.getpid():
|
||||
reasons.append(
|
||||
"workflow load proof was recorded in a different process "
|
||||
"(fail closed)"
|
||||
)
|
||||
return reasons
|
||||
if load.get("prompt_conflicts_with_workflow"):
|
||||
reasons.extend(load.get("prompt_conflict_reasons") or [
|
||||
"active prompt conflicts with loaded review-merge workflow"
|
||||
])
|
||||
if project_root:
|
||||
try:
|
||||
current = build_canonical_workflow_metadata(project_root)
|
||||
except OSError as exc:
|
||||
reasons.append(f"cannot re-verify workflow hash: {exc}")
|
||||
return reasons
|
||||
if current["workflow_hash"] != load.get("workflow_hash"):
|
||||
reasons.append(
|
||||
"stored workflow hash is stale; reload via "
|
||||
f"{LOAD_TOOL_NAME} (fail closed)"
|
||||
)
|
||||
if current["final_report_schema_hash"] != load.get(
|
||||
"final_report_schema_hash"):
|
||||
reasons.append(
|
||||
"stored final-report schema hash is stale; reload via "
|
||||
f"{LOAD_TOOL_NAME} (fail closed)"
|
||||
)
|
||||
return reasons
|
||||
|
||||
|
||||
def review_workflow_load_blockers(
|
||||
project_root: str | None = None,
|
||||
) -> list[str]:
|
||||
"""Reasons reviewer mutations must fail closed."""
|
||||
status = workflow_load_status(project_root)
|
||||
if not status.get("workflow_load_proof_present"):
|
||||
return list(status.get("reasons") or [])
|
||||
if not status.get("workflow_load_valid"):
|
||||
return list(status.get("reasons") or [])
|
||||
return []
|
||||
|
||||
|
||||
def recovery_handoff_without_replay() -> list[str]:
|
||||
"""Safe next-step lines that must not include approve/merge replay."""
|
||||
return [
|
||||
"Reload the canonical workflow via gitea_load_review_workflow, then "
|
||||
"rerun the full review-merge workflow from inventory.",
|
||||
"Do not call gitea_submit_pr_review, gitea_mark_final_review_decision, "
|
||||
"or gitea_merge_pr until workflow-load proof is present.",
|
||||
"Do not include approve/merge replay commands in the recovery handoff.",
|
||||
]
|
||||
+108
-1
@@ -217,12 +217,24 @@ def assess_acquire_lease(
|
||||
candidate_head: str | None,
|
||||
target_branch: str,
|
||||
target_branch_sha: str | None,
|
||||
pr_merged_or_closed: bool = False,
|
||||
now: datetime | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Fail closed when another session holds an active lease."""
|
||||
"""Fail closed when another session holds an active lease.
|
||||
|
||||
When *pr_merged_or_closed* is true the PR has already merged/closed, so any
|
||||
reviewer-lease acquisition or adoption for merge work is moot: fail closed
|
||||
with a ``post_merge_moot`` reason and never mint a lease body (#515).
|
||||
"""
|
||||
now = now or datetime.now(timezone.utc)
|
||||
reasons: list[str] = []
|
||||
existing = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
|
||||
post_merge_moot = bool(pr_merged_or_closed)
|
||||
if post_merge_moot:
|
||||
reasons.append(
|
||||
f"post_merge_moot: PR #{pr_number} is already merged/closed; reviewer "
|
||||
"lease adoption for merge is moot (fail closed)"
|
||||
)
|
||||
if existing:
|
||||
owner_session = (existing.get("session_id") or "").strip()
|
||||
freshness = existing.get("freshness") or classify_lease_freshness(existing, now=now)
|
||||
@@ -270,6 +282,101 @@ def assess_acquire_lease(
|
||||
"existing_lease": existing,
|
||||
"lease_body": body,
|
||||
"session_id": session_id,
|
||||
"post_merge_moot": post_merge_moot,
|
||||
}
|
||||
|
||||
|
||||
def assess_post_merge_moot_lease(
|
||||
comments: list[dict],
|
||||
*,
|
||||
pr_number: int,
|
||||
pr_merged: bool = False,
|
||||
pr_state: str | None = None,
|
||||
merge_commit_sha: str | None = None,
|
||||
now: datetime | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Assess a reviewer lease left lingering on an already-merged/closed PR (#515).
|
||||
|
||||
Read-first and fail-safe:
|
||||
|
||||
- Only treats a lease as moot when the live PR state is merged/closed.
|
||||
- Never proposes touching an *active* lease while the PR is still open
|
||||
(``cleanup_allowed`` stays false and a refusal reason is returned).
|
||||
- When the PR is merged/closed and a lease is still active, ``cleanup_allowed``
|
||||
is true and a terminal ``phase: released`` lease body (``blocker:
|
||||
post-merge-moot``) is provided so the moot lease can be neutralised by an
|
||||
append-only comment — never by deleting a foreign session's comment, and
|
||||
never by adopting or merging.
|
||||
|
||||
Posting the released body makes that lease terminal, so a subsequent call
|
||||
finds no active lease and reports nothing left to clean (idempotent).
|
||||
"""
|
||||
now = now or datetime.now(timezone.utc)
|
||||
merged_or_closed = bool(pr_merged) or (
|
||||
str(pr_state or "").strip().lower() == "closed"
|
||||
)
|
||||
active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
|
||||
# The newest lease comment is authoritative: once a terminal marker
|
||||
# (released/done/blocked) is the latest entry, the lease is resolved even if
|
||||
# an earlier non-terminal comment from the same session still lingers. This
|
||||
# keeps post-merge cleanup idempotent.
|
||||
entries = _lease_entries(comments, pr_number=pr_number)
|
||||
newest = entries[-1] if entries else None
|
||||
newest_terminal = bool(newest) and (
|
||||
(newest.get("phase") or "").strip().lower() in _TERMINAL_PHASES
|
||||
)
|
||||
reasons: list[str] = []
|
||||
cleanup_allowed = False
|
||||
release_body: str | None = None
|
||||
is_moot = bool(active) and merged_or_closed and not newest_terminal
|
||||
|
||||
if not merged_or_closed:
|
||||
if active:
|
||||
reasons.append(
|
||||
f"PR #{pr_number} is still open; refusing to touch active reviewer "
|
||||
"lease (fail closed)"
|
||||
)
|
||||
else:
|
||||
reasons.append(
|
||||
f"PR #{pr_number} is still open; no post-merge lease cleanup applicable"
|
||||
)
|
||||
elif newest_terminal:
|
||||
reasons.append(
|
||||
f"PR #{pr_number} reviewer lease already released/terminal; nothing to clean"
|
||||
)
|
||||
elif active:
|
||||
cleanup_allowed = True
|
||||
release_body = format_lease_body(
|
||||
repo=active.get("repo") or "",
|
||||
pr_number=pr_number,
|
||||
issue_number=active.get("issue_number"),
|
||||
reviewer_identity=active.get("reviewer_identity") or "",
|
||||
profile=active.get("profile") or "unknown",
|
||||
session_id=active.get("session_id") or "",
|
||||
worktree=active.get("worktree") or "",
|
||||
phase="released",
|
||||
candidate_head=active.get("candidate_head"),
|
||||
target_branch=active.get("target_branch") or "master",
|
||||
target_branch_sha=active.get("target_branch_sha"),
|
||||
last_activity=now,
|
||||
blocker="post-merge-moot",
|
||||
)
|
||||
else:
|
||||
reasons.append(
|
||||
f"PR #{pr_number} is merged/closed but no active reviewer lease remains; "
|
||||
"nothing to clean"
|
||||
)
|
||||
|
||||
return {
|
||||
"pr_number": pr_number,
|
||||
"pr_state": pr_state,
|
||||
"pr_merged_or_closed": merged_or_closed,
|
||||
"merge_commit_sha": merge_commit_sha,
|
||||
"active_lease": active,
|
||||
"is_moot": is_moot,
|
||||
"cleanup_allowed": cleanup_allowed,
|
||||
"release_body": release_body,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -304,6 +304,40 @@ If any required mutation capability is missing:
|
||||
* include safe next action (profile switch, human close, or dedicated reconciler
|
||||
profile)
|
||||
|
||||
## 15A. Audit vs cleanup phase (#419)
|
||||
|
||||
Reconciliation audits are **read-only** unless a separate cleanup phase is
|
||||
explicitly authorized.
|
||||
|
||||
**Audit phase forbids** (``audit_reconciliation_mode.check_audit_mutation_allowed``
|
||||
fails closed):
|
||||
|
||||
* ``gitea_delete_branch``
|
||||
* ``git branch -D``
|
||||
* ``git worktree remove``
|
||||
* pushes
|
||||
* issue/PR mutations
|
||||
* file edits
|
||||
|
||||
Dry-run merged-cleanup reconciliation (``gitea_reconcile_merged_cleanups`` with
|
||||
``dry_run=True``) stays in audit phase. Execution requires:
|
||||
|
||||
1. Operator approval or workflow authorization
|
||||
2. Exact ``delete_branch`` capability proof (``gitea.branch.delete``)
|
||||
3. Proof branch/worktree is safe to remove
|
||||
4. Before/after state snapshot
|
||||
|
||||
Call ``gitea_authorize_reconciliation_cleanup_phase`` before any cleanup
|
||||
mutation. Final reports must not claim ``no mutations`` if cleanup occurred.
|
||||
Classify cleanup mutations as:
|
||||
|
||||
* remote branch deletion → **External-state mutations**
|
||||
* local branch deletion → **Git ref mutations**
|
||||
* worktree removal → **Cleanup mutations**
|
||||
|
||||
``audit_reconciliation_mode.assess_audit_reconciliation_report`` validates
|
||||
these boundaries in final reports.
|
||||
|
||||
## 16. Mutation classification
|
||||
|
||||
Use precise mutation categories in the final report:
|
||||
|
||||
@@ -108,6 +108,10 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"permission": "gitea.read",
|
||||
"role": "author",
|
||||
},
|
||||
"reconciliation_cleanup": {
|
||||
"permission": "gitea.branch.delete",
|
||||
"role": "author",
|
||||
},
|
||||
"work_issue": {
|
||||
"permission": "gitea.pr.create",
|
||||
"role": "author",
|
||||
|
||||
+6
-4
@@ -157,6 +157,7 @@ class _AuditWiringBase(unittest.TestCase):
|
||||
|
||||
def tearDown(self):
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
mcp_server.review_workflow_load.clear_review_workflow_load()
|
||||
self._dir.cleanup()
|
||||
|
||||
def _env(self, **extra):
|
||||
@@ -291,12 +292,11 @@ class TestGatedToolAudit(_AuditWiringBase):
|
||||
|
||||
def setUp(self):
|
||||
super().setUp()
|
||||
from mcp_server import init_review_decision_lock
|
||||
from tests.test_mcp_server import _install_owned_reviewer_lease
|
||||
from tests.test_mcp_server import _init_reviewer_session, _install_owned_reviewer_lease
|
||||
import reviewer_pr_lease
|
||||
|
||||
# init_review_decision_lock clears any prior session lease (#407).
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
# Session init clears any prior session lease (#407) and loads workflow (#389).
|
||||
_init_reviewer_session("prgs")
|
||||
self._lease_patch = _install_owned_reviewer_lease(8)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
@@ -337,6 +337,7 @@ class TestGatedToolAudit(_AuditWiringBase):
|
||||
env = self._env(GITEA_PROFILE_NAME="gitea-merger",
|
||||
GITEA_ALLOWED_OPERATIONS="read,merge")
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8",
|
||||
expected_head_sha="abc123", remote="prgs")
|
||||
self.assertTrue(r["performed"])
|
||||
@@ -356,6 +357,7 @@ class TestGatedToolAudit(_AuditWiringBase):
|
||||
env = self._env(GITEA_PROFILE_NAME="gitea-merger",
|
||||
GITEA_ALLOWED_OPERATIONS="read,merge")
|
||||
with patch.dict(os.environ, env, clear=True):
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
r = gitea_merge_pr(pr_number=8, confirmation="MERGE PR 8", remote="prgs")
|
||||
self.assertFalse(r["performed"])
|
||||
recs = self._records()
|
||||
|
||||
@@ -0,0 +1,291 @@
|
||||
"""Tests for audit vs cleanup reconciliation mode (#419)."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import audit_reconciliation_mode as arm
|
||||
import mcp_server
|
||||
from audit_reconciliation_mode import (
|
||||
AUDIT_FORBIDDEN_TASKS,
|
||||
PHASE_AUDIT,
|
||||
PHASE_CLEANUP,
|
||||
assess_audit_command_allowed,
|
||||
assess_audit_reconciliation_report,
|
||||
authorize_cleanup_phase,
|
||||
check_audit_mutation_allowed,
|
||||
check_cleanup_execution_allowed,
|
||||
classify_cleanup_mutation,
|
||||
clear_phase,
|
||||
enter_audit_phase,
|
||||
)
|
||||
from final_report_validator import assess_final_report_validator
|
||||
from review_proofs import assess_audit_reconciliation_report as proofs_assess
|
||||
from task_capability_map import required_permission, required_role
|
||||
|
||||
DELETE_PROFILE = {
|
||||
"profile_name": "prgs-author-delete",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-author-delete",
|
||||
}
|
||||
|
||||
READ_PROFILE = {
|
||||
"profile_name": "prgs-author",
|
||||
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
|
||||
"forbidden_operations": ["gitea.branch.delete"],
|
||||
"audit_label": "prgs-author",
|
||||
}
|
||||
|
||||
READ_ENV = {
|
||||
"GITEA_MCP_CONFIG": os.path.join(
|
||||
os.path.dirname(__file__), "..", "profiles.json"
|
||||
),
|
||||
"GITEA_MCP_PROFILE": "prgs-author",
|
||||
}
|
||||
|
||||
|
||||
def _authorize_cleanup(**kwargs):
|
||||
defaults = {
|
||||
"operator_approved": True,
|
||||
"delete_capability_proven": True,
|
||||
"safety_proof": {"safe_to_delete_remote": True},
|
||||
"before_after_snapshot": {
|
||||
"before": "remote branch exists",
|
||||
"after": "remote branch absent",
|
||||
},
|
||||
}
|
||||
defaults.update(kwargs)
|
||||
return authorize_cleanup_phase(**defaults)
|
||||
|
||||
|
||||
def _cleanup_report(**overrides):
|
||||
base = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"Workflow source: workflows/reconcile-landed-pr.md\n"
|
||||
"Audit phase: read-only assessment\n"
|
||||
"Cleanup phase authorized: true\n"
|
||||
"Delete-branch capability proven: true\n"
|
||||
"Branch safe to remove: true\n"
|
||||
"Before/after state snapshot: remote branch feat/x present → absent\n"
|
||||
"External-state mutations: deleted remote branch feat/x\n"
|
||||
"Git ref mutations: none\n"
|
||||
"Cleanup mutations: removed worktree branches/feat-x\n"
|
||||
)
|
||||
for key, value in overrides.items():
|
||||
base = base.replace(key, value)
|
||||
return base
|
||||
|
||||
|
||||
class TestAuditPhaseGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile-landed-pr")
|
||||
|
||||
def tearDown(self):
|
||||
clear_phase()
|
||||
|
||||
def test_audit_blocks_delete_branch_task(self):
|
||||
allowed, reasons = check_audit_mutation_allowed("delete_branch")
|
||||
self.assertFalse(allowed)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
def test_audit_blocks_worktree_shell_commands(self):
|
||||
allowed, reasons = assess_audit_command_allowed(
|
||||
"git worktree remove branches/feat-x"
|
||||
)
|
||||
self.assertFalse(allowed)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
def test_audit_blocks_local_branch_delete_command(self):
|
||||
allowed, reasons = assess_audit_command_allowed("git branch -D feat/x")
|
||||
self.assertFalse(allowed)
|
||||
|
||||
def test_audit_allows_read_tasks(self):
|
||||
allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr")
|
||||
self.assertTrue(allowed)
|
||||
|
||||
def test_forbidden_set_covers_issue_and_pr_mutations(self):
|
||||
for task in ("close_pr", "comment_issue", "commit_files", "push_branch"):
|
||||
self.assertIn(task, AUDIT_FORBIDDEN_TASKS)
|
||||
|
||||
|
||||
class TestCleanupAuthorization(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile_merged_cleanups")
|
||||
|
||||
def tearDown(self):
|
||||
clear_phase()
|
||||
|
||||
def test_cleanup_without_approval_blocked(self):
|
||||
result = authorize_cleanup_phase(
|
||||
delete_capability_proven=True,
|
||||
safety_proof={"safe_to_delete_remote": True},
|
||||
before_after_snapshot={"before": "a", "after": "b"},
|
||||
)
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_cleanup_without_capability_proof_blocked(self):
|
||||
result = _authorize_cleanup(delete_capability_proven=False)
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_cleanup_without_snapshot_blocked(self):
|
||||
result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""})
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_authorized_cleanup_switches_phase(self):
|
||||
result = _authorize_cleanup()
|
||||
self.assertTrue(result["authorized"])
|
||||
self.assertEqual(result["phase"], PHASE_CLEANUP)
|
||||
|
||||
def test_cleanup_execution_allowed_only_after_authorization(self):
|
||||
self.assertFalse(check_cleanup_execution_allowed()[0])
|
||||
_authorize_cleanup()
|
||||
self.assertTrue(check_cleanup_execution_allowed()[0])
|
||||
|
||||
|
||||
class TestReportVerifier(unittest.TestCase):
|
||||
def test_false_no_mutations_after_cleanup_blocked(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"No mutations performed.\n"
|
||||
"delete_remote_branch feat/dup\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("no mutations", result["reasons"][0].lower())
|
||||
|
||||
def test_cleanup_without_authorization_fields_blocked(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"remove_local_worktree branches/feat-x\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_authorized_cleanup_report_passes(self):
|
||||
result = assess_audit_reconciliation_report(_cleanup_report())
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_mutation_classification_enforced(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"Cleanup phase authorized: true\n"
|
||||
"Delete-branch capability proven: true\n"
|
||||
"Branch safe to remove: true\n"
|
||||
"Before/after state snapshot: present\n"
|
||||
"remove_local_worktree branches/feat-x\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_proofs_export_matches_module(self):
|
||||
report = "No mutations performed.\ndelete_remote_branch feat/dup"
|
||||
self.assertEqual(
|
||||
proofs_assess(report)["proven"],
|
||||
assess_audit_reconciliation_report(report)["proven"],
|
||||
)
|
||||
|
||||
def test_final_report_validator_includes_boundary_rule(self):
|
||||
report = "No mutations performed.\ndelete_remote_branch feat/dup"
|
||||
result = assess_final_report_validator(
|
||||
report_text=report,
|
||||
task_kind="reconcile_already_landed",
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("reconcile.audit_cleanup_boundary", rule_ids)
|
||||
|
||||
|
||||
class TestMutationClassification(unittest.TestCase):
|
||||
def test_remote_delete_is_external_state(self):
|
||||
self.assertEqual(
|
||||
classify_cleanup_mutation("delete_remote_branch"),
|
||||
"external-state",
|
||||
)
|
||||
|
||||
def test_worktree_remove_is_cleanup(self):
|
||||
self.assertEqual(
|
||||
classify_cleanup_mutation("remove_local_worktree"),
|
||||
"cleanup",
|
||||
)
|
||||
|
||||
|
||||
class TestMcpGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile_merged_cleanups")
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_auth = patch(
|
||||
"mcp_server.get_auth_header", return_value="token test"
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
clear_phase()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_delete_branch_blocked_in_audit_phase(self, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["audit_phase"], PHASE_AUDIT)
|
||||
self.mock_api.assert_not_called()
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile):
|
||||
with self.assertRaises(ValueError):
|
||||
mcp_server.gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=False,
|
||||
remote="prgs",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=READ_PROFILE)
|
||||
def test_cleanup_auth_fails_without_delete_capability(self, _profile):
|
||||
result = mcp_server.gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved=True,
|
||||
delete_capability_proven=True,
|
||||
safe_to_delete_remote=True,
|
||||
before_state="exists",
|
||||
after_state="gone",
|
||||
)
|
||||
self.assertFalse(result["authorized"])
|
||||
self.assertFalse(result["delete_capability_verified"])
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_delete_branch_allowed_after_cleanup_authorization(self, _profile):
|
||||
mcp_server.gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved=True,
|
||||
delete_capability_proven=True,
|
||||
safe_to_delete_remote=True,
|
||||
before_state="exists",
|
||||
after_state="gone",
|
||||
)
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
self.mock_api.return_value = {}
|
||||
result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
|
||||
|
||||
class TestTaskCapabilityMap(unittest.TestCase):
|
||||
def test_reconciliation_cleanup_maps_delete_permission(self):
|
||||
self.assertEqual(
|
||||
required_permission("reconciliation_cleanup"),
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.assertEqual(required_role("reconciliation_cleanup"), "author")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -22,6 +22,7 @@ from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import mcp_server # noqa: E402
|
||||
from mcp_server import ( # noqa: E402
|
||||
gitea_check_pr_eligibility,
|
||||
gitea_merge_pr,
|
||||
|
||||
@@ -87,6 +87,14 @@ def test_reconcile_landed_workflow_contract():
|
||||
assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
|
||||
assert "RECOVERY_HANDOFF_ONLY" in text
|
||||
assert "resolve_partial_reconciliation_plan" in text
|
||||
assert "check_audit_mutation_allowed" in text
|
||||
assert "gitea_authorize_reconciliation_cleanup_phase" in text
|
||||
|
||||
|
||||
def test_audit_reconciliation_verifier_exported():
|
||||
from review_proofs import assess_audit_reconciliation_report
|
||||
|
||||
assert callable(assess_audit_reconciliation_report)
|
||||
|
||||
|
||||
def test_create_issue_workflow_contract():
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
"""Regression tests for gitea_lock_issue MCP tool registration (#521)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import tempfile
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import issue_lock_store
|
||||
import mcp_server
|
||||
from mcp_server import gitea_create_pr, gitea_lock_issue
|
||||
|
||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||
|
||||
ISSUE_WRITE_ENV = {
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
|
||||
),
|
||||
}
|
||||
|
||||
CREATE_PR_ENV = {
|
||||
"GITEA_PROFILE_NAME": "author-test",
|
||||
"GITEA_ALLOWED_OPERATIONS": (
|
||||
"gitea.read,gitea.pr.create,gitea.branch.push,"
|
||||
"gitea.issue.create,gitea.issue.close,gitea.issue.comment"
|
||||
),
|
||||
"GITEA_FORBIDDEN_OPERATIONS": (
|
||||
"gitea.pr.approve,gitea.pr.merge,gitea.pr.review"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
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",
|
||||
}
|
||||
|
||||
|
||||
def _registered_tool_names() -> set[str]:
|
||||
manager = mcp_server.mcp._tool_manager
|
||||
tools = getattr(manager, "_tools", None) or {}
|
||||
return set(tools.keys())
|
||||
|
||||
|
||||
class TestLockIssueMcpRegistration(unittest.TestCase):
|
||||
def test_gitea_lock_issue_registered_as_public_mcp_tool(self):
|
||||
names = _registered_tool_names()
|
||||
self.assertIn("gitea_lock_issue", names)
|
||||
|
||||
def test_internal_list_open_pulls_not_exposed_as_mcp_tool(self):
|
||||
names = _registered_tool_names()
|
||||
self.assertNotIn("_list_open_pulls", names)
|
||||
|
||||
|
||||
class TestCreatePrLockRegistrationFlow(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._lock_dir = tempfile.TemporaryDirectory()
|
||||
self._env_patcher = patch.dict(
|
||||
os.environ,
|
||||
{
|
||||
**CREATE_PR_ENV,
|
||||
"GITEA_ISSUE_LOCK_DIR": self._lock_dir.name,
|
||||
},
|
||||
clear=True,
|
||||
)
|
||||
self._env_patcher.start()
|
||||
self._dup_fetcher_patcher = patch(
|
||||
"mcp_server.issue_duplicate_context_fetcher",
|
||||
return_value=([], [], {"status": "not_claimed"}),
|
||||
)
|
||||
self._dup_fetcher_patcher.start()
|
||||
|
||||
def tearDown(self):
|
||||
self._dup_fetcher_patcher.stop()
|
||||
self._env_patcher.stop()
|
||||
self._lock_dir.cleanup()
|
||||
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_still_fails_closed_without_issue_lock(self, _auth, _role):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
gitea_create_pr(
|
||||
title="feat: X Closes #521",
|
||||
head="feat/issue-521-lock-issue-registration",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertIn("Issue lock is missing", str(ctx.exception))
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
)
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []))
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_pr_proceeds_after_valid_issue_lock(
|
||||
self, _auth, _role, _api, _git_state, mock_api_request
|
||||
):
|
||||
worktree = os.path.realpath(os.getcwd())
|
||||
mock_api_request.return_value = {"number": 521, "html_url": "https://example/pr/521"}
|
||||
lock_res = gitea_lock_issue(
|
||||
issue_number=521,
|
||||
branch_name="feat/issue-521-lock-issue-registration",
|
||||
remote="prgs",
|
||||
worktree_path=worktree,
|
||||
)
|
||||
self.assertTrue(lock_res["success"])
|
||||
lock_path = lock_res["lock_file_path"]
|
||||
self.assertTrue(os.path.exists(lock_path))
|
||||
lock = issue_lock_store.read_lock_file(lock_path)
|
||||
self.assertEqual(lock["issue_number"], 521)
|
||||
|
||||
res = gitea_create_pr(
|
||||
title="fix: restore lock tool registration Closes #521",
|
||||
head="feat/issue-521-lock-issue-registration",
|
||||
remote="prgs",
|
||||
worktree_path=worktree,
|
||||
)
|
||||
self.assertEqual(res["number"], 521)
|
||||
@@ -58,6 +58,12 @@ _NO_BLOCKER_FEEDBACK = {
|
||||
}
|
||||
|
||||
|
||||
def _init_reviewer_session(remote="prgs"):
|
||||
"""Seed review decision lock and required workflow-load proof (#389)."""
|
||||
init_review_decision_lock(remote, "review_pr")
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
|
||||
|
||||
def _mark_request_changes_ready(pr_number=8, **kwargs):
|
||||
"""Mark a request_changes decision ready with the #332 duplicate-
|
||||
suppression feedback fetch stubbed to 'no existing blocker'."""
|
||||
@@ -179,6 +185,7 @@ def _seed_ready_review_decision(
|
||||
"correction_authorized": False,
|
||||
"correction_reason": None,
|
||||
})
|
||||
_m.gitea_load_review_workflow()
|
||||
|
||||
|
||||
# Issue-write tools are profile-gated (#69).
|
||||
@@ -671,6 +678,7 @@ class TestMergePR(unittest.TestCase):
|
||||
def setUp(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
self._lease_patch = _install_owned_reviewer_lease(8)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
@@ -1876,7 +1884,7 @@ class TestReviewDecisionValidationGate(unittest.TestCase):
|
||||
def setUp(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
_init_reviewer_session("prgs")
|
||||
self._lease_patch = _install_owned_reviewer_lease(
|
||||
self.PR, head_sha=self.SHA,
|
||||
)
|
||||
@@ -1996,7 +2004,7 @@ class TestSubmitPrReview(unittest.TestCase):
|
||||
def setUp(self):
|
||||
import reviewer_pr_lease
|
||||
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
_init_reviewer_session("prgs")
|
||||
self._lease_patch = _install_owned_reviewer_lease(8)
|
||||
self._lease_patch.start()
|
||||
self._auth_identity_patch = patch(
|
||||
@@ -2416,7 +2424,7 @@ class TestSubmitPrReview(unittest.TestCase):
|
||||
os.remove(spoof_path)
|
||||
|
||||
def test_mark_final_decision_rejects_remote_mismatch(self):
|
||||
init_review_decision_lock("prgs", "review_pr")
|
||||
_init_reviewer_session("prgs")
|
||||
r = gitea_mark_final_review_decision(8, "approve", remote="dadeschools", expected_head_sha="abc123")
|
||||
self.assertFalse(r["marked_ready"])
|
||||
self.assertTrue(any("does not match locked remote" in x for x in r["reasons"]))
|
||||
@@ -2500,6 +2508,7 @@ if __name__ == "__main__":
|
||||
class TestTrackerHygieneCleanup(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_auth = patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start()
|
||||
patch("gitea_audit.audit_enabled", return_value=True).start()
|
||||
@@ -3879,7 +3888,7 @@ class TestPreflightVerification(unittest.TestCase):
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = " M reviewer_edit.py\n"
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity()
|
||||
self.assertIn("Reviewer profile is forbidden from modifying tracked workspace files", str(ctx.exception))
|
||||
self.assertIn("forbidden from modifying tracked workspace files", str(ctx.exception))
|
||||
self.assertIn("reviewer_edit.py", str(ctx.exception))
|
||||
|
||||
# Foreign pre-existing dirty state does not block when unchanged.
|
||||
@@ -3945,7 +3954,8 @@ class TestPreflightVerification(unittest.TestCase):
|
||||
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)
|
||||
self.assertIn("resolved workspace", msg)
|
||||
self.assertIn(worktree, msg)
|
||||
self.assertIn("worktree_path argument", msg)
|
||||
self.assertIn("task_file.py", msg)
|
||||
self.assertIn("author namespace", msg)
|
||||
|
||||
@@ -0,0 +1,240 @@
|
||||
"""Tests for namespace-scoped MCP workspace binding (#510)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_mcp_server as srv # noqa: E402
|
||||
import namespace_workspace_binding as nwb # noqa: E402
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parent.parent
|
||||
if REPO_ROOT.parent.name == "branches":
|
||||
CONTROL_ROOT = str(REPO_ROOT.parent.parent)
|
||||
else:
|
||||
CONTROL_ROOT = str(REPO_ROOT)
|
||||
|
||||
AUTHOR_DIRTY = f"{CONTROL_ROOT}/branches/mcp-author-worktree"
|
||||
MERGER_CLEAN = f"{CONTROL_ROOT}/branches/merge-pr487-submit"
|
||||
REVIEWER_CLEAN = f"{CONTROL_ROOT}/branches/review-pr487-submit"
|
||||
RECONCILER_CLEAN = f"{CONTROL_ROOT}/branches/reconcile-pr487"
|
||||
MCP_PROCESS_ROOT = CONTROL_ROOT
|
||||
|
||||
|
||||
class TestNamespaceWorkspaceModule(unittest.TestCase):
|
||||
def test_author_env_ignored_for_merger_namespace(self):
|
||||
workspace, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="merger",
|
||||
worktree_path=None,
|
||||
process_project_root=MCP_PROCESS_ROOT,
|
||||
env={
|
||||
nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
|
||||
nwb.MERGER_WORKTREE_ENV: MERGER_CLEAN,
|
||||
},
|
||||
profile_name="gitea-merger",
|
||||
)
|
||||
self.assertEqual(workspace, os.path.realpath(MERGER_CLEAN))
|
||||
self.assertEqual(source, f"{nwb.MERGER_WORKTREE_ENV} environment variable")
|
||||
|
||||
def test_author_env_ignored_for_reviewer_namespace(self):
|
||||
workspace, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reviewer",
|
||||
worktree_path=None,
|
||||
process_project_root=MCP_PROCESS_ROOT,
|
||||
env={
|
||||
nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
|
||||
nwb.REVIEWER_WORKTREE_ENV: REVIEWER_CLEAN,
|
||||
},
|
||||
)
|
||||
self.assertEqual(workspace, os.path.realpath(REVIEWER_CLEAN))
|
||||
self.assertEqual(source, f"{nwb.REVIEWER_WORKTREE_ENV} environment variable")
|
||||
|
||||
def test_author_env_ignored_for_reconciler_namespace(self):
|
||||
workspace, source = nwb.resolve_namespace_workspace(
|
||||
role_kind="reconciler",
|
||||
worktree_path=None,
|
||||
process_project_root=MCP_PROCESS_ROOT,
|
||||
env={
|
||||
nwb.AUTHOR_WORKTREE_ENV: AUTHOR_DIRTY,
|
||||
nwb.RECONCILER_WORKTREE_ENV: RECONCILER_CLEAN,
|
||||
},
|
||||
)
|
||||
self.assertEqual(workspace, os.path.realpath(RECONCILER_CLEAN))
|
||||
self.assertEqual(source, f"{nwb.RECONCILER_WORKTREE_ENV} environment variable")
|
||||
|
||||
def test_merger_profile_maps_to_merger_namespace(self):
|
||||
role = nwb.normalize_role_kind("reviewer", profile_name="gitea-merger")
|
||||
self.assertEqual(role, "merger")
|
||||
|
||||
def test_error_message_includes_path_and_binding_source(self):
|
||||
msg = nwb.format_namespace_workspace_binding_error(
|
||||
role_kind="merger",
|
||||
workspace_path=AUTHOR_DIRTY,
|
||||
binding_source=f"{nwb.AUTHOR_WORKTREE_ENV} environment variable",
|
||||
dirty_files=["gitea_mcp_server.py"],
|
||||
ignored_bindings=[f"{nwb.AUTHOR_WORKTREE_ENV}={AUTHOR_DIRTY} (ignored for merger namespace)"],
|
||||
)
|
||||
self.assertIn(AUTHOR_DIRTY, msg)
|
||||
self.assertIn("via", msg.lower())
|
||||
self.assertIn(nwb.AUTHOR_WORKTREE_ENV, msg)
|
||||
self.assertIn("Do not clean or reset foreign role worktrees", msg)
|
||||
|
||||
|
||||
class TestNamespaceWorkspaceIntegration(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._saved = {
|
||||
"whoami_called": srv._preflight_whoami_called,
|
||||
"capability_called": srv._preflight_capability_called,
|
||||
"resolved_role": srv._preflight_resolved_role,
|
||||
"whoami_violation": srv._preflight_whoami_violation,
|
||||
"capability_violation": srv._preflight_capability_violation,
|
||||
"in_test": srv._preflight_in_test_mode,
|
||||
}
|
||||
srv._preflight_whoami_called = True
|
||||
srv._preflight_capability_called = True
|
||||
srv._preflight_whoami_violation = False
|
||||
srv._preflight_capability_violation = False
|
||||
srv._preflight_in_test_mode = lambda: False
|
||||
self._env_patch = mock.patch.dict(os.environ, {"GITEA_TEST_PORCELAIN": ""}, clear=False)
|
||||
self._env_patch.start()
|
||||
|
||||
def tearDown(self):
|
||||
srv._preflight_whoami_called = self._saved["whoami_called"]
|
||||
srv._preflight_capability_called = self._saved["capability_called"]
|
||||
srv._preflight_resolved_role = self._saved["resolved_role"]
|
||||
srv._preflight_whoami_violation = self._saved["whoami_violation"]
|
||||
srv._preflight_capability_violation = self._saved["capability_violation"]
|
||||
srv._preflight_in_test_mode = self._saved["in_test"]
|
||||
self._env_patch.stop()
|
||||
for key in (
|
||||
nwb.AUTHOR_WORKTREE_ENV,
|
||||
nwb.MERGER_WORKTREE_ENV,
|
||||
nwb.REVIEWER_WORKTREE_ENV,
|
||||
nwb.RECONCILER_WORKTREE_ENV,
|
||||
nwb.ACTIVE_WORKTREE_ENV,
|
||||
):
|
||||
os.environ.pop(key, None)
|
||||
|
||||
def _merger_profile(self):
|
||||
return {
|
||||
"profile_name": "gitea-merger",
|
||||
"allowed_operations": ["gitea.pr.merge", "gitea.read"],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||
}
|
||||
|
||||
def _reviewer_profile(self):
|
||||
return {
|
||||
"profile_name": "prgs-reviewer",
|
||||
"allowed_operations": ["gitea.pr.approve", "gitea.pr.review", "gitea.read"],
|
||||
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
|
||||
}
|
||||
|
||||
def _reconciler_profile(self):
|
||||
return {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"allowed_operations": ["gitea.pr.close", "gitea.read"],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.create",
|
||||
],
|
||||
}
|
||||
|
||||
def _author_profile(self):
|
||||
return {
|
||||
"profile_name": "prgs-author",
|
||||
"allowed_operations": ["gitea.pr.create", "gitea.branch.push", "gitea.read"],
|
||||
"forbidden_operations": ["gitea.pr.merge", "gitea.pr.approve"],
|
||||
}
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_author_worktree_does_not_block_merger_with_clean_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
||||
srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_author_worktree_does_not_block_reviewer_with_clean_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
os.environ[nwb.REVIEWER_WORKTREE_ENV] = REVIEWER_CLEAN
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._reviewer_profile()):
|
||||
srv.verify_preflight_purity("prgs", worktree_path=REVIEWER_CLEAN)
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_author_worktree_does_not_block_reconciler_with_clean_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
os.environ[nwb.RECONCILER_WORKTREE_ENV] = RECONCILER_CLEAN
|
||||
srv._preflight_resolved_role = "reconciler"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._reconciler_profile()):
|
||||
srv.verify_preflight_purity("prgs", worktree_path=RECONCILER_CLEAN)
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_dirty_active_task_workspace_still_blocks_mutations(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
dirty = " M namespace_workspace_binding.py\n"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
||||
with mock.patch("gitea_mcp_server._get_workspace_porcelain", return_value=dirty):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.verify_preflight_purity("prgs", worktree_path=MERGER_CLEAN)
|
||||
self.assertIn(MERGER_CLEAN, str(ctx.exception))
|
||||
self.assertIn("binding", str(ctx.exception).lower())
|
||||
|
||||
def test_root_workspace_mutation_still_blocked_for_author(self):
|
||||
srv._preflight_resolved_role = "author"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", CONTROL_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._author_profile()):
|
||||
with mock.patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master"},
|
||||
):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.verify_preflight_purity("prgs")
|
||||
self.assertIn("stable control checkout", str(ctx.exception))
|
||||
|
||||
@mock.patch("subprocess.run")
|
||||
@mock.patch("os.path.isdir", return_value=True)
|
||||
@mock.patch("os.path.exists", return_value=True)
|
||||
def test_pr487_style_merge_binds_clean_merger_workspace(
|
||||
self, _exists, _isdir, mock_run
|
||||
):
|
||||
mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
|
||||
os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
|
||||
srv._preflight_resolved_role = "reviewer"
|
||||
with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
|
||||
with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
|
||||
resolved = srv._verify_role_mutation_workspace("prgs")
|
||||
self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))
|
||||
@@ -99,6 +99,7 @@ class PermissionReportBase(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
self._remotes.stop()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
mcp_server.review_workflow_load.clear_review_workflow_load()
|
||||
gitea_config._active_profile_override = None
|
||||
self._dir.cleanup()
|
||||
|
||||
@@ -250,6 +251,8 @@ class TestEligibilityDenialReport(PermissionReportBase):
|
||||
return {"login": "author-user"}
|
||||
return PR_PAYLOAD
|
||||
mock_api.side_effect = fake_api
|
||||
mcp_server.init_review_decision_lock("prgs", "review_pr")
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
with patch.dict(os.environ, self._env("author-profile")):
|
||||
res = mcp_server.gitea_merge_pr(
|
||||
pr_number=42, confirmation="MERGE PR 42",
|
||||
|
||||
@@ -0,0 +1,257 @@
|
||||
"""Post-merge moot reviewer-lease handling (#515).
|
||||
|
||||
Covers:
|
||||
a. Reviewer-lease acquisition/adoption is refused on an already-merged/closed
|
||||
PR (fail closed, no mutation).
|
||||
b. The post-merge moot cleanup path is safe and idempotent.
|
||||
c. An active foreign lease on an *open* PR is never force-cleaned.
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from datetime import datetime, timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import reviewer_pr_lease as leases # noqa: E402
|
||||
from mcp_server import ( # noqa: E402
|
||||
gitea_acquire_reviewer_pr_lease,
|
||||
gitea_cleanup_post_merge_moot_lease,
|
||||
)
|
||||
|
||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||
MERGER_ENV = {
|
||||
"GITEA_PROFILE_NAME": "prgs-merger",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
|
||||
}
|
||||
PR = 487
|
||||
ISSUE = 485
|
||||
SESSION = "97274-676d20a825c4"
|
||||
|
||||
|
||||
def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
|
||||
candidate_head="a" * 40):
|
||||
body = leases.format_lease_body(
|
||||
repo="Scaled-Tech-Consulting/Gitea-Tools",
|
||||
pr_number=pr_number,
|
||||
issue_number=ISSUE,
|
||||
reviewer_identity="sysadmin",
|
||||
profile="prgs-reviewer",
|
||||
session_id=session_id,
|
||||
worktree="branches/review-pr487",
|
||||
phase=phase,
|
||||
candidate_head=candidate_head,
|
||||
target_branch="master",
|
||||
target_branch_sha="b" * 40,
|
||||
last_activity=datetime.now(timezone.utc),
|
||||
)
|
||||
return {"id": 6603, "body": body, "user": {"login": "sysadmin"}}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Pure logic
|
||||
# --------------------------------------------------------------------------- #
|
||||
class TestAcquireRefusedOnMergedPR(unittest.TestCase):
|
||||
def test_acquire_refused_when_pr_merged_or_closed(self):
|
||||
result = leases.assess_acquire_lease(
|
||||
[_lease_comment()],
|
||||
pr_number=PR,
|
||||
reviewer_identity="sysadmin",
|
||||
profile="prgs-merger",
|
||||
session_id="new-session",
|
||||
repo="Scaled-Tech-Consulting/Gitea-Tools",
|
||||
issue_number=ISSUE,
|
||||
worktree="branches/merge-pr487",
|
||||
candidate_head="a" * 40,
|
||||
target_branch="master",
|
||||
target_branch_sha="b" * 40,
|
||||
pr_merged_or_closed=True,
|
||||
)
|
||||
self.assertFalse(result["acquire_allowed"])
|
||||
self.assertTrue(result["post_merge_moot"])
|
||||
self.assertIsNone(result["lease_body"])
|
||||
self.assertTrue(any("post_merge_moot" in r for r in result["reasons"]))
|
||||
|
||||
def test_acquire_still_allowed_on_open_pr_without_flag(self):
|
||||
result = leases.assess_acquire_lease(
|
||||
[],
|
||||
pr_number=PR,
|
||||
reviewer_identity="sysadmin",
|
||||
profile="prgs-reviewer",
|
||||
session_id="s1",
|
||||
repo="Scaled-Tech-Consulting/Gitea-Tools",
|
||||
issue_number=ISSUE,
|
||||
worktree="branches/review-pr487",
|
||||
candidate_head="a" * 40,
|
||||
target_branch="master",
|
||||
target_branch_sha="b" * 40,
|
||||
)
|
||||
self.assertTrue(result["acquire_allowed"])
|
||||
self.assertFalse(result["post_merge_moot"])
|
||||
|
||||
|
||||
class TestPostMergeMootAssessment(unittest.TestCase):
|
||||
def test_merged_pr_with_active_lease_is_moot_and_cleanable(self):
|
||||
a = leases.assess_post_merge_moot_lease(
|
||||
[_lease_comment()],
|
||||
pr_number=PR,
|
||||
pr_merged=True,
|
||||
pr_state="closed",
|
||||
merge_commit_sha="c" * 40,
|
||||
)
|
||||
self.assertTrue(a["pr_merged_or_closed"])
|
||||
self.assertTrue(a["is_moot"])
|
||||
self.assertTrue(a["cleanup_allowed"])
|
||||
self.assertIsNotNone(a["release_body"])
|
||||
self.assertIn("phase: released", a["release_body"])
|
||||
self.assertIn("blocker: post-merge-moot", a["release_body"])
|
||||
|
||||
def test_open_pr_active_lease_never_cleaned(self):
|
||||
a = leases.assess_post_merge_moot_lease(
|
||||
[_lease_comment()],
|
||||
pr_number=PR,
|
||||
pr_merged=False,
|
||||
pr_state="open",
|
||||
)
|
||||
self.assertFalse(a["pr_merged_or_closed"])
|
||||
self.assertFalse(a["is_moot"])
|
||||
self.assertFalse(a["cleanup_allowed"])
|
||||
self.assertIsNone(a["release_body"])
|
||||
self.assertTrue(any("still open" in r for r in a["reasons"]))
|
||||
|
||||
def test_merged_pr_without_lease_nothing_to_clean(self):
|
||||
a = leases.assess_post_merge_moot_lease(
|
||||
[], pr_number=PR, pr_merged=True, pr_state="closed")
|
||||
self.assertTrue(a["pr_merged_or_closed"])
|
||||
self.assertFalse(a["is_moot"])
|
||||
self.assertFalse(a["cleanup_allowed"])
|
||||
self.assertTrue(any("nothing to clean" in r for r in a["reasons"]))
|
||||
|
||||
def test_cleanup_is_idempotent(self):
|
||||
"""After the released marker is posted, a re-assess finds nothing to clean."""
|
||||
first = leases.assess_post_merge_moot_lease(
|
||||
[_lease_comment()], pr_number=PR, pr_merged=True, pr_state="closed")
|
||||
self.assertTrue(first["cleanup_allowed"])
|
||||
released_comment = {
|
||||
"id": 7000, "body": first["release_body"], "user": {"login": "sysadmin"}}
|
||||
second = leases.assess_post_merge_moot_lease(
|
||||
[_lease_comment(), released_comment],
|
||||
pr_number=PR, pr_merged=True, pr_state="closed")
|
||||
self.assertFalse(second["is_moot"])
|
||||
self.assertFalse(second["cleanup_allowed"])
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# Server tools
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
|
||||
"""Build an api_request side effect keyed on method + url."""
|
||||
calls = {"post": []}
|
||||
|
||||
def _side(method, url, auth=None, payload=None, *a, **k):
|
||||
m = (method or "").upper()
|
||||
if m == "POST":
|
||||
calls["post"].append({"url": url, "payload": payload})
|
||||
return {"id": posted_id}
|
||||
if "/comments" in url:
|
||||
return list(comments)
|
||||
if "/pulls/" in url:
|
||||
pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40}
|
||||
if pr_merged:
|
||||
pr["merged"] = True
|
||||
pr["merged_at"] = "2026-07-08T07:46:04Z"
|
||||
return pr
|
||||
if "/issues/" in url:
|
||||
return {"state": "closed" if pr_merged else "open", "number": ISSUE}
|
||||
return {}
|
||||
|
||||
return _side, calls
|
||||
|
||||
|
||||
class TestAcquireToolRefusesMergedPR(unittest.TestCase):
|
||||
def setUp(self):
|
||||
leases.clear_session_lease()
|
||||
|
||||
@patch("mcp_server.verify_preflight_purity", return_value=None)
|
||||
@patch("mcp_server._authenticated_username", return_value="sysadmin")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("mcp_server.api_request")
|
||||
def test_acquire_tool_fails_closed_on_merged_pr_without_posting(
|
||||
self, mock_api, _auth, _user, _purity):
|
||||
side, calls = _api_side_effect(
|
||||
pr_state="closed", pr_merged=True, comments=[])
|
||||
mock_api.side_effect = side
|
||||
with patch.dict(os.environ, MERGER_ENV, clear=True):
|
||||
result = gitea_acquire_reviewer_pr_lease(
|
||||
pr_number=PR,
|
||||
worktree="branches/merge-pr487",
|
||||
candidate_head="a" * 40,
|
||||
issue_number=ISSUE,
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertFalse(result["acquired"])
|
||||
self.assertTrue(result.get("post_merge_moot"))
|
||||
self.assertEqual(calls["post"], [], "must not post a lease comment")
|
||||
|
||||
|
||||
class TestCleanupTool(unittest.TestCase):
|
||||
def setUp(self):
|
||||
leases.clear_session_lease()
|
||||
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("mcp_server.api_request")
|
||||
def test_read_only_reports_moot_without_mutating(self, mock_api, _auth):
|
||||
side, calls = _api_side_effect(
|
||||
pr_state="closed", pr_merged=True, comments=[_lease_comment()])
|
||||
mock_api.side_effect = side
|
||||
with patch.dict(os.environ, MERGER_ENV, clear=True):
|
||||
result = gitea_cleanup_post_merge_moot_lease(
|
||||
pr_number=PR, apply=False, remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertTrue(result["pr_merged_or_closed"])
|
||||
self.assertTrue(result["lease_moot"])
|
||||
self.assertFalse(result["cleanup_performed"])
|
||||
self.assertTrue(result["no_merge_or_adoption"])
|
||||
self.assertEqual(result["mode"], "read_only")
|
||||
self.assertEqual(calls["post"], [])
|
||||
|
||||
@patch("mcp_server.verify_preflight_purity", return_value=None)
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("mcp_server.api_request")
|
||||
def test_apply_posts_released_marker_on_merged_pr(
|
||||
self, mock_api, _auth, _purity):
|
||||
side, calls = _api_side_effect(
|
||||
pr_state="closed", pr_merged=True, comments=[_lease_comment()])
|
||||
mock_api.side_effect = side
|
||||
with patch.dict(os.environ, MERGER_ENV, clear=True):
|
||||
result = gitea_cleanup_post_merge_moot_lease(
|
||||
pr_number=PR, apply=True, remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertTrue(result["cleanup_performed"])
|
||||
self.assertEqual(result["released_comment_id"], 9999)
|
||||
self.assertEqual(len(calls["post"]), 1)
|
||||
self.assertIn("phase: released", calls["post"][0]["payload"]["body"])
|
||||
self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"])
|
||||
|
||||
@patch("mcp_server.verify_preflight_purity", return_value=None)
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
@patch("mcp_server.api_request")
|
||||
def test_apply_refuses_to_clean_open_pr(self, mock_api, _auth, _purity):
|
||||
side, calls = _api_side_effect(
|
||||
pr_state="open", pr_merged=False, comments=[_lease_comment()])
|
||||
mock_api.side_effect = side
|
||||
with patch.dict(os.environ, MERGER_ENV, clear=True):
|
||||
result = gitea_cleanup_post_merge_moot_lease(
|
||||
pr_number=PR, apply=True, remote="prgs")
|
||||
self.assertFalse(result["cleanup_performed"])
|
||||
self.assertFalse(result["pr_merged_or_closed"])
|
||||
self.assertEqual(calls["post"], [], "never force-clean an open PR lease")
|
||||
self.assertTrue(
|
||||
any("still open" in r for r in result.get("cleanup_skipped_reason", [])))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -11,6 +11,7 @@ from mcp_server import (
|
||||
gitea_view_pr,
|
||||
gitea_review_pr,
|
||||
gitea_check_pr_eligibility,
|
||||
gitea_load_review_workflow,
|
||||
)
|
||||
import gitea_config
|
||||
|
||||
|
||||
@@ -0,0 +1,153 @@
|
||||
"""Tests for canonical review workflow load proof (#389)."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import review_workflow_load
|
||||
import mcp_server
|
||||
|
||||
|
||||
class TestReviewWorkflowLoadModule(unittest.TestCase):
|
||||
def setUp(self):
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
|
||||
def test_load_records_hash_and_schema(self):
|
||||
root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
|
||||
recorded = review_workflow_load.record_review_workflow_load(root)
|
||||
self.assertEqual(
|
||||
recorded["workflow_source"],
|
||||
review_workflow_load.WORKFLOW_REL_PATH,
|
||||
)
|
||||
self.assertEqual(recorded["task_mode"], "review-merge-pr")
|
||||
self.assertRegex(recorded["workflow_hash"], r"^[0-9a-f]{12}$")
|
||||
self.assertEqual(
|
||||
recorded["final_report_schema_path"],
|
||||
review_workflow_load.SCHEMA_REL_PATH,
|
||||
)
|
||||
status = review_workflow_load.workflow_load_status(root)
|
||||
self.assertTrue(status["workflow_load_proof_present"])
|
||||
self.assertTrue(status["workflow_load_valid"])
|
||||
|
||||
def test_stale_session_pid_blocks(self):
|
||||
root = str(__import__("pathlib").Path(__file__).resolve().parent.parent)
|
||||
review_workflow_load.record_review_workflow_load(root)
|
||||
review_workflow_load._REVIEW_WORKFLOW_LOAD["session_pid"] = 0
|
||||
blockers = review_workflow_load.review_workflow_load_blockers(root)
|
||||
self.assertTrue(any("different process" in b for b in blockers))
|
||||
|
||||
def test_prompt_conflict_detected(self):
|
||||
conflict, reasons = review_workflow_load.assess_prompt_conflict(
|
||||
"Run work-issue author implementation only")
|
||||
self.assertTrue(conflict)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
|
||||
class TestReviewWorkflowLoadGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
review_workflow_load.clear_review_workflow_load()
|
||||
mcp_server.init_review_decision_lock("prgs", "review_pr")
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", "reviewer")
|
||||
|
||||
def _load_workflow(self):
|
||||
return mcp_server.gitea_load_review_workflow()
|
||||
|
||||
def test_mcp_helper_returns_required_fields(self):
|
||||
res = self._load_workflow()
|
||||
self.assertTrue(res["success"])
|
||||
self.assertTrue(res["loaded"])
|
||||
self.assertIn("workflow_source", res)
|
||||
self.assertIn("workflow_hash", res)
|
||||
self.assertIn("final_report_schema_path", res)
|
||||
self.assertIn("final_report_schema_hash", res)
|
||||
|
||||
def test_mark_final_blocked_without_load(self):
|
||||
res = mcp_server.gitea_mark_final_review_decision(
|
||||
42, "approve", remote="prgs")
|
||||
self.assertFalse(res["marked_ready"])
|
||||
self.assertTrue(any(
|
||||
"gitea_load_review_workflow" in r for r in res["reasons"]))
|
||||
self.assertTrue(any(
|
||||
"approve/merge replay" in r.lower() or "Do not call" in r
|
||||
for r in res["reasons"]))
|
||||
|
||||
def test_submit_review_blocked_without_load(self):
|
||||
with patch("mcp_server.gitea_check_pr_eligibility") as elig:
|
||||
elig.return_value = {
|
||||
"eligible": True,
|
||||
"authenticated_user": "rev",
|
||||
"profile_name": "prgs-reviewer",
|
||||
"pr_author": "author",
|
||||
"head_sha": "abc123",
|
||||
"reasons": [],
|
||||
}
|
||||
res = mcp_server.gitea_submit_pr_review(
|
||||
42,
|
||||
"approve",
|
||||
remote="prgs",
|
||||
final_review_decision_ready=True,
|
||||
)
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertTrue(any(
|
||||
"gitea_load_review_workflow" in r for r in res["reasons"]))
|
||||
|
||||
def test_merge_blocked_without_load(self):
|
||||
res = mcp_server.gitea_merge_pr(
|
||||
42,
|
||||
confirmation="MERGE PR 42",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(res["performed"])
|
||||
self.assertTrue(any(
|
||||
"gitea_load_review_workflow" in r for r in res["reasons"]))
|
||||
|
||||
def test_resolve_capability_reports_missing_load(self):
|
||||
with patch.object(mcp_server, "_ensure_matching_profile"):
|
||||
with patch.object(
|
||||
mcp_server.gitea_config, "is_runtime_switching_enabled",
|
||||
return_value=False):
|
||||
with patch.object(
|
||||
mcp_server, "_authenticated_username",
|
||||
return_value="rev"):
|
||||
res = mcp_server.gitea_resolve_task_capability(
|
||||
"review_pr", remote="prgs")
|
||||
proof = res.get("workflow_load_proof") or {}
|
||||
self.assertFalse(proof.get("workflow_load_valid"))
|
||||
self.assertTrue(any(
|
||||
"gitea_load_review_workflow" in g
|
||||
for g in res.get("task_role_guidance") or []))
|
||||
|
||||
def test_init_review_lock_clears_prior_load(self):
|
||||
self._load_workflow()
|
||||
mcp_server.init_review_decision_lock("prgs", "review_pr")
|
||||
blockers = review_workflow_load.review_workflow_load_blockers(
|
||||
str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
self.assertTrue(blockers)
|
||||
|
||||
def test_dry_run_allowed_without_load(self):
|
||||
with patch("mcp_server.get_auth_header", return_value="Basic dGVzdA=="), \
|
||||
patch("mcp_server._list_pr_lease_comments", return_value=[]), \
|
||||
patch("mcp_server.gitea_check_pr_eligibility") as elig:
|
||||
elig.return_value = {
|
||||
"eligible": True,
|
||||
"authenticated_user": "rev",
|
||||
"profile_name": "prgs-reviewer",
|
||||
"pr_author": "author",
|
||||
"head_sha": "abc123",
|
||||
"reasons": [],
|
||||
}
|
||||
res = mcp_server.gitea_dry_run_pr_review(
|
||||
42, "approve", remote="prgs")
|
||||
self.assertNotIn(
|
||||
"gitea_load_review_workflow",
|
||||
" ".join(res.get("reasons") or []),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -37,6 +37,7 @@ def _lock(mutations=None, correction=False):
|
||||
|
||||
def _seed(mutations=None, correction=False):
|
||||
mcp_server._save_review_decision_lock(_lock(mutations, correction))
|
||||
mcp_server.gitea_load_review_workflow()
|
||||
|
||||
|
||||
APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
|
||||
@@ -48,6 +49,7 @@ RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
|
||||
class TestTerminalHardStopReasons(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
mcp_server.review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def test_no_lock_no_reasons(self):
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
@@ -93,6 +95,7 @@ class TestTerminalHardStopReasons(unittest.TestCase):
|
||||
class TestMergeHardStopWiring(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
mcp_server.review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def test_merge_blocked_after_request_changes(self):
|
||||
_seed([RC_A])
|
||||
@@ -114,6 +117,7 @@ class TestMergeHardStopWiring(unittest.TestCase):
|
||||
class TestMarkFinalHardStopWiring(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
mcp_server.review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def test_mark_ready_blocked_after_terminal_mutation(self):
|
||||
_seed([RC_A])
|
||||
@@ -146,6 +150,7 @@ def _mark(action, pr_number=6, **kwargs):
|
||||
class TestDuplicateRequestChangesSuppression(unittest.TestCase):
|
||||
def tearDown(self):
|
||||
mcp_server._save_review_decision_lock(None)
|
||||
mcp_server.review_workflow_load.clear_review_workflow_load()
|
||||
|
||||
def test_duplicate_request_changes_blocked_at_same_head(self):
|
||||
_seed()
|
||||
|
||||
Reference in New Issue
Block a user