fix: resolve conflicts for PR #554

Merge prgs/master into PR branch to restore mergeability.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
2026-07-08 22:28:13 -04:00
co-authored by Claude Opus 4.8
25 changed files with 2361 additions and 20 deletions
+19
View File
@@ -13,6 +13,25 @@ credentials.** Every test mocks the HTTP client and the keychain/auth lookup.
## 1. Standard test commands
### Canonical runner: `./run-tests.sh`
The canonical full-validation command is the root-level runner. It invokes the
project virtualenv interpreter and passes any extra arguments straight through
to `pytest`:
```bash
# Full validation
./run-tests.sh
# Focused validation (extra args forward to pytest)
./run-tests.sh tests/test_mcp_server.py -q
```
`run-tests.sh` runs `venv/bin/python -m pytest "$@"` and fails with a clear
setup message if the virtualenv Python is missing (so a session never silently
falls back to the wrong interpreter). The explicit `venv/bin/python -m pytest`
forms below remain valid and equivalent.
The test suite needs the project virtualenv (it provides the MCP SDK):
```bash
+67
View File
@@ -0,0 +1,67 @@
# MCP operator shell menu
## Purpose
`./mcp-menu.sh` is a repository-root terminal menu for onboarding and operating
the Gitea-Tools MCP/Gitea workflow without memorizing every prompt, script path,
or runbook section.
It is intentionally **safe by default**: status checks and copy-paste workflow
prompts. It does not delete branches, force-push, edit lock files, or bypass
sanctioned MCP tools.
## How to run
From the repository root:
```bash
./mcp-menu.sh
```
The script must be executable (`chmod +x mcp-menu.sh`). It uses bash with
`set -euo pipefail`.
## Safety rules
- **Read-only by default** — root checkout health is inspection only.
- **No destructive git** — no `git push --force`, branch deletion, or
`--delete` refspecs.
- **No lock-file editing** — issue locks are acquired only through
`gitea_lock_issue`.
- **No raw API bypass** — prompts direct operators to sanctioned MCP tools.
- **Remote mutations require confirmation** — any future menu action that would
mutate remote or server state must be clearly labeled and require explicit
operator confirmation before running.
- **Author work stays under `branches/`** — the root checkout is a stable
control checkout on `master` / `prgs/master`.
## Menu options
| Option | Description |
|--------|-------------|
| Project status / root checkout health | Shows cwd, branch, `git status --short --branch`, HEAD SHA, `prgs/master` SHA, and warnings when the root checkout is dirty or off `master`. |
| Author workflow prompts | Ready-to-copy prompts for issue work, conflict-fix sessions, and root checkout recovery. |
| Reviewer workflow prompts | PR review prompt (review-only; no merge). |
| Merger workflow prompts | PR merge prompt (merge gates and explicit approval). |
| Reconciler workflow prompts | Already-landed / closed PR reconciliation prompt. |
| Onboarding new project | Checklist prompt for adding a repository to the MCP workflow. |
| Proxmox deployment placeholder | **Not implemented** — informational message only. |
| Create Proxmox LXC placeholder | **Not implemented** — informational message only. |
| Run tests | Runs `./run-tests.sh` when present; otherwise `venv/bin/python -m pytest`; otherwise fails closed with a clear error. |
| Exit | Quit the menu. |
## Placeholder-only entries
**Proxmox deployment** and **Create Proxmox LXC** are placeholders until
dedicated issues implement sanctioned automation. The menu prints a clear
message and does not invoke deploy scripts.
## Related documentation
- [`docs/llm-workflow-runbooks.md`](llm-workflow-runbooks.md) — Gitea-specific workflow runbooks
- [`skills/llm-project-workflow/SKILL.md`](../skills/llm-project-workflow/SKILL.md) — portable workflow skill
- [`skills/llm-project-workflow/workflows/`](../skills/llm-project-workflow/workflows/) — canonical task workflows
## Tests
Hermetic coverage lives in `tests/test_mcp_menu_script.py`.
+17
View File
@@ -1145,6 +1145,22 @@ def _rule_reviewer_review_mutation(
)
def _rule_reviewer_mutation_capability_proof(report_text: str) -> list[dict[str, str]]:
from reviewer_mutation_capability_proof import assess_mutation_capability_proof
result = assess_mutation_capability_proof(report_text)
if not result.get("block"):
return []
return _findings_from_reasons(
"reviewer.mutation_capability_proof",
result.get("reasons") or [],
field="Capabilities proven",
severity="block",
safe_next_action=result.get("safe_next_action")
or "document exact per-mutation capability proof before each mutation",
)
def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
result = assess_post_merge_cleanup_proof(report_text)
if not result.get("block"):
@@ -1207,6 +1223,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
_rule_reviewer_workflow_load_boundary,
_rule_reviewer_mutation_ledger,
_rule_reviewer_review_mutation,
_rule_reviewer_mutation_capability_proof,
_rule_reviewer_post_merge_cleanup_proof,
*_SHARED_CLEANUP_PROOF_RULES,
_rule_reviewer_stale_head_proof,
+311 -13
View File
@@ -211,6 +211,28 @@ def _effective_workspace_role() -> str:
)
def _actual_profile_role() -> str:
"""Resolve the workspace role from the ACTIVE PROFILE alone (#540).
Unlike :func:`_effective_workspace_role`, this never consults
``_preflight_resolved_role``. A task capability whose ``required_role_kind``
is ``author`` (e.g. ``comment_issue``) stamps ``_preflight_resolved_role =
"author"``; the #274/#475 role exemptions must key off the real profile
identity so that stamp cannot poison a genuine reviewer/merger/reconciler
session into being treated as an author. Author profiles still classify as
``author`` here, so author blocking is preserved.
"""
profile = get_profile()
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()
@@ -541,9 +563,19 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
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).
The exemption honours BOTH the effective workspace role and the actual
profile role (#540). ``comment_issue`` preflight stamps
``_preflight_resolved_role = "author"`` (its ``required_role_kind``), which
would otherwise poison :func:`_effective_workspace_role` into classifying a
genuine reconciler as an author and defeat this exemption. Keying off the
real profile role as well preserves the exemption without weakening author
blocking an actual author profile classifies as ``author`` in both.
"""
role = _effective_workspace_role()
if role in nwb.NON_AUTHOR_ROLES:
if (
_effective_workspace_role() in nwb.NON_AUTHOR_ROLES
or _actual_profile_role() in nwb.NON_AUTHOR_ROLES
):
return
ctx = _resolve_namespace_mutation_context(worktree_path)
workspace = ctx["workspace_path"]
@@ -710,6 +742,7 @@ def _enforce_root_checkout_guard(worktree_path: str | None = None) -> None:
porcelain_status=git_state.get("porcelain_status") or "",
remote_master_sha=remote_master_sha,
resolved_role=_preflight_resolved_role,
actual_role=_actual_profile_role(),
)
if assessment["block"]:
raise RuntimeError(root_checkout_guard.format_root_checkout_guard_error(assessment))
@@ -748,6 +781,7 @@ import merge_approval_gate # noqa: E402
import already_landed_reconcile # noqa: E402
import author_mutation_worktree # noqa: E402
import root_checkout_guard # noqa: E402
import remote_repo_guard # noqa: E402
import issue_claim_heartbeat # noqa: E402
import issue_work_duplicate_gate # noqa: E402
import reviewer_pr_lease # noqa: E402
@@ -1162,11 +1196,49 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
if remote not in REMOTES:
raise ValueError(f"Unknown remote '{remote}'. Choose from: {list(REMOTES)}")
profile = REMOTES[remote]
return (
host or profile["host"],
org or profile["org"],
repo or profile["repo"],
resolved_host = host or profile["host"]
resolved_org = org or profile["org"]
resolved_repo = repo or profile["repo"]
_enforce_remote_repo_guard(
remote,
resolved_org,
resolved_repo,
org_explicit=org is not None,
repo_explicit=repo is not None,
)
return (resolved_host, resolved_org, resolved_repo)
def _enforce_remote_repo_guard(
remote: str,
resolved_org: str,
resolved_repo: str,
*,
org_explicit: bool,
repo_explicit: bool,
) -> None:
"""Fail closed on a remote/repo mismatch vs. the local git remote (#530).
Best-effort: bypassed under pytest unless ``GITEA_FORCE_REMOTE_REPO_CHECK`` is
set, so the unit suite (which calls tools with bare remotes against mocked APIs)
is unaffected. In production it protects every read/lookup/mutation tool because
they all resolve targets through :func:`_resolve`.
"""
if "pytest" in sys.modules and not os.environ.get(
"GITEA_FORCE_REMOTE_REPO_CHECK"
):
return
local_remote_url = _local_git_remote_url(remote)
assessment = remote_repo_guard.assess_remote_repo_match(
remote=remote,
resolved_org=resolved_org,
resolved_repo=resolved_repo,
local_remote_url=local_remote_url,
org_explicit=org_explicit,
repo_explicit=repo_explicit,
)
if assessment["block"]:
raise RuntimeError(remote_repo_guard.format_remote_repo_guard_error(assessment))
def _auth(host: str) -> str:
@@ -2409,10 +2481,18 @@ def _review_decision_session_reasons(lock: dict | None) -> list[str]:
return reasons
def init_review_decision_lock(remote: str | None, task: str | None):
def init_review_decision_lock(remote: str | None, task: str | None, force: bool = True):
"""Seed read-only-until-ready state for reviewer PR review tasks."""
if task != "review_pr":
return
if not force:
lock = _load_review_decision_lock()
if lock is not None:
if lock.get("remote") == remote and lock.get("session_pid") == os.getpid():
env_lock = (os.environ.get(SESSION_PROFILE_LOCK_ENV) or "").strip()
stored_lock = (lock.get("session_profile_lock") or "").strip()
if not env_lock or not stored_lock or env_lock == stored_lock:
return
review_workflow_load.clear_review_workflow_load()
profile = get_profile()
profile_name = (profile.get("profile_name") or "").strip()
@@ -2805,6 +2885,85 @@ def gitea_get_pr_review_feedback(
}
def _fetch_pr_lease_comments_safe(
pr_number: int,
*,
remote: str,
host: str | None,
org: str | None,
repo: str | None,
limit: int = 100,
require_open: bool = False,
) -> dict:
"""Fetch PR thread comments with structured fail-closed errors (#519)."""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
resolved_repo = f"{o}/{r}"
pr_url = f"{repo_api_url(h, o, r)}/pulls/{pr_number}"
try:
pr = api_request("GET", pr_url, auth)
except RuntimeError as exc:
return {
"success": False,
"comments": [],
"reasons": [
"PR lookup failed before conflict-fix push assessment "
f"(pr_number={pr_number}, repo={resolved_repo}, remote={remote}): "
f"{_redact(str(exc))}"
],
"pr_lookup": "failed",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
}
pr_state = (pr.get("state") or "").strip().lower()
if require_open and pr_state != "open":
return {
"success": False,
"comments": [],
"reasons": [
f"PR #{pr_number} on {resolved_repo} is not open "
f"(state={pr_state or 'unknown'})"
],
"pr_lookup": "not_open",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
"head_sha": (pr.get("head") or {}).get("sha"),
}
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
try:
comments = api_request("GET", api, auth)
except RuntimeError as exc:
return {
"success": False,
"comments": [],
"reasons": [
"PR comment fetch failed during conflict-fix push assessment "
f"(pr_number={pr_number}, issue_index={pr_number}, "
f"repo={resolved_repo}, remote={remote}): "
f"{_redact(str(exc))}"
],
"pr_lookup": "ok",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
"head_sha": (pr.get("head") or {}).get("sha"),
}
if not isinstance(comments, list):
comments = []
return {
"success": True,
"comments": list(comments[:limit]),
"reasons": [],
"pr_lookup": "ok",
"resolved_repo": resolved_repo,
"remote": remote,
"pr_number": pr_number,
"head_sha": (pr.get("head") or {}).get("sha"),
}
def _list_pr_lease_comments(
pr_number: int,
*,
@@ -2814,7 +2973,15 @@ def _list_pr_lease_comments(
repo: str | None,
limit: int = 100,
) -> list[dict]:
"""Fetch PR/issue thread comments used for reviewer/conflict-fix leases."""
"""Fetch PR/issue thread comments used for reviewer/conflict-fix leases.
Intentionally does **not** pre-fetch the PR via GET /pulls/{n}. Shared
callers (merge approval feedback, reviewer lease gates) depend on a single
comments GET so mock sequences and fail-open #485 non-list handling stay
stable. Structured PR-lookup failures belong only to
:func:`_fetch_pr_lease_comments_safe` used by conflict-fix push assessment
(#519).
"""
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
@@ -5591,7 +5758,6 @@ def gitea_adopt_merger_pr_lease(
_verify_role_mutation_workspace(
remote, worktree=worktree, task="adopt_merger_pr_lease"
)
verify_preflight_purity(remote, task="adopt_merger_pr_lease")
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
profile = get_profile()
@@ -5939,6 +6105,118 @@ def gitea_cleanup_post_merge_moot_lease(
return report
@mcp.tool()
def gitea_release_reviewer_pr_lease(
pr_number: int,
worktree: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> dict:
"""Safely release an active reviewer lease owned by the current session on an open PR.
This provides a canonical way to clean up reviewer leases when a review fails
before submission.
Args:
pr_number: The PR number whose reviewer lease to release.
worktree: Optional worktree path (resolves automatically if not supplied).
remote: Known instance 'dadeschools' or 'prgs'.
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
Returns:
dict reporting release status.
"""
_verify_role_mutation_workspace(
remote, worktree=worktree, task="review_pr"
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
comments = _fetch_pr_comments(
pr_number, remote=remote, host=host, org=org, repo=repo
)
active = reviewer_pr_lease.find_active_reviewer_lease(
comments, pr_number=pr_number
)
if not active:
return {
"success": True,
"released": False,
"reasons": ["no active reviewer lease found on PR"],
}
identity = _authenticated_username(remote) or ""
session = reviewer_pr_lease.get_session_lease() or {}
session_id = session.get("session_id")
owner_identity = active.get("reviewer_identity")
owner_session_id = active.get("session_id")
# Authorize release: identity matches, session_id matches, or lease is expired/reclaimable
authorized = False
reasons = []
if owner_identity and identity and owner_identity == identity:
authorized = True
elif owner_session_id and session_id and owner_session_id == session_id:
authorized = True
else:
freshness = reviewer_pr_lease.classify_lease_freshness(active)
if freshness in {"expired", "reclaimable"}:
authorized = True
if not authorized:
return {
"success": False,
"released": False,
"reasons": [
f"unauthorized to release active lease: owned by {owner_identity} "
f"(session {owner_session_id})"
],
}
# Format and post release comment
body = reviewer_pr_lease.format_lease_body(
repo=active.get("repo") or f"{o}/{r}",
pr_number=pr_number,
issue_number=active.get("issue_number"),
reviewer_identity=owner_identity or identity,
profile=active.get("profile") or "reviewer",
session_id=owner_session_id or session_id or "unknown",
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"),
blocker="manual-release",
)
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": "release_reviewer_pr_lease"},
):
posted = api_request("POST", comment_url, auth, {"body": body})
reviewer_pr_lease.clear_session_lease()
return {
"success": True,
"released": True,
"comment_id": posted.get("id"),
"reasons": [],
}
@mcp.tool()
def gitea_list_issue_comments(
issue_number: int,
@@ -6010,6 +6288,7 @@ def gitea_create_issue_comment(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""Post a markdown comment to a Gitea issue's discussion thread.
@@ -6031,6 +6310,7 @@ def gitea_create_issue_comment(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
worktree_path: Optional path to verify branches-only guard.
Returns:
dict with 'success', 'comment_id', and 'issue_number' ('url' only
@@ -6039,7 +6319,7 @@ def gitea_create_issue_comment(
(permission blocks also carry a structured 'permission_report',
#142).
"""
verify_preflight_purity(remote, task="comment_issue")
verify_preflight_purity(remote, worktree_path=worktree_path, task="comment_issue")
gate_reasons = _profile_operation_gate("gitea.issue.comment")
reasons = list(gate_reasons)
if not (body or "").strip():
@@ -7838,22 +8118,39 @@ def gitea_assess_conflict_fix_push(
"reasons": read_block,
"permission_report": _permission_block_report("gitea.read"),
}
comments = _list_pr_lease_comments(
fetched = _fetch_pr_lease_comments_safe(
pr_number,
remote=remote,
host=host,
org=org,
repo=repo,
require_open=True,
)
return pr_work_lease.assess_conflict_fix_push(
if not fetched["success"]:
return {
"push_allowed": False,
"block": True,
"reasons": fetched["reasons"],
"pr_lookup": fetched.get("pr_lookup"),
"resolved_repo": fetched.get("resolved_repo"),
"remote": fetched.get("remote"),
"pr_number": pr_number,
"assessment_failed": True,
}
assessment = pr_work_lease.assess_conflict_fix_push(
pr_number=pr_number,
comments=comments,
comments=fetched["comments"],
branch_head_before=branch_head_before,
branch_head_after=branch_head_after,
worktree_path=worktree_path,
push_cwd=push_cwd,
is_fast_forward=is_fast_forward,
)
assessment["pr_lookup"] = fetched.get("pr_lookup")
assessment["resolved_repo"] = fetched.get("resolved_repo")
assessment["remote"] = fetched.get("remote")
assessment["live_pr_head_sha"] = fetched.get("head_sha")
return assessment
@mcp.tool()
@@ -8491,6 +8788,7 @@ def gitea_resolve_task_capability(
init_review_decision_lock(
remote if remote in REMOTES else None,
task,
force=False,
)
record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
Executable
+232
View File
@@ -0,0 +1,232 @@
#!/usr/bin/env bash
# mcp-menu.sh — Repository-root operator menu for MCP/Gitea workflow onboarding.
#
# Safe by default: read-only status and copy-paste prompts unless an action is
# explicitly labeled and confirmed. No branch deletion, force-push, lock-file
# editing, or raw API bypass.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$SCRIPT_DIR"
pause() {
read -r -p "Press Enter to return to the menu..."
}
print_banner() {
printf '\n=== Gitea-Tools MCP Operator Menu ===\n'
printf 'Repository: %s\n' "$REPO_ROOT"
printf 'Safe by default — destructive actions require explicit confirmation.\n\n'
}
show_root_checkout_health() {
printf '\n--- Project status / root checkout health ---\n\n'
printf 'Current directory: %s\n' "$(pwd)"
local branch head_sha prgs_master_sha dirty
branch="$(git -C "$REPO_ROOT" branch --show-current 2>/dev/null || true)"
if [[ -z "$branch" ]]; then
branch="(detached HEAD)"
fi
printf 'Current branch: %s\n' "$branch"
printf '\nGit status (short, branch):\n'
git -C "$REPO_ROOT" status --short --branch || true
head_sha="$(git -C "$REPO_ROOT" rev-parse HEAD 2>/dev/null || echo 'unknown')"
printf '\nHEAD SHA: %s\n' "$head_sha"
if git -C "$REPO_ROOT" rev-parse --verify prgs/master >/dev/null 2>&1; then
prgs_master_sha="$(git -C "$REPO_ROOT" rev-parse prgs/master)"
printf 'prgs/master SHA: %s\n' "$prgs_master_sha"
if [[ "$head_sha" != "$prgs_master_sha" ]]; then
printf '\nWARNING: root checkout HEAD does not match prgs/master.\n'
printf 'Keep the stable control checkout on master/prgs/master.\n'
fi
else
printf 'prgs/master SHA: unavailable (remote ref not fetched)\n'
fi
if [[ -n "$(git -C "$REPO_ROOT" status --porcelain 2>/dev/null || true)" ]]; then
printf '\nWARNING: root checkout has uncommitted changes (dirty).\n'
printf 'Author mutations belong in a session worktree under branches/.\n'
fi
if [[ "$branch" != "master" && "$branch" != "main" && "$branch" != "dev" ]]; then
printf '\nWARNING: root checkout is not on a stable base branch (master/main/dev).\n'
printf 'Return to master before using the control checkout.\n'
fi
pause
}
print_prompt_block() {
local title="$1"
local body="$2"
printf '\n--- %s ---\n\n' "$title"
printf '%s\n' "$body"
printf '\n(Copy the prompt above into your LLM session.)\n'
pause
}
show_author_prompts() {
while true; do
printf '\n--- Author workflow prompts ---\n'
printf ' 1) Work issue (author/coder)\n'
printf ' 2) Conflict-fix author session\n'
printf ' 3) Root checkout recovery session\n'
printf ' 0) Back\n'
read -r -p 'Choice: ' choice
case "$choice" in
1)
print_prompt_block "Author — work issue" \
"You are the AUTHOR session for <org>/<repo>.
Goal: implement issue #<N> only.
Workflow:
1. Preflight: prove identity, work_issue/create_pr capability, clean session worktree under branches/.
2. gitea_lock_issue for issue #<N> and branch feat/issue-<N>-<short-desc>.
3. Implement in the locked worktree only — never mutate the root control checkout.
4. Validate, commit, push, gitea_create_pr. Final report with issue, branch, SHA, PR, tests, mutation ledger."
;;
2)
print_prompt_block "Author — conflict-fix session" \
"You are the AUTHOR session for <org>/<repo> in conflict-fix mode.
Goal: resolve merge conflicts on PR #<P> / branch <branch> only.
Workflow:
1. Preflight: prove author identity and exact push/commit capability for the locked PR branch.
2. Confirm conflict-fix lease and stale-head protection before pushing.
3. Work only in the session-owned worktree under branches/ — never the root checkout.
4. Rebase or merge target branch, run tests, push, update PR. No force-push without explicit operator approval.
5. Final report: conflict resolution proof, new HEAD SHA, tests, mutation ledger."
;;
3)
print_prompt_block "Author — root checkout recovery" \
"You are a RECOVERY session for <org>/<repo>.
Goal: restore the stable root control checkout to clean master/prgs/master.
Workflow:
1. Inspect root checkout: branch, git status --short --branch, HEAD vs prgs/master.
2. Do not implement features from the root checkout. Stash or move work to branches/<session> first.
3. Return root to master (or main/dev per project policy) matching prgs/master with no dirty tracked files.
4. Report before/after branch, SHA, dirty state, and safe next action for author worktree creation."
;;
0) return ;;
*) printf 'Invalid choice.\n' ;;
esac
done
}
show_reviewer_prompts() {
print_prompt_block "Reviewer — PR review" \
"You are the REVIEWER session for <org>/<repo>.
Goal: review PR #<P> only — do not merge unless explicitly switched to merger mode.
Workflow:
1. Load canonical workflow: skills/llm-project-workflow/workflows/review-merge-pr.md
2. Preflight: prove reviewer identity, review_pr capability, clean review worktree.
3. gitea_view_pr, validate scope, run required checks in the correct worktree.
4. gitea_review_pr with approve, request-changes, or comment as warranted.
5. Final report: PR head SHA, verdict, validation evidence, mutation ledger. No merge in reviewer-only runs."
}
show_merger_prompts() {
print_prompt_block "Merger — PR merge" \
"You are the MERGER session for <org>/<repo>.
Goal: merge PR #<P> only after every gate passes.
Workflow:
1. Load canonical workflow: skills/llm-project-workflow/workflows/review-merge-pr.md
2. Preflight: prove merger identity and exact merge_pr capability for the current PR head SHA.
3. Confirm approval pins the current head SHA; re-validate if the branch moved.
4. gitea_merge_pr only on explicit operator approval after all gates pass.
5. Final report: merged SHA, cleanup handoff, mutation ledger."
}
show_reconciler_prompts() {
print_prompt_block "Reconciler — already-landed / closed PR cleanup" \
"You are the RECONCILER session for <org>/<repo>.
Goal: reconcile already-landed open PRs — close or comment only when exact capability is proven.
Workflow:
1. Load canonical workflow: skills/llm-project-workflow/workflows/reconcile-landed-pr.md
2. Preflight: prove reconciler identity and gitea.pr.close (or authorized close) capability.
3. Do not review, merge, implement code, or create branches.
4. gitea_scan_already_landed_open_prs / gitea_reconcile_already_landed_pr as appropriate.
5. Final report: PR numbers handled, close proof, mutation ledger."
}
show_onboarding_prompt() {
print_prompt_block "Onboarding — new project to MCP workflow" \
"Onboard <org>/<repo> into the MCP Control Plane workflow.
Checklist:
1. Prove identity and task capability via gitea_whoami and gitea_resolve_task_capability.
2. Configure separate MCP namespaces/profiles: author, reviewer, merger/reconciler as needed.
3. Register gitea-tools (and jenkins-mcp / glitchtip-mcp if applicable) in the client MCP config.
4. Copy skills/llm-project-workflow/SKILL.md guidance into the target repo or ECC install.
5. Verify gitea_get_runtime_context, gitea_lock_issue, and worktree rules under branches/.
6. Run ./mcp-menu.sh for day-to-day prompts; use docs/mcp-menu.md and docs/llm-workflow-runbooks.md.
Canonical router: skills/llm-project-workflow/SKILL.md"
}
show_proxmox_placeholder() {
printf '\n--- Proxmox deployment (placeholder) ---\n\n'
printf 'Push this project to Proxmox — TODO / issue-backed\n'
printf 'Create Proxmox LXC — TODO / issue-backed\n\n'
printf 'These actions are NOT implemented yet.\n'
printf 'Track deployment automation in dedicated Gitea issues before enabling here.\n'
printf 'This menu will not run deploy scripts until sanctioned tooling exists.\n'
pause
}
run_tests() {
printf '\n--- Run tests ---\n\n'
if [[ -x "$REPO_ROOT/run-tests.sh" ]]; then
printf 'Running ./run-tests.sh ...\n\n'
(cd "$REPO_ROOT" && ./run-tests.sh)
pause
return
fi
if [[ -x "$REPO_ROOT/venv/bin/python" ]]; then
printf 'run-tests.sh not found; falling back to venv/bin/python -m pytest\n\n'
(cd "$REPO_ROOT" && ./venv/bin/python -m pytest)
pause
return
fi
printf 'ERROR: No test runner available (fail closed).\n' >&2
printf 'Expected ./run-tests.sh or venv/bin/python for pytest fallback.\n' >&2
exit 1
}
main_menu() {
while true; do
print_banner
printf ' 1) Project status / root checkout health\n'
printf ' 2) Author workflow prompts\n'
printf ' 3) Reviewer workflow prompts\n'
printf ' 4) Merger workflow prompts\n'
printf ' 5) Reconciler workflow prompts\n'
printf ' 6) Onboarding new project to this MCP workflow\n'
printf ' 7) Proxmox deployment menu placeholder\n'
printf ' 8) Create Proxmox LXC placeholder\n'
printf ' 9) Run tests\n'
printf ' 0) Exit\n'
read -r -p 'Choice: ' choice
case "$choice" in
1) show_root_checkout_health ;;
2) show_author_prompts ;;
3) show_reviewer_prompts ;;
4) show_merger_prompts ;;
5) show_reconciler_prompts ;;
6) show_onboarding_prompt ;;
7|8) show_proxmox_placeholder ;;
9) run_tests ;;
0) printf 'Goodbye.\n'; exit 0 ;;
*) printf 'Invalid choice.\n'; pause ;;
esac
done
}
main_menu
+98
View File
@@ -0,0 +1,98 @@
"""Remote/repo mismatch guard (#530).
Bare ``remote`` names resolve to a default ``org``/``repo`` via the ``REMOTES``
table in :mod:`gitea_auth`. For the ``prgs`` instance the hardcoded default repo
is ``Timesheet``, but the tools in this project operate on
``Scaled-Tech-Consulting/Gitea-Tools``. When a session runs inside a Gitea-Tools
worktree and calls a tool with a bare ``remote=prgs`` (no explicit ``org``/``repo``),
the resolved target silently points at the wrong repository, producing false 404s
and risking mutation of a different repo.
This module provides a pure assessment that compares the MCP-resolved ``org/repo``
against the local git remote URL and fails closed on a genuine mismatch, unless the
caller supplied explicit ``org``/``repo`` (in which case their intent is authoritative)
or the local remote URL is unavailable (best-effort corroboration only).
"""
from __future__ import annotations
REMEDIATION = (
"Pass explicit org= and repo= matching the local git remote, "
"e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools."
)
def assess_remote_repo_match(
*,
remote: str,
resolved_org: str,
resolved_repo: str,
local_remote_url: str | None,
org_explicit: bool,
repo_explicit: bool,
) -> dict:
"""Fail closed when the resolved org/repo disagrees with the local git remote.
The guard is intentionally conservative:
* When the caller passed both ``org`` and ``repo`` explicitly, their intent is
authoritative and the guard never blocks.
* When the local git remote URL is unavailable (``None``/empty), corroboration
is impossible, so the guard does not block (best-effort only).
* Otherwise, the resolved ``org/repo`` slug must appear in the local remote URL
(case-insensitive); if it does not, the guard blocks.
"""
reasons: list[str] = []
if org_explicit and repo_explicit:
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
url = (local_remote_url or "").strip()
if not url:
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
expected_slug = f"{resolved_org}/{resolved_repo}".lower()
if expected_slug in url.lower():
return _assessment(True, reasons, remote, resolved_org, resolved_repo, local_remote_url)
reasons.append(
f"MCP-resolved repository '{resolved_org}/{resolved_repo}' for remote "
f"'{remote}' does not match the local git remote URL '{url}'"
)
return _assessment(False, reasons, remote, resolved_org, resolved_repo, local_remote_url)
def format_remote_repo_guard_error(assessment: dict) -> str:
"""Single RuntimeError message for the MCP resolver gate."""
reasons = "; ".join(
assessment.get("reasons") or ["remote/repo resolution mismatch"]
)
resolved = (
f"{assessment.get('resolved_org')}/{assessment.get('resolved_repo')}"
)
local = assessment.get("local_remote_url") or "(unknown)"
return (
f"Remote/repo guard (#530): {reasons}. "
f"Resolved target: {resolved}; local git remote: {local}. "
f"{REMEDIATION}"
)
def _assessment(
proven: bool,
reasons: list[str],
remote: str,
resolved_org: str,
resolved_repo: str,
local_remote_url: str | None,
) -> dict:
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"remote": remote,
"resolved_org": resolved_org,
"resolved_repo": resolved_repo,
"local_remote_url": local_remote_url,
"remediation": REMEDIATION,
}
+9
View File
@@ -5617,3 +5617,12 @@ def assess_proof_backed_handoff_report(report_text, **kwargs):
from reviewer_proof_backed_handoff import assess_proof_backed_handoff_report as _assess
return _assess(report_text, **kwargs)
def assess_mutation_capability_proof(report_text, **kwargs):
"""#405: exact per-mutation capability proof in reviewer final reports."""
from reviewer_mutation_capability_proof import (
assess_mutation_capability_proof as _assess,
)
return _assess(report_text, **kwargs)
+129
View File
@@ -0,0 +1,129 @@
"""Exact per-mutation capability proof verifier for reviewer reports (#405).
A reviewer final report may prove ``review_pr`` capability and then also merge
a PR or delete a remote branch. Merge and branch deletion are separate
mutations that require their own exact capability proof a nearby capability
must never authorize a different operation. This verifier requires a
mutation-capability table pairing every performed mutation with the exact
task/permission resolved *before* that mutation.
"""
from __future__ import annotations
import re
# Mutations this verifier tracks, with the exact capability tokens that
# authorize each. A row for the mutation must cite one of its own tokens;
# tokens from a different mutation (a "nearby capability") never count.
_REVIEW_TOKENS = ("review_pr", "gitea.pr.review", "gitea.pr.approve",
"gitea.pr.request_changes", "request_changes_pr", "approve_pr")
_MERGE_TOKENS = ("merge_pr", "gitea.pr.merge")
_DELETE_TOKENS = ("delete_branch", "gitea.branch.delete")
# Detect that a mutation was actually performed (not merely mentioned as a
# non-goal or skipped).
_MERGE_PERFORMED = re.compile(
r"(?:gitea_merge_pr\b(?![^\n]*\b(?:not called|skipped|blocked)\b)|"
r"^\s*[-*]?\s*merge result\s*:\s*merged\b|"
r"\bpr merged\b|\bmerge commit\s*(?:sha)?\s*[:=]?\s*[0-9a-f]{7,})",
re.IGNORECASE | re.MULTILINE,
)
_DELETE_PERFORMED = re.compile(
r"(?:gitea_delete_branch\b(?![^\n]*\b(?:not called|skipped|blocked)\b)|"
r"^\s*[-*]?\s*(?:remote )?branch deleted\s*:|"
r"\bdeleted (?:the )?(?:remote )?branch\b|"
r"^\s*[-*]?\s*branch deletion\s*:\s*(?!skipped|none|not)\S)",
re.IGNORECASE | re.MULTILINE,
)
_REVIEW_PERFORMED = re.compile(
r"(?:gitea_submit_pr_review\b|gitea_mark_final_review_decision\b|"
r"^\s*[-*]?\s*review (?:decision|verdict|mutation)\s*:\s*"
r"(?:approved|request[_ ]changes)\b|\breview submitted\b)",
re.IGNORECASE | re.MULTILINE,
)
# Post-hoc proof: capability resolved *after* the mutation is never valid.
_POST_HOC = re.compile(
r"capabilit(?:y|ies)\s+(?:resolved|proven|checked)\s+(?:after|post[- ])\s*"
r"(?:the\s+)?(?:merge|deletion|delete|mutation|review)",
re.IGNORECASE,
)
# The report must carry an explicit mutation-capability table.
_TABLE_MARKER = re.compile(
r"mutation[- ]capability(?:\s+table)?|capability[- ]per[- ]mutation",
re.IGNORECASE,
)
def _tokens_present(text: str, tokens: tuple[str, ...]) -> bool:
low = text.lower()
return any(tok.lower() in low for tok in tokens)
def assess_mutation_capability_proof(report_text: str) -> dict:
"""Validate exact per-mutation capability proof in a reviewer report.
Returns ``{proven, block, reasons, safe_next_action}``. A report that
performs no mutation beyond an ordinary review passes only when its
review capability is cited; merge/delete each demand their own exact
capability row. Fail closed on nearby-capability substitution, a
missing table, missing rows, or post-hoc proof.
"""
text = report_text or ""
reasons: list[str] = []
merged = bool(_MERGE_PERFORMED.search(text))
deleted = bool(_DELETE_PERFORMED.search(text))
reviewed = bool(_REVIEW_PERFORMED.search(text))
extra_mutation = merged or deleted
if _POST_HOC.search(text):
reasons.append(
"capability proof recorded after the mutation; exact capability "
"must be resolved before each mutation"
)
# A review-only report needs its review capability cited; no table required.
if reviewed and not _tokens_present(text, _REVIEW_TOKENS):
reasons.append(
"review mutation performed without exact review capability proof "
"(review_pr / gitea.pr.review)"
)
if extra_mutation and not _TABLE_MARKER.search(text):
reasons.append(
"mutation beyond review performed without a mutation-capability "
"table (mutation, exact task/capability, result, order-before)"
)
if merged:
if not _tokens_present(text, _MERGE_TOKENS):
reasons.append(
"merge performed without exact merge capability proof "
"(merge_pr / gitea.pr.merge); nearby review_pr does not "
"authorize merge"
)
if deleted:
if not _tokens_present(text, _DELETE_TOKENS):
reasons.append(
"branch deletion performed without exact delete capability "
"proof (delete_branch / gitea.branch.delete); nearby "
"merge_pr does not authorize branch deletion"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"safe_next_action": (
"proceed"
if proven
else "add a mutation-capability table with the exact resolved "
"task/permission and pre-mutation order for every mutation; "
"skip any mutation whose exact capability is unproven"
),
}
+17 -2
View File
@@ -58,15 +58,30 @@ def assess_root_checkout_guard(
porcelain_status: str,
remote_master_sha: str | None,
resolved_role: str | None = None,
actual_role: str | None = None,
) -> dict:
"""Fail closed when the control checkout is not clean master/prgs/master."""
"""Fail closed when the control checkout is not clean master/prgs/master.
``resolved_role`` is the preflight-resolved *task* role and ``actual_role``
is the *active profile* role (#540). The reconciler exemption honours either
signal so a ``comment_issue`` preflight (which stamps the task role as
``author``) cannot strip a genuine reconciler of its exemption. An actual
author profile classifies as ``author`` in both signals, so author blocking
on a contaminated control checkout is preserved.
Merger *strictness* (a merger must not be auto-exempted by working from a
``branches/`` worktree) stays keyed on the resolved task role: merge
operations resolve their own task role, and widening the merger test with
``actual_role`` would wrongly subject a merger operating from its clean
workspace under a non-merge task to full control-checkout checks.
"""
reasons: list[str] = []
root = os.path.realpath(canonical_repo_root)
workspace = os.path.realpath(workspace_path)
branch = (current_branch or "").strip()
dirty_files = parse_dirty_tracked_files(porcelain_status)
if resolved_role == "reconciler":
if resolved_role == "reconciler" or actual_role == "reconciler":
return _assessment(True, [], root, workspace, branch, head_sha, dirty_files)
if resolved_role != "merger" and is_path_under_branches(workspace, root):
Executable
+13
View File
@@ -0,0 +1,13 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PYTHON="$ROOT_DIR/venv/bin/python"
if [[ ! -x "$PYTHON" ]]; then
echo "ERROR: expected virtualenv Python at $PYTHON" >&2
echo "Create the venv first, then run: venv/bin/python -m pytest" >&2
exit 1
fi
exec "$PYTHON" -m pytest "$@"
@@ -11,6 +11,10 @@ Final report schema: `schemas/review-merge-final-report.md`.
Rules (llm-project-workflow):
- **BLOCKED + DIAGNOSE default (required):** If any required step (load workflow, lease, profile/role, capability, worktree under branches/, preflight, tool, instruction, etc.) cannot be performed, STOP. State BLOCKED. Use [`blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template exactly. Only safe non-mutating recovery. Report. Do not continue or fallback.
- Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo
and is blocked when it disagrees with the local git remote URL.
- Only an eligible, NON-author reviewer merges. If authenticated user == PR
author → STOP.
- Do not merge unless the PR is open, mergeable, and its checks/review pass.
@@ -36,6 +36,10 @@ Final report schema: `schemas/review-merge-final-report.md`.
Rules (llm-project-workflow):
- **BLOCKED + DIAGNOSE default (required):** If any required step (load workflow, lease, profile/role, capability, worktree under branches/, preflight, tool, instruction, etc.) cannot be performed, STOP. State BLOCKED. Use [`blocked-diagnose-report.md`](../templates/blocked-diagnose-report.md) template exactly. Only safe non-mutating recovery. Report. Do not continue or fallback.
- Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo
and is blocked when it disagrees with the local git remote URL.
- Review in a SEPARATE detached review worktree, never the author's folder.
- Worktree safety (#233): before checkout, diff, validation, review, or merge,
report the starting worktree path and whether it was dirty. If unrelated
@@ -16,6 +16,10 @@ Rules (llm-project-workflow):
- Work only in an isolated branch worktree under branches/. The main checkout
is orchestration/status only.
- Do not self-review or self-merge.
- Repository targeting (#530): pass explicit `remote=`, `org=`, and `repo=` on
every gitea-tools call (e.g. `remote=prgs org=Scaled-Tech-Consulting
repo=Gitea-Tools`). A bare `remote=prgs` can resolve to the wrong default repo
and is blocked when it disagrees with the local git remote URL.
Steps:
0. Work Selection Rule — before any claim, branch, or file edits, acquire or
@@ -1099,6 +1099,26 @@ Use precise wording:
Do not collapse review, merge, cleanup, or external-state mutations into vague wording.
## 31B. Mutation-capability table (#405)
Every performed mutation requires exact capability proof resolved **before** that
mutation executes. Nearby capabilities never authorize a different operation —
`review_pr` does not authorize `merge_pr`, and `merge_pr` does not authorize
`delete_branch` / `gitea.branch.delete`.
When any mutation beyond a bare review occurs (merge, branch delete, issue
close/comment, etc.), the final report must include a **mutation-capability table**
with one row per performed mutation:
* mutation (tool/action name)
* exact task/capability resolved (for example `merge_pr` / `gitea.pr.merge`)
* result
* order/timestamp proof that capability was resolved before the mutation
If exact capability proof is missing, skip the mutation or stop the workflow —
never claim a performed mutation without its row. Post-hoc capability proof after
the mutation fails validation.
## 31A. Local artifact and report consistency rule
Do not create local walkthrough, notes, markdown, JSON, or report artifacts during reviewer runs unless the canonical workflow or operator explicitly requires it.
@@ -628,9 +628,14 @@ When pushing to an existing PR branch to resolve merge conflicts:
* session worktree path
* push cwd
* whether the push is fast-forward
3. Do not push when a reviewer holds an active lease on the same PR.
4. Do not force-push.
5. Do not push from the main checkout or wrong cwd.
* explicit `remote`, `org`, and `repo` when not using defaults
3. If assessment returns `assessment_failed: true` or `pr_lookup: failed`, stop
and produce a recovery handoff with the structured `reasons` and
`resolved_repo` fields — do not treat an MCP HTTP 500 as proof the push was
unsafe or safe (#519).
4. Do not push when a reviewer holds an active lease on the same PR.
5. Do not force-push.
6. Do not push from the main checkout or wrong cwd.
Conflict-fix final reports must state:
+139
View File
@@ -0,0 +1,139 @@
"""Regression tests for gitea_assess_conflict_fix_push structured failures (#519)."""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from mcp_server import ( # noqa: E402
_fetch_pr_lease_comments_safe,
gitea_assess_conflict_fix_push,
)
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
AUTHOR_ENV = {
"GITEA_PROFILE_NAME": "prgs-author",
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.branch.push,gitea.pr.create",
}
HEAD_BEFORE = "dad1dc8d5108ab01ed83065334116d7425a4471c"
HEAD_AFTER = "3f3d6cb35d0fe225dcb236247f2a2d0ec193fa35"
OPEN_PR = {
"number": 508,
"state": "open",
"head": {"sha": HEAD_AFTER},
}
class TestFetchPrLeaseCommentsSafe(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_pr_lookup_404_returns_structured_failure(self, _auth, mock_api):
mock_api.side_effect = RuntimeError(
'HTTP 404: {"message":"issue does not exist","index":508}'
)
result = _fetch_pr_lease_comments_safe(
508, remote="prgs", host=None, org=None, repo=None)
self.assertFalse(result["success"])
self.assertEqual(result["comments"], [])
self.assertTrue(result["reasons"])
self.assertIn("PR lookup failed", result["reasons"][0])
self.assertEqual(result["pr_lookup"], "failed")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_comment_fetch_404_after_pr_lookup(self, _auth, mock_api):
def _api(method, url, auth, *args, **kwargs):
if "/pulls/" in url:
return OPEN_PR
raise RuntimeError(
'HTTP 404: {"message":"issue does not exist","index":508}'
)
mock_api.side_effect = _api
result = _fetch_pr_lease_comments_safe(
508, remote="prgs", host=None, org=None, repo=None)
self.assertFalse(result["success"])
self.assertIn("comment fetch failed", result["reasons"][0].lower())
self.assertEqual(result["pr_lookup"], "ok")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_success_returns_comments_and_head_sha(self, _auth, mock_api):
comment = {"id": 1, "body": "<!-- mcp-conflict-fix-lease:v1 -->"}
def _api(method, url, auth, *args, **kwargs):
if "/pulls/" in url:
return OPEN_PR
return [comment]
mock_api.side_effect = _api
result = _fetch_pr_lease_comments_safe(
508, remote="prgs", host=None, org=None, repo=None)
self.assertTrue(result["success"])
self.assertEqual(result["comments"], [comment])
self.assertEqual(result["head_sha"], OPEN_PR["head"]["sha"])
class TestAssessConflictFixPushTool(unittest.TestCase):
@patch("mcp_server._fetch_pr_lease_comments_safe")
@patch("mcp_server._profile_operation_gate", return_value=None)
def test_returns_structured_block_when_pr_lookup_fails(
self, _gate, mock_fetch,
):
mock_fetch.return_value = {
"success": False,
"comments": [],
"reasons": ["PR lookup failed"],
"pr_lookup": "failed",
"resolved_repo": "Scaled-Tech-Consulting/Gitea-Tools",
"remote": "prgs",
"pr_number": 508,
}
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
result = gitea_assess_conflict_fix_push(
pr_number=508,
branch_head_before=HEAD_BEFORE,
branch_head_after=HEAD_AFTER,
worktree_path="/proj/branches/fix-pr508",
push_cwd="/proj/branches/fix-pr508",
is_fast_forward=True,
remote="prgs",
)
self.assertFalse(result["push_allowed"])
self.assertTrue(result.get("assessment_failed"))
self.assertEqual(result["pr_lookup"], "failed")
@patch("mcp_server._fetch_pr_lease_comments_safe")
@patch("mcp_server._profile_operation_gate", return_value=None)
def test_valid_push_assessment_when_pr_and_comments_resolve(
self, _gate, mock_fetch,
):
mock_fetch.return_value = {
"success": True,
"comments": [],
"reasons": [],
"pr_lookup": "ok",
"resolved_repo": "Scaled-Tech-Consulting/Gitea-Tools",
"remote": "prgs",
"pr_number": 508,
"head_sha": HEAD_AFTER,
}
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
result = gitea_assess_conflict_fix_push(
pr_number=508,
branch_head_before=HEAD_BEFORE,
branch_head_after=HEAD_AFTER,
worktree_path="/proj/branches/fix-pr508",
push_cwd="/proj/branches/fix-pr508",
is_fast_forward=True,
remote="prgs",
)
self.assertTrue(result["push_allowed"])
self.assertEqual(result.get("live_pr_head_sha"), HEAD_AFTER)
if __name__ == "__main__":
unittest.main()
+271
View File
@@ -0,0 +1,271 @@
"""Regression tests for #540: comment_issue preflight must not poison the
actual reconciler role for #274 branch-only / #475 root-checkout exemptions.
`resolve_task_capability("comment_issue")` stamps
``_preflight_resolved_role = "author"`` because ``comment_issue`` has
``required_role_kind = author``. Before the fix, ``_effective_workspace_role``
preferred that stamp and a genuine ``prgs-reconciler`` session was treated as an
author inside the #274 branch-only mutation guard and the #475 root checkout
guard, blocking ``gitea_create_issue_comment`` from the control checkout.
The fix keys the role exemptions off the *actual profile role* as well, so the
exemption survives the poisoned task role while author profiles stay blocked.
"""
from __future__ import annotations
import os
import sys
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import gitea_mcp_server as srv # noqa: E402
import root_checkout_guard as rcg # noqa: E402
FAKE_AUTH = "token test"
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
MASTER_SHA = "a" * 40
OTHER_SHA = "b" * 40
RECONCILER_PROFILE = {
"profile_name": "prgs-reconciler",
"allowed_operations": [
"gitea.read",
"gitea.pr.close",
"gitea.pr.comment",
"gitea.issue.comment",
],
"forbidden_operations": [
"gitea.pr.approve",
"gitea.pr.merge",
"gitea.pr.create",
"gitea.branch.push",
"gitea.repo.commit",
],
"audit_label": "prgs-reconciler",
}
AUTHOR_PROFILE = {
"profile_name": "prgs-author",
"allowed_operations": [
"gitea.read",
"gitea.issue.comment",
"gitea.issue.create",
"gitea.pr.create",
"gitea.branch.push",
"gitea.repo.commit",
],
"forbidden_operations": [
"gitea.pr.approve",
"gitea.pr.merge",
],
"audit_label": "prgs-author",
}
REVIEWER_PROFILE = {
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.read", "gitea.pr.approve", "gitea.pr.merge"],
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
"audit_label": "prgs-reviewer",
}
class TestActualProfileRole(unittest.TestCase):
"""`_actual_profile_role` ignores the poisoned preflight task role (#540)."""
def tearDown(self):
srv._preflight_resolved_role = None
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
def test_reconciler_profile_role_survives_author_task_stamp(self, _profile):
srv._preflight_resolved_role = "author" # comment_issue poison
self.assertEqual(srv._actual_profile_role(), "reconciler")
# _effective_workspace_role is still poisoned to author (unchanged #510)...
self.assertEqual(srv._effective_workspace_role(), "author")
@patch("gitea_mcp_server.get_profile", return_value=AUTHOR_PROFILE)
def test_author_profile_role_is_author(self, _profile):
srv._preflight_resolved_role = "author"
self.assertEqual(srv._actual_profile_role(), "author")
@patch("gitea_mcp_server.get_profile", return_value=REVIEWER_PROFILE)
def test_reviewer_profile_role_is_reviewer(self, _profile):
srv._preflight_resolved_role = "author"
self.assertEqual(srv._actual_profile_role(), "reviewer")
class TestBranchesOnlyExemptionRealRole(unittest.TestCase):
"""#274 branches-only exemption keys off the actual profile role (#540)."""
def tearDown(self):
srv._preflight_resolved_role = None
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
def test_reconciler_exempt_despite_author_task_stamp(self, _profile):
srv._preflight_resolved_role = "author" # poison
# Must return without raising and without resolving an author worktree.
with patch("gitea_mcp_server._resolve_namespace_mutation_context") as ctx:
srv._enforce_branches_only_author_mutation()
ctx.assert_not_called()
@patch("gitea_mcp_server.get_profile", return_value=AUTHOR_PROFILE)
def test_author_not_exempt(self, _profile):
srv._preflight_resolved_role = "author"
sentinel = RuntimeError("author-mutation-guard-reached")
def _blow_up(*_a, **_k):
raise sentinel
# Author is not exempt: the guard proceeds to resolve/assess the
# workspace (proven by reaching the patched context resolver).
with patch(
"gitea_mcp_server._resolve_namespace_mutation_context",
side_effect=_blow_up,
):
with self.assertRaises(RuntimeError) as raised:
srv._enforce_branches_only_author_mutation()
self.assertIs(raised.exception, sentinel)
class TestRootCheckoutGuardRealRole(unittest.TestCase):
"""#475 root guard honours the actual profile role too (#540)."""
def _assess(self, **kwargs):
defaults = {
"workspace_path": CONTROL_CHECKOUT_ROOT,
"canonical_repo_root": CONTROL_CHECKOUT_ROOT,
"current_branch": "feat/some-branch",
"head_sha": OTHER_SHA,
"porcelain_status": " M gitea_mcp_server.py\n",
"remote_master_sha": MASTER_SHA,
"resolved_role": "author", # poisoned task role
}
defaults.update(kwargs)
return rcg.assess_root_checkout_guard(**defaults)
def test_actual_reconciler_exempt_despite_poisoned_resolved_author(self):
result = self._assess(actual_role="reconciler")
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_actual_author_still_blocked_on_contaminated_root(self):
result = self._assess(actual_role="author")
self.assertTrue(result["block"])
def test_merger_strictness_stays_keyed_on_resolved_task_role(self):
# When the merge task resolves the merger role, the branches/ auto
# exemption is denied and a clean control checkout is required.
blocked = self._assess(
workspace_path=f"{CONTROL_CHECKOUT_ROOT}/branches/review-pr-1",
current_branch="review-pr-1",
porcelain_status="",
head_sha=OTHER_SHA,
resolved_role="merger",
)
self.assertTrue(blocked["block"])
def test_actual_merger_does_not_over_tighten_non_merge_task(self):
# A merger profile whose current task did NOT resolve to merger keeps
# the branches/ workspace exemption (regression guard for #540): the
# actual_role signal must not force merger strictness here.
result = self._assess(
workspace_path=f"{CONTROL_CHECKOUT_ROOT}/branches/review-pr-1",
current_branch="review-pr-1",
porcelain_status="",
head_sha=OTHER_SHA,
resolved_role="reviewer",
actual_role="merger",
)
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_backward_compatible_without_actual_role(self):
# No actual_role supplied: behaviour falls back to resolved_role.
result = self._assess(resolved_role="reconciler")
self.assertTrue(result["proven"])
class TestReconcilerCommentThroughCanonicalPath(unittest.TestCase):
"""Integration: reconciler comment survives the poisoned author task role."""
def setUp(self):
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_whoami_violation = False
srv._preflight_capability_violation = False
srv._preflight_capability_baseline_porcelain = ""
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
def tearDown(self):
srv._preflight_in_test_mode = self._orig_in_test
srv._preflight_resolved_role = None
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@patch(
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
return_value=MASTER_SHA,
)
@patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"head_sha": MASTER_SHA,
"porcelain_status": "",
},
)
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
def test_reconciler_comment_from_control_checkout_succeeds(
self, _profile, _git, _remote_sha, mock_api, _auth, _porcelain
):
srv._preflight_resolved_role = "author" # comment_issue poison
mock_api.return_value = {"id": 9001, "html_url": "https://x/y"}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
os.environ.pop("GITEA_RECONCILER_WORKTREE", None)
result = srv.gitea_create_issue_comment(
515, "canonical reconciler audit", remote="prgs"
)
self.assertTrue(result["success"])
self.assertEqual(result["comment_id"], 9001)
mock_api.assert_called_once()
@patch("gitea_mcp_server._get_workspace_porcelain", return_value="")
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@patch(
"gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
return_value=MASTER_SHA,
)
@patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
return_value={
"current_branch": "master",
"head_sha": MASTER_SHA,
"porcelain_status": "",
},
)
@patch("gitea_mcp_server.get_profile", return_value=AUTHOR_PROFILE)
def test_author_comment_from_control_checkout_blocked(
self, _profile, _git, _remote_sha, mock_api, _auth, _porcelain
):
srv._preflight_resolved_role = "author"
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch.dict(os.environ, {}, clear=False):
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
with self.assertRaises(RuntimeError):
srv.gitea_create_issue_comment(
515, "author note", remote="prgs"
)
mock_api.assert_not_called()
if __name__ == "__main__":
unittest.main()
+257
View File
@@ -0,0 +1,257 @@
import os
import sys
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import gitea_mcp_server as srv
FAKE_AUTH = {"Authorization": "token test-token"}
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
class TestIssueCommentWorkspaceGuard(unittest.TestCase):
AUTHOR_ENV = {
"GITEA_PROFILE_NAME": "gitea-author",
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
}
def setUp(self):
srv._preflight_whoami_called = True
srv._preflight_capability_called = True
srv._preflight_resolved_role = "author"
srv._preflight_resolved_task = "comment_issue"
srv._preflight_whoami_violation = False
srv._preflight_capability_violation = False
srv._preflight_reviewer_violation_files = []
self._orig_in_test = srv._preflight_in_test_mode
srv._preflight_in_test_mode = lambda: False
self.addCleanup(self._restore_preflight_mode)
def _restore_preflight_mode(self):
srv._preflight_in_test_mode = self._orig_in_test
def _git_state(self, valid_worktree: str):
def side_effect(path):
real = os.path.realpath(path)
if real == os.path.realpath(CONTROL_CHECKOUT_ROOT):
return {
"current_branch": "master",
"head_sha": "a" * 40,
"porcelain_status": "",
}
if real == os.path.realpath(valid_worktree):
return {
"current_branch": "fix/issue-560-issue-comment-worktree-path",
"head_sha": "b" * 40,
"porcelain_status": "",
}
return {
"current_branch": "other",
"head_sha": "c" * 40,
"porcelain_status": "",
}
return side_effect
def _subprocess(self, valid_worktree: str, outside_worktree: str | None = None):
def side_effect(cmd, *args, **kwargs):
result = MagicMock(returncode=0, stdout="")
cwd = ""
if isinstance(cmd, list) and "-C" in cmd:
cwd = os.path.realpath(cmd[cmd.index("-C") + 1])
if isinstance(cmd, list) and "--show-toplevel" in cmd:
if cwd == os.path.realpath(valid_worktree):
result.stdout = f"{valid_worktree}\n"
elif outside_worktree and cwd == os.path.realpath(outside_worktree):
result.stdout = f"{outside_worktree}\n"
else:
result.stdout = f"{CONTROL_CHECKOUT_ROOT}\n"
return result
if isinstance(cmd, list) and "--git-common-dir" in cmd:
if cwd == os.path.realpath(valid_worktree):
result.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
elif outside_worktree and cwd == os.path.realpath(outside_worktree):
result.stdout = "/tmp/other-repo/.git\n"
else:
result.stdout = f"{CONTROL_CHECKOUT_ROOT}/.git\n"
return result
return result
return side_effect
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
def test_comment_rejects_control_checkout_without_worktree_path(
self, _remote_sha, mock_api, _auth
):
valid_worktree = os.path.join(
CONTROL_CHECKOUT_ROOT,
"branches",
"issue-560-issue-comment-worktree-path",
)
mock_api.return_value = {"id": 42}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
side_effect=self._git_state(valid_worktree),
):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
remote="prgs",
)
self.assertIn("stable control checkout", str(ctx.exception))
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
@patch("os.path.isdir", return_value=True)
@patch("os.path.exists", return_value=True)
def test_comment_accepts_valid_branches_worktree_path_with_explicit_repo(
self, _exists, _isdir, _remote_sha, mock_api, _auth
):
valid_worktree = os.path.join(
CONTROL_CHECKOUT_ROOT,
"branches",
"issue-560-issue-comment-worktree-path",
)
mock_api.return_value = {"id": 43}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
side_effect=self._git_state(valid_worktree),
):
with patch(
"gitea_mcp_server.subprocess.run",
side_effect=self._subprocess(valid_worktree),
):
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
result = srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
worktree_path=valid_worktree,
)
self.assertTrue(result["success"])
self.assertTrue(result["performed"])
self.assertEqual(result["comment_id"], 43)
method, url, _auth_arg, payload = mock_api.call_args[0]
self.assertEqual(method, "POST")
self.assertIn("/repos/Scaled-Tech-Consulting/Gitea-Tools/issues/557/comments", url)
self.assertEqual(payload, {"body": "canonical evidence"})
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
@patch("os.path.isdir", return_value=True)
@patch("os.path.exists", return_value=True)
def test_comment_rejects_outside_repo_worktree_path(
self, _exists, _isdir, _remote_sha, mock_api, _auth
):
valid_worktree = os.path.join(
CONTROL_CHECKOUT_ROOT,
"branches",
"issue-560-issue-comment-worktree-path",
)
outside_worktree = "/tmp/not-gitea-tools-worktree"
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
side_effect=self._git_state(valid_worktree),
):
with patch(
"gitea_mcp_server.subprocess.run",
side_effect=self._subprocess(valid_worktree, outside_worktree),
):
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
remote="prgs",
worktree_path=outside_worktree,
)
self.assertIn("does not belong to the target repository", str(ctx.exception))
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
def test_comment_still_requires_capability_preflight(self, mock_api, _auth):
srv._preflight_capability_called = False
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with self.assertRaises(RuntimeError) as ctx:
srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
remote="prgs",
worktree_path=os.path.join(
CONTROL_CHECKOUT_ROOT,
"branches",
"issue-560-issue-comment-worktree-path",
),
)
self.assertIn("Task capability", str(ctx.exception))
mock_api.assert_not_called()
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
@patch("gitea_mcp_server.api_request")
@patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
@patch("os.path.isdir", return_value=True)
@patch("os.path.exists", return_value=True)
def test_comment_still_requires_issue_comment_permission(
self, _exists, _isdir, _remote_sha, mock_api, _auth
):
valid_worktree = os.path.join(
CONTROL_CHECKOUT_ROOT,
"branches",
"issue-560-issue-comment-worktree-path",
)
denied_env = {
"GITEA_PROFILE_NAME": "gitea-author",
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
}
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
with patch(
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
side_effect=self._git_state(valid_worktree),
):
with patch(
"gitea_mcp_server.subprocess.run",
side_effect=self._subprocess(valid_worktree),
):
with patch.dict(os.environ, denied_env, clear=True):
result = srv.gitea_create_issue_comment(
issue_number=557,
body="canonical evidence",
remote="prgs",
worktree_path=valid_worktree,
)
self.assertFalse(result["success"])
self.assertFalse(result["performed"])
self.assertIn("permission_report", result)
self.assertEqual(
result["permission_report"]["missing_permission"],
"gitea.issue.comment",
)
mock_api.assert_not_called()
if __name__ == "__main__":
unittest.main()
+124
View File
@@ -0,0 +1,124 @@
"""Hermetic tests for repository-root MCP operator menu (#478)."""
import os
import stat
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SCRIPT = REPO_ROOT / "mcp-menu.sh"
DOCS = REPO_ROOT / "docs" / "mcp-menu.md"
REQUIRED_MENU_LABELS = (
"Project status / root checkout health",
"Author workflow prompts",
"Reviewer workflow prompts",
"Merger workflow prompts",
"Reconciler workflow prompts",
"Onboarding new project to this MCP workflow",
"Proxmox deployment menu placeholder",
"Create Proxmox LXC placeholder",
"Run tests",
"Exit",
)
DANGEROUS_PATTERNS = (
"git push --force",
"git push -f",
"--force-with-lease",
"delete-branch",
"gitea_delete_branch",
"issue-locks",
"issue_lock_store",
"curl ",
"wget ",
)
class TestMcpMenuScript(unittest.TestCase):
def setUp(self):
self.assertTrue(SCRIPT.is_file(), "mcp-menu.sh must exist at repo root")
self.content = SCRIPT.read_text(encoding="utf-8")
def test_script_exists_at_repo_root(self):
self.assertEqual(SCRIPT.name, "mcp-menu.sh")
self.assertEqual(SCRIPT.parent, REPO_ROOT)
def test_executable_bit_is_set(self):
mode = SCRIPT.stat().st_mode
self.assertTrue(mode & stat.S_IXUSR, "mcp-menu.sh must be executable by owner")
def test_shebang(self):
first_line = self.content.splitlines()[0]
self.assertEqual(first_line, "#!/usr/bin/env bash")
def test_uses_set_euo_pipefail(self):
self.assertIn("set -euo pipefail", self.content)
def test_no_dangerous_commands(self):
lowered = self.content.lower()
for pattern in DANGEROUS_PATTERNS:
with self.subTest(pattern=pattern):
self.assertNotIn(pattern.lower(), lowered)
def test_no_branch_deletion_verbs(self):
for token in ("git branch -D", "git branch -d", "push :refs"):
with self.subTest(token=token):
self.assertNotIn(token, self.content)
def test_contains_required_menu_labels(self):
for label in REQUIRED_MENU_LABELS:
with self.subTest(label=label):
self.assertIn(label, self.content)
def test_run_tests_prefers_run_tests_sh(self):
self.assertIn('run-tests.sh', self.content)
run_tests_idx = self.content.index("run_tests()")
run_tests_body = self.content[run_tests_idx : run_tests_idx + 800]
pytest_idx = run_tests_body.find("pytest")
run_tests_sh_idx = run_tests_body.find("run-tests.sh")
self.assertGreater(pytest_idx, 0)
self.assertGreater(run_tests_sh_idx, 0)
self.assertLess(run_tests_sh_idx, pytest_idx)
def test_run_tests_fail_closed_without_runner(self):
self.assertIn("fail closed", self.content.lower())
self.assertIn("exit 1", self.content)
def test_proxmox_entries_are_placeholders(self):
self.assertIn("TODO / issue-backed", self.content)
self.assertIn("NOT implemented", self.content)
def test_root_health_shows_required_fields(self):
health_fn = self._extract_function("show_root_checkout_health")
for snippet in (
"status --short --branch",
"rev-parse HEAD",
"prgs/master",
"WARNING: root checkout",
):
with self.subTest(snippet=snippet):
self.assertIn(snippet, health_fn)
def test_docs_mention_mcp_menu_sh(self):
self.assertTrue(DOCS.is_file(), "docs/mcp-menu.md must exist")
docs_text = DOCS.read_text(encoding="utf-8")
self.assertIn("./mcp-menu.sh", docs_text)
self.assertIn("placeholder", docs_text.lower())
def _extract_function(self, name: str) -> str:
marker = f"{name}() {{"
start = self.content.index(marker)
depth = 0
for idx in range(start, len(self.content)):
char = self.content[idx]
if char == "{":
depth += 1
elif char == "}":
depth -= 1
if depth == 0:
return self.content[start : idx + 1]
self.fail(f"Could not parse function {name}")
if __name__ == "__main__":
unittest.main()
+164
View File
@@ -3965,3 +3965,167 @@ class TestPreflightVerification(unittest.TestCase):
self.assertIn("worktree_path argument", msg)
self.assertIn("task_file.py", msg)
self.assertIn("author namespace", msg)
class TestIssue546Deadlock(unittest.TestCase):
def setUp(self):
import reviewer_pr_lease
import review_workflow_load
import mcp_server
reviewer_pr_lease.clear_session_lease()
review_workflow_load.clear_review_workflow_load()
mcp_server._preflight_capability_called = False
mcp_server._preflight_resolved_role = None
mcp_server._REVIEW_DECISION_LOCK = None
self.auth_user_patch = patch("mcp_server._authenticated_username", return_value="reviewer-bot")
self.auth_user_patch.start()
self.addCleanup(self.auth_user_patch.stop)
self.verify_purity_patch = patch("mcp_server.verify_preflight_purity", return_value=None)
self.verify_purity_patch.start()
self.addCleanup(self.verify_purity_patch.stop)
self.verify_workspace_patch = patch("mcp_server._verify_role_mutation_workspace", return_value="/workspace")
self.verify_workspace_patch.start()
self.addCleanup(self.verify_workspace_patch.stop)
self.get_profile_patch = patch("mcp_server.get_profile", return_value={
"profile_name": "prgs-reviewer",
"allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.review", "gitea.pr.approve"],
"forbidden_operations": [],
})
self.get_profile_patch.start()
self.addCleanup(self.get_profile_patch.stop)
self.pr_work_lease_patch = patch(
"mcp_server._pr_work_lease_reviewer_block",
return_value=dict(_NO_PR_WORK_LEASE_BLOCK),
)
self.pr_work_lease_patch.start()
self.addCleanup(self.pr_work_lease_patch.stop)
self.auth_header_patch = patch("mcp_server.get_auth_header", return_value="token test")
self.auth_header_patch.start()
self.addCleanup(self.auth_header_patch.stop)
def test_workflow_no_deadlock(self):
import mcp_server
import reviewer_pr_lease
# 1. Resolve capability
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
self.assertTrue(res["allowed_in_current_session"])
# 2. Load review workflow
res = mcp_server.gitea_load_review_workflow()
self.assertTrue(res["success"])
# Mock API requests for lease comments and PR retrieval
comments = []
def mock_api_side(method, url, auth, data=None):
if "/api/v1/user" in url:
return {"login": "reviewer-bot"}
if "/pulls/" in url:
if "/reviews" in url:
if method == "POST":
return {"id": 2001, "state": "APPROVED"}
return [{"id": 2001, "user": {"login": "reviewer-bot"}, "state": "APPROVED", "commit_id": "abc123"}]
return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc123"}, "mergeable": True}
if "/comments" in url:
if method == "POST":
comments.append({"id": 1001, "body": data["body"], "user": {"login": "reviewer-bot"}})
return {"id": 1001}
return comments
return {}
with patch("mcp_server.api_request", side_effect=mock_api_side):
# 3. Acquire reviewer lease
res = mcp_server.gitea_acquire_reviewer_pr_lease(
pr_number=550,
worktree="/workspace",
candidate_head="abc123",
remote="prgs",
)
self.assertTrue(res["success"])
self.assertEqual(res["comment_id"], 1001)
# verify lease is recorded in session
session_lease = reviewer_pr_lease.get_session_lease()
self.assertIsNotNone(session_lease)
self.assertEqual(session_lease["pr_number"], 550)
# 4. Resolve capability again (simulating subsequent tool call preflight check/token refresh)
# This should NOT clear the session lease or decision lock!
mcp_server._preflight_capability_called = False # simulate clearance from prior mutation
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
self.assertTrue(res["allowed_in_current_session"])
self.assertIsNotNone(reviewer_pr_lease.get_session_lease()) # Preserved!
# 5. Mark final review decision APPROVED
res = mcp_server.gitea_mark_final_review_decision(
pr_number=550,
action="approve",
expected_head_sha="abc123",
remote="prgs",
)
self.assertTrue(res["marked_ready"])
# 6. Submit review (should succeed without deadlock)
res = mcp_server.gitea_submit_pr_review(
pr_number=550,
action="approve",
body="APPROVED",
expected_head_sha="abc123",
final_review_decision_ready=True,
remote="prgs",
worktree_path="/workspace",
)
self.assertTrue(res["performed"])
def test_release_reviewer_pr_lease(self):
import mcp_server
import reviewer_pr_lease
# Resolve capability and load workflow
mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
mcp_server.gitea_load_review_workflow()
comments = [_reviewer_lease_comment(550, session_id=_DEFAULT_LEASE_SESSION, head_sha="abc123")]
posted_comments = []
def mock_api_side(method, url, auth, data=None):
if "/api/v1/user" in url:
return {"login": "reviewer-bot"}
if "/pulls/" in url:
return {"user": {"login": "author-user"}, "state": "open", "head": {"sha": "abc123"}, "mergeable": True}
if "/comments" in url:
if method == "POST":
new_c = {"id": 1002, "body": data["body"], "user": {"login": "reviewer-bot"}}
comments.append(new_c)
posted_comments.append(new_c)
return {"id": 1002}
return comments
return {}
# Set session lease in memory
import merger_lease_adoption as mla
reviewer_pr_lease.record_session_lease({
"pr_number": 550,
"session_id": _DEFAULT_LEASE_SESSION,
"candidate_head": "abc123",
"target_branch": "master",
"comment_id": 9001,
}, lease_provenance=mla.build_lease_provenance(source=mla.SOURCE_ACQUIRE, comment_id=9001))
with patch("mcp_server.api_request", side_effect=mock_api_side):
# Release lease
res = mcp_server.gitea_release_reviewer_pr_lease(pr_number=550, remote="prgs")
self.assertTrue(res["success"])
self.assertTrue(res["released"])
self.assertEqual(res["comment_id"], 1002)
# verify lease is cleared in session
self.assertIsNone(reviewer_pr_lease.get_session_lease())
# verify comment phase is released
self.assertIn("phase: released", posted_comments[0]["body"])
+64 -1
View File
@@ -258,7 +258,9 @@ class TestMergerHandoffMutationGate(unittest.TestCase):
class TestAdoptTool(unittest.TestCase):
def setUp(self):
leases.clear_session_lease()
patch("mcp_server.verify_preflight_purity", return_value=None).start()
# Patch only the role workspace verifier (which internally does the single
# verify_preflight_purity); the direct duplicate call was removed to prevent
# capability state clearing. See fix for #548.
patch("mcp_server._verify_role_mutation_workspace", return_value=None).start()
patch("gitea_audit.audit_enabled", return_value=False).start()
mcp_server = __import__("mcp_server")
@@ -363,6 +365,67 @@ class TestAdoptTool(unittest.TestCase):
)
self.assertFalse(gate["block"])
def test_adopt_preserves_preflight_capability_state_no_duplicate_clear(self):
"""Prove fix for #548: adopt calls _verify (single purity) and does not
trigger duplicate verify that would clear capability state, avoiding
need for temp local patches or raw fallbacks.
"""
from unittest.mock import patch, MagicMock
import mcp_server as mcp_server_mod
# fresh preflight state (as controller/reconciler would have from resolve)
mcp_server_mod.record_preflight_check("whoami")
mcp_server_mod.record_preflight_check(
"capability", "merger", resolved_task="adopt_merger_pr_lease"
)
initial_cap_called = mcp_server_mod._preflight_capability_called
verify_patch = patch("mcp_server.verify_preflight_purity", wraps=mcp_server_mod.verify_preflight_purity)
def _role_side(*a, **k):
# simulate the real _verify which calls verify once with path
mcp_server_mod.verify_preflight_purity(k.get("remote") or a[0] if a else None, worktree_path=k.get("worktree") or "branches/work", task=k.get("task"))
return "branches/work"
role_patch = patch("mcp_server._verify_role_mutation_workspace", side_effect=_role_side)
with verify_patch as vmock, role_patch, \
patch("mcp_server.get_auth_header", return_value="token test"), \
patch("mcp_server._authenticated_username", return_value="sysadmin"):
# minimal mocks to let adopt reach end without full api
with patch("mcp_server.get_profile") as gp:
gp.return_value = {
"profile_name": "prgs-merger",
"allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.merge"],
"forbidden_operations": [],
}
with patch("mcp_server.api_request") as api:
def _side(url, *a, **k):
if "/pulls/" in str(url) and "/comments" not in str(url):
return {"state": "open", "head": {"sha": "deadbeef"*5}, "merged": False, "number": 999}
if "/comments" in str(url):
return [] # list of comments for fetch
return {"id": 42}
api.side_effect = _side
with patch("mcp_server.gitea_get_pr_review_feedback", return_value={"approval_at_current_head": True}):
with patch("merger_lease_adoption.assess_adopt_merger_lease", return_value={
"adopt_allowed": True,
"active_lease": {"session_id": "rev", "profile": "prgs-reviewer", "reviewer_identity": "sysadmin", "issue_number": 548, "target_branch": "master"},
"adoption_body": "<!-- mcp-review-lease-adoption:v1 --> adopted",
}):
res = mcp_server_mod.gitea_adopt_merger_pr_lease(
pr_number=999,
worktree="branches/merge-test",
expected_head_sha="deadbeef"*5,
issue_number=548,
remote="prgs",
)
self.assertTrue(res.get("success"))
# verify_preflight_purity should have been invoked exactly once (via the _verify_role path)
# not twice (old dup would have cleared state mid-way, requiring temp patches)
self.assertEqual(vmock.call_count, 1)
# capability state management: the single verify consumes (clears) as designed;
# prior state was valid, no unexpected clear between "checks" or hidden reset
# (the removed dup line was the source of the #548 clearing bug)
self.assertTrue(initial_cap_called)
# final may be set by other records in flow, but key is single invocation and success without bypasses
if __name__ == "__main__":
unittest.main()
@@ -26,6 +26,11 @@ class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
result = _list_pr_lease_comments(
12, remote="prgs", host=None, org=None, repo=None)
self.assertEqual(result, [])
# Single comments GET only — no pre-flight PR lookup (#519 isolation).
mock_api.assert_called_once()
_method, url, _auth_arg = mock_api.call_args[0][:3]
self.assertIn("/issues/12/comments", url)
self.assertNotIn("/pulls/", url)
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -73,4 +78,4 @@ class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
unittest.main()
+178
View File
@@ -0,0 +1,178 @@
"""Regression coverage for the remote/repo mismatch guard (#530).
Bare ``remote=prgs`` historically resolved to ``Scaled-Tech-Consulting/Timesheet``
(the hardcoded ``REMOTES`` default) instead of ``Scaled-Tech-Consulting/Gitea-Tools``,
which is what the local git remote actually points at. That silent mismatch caused
false 404s and risked mutating the wrong repository. The guard fails closed when the
MCP-resolved org/repo disagrees with the local git remote URL and the caller did not
pass explicit ``org``/``repo``.
"""
import os
import unittest
from unittest import mock
import remote_repo_guard
LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
class TestAssessRemoteRepoMatch(unittest.TestCase):
def test_explicit_org_and_repo_skips_guard(self):
# Caller supplied both org and repo explicitly: never block, even on mismatch.
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Timesheet",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=True,
repo_explicit=True,
)
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
self.assertEqual(assessment["reasons"], [])
def test_missing_local_remote_url_skips_guard(self):
# Best-effort lookup: no local remote URL (non-checkout usage) => do not block.
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Timesheet",
local_remote_url=None,
org_explicit=False,
repo_explicit=False,
)
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
def test_matching_repo_is_proven(self):
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Gitea-Tools",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=False,
repo_explicit=False,
)
self.assertTrue(assessment["proven"])
self.assertFalse(assessment["block"])
def test_regression_prgs_default_timesheet_vs_local_gitea_tools(self):
# The exact #530 scenario: default repo Timesheet, local remote Gitea-Tools.
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Timesheet",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=False,
repo_explicit=False,
)
self.assertFalse(assessment["proven"])
self.assertTrue(assessment["block"])
self.assertTrue(assessment["reasons"])
self.assertEqual(assessment["resolved_repo"], "Timesheet")
self.assertEqual(assessment["resolved_org"], "Scaled-Tech-Consulting")
def test_only_org_explicit_still_checks_repo(self):
# Repo left as default => still guarded even if org was explicit.
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Timesheet",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=True,
repo_explicit=False,
)
self.assertTrue(assessment["block"])
def test_case_insensitive_match(self):
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="scaled-tech-consulting",
resolved_repo="gitea-tools",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=False,
repo_explicit=False,
)
self.assertTrue(assessment["proven"])
def test_format_error_mentions_resolved_and_local(self):
assessment = remote_repo_guard.assess_remote_repo_match(
remote="prgs",
resolved_org="Scaled-Tech-Consulting",
resolved_repo="Timesheet",
local_remote_url=LOCAL_GITEA_TOOLS_URL,
org_explicit=False,
repo_explicit=False,
)
message = remote_repo_guard.format_remote_repo_guard_error(assessment)
self.assertIn("Timesheet", message)
self.assertIn("Gitea-Tools", message)
self.assertIn("org=", message)
self.assertIn("repo=", message)
class TestResolveServerWiring(unittest.TestCase):
"""The guard is wired into gitea_mcp_server._resolve, so every read/lookup/
mutation tool that resolves a target fails closed on a repo mismatch.
Under pytest the guard is bypassed unless GITEA_FORCE_REMOTE_REPO_CHECK is set,
so these tests set the force flag and patch the local remote lookup + REMOTES.
"""
def setUp(self):
import gitea_mcp_server
self.server = gitea_mcp_server
force = mock.patch.dict(
os.environ, {"GITEA_FORCE_REMOTE_REPO_CHECK": "1"}, clear=False
)
force.start()
self.addCleanup(force.stop)
url_patch = mock.patch.object(
gitea_mcp_server,
"_local_git_remote_url",
return_value=LOCAL_GITEA_TOOLS_URL,
)
url_patch.start()
self.addCleanup(url_patch.stop)
def _set_prgs_default_repo(self, repo):
original = dict(self.server.REMOTES["prgs"])
self.addCleanup(self.server.REMOTES.__setitem__, "prgs", original)
self.server.REMOTES["prgs"] = {**original, "repo": repo}
def test_resolve_blocks_on_default_repo_mismatch(self):
self._set_prgs_default_repo("Timesheet")
with self.assertRaises(RuntimeError) as ctx:
self.server._resolve("prgs", None, None, None)
self.assertIn("Timesheet", str(ctx.exception))
self.assertIn("Gitea-Tools", str(ctx.exception))
def test_resolve_allows_explicit_org_and_repo(self):
self._set_prgs_default_repo("Timesheet")
# Explicit org/repo is authoritative even if it does not match local remote.
host, org, repo = self.server._resolve(
"prgs", None, "Scaled-Tech-Consulting", "Timesheet"
)
self.assertEqual((org, repo), ("Scaled-Tech-Consulting", "Timesheet"))
def test_resolve_allows_matching_default(self):
self._set_prgs_default_repo("Gitea-Tools")
host, org, repo = self.server._resolve("prgs", None, None, None)
self.assertEqual((org, repo), ("Scaled-Tech-Consulting", "Gitea-Tools"))
def test_lookup_tool_cannot_silently_query_different_repo(self):
# gitea_view_issue resolves the target via _resolve first, so a bare
# remote=prgs pointing at the wrong default repo fails closed before any
# API call is made.
self._set_prgs_default_repo("Timesheet")
with self.assertRaises(RuntimeError):
self.server.gitea_view_issue(issue_number=1, remote="prgs")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,141 @@
"""Tests for exact per-mutation capability proof in reviewer reports (#405)."""
import sys
import unittest
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from reviewer_mutation_capability_proof import assess_mutation_capability_proof
from review_proofs import assess_mutation_capability_proof as proofs_assess
from final_report_validator import assess_final_report_validator
REVIEW_ONLY = """
Review decision: approved
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | resolved before review
"""
MERGE_PROVEN = """
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | resolved before merge
"""
MERGE_AND_DELETE_PROVEN = """
Review decision: approved
Merge result: merged 0123456789ab
Remote branch deleted: feat/issue-x
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
- gitea_delete_branch | delete_branch (gitea.branch.delete) | deleted | before delete
"""
class TestModule(unittest.TestCase):
def test_review_only_with_capability_passes(self):
r = assess_mutation_capability_proof(REVIEW_ONLY)
self.assertTrue(r["proven"], r["reasons"])
def test_merge_with_exact_capability_passes(self):
r = assess_mutation_capability_proof(MERGE_PROVEN)
self.assertTrue(r["proven"], r["reasons"])
def test_merge_and_delete_fully_proven_passes(self):
r = assess_mutation_capability_proof(MERGE_AND_DELETE_PROVEN)
self.assertTrue(r["proven"], r["reasons"])
def test_review_pr_does_not_authorize_merge(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("merge" in x.lower() for x in r["reasons"]), r["reasons"])
def test_merge_pr_does_not_authorize_branch_deletion(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Remote branch deleted: feat/issue-x
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("delet" in x.lower() for x in r["reasons"]), r["reasons"])
def test_delete_skipped_when_capability_missing_passes(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Branch deletion: skipped delete_branch capability not available
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | before merge
"""
r = assess_mutation_capability_proof(report)
self.assertTrue(r["proven"], r["reasons"])
def test_missing_table_when_merging_blocks(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
merge_pr gitea.pr.merge resolved
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("table" in x.lower() for x in r["reasons"]), r["reasons"])
def test_post_hoc_proof_blocks(self):
report = """
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_merge_pr | merge_pr (gitea.pr.merge) | merged | capability resolved after merge
"""
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
self.assertTrue(any("after" in x.lower() for x in r["reasons"]), r["reasons"])
def test_review_without_capability_blocks(self):
report = "Review decision: approved\nreview submitted\n"
r = assess_mutation_capability_proof(report)
self.assertFalse(r["proven"])
def test_no_mutation_no_requirement(self):
r = assess_mutation_capability_proof("Selected PR: #1\nSkipped, no action.")
self.assertTrue(r["proven"], r["reasons"])
class TestWiring(unittest.TestCase):
def test_review_proofs_wrapper_matches_module(self):
self.assertEqual(
proofs_assess(MERGE_PROVEN)["proven"],
assess_mutation_capability_proof(MERGE_PROVEN)["proven"],
)
def test_final_report_validator_flags_nearby_capability_merge(self):
report = """
## Controller Handoff
- Task: review_pr
Review decision: approved
Merge result: merged 0123456789ab
Mutation capability table:
- gitea_submit_pr_review | review_pr (gitea.pr.review) | submitted | before review
"""
result = assess_final_report_validator(report, "review_pr")
rule_ids = [f["rule_id"] for f in result.get("findings", [])]
self.assertIn("reviewer.mutation_capability_proof", rule_ids)
if __name__ == "__main__":
unittest.main()
+65
View File
@@ -0,0 +1,65 @@
"""Contract checks for the root-level `run-tests.sh` convenience runner (#473).
`run-tests.sh` is the canonical full-validation entry point. It must invoke the
project virtualenv interpreter, forward extra args to pytest, fail closed when
the venv Python is missing, and stay repo-local (no network, no lock files).
These checks pin that behavior so it cannot silently regress, and confirm the
developer testing guide names the runner.
"""
import stat
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
SCRIPT = REPO_ROOT / "run-tests.sh"
GUIDE = REPO_ROOT / "docs" / "developer-testing-guidelines.md"
def _script_text() -> str:
return SCRIPT.read_text(encoding="utf-8")
def test_run_tests_script_exists():
assert SCRIPT.is_file(), "run-tests.sh must exist at the repository root"
def test_run_tests_script_is_executable():
mode = SCRIPT.stat().st_mode
assert mode & stat.S_IXUSR, "run-tests.sh must be executable (chmod +x)"
def test_run_tests_script_has_strict_bash_flags():
text = _script_text()
assert text.startswith("#!/usr/bin/env bash"), "must use the bash shebang"
assert "set -euo pipefail" in text, "must set -euo pipefail"
def test_run_tests_script_uses_venv_pytest_and_forwards_args():
text = _script_text()
# Resolves the venv interpreter relative to the script's own directory.
assert "venv/bin/python" in text, "must use the project virtualenv Python"
assert "-m pytest" in text, "must run pytest via the module"
assert '"$@"' in text, "must forward extra CLI args to pytest"
def test_run_tests_script_fails_closed_without_venv():
text = _script_text()
# Missing venv must be an explicit, non-zero-exit error, not a silent
# fallback to the wrong interpreter.
assert "if [[ ! -x" in text, "must guard on an executable venv Python"
assert "exit 1" in text, "must exit non-zero when the venv is missing"
assert "ERROR" in text, "must print a clear error message"
def test_run_tests_script_stays_repo_local():
text = _script_text()
# No network calls, no lock-file writes from the runner itself.
for forbidden in ("curl", "wget", "gitea_issue_lock.json"):
assert forbidden not in text, f"runner must not reference {forbidden!r}"
def test_guide_names_canonical_runner():
text = " ".join(GUIDE.read_text(encoding="utf-8").split())
assert "./run-tests.sh" in text, "testing guide must name ./run-tests.sh"
assert "./run-tests.sh tests/test_mcp_server.py -q" in text, (
"testing guide must show the focused-validation example"
)