Compare commits

..
Author SHA1 Message Date
sysadminandClaude Opus 4.8 eb97060867 docs(wiki): refresh Gitea-Tools wiki proof head to d1f0693 (#232)
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-06 01:56:10 -04:00
11 changed files with 39 additions and 1121 deletions
+1 -2
View File
@@ -32,7 +32,6 @@ Wiki-related issues cannot be closed until the live Wiki is verified — see
| Repository | `docs/wiki/` source | Gitea Wiki published | Proof |
|---|---|---|---|
| `Scaled-Tech-Consulting/Gitea-Tools` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools/wiki/Home); 10 pages; wiki git log head `d1f0693` |
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home); 10 pages (History, Home, Identity-and-Profiles, MCP-Tools, Open-Decisions, Operator-Guide, Repositories, Runbooks, Safety-and-Gates, Workflow); wiki git log head `ef3dec2` |
| `Scaled-Tech-Consulting/mcp-control-plane` | yes (10 pages) | published (verified 2026-07-06) | [Wiki Home](https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane/wiki/Home) |
Update this table whenever a wiki is published, re-synced, or found stale.
+1 -9
View File
@@ -205,7 +205,7 @@ def selected_profile_name():
def is_runtime_switching_enabled(path=None):
"""Check if runtime profile switching is enabled in config."""
"""Check if runtime profile switching is explicitly enabled in config."""
try:
config = load_config(path)
except Exception:
@@ -213,18 +213,10 @@ def is_runtime_switching_enabled(path=None):
if not config:
return False
rules = config.get("rules") or {}
if rules.get("allow_runtime_switching") is False:
return False
if config.get("allow_runtime_switching") is False:
return False
if rules.get("allow_runtime_switching") is True:
return True
if config.get("allow_runtime_switching") is True:
return True
# Default to True if multiple profiles exist in the config
profiles = config.get("profiles") or {}
if len(profiles) > 1:
return True
return False
+4 -154
View File
@@ -259,7 +259,6 @@ import issue_duplicate_gate # noqa: E402
import role_session_router # noqa: E402
import role_namespace_gate # noqa: E402
import task_capability_map # noqa: E402
import review_proofs # noqa: E402
# Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue,
@@ -2275,23 +2274,6 @@ def gitea_merge_pr(
return result
def _local_git_remote_url(remote_name: str) -> str | None:
"""Best-effort local ``git remote get-url`` for trust-gate corroboration."""
try:
proc = subprocess.run(
["git", "remote", "get-url", remote_name],
capture_output=True,
text=True,
cwd=PROJECT_ROOT,
)
if proc.returncode != 0:
return None
url = (proc.stdout or "").strip()
return url or None
except Exception:
return None
@mcp.tool()
def gitea_review_pr(
pr_number: int,
@@ -2359,8 +2341,6 @@ def gitea_review_pr(
prs_found_count = 0
pr_details_list = []
inventory_msg = ""
inventory_trust_gate = None
prs: list = []
h, o, r = _resolve(remote, host, org, repo)
auth = None
@@ -2445,28 +2425,7 @@ def gitea_review_pr(
if inventory_msg:
report_lines.append(inventory_msg)
else:
inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
prs if inventory_attempted else None,
remote=remote,
org=o,
repo=r,
state="open",
authenticated_profile=profile,
local_remote_url=_local_git_remote_url(remote),
has_finality_metadata=True,
)
report_lines.extend(
review_proofs.format_pr_inventory_trust_gate_report(
inventory_trust_gate
)
)
if inventory_trust_gate.get("status") == "trusted_empty":
report_lines.append("Open PRs found: 0 (trusted_empty)")
else:
report_lines.append(
"Empty-queue claim blocked: "
"pr_inventory_trust_gate did not return trusted_empty"
)
report_lines.append("Open PRs found: 0")
else:
report_lines.append(inventory_msg)
@@ -2758,76 +2717,6 @@ def _permission_block_report(required_operation: str,
return report
def _role_for_operation(op: str) -> str | None:
# Normalize op first
try:
op_n = gitea_config.normalize_operation(op)
except Exception:
op_n = op
if op_n in ("gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review", "gitea.pr.request_changes"):
return "reviewer"
if op_n in (
"gitea.issue.create",
"gitea.issue.comment",
"gitea.issue.close",
"gitea.branch.create",
"gitea.branch.push",
"gitea.pr.create",
"gitea.pr.comment",
"gitea.pr.close",
"gitea.branch.delete",
"gitea.repo.commit"
):
return "author"
return None
def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
"""Try to find a profile in config that allows op and has valid credentials.
If found, switch to it, clear identity cache, and return True.
Otherwise return False.
"""
role = _role_for_operation(op)
if not role:
return False
if not gitea_config.is_runtime_switching_enabled():
return False
config = gitea_config.load_config()
if not config or "profiles" not in config:
return False
for p_name, p_data in config["profiles"].items():
# Role classification matching
p_role = p_data.get("role") or _role_kind(p_data.get("allowed_operations", []), p_data.get("forbidden_operations", []))
if p_role != role:
continue
p_allowed = p_data.get("allowed_operations") or []
p_forbidden = p_data.get("forbidden_operations") or []
p_allowed_n = []
for op_val in p_allowed:
try:
p_allowed_n.append(gitea_config.normalize_operation(op_val))
except Exception:
pass
p_forbidden_n = []
for op_val in p_forbidden:
try:
p_forbidden_n.append(gitea_config.normalize_operation(op_val))
except Exception:
pass
ok, _ = gitea_config.check_operation(op, p_allowed_n, p_forbidden_n)
if ok:
try:
tok = gitea_config.resolve_token(p_data)
if tok:
gitea_config._active_profile_override = p_name
_IDENTITY_CACHE.clear()
return True
except Exception:
pass
return False
def _profile_operation_gate(op: str) -> list[str]:
"""Profile permission check for a single gated operation (#126, #216).
@@ -2845,17 +2734,6 @@ def _profile_operation_gate(op: str) -> list[str]:
op, profile["allowed_operations"], profile["forbidden_operations"])
if op_ok:
return []
if _try_auto_switch_for_operation(op):
try:
profile = get_profile()
op_ok, op_reason = gitea_config.check_operation(
op, profile["allowed_operations"], profile["forbidden_operations"])
if op_ok:
return []
except Exception as exc:
return [f"profile could not be resolved (fail closed): {_redact(str(exc))}"]
if op_reason == "no-allowed-operations":
return ["profile has no configured allowed operations (fail closed)"]
if op_reason == "forbidden":
@@ -4507,12 +4385,6 @@ def gitea_resolve_task_capability(
required_permission, active_allowed, active_forbidden
)
switching = gitea_config.is_runtime_switching_enabled()
available_in_session = allowed_in_current_session
configured = False
restart_required = False
reason_msg = None
# Find matching configured profiles
matching_profiles = []
if config and "profiles" in config:
@@ -4535,24 +4407,7 @@ def gitea_resolve_task_capability(
if ok:
matching_profiles.append(p_name)
if not allowed_in_current_session and switching:
# Try to auto switch
if _try_auto_switch_for_operation(required_permission, h):
# Successfully switched!
profile = get_profile()
username = _authenticated_username(h) if h else None
active_allowed = profile.get("allowed_operations") or []
active_forbidden = profile.get("forbidden_operations") or []
allowed_in_current_session = True
available_in_session = True
else:
# Check if any profile in config allows it (meaning it is configured but unattached)
if matching_profiles:
configured = True
restart_required = True
available_in_session = False
reason_msg = f"{required_role.capitalize()} profile exists but MCP server was added after session startup and is not attached"
switching = gitea_config.is_runtime_switching_enabled()
different_namespace_required = False
next_safe_action = "None; ready for operations."
@@ -4630,19 +4485,14 @@ def gitea_resolve_task_capability(
"active_identity": username,
"active_profile_allowed_operations": active_allowed,
"allowed_in_current_session": allowed_in_current_session,
"available_in_session": available_in_session,
"configured": configured,
"restart_required": restart_required,
"stop_required": stop_required or restart_required,
"stop_required": stop_required,
"task_role_guidance": task_role_guidance,
"matching_configured_profile": matching_profiles,
"runtime_switching_supported": switching,
"different_mcp_namespace_required": different_namespace_required,
"exact_safe_next_action": next_safe_action,
}
if reason_msg:
result["reason"] = reason_msg
if stop_required or restart_required:
if stop_required:
terminal = capability_stop_terminal.enter_from_capability_result(result)
if terminal:
result["terminal_mode"] = True
+1 -116
View File
@@ -17,7 +17,6 @@ here weakens or replaces them.
import re
import issue_duplicate_gate
from reviewer_worktree import assess_reviewer_worktree_proof
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$")
@@ -638,7 +637,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
role_boundary=None, review_mutation=None,
report_text=None, review_decision_lock=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None, worktree_proof=None):
sweep_proof=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -708,14 +707,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"downgraded": True,
"reasons": ["review mutation proof not provided (#211)"],
}
if worktree_proof is not None:
worktree = assess_reviewer_worktree_proof(worktree_proof)
else:
worktree = {
"proven": False,
"block": True,
"reasons": ["reviewer worktree proof not provided (#233)"],
}
capability_proven = bool(capability_evidence.get("proven"))
sweep_proven = bool(sweep.get("proven"))
@@ -728,7 +719,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"reasons": ["review mutation proof missing"],
}
review_mutation_complete = bool(review_mutation.get("complete"))
worktree_proven = bool(worktree.get("proven"))
downgrade_reasons = []
if not identity_eligible:
@@ -782,11 +772,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
if not review_mutation_complete:
downgrade_reasons.append("review mutation proof missing or incomplete (#211)")
downgrade_reasons.extend(review_mutation.get("reasons", []))
if not worktree_proven:
downgrade_reasons.append(
"reviewer worktree safety proof missing or failed (#233)"
)
downgrade_reasons.extend(worktree.get("reasons", []))
merge_allowed = (
identity_eligible
@@ -797,7 +782,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
and validation.get("verdict") != "invalid"
# #179: no merge without a proven final live-state recheck.
and live_state_proven
and worktree_proven
)
violations = []
@@ -841,11 +825,6 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"live_state_recheck_proven": live_state_proven,
"role_boundary_clean": role_boundary_clean,
"review_mutation_complete": review_mutation_complete,
"worktree_proof_proven": worktree_proven,
"worktree_scratch_used": bool(worktree.get("scratch_used")),
"unrelated_mutations_avoided": bool(
worktree.get("unrelated_mutations_avoided")
),
}
@@ -992,12 +971,6 @@ HANDOFF_ROLE_FIELDS = {
("Selected PR", ("selected pr",)),
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
("Worktree path", ("worktree path", "starting worktree path")),
("Worktree dirty", ("worktree dirty", "whether worktree was dirty")),
("Scratch worktree used", ("scratch worktree used", "scratch clone used",
"scratch worktree")),
("Unrelated local mutations", ("unrelated local mutations",
"unrelated files modified")),
("Review decision", ("review decision", "decision")),
("Merge result", ("merge result",)),
("Linked issue status", ("linked issue status", "linked issue")),
@@ -1005,7 +978,6 @@ HANDOFF_ROLE_FIELDS = {
),
"author": (
("Selected issue", ("selected issue",)),
("Issue lock proof", ("issue lock proof", "lock before diff")),
("Claim/comment status", ("claim/comment status", "claim status",
"claim")),
("PR number opened", ("pr number opened", "pr opened", "pr number")),
@@ -1312,93 +1284,6 @@ def pr_inventory_trust_gate(
}
def _split_repo_slug(full_repo: str) -> tuple[str | None, str | None]:
parts = (full_repo or "").split("/", 1)
if len(parts) == 2:
return parts[0].strip() or None, parts[1].strip() or None
return None, None
def assess_reviewer_queue_inventory(
repo_reports: list[dict] | None,
required_repos: list[str] | None = None,
*,
user_context: str | None = None,
) -> dict:
"""Canonical reviewer queue path: completeness plus per-repo trust gates (#196).
Any repository reporting ``open_pr_count == 0`` must pass
``pr_inventory_trust_gate`` with ``trusted_empty`` before an empty-queue
claim is allowed. A bare ``[]`` from ``gitea_list_prs`` is never sufficient.
"""
required = list(required_repos or [
"Scaled-Tech-Consulting/Gitea-Tools",
"Scaled-Tech-Consulting/mcp-control-plane",
])
completeness = assess_inventory_completeness(repo_reports, required)
trust_gates: dict[str, dict] = {}
blockers: list[str] = []
can_claim_empty = bool(completeness.get("complete"))
for report in repo_reports or []:
repo = (report.get("repo") or "").strip()
count = report.get("open_pr_count")
if not isinstance(count, int) or count != 0:
continue
org, repo_name = _split_repo_slug(repo)
list_response = report.get("list_prs_response")
if list_response is None:
list_response = []
gate = pr_inventory_trust_gate(
list_response,
remote=report.get("remote"),
org=org,
repo=repo_name,
state=report.get("state_filter"),
authenticated_profile=report.get("authenticated_profile"),
local_remote_url=report.get("local_remote_url"),
user_context=user_context or report.get("user_context"),
corroboration_open_pr_counter=report.get(
"corroboration_open_pr_counter"
),
has_finality_metadata=report.get("pagination_complete") is True,
)
trust_gates[repo] = gate
status = gate.get("status")
if status != "trusted_empty":
can_claim_empty = False
blockers.append(
f"repository '{repo}': empty PR list trust gate is "
f"'{status}'; cannot claim 'no open PRs'"
)
blockers.extend(gate.get("reasons") or [])
return {
"complete": bool(completeness.get("complete")),
"can_claim_empty_queue": can_claim_empty,
"can_claim_exhaustive": (
bool(completeness.get("can_claim_exhaustive")) and can_claim_empty
),
"inventory_reasons": list(completeness.get("reasons") or []),
"trust_gates": trust_gates,
"blockers": blockers,
"reasons": list(completeness.get("reasons") or []) + blockers,
}
def format_pr_inventory_trust_gate_report(gate: dict) -> list[str]:
"""Render trust-gate lines for MCP inventory output."""
lines = [f"pr_inventory_trust_gate.status: {gate.get('status', 'unknown')}"]
if gate.get("corroborated"):
lines.append("pr_inventory_trust_gate.corroborated: true")
for reason in gate.get("reasons") or []:
lines.append(f"pr_inventory_trust_gate.reason: {reason}")
return lines
def assess_duplicate_search_proof(report_text, matches):
"""#207: reject LLM duplicate summaries that omit known title matches."""
return issue_duplicate_gate.assess_duplicate_search_proof(
-219
View File
@@ -1,219 +0,0 @@
"""Fail-closed reviewer worktree and local-git safety proofs (#233).
Reviewer sessions must never stash, reset, or otherwise manipulate unrelated
local changes from another session. When the active worktree has dirty tracked
files outside the PR scope, the workflow must stop or switch to a disposable
scratch worktree (``scripts/worktree-review``).
"""
from __future__ import annotations
import re
import shlex
# Subcommands that mutate unrelated local state — forbidden for reviewers.
_FORBIDDEN_REVIEWER_GIT = re.compile(
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
r"(?:stash(?:\s+(?:push|pop|drop|apply|clear|list))?|"
r"checkout\s+--|"
r"restore\s+|"
r"reset(?:\s+(?:--hard|--soft|--mixed|--merge))?|"
r"clean(?:\s+(?:-f|-fd|-fdx|-x|-d|-n))*|"
r"cherry-pick|"
r"rebase|"
r"merge|"
r"commit(?:\s+(?:--amend|-a|-am))?|"
r"push(?:\s+(?:--force|--force-with-lease))?|"
r"branch\s+-[dD]|"
r"worktree\s+remove)",
re.IGNORECASE,
)
# Read-only git operations reviewers may use for validation.
_READONLY_REVIEWER_GIT = re.compile(
r"\bgit\b(?:\s+(?:-C\s+\S+\s+)?)?"
r"(?:fetch|status|diff|log|show|rev-parse|branch(?:\s+--show-current)?|"
r"worktree\s+list|worktree\s+add)",
re.IGNORECASE,
)
def parse_dirty_tracked_files(porcelain: str) -> list[str]:
"""Return tracked paths with local modifications from ``git status --porcelain``.
Untracked entries (``??``) are ignored — they do not block reviewer work
when a scratch worktree is used, and authors may have unrelated untracked
files without implying reviewer interference.
"""
paths: list[str] = []
for line in (porcelain or "").splitlines():
if not line or len(line) < 4:
continue
if line.startswith("??"):
continue
path = line[3:].strip()
if " -> " in path:
path = path.split(" -> ", 1)[1].strip()
if path:
paths.append(path)
return paths
def files_outside_pr_scope(
dirty_files: list[str] | None,
pr_scope_files: list[str] | None,
) -> list[str]:
"""Dirty tracked files not explained by the PR diff file set."""
dirty = [p for p in (dirty_files or []) if p]
scope = {p for p in (pr_scope_files or []) if p}
if not dirty:
return []
if not scope:
return list(dirty)
return [path for path in dirty if path not in scope]
def is_forbidden_reviewer_git_command(command: str) -> bool:
"""True when a shell command would mutate unrelated local/remote git state."""
text = (command or "").strip()
if not text:
return False
return bool(_FORBIDDEN_REVIEWER_GIT.search(text))
def is_readonly_reviewer_git_command(command: str) -> bool:
"""True when the command is an explicitly allowed read-only git operation."""
text = (command or "").strip()
if not text:
return False
if is_forbidden_reviewer_git_command(text):
return False
return bool(_READONLY_REVIEWER_GIT.search(text))
def assess_reviewer_git_command_log(commands: list[str] | None) -> dict:
"""Fail closed when reviewer shell history includes forbidden git mutations."""
forbidden = [
cmd for cmd in (commands or []) if is_forbidden_reviewer_git_command(cmd)
]
if forbidden:
return {
"proven": False,
"block": True,
"forbidden_commands": forbidden,
"reasons": [
"reviewer workflow executed forbidden local git mutation: "
f"{cmd!r}"
for cmd in forbidden
],
"safe_next_action": (
"stop; report worktree interference; do not stash/reset/checkout "
"unrelated files — use scripts/worktree-review instead"
),
}
return {
"proven": True,
"block": False,
"forbidden_commands": [],
"reasons": [],
"safe_next_action": "proceed",
}
def assess_reviewer_worktree_proof(proof: dict | None) -> dict:
"""Evaluate reviewer worktree safety before checkout/diff/validation/review.
*proof* keys:
- ``worktree_path`` (required)
- ``porcelain_status`` or ``dirty_files``
- ``pr_scope_files`` (paths in the PR diff)
- ``scratch_used`` (bool)
- ``scratch_path`` (when scratch_used)
- ``git_commands`` (shell commands executed this session)
- ``unrelated_mutations_claimed`` (bool) — stash/reset/drop reported
"""
proof = dict(proof or {})
reasons: list[str] = []
worktree_path = (proof.get("worktree_path") or "").strip()
if not worktree_path:
reasons.append("reviewer worktree path not reported; fail closed")
if proof.get("dirty_files") is not None:
dirty_files = list(proof.get("dirty_files") or [])
else:
dirty_files = parse_dirty_tracked_files(proof.get("porcelain_status") or "")
pr_scope = list(proof.get("pr_scope_files") or [])
unrelated = files_outside_pr_scope(dirty_files, pr_scope)
scratch_used = bool(proof.get("scratch_used"))
scratch_path = (proof.get("scratch_path") or "").strip()
is_dirty = bool(dirty_files)
unrelated_dirty = bool(unrelated)
if unrelated_dirty and not scratch_used:
reasons.append(
"worktree has dirty tracked files outside PR scope "
f"({', '.join(unrelated)}); stop or use a scratch worktree"
)
if scratch_used and not scratch_path:
reasons.append(
"scratch worktree was used but scratch_path was not reported"
)
if proof.get("unrelated_mutations_claimed"):
reasons.append(
"reviewer reported stash/reset/checkout cleanup of unrelated "
"local changes; this is forbidden"
)
command_assessment = assess_reviewer_git_command_log(
list(proof.get("git_commands") or [])
)
if command_assessment["block"]:
reasons.extend(command_assessment["reasons"])
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"worktree_path": worktree_path or None,
"is_dirty": is_dirty,
"dirty_files": dirty_files,
"unrelated_dirty_files": unrelated,
"scratch_used": scratch_used,
"scratch_path": scratch_path or None,
"unrelated_mutations_avoided": not bool(
proof.get("unrelated_mutations_claimed")
or command_assessment.get("forbidden_commands")
),
"safe_next_action": (
"proceed"
if proven
else command_assessment.get("safe_next_action")
or "stop; use scripts/worktree-review or report dirty worktree"
),
"forbidden_commands": command_assessment.get("forbidden_commands", []),
}
def assess_author_worktree_continuity(proof: dict | None) -> dict:
"""Authors may keep dirty feature worktrees; reviewers may not manipulate them.
This helper only proves the task role is author when dirty unrelated files
exist — it does not grant reviewers an exception.
"""
proof = dict(proof or {})
role = (proof.get("task_role") or "").strip().lower()
dirty_files = list(proof.get("dirty_files") or [])
if role == "author" and dirty_files:
return {
"allowed": True,
"reasons": [
"author task may continue with dirty tracked files in its own "
"worktree; reviewer interference rules do not apply"
],
}
if role == "reviewer" and dirty_files:
return assess_reviewer_worktree_proof(proof)
return {"allowed": True, "reasons": []}
@@ -17,24 +17,9 @@ Repo name disambiguation (Gitea-Tools blind review hardening):
configured repos were not checked. This is not a complete queue inventory."
- A single-repo "no open PRs" result MUST NOT be reported as global "no open PRs"
if the other configured repo was not inventoried.
- PR inventory trust gate (#196): before reporting "no open PRs" or "queue empty",
the workflow must run `pr_inventory_trust_gate` (via the live inventory path or
`review_proofs.assess_reviewer_queue_inventory`). Only `trusted_empty` allows a
clean empty-queue stop. Report `pr_inventory_trust_gate.status`, reasons, and
corroboration in the final report. A bare `[]` from `gitea_list_prs` is never
sufficient proof.
Rules (llm-project-workflow):
- 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
tracked files exist outside the PR scope, STOP or run
`scripts/worktree-review <pr-head-branch>` and validate in the scratch path.
NEVER run `git stash`, `git stash pop/drop`, `git checkout --`, `git reset`,
or `git clean` to manage another session's dirty files.
- Final report must state: Worktree path, Worktree dirty (yes/no),
Scratch worktree used (yes/no + path if yes), and confirm no unrelated local
files were modified, stashed, reset, or dropped.
- You must NOT be the PR author. If the authenticated user == PR author, stop.
A different LLM-Agent-SHA does NOT make you a different actor — only a
different authenticated Gitea user does (docs/llm-agent-sha.md).
+13 -49
View File
@@ -65,20 +65,6 @@ CREATE_PR_ENV = {
),
}
ISSUE_LOCK_FILE = "/tmp/gitea_issue_lock.json"
def _sample_issue_lock(issue_number=123, branch_name="feat/x", **overrides):
record = {
"issue_number": issue_number,
"branch_name": branch_name,
"remote": "dadeschools",
"org": "Scaled-Tech-Consulting",
"repo": "Gitea-Tools",
}
record.update(overrides)
return record
# ---------------------------------------------------------------------------
# Create Issue
@@ -141,19 +127,15 @@ class TestCreatePR(unittest.TestCase):
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_creates_pr(self, mock_open, mock_exists, _auth, mock_api, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertEqual(result["number"], 3)
self.assertNotIn("url", result)
mock_exists.assert_called_with(ISSUE_LOCK_FILE)
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
payload = mock_api.call_args[0][3]
self.assertEqual(payload["head"], "feat/x")
self.assertEqual(payload["base"], "main")
self.assertIn("Closes #123", payload["title"])
@patch("mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
return_value=(True, []))
@@ -162,28 +144,13 @@ class TestCreatePR(unittest.TestCase):
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_reveal_opt_in_includes_url(self, mock_open, mock_exists, _auth, mock_api, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
mock_open.return_value.__enter__.return_value.read.return_value = '{"issue_number": 123, "branch_name": "feat/x"}'
mock_api.return_value = {"number": 3, "html_url": "https://example.com/pulls/3"}
env = {**CREATE_PR_ENV, "GITEA_MCP_REVEAL_ENDPOINTS": "1"}
with patch.dict(os.environ, env, clear=True):
result = gitea_create_pr(title="feat: X Closes #123", head="feat/x", base="main")
self.assertIn("pulls/3", result["url"])
@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)
@patch("os.path.exists", return_value=True)
@patch("builtins.open")
def test_create_pr_locked_issue_mismatch_fails(self, mock_open, mock_exists, _auth, _role):
lock_json = json.dumps(_sample_issue_lock(issue_number=123, branch_name="feat/x"))
mock_open.return_value.__enter__.return_value.read.return_value = lock_json
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #999", head="feat/x", base="main")
self.assertIn("Closes #123", str(ctx.exception))
mock_open.assert_called_with(ISSUE_LOCK_FILE, "r", encoding="utf-8")
# ---------------------------------------------------------------------------
# Close Issue
@@ -2803,8 +2770,8 @@ class TestIssueLocking(unittest.TestCase):
"""Test issue locking and PR gating constraints."""
def tearDown(self):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
@patch("mcp_server.api_get_all")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -2812,7 +2779,7 @@ class TestIssueLocking(unittest.TestCase):
mock_api.return_value = [] # no open PRs
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
self.assertTrue(res["success"])
self.assertTrue(os.path.exists(ISSUE_LOCK_FILE))
self.assertTrue(os.path.exists("/tmp/gitea_issue_lock.json"))
def test_lock_issue_mismatch_branch_fails(self):
with self.assertRaises(ValueError) as ctx:
@@ -2849,8 +2816,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_lock_fails(self, _auth, _role):
if os.path.exists(ISSUE_LOCK_FILE):
os.remove(ISSUE_LOCK_FILE)
if os.path.exists("/tmp/gitea_issue_lock.json"):
os.remove("/tmp/gitea_issue_lock.json")
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(RuntimeError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-mutations", remote="prgs")
@@ -2860,9 +2827,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_branch_mismatch_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X Closes #196", head="feat/issue-196-different", remote="prgs")
@@ -2872,9 +2838,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_forbidden_terms_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
for term in ("equivalent to #196", "related to #196", "same as #196"):
with self.assertRaises(ValueError) as ctx:
@@ -2885,9 +2850,8 @@ class TestIssueLocking(unittest.TestCase):
return_value=(True, []))
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_create_pr_missing_closes_ref_fails(self, _auth, _role):
with open(ISSUE_LOCK_FILE, "w", encoding="utf-8") as f:
json.dump(_sample_issue_lock(
issue_number=196, branch_name="feat/issue-196-mutations"), f)
with open("/tmp/gitea_issue_lock.json", "w", encoding="utf-8") as f:
json.dump({"issue_number": 196, "branch_name": "feat/issue-196-mutations"}, f)
with patch.dict(os.environ, CREATE_PR_ENV, clear=True):
with self.assertRaises(ValueError) as ctx:
gitea_create_pr(title="feat: X refs #196", head="feat/issue-196-mutations", remote="prgs")
-181
View File
@@ -1,181 +0,0 @@
"""Tests for operation-scoped role selection and automatic dispatch switching (#228)."""
import os
import sys
import json
import tempfile
import unittest
from unittest.mock import patch, MagicMock
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_config
import gitea_auth
import mcp_server
from reviewer_worktree import assess_author_worktree_continuity
CONFIG_TEST = {
"version": 2,
"contexts": {
"ctx": {
"enabled": True,
"gitea": {
"enabled": True,
"base_url": "https://gitea.example.com"
}
}
},
"profiles": {
"author-profile": {
"enabled": True,
"context": "ctx",
"role": "author",
"username": "author-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
"allowed_operations": ["gitea.read", "gitea.issue.create", "gitea.pr.create", "gitea.branch.push", "gitea.issue.comment"],
"forbidden_operations": ["gitea.pr.approve", "gitea.pr.merge"],
"execution_profile": "author-profile"
},
"reviewer-profile": {
"enabled": True,
"context": "ctx",
"role": "reviewer",
"username": "reviewer-user",
"auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
"allowed_operations": ["gitea.read", "gitea.pr.review", "gitea.pr.approve", "gitea.pr.merge", "gitea.issue.comment"],
"forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
"execution_profile": "reviewer-profile"
}
}
}
class TestOperationScopedRoles(unittest.TestCase):
def setUp(self):
self._remotes_patch = patch.dict(mcp_server.REMOTES, {
"dadeschools": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"},
"prgs": {"host": "gitea.example.com", "org": "Example-Org", "repo": "Example-Repo"}
})
self._remotes_patch.start()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
self._dir = tempfile.TemporaryDirectory()
self.config_path = os.path.join(self._dir.name, "profiles.json")
self._write_config(CONFIG_TEST)
def tearDown(self):
self._remotes_patch.stop()
mcp_server._IDENTITY_CACHE.clear()
gitea_config._active_profile_override = None
mcp_server._MUTATION_AUTHORITY = None
self._dir.cleanup()
def _write_config(self, obj):
with open(self.config_path, "w", encoding="utf-8") as fh:
fh.write(json.dumps(obj))
def _env(self, profile="author-profile", with_reviewer_token=True):
env = {
"GITEA_MCP_CONFIG": self.config_path,
"GITEA_MCP_PROFILE": profile,
"GITEA_TOKEN_AUTHOR": "author-pass",
}
if with_reviewer_token:
env["GITEA_TOKEN_REVIEWER"] = "reviewer-pass"
return env
@patch("mcp_server.api_request")
def test_auto_switch_to_reviewer(self, mock_api):
# mock identity resolution to return username matching profile
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
# initially we are author-profile
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
self.assertEqual(mcp_server.get_profile()["profile_name"], "author-profile")
# resolve a reviewer task (review_pr)
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
# verify it automatically switched to reviewer-profile
self.assertTrue(res["allowed_in_current_session"])
self.assertTrue(res["available_in_session"])
self.assertEqual(res["active_profile"], "reviewer-profile")
self.assertEqual(res["active_identity"], "reviewer-user")
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
@patch("mcp_server.api_request")
def test_auto_switch_to_author(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("reviewer-profile")):
# initially we are reviewer-profile
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
# resolve an author task (create_issue)
res = mcp_server.gitea_resolve_task_capability(task="create_issue", remote="prgs")
# verify it automatically switched to author-profile
self.assertTrue(res["allowed_in_current_session"])
self.assertTrue(res["available_in_session"])
self.assertEqual(res["active_profile"], "author-profile")
self.assertEqual(res["active_identity"], "author-user")
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
@patch("mcp_server.api_request")
def test_restart_required_when_unattached(self, mock_api):
mock_api.side_effect = lambda method, url, header: {"login": "author-user"}
# launch without reviewer token in env
with patch.dict(os.environ, self._env("author-profile", with_reviewer_token=False)):
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
# resolve a reviewer task (review_pr)
res = mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
# verify it did NOT switch and reports restart_required
self.assertFalse(res["allowed_in_current_session"])
self.assertFalse(res["available_in_session"])
self.assertTrue(res["configured"])
self.assertTrue(res["restart_required"])
self.assertTrue(res["stop_required"])
self.assertIn("Reviewer profile exists but MCP server was added after session startup and is not attached", res["reason"])
def test_author_continuity_dirty_worktree(self):
# author is allowed to keep dirty worktree
res = assess_author_worktree_continuity({
"task_role": "author",
"dirty_files": ["review_proofs.py"],
})
self.assertTrue(res["allowed"])
# reviewer is blocked by reviewer worktree proof
res2 = assess_author_worktree_continuity({
"task_role": "reviewer",
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
})
self.assertFalse(res2["proven"])
@patch("mcp_server.api_request")
def test_mutating_actions_auto_switch(self, mock_api):
mock_api.side_effect = lambda method, url, header: (
{"login": "reviewer-user"} if "reviewer-pass" in str(header) else {"login": "author-user"}
)
with patch.dict(os.environ, self._env("author-profile")):
# verify we are author-profile
self.assertEqual(gitea_config.selected_profile_name(), "author-profile")
# call a reviewer mutation check helper (like _profile_permission_block with reviewer permission)
blocked = mcp_server._profile_permission_block("gitea.pr.merge", remote="prgs")
self.assertIsNone(blocked) # should switch to reviewer-profile and allow it (no permission block)
# verify we dynamically switched to reviewer-profile
self.assertEqual(gitea_config.selected_profile_name(), "reviewer-profile")
if __name__ == "__main__":
unittest.main()
+1 -68
View File
@@ -261,79 +261,12 @@ class TestPRQueueInventory(unittest.TestCase):
for call in mock_api.call_args_list:
self.assertEqual(call.args[0], "GET")
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_empty_inventory_runs_trust_gate_and_blocks_without_trusted_empty(
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = None
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertFalse(result["success"])
msg = result["message"]
self.assertIn("pr_inventory_trust_gate.status: untrusted_empty", msg)
self.assertIn("Empty-queue claim blocked", msg)
self.assertNotIn("Open PRs found: 0 (trusted_empty)", msg)
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_empty_inventory_trusted_empty_when_gate_passes(
self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
)
result = gitea_review_pr(
pr_number=1,
event="APPROVE",
remote="prgs",
org="Scaled-Tech-Consulting",
repo="Gitea-Tools",
)
self.assertFalse(result["success"])
msg = result["message"]
self.assertIn("pr_inventory_trust_gate.status: trusted_empty", msg)
self.assertIn("Open PRs found: 0 (trusted_empty)", msg)
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all, mock_local_url):
def test_author_profiles_cannot_approve_request_changes_merge_or_bypass_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
"""Author profiles still cannot approve, request_changes, merge, or bypass gates even with inventory."""
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
)
for event in ["APPROVE", "REQUEST_CHANGES"]:
mock_get_profile.return_value = {
"profile_name": "gitea-author",
+18 -142
View File
@@ -26,7 +26,6 @@ from review_proofs import ( # noqa: E402
assess_capability_proof,
assess_controller_handoff,
assess_inventory_completeness,
assess_reviewer_queue_inventory,
assess_live_state_recheck,
assess_review_mutation_final_report,
assess_role_boundary,
@@ -184,24 +183,6 @@ def _good_review_mutation():
return assess_review_mutation_final_report(report, lock)
def _good_worktree(**overrides):
proof = {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
"git_commands": [
"git fetch prgs master feat/issue-224-wiki-proof-refresh",
"git diff prgs/master...prgs/feat/issue-224-wiki-proof-refresh",
],
}
proof.update(overrides)
from reviewer_worktree import assess_reviewer_worktree_proof # noqa: E402
return assess_reviewer_worktree_proof(proof)
def _good_role_boundary_179(**overrides):
kwargs = {
"task_role": "reviewer",
@@ -614,13 +595,6 @@ class TestFinalReport(unittest.TestCase):
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
"worktree_proof": {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
},
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -925,17 +899,12 @@ class TestControllerHandoff(unittest.TestCase):
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Pinned reviewed head", result["missing_fields"])
self.assertIn("Worktree path", result["missing_fields"])
self.assertIn("Merge result", result["missing_fields"])
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Worktree path: /repo/branches/review-pr-999",
"- Worktree dirty: no",
"- Scratch worktree used: yes (/repo/branches/review-pr-999)",
"- Unrelated local mutations: none",
"- Review decision: approve",
"- Merge result: merged",
"- Linked issue status: closed",
@@ -944,46 +913,24 @@ class TestControllerHandoff(unittest.TestCase):
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def _author_role_fields(self, issue_number=182, pr_number=999):
return [
f"- Selected issue: #{issue_number}",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
f"- PR number opened: #{pr_number}",
"- No review/merge: confirmed",
]
def test_handoff_role_fields_author_includes_issue_lock_proof(self):
from review_proofs import HANDOFF_ROLE_FIELDS
names = [name for name, _ in HANDOFF_ROLE_FIELDS["author"]]
self.assertIn("Issue lock proof", names)
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join(self._author_role_fields())
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_requires_issue_lock_proof(self):
without_lock = self.BASE_HANDOFF + "\n" + "\n".join([
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
result = assess_controller_handoff(without_lock, role="author")
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Issue lock proof", result["missing_fields"])
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_author_role_rejects_equivalent_or_multiple_issues(self):
# 1. equivalent reference blocked
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: Issue #194 / #196 equivalent",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -995,7 +942,6 @@ class TestControllerHandoff(unittest.TestCase):
# 2. multiple issues blocked
incomplete_multi = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #194, #196",
"- Issue lock proof: lock before diff on feat/x @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
@@ -1007,7 +953,6 @@ class TestControllerHandoff(unittest.TestCase):
def test_author_role_rejects_fuzzy_pr_number(self):
incomplete_pr = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Issue lock proof: lock before diff on feat/issue-196 @ master",
"- Claim/comment status: comment-claimed",
"- PR number opened: PR #203 / #204 equivalent",
"- No review/merge: confirmed",
@@ -1039,17 +984,23 @@ class TestControllerHandoff(unittest.TestCase):
def test_handoff_rejects_none_workspace_mutations_when_local_edits_exist(self):
# 1. Workspace mutations: none is rejected when local_edits is True
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join(
self._author_role_fields(issue_number=196, pr_number=203))
incomplete_eq = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res = assess_controller_handoff(incomplete_eq, role="author", local_edits=True)
self.assertEqual(res["verdict"], "incomplete")
self.assertIn("Workspace mutations", res["missing_fields"])
# 2. Workspace mutations: edited files is allowed when local_edits is True
complete_eq = self.BASE_HANDOFF.replace(
"- Workspace mutations: none",
"- Workspace mutations: edited review_proofs.py",
) + "\n" + "\n".join(self._author_role_fields(issue_number=196, pr_number=203))
complete_eq = self.BASE_HANDOFF.replace("- Workspace mutations: none", "- Workspace mutations: edited review_proofs.py") + "\n" + "\n".join([
"- Selected issue: #196",
"- Claim/comment status: comment-claimed",
"- PR number opened: #203",
"- No review/merge: confirmed",
])
res2 = assess_controller_handoff(complete_eq, role="author", local_edits=True)
self.assertEqual(res2["verdict"], "complete")
@@ -1212,74 +1163,6 @@ class TestPRInventoryTrustGate(unittest.TestCase):
self.assertTrue(res["corroborated"])
class TestAssessReviewerQueueInventory(unittest.TestCase):
"""Issue #196: trust gate wired into canonical queue inventory."""
def _repo_report(self, **overrides):
report = {
"repo": "Scaled-Tech-Consulting/Gitea-Tools",
"state_filter": "open",
"pagination_complete": True,
"open_pr_count": 0,
"list_prs_response": [],
"remote": "prgs",
"authenticated_profile": {
"profile_name": "prgs-reviewer",
"allowed_operations": ["read", "gitea.read"],
},
"local_remote_url": (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
),
}
report.update(overrides)
return report
def test_empty_without_corroboration_blocks_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(pagination_complete=False),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=0,
pagination_complete=False,
),
])
self.assertFalse(result["can_claim_empty_queue"])
self.assertIn("untrusted_empty", str(result["trust_gates"]))
def test_trusted_empty_with_finality_allows_empty_queue_claim(self):
result = assess_reviewer_queue_inventory([
self._repo_report(),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
local_remote_url=(
"https://gitea.prgs.cc/Scaled-Tech-Consulting/"
"mcp-control-plane.git"
),
),
])
self.assertTrue(result["can_claim_empty_queue"])
self.assertTrue(result["can_claim_exhaustive"])
def test_user_context_indicating_open_prs_blocks_empty_claim(self):
result = assess_reviewer_queue_inventory(
[self._repo_report()],
user_context="please review open PR #195 in the queue",
)
self.assertFalse(result["can_claim_empty_queue"])
self.assertTrue(result["blockers"])
def test_nonempty_inventory_skips_empty_trust_gate_block(self):
result = assess_reviewer_queue_inventory([
self._repo_report(open_pr_count=2),
self._repo_report(
repo="Scaled-Tech-Consulting/mcp-control-plane",
open_pr_count=1,
),
])
self.assertTrue(result["complete"])
self.assertEqual(result["trust_gates"], {})
class TestCapabilityEvidence(unittest.TestCase):
"""#179 gap 1: capability claims need exact evidence."""
@@ -1423,13 +1306,6 @@ class TestFinalReport179Bar(unittest.TestCase):
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
"worktree_proof": {
"worktree_path": "/repo/branches/review-feat-issue-224",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
},
}
kwargs.update(overrides)
return build_final_report(**kwargs)
-166
View File
@@ -1,166 +0,0 @@
"""Tests for reviewer worktree safety proofs (Issue #233)."""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
from reviewer_worktree import ( # noqa: E402
assess_author_worktree_continuity,
assess_reviewer_git_command_log,
assess_reviewer_worktree_proof,
files_outside_pr_scope,
is_forbidden_reviewer_git_command,
is_readonly_reviewer_git_command,
parse_dirty_tracked_files,
)
class TestParseDirtyTrackedFiles(unittest.TestCase):
def test_ignores_untracked_files(self):
porcelain = "?? untracked.txt\n M tracked.py\n"
self.assertEqual(parse_dirty_tracked_files(porcelain), ["tracked.py"])
def test_parses_renamed_paths(self):
porcelain = "R old.py -> new.py\n"
self.assertEqual(parse_dirty_tracked_files(porcelain), ["new.py"])
class TestFilesOutsidePrScope(unittest.TestCase):
def test_all_dirty_in_scope_is_clean(self):
self.assertEqual(
files_outside_pr_scope(
["docs/wiki/Repositories.md"],
["docs/wiki/Repositories.md"],
),
[],
)
def test_unrelated_dirty_files_detected(self):
self.assertEqual(
files_outside_pr_scope(
["review_proofs.py", "docs/wiki/Repositories.md"],
["docs/wiki/Repositories.md"],
),
["review_proofs.py"],
)
class TestForbiddenGitCommands(unittest.TestCase):
def test_blocks_stash_operations(self):
self.assertTrue(is_forbidden_reviewer_git_command("git stash"))
self.assertTrue(
is_forbidden_reviewer_git_command(
'git stash push -m "reviewer-temp-stash" -- tests/test_mcp_server.py'
)
)
self.assertTrue(is_forbidden_reviewer_git_command("git stash pop"))
self.assertTrue(is_forbidden_reviewer_git_command("git stash drop"))
def test_blocks_checkout_reset_and_clean(self):
self.assertTrue(
is_forbidden_reviewer_git_command(
"git checkout -- review_proofs.py tests/test_mcp_server.py"
)
)
self.assertTrue(is_forbidden_reviewer_git_command("git reset --hard"))
self.assertTrue(is_forbidden_reviewer_git_command("git clean -fd"))
def test_allows_readonly_commands(self):
for cmd in (
"git fetch prgs master",
"git status --porcelain",
"git diff prgs/master...HEAD",
"git rev-parse HEAD",
"git -C /repo log -1",
):
with self.subTest(cmd=cmd):
self.assertFalse(is_forbidden_reviewer_git_command(cmd))
self.assertTrue(is_readonly_reviewer_git_command(cmd))
class TestAssessReviewerWorktreeProof(unittest.TestCase):
def test_clean_worktree_proceeds(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo/branches/review-pr-231",
"porcelain_status": "",
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
"git_commands": ["git fetch prgs master", "git diff prgs/master...HEAD"],
})
self.assertTrue(result["proven"])
self.assertFalse(result["block"])
def test_dirty_unrelated_without_scratch_blocks(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py", "tests/test_review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": False,
})
self.assertFalse(result["proven"])
self.assertTrue(result["block"])
self.assertIn("review_proofs.py", result["unrelated_dirty_files"][0])
def test_scratch_worktree_allows_dirty_main_repo(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo",
"dirty_files": ["review_proofs.py"],
"pr_scope_files": ["docs/wiki/Repositories.md"],
"scratch_used": True,
"scratch_path": "/repo/branches/review-feat-issue-224",
})
self.assertTrue(result["proven"])
def test_forbidden_command_history_blocks(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo/branches/review-pr-231",
"porcelain_status": "",
"git_commands": ["git stash push -m temp -- review_proofs.py"],
})
self.assertFalse(result["proven"])
self.assertTrue(result["forbidden_commands"])
def test_unrelated_mutations_claimed_blocks(self):
result = assess_reviewer_worktree_proof({
"worktree_path": "/repo",
"porcelain_status": "",
"unrelated_mutations_claimed": True,
})
self.assertFalse(result["proven"])
class TestAuthorContinuity(unittest.TestCase):
def test_author_may_keep_dirty_worktree(self):
result = assess_author_worktree_continuity({
"task_role": "author",
"dirty_files": ["feat.py"],
})
self.assertTrue(result["allowed"])
def test_reviewer_dirty_worktree_uses_reviewer_gate(self):
result = assess_author_worktree_continuity({
"task_role": "reviewer",
"worktree_path": "/repo",
"dirty_files": ["other.py"],
"pr_scope_files": ["docs/a.md"],
"scratch_used": False,
})
self.assertFalse(result["proven"])
class TestAssessReviewerGitCommandLog(unittest.TestCase):
def test_empty_log_is_clean(self):
result = assess_reviewer_git_command_log([])
self.assertTrue(result["proven"])
def test_mixed_log_blocks_on_forbidden(self):
result = assess_reviewer_git_command_log([
"git fetch prgs",
"git stash",
])
self.assertFalse(result["proven"])
self.assertEqual(len(result["forbidden_commands"]), 1)
if __name__ == "__main__":
unittest.main()