Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

fix: resolve conflicts for issue 330

Browse Source 
Merge prgs/master into feat/issue-330-proof-wording-enforcement and keep
both assess_proof_wording (#330) and git ref mutation reporting (#297).

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
sysadmin
2026-07-07 05:16:57 -04:00
co-authored by Claude Opus 4.8
parent 46703eac3f 407483f6e0
commit 94b022b24d
14 changed files with 1441 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture/glitchtip-to-gitea-workflow-design.md
+38 -8
View File
@@ -1,21 +1,34 @@
# GlitchTip-to-Gitea Issue Filing Workflow Design
# GlitchTip-to-Gitea Issue Filing Workflow Contract
- **Status:** Design (no implementation in this repo)
- **Issue:** #74 (parent umbrella: #75)
- **Status:** Contract for the library-only implementation in
`Scaled-Tech-Consulting/mcp-control-plane` (#57)
- **Issue:** #153 (supersedes #74/#78 as the Gitea-Tools tracker)
- **Related:** #78 (deduplication design, child of #74)
- **Date:** 2026-07-02
- **Date:** 2026-07-07
## 1. Boundary and Orchestration
* **GlitchTip-to-Gitea filing is NOT a GlitchTip MCP capability.** The `glitchtip-mcp` boundary remains strictly read-only per ADR-0001.
* The filing capability lives in an **orchestrator / runbook / release workflow**. It **composes** separate GlitchTip **read** tools (from the `glitchtip-mcp` server) and Gitea **issue** tools (from the `gitea-mcp` server).
* The filing capability lives in a **library-only orchestrator** in
`mcp-control-plane` and is not exposed through `glitchtip-mcp`.
* The orchestrator composes the real GlitchTip read path with Gitea
issue-write tools. It must not use mocked GlitchTip issue data in production
filing paths.
* The orchestrator **must not centralize credentials** into a single server. The GlitchTip MCP holds only GlitchTip tokens, and the Gitea MCP holds only Gitea tokens.
* If a future MCP surface exposes filing, it must be a separate write-boundary
server/profile with explicit Gitea issue-write permission and the same audit
gates. It must not be added to the read-only `glitchtip-mcp` surface.
## 2. Invocation and Safety
* **Explicit invocation only:** There is no automatic, unsupervised filing in phase 1. A human or an explicitly-triggered automation must initiate the workflow.
* **Dry-run / Preview required:** The orchestrator must present a preview of the drafted Gitea issue (title, body, labels) and obtain explicit confirmation before calling the Gitea mutation tool to file the issue.
* **Gitea Profile Checks & Audit Logging:** The actual Gitea issue creation relies on `gitea-mcp`, and therefore inherently subjects the mutation to Gitea profile checks and audit logging as standard.
* **Gitea Profile Checks & Audit Logging:** The actual Gitea issue creation
relies on `gitea-mcp`, and therefore must pass Gitea profile checks and
fail-closed mutation audit before create/link/comment actions.
* **Deduplication before create:** The orchestrator must run the
GlitchTip-to-Gitea dedup/linking logic before any Gitea create action. Create,
link, and skip decisions must be represented in tests.
## 3. Gitea Issue Format
@@ -51,6 +64,23 @@ To prevent PII or secret leakage into Gitea, the orchestrator and the underlying
The principle is: **"Link, don't dump"**. The generated issue acts as an alert/pointer, while the raw context remains protected inside GlitchTip.
## 5. Deduplication and Linking (Deferred)
## 5. Deduplication and Linking
Deduplication logic (e.g. searching existing Gitea issues, managing GlitchTip issue IDs, and race condition handling) is specifically handled by **Issue #78** and will augment this design.
Deduplication logic (e.g. searching existing Gitea issues, managing GlitchTip
issue IDs, and race condition handling) is integrated into the library-only
filing orchestrator in `mcp-control-plane`. The orchestrator must reuse that
dedup/linking path instead of duplicating or bypassing it.
## 6. Gitea-Tools Acceptance Contract for #153
Gitea-Tools does not host the filing implementation. This repository owns the
operator-facing contract:
* `glitchtip-mcp` remains read-only and exposes only GlitchTip inspection tools.
* Filing is implemented in `mcp-control-plane`, not in this Gitea MCP runtime.
* Filing uses the real GlitchTip read path, not mocked issue data.
* Dedup runs before any Gitea create action.
* Create/link/skip decisions and audit failure are covered by orchestrator tests.
* Mutation audit fails closed before any Gitea create, link, or comment mutation.
* Any future MCP exposure requires a separate write-boundary server/profile and
must not add Gitea write credentials to `glitchtip-mcp`.
docs/mcp-client-registration.md
+9 -2
View File
@@ -95,7 +95,14 @@ back to shell commands, raw service APIs, or unrelated MCP servers.
- Do not register `jenkins-write-mcp` until an operator approves a trigger
profile; no shipped profile carries trigger capability by default.
- `glitchtip-mcp` remains read-only. It must not file or mutate Gitea issues.
- GlitchTip-to-Gitea filing is a separate orchestrator that composes GlitchTip
read tools with Gitea issue-write tools.
- GlitchTip-to-Gitea filing is a separate library-only orchestrator in
`mcp-control-plane` that composes the real GlitchTip read path with Gitea
issue-write tools.
- The filing orchestrator must run GlitchTip/Gitea dedup before any Gitea
create action, and its create/link/skip decisions must be covered by tests.
- The filing orchestrator must fail closed on mutation-audit failure before
any Gitea create, link, or comment mutation.
- If filing is ever MCP-exposed, it must use a separate write-boundary
server/profile. It must not be exposed by `glitchtip-mcp`.
- Gitea credentials never enter Jenkins or GlitchTip runtimes.
- Jenkins and GlitchTip credentials never enter the Gitea MCP runtime.
gitea_auth.py
+37
View File
@@ -420,6 +420,43 @@ def api_get_all(url, auth_header, *, limit=None, page_size=50, max_pages=100,
return results
def api_fetch_page(url, auth_header, *, page=1, limit=50, **kwargs):
"""Fetch one page from a Gitea list endpoint with explicit pagination metadata.
Returns ``(items, pagination)`` where *pagination* includes ``has_more``,
``next_page``, and ``is_final_page`` derived from the returned page length.
"""
page = max(1, int(page))
limit = max(1, min(50, int(limit)))
page_url = _add_query(url, page=page, limit=limit)
data = api_request("GET", page_url, auth_header, **kwargs)
if data is None:
pagination = {
"page": page,
"per_page": limit,
"returned_count": 0,
"has_more": False,
"next_page": None,
"is_final_page": True,
}
return [], pagination
if not isinstance(data, list):
raise RuntimeError(
f"expected a list page from Gitea, got {type(data).__name__}"
)
returned = len(data)
has_more = returned >= limit
pagination = {
"page": page,
"per_page": limit,
"returned_count": returned,
"has_more": has_more,
"next_page": page + 1 if has_more else None,
"is_final_page": not has_more,
}
return data, pagination
def gitea_url(host, path):
"""Build a full URL for *host* and *path*, using http for loopback and https for others."""
if not path.startswith("/"):
gitea_mcp_server.py
+213 -28
View File
@@ -461,6 +461,7 @@ from gitea_auth import ( # noqa: E402
get_auth_header,
api_request,
api_get_all,
api_fetch_page,
repo_api_url,
get_profile,
gitea_url,
@@ -1119,6 +1120,56 @@ def gitea_create_pr(
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
def _format_list_pr_entry(pr: dict) -> dict:
head_obj = pr.get("head") or {}
base_obj = pr.get("base") or {}
entry = {
"number": pr["number"],
"title": pr["title"],
"state": pr["state"],
"head": head_obj.get("ref") if isinstance(head_obj, dict) else head_obj,
"base": base_obj.get("ref") if isinstance(base_obj, dict) else base_obj,
"mergeable": pr.get("mergeable"),
"updated_at": pr.get("updated_at"),
"head_sha": head_obj.get("sha") if isinstance(head_obj, dict) else None,
"author": (pr.get("user") or {}).get("login"),
"labels": [
label["name"]
for label in pr.get("labels", [])
if isinstance(label, dict) and label.get("name")
],
"body": pr.get("body") or "",
}
return _with_optional_url(entry, pr.get("html_url"))
def _list_prs_pagination_envelope(
prs: list[dict],
*,
page: int,
per_page: int,
pages_fetched: int,
is_final_page: bool,
has_more: bool,
next_page: int | None,
inventory_complete: bool,
) -> dict:
return {
"prs": prs,
"pagination": {
"page": page,
"per_page": per_page,
"returned_count": len(prs),
"has_more": has_more,
"next_page": next_page,
"is_final_page": is_final_page,
"pages_fetched": pages_fetched,
"inventory_complete": inventory_complete,
"total_count": len(prs) if inventory_complete else None,
},
}
@mcp.tool()
def gitea_list_prs(
state: str = "open",
@@ -1126,7 +1177,9 @@ def gitea_list_prs(
host: str | None = None,
org: str | None = None,
repo: str | None = None,
) -> list[dict]:
page: int | None = None,
per_page: int = 50,
) -> dict:
"""List pull requests on a Gitea repository.
Args:
@@ -1135,11 +1188,16 @@ def gitea_list_prs(
host: Override the Gitea host.
org: Override the owner/organization.
repo: Override the repository name.
page: Optional 1-based page for explicit single-page fetch. When
omitted, all pages are traversed and ``inventory_complete`` is set.
per_page: Page size (1–50; Gitea caps at 50).
Returns:
List of dicts with 'number', 'title', 'state', 'head', 'base',
'mergeable', 'updated_at' ('url' only with the reveal opt-in).
The additional 'updated_at' aids stale/conflicting queue detection.
Dict with ``prs`` (list of PR summaries) and ``pagination`` metadata
(``has_more``, ``next_page``, ``is_final_page``, ``inventory_complete``,
``pages_fetched``, ``total_count`` when complete). Each PR entry carries
``number``, ``title``, ``state``, ``head``, ``base``, ``mergeable``,
``updated_at`` ('url' only with the reveal opt-in).
"""
allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
"list_prs"
@@ -1149,20 +1207,48 @@ def gitea_list_prs(
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
prs = api_get_all(url, auth)
results = []
for pr in prs:
entry = {
"number": pr["number"],
"title": pr["title"],
"state": pr["state"],
"head": pr["head"]["ref"],
"base": pr["base"]["ref"],
"mergeable": pr.get("mergeable"),
"updated_at": pr.get("updated_at"),
}
results.append(_with_optional_url(entry, pr.get("html_url")))
return results
if page is not None:
raw_page, pagination = api_fetch_page(
url, auth, page=page, limit=per_page
)
prs = [_format_list_pr_entry(pr) for pr in raw_page]
return _list_prs_pagination_envelope(
prs,
page=pagination["page"],
per_page=pagination["per_page"],
pages_fetched=1,
is_final_page=pagination["is_final_page"],
has_more=pagination["has_more"],
next_page=pagination["next_page"],
inventory_complete=False,
)
all_raw: list[dict] = []
current_page = 1
pages_fetched = 0
last_meta: dict = {}
while pages_fetched < 100:
raw_page, last_meta = api_fetch_page(
url, auth, page=current_page, limit=per_page
)
pages_fetched += 1
all_raw.extend(raw_page)
if last_meta["is_final_page"]:
break
current_page += 1
prs = [_format_list_pr_entry(pr) for pr in all_raw]
return _list_prs_pagination_envelope(
prs,
page=1,
per_page=per_page,
pages_fetched=pages_fetched,
is_final_page=bool(last_meta.get("is_final_page")),
has_more=False,
next_page=None,
inventory_complete=bool(last_meta.get("is_final_page")),
)
@mcp.tool()
@@ -1678,6 +1764,51 @@ def record_live_review_mutation(pr_number: int, action: str, review_id: int | No
_save_review_decision_lock(lock)
def terminal_review_hard_stop_reasons(pr_number: int, operation: str) -> list[str]:
"""Session hard-stop after a terminal live review mutation (#332).
After a terminal verdict is consumed in this run, the only permitted
continuation is the merge sequence for the same PR that was approved.
A REQUEST_CHANGES (or an approval of a different PR) blocks every
further review/mark-ready/merge mutation. An operator-approved
correction (#211) re-opens the review path only — never a cross-PR
merge.
*operation* is one of 'merge', 'mark_ready', or 'review'.
Returns [] when the operation may proceed.
"""
lock = _load_review_decision_lock()
if lock is None:
return []
terminals = [
m for m in (lock.get("live_mutations") or [])
if m.get("action") in _TERMINAL_REVIEW_ACTIONS
]
if not terminals:
return []
last = terminals[-1]
if (
operation == "merge"
and last.get("action") == "approve"
and last.get("pr_number") == pr_number
):
return []
if operation in ("mark_ready", "review") and lock.get("correction_authorized"):
return []
if last.get("action") == "approve":
guidance = (
f"only the merge sequence for approved PR "
f"#{last.get('pr_number')} may continue"
)
else:
guidance = "the session must stop and produce a final report"
return [
"terminal review mutation already consumed in this run "
f"({last.get('action')} on PR #{last.get('pr_number')}); {guidance} "
"(fail closed, #332)"
]
# Patterns scrubbed from any surfaced error text so a credential can never leak.
_SECRET_PREFIXES = ("token ", "Basic ")
@@ -2093,6 +2224,9 @@ def gitea_mark_final_review_decision(
}
org = resolved_org
repo = resolved_repo
hard_stop = terminal_review_hard_stop_reasons(pr_number, "mark_ready")
if hard_stop:
return {"marked_ready": False, "reasons": hard_stop}
if lock.get("live_mutations") and not lock.get("correction_authorized"):
return {
"marked_ready": False,
@@ -2110,6 +2244,33 @@ def gitea_mark_final_review_decision(
f"{sorted(_REVIEW_ACTIONS)}"
],
}
if action == "request_changes":
# Duplicate request-changes suppression (#332): an unresolved
# REQUEST_CHANGES at the current head must not be duplicated.
feedback = gitea_get_pr_review_feedback(
pr_number, remote=remote, org=org, repo=repo)
if not feedback.get("success"):
return {
"marked_ready": False,
"reasons": [
"could not verify existing request-changes state for "
f"PR #{pr_number}; refusing to submit a possibly "
"duplicate REQUEST_CHANGES (fail closed, #332)"
],
}
if (
feedback.get("has_blocking_change_requests")
and not feedback.get("review_feedback_stale")
):
return {
"marked_ready": False,
"reasons": [
f"PR #{pr_number} already has an unresolved "
"REQUEST_CHANGES at the current head SHA; do not "
"duplicate the request-changes review (fail closed, "
"#332)"
],
}
lock["final_review_decision_ready"] = True
lock["ready_pr_number"] = pr_number
lock["ready_action"] = action
@@ -2679,6 +2840,14 @@ def gitea_merge_pr(
)
return result
# Gate 2b — session hard-stop after a terminal review mutation (#332):
# merge may only continue for the same PR that was just approved.
# Local in-process check; zero API calls.
hard_stop = terminal_review_hard_stop_reasons(pr_number, "merge")
if hard_stop:
reasons.extend(hard_stop)
return result
# Gate 3 — reuse #14 eligibility (identity + profile + merge-allowed +
# author + self-merge block + open + mergeable/unknown fail-closed).
# Read-only GETs only.
@@ -2929,19 +3098,25 @@ def gitea_review_pr(
if can_read and auth:
inventory_attempted = True
try:
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
prs = api_get_all(url, auth)
prs_found_count = len(prs)
for pr in prs:
labels = [l["name"] for l in pr.get("labels", [])]
prs = gitea_list_prs(
state="open",
remote=remote,
host=h,
org=o,
repo=r,
)
prs_list = prs.get("prs") or []
prs_found_count = len(prs_list)
for pr in prs_list:
labels = list(pr.get("labels") or [])
body_text = pr.get("body") or ""
linked = []
# Simple extraction of linked issues from body
for match in re.findall(r'#(\d+)', body_text):
linked.append(f"#{match}")
pr_head_sha = (pr.get("head") or {}).get("sha")
pr_author = (pr.get("user") or {}).get("login")
pr_head_sha = pr.get("head_sha")
pr_author = pr.get("author")
# Determine review block status
req_non_author = False
@@ -2952,8 +3127,8 @@ def gitea_review_pr(
"number": pr["number"],
"title": pr["title"],
"author": pr_author,
"base": pr["base"]["ref"],
"head": pr["head"]["ref"],
"base": pr["base"],
"head": pr["head"],
"mergeable": pr.get("mergeable"),
"linked_issues": list(set(linked)),
"labels": labels,
@@ -2994,6 +3169,16 @@ def gitea_review_pr(
if inventory_msg:
report_lines.append(inventory_msg)
else:
pagination = (
prs.get("pagination") if isinstance(prs, dict) else {}
) or {}
has_finality_metadata = bool(
pagination.get("inventory_complete")
or (
pagination.get("is_final_page")
and not pagination.get("has_more")
)
)
inventory_trust_gate = review_proofs.pr_inventory_trust_gate(
prs if inventory_attempted else None,
remote=remote,
@@ -3002,7 +3187,7 @@ def gitea_review_pr(
state="open",
authenticated_profile=profile,
local_remote_url=_local_git_remote_url(remote),
has_finality_metadata=True,
has_finality_metadata=has_finality_metadata,
)
report_lines.extend(
review_proofs.format_pr_inventory_trust_gate_report(
review_proofs.py
+402 -3
View File
@@ -968,6 +968,90 @@ def assess_mutation_ledger_report(
}
def assess_list_prs_pagination_proof(
list_prs_response: dict | list | None,
*,
inventory_complete_claimed: bool = False,
) -> dict:
"""#340: verify ``gitea_list_prs`` pagination metadata proves inventory scope.
Accepts the wrapped ``{"prs": [...], "pagination": {...}}`` response from
``gitea_list_prs``. Legacy bare lists fail closed when completeness is
claimed. Single-page fetches may not claim full inventory unless
``inventory_complete`` or ``is_final_page`` is true.
"""
reasons: list[str] = []
if list_prs_response is None:
reasons.append("gitea_list_prs response missing; fail closed")
return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
if isinstance(list_prs_response, list):
pagination = None
if inventory_complete_claimed:
reasons.append(
"bare PR list lacks pagination metadata; cannot prove inventory "
"completeness"
)
return {
"proven": not reasons,
"block": bool(reasons),
"reasons": reasons,
"pagination": pagination,
}
if not isinstance(list_prs_response, dict):
reasons.append("gitea_list_prs response is not a dict or list; fail closed")
return {"proven": False, "block": True, "reasons": reasons, "pagination": None}
prs = list_prs_response.get("prs")
pagination = list_prs_response.get("pagination")
if not isinstance(prs, list):
reasons.append("gitea_list_prs response missing 'prs' list")
if not isinstance(pagination, dict):
reasons.append("gitea_list_prs response missing 'pagination' metadata")
return {
"proven": False,
"block": True,
"reasons": reasons,
"pagination": pagination if isinstance(pagination, dict) else None,
}
required_keys = ("page", "per_page", "returned_count", "has_more", "is_final_page")
for key in required_keys:
if key not in pagination:
reasons.append(f"pagination metadata missing '{key}'")
if inventory_complete_claimed:
complete = pagination.get("inventory_complete") is True
final_page = pagination.get("is_final_page") is True and pagination.get("has_more") is False
if not (complete or final_page):
reasons.append(
"inventory completeness claimed but pagination metadata does not "
"prove final page (inventory_complete or is_final_page with "
"has_more=false required)"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"pagination": pagination,
}
def _prs_from_list_response(list_prs_response: list | dict | None) -> list | None:
"""Normalize ``gitea_list_prs`` output to a PR list."""
if list_prs_response is None:
return None
if isinstance(list_prs_response, list):
return list_prs_response
if isinstance(list_prs_response, dict):
prs = list_prs_response.get("prs")
return prs if isinstance(prs, list) else None
return None
def assess_role_boundary(proof=None, *, task_role=None, namespaces_used=None,
justification=None):
"""Assess reviewer/author role separation for blind queue workflows.
@@ -1583,8 +1667,55 @@ HANDOFF_ROLE_FIELDS = {
)),
("Related issues", ("related issues",)),
),
"create_issue": (
("Active profile", ("active profile",)),
("Runtime context", ("runtime context",)),
("Requested issue task", ("requested issue task",)),
("Workflow source", ("workflow source",)),
("Capability proof", ("capability proof",)),
("Duplicate search terms", ("duplicate search terms",)),
("Duplicate search pagination proof", (
"duplicate search pagination proof",
)),
("Duplicates found", ("duplicates found",)),
("Issues created", ("issues created",)),
("Issues commented", ("issues commented",)),
("Issues edited", ("issues edited",)),
("Issues skipped as duplicates", (
"issues skipped as duplicates",
)),
("File edits by issue creator", ("file edits by issue creator",)),
("Safety statement", ("safety statement",)),
),
}
CREATE_ISSUE_WORKFLOW_MARKERS = (
"workflows/create-issue.md",
"create-issue.md",
)
CREATE_ISSUE_TASK_MARKERS = (
"create-issue",
"create issue",
)
CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS = (
("Selected issue", ("selected issue",)),
("Branch name", ("branch name",)),
("PR number", ("pr number",)),
("PR URL", ("pr url",)),
("Commit SHA", ("commit sha",)),
("Push result", ("push result",)),
("Baseline comparison", ("baseline comparison",)),
("Selected PR", ("selected pr",)),
("Review decision", ("review decision",)),
("Merge result", ("merge result",)),
("Merge preflight", ("merge preflight",)),
("Already-landed gate", ("already-landed gate",)),
("Pinned reviewed head", ("pinned reviewed head",)),
("Terminal review mutation", ("terminal review mutation",)),
)
def _handoff_section_lines(report_text):
"""Return the lines of the Controller Handoff section, or None."""
@@ -1635,6 +1766,11 @@ def assess_controller_handoff(report_text, role=None, local_edits=False):
fields_dict[label] = v.strip()
required = list(HANDOFF_BASE_FIELDS)
if role == "create_issue":
required = [
field for field in required
if field[0] not in ("Workspace mutations", "Mutations")
]
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
missing = []
@@ -1822,14 +1958,19 @@ def pr_inventory_trust_gate(
"queue_target_lock": lock_status,
}
if list_prs_response is None or not isinstance(list_prs_response, list):
prs_list = _prs_from_list_response(list_prs_response)
pagination_meta = (
list_prs_response.get("pagination")
if isinstance(list_prs_response, dict) else {}
) or {}
if prs_list is None:
return {
"status": "inventory_error",
"reasons": ["PR list response is invalid (not a list or None)"],
"reasons": ["PR list response is invalid (missing prs list or None)"],
"corroborated": False,
}
if len(list_prs_response) > 0:
if len(prs_list) > 0:
return {
"status": "trusted_nonempty",
"reasons": [],
@@ -1860,6 +2001,13 @@ def pr_inventory_trust_gate(
corroborated = False
if has_finality_metadata:
corroborated = True
elif pagination_meta.get("inventory_complete") is True:
corroborated = True
elif (
pagination_meta.get("is_final_page") is True
and pagination_meta.get("has_more") is False
):
corroborated = True
elif corroboration_open_pr_counter == 0:
corroborated = True
else:
@@ -2659,6 +2807,141 @@ def assess_issue_filing_duplicate_summary(
}
def assess_create_issue_workflow_source(report_text: str) -> dict:
"""#337: create-issue reports must cite the canonical workflow file."""
lower = (report_text or "").lower()
reasons = []
if not any(marker in lower for marker in CREATE_ISSUE_WORKFLOW_MARKERS):
reasons.append(
"create-issue report missing workflow source "
"(workflows/create-issue.md)"
)
if not any(marker in lower for marker in CREATE_ISSUE_TASK_MARKERS):
reasons.append(
"create-issue report missing task mode declaration "
"(create-issue)"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_create_issue_mode_isolation(report_text: str) -> dict:
"""#337: reject work-issue/review-merge handoff fields in create-issue runs."""
section = _handoff_section_lines(report_text)
reasons = []
if section is None:
return {
"complete": True,
"downgraded": False,
"reasons": [],
}
labels = []
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
for name, aliases in CREATE_ISSUE_FORBIDDEN_CROSS_MODE_FIELDS:
if any(
label.startswith(alias)
for label in labels
for alias in aliases
):
reasons.append(
f"create-issue handoff must not include cross-mode field "
f"'{name}'"
)
legacy_workspace = any(
label.startswith("workspace mutations") for label in labels
)
precise_categories = any(
label.startswith("file edits by issue creator")
for label in labels
)
if legacy_workspace and not precise_categories:
reasons.append(
"create-issue handoff must use precise mutation categories "
"instead of legacy 'Workspace mutations'"
)
return {
"complete": not reasons,
"downgraded": bool(reasons),
"reasons": reasons,
}
def assess_create_issue_final_report(
report_text: str,
*,
issue_number: int | None = None,
issue_title: str | None = None,
action: str = "created",
mutations: list[str] | None = None,
resolved_capabilities: dict | None = None,
issues_searched: int | None = None,
closest_matches: list[dict] | None = None,
duplicate_result: dict | None = None,
performed_mutations: list[str] | None = None,
) -> dict:
"""#337: composite verifier for create-issue final reports."""
checks = {
"workflow_source": assess_create_issue_workflow_source(report_text),
"mode_isolation": assess_create_issue_mode_isolation(report_text),
"controller_handoff": assess_controller_handoff(
report_text, role="create_issue"
),
"issue_reference": assess_issue_filing_issue_reference(
report_text,
issue_number=issue_number,
issue_title=issue_title,
action=action,
),
"mutation_capability": assess_issue_filing_mutation_capability(
report_text,
mutations or performed_mutations,
resolved_capabilities,
),
"mutation_scope": assess_issue_filing_mutation_scope(
report_text,
performed_mutations=performed_mutations or mutations,
),
"duplicate_summary": assess_issue_filing_duplicate_summary(
report_text,
issues_searched=issues_searched,
closest_matches=closest_matches,
duplicate_result=duplicate_result,
),
}
reasons = []
downgraded = False
for name, result in checks.items():
verdict = result.get("verdict")
if verdict in ("missing", "incomplete"):
downgraded = True
reasons.extend(result.get("reasons") or [])
elif result.get("downgraded") or not result.get("complete", True):
downgraded = True
reasons.extend(
f"{name}: {r}" for r in (result.get("reasons") or [])
)
grade = "A" if not downgraded else "downgraded"
return {
"grade": grade,
"downgraded": downgraded,
"checks": checks,
"reasons": reasons,
"complete": not downgraded,
}
def assess_issue_filing_final_report(
report_text: str,
*,
@@ -2952,3 +3235,119 @@ def assess_proof_wording(
"flagged_rules": flagged_rules,
"reasons": violations,
}
# ---------------------------------------------------------------------------
# Git ref mutations (#297)
# ---------------------------------------------------------------------------
_GIT_REF_MUTATIONS_FIELD_RE = re.compile(
r"^\s*git ref mutations\s*:\s*(.+?)\s*$", re.I | re.M)
_GIT_WORKTREE_MUTATIONS_NONE_RE = re.compile(
r"^\s*git/worktree mutations\s*:\s*none\b", re.I | re.M)
_GIT_REF_MUTATING_FIRST_WORDS = frozenset(
{"fetch", "pull", "push", "merge", "update-ref"})
_GIT_REF_MUTATING_PREFIXES = ("remote update", "branch -d", "branch -D",
"reset --hard")
def _command_text(entry):
if isinstance(entry, dict):
return str(entry.get("command") or "").strip()
return str(entry or "").strip()
def _git_subcommand_line(command):
tokens = command.split()
if not tokens or tokens[0] != "git":
return None
return " ".join(tokens[1:])
def git_ref_mutating_commands(command_log):
"""Return logged commands that update git refs (#297).
``git fetch`` and friends edit no files but move refs; they are never
read-only diagnostics.
"""
out = []
for entry in command_log or []:
command = _command_text(entry)
rest = _git_subcommand_line(command)
if rest is None:
continue
first = rest.split()[0] if rest.split() else ""
if first in _GIT_REF_MUTATING_FIRST_WORDS:
out.append(command)
continue
if any(rest.startswith(prefix) for prefix in _GIT_REF_MUTATING_PREFIXES):
out.append(command)
return out
def _fetch_targets(ref_commands):
"""Extract ``<remote>/<branch>`` targets from git fetch commands."""
targets = []
for command in ref_commands:
rest = _git_subcommand_line(command) or ""
tokens = rest.split()
if not tokens or tokens[0] != "fetch":
continue
args = [t for t in tokens[1:] if not t.startswith("-")]
if len(args) >= 2:
targets.append(f"{args[0]}/{args[1]}")
return targets
def assess_git_ref_mutation_report(report_text, *, command_log=None):
"""#297: ref updates are Git ref mutations, not read-only diagnostics.
When the command log holds ref-updating commands, the report must carry
a non-none ``Git ref mutations`` entry (naming ``fetched
<remote>/<branch>`` for targeted fetches), may not claim ``Git/worktree
mutations: None``, and may not list the command as read-only.
"""
text = report_text or ""
lower = text.lower()
reasons = []
ref_commands = git_ref_mutating_commands(command_log)
fetch_targets = _fetch_targets(ref_commands)
if ref_commands:
field = _GIT_REF_MUTATIONS_FIELD_RE.search(text)
value = field.group(1).strip().lower() if field else None
if not value or value == "none":
reasons.append(
"ref-updating commands ran but the report has no "
"'Git ref mutations' entry"
)
for target in fetch_targets:
if "fetched" not in lower or target.lower() not in lower:
reasons.append(
"git fetch ran; report must state "
f"'Git ref mutations: fetched {target}'"
)
if _GIT_WORKTREE_MUTATIONS_NONE_RE.search(text):
reasons.append(
"'Git/worktree mutations: None' rejected: ref-updating "
"commands ran"
)
for line in text.splitlines():
if "read-only" not in line.lower():
continue
for command in ref_commands:
if command.lower() in line.lower():
reasons.append(
"read-only diagnostics may not include ref-updating "
f"command '{command}'"
)
proven = not reasons
return {
"proven": proven,
"block": not proven,
"reasons": reasons,
"ref_mutating_commands": ref_commands,
"fetch_targets": fetch_targets,
}
tests/test_capability_stop_terminal.py
+7 -3
View File
@@ -99,12 +99,16 @@ class TestCapabilityStopTerminal(unittest.TestCase):
self.assertIn("Cannot perform reviewer task", str(ctx.exception))
self.assertIn("review_pr", str(ctx.exception))
@patch("mcp_server.api_get_all", return_value=[])
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
def test_list_prs_allowed_after_author_task_clears_stale_denial(
self, _auth, _api, _get_all,
self, _auth, _api, mock_fetch,
):
mock_fetch.return_value = ([], {
"page": 1, "per_page": 50, "returned_count": 0,
"has_more": False, "next_page": None, "is_final_page": True,
})
with patch.dict(os.environ, self._env()):
denied = mcp_server.gitea_resolve_task_capability(
task="review_pr", remote="prgs"
@@ -120,7 +124,7 @@ class TestCapabilityStopTerminal(unittest.TestCase):
self.assertFalse(capability_stop_terminal.is_active())
prs = mcp_server.gitea_list_prs(remote="prgs")
self.assertEqual(prs, [])
self.assertEqual(prs["prs"], [])
@patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
@patch("mcp_server.get_auth_header", return_value="token author-pass")
tests/test_external_mcp_registration_docs.py
+38
View File
@@ -59,6 +59,22 @@ def test_registration_doc_preserves_boundaries():
assert phrase in text
def test_registration_doc_pins_glitchtip_filing_boundary():
text = " ".join(_doc_text().split())
for phrase in (
"GlitchTip-to-Gitea filing is a separate library-only orchestrator in "
"`mcp-control-plane`",
"real GlitchTip read path",
"dedup before any Gitea create action",
"create/link/skip decisions",
"fail closed on mutation-audit failure before any Gitea create, link, "
"or comment mutation",
"must use a separate write-boundary server/profile",
"must not be exposed by `glitchtip-mcp`",
):
assert phrase in text
def test_registration_doc_separates_jenkins_trigger_from_read_surface():
text = _doc_text()
assert "jenkins_trigger_build" in text
@@ -100,3 +116,25 @@ def test_related_docs_keep_jenkins_trigger_off_read_surface():
assert "jenkins-write-mcp" in text
assert "jenkins_mcp.write_server" in text
assert "jenkins-mcp" in text
def test_glitchtip_filing_contract_doc_covers_issue_153_acceptance():
text = (
REPO_ROOT / "docs" / "architecture" /
"glitchtip-to-gitea-workflow-design.md"
).read_text(encoding="utf-8")
flattened = " ".join(text.split())
for phrase in (
"**Status:** Contract for the library-only implementation in "
"`Scaled-Tech-Consulting/mcp-control-plane` (#57)",
"**Issue:** #153",
"not exposed through `glitchtip-mcp`",
"real GlitchTip read path",
"must not use mocked GlitchTip issue data",
"separate write-boundary server/profile",
"fail-closed mutation audit before create/link/comment actions",
"Deduplication before create",
"Create, link, and skip decisions must be represented in tests",
"Mutation audit fails closed",
):
assert phrase in flattened
tests/test_git_ref_mutation_report.py
+101
View File
@@ -0,0 +1,101 @@
"""Git ref mutations must be reported, never hidden as diagnostics (#297).
``git fetch`` updates remote-tracking refs even though it edits no files.
Reports that ran fetch (or any ref-updating command) must carry a
``Git ref mutations`` entry and may not claim ``Git/worktree mutations:
None`` or classify the fetch as read-only diagnostics.
"""
import sys
import unittest
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import review_proofs
def _assess(report, commands):
return review_proofs.assess_git_ref_mutation_report(
report, command_log=commands)
class TestGitRefMutationDetection(unittest.TestCase):
def test_fetch_only_review_with_proper_ledger_passes(self):
report = "\n".join([
"Git ref mutations: fetched prgs/master",
"Review decision: APPROVE",
])
res = _assess(report, ["git fetch prgs master"])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(res["ref_mutating_commands"], ["git fetch prgs master"])
def test_fetch_without_ref_mutation_line_is_blocked(self):
report = "Review decision: APPROVE\nMutations: None\n"
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("git ref mutations" in r.lower() for r in res["reasons"]))
def test_fetch_with_remote_branch_requires_fetched_target_in_report(self):
report = "Git ref mutations: some refs updated\n"
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("prgs/master" in r for r in res["reasons"]))
def test_git_worktree_mutations_none_rejected_when_fetch_occurred(self):
report = "\n".join([
"Git ref mutations: fetched prgs/master",
"Git/worktree mutations: None",
])
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("git/worktree mutations" in r.lower() for r in res["reasons"]))
def test_fetch_classified_as_read_only_diagnostics_rejected(self):
report = "\n".join([
"Git ref mutations: fetched prgs/master",
"Read-only diagnostics: git fetch prgs master, git status",
])
res = _assess(report, ["git fetch prgs master"])
self.assertFalse(res["proven"])
self.assertTrue(
any("read-only" in r.lower() for r in res["reasons"]))
def test_worktree_creation_is_not_a_ref_mutation(self):
report = "Worktree mutations: created branches/review-pr9\n"
res = _assess(report, ["git worktree add branches/review-pr9 prgs/master"])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(res["ref_mutating_commands"], [])
def test_merge_command_counts_as_ref_mutation(self):
report = "Review decision: APPROVE\n"
res = _assess(report, ["git merge --no-ff feat/x"])
self.assertFalse(res["proven"])
self.assertIn("git merge --no-ff feat/x", res["ref_mutating_commands"])
def test_cleanup_branch_delete_counts_as_ref_mutation(self):
report = "Cleanup mutations: deleted branch feat/x\n"
res = _assess(report, ["git branch -d feat/x"])
self.assertFalse(res["proven"])
self.assertIn("git branch -d feat/x", res["ref_mutating_commands"])
def test_no_mutation_blocked_handoff_passes(self):
report = "\n".join([
"Git/worktree mutations: None",
"Current status: blocked handoff, no mutations performed",
])
res = _assess(report, ["git status --porcelain", "git log --oneline -1"])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(res["ref_mutating_commands"], [])
def test_command_log_dict_entries_supported(self):
report = "Git ref mutations: fetched prgs/master\n"
res = _assess(report, [{"command": "git fetch prgs master"}])
self.assertTrue(res["proven"], res["reasons"])
self.assertEqual(
res["ref_mutating_commands"], ["git fetch prgs master"])
if __name__ == "__main__":
unittest.main()
tests/test_list_prs_pagination.py
+153
View File
@@ -0,0 +1,153 @@
"""Tests for gitea_list_prs pagination metadata (#340)."""
import os
import sys
import unittest
from unittest.mock import patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
import gitea_auth # noqa: E402
from mcp_server import gitea_list_prs # noqa: E402
from review_proofs import assess_list_prs_pagination_proof # noqa: E402
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
URL = "https://gitea.example.com/api/v1/repos/o/r/pulls?state=open"
SAMPLE_PR = {
"number": 1,
"title": "PR 1",
"state": "open",
"head": {"ref": "branch1"},
"base": {"ref": "master"},
"mergeable": True,
"updated_at": "2026-07-05T12:00:00Z",
}
class TestApiFetchPage(unittest.TestCase):
@patch("gitea_auth.api_request")
def test_final_page_metadata(self, mock_req):
mock_req.return_value = [SAMPLE_PR]
items, meta = gitea_auth.api_fetch_page(URL, FAKE_AUTH, page=1, limit=50)
self.assertEqual(len(items), 1)
self.assertFalse(meta["has_more"])
self.assertTrue(meta["is_final_page"])
self.assertIsNone(meta["next_page"])
@patch("gitea_auth.api_request")
def test_full_page_has_more(self, mock_req):
mock_req.return_value = [{"id": i} for i in range(2)]
items, meta = gitea_auth.api_fetch_page(URL, FAKE_AUTH, page=1, limit=2)
self.assertEqual(len(items), 2)
self.assertTrue(meta["has_more"])
self.assertFalse(meta["is_final_page"])
self.assertEqual(meta["next_page"], 2)
class TestGiteaListPrsPagination(unittest.TestCase):
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_explicit_page_returns_has_more(self, _auth, mock_fetch):
mock_fetch.return_value = (
[SAMPLE_PR],
{
"page": 1,
"per_page": 50,
"returned_count": 1,
"has_more": False,
"next_page": None,
"is_final_page": True,
},
)
with patch.dict(os.environ, {}, clear=True):
result = gitea_list_prs(page=1, per_page=50)
self.assertEqual(len(result["prs"]), 1)
self.assertFalse(result["pagination"]["inventory_complete"])
self.assertTrue(result["pagination"]["is_final_page"])
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_multi_page_fetch_all_marks_inventory_complete(self, _auth, mock_fetch):
mock_fetch.side_effect = [
(
[dict(SAMPLE_PR, number=i) for i in (1, 2)],
{
"page": 1,
"per_page": 2,
"returned_count": 2,
"has_more": True,
"next_page": 2,
"is_final_page": False,
},
),
(
[dict(SAMPLE_PR, number=3)],
{
"page": 2,
"per_page": 2,
"returned_count": 1,
"has_more": False,
"next_page": None,
"is_final_page": True,
},
),
]
with patch.dict(os.environ, {}, clear=True):
result = gitea_list_prs(per_page=2)
self.assertEqual([p["number"] for p in result["prs"]], [1, 2, 3])
self.assertTrue(result["pagination"]["inventory_complete"])
self.assertEqual(result["pagination"]["pages_fetched"], 2)
self.assertEqual(result["pagination"]["total_count"], 3)
class TestListPrsPaginationProof(unittest.TestCase):
def test_wrapped_final_page_passes_completeness_claim(self):
response = {
"prs": [],
"pagination": {
"page": 1,
"per_page": 50,
"returned_count": 0,
"has_more": False,
"next_page": None,
"is_final_page": True,
"inventory_complete": True,
"pages_fetched": 1,
"total_count": 0,
},
}
result = assess_list_prs_pagination_proof(
response, inventory_complete_claimed=True
)
self.assertTrue(result["proven"])
def test_bare_list_blocks_completeness_claim(self):
result = assess_list_prs_pagination_proof(
[{"number": 1}], inventory_complete_claimed=True
)
self.assertFalse(result["proven"])
self.assertIn("pagination metadata", " ".join(result["reasons"]))
def test_single_page_without_complete_blocks_claim(self):
response = {
"prs": [{"number": 1}],
"pagination": {
"page": 1,
"per_page": 50,
"returned_count": 1,
"has_more": True,
"next_page": 2,
"is_final_page": False,
"inventory_complete": False,
"pages_fetched": 1,
"total_count": None,
},
}
result = assess_list_prs_pagination_proof(
response, inventory_complete_claimed=True
)
self.assertFalse(result["proven"])
if __name__ == "__main__":
unittest.main()
tests/test_llm_workflow_split.py
+26 -2
View File
@@ -1,4 +1,4 @@
"""Doc-contract checks for task-specific LLM workflow split (#333)."""
"""Doc-contract checks for task-specific LLM workflow split (#333, #334)."""
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
@@ -46,15 +46,33 @@ def test_skill_is_router_not_monolithic_review_body():
assert "gitea_submit_pr_review" not in SKILL
def test_skill_still_declares_controller_handoff_contract():
assert "## Controller Handoff" in SKILL
assert "assess_controller_handoff" in SKILL
assert "## Global LLM Worktree Rule" in SKILL
def test_review_merge_workflow_contract():
text = (SKILL_DIR / "workflows" / "review-merge-pr.md").read_text(encoding="utf-8")
assert "canonical: true" in text
assert "## 0. Load the canonical workflow first" in text
assert "## 26A. Terminal review mutation hard-stop" in text
assert "## 11A. Skipped PRs are read-only" in text
assert "## 35. Duplicate request-changes prevention" in text
assert "## 37. Controller handoff schema" in text
assert "INVENTORY_PAGINATION_UNPROVEN" in text
assert "ALREADY_LANDED_RECONCILE_REQUIRED" in text
def test_review_merge_final_report_schema_exists():
path = SKILL_DIR / "schemas" / "review-merge-final-report.md"
text = path.read_text(encoding="utf-8")
assert "Controller Handoff" in text
assert "Candidate head SHA:" in text
assert "Terminal review mutation:" in text
assert "Workspace mutations" in text
def test_reconcile_landed_workflow_contract():
text = (SKILL_DIR / "workflows" / "reconcile-landed-pr.md").read_text(encoding="utf-8")
assert "canonical: true" in text
@@ -97,4 +115,10 @@ def test_merge_pr_template_references_workflow():
def test_start_issue_template_references_work_issue_workflow():
text = (SKILL_DIR / "templates" / "start-issue.md").read_text(encoding="utf-8")
assert "workflows/work-issue.md" in text
assert "workflows/work-issue.md" in text
def test_create_issue_final_report_verifier_exported():
from review_proofs import assess_create_issue_final_report
assert callable(assess_create_issue_final_report)
tests/test_mcp_server.py
+45 -21
View File
@@ -48,6 +48,21 @@ import mcp_server
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
_NO_BLOCKER_FEEDBACK = {
"success": True,
"has_blocking_change_requests": False,
"review_feedback_stale": False,
}
def _mark_request_changes_ready(pr_number=8, **kwargs):
"""Mark a request_changes decision ready with the #332 duplicate-
suppression feedback fetch stubbed to 'no existing blocker'."""
with patch("mcp_server.gitea_get_pr_review_feedback",
return_value=dict(_NO_BLOCKER_FEEDBACK)):
return gitea_mark_final_review_decision(
pr_number, "request_changes", **kwargs)
def _formal_review(reviewer, verdict, sha="abc123", review_id=1):
return {
@@ -400,40 +415,49 @@ class TestMirrorRefs(unittest.TestCase):
# ---------------------------------------------------------------------------
class TestListPRs(unittest.TestCase):
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_list_prs(self, _auth, mock_api):
mock_api.return_value = [
{
def test_list_prs(self, _auth, mock_fetch):
mock_fetch.return_value = (
[{
"number": 1, "title": "PR 1", "state": "open",
"head": {"ref": "branch1"}, "base": {"ref": "main"},
"html_url": "http://url1", "mergeable": True,
"updated_at": "2026-07-05T12:00:00Z"
}
]
}],
{
"page": 1, "per_page": 50, "returned_count": 1,
"has_more": False, "next_page": None, "is_final_page": True,
},
)
with patch.dict(os.environ, {}, clear=True):
result = gitea_list_prs()
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["number"], 1)
self.assertEqual(result[0]["head"], "branch1")
self.assertNotIn("url", result[0])
self.assertEqual(len(result["prs"]), 1)
self.assertEqual(result["prs"][0]["number"], 1)
self.assertEqual(result["prs"][0]["head"], "branch1")
self.assertNotIn("url", result["prs"][0])
self.assertTrue(result["pagination"]["inventory_complete"])
# Staleness fields for queue reconciliation (#166)
self.assertEqual(result[0]["updated_at"], "2026-07-05T12:00:00Z")
self.assertEqual(result["prs"][0]["updated_at"], "2026-07-05T12:00:00Z")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_list_prs_reveal_opt_in_includes_url(self, _auth, mock_api):
mock_api.return_value = [
{
def test_list_prs_reveal_opt_in_includes_url(self, _auth, mock_fetch):
mock_fetch.return_value = (
[{
"number": 1, "title": "PR 1", "state": "open",
"head": {"ref": "branch1"}, "base": {"ref": "main"},
"html_url": "http://url1", "mergeable": True,
"updated_at": "2026-07-05T12:00:00Z"
}
]
}],
{
"page": 1, "per_page": 50, "returned_count": 1,
"has_more": False, "next_page": None, "is_final_page": True,
},
)
with patch.dict(os.environ, {"GITEA_MCP_REVEAL_ENDPOINTS": "1"}, clear=True):
result = gitea_list_prs()
self.assertEqual(result[0]["url"], "http://url1")
self.assertEqual(result["prs"][0]["url"], "http://url1")
# ---------------------------------------------------------------------------
@@ -1830,7 +1854,7 @@ class TestSubmitPrReview(unittest.TestCase):
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
def test_request_changes_succeeds_when_eligible(self, _auth, mock_api):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
_mark_request_changes_ready(remote="prgs")
mock_api.side_effect = [
{"login": "reviewer-bot"}, self._pr("author-bot"),
{"id": 9, "state": "REQUEST_CHANGES"},
@@ -1853,7 +1877,7 @@ class TestSubmitPrReview(unittest.TestCase):
self.assertEqual(post_calls[0].args[3]["event"], "REQUEST_CHANGES")
def test_request_changes_blocked_without_eligibility(self):
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
_mark_request_changes_ready(remote="prgs")
with patch("mcp_server.get_auth_header", return_value=FAKE_AUTH) as _a, \
patch("mcp_server.api_request") as mock_api:
mock_api.side_effect = [{"login": "reviewer-bot"}, self._pr("author-bot")]
@@ -2135,7 +2159,7 @@ class TestSubmitPrReview(unittest.TestCase):
reason="operator approved correcting mistaken approve",
operator_authorized=True,
)
gitea_mark_final_review_decision(8, "request_changes", remote="prgs")
_mark_request_changes_ready(remote="prgs")
second = gitea_submit_pr_review(
pr_number=8, action="request_changes", remote="prgs",
final_review_decision_ready=True,
tests/test_pr_queue_inventory.py
+67 -55
View File
@@ -16,38 +16,50 @@ import gitea_config
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
def _final_page_fetch(items):
return (items, {
"page": 1,
"per_page": 50,
"returned_count": len(items),
"has_more": False,
"next_page": None,
"is_final_page": True,
})
class TestPRQueueInventory(unittest.TestCase):
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profile_can_list_open_prs(self, mock_get_profile, _auth, mock_get_all):
def test_author_profile_can_list_open_prs(self, mock_get_profile, _auth, mock_fetch):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read", "gitea.pr.comment"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
mock_fetch.return_value = _final_page_fetch([
{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True}
]
])
result = gitea_list_prs(state="open", remote="prgs")
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["number"], 1)
self.assertEqual(len(result["prs"]), 1)
self.assertEqual(result["prs"][0]["number"], 1)
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
@patch("gitea_config.is_runtime_switching_enabled", return_value=True)
def test_author_profile_can_produce_pending_reviewer_queue_report(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all):
def test_author_profile_can_produce_pending_reviewer_queue_report(self, mock_switch, mock_get_profile, _auth, mock_api, mock_fetch):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
mock_fetch.return_value = _final_page_fetch([
{
"number": 1,
"title": "PR 1",
@@ -72,7 +84,7 @@ class TestPRQueueInventory(unittest.TestCase):
"body": "",
"updated_at": "2026-07-05T11:00:00Z"
}
]
])
mock_api.return_value = {"login": "jcwalker3"} # whoami
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
@@ -86,18 +98,18 @@ class TestPRQueueInventory(unittest.TestCase):
# updated_at surfaced for reconciliation (#166)
self.assertIn("Updated: 2026-07-05T10:00:00Z", result["message"])
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profile_never_calls_mutation_tools_during_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_author_profile_never_calls_mutation_tools_during_inventory(self, mock_get_profile, _auth, mock_api, mock_fetch):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_fetch.return_value = _final_page_fetch([])
mock_api.return_value = {"login": "jcwalker3"}
gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
@@ -105,20 +117,20 @@ class TestPRQueueInventory(unittest.TestCase):
method = call.args[0]
self.assertEqual(method, "GET")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_reviewer_profile_can_proceed_from_inventory_to_review_gates(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_reviewer_profile_can_proceed_from_inventory_to_review_gates(self, mock_get_profile, _auth, mock_api, mock_fetch):
mock_get_profile.return_value = {
"profile_name": "gitea-reviewer",
"allowed_operations": ["read", "approve", "review"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
mock_fetch.return_value = _final_page_fetch([
{"number": 1, "title": "PR 1", "state": "open", "head": {"ref": "branch1", "sha": "abc1"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "other_user"}}
]
])
# mock_api: inventory whoami, eligibility whoami, eligibility PR,
# POST review (#244: state + visible-verdict GET reviews).
mock_api.side_effect = [
@@ -152,19 +164,19 @@ class TestPRQueueInventory(unittest.TestCase):
self.assertIn("Repository:", result["message"])
self.assertIn("Successfully submitted review", result["message"])
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
@patch("gitea_config.is_runtime_switching_enabled", return_value=False)
def test_static_runtime_rejects_or_hides_profile_activation(self, mock_switch, mock_get_profile, _auth, mock_api, mock_get_all):
def test_static_runtime_rejects_or_hides_profile_activation(self, mock_switch, mock_get_profile, _auth, mock_api, mock_fetch):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
"allowed_operations": ["read"],
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_fetch.return_value = _final_page_fetch([])
mock_api.return_value = {"login": "jcwalker3"}
result = gitea_review_pr(pr_number=1, event="APPROVE", remote="prgs")
@@ -200,11 +212,11 @@ class TestPRQueueInventory(unittest.TestCase):
self.assertFalse(result["eligible"])
self.assertEqual(mock_api.call_count, 0)
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_inventory_failure_output_is_redacted_no_raw_exception_leak(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_inventory_failure_output_is_redacted_no_raw_exception_leak(self, mock_get_profile, _auth, mock_api, mock_fetch):
"""Inventory failure path must use redaction and not leak raw exception text."""
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -213,7 +225,7 @@ class TestPRQueueInventory(unittest.TestCase):
"base_url": None,
}
# Exception contains something that should be redacted or not leaked raw
mock_get_all.side_effect = Exception("connection error token supersecrettoken123 at https://internal.example/repo")
mock_fetch.side_effect = Exception("connection error token supersecrettoken123 at https://internal.example/repo")
mock_api.return_value = {"login": "jcwalker3"}
result = gitea_review_pr(pr_number=42, event="APPROVE", remote="prgs")
@@ -228,11 +240,11 @@ class TestPRQueueInventory(unittest.TestCase):
# uses project's redaction pattern (token prefix -> [REDACTED])
# even if url leaks, the exception is redacted per pattern
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_inventory_failure_output_redacts_real_service_urls(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_inventory_failure_output_redacts_real_service_urls(self, mock_get_profile, _auth, mock_api, mock_fetch):
"""Failure paths must fully redact real service hostnames and URLs from inventory output."""
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -240,7 +252,7 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.side_effect = Exception("failed to connect to https://gitea.prgs.cc/api/v1/repos/x")
mock_fetch.side_effect = Exception("failed to connect to https://gitea.prgs.cc/api/v1/repos/x")
mock_api.return_value = {"login": "jcwalker3"}
result = gitea_review_pr(pr_number=42, event="APPROVE", remote="prgs")
@@ -250,11 +262,11 @@ class TestPRQueueInventory(unittest.TestCase):
self.assertIn("[REDACTED_URL]", msg)
self.assertNotIn("gitea.prgs.cc", msg)
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_author_profiles_can_perform_read_only_pr_inventory(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_author_profiles_can_perform_read_only_pr_inventory(self, mock_get_profile, _auth, mock_api, mock_fetch):
"""Author profiles with read can perform read-only PR inventory (report produced, no mutations)."""
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -262,9 +274,9 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [
mock_fetch.return_value = _final_page_fetch([
{"number": 7, "title": "Some PR", "state": "open", "head": {"ref": "f", "sha": "def"}, "base": {"ref": "master"}, "mergeable": True, "user": {"login": "someone"}, "labels": [], "body": ""}
]
])
mock_api.return_value = {"login": "jcwalker3"}
result = gitea_review_pr(pr_number=7, event="APPROVE", remote="prgs")
@@ -278,12 +290,12 @@ class TestPRQueueInventory(unittest.TestCase):
self.assertEqual(call.args[0], "GET")
@patch("mcp_server._local_git_remote_url")
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@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
self, mock_get_profile, _auth, mock_api, mock_fetch, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -291,7 +303,7 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_fetch.return_value = _final_page_fetch([])
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = None
@@ -309,12 +321,12 @@ class TestPRQueueInventory(unittest.TestCase):
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_fetch_page")
@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
self, mock_get_profile, _auth, mock_api, mock_fetch, mock_local_url
):
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -322,7 +334,7 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_fetch.return_value = _final_page_fetch([])
mock_api.return_value = {"login": "jcwalker3"}
mock_local_url.return_value = (
"https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
@@ -341,11 +353,11 @@ class TestPRQueueInventory(unittest.TestCase):
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_fetch_page")
@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_fetch, mock_local_url):
"""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"
@@ -357,7 +369,7 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = []
mock_fetch.return_value = _final_page_fetch([])
mock_api.return_value = {"login": "jcwalker3"}
result = gitea_review_pr(pr_number=1, event=event, remote="prgs")
@@ -377,10 +389,10 @@ class TestPRQueueInventory(unittest.TestCase):
# Tests use direct list/view calls + inventory path to ensure fields and
# live data are available for detection without trusting prior handoffs.
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_r5_s1_stale_prior_handoff_merged_but_live_pr_open(self, mock_get_profile, _auth, mock_get_all):
def test_r5_s1_stale_prior_handoff_merged_but_live_pr_open(self, mock_get_profile, _auth, mock_fetch):
"""Scenario 1: stale prior handoff says merged but live PR list says open."""
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -388,17 +400,17 @@ class TestPRQueueInventory(unittest.TestCase):
"forbidden_operations": [],
"base_url": None,
}
mock_get_all.return_value = [{
mock_fetch.return_value = _final_page_fetch([{
"number": 99, "title": "Stale merge claim", "state": "open",
"head": {"ref": "f99", "sha": "sha99"}, "base": {"ref": "master"},
"mergeable": True, "user": {"login": "other"},
"labels": [], "body": "Fixes #200",
"updated_at": "2026-07-05T22:00:00Z"
}]
}])
prs = gitea_list_prs(state="open", remote="prgs")
self.assertEqual(len(prs), 1)
self.assertEqual(prs[0]["state"], "open")
self.assertIn("updated_at", prs[0]) # enables staleness check vs prior "merged"
self.assertEqual(len(prs["prs"]), 1)
self.assertEqual(prs["prs"][0]["state"], "open")
self.assertIn("updated_at", prs["prs"][0]) # enables staleness check vs prior "merged"
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@@ -424,11 +436,11 @@ class TestPRQueueInventory(unittest.TestCase):
self.assertIsNotNone(pr.get("closed_at"))
# live closed would contradict prior "open" handoff
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_r5_s3_list_prs_and_view_pr_disagree(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_r5_s3_list_prs_and_view_pr_disagree(self, mock_get_profile, _auth, mock_api, mock_fetch):
"""Scenario 3: gitea_list_prs and gitea_view_pr disagree on state/details."""
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -437,12 +449,12 @@ class TestPRQueueInventory(unittest.TestCase):
"base_url": None,
}
# list says open
mock_get_all.return_value = [{
mock_fetch.return_value = _final_page_fetch([{
"number": 77, "title": "Disagree", "state": "open",
"head": {"ref": "f77", "sha": "s77"}, "base": {"ref": "master"},
"mergeable": True, "user": {"login": "x"},
"labels": [], "body": "", "updated_at": "2026-07-05T23:00:00Z"
}]
}])
# view says closed (simulating inconsistency)
mock_api.return_value = {
"number": 77, "title": "Disagree", "state": "closed",
@@ -453,7 +465,7 @@ class TestPRQueueInventory(unittest.TestCase):
}
listed = gitea_list_prs(state="open", remote="prgs")
viewed = gitea_view_pr(pr_number=77, remote="prgs")
self.assertEqual(listed[0]["state"], "open")
self.assertEqual(listed["prs"][0]["state"], "open")
self.assertEqual(viewed["state"], "closed")
# caller must reconcile; mismatch is hard blocker per rules
@@ -483,11 +495,11 @@ class TestPRQueueInventory(unittest.TestCase):
self.assertIsNone(pr.get("merge_commit_sha"))
# post-merge view must surface missing commit for detection
@patch("mcp_server.api_get_all")
@patch("mcp_server.api_fetch_page")
@patch("mcp_server.api_request")
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
@patch("mcp_server.get_profile")
def test_r5_s5_linked_issue_remains_open_after_claimed_merge(self, mock_get_profile, _auth, mock_api, mock_get_all):
def test_r5_s5_linked_issue_remains_open_after_claimed_merge(self, mock_get_profile, _auth, mock_api, mock_fetch):
"""Scenario 5: linked issue remains open after claimed merge."""
mock_get_profile.return_value = {
"profile_name": "gitea-author",
@@ -496,12 +508,12 @@ class TestPRQueueInventory(unittest.TestCase):
"base_url": None,
}
# PR claims merge, links #300
mock_get_all.return_value = [{
mock_fetch.return_value = _final_page_fetch([{
"number": 300, "title": "PR for #300", "state": "closed",
"head": {"ref": "f300", "sha": "s3"}, "base": {"ref": "master"},
"mergeable": True, "user": {"login": "u"},
"labels": [], "body": "Fixes #300", "updated_at": "2026-07-05T20:20:00Z"
}]
}])
# But linked issue (via list_issues or view) still open - here we use view_pr body to confirm link
# For this test we assert the PR data shows link, caller must check issue state live
mock_api.return_value = {
tests/test_review_proofs.py
+118
View File
@@ -29,6 +29,9 @@ from review_proofs import ( # noqa: E402
assess_contradictory_no_pr_claim,
assess_continuation_mode_report,
assess_controller_handoff,
assess_create_issue_final_report,
assess_create_issue_mode_isolation,
assess_create_issue_workflow_source,
assess_edited_pr_inventory_coverage,
assess_empty_queue_report,
assess_fresh_issue_selection,
@@ -2042,5 +2045,120 @@ class TestIssueFilingFinalReport(unittest.TestCase):
self.assertTrue(result["downgraded"])
class TestCreateIssueFinalReport(unittest.TestCase):
"""#337: create-issue final-report verifier coverage."""
ISSUE_TITLE = "Extract create-issue workflow from monolithic LLM project skill"
CAPABILITIES = {
"create_issue": {
"requested_task": "create_issue",
"required_operation_permission": "gitea.issue.create",
"allowed_in_current_session": True,
},
}
CLOSEST = [{"number": 333, "title": "Split monolithic workflow", "state": "closed"}]
def _good_report(self):
return "\n".join([
"Task mode: create-issue",
"Workflow source: skills/llm-project-workflow/workflows/create-issue.md",
f"Updated issue #337 — `{self.ISSUE_TITLE}` (remaining verifier AC).",
"Duplicate check: searched 54 open issues.",
"Closest existing: #333 — Split monolithic workflow (partial overlap).",
"Why new issue justified: dedicated create-issue extraction.",
"Only mutation: issue creation (gitea.issue.create).",
"Confirm no labels, comments, PRs, reviews, merges, or closes.",
"## Controller Handoff",
"- Task: create-issue",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: jcwalker3 / prgs-author",
"- Active profile: prgs-author",
"- Runtime context: prgs-author on prgs",
"- Requested issue task: extract create-issue workflow verifier",
"- Workflow source: workflows/create-issue.md @ fd396df",
"- Capability proof: create_issue allowed (gitea.issue.create)",
"- Duplicate search terms: create-issue, workflow extract",
"- Duplicate search pagination proof: 54 open issues < limit 500",
"- Duplicates found: #333 partial only",
"- Issues created: none (implementation PR for remaining AC)",
"- Issues commented: none",
"- Issues edited: none",
"- Issues skipped as duplicates: none",
"- File edits by issue creator: review_proofs.py, tests",
"- Worktree/index mutations: feat/issue-337 worktree created",
"- Git ref mutations: fetch prgs/master",
"- MCP/Gitea mutations: gitea_mark_issue start on #337",
"- Issue mutations: claimed #337",
"- Label/assignment/milestone mutations: none",
"- External-state mutations: none",
"- Read-only diagnostics: whoami, list issues",
"- Blockers: none",
"- Current status: implementing verifier",
"- Safe next action: open PR",
"- Safety statement: no review/merge/repo main-checkout edits",
"- Issue/PR: #337",
"- Branch/SHA: feat/issue-337-create-issue-verifier",
"- Files changed: review_proofs.py, tests",
"- Validation: pytest tests/test_review_proofs.py",
"- Mutations: create_issue capability resolved; mark_issue only",
"- Next: open PR",
"- Safety: no review/merge",
])
def test_complete_create_issue_report_earns_a(self):
result = assess_create_issue_final_report(
self._good_report(),
issue_number=337,
issue_title=self.ISSUE_TITLE,
action="updated",
mutations=["create_issue"],
resolved_capabilities=self.CAPABILITIES,
issues_searched=54,
closest_matches=self.CLOSEST,
performed_mutations=["create_issue"],
)
self.assertEqual(result["grade"], "A")
self.assertFalse(result["downgraded"])
def test_missing_workflow_source_downgrades(self):
report = self._good_report().replace("workflows/create-issue.md", "SKILL.md only")
result = assess_create_issue_workflow_source(report)
self.assertTrue(result["downgraded"])
def test_work_issue_field_in_handoff_downgrades(self):
report = self._good_report().replace(
"- Safety statement:",
"- Selected issue: #123\n- Safety statement:",
)
result = assess_create_issue_mode_isolation(report)
self.assertTrue(result["downgraded"])
self.assertTrue(
any("Selected issue" in r for r in result["reasons"])
)
def test_review_merge_field_in_handoff_downgrades(self):
report = self._good_report().replace(
"- Safety statement:",
"- Review decision: APPROVE\n- Safety statement:",
)
result = assess_create_issue_mode_isolation(report)
self.assertTrue(result["downgraded"])
def test_legacy_workspace_mutations_without_precise_categories_downgrades(self):
report = self._good_report().replace(
"- File edits by issue creator:",
"- Workspace mutations: none\n- File edits by issue creator:",
)
lines = report.splitlines()
cleaned = [
line for line in lines
if not line.strip().startswith("- File edits by issue creator:")
]
report = "\n".join(cleaned)
result = assess_create_issue_mode_isolation(report)
self.assertTrue(result["downgraded"])
if __name__ == "__main__":
unittest.main()
tests/test_terminal_review_hard_stop.py
+187
View File
@@ -0,0 +1,187 @@
"""Tests for the session hard-stop after a terminal review mutation (#332).
Extends the single-terminal-decision machinery (#211): after a live
REQUEST_CHANGES the run must stop; after an APPROVE only the merge sequence
for that same PR may continue; a different PR can never be reviewed or
merged in the same run; duplicate REQUEST_CHANGES at an unchanged head is
suppressed.
"""
import os
import unittest
from unittest.mock import patch
import mcp_server
def _lock(mutations=None, correction=False):
return {
"task": "review_pr",
"remote": "prgs",
"session_pid": os.getpid(),
"session_profile": "prgs-reviewer",
"session_profile_lock": "",
"final_review_decision_ready": False,
"ready_pr_number": None,
"ready_action": None,
"ready_expected_head_sha": None,
"ready_remote": None,
"ready_org": None,
"ready_repo": None,
"live_mutations": list(mutations or []),
"correction_authorized": correction,
"correction_reason": None,
}
def _seed(mutations=None, correction=False):
mcp_server._save_review_decision_lock(_lock(mutations, correction))
APPROVED_A = {"pr_number": 5, "action": "approve", "review_id": 1,
"review_state": "approve"}
RC_A = {"pr_number": 5, "action": "request_changes", "review_id": 2,
"review_state": "request_changes"}
class TestTerminalHardStopReasons(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_no_lock_no_reasons(self):
mcp_server._save_review_decision_lock(None)
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
def test_no_terminal_mutation_no_reasons(self):
_seed()
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
def test_request_changes_blocks_everything(self):
_seed([RC_A])
for op in ("merge", "mark_ready", "review"):
for pr in (5, 6):
reasons = mcp_server.terminal_review_hard_stop_reasons(pr, op)
self.assertTrue(reasons, f"{op}/{pr}")
self.assertIn("request_changes on PR #5", reasons[0])
self.assertIn("#332", reasons[0])
def test_approve_allows_same_pr_merge_only(self):
_seed([APPROVED_A])
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "merge"), [])
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "mark_ready"))
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "review"))
def test_correction_reopens_review_path_only(self):
_seed([RC_A], correction=True)
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "mark_ready"), [])
self.assertEqual(
mcp_server.terminal_review_hard_stop_reasons(5, "review"), [])
# correction never opens a cross-PR merge
self.assertTrue(
mcp_server.terminal_review_hard_stop_reasons(6, "merge"))
class TestMergeHardStopWiring(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_merge_blocked_after_request_changes(self):
_seed([RC_A])
result = mcp_server.gitea_merge_pr(
pr_number=6, confirmation="MERGE PR 6", remote="prgs")
self.assertFalse(result["performed"])
self.assertTrue(
any("#332" in r for r in result["reasons"]), result["reasons"])
def test_merge_of_other_pr_blocked_after_approve(self):
_seed([APPROVED_A])
result = mcp_server.gitea_merge_pr(
pr_number=6, confirmation="MERGE PR 6", remote="prgs")
self.assertFalse(result["performed"])
self.assertTrue(
any("#332" in r for r in result["reasons"]), result["reasons"])
class TestMarkFinalHardStopWiring(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_mark_ready_blocked_after_terminal_mutation(self):
_seed([RC_A])
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="approve", remote="prgs")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("#332" in r for r in result["reasons"]), result["reasons"])
def _feedback(blocking, stale=False, success=True):
return {
"success": success,
"has_blocking_change_requests": blocking,
"review_feedback_stale": stale,
"current_head_sha": "abc123",
}
class TestDuplicateRequestChangesSuppression(unittest.TestCase):
def tearDown(self):
mcp_server._save_review_decision_lock(None)
def test_duplicate_request_changes_blocked_at_same_head(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=True, stale=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("duplicate" in r for r in result["reasons"]),
result["reasons"])
def test_request_changes_allowed_when_prior_blocker_stale(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=True, stale=True)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertTrue(result["marked_ready"], result.get("reasons"))
def test_request_changes_allowed_when_no_blocker(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertTrue(result["marked_ready"], result.get("reasons"))
def test_request_changes_fails_closed_when_feedback_unavailable(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
return_value=_feedback(blocking=False,
success=False)):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="request_changes", remote="prgs")
self.assertFalse(result["marked_ready"])
self.assertTrue(
any("could not verify" in r for r in result["reasons"]),
result["reasons"])
def test_approve_does_not_fetch_feedback(self):
_seed()
with patch.object(mcp_server, "gitea_get_pr_review_feedback",
side_effect=AssertionError("must not be called")):
result = mcp_server.gitea_mark_final_review_decision(
pr_number=6, action="approve", remote="prgs")
self.assertTrue(result["marked_ready"], result.get("reasons"))
if __name__ == "__main__":
unittest.main()