Compare commits
17
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
00efda0cfb | ||
|
sysadmin
|
ce61e424f6 | ||
|
sysadminandClaude Opus 4.8
|
2e4ebc6434 | ||
|
sysadmin
|
f89949cea9 | ||
|
sysadmin
|
b3edd53185 | ||
|
sysadmin
|
9ada4762c5 | ||
|
sysadminandClaude Opus 4.8
|
3bce0a55fb | ||
|
sysadminandClaude Opus 4.8
|
afb4ba563c | ||
|
sysadmin
|
48cf3a334b | ||
|
sysadminandClaude Opus 4.8
|
a8bcbdbcf2 | ||
|
sysadminandClaude Opus 4.8
|
2d4ab4e54e | ||
|
sysadminandClaude Opus 4.8
|
514eae84f9 | ||
|
sysadmin
|
28f0f84de8 | ||
|
sysadminandClaude Opus 4.8
|
08202f7eaa | ||
 jcwalker3andClaude Opus 4.8
|
3496fc3952 | ||
|
sysadminandClaude Opus 4.8
|
c7b462a034 | ||
|
sysadminandClaude Opus 4.8
|
1320a55a7e |
@@ -1,408 +0,0 @@
|
||||
"""Fail-closed validation for workflow-changing Gitea comments (#496)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
VALID_ROLES = frozenset({
|
||||
"controller",
|
||||
"author",
|
||||
"reviewer",
|
||||
"merger",
|
||||
"reconciler",
|
||||
"user",
|
||||
})
|
||||
|
||||
_BASE_REQUIRED = ("STATE", "WHO_IS_NEXT", "NEXT_ACTION", "NEXT_PROMPT", "WHY")
|
||||
|
||||
_ISSUE_REQUIRED = _BASE_REQUIRED + ("BLOCKERS", "VALIDATION")
|
||||
_PR_REQUIRED = _BASE_REQUIRED + (
|
||||
"ISSUE",
|
||||
"HEAD_SHA",
|
||||
"REVIEW_STATUS",
|
||||
"MERGE_READY",
|
||||
"BLOCKERS",
|
||||
"VALIDATION",
|
||||
)
|
||||
_SUPERSESSION_REQUIRED = _BASE_REQUIRED + (
|
||||
"CANONICAL_ITEM",
|
||||
"SUPERSEDED_ITEM",
|
||||
"CLOSE_OR_KEEP_OPEN",
|
||||
)
|
||||
|
||||
_MACHINE_MARKERS = (
|
||||
"<!-- mcp-review-lease:v1 -->",
|
||||
"<!-- mcp-conflict-fix-lease:v1 -->",
|
||||
"<!-- gitea-issue-claim-heartbeat:v1 -->",
|
||||
)
|
||||
|
||||
_WORKFLOW_TRIGGERS = re.compile(
|
||||
r"\b(?:"
|
||||
r"blocked|unblocked|ready(?:\s+for\s+(?:review|merge|author))?|"
|
||||
r"ready-to-merge|approved|approve|request\s+changes|changes\s+requested|"
|
||||
r"superseded|duplicate|canonical|next\s+action|next\s+actor|who\s+is\s+next|"
|
||||
r"author\s+should|reviewer\s+should|merger\s+should|reconciler\s+should|"
|
||||
r"controller\s+should|issue\s+complete|pr\s+open|pr\s+merged|close\s+this|"
|
||||
r"do\s+not\s+merge|needs?\s+rebase|stale\s+approval|contaminated\s+review|"
|
||||
r"merge\s+ready|ready\s+for\s+merge"
|
||||
r")\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
_CANONICAL_HEADINGS = (
|
||||
"## Canonical Issue State",
|
||||
"## Canonical PR State",
|
||||
"## Canonical Discussion Summary",
|
||||
)
|
||||
|
||||
_FIELD_RE = re.compile(
|
||||
r"^([A-Z][A-Z0-9_]*)\s*:\s*(.*)$",
|
||||
re.MULTILINE,
|
||||
)
|
||||
|
||||
_VAGUE_NEXT_ACTIONS = frozenset({
|
||||
"continue",
|
||||
"handle this",
|
||||
"fix it",
|
||||
"follow up",
|
||||
"follow-up",
|
||||
"see above",
|
||||
"see review",
|
||||
"tbd",
|
||||
"todo",
|
||||
"as needed",
|
||||
"proceed",
|
||||
"next steps",
|
||||
"will check",
|
||||
"investigate",
|
||||
})
|
||||
|
||||
_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
|
||||
_SHORT_SHA_RE = re.compile(r"\b[0-9a-f]{7,40}\b", re.IGNORECASE)
|
||||
_PR_REF_RE = re.compile(r"(?:PR\s*#|pull\s*#)\d+|\b#\d{2,}\b", re.IGNORECASE)
|
||||
_APPROVAL_PROOF_RE = re.compile(
|
||||
r"\b(?:approved|approval_at_current_head|APPROVE)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_UNBLOCK_RE = re.compile(
|
||||
r"\b(?:unblock|until|after|once|when|requires?|must)\b",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _parse_fields(body: str) -> dict[str, str]:
|
||||
"""Parse KEY: value fields, including values continued on following lines."""
|
||||
fields: dict[str, str] = {}
|
||||
current_key: str | None = None
|
||||
current_lines: list[str] = []
|
||||
|
||||
def _flush() -> None:
|
||||
nonlocal current_key, current_lines
|
||||
if current_key is not None:
|
||||
fields[current_key] = "\n".join(current_lines).strip()
|
||||
current_key = None
|
||||
current_lines = []
|
||||
|
||||
for line in (body or "").splitlines():
|
||||
if line.startswith("## "):
|
||||
_flush()
|
||||
continue
|
||||
match = re.match(r"^([A-Z][A-Z0-9_]*)\s*:\s*(.*)$", line)
|
||||
if match:
|
||||
_flush()
|
||||
current_key = match.group(1).strip().upper()
|
||||
rest = match.group(2)
|
||||
current_lines = [rest] if rest else []
|
||||
elif current_key is not None:
|
||||
current_lines.append(line)
|
||||
_flush()
|
||||
return fields
|
||||
|
||||
|
||||
def _is_machine_generated(body: str) -> bool:
|
||||
text = body or ""
|
||||
return any(marker in text for marker in _MACHINE_MARKERS)
|
||||
|
||||
|
||||
def _has_canonical_heading(body: str) -> bool:
|
||||
return any(h in (body or "") for h in _CANONICAL_HEADINGS)
|
||||
|
||||
|
||||
def is_workflow_changing_comment(body: str) -> bool:
|
||||
"""True when comment text implies a workflow/state transition."""
|
||||
text = (body or "").strip()
|
||||
if not text:
|
||||
return False
|
||||
if _is_machine_generated(text):
|
||||
return False
|
||||
if _has_canonical_heading(text):
|
||||
return True
|
||||
if _WORKFLOW_TRIGGERS.search(text):
|
||||
return True
|
||||
fields = _parse_fields(text)
|
||||
if "STATE" in fields or "WHO_IS_NEXT" in fields:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def infer_comment_context(body: str, *, explicit: str | None = None) -> str:
|
||||
if explicit:
|
||||
return explicit
|
||||
text = body or ""
|
||||
if "## Canonical Discussion Summary" in text:
|
||||
return "discussion_summary"
|
||||
if "## Canonical PR State" in text:
|
||||
return "pr_comment"
|
||||
if "## Canonical Issue State" in text:
|
||||
return "issue_comment"
|
||||
fields = _parse_fields(text)
|
||||
if fields.get("CANONICAL_ITEM") or fields.get("SUPERSEDED_ITEM"):
|
||||
return "supersession"
|
||||
if any(k in fields for k in ("HEAD_SHA", "REVIEW_STATUS", "MERGE_READY", "ISSUE")):
|
||||
return "pr_comment"
|
||||
return "issue_comment"
|
||||
|
||||
|
||||
def _required_fields_for_context(context: str) -> tuple[str, ...]:
|
||||
if context in ("pr_comment", "pr_review"):
|
||||
return _PR_REQUIRED
|
||||
if context == "supersession":
|
||||
return _SUPERSESSION_REQUIRED
|
||||
if context == "discussion_summary":
|
||||
return _BASE_REQUIRED + ("DECISION", "SUBSTANTIVE_COMMENTS")
|
||||
return _ISSUE_REQUIRED
|
||||
|
||||
|
||||
def _is_vague_next_action(value: str) -> bool:
|
||||
normalized = re.sub(r"\s+", " ", (value or "").strip().lower())
|
||||
normalized = normalized.rstrip(".")
|
||||
if not normalized:
|
||||
return True
|
||||
if normalized in _VAGUE_NEXT_ACTIONS:
|
||||
return True
|
||||
if len(normalized) < 12:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _next_prompt_ok(value: str) -> bool:
|
||||
text = (value or "").strip()
|
||||
if len(text) < 40:
|
||||
return False
|
||||
if text.lower() in {"n/a", "none", "tbd", "todo"}:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _state_value(fields: dict[str, str]) -> str:
|
||||
return (fields.get("STATE") or "").strip().lower()
|
||||
|
||||
|
||||
def _suggested_template(context: str) -> str:
|
||||
if context == "pr_comment" or context == "pr_review":
|
||||
return (
|
||||
"## Canonical PR State\n\n"
|
||||
"STATE:\n"
|
||||
"WHO_IS_NEXT:\n"
|
||||
"NEXT_ACTION:\n"
|
||||
"NEXT_PROMPT:\n"
|
||||
"```text\n<paste-ready prompt>\n```\n"
|
||||
"WHAT_HAPPENED:\n"
|
||||
"WHY:\n"
|
||||
"ISSUE:\n"
|
||||
"HEAD_SHA:\n"
|
||||
"REVIEW_STATUS:\n"
|
||||
"MERGE_READY:\n"
|
||||
"BLOCKERS:\n"
|
||||
"VALIDATION:\n"
|
||||
"LAST_UPDATED_BY:\n"
|
||||
)
|
||||
if context == "supersession":
|
||||
return (
|
||||
"## Canonical PR State\n\n"
|
||||
"STATE:\nsuperseded\n"
|
||||
"WHO_IS_NEXT:\nreconciler\n"
|
||||
"NEXT_ACTION:\n"
|
||||
"NEXT_PROMPT:\n"
|
||||
"WHY:\n"
|
||||
"CANONICAL_ITEM:\n"
|
||||
"SUPERSEDED_ITEM:\n"
|
||||
"CLOSE_OR_KEEP_OPEN:\n"
|
||||
)
|
||||
if context == "discussion_summary":
|
||||
return (
|
||||
"## Canonical Discussion Summary\n\n"
|
||||
"STATE:\n"
|
||||
"WHO_IS_NEXT:\n"
|
||||
"DECISION:\n"
|
||||
"WHY:\n"
|
||||
"SUBSTANTIVE_COMMENTS:\n"
|
||||
"NEXT_ACTION:\n"
|
||||
"NEXT_PROMPT:\n"
|
||||
)
|
||||
return (
|
||||
"## Canonical Issue State\n\n"
|
||||
"STATE:\n"
|
||||
"WHO_IS_NEXT:\n"
|
||||
"NEXT_ACTION:\n"
|
||||
"NEXT_PROMPT:\n"
|
||||
"```text\n<paste-ready prompt>\n```\n"
|
||||
"WHAT_HAPPENED:\n"
|
||||
"WHY:\n"
|
||||
"RELATED_PRS:\n"
|
||||
"BLOCKERS:\n"
|
||||
"VALIDATION:\n"
|
||||
"LAST_UPDATED_BY:\n"
|
||||
)
|
||||
|
||||
|
||||
def _build_correction_message(
|
||||
*,
|
||||
missing_fields: list[str],
|
||||
vague_fields: list[str],
|
||||
extra_reasons: list[str],
|
||||
context: str,
|
||||
) -> str:
|
||||
parts = [
|
||||
"Canonical comment validation failed (fail closed before posting).",
|
||||
]
|
||||
if missing_fields:
|
||||
parts.append("Missing fields: " + ", ".join(missing_fields) + ".")
|
||||
if vague_fields:
|
||||
parts.append("Vague or invalid fields: " + ", ".join(vague_fields) + ".")
|
||||
parts.extend(extra_reasons)
|
||||
parts.append("Fill the suggested template and retry.")
|
||||
parts.append("Suggested template:\n" + _suggested_template(context))
|
||||
return "\n".join(parts)
|
||||
|
||||
|
||||
def assess_canonical_comment(
|
||||
body: str,
|
||||
*,
|
||||
context: str | None = None,
|
||||
force_workflow: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate outgoing comment text before a Gitea mutation."""
|
||||
text = (body or "").strip()
|
||||
ctx = infer_comment_context(text, explicit=context)
|
||||
|
||||
if not text:
|
||||
return {
|
||||
"allowed": True,
|
||||
"is_workflow_comment": False,
|
||||
"context": ctx,
|
||||
"missing_fields": [],
|
||||
"vague_fields": [],
|
||||
"correction_message": "",
|
||||
"suggested_template": "",
|
||||
}
|
||||
|
||||
if _is_machine_generated(text):
|
||||
return {
|
||||
"allowed": True,
|
||||
"is_workflow_comment": False,
|
||||
"context": ctx,
|
||||
"missing_fields": [],
|
||||
"vague_fields": [],
|
||||
"correction_message": "",
|
||||
"suggested_template": "",
|
||||
}
|
||||
|
||||
workflow = force_workflow or is_workflow_changing_comment(text)
|
||||
if not workflow:
|
||||
return {
|
||||
"allowed": True,
|
||||
"is_workflow_comment": False,
|
||||
"context": ctx,
|
||||
"missing_fields": [],
|
||||
"vague_fields": [],
|
||||
"correction_message": "",
|
||||
"suggested_template": "",
|
||||
}
|
||||
|
||||
fields = _parse_fields(text)
|
||||
required = _required_fields_for_context(ctx)
|
||||
missing = [name for name in required if not (fields.get(name) or "").strip()]
|
||||
|
||||
vague: list[str] = []
|
||||
extra: list[str] = []
|
||||
|
||||
who = (fields.get("WHO_IS_NEXT") or "").strip().lower()
|
||||
if who and who not in VALID_ROLES:
|
||||
vague.append("WHO_IS_NEXT")
|
||||
extra.append(
|
||||
f"WHO_IS_NEXT must be one of: {', '.join(sorted(VALID_ROLES))}."
|
||||
)
|
||||
|
||||
next_action = fields.get("NEXT_ACTION") or ""
|
||||
if next_action and _is_vague_next_action(next_action):
|
||||
vague.append("NEXT_ACTION")
|
||||
|
||||
next_prompt = fields.get("NEXT_PROMPT") or ""
|
||||
if not _next_prompt_ok(next_prompt):
|
||||
if "NEXT_PROMPT" not in missing:
|
||||
vague.append("NEXT_PROMPT")
|
||||
|
||||
state = _state_value(fields)
|
||||
blockers = (fields.get("BLOCKERS") or "").strip()
|
||||
if "blocked" in state:
|
||||
if not blockers or blockers.lower() in {"none", "n/a"}:
|
||||
missing.append("BLOCKERS (unblock condition)")
|
||||
elif not _UNBLOCK_RE.search(blockers):
|
||||
vague.append("BLOCKERS")
|
||||
extra.append("BLOCKED state requires an explicit unblock condition in BLOCKERS.")
|
||||
|
||||
if "superseded" in state:
|
||||
canon = (fields.get("CANONICAL_ITEM") or fields.get("SUPERSEDED_BY") or "").strip()
|
||||
superseded = (fields.get("SUPERSEDED_ITEM") or fields.get("SUPERSEDES") or "").strip()
|
||||
if not canon and "CANONICAL_ITEM" not in missing:
|
||||
missing.append("CANONICAL_ITEM")
|
||||
if not superseded and "SUPERSEDED_ITEM" not in missing:
|
||||
missing.append("SUPERSEDED_ITEM")
|
||||
|
||||
if "ready-to-merge" in state or "ready to merge" in state:
|
||||
head_sha = fields.get("HEAD_SHA") or ""
|
||||
merge_ready = fields.get("MERGE_READY") or ""
|
||||
review_status = fields.get("REVIEW_STATUS") or ""
|
||||
validation = fields.get("VALIDATION") or ""
|
||||
proof_blob = " ".join((head_sha, merge_ready, review_status, validation))
|
||||
has_sha = bool(_FULL_SHA_RE.search(proof_blob) or _SHORT_SHA_RE.search(proof_blob))
|
||||
has_approval = bool(_APPROVAL_PROOF_RE.search(proof_blob))
|
||||
if not has_sha or not has_approval:
|
||||
extra.append(
|
||||
"ready-to-merge STATE requires approval proof and HEAD_SHA in "
|
||||
"REVIEW_STATUS, MERGE_READY, HEAD_SHA, or VALIDATION."
|
||||
)
|
||||
|
||||
if ctx in ("pr_comment", "pr_review"):
|
||||
head_sha = (fields.get("HEAD_SHA") or "").strip()
|
||||
if not head_sha or not _SHORT_SHA_RE.search(head_sha):
|
||||
if "HEAD_SHA" not in missing:
|
||||
missing.append("HEAD_SHA")
|
||||
|
||||
if ctx == "issue_comment" and _PR_REF_RE.search(text):
|
||||
related = (fields.get("RELATED_PRS") or "").strip()
|
||||
if not related or related.lower() in {"none", "n/a", "-"}:
|
||||
missing.append("RELATED_PRS")
|
||||
|
||||
allowed = not missing and not vague and not extra
|
||||
correction = ""
|
||||
if not allowed:
|
||||
correction = _build_correction_message(
|
||||
missing_fields=missing,
|
||||
vague_fields=vague,
|
||||
extra_reasons=extra,
|
||||
context=ctx,
|
||||
)
|
||||
|
||||
return {
|
||||
"allowed": allowed,
|
||||
"is_workflow_comment": True,
|
||||
"context": ctx,
|
||||
"missing_fields": missing,
|
||||
"vague_fields": vague,
|
||||
"extra_reasons": extra,
|
||||
"correction_message": correction,
|
||||
"suggested_template": _suggested_template(ctx) if not allowed else "",
|
||||
}
|
||||
@@ -325,6 +325,17 @@ shared state and manual writes can clobber another session's live lease. Use
|
||||
3. Operator override only when explicitly authorized — record
|
||||
`External-state mutations` and `operator override proof` in the final report.
|
||||
|
||||
**Adoption proof in the live lock response (#477):** when `gitea_lock_issue`
|
||||
adopts an existing own branch, the response carries an `adoption` block with
|
||||
citable fields — `adoption_decision` (`ADOPT`), `adopted` (`true`),
|
||||
`adopted_branch`, `adopted_branch_head`, `matcher_summary` (boundary-safe reason
|
||||
the branch qualified), `competing_branch_check`, and `safe_next_action`. A normal
|
||||
lock instead returns an `adoption_check` block with `adoption_decision`
|
||||
(`NO_MATCH`) and `adopted: false`, so a non-adoption response can never be misread
|
||||
as claiming adoption. Recovery reports should quote the live lock response
|
||||
`adoption`/`adoption_check` block directly instead of inferring adoption from
|
||||
separate offline checks.
|
||||
|
||||
`gitea_create_pr` rejects lock files that lack sanctioned `lock_provenance`
|
||||
metadata. Final-report validation blocks handoffs that hide lock read/write/delete
|
||||
under `External-state mutations: none` or mix author PR creation with reviewer
|
||||
@@ -481,6 +492,25 @@ Root-level matches are listed in `.gitignore` so they never get committed.
|
||||
`gitea_get_runtime_context` and `gitea_lock_issue` surface **warnings** (not
|
||||
hard blocks) when these artifacts are still present.
|
||||
|
||||
## Capability preflight lifetime (#470)
|
||||
|
||||
After `gitea_resolve_task_capability(task=…)` proves the mutation is allowed,
|
||||
interleaved **read-only** calls preserve that proof until a gated mutation
|
||||
consumes it:
|
||||
|
||||
- Safe reads: `gitea_whoami`, `gitea_view_pr`, `gitea_view_issue`, `gitea_list_*`,
|
||||
`gitea_get_runtime_context`, `gitea_check_pr_eligibility`, and related
|
||||
read-only inventory/eligibility tools (see `preflight_contract.py`).
|
||||
- Each mutation consumes the proof once; call `gitea_resolve_task_capability`
|
||||
again immediately before the next mutation on the same task.
|
||||
- Resolving capability for a **different** task replaces the prior task binding.
|
||||
- Workspace edits before resolve, profile switches, or a dirty whoami baseline
|
||||
invalidate proof (fail closed).
|
||||
|
||||
If a mutation fails with “capability has not been resolved” or “task mismatch”,
|
||||
re-run `gitea_resolve_task_capability(task="<mutation>")` immediately before
|
||||
retrying — do not guess or skip the resolve step.
|
||||
|
||||
Implementation work and review work must use separate branch folders. For
|
||||
example, an implementation branch might live under
|
||||
`branches/fix-issue-123-example`, while a review branch for the resulting PR
|
||||
@@ -749,55 +779,6 @@ touched release state names the exact tag/commit and why. Design debates
|
||||
belong in **discussion/RFC issues** (e.g. #100 `profiles.json v2`) — comment
|
||||
on the issue, create no branches/PRs, and end the comment with this handoff.
|
||||
|
||||
## Canonical comment validation (#496)
|
||||
|
||||
Workflow-changing issue/PR/review comments must carry durable next-action
|
||||
state. Casual discussion is still allowed.
|
||||
|
||||
The MCP server runs `canonical_comment_validator.assess_canonical_comment`
|
||||
**before** posting through:
|
||||
|
||||
- `gitea_create_issue_comment`
|
||||
- `gitea_submit_pr_review` / `gitea_dry_run_pr_review` (non-empty review bodies)
|
||||
- `gitea_reconcile_already_landed_pr` when `post_comment=True`
|
||||
- internal structured comment helpers (machine lease/heartbeat markers stay exempt)
|
||||
|
||||
Detection examples:
|
||||
|
||||
- **Allowed:** `Thanks, I will check this.`
|
||||
- **Rejected:** `Blocked, author should fix.` (workflow trigger without canonical fields)
|
||||
|
||||
When validation fails, the tool returns `canonical_comment_validation` with
|
||||
`allowed: false`, `missing_fields`, `vague_fields`, `correction_message`, and
|
||||
`suggested_template`. **No Gitea API call is made.**
|
||||
|
||||
Minimum workflow comment fields:
|
||||
|
||||
```text
|
||||
STATE:
|
||||
WHO_IS_NEXT:
|
||||
NEXT_ACTION:
|
||||
NEXT_PROMPT:
|
||||
WHY:
|
||||
```
|
||||
|
||||
`WHO_IS_NEXT` must be one of: `controller`, `author`, `reviewer`, `merger`,
|
||||
`reconciler`, `user`.
|
||||
|
||||
Issue comments also require `RELATED_PRS`, `BLOCKERS`, and `VALIDATION` when
|
||||
they mention PR work. PR comments/reviews also require `ISSUE`, `HEAD_SHA`,
|
||||
`REVIEW_STATUS`, `MERGE_READY`, `BLOCKERS`, and `VALIDATION`.
|
||||
|
||||
Special states:
|
||||
|
||||
- `STATE: blocked` — `BLOCKERS` must name an explicit unblock condition.
|
||||
- `STATE: superseded` — requires `CANONICAL_ITEM` and `SUPERSEDED_ITEM`.
|
||||
- `STATE: ready-to-merge` — requires approval proof and head SHA in
|
||||
`HEAD_SHA`, `REVIEW_STATUS`, `MERGE_READY`, or `VALIDATION`.
|
||||
|
||||
Final reports must not claim a comment was posted when
|
||||
`canonical_comment_validation.allowed` is false (#496 AC14).
|
||||
|
||||
## Fail-closed behavior
|
||||
|
||||
Before any mutating action the workflow verifies identity, active profile,
|
||||
|
||||
+37
-65
@@ -175,23 +175,6 @@ _PAGINATION_PROOF_RE = re.compile(
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_GIT_FETCH_RE = re.compile(r"\bgit\s+fetch\b", re.IGNORECASE)
|
||||
_CANONICAL_VALIDATION_REJECTED_RE = re.compile(
|
||||
r"canonical comment validation failed|"
|
||||
r"canonical_comment_validation|"
|
||||
r'"allowed"\s*:\s*false',
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_COMMENT_POSTED_CLAIM_RE = re.compile(
|
||||
r"(?:issue comments posted\s*:\s+(?!none\b)\S|"
|
||||
r"pr comments posted\s*:\s+(?!none\b)\S|"
|
||||
r"comment_id\s*[:=]\s*\d+|"
|
||||
r"gitea comment (?:was )?posted|"
|
||||
r"posted (?:issue|pr|canonical) (?:state )?comment|"
|
||||
r"comment posted successfully|"
|
||||
r"mcp/gitea mutations\s*:\s*[^;\n]*comment posted|"
|
||||
r"reconciliation mutations\s*:\s*[^;\n]*comment posted)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_READONLY_DIAG_RE = re.compile(
|
||||
r"read[- ]only diagnostics\s*:\s*(.+)$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
@@ -346,43 +329,6 @@ def _rule_shared_email_disclosure(report_text: str) -> list[dict[str, str]]:
|
||||
)
|
||||
|
||||
|
||||
def _rule_shared_canonical_comment_post_claim(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list[dict] | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
"""#496: reports must not claim comment posted when validator rejected."""
|
||||
text = report_text or ""
|
||||
rejected_in_report = bool(_CANONICAL_VALIDATION_REJECTED_RE.search(text))
|
||||
rejected_in_log = False
|
||||
if action_log:
|
||||
for entry in action_log:
|
||||
validation = entry.get("canonical_comment_validation") or {}
|
||||
if validation.get("allowed") is False:
|
||||
rejected_in_log = True
|
||||
break
|
||||
result = entry.get("result") or {}
|
||||
nested = result.get("canonical_comment_validation") or {}
|
||||
if nested.get("allowed") is False:
|
||||
rejected_in_log = True
|
||||
break
|
||||
if not (rejected_in_report or rejected_in_log):
|
||||
return []
|
||||
if not _COMMENT_POSTED_CLAIM_RE.search(text):
|
||||
return []
|
||||
return [
|
||||
validator_finding(
|
||||
"shared.canonical_comment_post_claim",
|
||||
"block",
|
||||
"MCP/Gitea mutations",
|
||||
"report claims a Gitea comment was posted while canonical comment "
|
||||
"validation rejected the workflow comment",
|
||||
"do not claim comment posted when validator fail-closed; repair "
|
||||
"the canonical comment body and retry posting",
|
||||
)
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_legacy_workspace_mutations(
|
||||
report_text: str,
|
||||
*,
|
||||
@@ -546,6 +492,42 @@ def _rule_reviewer_validation_failure_history(
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_validation_cwd_proof(
|
||||
report_text: str,
|
||||
*,
|
||||
validation_session: dict | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report
|
||||
|
||||
session = validation_session or {}
|
||||
claims = (
|
||||
session.get("validation_ran")
|
||||
or session.get("command")
|
||||
or session.get("baseline_validation_ran")
|
||||
)
|
||||
if not claims and "validation command:" not in (report_text or "").lower():
|
||||
return []
|
||||
|
||||
result = assess_validation_cwd_proof_report(
|
||||
report_text,
|
||||
validation_session=session,
|
||||
)
|
||||
if result.get("proven") or not result.get("claims_validation"):
|
||||
return []
|
||||
severity = "block" if result.get("violations") else "downgrade"
|
||||
return [
|
||||
validator_finding(
|
||||
"reviewer.validation_cwd_proof",
|
||||
severity,
|
||||
"Validation cwd/HEAD proof",
|
||||
reason,
|
||||
result.get("safe_next_action")
|
||||
or "document pwd, HEAD SHA, and explicit cwd before validation",
|
||||
)
|
||||
for reason in (result.get("violations") or result.get("reasons") or ["incomplete"])
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_stale_head_proof(report_text: str) -> list[dict[str, str]]:
|
||||
from pr_work_lease import assess_reviewer_stale_head_final_report
|
||||
|
||||
@@ -1066,15 +1048,10 @@ _SHARED_ISSUE_LOCK_RULES = (
|
||||
_rule_shared_author_reviewer_same_run,
|
||||
)
|
||||
|
||||
_SHARED_CANONICAL_COMMENT_RULES = (
|
||||
_rule_shared_canonical_comment_post_claim,
|
||||
)
|
||||
|
||||
_RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
"review_pr": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
_rule_reviewer_legacy_workspace_mutations,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
@@ -1082,6 +1059,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_git_fetch_readonly,
|
||||
_rule_reviewer_validation_command,
|
||||
_rule_reviewer_validation_failure_history,
|
||||
_rule_reviewer_validation_cwd_proof,
|
||||
_rule_reviewer_validation_structured,
|
||||
_rule_reviewer_linked_issue,
|
||||
_rule_reviewer_baseline_on_failure,
|
||||
@@ -1098,7 +1076,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
"reconcile_already_landed": [
|
||||
_rule_reconcile_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
_rule_reconcile_stale_author_fields,
|
||||
_rule_reconcile_eligible_reviewed,
|
||||
@@ -1111,14 +1088,12 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
"author_issue": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
],
|
||||
"work_issue": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
_rule_conflict_fix_push_proof,
|
||||
@@ -1126,20 +1101,17 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
"issue_filing": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
],
|
||||
"inventory": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
_rule_reconcile_pagination_proof,
|
||||
],
|
||||
"issue_selection": [
|
||||
_rule_shared_controller_handoff,
|
||||
_rule_shared_email_disclosure,
|
||||
*_SHARED_CANONICAL_COMMENT_RULES,
|
||||
*_SHARED_ISSUE_LOCK_RULES,
|
||||
],
|
||||
}
|
||||
|
||||
+134
-97
@@ -165,6 +165,7 @@ _preflight_capability_called = False
|
||||
_preflight_whoami_violation = False
|
||||
_preflight_capability_violation = False
|
||||
_preflight_resolved_role = None
|
||||
_preflight_resolved_task: str | None = None
|
||||
_process_start_porcelain: str | None = None
|
||||
_preflight_whoami_baseline_porcelain: str | None = None
|
||||
_preflight_capability_baseline_porcelain: str | None = None
|
||||
@@ -379,11 +380,30 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def record_preflight_check(type_name: str, resolved_role: str | None = None):
|
||||
def _clear_preflight_capability_state() -> None:
|
||||
"""Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
|
||||
global _preflight_capability_called, _preflight_capability_violation
|
||||
global _preflight_resolved_task
|
||||
global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
|
||||
global _preflight_reviewer_violation_files
|
||||
|
||||
_preflight_capability_called = False
|
||||
_preflight_capability_violation = False
|
||||
_preflight_capability_violation_files = []
|
||||
_preflight_capability_baseline_porcelain = None
|
||||
_preflight_resolved_task = None
|
||||
_preflight_reviewer_violation_files = []
|
||||
|
||||
|
||||
def record_preflight_check(
|
||||
type_name: str,
|
||||
resolved_role: str | None = None,
|
||||
resolved_task: str | None = None,
|
||||
):
|
||||
"""Record a pre-flight check (whoami or capability) with session-scoped deltas."""
|
||||
global _preflight_whoami_called, _preflight_capability_called
|
||||
global _preflight_whoami_violation, _preflight_capability_violation
|
||||
global _preflight_resolved_role
|
||||
global _preflight_resolved_role, _preflight_resolved_task
|
||||
global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
|
||||
global _preflight_whoami_violation_files, _preflight_capability_violation_files
|
||||
global _preflight_reviewer_violation_files
|
||||
@@ -391,13 +411,24 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
|
||||
current = _get_workspace_porcelain()
|
||||
|
||||
if type_name == "whoami":
|
||||
# Fresh whoami restarts the capability step and re-evaluates violations
|
||||
# instead of replaying a sticky record (#252).
|
||||
_preflight_capability_called = False
|
||||
_preflight_capability_violation = False
|
||||
_preflight_capability_violation_files = []
|
||||
_preflight_capability_baseline_porcelain = None
|
||||
_preflight_reviewer_violation_files = []
|
||||
# Re-evaluate whoami violations instead of replaying sticky state (#252).
|
||||
# Interleaved read-only whoami must not clear a valid capability proof (#469).
|
||||
saved_capability = None
|
||||
preserve_capability = (
|
||||
_preflight_capability_called
|
||||
and _preflight_whoami_called
|
||||
and not _preflight_whoami_violation
|
||||
)
|
||||
if preserve_capability:
|
||||
saved_capability = (
|
||||
_preflight_capability_violation,
|
||||
list(_preflight_capability_violation_files),
|
||||
_preflight_capability_baseline_porcelain,
|
||||
_preflight_resolved_role,
|
||||
_preflight_resolved_task,
|
||||
)
|
||||
else:
|
||||
_clear_preflight_capability_state()
|
||||
|
||||
process_start = _ensure_process_start_porcelain()
|
||||
whoami_delta = _new_tracked_changes_since(process_start, current)
|
||||
@@ -405,6 +436,20 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
|
||||
_preflight_whoami_violation_files = whoami_delta
|
||||
_preflight_whoami_baseline_porcelain = current
|
||||
_preflight_whoami_called = True
|
||||
|
||||
if (
|
||||
preserve_capability
|
||||
and saved_capability is not None
|
||||
and not whoami_delta
|
||||
):
|
||||
(
|
||||
_preflight_capability_violation,
|
||||
_preflight_capability_violation_files,
|
||||
_preflight_capability_baseline_porcelain,
|
||||
_preflight_resolved_role,
|
||||
_preflight_resolved_task,
|
||||
) = saved_capability
|
||||
_preflight_capability_called = True
|
||||
elif type_name == "capability":
|
||||
baseline = _preflight_whoami_baseline_porcelain or ""
|
||||
capability_delta = _new_tracked_changes_since(baseline, current)
|
||||
@@ -414,11 +459,18 @@ def record_preflight_check(type_name: str, resolved_role: str | None = None):
|
||||
_preflight_capability_called = True
|
||||
if resolved_role:
|
||||
_preflight_resolved_role = resolved_role
|
||||
if resolved_task:
|
||||
_preflight_resolved_task = resolved_task
|
||||
|
||||
|
||||
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
|
||||
"""#274: author mutations must run from a branches/ session worktree."""
|
||||
if _preflight_resolved_role == "reviewer":
|
||||
"""#274: author file/branch mutations must run from a branches/ worktree.
|
||||
|
||||
Reviewer and reconciler roles are exempt: reconciler ``close_pr`` is a
|
||||
Gitea metadata mutation and must not require ``GITEA_AUTHOR_WORKTREE``
|
||||
(#468).
|
||||
"""
|
||||
if _preflight_resolved_role in ("reviewer", "reconciler"):
|
||||
return
|
||||
ctx = _resolve_author_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
@@ -434,7 +486,11 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
|
||||
)
|
||||
|
||||
|
||||
def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
|
||||
def verify_preflight_purity(
|
||||
remote: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
task: str | None = None,
|
||||
):
|
||||
"""Verify that identity and capability were verified prior to session edits."""
|
||||
global _preflight_reviewer_violation_files
|
||||
|
||||
@@ -451,7 +507,27 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
|
||||
)
|
||||
if not _preflight_capability_called:
|
||||
raise RuntimeError(
|
||||
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
|
||||
preflight_contract.format_missing_capability_error(task)
|
||||
)
|
||||
if (
|
||||
task is not None
|
||||
and _preflight_resolved_task is not None
|
||||
and task != _preflight_resolved_task
|
||||
):
|
||||
raise RuntimeError(
|
||||
preflight_contract.format_task_mismatch_error(
|
||||
_preflight_resolved_task, task
|
||||
)
|
||||
)
|
||||
if (
|
||||
task is not None
|
||||
and _preflight_resolved_task is not None
|
||||
and task != _preflight_resolved_task
|
||||
):
|
||||
raise RuntimeError(
|
||||
"Pre-flight task mismatch: "
|
||||
f"resolved '{_preflight_resolved_task}' but mutation requires "
|
||||
f"'{task}' (fail closed)"
|
||||
)
|
||||
|
||||
ctx = _resolve_author_mutation_context(worktree_path)
|
||||
@@ -508,6 +584,7 @@ def verify_preflight_purity(remote: str | None = None, worktree_path: str | None
|
||||
)
|
||||
|
||||
_enforce_branches_only_author_mutation(worktree_path)
|
||||
_clear_preflight_capability_state()
|
||||
|
||||
from mcp.server.fastmcp import FastMCP # noqa: E402
|
||||
|
||||
@@ -539,6 +616,7 @@ import stacked_pr_support # noqa: E402
|
||||
import merge_approval_gate # noqa: E402
|
||||
import already_landed_reconcile # noqa: E402
|
||||
import author_mutation_worktree # noqa: E402
|
||||
import preflight_contract # noqa: E402
|
||||
import issue_claim_heartbeat # noqa: E402
|
||||
import issue_work_duplicate_gate # noqa: E402
|
||||
import reviewer_pr_lease # noqa: E402
|
||||
@@ -548,7 +626,6 @@ import reconciliation_workflow # noqa: E402
|
||||
import review_merge_state_machine # noqa: E402
|
||||
import pr_work_lease # noqa: E402
|
||||
import native_mcp_preference # noqa: E402
|
||||
import canonical_comment_validator as ccv # noqa: E402
|
||||
|
||||
|
||||
# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
|
||||
@@ -1224,7 +1301,7 @@ def gitea_create_issue(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
|
||||
base = repo_api_url(h, o, r)
|
||||
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
|
||||
closed_issues = api_get_all(
|
||||
@@ -1357,7 +1434,7 @@ def gitea_lock_issue(
|
||||
)
|
||||
else:
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
|
||||
verify_preflight_purity(remote, worktree_path=resolved_worktree)
|
||||
verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
|
||||
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=resolved_worktree,
|
||||
current_branch=git_state.get("current_branch"),
|
||||
@@ -1485,6 +1562,14 @@ def gitea_lock_issue(
|
||||
f"Adopted existing branch '{branch_name}' and locked issue "
|
||||
f"#{issue_number} for recovery (fail-closed check complete)."
|
||||
)
|
||||
else:
|
||||
# #477 AC2: normal (no-adoption) lock responses carry explicit,
|
||||
# adoption-free proof metadata so they stay clear and cannot be
|
||||
# misread as claiming a branch was adopted.
|
||||
result["adoption_check"] = issue_lock_adoption.build_non_adoption_lock_proof(
|
||||
issue_number=issue_number,
|
||||
branch_name=branch_name,
|
||||
)
|
||||
if agent_artifacts:
|
||||
result["warnings"] = [
|
||||
"Agent temp artifacts at repo root (delete before implementation): "
|
||||
@@ -1573,7 +1658,7 @@ def gitea_create_pr(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
|
||||
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
|
||||
@@ -2589,7 +2674,12 @@ def _list_pr_lease_comments(
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
api = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
|
||||
comments = api_request("GET", api, auth) or []
|
||||
comments = api_request("GET", api, auth)
|
||||
# Fail safe to no lease comments when the API returns a non-list payload
|
||||
# (e.g. an error object such as an HTTP 401 body): lease state can only be
|
||||
# proven from real comment entries, never inferred from an error shape (#485).
|
||||
if not isinstance(comments, list):
|
||||
return []
|
||||
return list(comments[:limit])
|
||||
|
||||
|
||||
@@ -2634,7 +2724,7 @@ def _evaluate_pr_review_submission(
|
||||
final_review_decision_ready: bool = False,
|
||||
) -> dict:
|
||||
"""Shared gate chain for live submit and dry-run review tools."""
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="review_pr")
|
||||
action = (action or "").strip().lower()
|
||||
result = {
|
||||
"requested_action": action,
|
||||
@@ -2747,13 +2837,6 @@ def _evaluate_pr_review_submission(
|
||||
result["pr_work_lease"] = lease_block
|
||||
return result
|
||||
|
||||
if (body or "").strip():
|
||||
gate = _canonical_comment_gate(body, context="pr_review")
|
||||
if gate["blocked"]:
|
||||
reasons.extend(gate["reasons"])
|
||||
result["canonical_comment_validation"] = gate["canonical_comment_validation"]
|
||||
return result
|
||||
|
||||
result["would_perform"] = True
|
||||
if not live:
|
||||
reasons.append(
|
||||
@@ -3148,12 +3231,12 @@ def gitea_edit_pr(
|
||||
if not payload:
|
||||
raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
closing = payload.get("state") == "closed"
|
||||
verify_preflight_purity(remote, task="close_pr" if closing else None)
|
||||
|
||||
# PR closure is a first-class capability, distinct from retitling or
|
||||
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
|
||||
# never touches the network.
|
||||
closing = payload.get("state") == "closed"
|
||||
if closing:
|
||||
gate_reasons = _profile_operation_gate("gitea.pr.close")
|
||||
if gate_reasons:
|
||||
@@ -3398,7 +3481,7 @@ def gitea_commit_files(
|
||||
branch="",
|
||||
)
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="commit_files")
|
||||
processed_files, source_proofs = _prepare_commit_payload_files(files)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
@@ -3499,7 +3582,7 @@ def gitea_merge_pr(
|
||||
reasons/gates passed or blocked, and merge result / merge commit if
|
||||
available. Never secrets.
|
||||
"""
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="merge_pr")
|
||||
do = (do or "").strip().lower()
|
||||
result = {
|
||||
"performed": False,
|
||||
@@ -4028,7 +4111,7 @@ def gitea_delete_branch(
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="delete_branch")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
import urllib.parse
|
||||
@@ -4137,7 +4220,7 @@ def gitea_reconcile_merged_cleanups(
|
||||
report["executed"] = False
|
||||
return {"success": True, "performed": False, **report}
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="reconcile_merged_cleanups")
|
||||
actions: list[dict] = []
|
||||
for entry in report.get("entries") or []:
|
||||
head_branch = entry.get("head_branch") or ""
|
||||
@@ -4451,7 +4534,7 @@ def gitea_reconcile_already_landed_pr(
|
||||
)
|
||||
return result
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="reconcile_already_landed_pr")
|
||||
|
||||
if post_comment and comment_body.strip():
|
||||
comment_block = _profile_operation_gate("gitea.pr.comment")
|
||||
@@ -4462,12 +4545,6 @@ def gitea_reconcile_already_landed_pr(
|
||||
"gitea.pr.comment"
|
||||
)
|
||||
return result
|
||||
gate = _canonical_comment_gate(comment_body)
|
||||
if gate["blocked"]:
|
||||
result["success"] = False
|
||||
result["reasons"].extend(gate["reasons"])
|
||||
result["canonical_comment_validation"] = gate["canonical_comment_validation"]
|
||||
return result
|
||||
comment_url = f"{base}/issues/{pr_number}/comments"
|
||||
with _audited(
|
||||
"comment_pr",
|
||||
@@ -4560,7 +4637,7 @@ def gitea_close_issue(
|
||||
task_capability_map.required_permission("close_issue"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="close_issue")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
|
||||
@@ -4988,7 +5065,7 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
"permission_report": _permission_block_report("gitea.pr.comment"),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="review_pr")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
profile = get_profile()
|
||||
@@ -5089,7 +5166,7 @@ def gitea_heartbeat_reviewer_pr_lease(
|
||||
],
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="review_pr")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
body = reviewer_pr_lease.format_lease_body(
|
||||
@@ -5206,7 +5283,12 @@ def gitea_list_issue_comments(
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
|
||||
comments = api_request("GET", api, auth) or []
|
||||
comments = api_request("GET", api, auth)
|
||||
# Fail safe to no comments when the API returns a non-list payload (e.g. an
|
||||
# error object such as an HTTP 401 body) so listing never crashes on a
|
||||
# malformed response (#485).
|
||||
if not isinstance(comments, list):
|
||||
comments = []
|
||||
reveal = _reveal_endpoints()
|
||||
out = []
|
||||
for c in comments[:limit]:
|
||||
@@ -5260,7 +5342,7 @@ def gitea_create_issue_comment(
|
||||
(permission blocks also carry a structured 'permission_report',
|
||||
#142).
|
||||
"""
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="comment_issue")
|
||||
gate_reasons = _profile_operation_gate("gitea.issue.comment")
|
||||
reasons = list(gate_reasons)
|
||||
if not (body or "").strip():
|
||||
@@ -5273,15 +5355,6 @@ def gitea_create_issue_comment(
|
||||
blocked["permission_report"] = _permission_block_report(
|
||||
"gitea.issue.comment")
|
||||
return blocked
|
||||
canon_gate = _canonical_comment_gate(body, context="issue_comment")
|
||||
if canon_gate["blocked"]:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"issue_number": issue_number,
|
||||
"reasons": canon_gate["reasons"],
|
||||
"canonical_comment_validation": canon_gate["canonical_comment_validation"],
|
||||
}
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
|
||||
@@ -6733,34 +6806,6 @@ def gitea_audit_config() -> dict:
|
||||
return report
|
||||
|
||||
|
||||
def _canonical_comment_gate(
|
||||
body: str,
|
||||
*,
|
||||
context: str | None = None,
|
||||
force_workflow: bool = False,
|
||||
) -> dict:
|
||||
"""Fail-closed canonical state validation before comment/review POST (#496)."""
|
||||
assessment = ccv.assess_canonical_comment(
|
||||
body,
|
||||
context=context,
|
||||
force_workflow=force_workflow,
|
||||
)
|
||||
if assessment.get("allowed"):
|
||||
return {
|
||||
"blocked": False,
|
||||
"canonical_comment_validation": assessment,
|
||||
"reasons": [],
|
||||
}
|
||||
correction = (assessment.get("correction_message") or "").strip()
|
||||
return {
|
||||
"blocked": True,
|
||||
"canonical_comment_validation": assessment,
|
||||
"reasons": [
|
||||
correction or "canonical comment validation failed (fail closed before posting)"
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _post_structured_issue_comment(
|
||||
*,
|
||||
issue_number: int,
|
||||
@@ -6770,18 +6815,8 @@ def _post_structured_issue_comment(
|
||||
org: str | None,
|
||||
repo: str | None,
|
||||
audit_op: str = "create_issue_comment",
|
||||
comment_context: str | None = None,
|
||||
) -> dict:
|
||||
"""Post an issue-thread comment after permission gates already passed."""
|
||||
gate = _canonical_comment_gate(body, context=comment_context)
|
||||
if gate["blocked"]:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"issue_number": issue_number,
|
||||
"reasons": gate["reasons"],
|
||||
"canonical_comment_validation": gate["canonical_comment_validation"],
|
||||
}
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
api = f"{repo_api_url(h, o, r)}/issues/{issue_number}/comments"
|
||||
@@ -6841,7 +6876,7 @@ def gitea_mark_issue(
|
||||
task_capability_map.required_permission("mark_issue"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -6919,7 +6954,7 @@ def gitea_post_heartbeat(
|
||||
task_capability_map.required_permission("post_heartbeat"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="post_heartbeat")
|
||||
active_profile = profile or get_profile().get("profile_name")
|
||||
body = issue_claim_heartbeat.format_heartbeat_body(
|
||||
kind="progress",
|
||||
@@ -6958,7 +6993,9 @@ def gitea_acquire_conflict_fix_lease(
|
||||
task_capability_map.required_permission("comment_issue"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
verify_preflight_purity(
|
||||
remote, worktree_path=worktree_path, task="comment_issue"
|
||||
)
|
||||
comments = _list_pr_lease_comments(
|
||||
pr_number,
|
||||
remote=remote,
|
||||
@@ -7162,7 +7199,7 @@ def gitea_cleanup_stale_claims(
|
||||
task_capability_map.required_permission("cleanup_stale_claims"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="cleanup_stale_claims")
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
@@ -7316,7 +7353,7 @@ def gitea_set_issue_labels(
|
||||
task_capability_map.required_permission("set_issue_labels"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote)
|
||||
verify_preflight_purity(remote, task="set_issue_labels")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -7553,7 +7590,7 @@ def gitea_resolve_task_capability(
|
||||
"exact_safe_next_action": next_safe_action,
|
||||
}
|
||||
|
||||
record_preflight_check("capability", required_role)
|
||||
record_preflight_check("capability", required_role, resolved_task=task)
|
||||
|
||||
# Try automatic dispatch switching
|
||||
_ensure_matching_profile(required_permission, required_role, remote, host)
|
||||
|
||||
@@ -20,6 +20,43 @@ ADOPT = "adopt_existing_branch"
|
||||
BLOCK_COMPETING = "block_competing_branch"
|
||||
NO_MATCH = "no_matching_branch"
|
||||
|
||||
# Citable decision labels aligned with the ``assess_own_branch_adoption``
|
||||
# outcomes, surfaced verbatim in the live ``gitea_lock_issue`` response so
|
||||
# recovery reports (#473-style) can quote the lock tool output directly
|
||||
# instead of inferring adoption from separate offline checks (#477).
|
||||
DECISION_LABELS = {
|
||||
ADOPT: "ADOPT",
|
||||
BLOCK_COMPETING: "BLOCK_COMPETING",
|
||||
NO_MATCH: "NO_MATCH",
|
||||
}
|
||||
|
||||
_SAFE_NEXT_ACTIONS = {
|
||||
ADOPT: (
|
||||
"Own existing branch adopted for lock recovery; proceed to "
|
||||
"gitea_create_pr for this issue and cite this adoption proof."
|
||||
),
|
||||
BLOCK_COMPETING: (
|
||||
"Competing same-issue branch(es) exist; resolve branch ownership "
|
||||
"before locking. No adoption performed (fail closed)."
|
||||
),
|
||||
NO_MATCH: (
|
||||
"No existing branch carries this issue marker; normal lock path "
|
||||
"applied. No adoption performed."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def decision_label(outcome: str) -> str:
|
||||
"""Map an ``assess_own_branch_adoption`` outcome to its citable label."""
|
||||
return DECISION_LABELS.get(outcome, "UNKNOWN")
|
||||
|
||||
|
||||
def safe_next_action(outcome: str) -> str:
|
||||
"""Return the safe next action string for an adoption *outcome*."""
|
||||
return _SAFE_NEXT_ACTIONS.get(
|
||||
outcome, "Unknown adoption outcome; treat as fail closed."
|
||||
)
|
||||
|
||||
|
||||
def _branch_name(entry) -> str:
|
||||
if isinstance(entry, dict):
|
||||
@@ -128,6 +165,42 @@ def assess_own_branch_adoption(
|
||||
}
|
||||
|
||||
|
||||
def _matcher_summary(issue_number: int, assessment: dict) -> str:
|
||||
"""Explain, citably, why the assessed branch did or did not qualify.
|
||||
|
||||
Names the numeric word-boundary rule so reports can show that
|
||||
``issue-42`` was not matched inside ``issue-420`` (#440 / #477 AC3).
|
||||
"""
|
||||
outcome = assessment.get("outcome")
|
||||
matched = assessment.get("matched_branch")
|
||||
competing = assessment.get("competing_branches") or []
|
||||
if outcome == ADOPT and matched:
|
||||
return (
|
||||
f"branch '{matched}' exactly matches the issue-{int(issue_number)} "
|
||||
f"marker (numeric word-boundary; 'issue-{int(issue_number)}' is not "
|
||||
f"matched inside 'issue-{int(issue_number)}0')"
|
||||
)
|
||||
if outcome == BLOCK_COMPETING:
|
||||
return (
|
||||
f"competing same-issue branch(es) {competing} carry the "
|
||||
f"issue-{int(issue_number)} marker but are not the requested "
|
||||
f"branch; ownership is ambiguous (fail closed)"
|
||||
)
|
||||
return (
|
||||
f"no existing branch carries the issue-{int(issue_number)} marker "
|
||||
f"under the numeric word-boundary rule"
|
||||
)
|
||||
|
||||
|
||||
def _competing_branch_check(assessment: dict) -> dict:
|
||||
"""Structured competing-branch verdict for the proof block."""
|
||||
competing = list(assessment.get("competing_branches") or [])
|
||||
return {
|
||||
"result": "blocked" if competing else "clear",
|
||||
"competing_branches": competing,
|
||||
}
|
||||
|
||||
|
||||
def build_adoption_proof(
|
||||
*,
|
||||
issue_number: int,
|
||||
@@ -143,7 +216,18 @@ def build_adoption_proof(
|
||||
Requirement #4: adoption results must carry issue number, branch name,
|
||||
branch head commit, adoption reason, no-existing-PR proof, no-competing-
|
||||
live-lock proof, and lock file path/status.
|
||||
|
||||
#477: additionally surface explicit, citable adoption-proof fields tied to
|
||||
the ``assess_own_branch_adoption`` outcome (``adoption_decision``,
|
||||
``adopted``, ``adopted_branch``, ``adopted_branch_head``,
|
||||
``matcher_summary``, ``competing_branch_check``, ``safe_next_action``) so a
|
||||
recovery session can quote the live lock response directly. The explicit
|
||||
fields are populated for any outcome; ``adopted_branch`` /
|
||||
``adopted_branch_head`` are set only when the outcome is ADOPT so a
|
||||
non-adoption proof can never be misread as claiming adoption.
|
||||
"""
|
||||
outcome = assessment.get("outcome")
|
||||
adopted = outcome == ADOPT
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"branch_name": branch_name,
|
||||
@@ -153,4 +237,36 @@ def build_adoption_proof(
|
||||
"no_competing_live_lock_proof": bool(competing_lock_checked),
|
||||
"lock_file_path": lock_file_path,
|
||||
"lock_file_status": lock_file_status,
|
||||
# Explicit citable fields (#477).
|
||||
"adoption_decision": decision_label(outcome),
|
||||
"adopted": adopted,
|
||||
"adopted_branch": branch_name if adopted else None,
|
||||
"adopted_branch_head": assessment.get("matched_head_sha") if adopted else None,
|
||||
"matcher_summary": _matcher_summary(issue_number, assessment),
|
||||
"competing_branch_check": _competing_branch_check(assessment),
|
||||
"safe_next_action": safe_next_action(outcome),
|
||||
}
|
||||
|
||||
|
||||
def build_non_adoption_lock_proof(*, issue_number: int, branch_name: str) -> dict:
|
||||
"""Safe, adoption-free proof metadata for a normal (NO_MATCH) lock.
|
||||
|
||||
Requirement #477 AC2: non-adoption lock responses must stay clear and must
|
||||
not imply adoption. This returns explicit ``adopted: False`` metadata with
|
||||
the ``NO_MATCH`` decision so a normal lock response can carry citable proof
|
||||
without ever asserting a branch was adopted.
|
||||
"""
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"branch_name": branch_name,
|
||||
"adoption_decision": DECISION_LABELS[NO_MATCH],
|
||||
"adopted": False,
|
||||
"adopted_branch": None,
|
||||
"adopted_branch_head": None,
|
||||
"matcher_summary": (
|
||||
f"no existing branch carries the issue-{int(issue_number)} marker; "
|
||||
f"normal lock path (no adoption)"
|
||||
),
|
||||
"competing_branch_check": {"result": "clear", "competing_branches": []},
|
||||
"safe_next_action": safe_next_action(NO_MATCH),
|
||||
}
|
||||
@@ -0,0 +1,66 @@
|
||||
"""Capability preflight lifetime contract (#470).
|
||||
|
||||
Defines which read-only MCP tools preserve an existing capability proof and the
|
||||
canonical sequencing operators must follow before mutations.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
# Read-only tools that must not invalidate capability proof (#470).
|
||||
# gitea_whoami is read-only but re-pins identity; capability is preserved when
|
||||
# identity was already verified clean (#469).
|
||||
READ_ONLY_PREFLIGHT_TOOLS = frozenset({
|
||||
"gitea_whoami",
|
||||
"gitea_get_authenticated_user",
|
||||
"gitea_get_current_user",
|
||||
"gitea_view_pr",
|
||||
"gitea_view_issue",
|
||||
"gitea_list_prs",
|
||||
"gitea_list_issues",
|
||||
"gitea_list_issue_comments",
|
||||
"gitea_get_runtime_context",
|
||||
"gitea_resolve_task_capability",
|
||||
"gitea_check_pr_eligibility",
|
||||
"gitea_get_pr_review_feedback",
|
||||
"gitea_assess_work_issue_duplicate",
|
||||
"gitea_route_task_session",
|
||||
"gitea_audit_config",
|
||||
"gitea_list_labels",
|
||||
"gitea_get_profile",
|
||||
"gitea_list_profiles",
|
||||
})
|
||||
|
||||
PREFLIGHT_CONTRACT_SUMMARY = (
|
||||
"Capability preflight is session-scoped per MCP process, profile, and task. "
|
||||
"After gitea_resolve_task_capability(task=...), interleaved read-only calls "
|
||||
"(whoami, view_*, list_*, get_runtime_context, eligibility checks) preserve "
|
||||
"the proof until a gated mutation consumes it. Each mutation consumes the proof "
|
||||
"once; re-resolve immediately before the next mutation. A new resolve for a "
|
||||
"different task replaces the prior task binding. Profile/session changes or "
|
||||
"workspace edits before resolve invalidate proof."
|
||||
)
|
||||
|
||||
|
||||
def format_missing_capability_error(task: str | None = None) -> str:
|
||||
base = (
|
||||
"Pre-flight order violation: Task capability "
|
||||
"(gitea_resolve_task_capability) has not been resolved (fail closed)"
|
||||
)
|
||||
if task:
|
||||
return (
|
||||
f"{base}. Re-run gitea_resolve_task_capability(task=\"{task}\") "
|
||||
"immediately before this mutation."
|
||||
)
|
||||
return (
|
||||
f"{base}. Re-run gitea_resolve_task_capability for the mutation task "
|
||||
"immediately before acting."
|
||||
)
|
||||
|
||||
|
||||
def format_task_mismatch_error(resolved: str, required: str) -> str:
|
||||
return (
|
||||
"Pre-flight task mismatch: "
|
||||
f"resolved '{resolved}' but mutation requires '{required}' (fail closed). "
|
||||
f"Re-run gitea_resolve_task_capability(task=\"{required}\") "
|
||||
"immediately before this mutation."
|
||||
)
|
||||
@@ -5515,6 +5515,13 @@ def assess_validation_failure_history_report(report_text, **kwargs):
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_validation_cwd_proof_report(report_text, **kwargs):
|
||||
"""#398: validation commands require explicit worktree cwd and HEAD proof."""
|
||||
from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report as _assess
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_already_landed_classification_report(report_text, **kwargs):
|
||||
"""#295: already-landed PRs are reconciliation-only, not review eligible."""
|
||||
from reviewer_already_landed_classification import (
|
||||
|
||||
@@ -0,0 +1,233 @@
|
||||
"""Explicit worktree and cwd proof for PR review validation (#398)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
|
||||
|
||||
_PWD_RE = re.compile(
|
||||
r"(?:^|\n)\s*(?:pwd|working\s+directory|cwd)\s*:\s*(\S+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_HEAD_RE = re.compile(
|
||||
r"(?:git\s+rev-parse\s+head|observed\s+head\s+sha)\s*:\s*([0-9a-f]{7,40})",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_EXPECTED_HEAD_RE = re.compile(
|
||||
r"(?:expected\s+(?:pr\s+)?head\s+sha|candidate\s+head\s+sha|pinned\s+head)\s*:\s*([0-9a-f]{7,40})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_STATUS_RE = re.compile(
|
||||
r"git\s+status\s+(?:--short\s+--branch|--short|-sb)\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_VALIDATION_CMD_RE = re.compile(
|
||||
r"validation\s+command\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_GIT_C_CMD_RE = re.compile(r"git\s+-C\s+\S+", re.IGNORECASE)
|
||||
_CD_CMD_RE = re.compile(r"(?:^|&&\s*)cd\s+\S+", re.IGNORECASE)
|
||||
_BASELINE_CWD_RE = re.compile(
|
||||
r"baseline\s+(?:worktree|working\s+directory|cwd)\s*:\s*(\S+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BASELINE_SHA_RE = re.compile(
|
||||
r"baseline\s+(?:target\s+)?sha\s*:\s*([0-9a-f]{7,40})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BASELINE_CMD_RE = re.compile(
|
||||
r"baseline\s+validation\s+command\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _normalize_path(path: str) -> str:
|
||||
return (path or "").replace("\\", "/").rstrip("/")
|
||||
|
||||
|
||||
def _path_under_branches(path: str, project_root: str | None = None) -> bool:
|
||||
normalized = _normalize_path(path)
|
||||
if not normalized:
|
||||
return False
|
||||
if "/branches/" in f"{normalized}/":
|
||||
return True
|
||||
if normalized.endswith("/branches"):
|
||||
return True
|
||||
if project_root:
|
||||
root = _normalize_path(project_root)
|
||||
if normalized.startswith(f"{root}/"):
|
||||
rel = normalized[len(root) + 1 :]
|
||||
return rel == "branches" or rel.startswith("branches/")
|
||||
return False
|
||||
|
||||
|
||||
def _expand_sha(sha: str) -> str:
|
||||
return (sha or "").strip().lower()
|
||||
|
||||
|
||||
def _sha_matches(expected: str, observed: str) -> bool:
|
||||
exp = _expand_sha(expected)
|
||||
obs = _expand_sha(observed)
|
||||
if not exp or not obs:
|
||||
return False
|
||||
if len(exp) == 40 and len(obs) == 40:
|
||||
return exp == obs
|
||||
return obs.startswith(exp) or exp.startswith(obs)
|
||||
|
||||
|
||||
def _command_has_explicit_cwd(command: str, cwd: str) -> bool:
|
||||
text = (command or "").strip()
|
||||
if not text:
|
||||
return False
|
||||
if _GIT_C_CMD_RE.search(text):
|
||||
return True
|
||||
if _CD_CMD_RE.search(text):
|
||||
return True
|
||||
if cwd and cwd in text:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def assess_validation_cwd_proof_report(
|
||||
report_text: str,
|
||||
*,
|
||||
validation_session: dict | None = None,
|
||||
project_root: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Require cwd/HEAD proof before reviewer validation claims (#398)."""
|
||||
text = report_text or ""
|
||||
session = dict(validation_session or {})
|
||||
reasons: list[str] = []
|
||||
violations: list[str] = []
|
||||
|
||||
claims_validation = bool(
|
||||
session.get("validation_ran")
|
||||
or _VALIDATION_CMD_RE.search(text)
|
||||
or session.get("command")
|
||||
)
|
||||
if not claims_validation:
|
||||
return {
|
||||
"proven": True,
|
||||
"block": False,
|
||||
"claims_validation": False,
|
||||
"reasons": [],
|
||||
"violations": [],
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
|
||||
expected_head = (
|
||||
session.get("expected_head_sha")
|
||||
or session.get("candidate_head_sha")
|
||||
or ""
|
||||
).strip()
|
||||
if not expected_head:
|
||||
match = _EXPECTED_HEAD_RE.search(text)
|
||||
expected_head = (match.group(1) if match else "").strip()
|
||||
|
||||
observed_head = (session.get("observed_head_sha") or "").strip()
|
||||
if not observed_head:
|
||||
match = _HEAD_RE.search(text)
|
||||
observed_head = (match.group(1) if match else "").strip()
|
||||
|
||||
cwd = (
|
||||
session.get("working_directory")
|
||||
or session.get("cwd")
|
||||
or session.get("pwd")
|
||||
or ""
|
||||
).strip()
|
||||
if not cwd:
|
||||
match = _PWD_RE.search(text)
|
||||
cwd = (match.group(1) if match else "").strip().rstrip(",.;")
|
||||
|
||||
command = (session.get("command") or "").strip()
|
||||
if not command:
|
||||
match = _VALIDATION_CMD_RE.search(text)
|
||||
command = (match.group(1) if match else "").strip().rstrip(".;")
|
||||
|
||||
if not cwd:
|
||||
reasons.append(
|
||||
"validation claimed without pwd/working-directory proof (#398)"
|
||||
)
|
||||
elif not _path_under_branches(cwd, project_root):
|
||||
violations.append(
|
||||
f"validation cwd {cwd!r} is not under branches/ (#398)"
|
||||
)
|
||||
reasons.append(
|
||||
"reviewer validation must run from a branches/ worktree, "
|
||||
"not the main checkout (#398)"
|
||||
)
|
||||
|
||||
if not observed_head:
|
||||
reasons.append(
|
||||
"validation claimed without git rev-parse HEAD / observed HEAD SHA "
|
||||
"proof (#398)"
|
||||
)
|
||||
elif expected_head and not _sha_matches(expected_head, observed_head):
|
||||
violations.append(
|
||||
f"observed HEAD {observed_head} does not match expected "
|
||||
f"PR head {expected_head} (#398)"
|
||||
)
|
||||
reasons.append("validation HEAD SHA must match pinned PR head (#398)")
|
||||
|
||||
if not _STATUS_RE.search(text) and session.get("git_status") is None:
|
||||
reasons.append(
|
||||
"validation claimed without git status --short --branch proof (#398)"
|
||||
)
|
||||
|
||||
if command and cwd and not _command_has_explicit_cwd(command, cwd):
|
||||
if session.get("tool_working_directory") is not True:
|
||||
reasons.append(
|
||||
"validation command must use git -C <worktree>, "
|
||||
"cd <worktree> && ..., or tool-provided cwd metadata (#398)"
|
||||
)
|
||||
|
||||
baseline_ran = bool(
|
||||
session.get("baseline_validation_ran")
|
||||
or _BASELINE_CMD_RE.search(text)
|
||||
)
|
||||
if baseline_ran:
|
||||
baseline_cwd = (session.get("baseline_worktree_path") or "").strip()
|
||||
if not baseline_cwd:
|
||||
match = _BASELINE_CWD_RE.search(text)
|
||||
baseline_cwd = (match.group(1) if match else "").strip().rstrip(",.;")
|
||||
if not baseline_cwd or not _path_under_branches(baseline_cwd, project_root):
|
||||
reasons.append(
|
||||
"baseline validation claimed without baseline worktree cwd "
|
||||
"under branches/ (#398)"
|
||||
)
|
||||
baseline_sha = (session.get("baseline_target_sha") or "").strip()
|
||||
if not baseline_sha:
|
||||
match = _BASELINE_SHA_RE.search(text)
|
||||
baseline_sha = (match.group(1) if match else "").strip()
|
||||
if not baseline_sha:
|
||||
reasons.append(
|
||||
"baseline validation claimed without baseline target SHA (#398)"
|
||||
)
|
||||
baseline_cmd = (session.get("baseline_command") or "").strip()
|
||||
if not baseline_cmd:
|
||||
match = _BASELINE_CMD_RE.search(text)
|
||||
baseline_cmd = (match.group(1) if match else "").strip()
|
||||
if not baseline_cmd:
|
||||
reasons.append(
|
||||
"baseline validation claimed without exact baseline command (#398)"
|
||||
)
|
||||
|
||||
proven = not reasons and not violations
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": bool(violations) or not proven,
|
||||
"claims_validation": True,
|
||||
"expected_head_sha": expected_head or None,
|
||||
"observed_head_sha": observed_head or None,
|
||||
"working_directory": cwd or None,
|
||||
"reasons": reasons,
|
||||
"violations": violations,
|
||||
"safe_next_action": (
|
||||
"before validation record pwd, git rev-parse HEAD, git status, "
|
||||
"expected PR head SHA; run commands with git -C or cd in the same line"
|
||||
if not proven
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
@@ -626,6 +626,28 @@ Do not claim “full-suite failures are pre-existing” unless baseline proof is
|
||||
|
||||
## 23. Validation command proof rule
|
||||
|
||||
Before any diff, test, or compile validation, record in the same command
|
||||
transcript or final report:
|
||||
|
||||
* `pwd` or explicit working directory
|
||||
* `git rev-parse HEAD`
|
||||
* `git status --short --branch`
|
||||
* expected PR head SHA (candidate head SHA)
|
||||
|
||||
Validation commands must use one of:
|
||||
|
||||
* `git -C <review_worktree> ...`
|
||||
* `cd <review_worktree> && ...` in the same command
|
||||
* tool-provided explicit working-directory metadata
|
||||
|
||||
Do not rely on inferred shell cwd from a prior command in a different block.
|
||||
|
||||
`gitea_validate_review_final_report` rejects validation claims without
|
||||
cwd/HEAD proof when `validation_session` is supplied.
|
||||
|
||||
Baseline validation must document baseline worktree path, baseline target SHA,
|
||||
cwd proof, exact baseline command, and baseline result using the same rules.
|
||||
|
||||
Report the exact validation command as executed.
|
||||
|
||||
Report the working directory where validation ran.
|
||||
@@ -1189,6 +1211,7 @@ Controller Handoff:
|
||||
* Files reviewed:
|
||||
* Validation:
|
||||
* Validation failure history:
|
||||
* Validation cwd/HEAD proof:
|
||||
* Official validation integrity status:
|
||||
* Terminal review mutation:
|
||||
* Review decision:
|
||||
|
||||
@@ -1,217 +0,0 @@
|
||||
"""Unit tests for canonical_comment_validator (#496)."""
|
||||
|
||||
import unittest
|
||||
|
||||
import canonical_comment_validator as ccv
|
||||
|
||||
FULL_SHA = "a" * 40
|
||||
|
||||
_VALID_ISSUE = f"""## Canonical Issue State
|
||||
|
||||
STATE: ready-for-author
|
||||
WHO_IS_NEXT: author
|
||||
NEXT_ACTION: Implement canonical comment validator and wire MCP gates
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
You are the author LLM for issue #496. Add canonical_comment_validator.py
|
||||
and wire fail-closed gates into gitea_create_issue_comment.
|
||||
```
|
||||
WHAT_HAPPENED: Issue filed for MCP validation wall
|
||||
WHY: Workflow comments must carry durable next-action state
|
||||
RELATED_PRS: none
|
||||
BLOCKERS: none
|
||||
VALIDATION: unit tests not yet run
|
||||
LAST_UPDATED_BY: prgs-author
|
||||
"""
|
||||
|
||||
_VALID_PR = f"""## Canonical PR State
|
||||
|
||||
STATE: ready-for-review
|
||||
WHO_IS_NEXT: reviewer
|
||||
NEXT_ACTION: Run focused pytest and review the validator wiring diff
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Review PR for issue #496. Confirm casual comments still post and workflow
|
||||
comments without canonical fields are rejected before the Gitea API call.
|
||||
```
|
||||
WHAT_HAPPENED: Author opened PR with validator module
|
||||
WHY: Fail-closed enforcement belongs at the MCP boundary
|
||||
ISSUE: #496
|
||||
HEAD_SHA: {FULL_SHA}
|
||||
REVIEW_STATUS: pending
|
||||
MERGE_READY: false
|
||||
BLOCKERS: none
|
||||
VALIDATION: not yet run
|
||||
LAST_UPDATED_BY: prgs-author
|
||||
"""
|
||||
|
||||
_VALID_SUPERSEDED = """## Canonical PR State
|
||||
|
||||
STATE: superseded
|
||||
WHO_IS_NEXT: reconciler
|
||||
NEXT_ACTION: Close superseded issue after canonical issue #495 lands
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Close issue #494 as superseded by #495 once templates and docs are merged.
|
||||
```
|
||||
WHY: Duplicate tracking issue replaced by durable canonical template work
|
||||
CANONICAL_ITEM: issue #495
|
||||
SUPERSEDED_ITEM: issue #494
|
||||
CLOSE_OR_KEEP_OPEN: close superseded issue #494 after #495 merges
|
||||
"""
|
||||
|
||||
_VALID_BLOCKED = """## Canonical Issue State
|
||||
|
||||
STATE: blocked
|
||||
WHO_IS_NEXT: author
|
||||
NEXT_ACTION: Fix failing validator unit tests before requesting re-review
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Run pytest tests/test_canonical_comment_validator.py and repair failures.
|
||||
```
|
||||
WHAT_HAPPENED: Validator tests failed in CI
|
||||
WHY: Blocked until test suite is green
|
||||
RELATED_PRS: PR #500
|
||||
BLOCKERS: pytest failures; unblock after all validator tests pass
|
||||
VALIDATION: pytest failed
|
||||
LAST_UPDATED_BY: prgs-author
|
||||
"""
|
||||
|
||||
_VALID_READY_TO_MERGE = f"""## Canonical PR State
|
||||
|
||||
STATE: ready-to-merge
|
||||
WHO_IS_NEXT: merger
|
||||
NEXT_ACTION: Merge PR after confirming approval_at_current_head
|
||||
NEXT_PROMPT:
|
||||
```text
|
||||
Merge PR #500 for issue #496 after live mergeable check passes.
|
||||
```
|
||||
WHAT_HAPPENED: Reviewer approved at current head
|
||||
WHY: All gates passed and head SHA is current
|
||||
ISSUE: #496
|
||||
HEAD_SHA: {FULL_SHA}
|
||||
REVIEW_STATUS: approved / approval_at_current_head
|
||||
MERGE_READY: true
|
||||
BLOCKERS: none
|
||||
VALIDATION: pytest passed; reviewer approved at head {FULL_SHA}
|
||||
LAST_UPDATED_BY: prgs-reviewer
|
||||
"""
|
||||
|
||||
|
||||
class TestCanonicalCommentDetection(unittest.TestCase):
|
||||
def test_casual_comment_not_workflow(self):
|
||||
self.assertFalse(ccv.is_workflow_changing_comment("Thanks, I will check this."))
|
||||
|
||||
def test_workflow_trigger_detected(self):
|
||||
self.assertTrue(ccv.is_workflow_changing_comment("Blocked, author should fix."))
|
||||
|
||||
def test_machine_marker_exempt(self):
|
||||
body = "<!-- mcp-review-lease:v1 -->\nphase: claimed"
|
||||
self.assertFalse(ccv.is_workflow_changing_comment(body))
|
||||
|
||||
|
||||
class TestCanonicalCommentValidation(unittest.TestCase):
|
||||
def test_casual_comment_allowed(self):
|
||||
result = ccv.assess_canonical_comment("Thanks, I will check this.")
|
||||
self.assertTrue(result["allowed"])
|
||||
self.assertFalse(result["is_workflow_comment"])
|
||||
|
||||
def test_workflow_comment_missing_fields_rejected(self):
|
||||
result = ccv.assess_canonical_comment("Blocked, author should fix.")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertTrue(result["is_workflow_comment"])
|
||||
self.assertIn("WHO_IS_NEXT", result["missing_fields"])
|
||||
self.assertIn("correction_message", result)
|
||||
self.assertIn("Suggested template", result["correction_message"])
|
||||
|
||||
def test_missing_next_prompt_rejected(self):
|
||||
body = """STATE: blocked
|
||||
WHO_IS_NEXT: author
|
||||
NEXT_ACTION: Repair the failing validator unit tests in CI
|
||||
WHY: Tests must pass before merge
|
||||
BLOCKERS: pytest failed; unblock after tests pass
|
||||
VALIDATION: failed
|
||||
"""
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("NEXT_PROMPT", result["missing_fields"] + result["vague_fields"])
|
||||
|
||||
def test_vague_next_action_rejected(self):
|
||||
body = _VALID_ISSUE.replace(
|
||||
"Implement canonical comment validator and wire MCP gates",
|
||||
"continue",
|
||||
)
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("NEXT_ACTION", result["vague_fields"])
|
||||
|
||||
def test_invalid_who_is_next_rejected(self):
|
||||
body = _VALID_ISSUE.replace("WHO_IS_NEXT: author", "WHO_IS_NEXT: llm")
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("WHO_IS_NEXT", result["vague_fields"])
|
||||
|
||||
def test_valid_issue_comment_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_ISSUE)
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_issue_comment_mentioning_pr_requires_related_prs(self):
|
||||
body = _VALID_ISSUE.replace("RELATED_PRS: none", "RELATED_PRS:")
|
||||
body = body.replace("Issue filed", "Issue filed; see PR #500")
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("RELATED_PRS", result["missing_fields"])
|
||||
|
||||
def test_pr_review_requires_head_sha(self):
|
||||
body = _VALID_PR.replace(f"HEAD_SHA: {FULL_SHA}", "HEAD_SHA:")
|
||||
result = ccv.assess_canonical_comment(body, context="pr_review")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("HEAD_SHA", result["missing_fields"])
|
||||
|
||||
def test_ready_to_merge_requires_approval_and_sha_proof(self):
|
||||
body = _VALID_READY_TO_MERGE.replace("approval_at_current_head", "pending")
|
||||
body = body.replace(f"HEAD_SHA: {FULL_SHA}", "HEAD_SHA: pending")
|
||||
body = body.replace("reviewer approved at head", "reviewer pending at head")
|
||||
body = body.replace("approved /", "pending /")
|
||||
result = ccv.assess_canonical_comment(body, context="pr_comment")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertTrue(result["extra_reasons"])
|
||||
|
||||
def test_ready_to_merge_valid_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_READY_TO_MERGE, context="pr_comment")
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_superseded_requires_canonical_and_superseded_items(self):
|
||||
body = _VALID_SUPERSEDED.replace("CANONICAL_ITEM: issue #495", "CANONICAL_ITEM:")
|
||||
result = ccv.assess_canonical_comment(body, context="supersession")
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("CANONICAL_ITEM", result["missing_fields"])
|
||||
|
||||
def test_superseded_valid_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_SUPERSEDED, context="supersession")
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_blocked_requires_unblock_condition(self):
|
||||
body = _VALID_BLOCKED.replace(
|
||||
"BLOCKERS: pytest failures; unblock after all validator tests pass",
|
||||
"BLOCKERS: pytest failures",
|
||||
)
|
||||
result = ccv.assess_canonical_comment(body)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertIn("BLOCKERS", result["vague_fields"])
|
||||
|
||||
def test_blocked_valid_allowed(self):
|
||||
result = ccv.assess_canonical_comment(_VALID_BLOCKED)
|
||||
self.assertTrue(result["allowed"])
|
||||
|
||||
def test_correction_lists_missing_fields(self):
|
||||
result = ccv.assess_canonical_comment("ready for merge")
|
||||
self.assertFalse(result["allowed"])
|
||||
message = result["correction_message"]
|
||||
self.assertIn("Missing fields", message)
|
||||
self.assertIn("WHO_IS_NEXT", message)
|
||||
self.assertIn("Suggested template", message)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -498,31 +498,6 @@ class TestCanonicalReconcileSchema(unittest.TestCase):
|
||||
)
|
||||
|
||||
|
||||
class TestCanonicalCommentPostClaim(unittest.TestCase):
|
||||
def test_blocks_post_claim_when_validator_rejected_in_report(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- MCP/Gitea mutations: issue comment posted",
|
||||
"- Issue comments posted: 1 canonical state comment",
|
||||
"canonical comment validation failed (fail closed before posting).",
|
||||
'- canonical_comment_validation: {"allowed": false}',
|
||||
])
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("shared.canonical_comment_post_claim", rule_ids)
|
||||
self.assertTrue(result["blocked"])
|
||||
|
||||
def test_allows_none_when_validator_rejected_without_post_claim(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Issue comments posted: none",
|
||||
"canonical comment validation failed (fail closed before posting).",
|
||||
])
|
||||
result = assess_final_report_validator(report, "work_issue")
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertNotIn("shared.canonical_comment_post_claim", rule_ids)
|
||||
|
||||
|
||||
class TestEntryPoint(unittest.TestCase):
|
||||
def test_unknown_task_kind_blocks(self):
|
||||
result = assess_final_report_validator("report", "unknown_mode")
|
||||
|
||||
@@ -11,6 +11,7 @@ from issue_lock_adoption import ( # noqa: E402
|
||||
NO_MATCH,
|
||||
assess_own_branch_adoption,
|
||||
build_adoption_proof,
|
||||
build_non_adoption_lock_proof,
|
||||
)
|
||||
|
||||
REQ = "feat/issue-420-server-code-parity"
|
||||
@@ -134,5 +135,93 @@ class TestBuildAdoptionProof(unittest.TestCase):
|
||||
self.assertTrue(proof["no_competing_live_lock_proof"])
|
||||
|
||||
|
||||
class TestExplicitAdoptionProofFields(unittest.TestCase):
|
||||
"""#477: explicit, citable adoption-proof fields for all outcomes."""
|
||||
|
||||
def _proof(self, assessment, branch):
|
||||
return build_adoption_proof(
|
||||
issue_number=420,
|
||||
branch_name=branch,
|
||||
assessment=assessment,
|
||||
open_pr_checked=True,
|
||||
competing_lock_checked=True,
|
||||
lock_file_path="/tmp/example-lock.json",
|
||||
lock_file_status="written",
|
||||
)
|
||||
|
||||
def test_adopt_proof_exposes_explicit_fields(self):
|
||||
assessment = assess_own_branch_adoption(
|
||||
issue_number=420,
|
||||
requested_branch=REQ,
|
||||
existing_branches=[{"name": REQ, "commit_sha": "934688a"}],
|
||||
)
|
||||
proof = self._proof(assessment, REQ)
|
||||
self.assertEqual(proof["adoption_decision"], "ADOPT")
|
||||
self.assertTrue(proof["adopted"])
|
||||
self.assertEqual(proof["adopted_branch"], REQ)
|
||||
self.assertEqual(proof["adopted_branch_head"], "934688a")
|
||||
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
|
||||
self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
|
||||
self.assertIn("gitea_create_pr", proof["safe_next_action"])
|
||||
self.assertIn("exactly matches", proof["matcher_summary"])
|
||||
|
||||
def test_block_proof_reports_competing_and_does_not_claim_adoption(self):
|
||||
assessment = assess_own_branch_adoption(
|
||||
issue_number=420,
|
||||
requested_branch=REQ,
|
||||
existing_branches=[{"name": "feat/issue-420-rogue"}],
|
||||
)
|
||||
proof = self._proof(assessment, REQ)
|
||||
self.assertEqual(proof["adoption_decision"], "BLOCK_COMPETING")
|
||||
self.assertFalse(proof["adopted"])
|
||||
self.assertIsNone(proof["adopted_branch"])
|
||||
self.assertIsNone(proof["adopted_branch_head"])
|
||||
self.assertEqual(proof["competing_branch_check"]["result"], "blocked")
|
||||
self.assertIn(
|
||||
"feat/issue-420-rogue",
|
||||
proof["competing_branch_check"]["competing_branches"],
|
||||
)
|
||||
self.assertIn("fail closed", proof["safe_next_action"])
|
||||
|
||||
def test_no_match_proof_does_not_claim_adoption(self):
|
||||
assessment = assess_own_branch_adoption(
|
||||
issue_number=420,
|
||||
requested_branch=REQ,
|
||||
existing_branches=[{"name": "feat/issue-999-unrelated"}],
|
||||
)
|
||||
proof = self._proof(assessment, REQ)
|
||||
self.assertEqual(proof["adoption_decision"], "NO_MATCH")
|
||||
self.assertFalse(proof["adopted"])
|
||||
self.assertIsNone(proof["adopted_branch"])
|
||||
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
|
||||
|
||||
def test_substring_collision_stays_boundary_safe(self):
|
||||
# issue-42 must not adopt/claim against an issue-420 branch (#440/#477).
|
||||
own = "feat/issue-42-widget"
|
||||
assessment = assess_own_branch_adoption(
|
||||
issue_number=42,
|
||||
requested_branch=own,
|
||||
existing_branches=[
|
||||
{"name": own, "commit_sha": "abc1234"},
|
||||
{"name": "feat/issue-420-server-code-parity"},
|
||||
],
|
||||
)
|
||||
proof = self._proof(assessment, own)
|
||||
self.assertEqual(proof["adoption_decision"], "ADOPT")
|
||||
self.assertEqual(proof["adopted_branch"], own)
|
||||
self.assertEqual(proof["competing_branch_check"]["competing_branches"], [])
|
||||
|
||||
def test_non_adoption_lock_proof_is_adoption_free(self):
|
||||
proof = build_non_adoption_lock_proof(
|
||||
issue_number=196, branch_name="feat/issue-196-mutations"
|
||||
)
|
||||
self.assertEqual(proof["adoption_decision"], "NO_MATCH")
|
||||
self.assertFalse(proof["adopted"])
|
||||
self.assertIsNone(proof["adopted_branch"])
|
||||
self.assertIsNone(proof["adopted_branch_head"])
|
||||
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
|
||||
self.assertIn("no adoption", proof["safe_next_action"].lower())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -213,6 +213,12 @@ def test_validation_failure_history_verifier_exported():
|
||||
assert callable(assess_validation_failure_history_report)
|
||||
|
||||
|
||||
def test_validation_cwd_proof_verifier_exported():
|
||||
from review_proofs import assess_validation_cwd_proof_report
|
||||
|
||||
assert callable(assess_validation_cwd_proof_report)
|
||||
|
||||
|
||||
def test_prior_blocker_skip_verifier_exported():
|
||||
from review_proofs import assess_prior_blocker_skip_proof
|
||||
|
||||
|
||||
+39
-30
@@ -3087,36 +3087,6 @@ class TestIssueCommentTools(unittest.TestCase):
|
||||
self.assertFalse(result["success"])
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_casual_comment_allowed(self, _auth, mock_api):
|
||||
mock_api.return_value = self._comment(556, "gitea-author", "casual")
|
||||
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
|
||||
result = gitea_create_issue_comment(
|
||||
issue_number=9,
|
||||
body="Thanks, I will check this.",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertTrue(result["success"])
|
||||
self.assertTrue(result["performed"])
|
||||
mock_api.assert_called_once()
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_workflow_comment_missing_fields_blocked(self, _auth, mock_api):
|
||||
with patch.dict(os.environ, self.AUTHOR_ENV, clear=True):
|
||||
result = gitea_create_issue_comment(
|
||||
issue_number=9,
|
||||
body="Blocked, author should fix.",
|
||||
remote="prgs",
|
||||
)
|
||||
self.assertFalse(result["success"])
|
||||
self.assertFalse(result["performed"])
|
||||
self.assertIn("canonical_comment_validation", result)
|
||||
self.assertFalse(result["canonical_comment_validation"]["allowed"])
|
||||
self.assertTrue(result["canonical_comment_validation"]["is_workflow_comment"])
|
||||
mock_api.assert_not_called()
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_create_missing_issue_error_is_redacted(self, _auth, mock_api):
|
||||
@@ -3490,6 +3460,45 @@ class TestIssueLocking(unittest.TestCase):
|
||||
self.assertIn("adoption", res)
|
||||
self.assertEqual(res["adoption"]["branch_head_commit"], "abc123")
|
||||
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
)
|
||||
@patch("mcp_server.api_get_all")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_lock_issue_adoption_response_has_explicit_proof(self, _auth, mock_api, _git_state):
|
||||
# #477 AC1: the live lock response must carry citable adoption proof.
|
||||
branch = "feat/issue-196-mutations"
|
||||
self.mock_dup_fetcher.return_value = ([], [branch], {"status": "not_claimed"})
|
||||
mock_api.return_value = [{"name": branch, "commit": {"id": "abc123"}}]
|
||||
res = gitea_lock_issue(issue_number=196, branch_name=branch, remote="prgs")
|
||||
proof = res["adoption"]
|
||||
self.assertEqual(proof["adoption_decision"], "ADOPT")
|
||||
self.assertTrue(proof["adopted"])
|
||||
self.assertEqual(proof["adopted_branch"], branch)
|
||||
self.assertEqual(proof["adopted_branch_head"], "abc123")
|
||||
self.assertEqual(proof["competing_branch_check"]["result"], "clear")
|
||||
self.assertIn("gitea_create_pr", proof["safe_next_action"])
|
||||
self.assertIn("196", proof["matcher_summary"])
|
||||
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
)
|
||||
@patch("mcp_server.api_get_all", return_value=[])
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_lock_issue_no_match_response_does_not_claim_adoption(self, _auth, _api, _git_state):
|
||||
# #477 AC2/AC3: a normal (NO_MATCH) lock must carry adoption-free proof
|
||||
# and must NOT expose an ``adoption`` block.
|
||||
res = gitea_lock_issue(issue_number=196, branch_name="feat/issue-196-mutations", remote="prgs")
|
||||
self.assertTrue(res["success"])
|
||||
self.assertNotIn("adoption", res)
|
||||
check = res["adoption_check"]
|
||||
self.assertEqual(check["adoption_decision"], "NO_MATCH")
|
||||
self.assertFalse(check["adopted"])
|
||||
self.assertIsNone(check["adopted_branch"])
|
||||
self.assertEqual(check["competing_branch_check"]["result"], "clear")
|
||||
|
||||
@patch(
|
||||
"mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value=_clean_master_git_state_for_lock(),
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Regression tests for non-list API payloads on PR/issue comment listing (#485)."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
from mcp_server import ( # noqa: E402
|
||||
_list_pr_lease_comments,
|
||||
gitea_list_issue_comments,
|
||||
)
|
||||
|
||||
FAKE_AUTH = "Basic dGVzdDp0ZXN0"
|
||||
AUTHOR_ENV = {
|
||||
"GITEA_PROFILE_NAME": "gitea-author",
|
||||
"GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.issue.comment",
|
||||
}
|
||||
|
||||
|
||||
class TestPrLeaseCommentsNonListGuard(unittest.TestCase):
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_list_pr_lease_comments_non_list_payload_returns_empty(self, _auth, mock_api):
|
||||
mock_api.return_value = {"message": "Unauthorized"}
|
||||
result = _list_pr_lease_comments(
|
||||
12, remote="prgs", host=None, org=None, repo=None)
|
||||
self.assertEqual(result, [])
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_list_pr_lease_comments_none_returns_empty(self, _auth, mock_api):
|
||||
mock_api.return_value = None
|
||||
result = _list_pr_lease_comments(
|
||||
12, remote="prgs", host=None, org=None, repo=None)
|
||||
self.assertEqual(result, [])
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_list_pr_lease_comments_list_payload_unchanged(self, _auth, mock_api):
|
||||
comment = {"id": 7, "body": "<!-- mcp-review-lease:v1 -->"}
|
||||
mock_api.return_value = [comment]
|
||||
result = _list_pr_lease_comments(
|
||||
12, remote="prgs", host=None, org=None, repo=None, limit=5)
|
||||
self.assertEqual(result, [comment])
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_list_issue_comments_non_list_payload_returns_empty(self, _auth, mock_api):
|
||||
mock_api.return_value = {"message": "Unauthorized"}
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
result = gitea_list_issue_comments(issue_number=9, remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["comments"], [])
|
||||
|
||||
@patch("mcp_server.api_request")
|
||||
@patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
|
||||
def test_list_issue_comments_list_payload_unchanged(self, _auth, mock_api):
|
||||
mock_api.return_value = [
|
||||
{
|
||||
"id": 101,
|
||||
"user": {"login": "alice"},
|
||||
"body": "hello",
|
||||
"created_at": "2026-07-03T00:00:00Z",
|
||||
"updated_at": "2026-07-03T01:00:00Z",
|
||||
}
|
||||
]
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
result = gitea_list_issue_comments(issue_number=9, remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(len(result["comments"]), 1)
|
||||
self.assertEqual(result["comments"][0]["author"], "alice")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,159 @@
|
||||
"""#469/#470: capability preflight lifetime across safe read-only calls."""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
|
||||
import gitea_mcp_server as mcp_server
|
||||
|
||||
|
||||
class TestPreflightReadSurvival(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.orig_whoami = mcp_server._preflight_whoami_called
|
||||
self.orig_capability = mcp_server._preflight_capability_called
|
||||
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
|
||||
self.orig_capability_violation = mcp_server._preflight_capability_violation
|
||||
self.orig_resolved_role = mcp_server._preflight_resolved_role
|
||||
self.orig_resolved_task = mcp_server._preflight_resolved_task
|
||||
self.orig_process_start = mcp_server._process_start_porcelain
|
||||
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
|
||||
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
|
||||
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
mcp_server._preflight_capability_violation = False
|
||||
mcp_server._preflight_resolved_role = None
|
||||
mcp_server._preflight_resolved_task = None
|
||||
mcp_server._process_start_porcelain = ""
|
||||
mcp_server._preflight_whoami_baseline_porcelain = None
|
||||
mcp_server._preflight_capability_baseline_porcelain = None
|
||||
|
||||
def tearDown(self):
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami
|
||||
mcp_server._preflight_capability_called = self.orig_capability
|
||||
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
|
||||
mcp_server._preflight_capability_violation = self.orig_capability_violation
|
||||
mcp_server._preflight_resolved_role = self.orig_resolved_role
|
||||
mcp_server._preflight_resolved_task = self.orig_resolved_task
|
||||
mcp_server._process_start_porcelain = self.orig_process_start
|
||||
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
|
||||
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
|
||||
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
|
||||
def test_interleaved_whoami_preserves_capability(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
||||
)
|
||||
self.assertTrue(mcp_server._preflight_capability_called)
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
||||
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_capability_called)
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
||||
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertFalse(mcp_server._preflight_capability_called)
|
||||
|
||||
def test_missing_capability_still_fails_closed(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertIn("has not been resolved", str(ctx.exception))
|
||||
|
||||
def test_task_mismatch_fails_closed(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="author", resolved_task="create_issue"
|
||||
)
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertIn("task mismatch", str(ctx.exception))
|
||||
|
||||
def test_capability_consumed_after_mutation_gate(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="author", resolved_task="create_issue"
|
||||
)
|
||||
mcp_server.verify_preflight_purity(task="create_issue")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="create_issue")
|
||||
self.assertIn("has not been resolved", str(ctx.exception))
|
||||
|
||||
def test_whoami_recovery_after_violation_clears_capability(self):
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_whoami_violation)
|
||||
|
||||
del os.environ["GITEA_TEST_FORCE_DIRTY"]
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertFalse(mcp_server._preflight_whoami_violation)
|
||||
self.assertFalse(mcp_server._preflight_capability_called)
|
||||
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reviewer", resolved_task="review_pr"
|
||||
)
|
||||
mcp_server.verify_preflight_purity(task="review_pr")
|
||||
|
||||
def test_close_pr_sequence_with_interleaved_reads(self):
|
||||
"""resolve(close_pr) → whoami/view reads → close_pr gate (#470)."""
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
||||
)
|
||||
# Simulate live-state revalidation between resolve and mutation.
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_capability_called)
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertFalse(mcp_server._preflight_capability_called)
|
||||
|
||||
def test_fresh_capability_resolve_replaces_prior_task(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="author", resolved_task="create_issue"
|
||||
)
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
||||
)
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
|
||||
def test_missing_capability_error_names_re_resolve(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
msg = str(ctx.exception)
|
||||
self.assertIn("gitea_resolve_task_capability", msg)
|
||||
self.assertIn('task="close_pr"', msg)
|
||||
|
||||
def test_task_mismatch_error_names_re_resolve(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="author", resolved_task="create_issue"
|
||||
)
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
msg = str(ctx.exception)
|
||||
self.assertIn("task mismatch", msg)
|
||||
self.assertIn('task="close_pr"', msg)
|
||||
|
||||
def test_consumed_capability_error_names_re_resolve(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
||||
)
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertIn('task="close_pr"', str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,89 @@
|
||||
"""Reconciler close_pr must not require author branches/ worktree (#468)."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
import gitea_mcp_server as srv
|
||||
|
||||
FAKE_AUTH = "token test"
|
||||
CONTROL_CHECKOUT_ROOT = str(Path(__file__).resolve().parents[3])
|
||||
|
||||
RECONCILER_PROFILE = {
|
||||
"profile_name": "prgs-reconciler",
|
||||
"allowed_operations": ["gitea.read", "gitea.pr.close", "gitea.pr.comment"],
|
||||
"forbidden_operations": [
|
||||
"gitea.pr.approve",
|
||||
"gitea.pr.merge",
|
||||
"gitea.pr.review",
|
||||
"gitea.pr.create",
|
||||
"gitea.branch.push",
|
||||
"gitea.repo.commit",
|
||||
],
|
||||
"audit_label": "prgs-reconciler",
|
||||
}
|
||||
|
||||
|
||||
class TestReconcilerCloseWorkspaceGuard(unittest.TestCase):
|
||||
def setUp(self):
|
||||
srv._preflight_whoami_called = True
|
||||
srv._preflight_capability_called = True
|
||||
srv._preflight_whoami_violation = False
|
||||
srv._preflight_capability_violation = False
|
||||
self._orig_in_test = srv._preflight_in_test_mode
|
||||
srv._preflight_in_test_mode = lambda: False
|
||||
|
||||
def tearDown(self):
|
||||
srv._preflight_in_test_mode = self._orig_in_test
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
||||
@patch("gitea_mcp_server.get_profile", return_value=RECONCILER_PROFILE)
|
||||
@patch("gitea_mcp_server.api_request")
|
||||
def test_reconciler_close_pr_from_control_checkout_succeeds(
|
||||
self, mock_api, _profile, _ns, _auth
|
||||
):
|
||||
srv._preflight_resolved_role = "reconciler"
|
||||
mock_api.return_value = {
|
||||
"number": 414,
|
||||
"title": "old",
|
||||
"body": "",
|
||||
"state": "closed",
|
||||
"html_url": "https://gitea.example.com/pulls/414",
|
||||
}
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
with patch.dict(os.environ, {}, clear=False):
|
||||
os.environ.pop("GITEA_AUTHOR_WORKTREE", None)
|
||||
os.environ.pop("GITEA_ACTIVE_WORKTREE", None)
|
||||
result = srv.gitea_edit_pr(414, state="closed", remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
self.assertEqual(result["state"], "closed")
|
||||
mock_api.assert_called_once()
|
||||
|
||||
@patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
|
||||
@patch("gitea_mcp_server._profile_permission_block", return_value=None)
|
||||
@patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
|
||||
@patch(
|
||||
"gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
|
||||
return_value=(True, []),
|
||||
)
|
||||
@patch("gitea_mcp_server.api_get_all", return_value=[])
|
||||
@patch(
|
||||
"gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
|
||||
return_value={"current_branch": "master"},
|
||||
)
|
||||
def test_author_create_issue_still_blocked_on_control_checkout(
|
||||
self, _git, _get_all, _role, _ns, _prof, _auth
|
||||
):
|
||||
srv._preflight_resolved_role = "author"
|
||||
with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
srv.gitea_create_issue(title="Test", body="body")
|
||||
self.assertIn("stable control checkout", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Tests for validation cwd/HEAD proof verifier (#398)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
from reviewer_validation_cwd_proof import assess_validation_cwd_proof_report # noqa: E402
|
||||
|
||||
ROOT = "/Users/jasonwalker/Development/Gitea-Tools"
|
||||
WORKTREE = f"{ROOT}/branches/review-feat-issue-398"
|
||||
HEAD = "f5953549aad5e822f14f52d3ea3c6d7990109384"
|
||||
|
||||
|
||||
def _proof_backed_report() -> str:
|
||||
return "\n".join([
|
||||
f"Candidate head SHA: {HEAD}",
|
||||
f"pwd: {WORKTREE}",
|
||||
f"git rev-parse HEAD: {HEAD}",
|
||||
"git status --short --branch: ## feat/issue-398...prgs/master",
|
||||
f"Validation command: cd {WORKTREE} && venv/bin/python -m pytest tests/ -q",
|
||||
"Result: 1497 passed, 6 skipped",
|
||||
])
|
||||
|
||||
|
||||
class TestValidationCwdProof(unittest.TestCase):
|
||||
def test_no_validation_claim_passes(self):
|
||||
result = assess_validation_cwd_proof_report("Review decision: approve")
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_missing_cwd_proof_fails(self):
|
||||
result = assess_validation_cwd_proof_report(
|
||||
f"Validation command: pytest tests/\nCandidate head SHA: {HEAD}",
|
||||
validation_session={"validation_ran": True, "expected_head_sha": HEAD},
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_main_checkout_cwd_blocks(self):
|
||||
result = assess_validation_cwd_proof_report(
|
||||
"\n".join([
|
||||
f"Candidate head SHA: {HEAD}",
|
||||
f"pwd: {ROOT}",
|
||||
f"git rev-parse HEAD: {HEAD}",
|
||||
"git status --short --branch: ## master",
|
||||
"Validation command: pytest tests/ -q",
|
||||
]),
|
||||
validation_session={"validation_ran": True, "expected_head_sha": HEAD},
|
||||
project_root=ROOT,
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["violations"])
|
||||
|
||||
def test_wrong_head_blocks(self):
|
||||
wrong = "a" * 40
|
||||
result = assess_validation_cwd_proof_report(
|
||||
"\n".join([
|
||||
f"Candidate head SHA: {HEAD}",
|
||||
f"pwd: {WORKTREE}",
|
||||
f"git rev-parse HEAD: {wrong}",
|
||||
"git status --short --branch: clean",
|
||||
f"Validation command: cd {WORKTREE} && pytest -q",
|
||||
]),
|
||||
validation_session={"validation_ran": True, "expected_head_sha": HEAD},
|
||||
project_root=ROOT,
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["violations"])
|
||||
|
||||
def test_fully_proof_backed_passes(self):
|
||||
result = assess_validation_cwd_proof_report(
|
||||
_proof_backed_report(),
|
||||
validation_session={"validation_ran": True, "expected_head_sha": HEAD},
|
||||
project_root=ROOT,
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_baseline_without_cwd_fails(self):
|
||||
result = assess_validation_cwd_proof_report(
|
||||
"\n".join([
|
||||
_proof_backed_report(),
|
||||
"Baseline validation command: pytest tests/ -q",
|
||||
]),
|
||||
validation_session={
|
||||
"validation_ran": True,
|
||||
"expected_head_sha": HEAD,
|
||||
"baseline_validation_ran": True,
|
||||
},
|
||||
project_root=ROOT,
|
||||
)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("baseline" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_baseline_with_full_proof_passes(self):
|
||||
result = assess_validation_cwd_proof_report(
|
||||
"\n".join([
|
||||
_proof_backed_report(),
|
||||
f"Baseline worktree: {ROOT}/branches/baseline-master-pr376",
|
||||
f"Baseline target SHA: {HEAD}",
|
||||
f"Baseline validation command: cd {ROOT}/branches/baseline-master-pr376 && pytest -q",
|
||||
]),
|
||||
validation_session={
|
||||
"validation_ran": True,
|
||||
"expected_head_sha": HEAD,
|
||||
"baseline_validation_ran": True,
|
||||
},
|
||||
project_root=ROOT,
|
||||
)
|
||||
self.assertTrue(result["proven"], result["reasons"])
|
||||
|
||||
def test_final_report_validator_integration(self):
|
||||
result = assess_final_report_validator(
|
||||
"Validation command: pytest tests/ -q",
|
||||
"review_pr",
|
||||
validation_session={"validation_ran": True},
|
||||
)
|
||||
self.assertTrue(result["blocked"] or result["downgraded"])
|
||||
self.assertTrue(
|
||||
any(
|
||||
f.get("rule_id") == "reviewer.validation_cwd_proof"
|
||||
for f in result.get("findings") or []
|
||||
)
|
||||
)
|
||||
|
||||
def test_exported_from_review_proofs(self):
|
||||
from review_proofs import assess_validation_cwd_proof_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user