diff --git a/author_proofs.py b/author_proofs.py new file mode 100644 index 0000000..e40d8e9 --- /dev/null +++ b/author_proofs.py @@ -0,0 +1,217 @@ +"""Fail-closed branch-identity proofs for author workflows (#177). + +Author-side counterpart of the reviewer proofs in ``review_proofs.py`` +(#173). During the #173 implementation itself, a commit landed on local +``master`` because the shared checkout's branch moved mid-session (origin +incident of #177). These helpers turn that from an after-the-fact repair +into a fail-closed gate: an author workflow must prove its local git state +before staging, committing, or pushing. + +The helpers are pure (no git calls): the workflow gathers the raw facts +(``git branch --show-current``, ``git rev-parse HEAD``, the push refspec, +the branch named in the issue claim) and passes them in, so the same logic +works from prompts, harness assertions, and tests. Shared-worktree branch +switches by other sessions are treated as expected events to detect, not +exceptional ones. Nothing here weakens the review/merge/permission gates. +""" + +PROTECTED_BRANCHES = frozenset( + {"master", "main", "develop", "development", "dev"} +) + + +def _clean(name): + return (name or "").strip() + + +def verify_branch_for_commit(current_branch, intended_branch): + """Required behavior 1: prove the branch before staging/committing. + + Proven only when both names are present, the intended branch is not a + protected branch, and the current branch equals the intended one (which + also rules out being on any protected branch). Returns {'proven', + 'block', 'reasons', 'current_branch', 'intended_branch'}. + """ + reasons = [] + current = _clean(current_branch) + intended = _clean(intended_branch) + + if not current: + reasons.append( + "current branch unknown (detached HEAD or state not read); " + "fail closed" + ) + if not intended: + reasons.append("intended feature branch not stated; fail closed") + if intended and intended in PROTECTED_BRANCHES: + reasons.append( + f"intended branch '{intended}' is a protected branch; author " + "work must target a feature branch" + ) + if current and current in PROTECTED_BRANCHES: + reasons.append( + f"current branch '{current}' is a protected branch; committing " + "here is blocked" + ) + if current and intended and current != intended: + reasons.append( + f"current branch '{current}' is not the intended feature branch " + f"'{intended}'; stop before staging/committing" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "current_branch": current or None, + "intended_branch": intended or None, + } + + +def detect_branch_drift(branch_at_validation, head_at_validation, + current_branch, current_head): + """Required behaviors 2–3: stop when branch or HEAD moved mid-session. + + Compares the branch name and HEAD SHA captured at validation time with + the state observed immediately before commit/push. Any difference — + including an external branch switch in a shared worktree — is drift and + blocks until reconciled. Missing state fails closed. + """ + reasons = [] + branch_then = _clean(branch_at_validation) + branch_now = _clean(current_branch) + head_then = _clean(head_at_validation).lower() + head_now = _clean(current_head).lower() + + if not branch_then or not head_then: + reasons.append("validation-time branch/HEAD not recorded; fail closed") + if not branch_now or not head_now: + reasons.append("current branch/HEAD not read; fail closed") + + if branch_then and branch_now and branch_then != branch_now: + reasons.append( + f"branch changed from '{branch_then}' to '{branch_now}' since " + "validation — possible external branch switch in a shared " + "worktree; stop and reconcile before committing" + ) + if head_then and head_now and head_then != head_now: + reasons.append( + "HEAD moved since validation; re-validate on the current HEAD " + "before committing" + ) + + drifted = bool(reasons) + return {"drifted": drifted, "block": drifted, "reasons": reasons} + + +def verify_push_target(current_branch, remote_target_branch, intended_branch): + """Acceptance: a push needs local, remote, and intended branches to match. + + Proven only when all three names are present, equal, and not a + protected branch — a feature-branch workflow never pushes a protected + branch, and never pushes to a refspec other than its own branch. + """ + reasons = [] + current = _clean(current_branch) + remote_target = _clean(remote_target_branch) + intended = _clean(intended_branch) + + if not current: + reasons.append("current branch unknown; fail closed") + if not remote_target: + reasons.append("remote target branch not stated; fail closed") + if not intended: + reasons.append("intended feature branch not stated; fail closed") + + for label, name in (("current", current), ("remote target", remote_target), + ("intended", intended)): + if name and name in PROTECTED_BRANCHES: + reasons.append( + f"{label} branch '{name}' is a protected branch; author " + "pushes to protected branches are blocked" + ) + + if current and remote_target and current != remote_target: + reasons.append( + f"push target '{remote_target}' does not match the local branch " + f"'{current}'" + ) + if current and intended and current != intended: + reasons.append( + f"local branch '{current}' does not match the intended feature " + f"branch '{intended}'" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + } + + +def assess_protected_branch_commit(commit_branch, pushed=False, + repair_reported=True): + """Required behavior 4: handle an accidental protected-branch commit. + + If a commit landed on a protected branch: it must never be pushed, a + repair is required, and the repair must be *reported* — silently + continuing after (or without) repair is a violation, as is having + pushed the accident. + """ + branch = _clean(commit_branch) + accident = branch in PROTECTED_BRANCHES + + violations = [] + if accident: + if pushed: + violations.append( + f"accidental commit on protected branch '{branch}' was " + "pushed; protected-branch pushes are forbidden" + ) + if not repair_reported: + violations.append( + "protected-branch commit repair was not reported; the " + "workflow must surface the accident and the repair steps, " + "never silently continue" + ) + + return { + "accident": accident, + "must_not_push": accident, + "repair_required": accident, + "violations": violations, + } + + +def build_commit_push_report(commit_proof, drift, push_proof, accident=None): + """Acceptance: final report carries branch proof before commit and push. + + Combines the individual proofs; any failed proof, detected drift, or + accident violation makes the status 'blocked' — the workflow stops and + reports instead of continuing. + """ + accident = accident or {"accident": False, "violations": []} + violations = list(accident.get("violations", [])) + + blocked = ( + not commit_proof.get("proven") + or drift.get("drifted") + or not push_proof.get("proven") + or bool(violations) + ) + + return { + "status": "blocked" if blocked else "ok", + "branch_proof_before_commit": bool(commit_proof.get("proven")), + "branch_proof_before_push": bool(push_proof.get("proven")), + "drift_detected": bool(drift.get("drifted")), + "protected_branch_accident": bool(accident.get("accident")), + "violations": violations, + "reasons": ( + list(commit_proof.get("reasons", [])) + + list(drift.get("reasons", [])) + + list(push_proof.get("reasons", [])) + ), + } diff --git a/mcp_server.py b/mcp_server.py index f74a1f2..b0179e4 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -2492,26 +2492,26 @@ _PROJECT_SKILLS = { "committed.", ], }, - "jenkins-readonly": { + "jenkins-mcp": { "description": "Read-only Jenkins CI inspection (jobs, builds, " - "logs). Actual server name: jenkins-mcp (see mcp-control-plane).", + "logs).", "when_to_use": "Checking CI state once Jenkins MCP tools exist.", "required_operations": ["jenkins.read"], "status": "designed-not-implemented", - "notes": "Server code exists in mcp-control-plane as jenkins-mcp (read tools + gated trigger); registration pending (#55); docs in Gitea-Tools use historical name. Report SKIPPED if not connected. Do not substitute shell/API. Trigger requires dedicated profile (see #56).", + "notes": "Server code exists in mcp-control-plane as jenkins-mcp (read tools + gated trigger); registration pending. To register for discoverability in clients (Codex/Gemini/Grok/etc.): add to client MCP config under the jenkins-mcp name, reconnect/reload the client session after registration. Report SKIPPED if not connected. Do not substitute shell/API. Trigger requires dedicated profile (see #56).", "steps": [ "Confirm a Jenkins MCP server is connected (jenkins-mcp); if not, report " "SKIPPED.", "Use read-only operations only; never trigger unless using dedicated profile + confirmation.", ], }, - "glitchtip-readonly": { - "description": "Read-only GlitchTip error/event inspection. Actual server name: glitchtip-mcp (see mcp-control-plane).", + "glitchtip-mcp": { + "description": "Read-only GlitchTip error/event inspection.", "when_to_use": "Investigating reported errors once GlitchTip MCP " "tools exist.", "required_operations": ["glitchtip.read"], "status": "designed-not-implemented", - "notes": "Server code exists in mcp-control-plane as glitchtip-mcp (read-only tools); registration pending (#55); filing orchestrator is partial in mcp-control-plane (see #57). Report SKIPPED if not connected. Filing to Gitea is separate orchestrator, not in this server.", + "notes": "Server code exists in mcp-control-plane as glitchtip-mcp (read-only tools); registration pending. To register for discoverability in clients (Codex/Gemini/Grok/etc.): add to client MCP config under the glitchtip-mcp name, reconnect/reload the client session after registration. Filing orchestrator is partial in mcp-control-plane (see #57). Report SKIPPED if not connected. Filing to Gitea is separate orchestrator, not in this server.", "steps": [ "Confirm a GlitchTip MCP server is connected (glitchtip-mcp); if not, report " "SKIPPED.", diff --git a/review_proofs.py b/review_proofs.py index bab0f00..718061d 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -330,9 +330,97 @@ def assess_self_review_contamination(reviewer_identity, pr_author, } +def assess_role_boundary(proof): + """Assess reviewer/author role separation for blind queue workflows. + + Issue #175 blocks a reviewer queue task from silently becoming author + implementation work. The workflow may use both namespaces only when that + mixed use is explicit, justified, and non-mutating; author mutations after + a reviewer queue task require an explicit operator authorization. + + *proof* keys: + ``task_role`` ('reviewer' or 'author'), ``task_kind`` (for example + 'blind_pr_queue_review'), ``reviewer_namespace_used``, + ``author_namespace_used``, ``author_mutations`` (list), ``review_mutations`` + (list), ``operator_authorized_author_work``, ``mixed_namespace_justification``, + ``scratch_evidence_claimed``, and ``scratch_evidence_durable``. + """ + proof = proof or {} + task_role = (proof.get("task_role") or "").strip().lower() + task_kind = (proof.get("task_kind") or "").strip().lower() + author_mutations = list(proof.get("author_mutations") or []) + review_mutations = list(proof.get("review_mutations") or []) + reviewer_used = bool(proof.get("reviewer_namespace_used")) + author_used = bool(proof.get("author_namespace_used")) + authorized = bool(proof.get("operator_authorized_author_work")) + mixed_justification = ( + proof.get("mixed_namespace_justification") or "" + ).strip() + scratch_claimed = bool(proof.get("scratch_evidence_claimed")) + scratch_durable = bool(proof.get("scratch_evidence_durable")) + + reasons = [] + violations = [] + + if task_role not in {"reviewer", "author"}: + reasons.append("task role missing or unknown; role boundary unproven") + + if task_role == "reviewer": + if author_mutations and not authorized: + violations.append( + "reviewer task performed author mutations without explicit " + "operator authorization" + ) + if author_used and not mixed_justification: + reasons.append( + "reviewer task used author namespace without an explicit " + "justification" + ) + if task_kind == "blind_pr_queue_review" and author_mutations: + if not authorized: + violations.append( + "blind PR queue review silently pivoted into author " + "implementation" + ) + elif task_role == "author": + if review_mutations: + violations.append( + "author task performed reviewer-only mutations" + ) + + if reviewer_used and author_used and not mixed_justification: + reasons.append( + "mixed reviewer+author namespace use was not reported as a " + "role-boundary event" + ) + + if scratch_claimed and not scratch_durable: + reasons.append( + "scratch-only notes were claimed as durable evidence" + ) + + if violations: + status = "violation" + safe_next_action = "stop; report role-boundary violation" + elif reasons: + status = "warning" + safe_next_action = "downgrade final report; do not claim A-level proof" + else: + status = "clean" + safe_next_action = "proceed" + + return { + "status": status, + "clean": status == "clean", + "reasons": reasons, + "violations": violations, + "safe_next_action": safe_next_action, + } + + def build_final_report(checkout_proof, inventory, validation, contamination, identity_eligible, merge_performed, - issue_status_verified): + issue_status_verified, role_boundary=None): """Required behavior 6 + acceptance criteria: one report, distinct proofs. Combines the individual proof verdicts into the final-report fields the @@ -348,6 +436,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, checkout_proven = bool(checkout_proof.get("proven")) validation_claimable = bool(validation.get("claimable")) validation_strong = validation.get("verdict") == "strong" + role_boundary = role_boundary or { + "status": "warning", + "reasons": ["role-boundary proof missing"], + "violations": [], + } + role_status = role_boundary.get("status", "warning") downgrade_reasons = [] if not identity_eligible: @@ -367,6 +461,9 @@ def build_final_report(checkout_proof, inventory, validation, contamination, downgrade_reasons.append( f"session contamination status is '{contamination_status}'" ) + if role_status != "clean": + downgrade_reasons.append(f"role boundary status is '{role_status}'") + downgrade_reasons.extend(role_boundary.get("reasons", [])) if not issue_status_verified: downgrade_reasons.append("linked issue status not verified") @@ -374,6 +471,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, identity_eligible and checkout_proven and contamination_status == "clean" + and role_status == "clean" and validation_claimable and validation.get("verdict") != "invalid" ) @@ -384,6 +482,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "merge was performed/claimed although the proofs did not allow " "one; this run is blocked, not graded" ) + violations.extend(role_boundary.get("violations", [])) if violations: grade = "blocked" @@ -400,6 +499,7 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "pr_author_distinct_from_reviewer": contamination_status in ("clean",), "session_contamination": contamination_status, + "role_boundary": role_status, "inventory_complete": bool(inventory.get("complete")), "validated_on_pinned_head": checkout_proven and validation_claimable, "validation_passed": @@ -562,3 +662,97 @@ def assess_controller_handoff(report_text, role=None): "missing_fields": [], "reasons": [], } + + +# ── PR Inventory Trust Gate (Issue #194) ────────────────────────────────────── +# +# A reviewer agent may not convert an empty PR list response into a definitive +# "no open PRs" conclusion unless the inventory result is independently proven +# trustworthy. + +def pr_inventory_trust_gate( + list_prs_response: list | None, + remote: str | None = None, + org: str | None = None, + repo: str | None = None, + state: str | None = None, + authenticated_profile: dict | None = None, + local_remote_url: str | None = None, + user_context: str | None = None, + corroboration_open_pr_counter: int | None = None, + has_finality_metadata: bool = False, +) -> dict: + """Evaluate whether an empty PR list is trusted or untrusted. + + Returns a dict with 'status', 'reasons', and 'corroborated'. + """ + if list_prs_response is None or not isinstance(list_prs_response, list): + return { + "status": "inventory_error", + "reasons": ["PR list response is invalid (not a list or None)"], + "corroborated": False, + } + + if len(list_prs_response) > 0: + return { + "status": "trusted_nonempty", + "reasons": [], + "corroborated": False, + } + + reasons = [] + + # 1. Exact remote, owner, repo, and state filter resolved correctly + if not remote or remote not in ("dadeschools", "prgs"): + reasons.append("remote instance is invalid or unresolved") + if not org or not org.strip(): + reasons.append("owner/org is invalid or unresolved") + if not repo or not repo.strip(): + reasons.append("repository name is invalid or unresolved") + if state != "open": + reasons.append("state filter is not 'open'") + + # 2. Authenticated profile permission check + if not authenticated_profile or not isinstance(authenticated_profile, dict): + reasons.append("authenticated profile is missing or invalid") + else: + allowed = authenticated_profile.get("allowed_operations") or [] + if "gitea.read" not in allowed and "read" not in allowed: + reasons.append("authenticated profile lacks read permissions") + + # 3. Pagination/finality metadata or independent read path corroboration + corroborated = False + if has_finality_metadata: + corroborated = True + elif corroboration_open_pr_counter == 0: + corroborated = True + else: + reasons.append("pagination finality not proven and open_pr_counter corroboration is missing or non-zero") + + # 4. Local checkout remote URL matching the target repo + if not local_remote_url or not isinstance(local_remote_url, str): + reasons.append("local checkout remote URL is missing or invalid") + else: + expected = f"{org}/{repo}".lower() + if expected not in local_remote_url.lower(): + reasons.append(f"local remote URL does not match target repository '{org}/{repo}'") + + # 5. User context check (indicators that PRs should exist) + if user_context and isinstance(user_context, str): + indicators = ["pr #", "pull request #", "open pr", "pr queue"] + found = [ind for ind in indicators if ind in user_context.lower()] + if found: + reasons.append(f"user context indicates open PRs should exist (matched: {', '.join(found)})") + + if reasons: + return { + "status": "untrusted_empty", + "reasons": reasons, + "corroborated": corroborated, + } + + return { + "status": "trusted_empty", + "reasons": [], + "corroborated": corroborated, + } diff --git a/skills/llm-project-workflow/SKILL.md b/skills/llm-project-workflow/SKILL.md index 3a71e50..ad0d098 100644 --- a/skills/llm-project-workflow/SKILL.md +++ b/skills/llm-project-workflow/SKILL.md @@ -150,12 +150,33 @@ Worktree folder = branch with `/` replaced by `-` 6. Implement the narrow scope only — no unrelated refactors or formatting churn. 7. Add/update focused tests when behavior changes. 8. Run the checks (tests, compile/lint, `git diff --check`, secret scan). -9. Commit with an issue-linked message. -10. Push the branch. -11. Open a PR to `master`. -12. **If you are the author, stop before review/merge.** -13. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism. -14. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include: + Record the branch name and `HEAD` SHA at validation time — the drift + check in step 9 compares against exactly this state. +9. **Branch proof before commit (#177):** prove and state, immediately + before staging/committing (`author_proofs.verify_branch_for_commit`, + `author_proofs.detect_branch_drift`): + - current branch (`git branch --show-current`) equals the intended + feature branch from the issue claim + - current branch is not `master`, `main`, `develop`, `development`, or + `dev` + - branch and `HEAD` have not changed since validation (step 8) — in a + shared checkout another session may switch branches mid-session; + treat that as expected and **stop before committing** when detected + If any check fails, stop and reconcile; do not commit. +10. Commit with an issue-linked message. +11. **Branch proof before push (#177):** prove that the local branch, the + push target branch, and the intended issue branch all match, and that + none of them is a protected branch + (`author_proofs.verify_push_target`). If a commit accidentally landed + on a protected branch, do **not** push: report the accident and the + exact repair steps (`author_proofs.assess_protected_branch_commit`) — + never silently continue after a repair. +12. Push the branch. +13. Open a PR to `master`. The final report must include the branch proofs + from steps 9 and 11 (`author_proofs.build_commit_push_report`). +14. **If you are the author, stop before review/merge.** +15. **Normal issue work must not directly push to `master`.** PR content should be merged through the forge PR merge mechanism. +16. Direct push to `master` is allowed only as a documented recovery exception. If used, the final report must include: - why the PR merge path could not be used - exact commits pushed - PR metadata state @@ -196,19 +217,27 @@ Worktree folder = branch with `/` replaced by `-` Both configured repos must be reported with state filter, pagination proof, and open-PR count (`review_proofs.assess_inventory_completeness` and `resolve_repos_from_user_reference`). -7. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files. -8. Run the tests. Validation reporting must include the exact command and +7. **Role-boundary proof (#175):** a reviewer queue task must not silently + become author implementation. If no eligible PR exists, stop with the + queue report. Do not claim issues, create branches, commit, push, or open + PRs unless the operator explicitly retasks the run as author work. Mixed + reviewer+author namespace use must be reported with a justification, and + scratch-only notes are not durable evidence unless posted or committed + intentionally (`review_proofs.assess_role_boundary`). +8. Inspect the full diff; confirm scope matches the linked issue; flag unrelated files. +9. Run the tests. Validation reporting must include the exact command and exact results: pass/fail, counts of tests passed/skipped/failed, any ignored paths and why they are safe to ignore, and whether the command differs from the repository's canonical validation command. Only claim a validation result after the command has completed and its output has been read (`review_proofs.assess_validation_report`). -9. **Do not merge if checks fail. Do not merge if the reviewer is the author.** -10. The final report must distinguish (`review_proofs.build_final_report`): +10. **Do not merge if checks fail. Do not merge if the reviewer is the author.** +11. The final report must distinguish (`review_proofs.build_final_report`): identity eligible; PR author different from reviewer; session - contamination absent (with evidence); validation performed on the pinned - head; merge performed; issue status verified. If any proof is missing, - stop or downgrade the result instead of merging confidently. + contamination absent (with evidence); role boundary clean; validation + performed on the pinned head; merge performed; issue status verified. If + any proof is missing, stop or downgrade the result instead of merging + confidently. ## G. Merge / cleanup workflow diff --git a/skills/llm-project-workflow/templates/review-pr.md b/skills/llm-project-workflow/templates/review-pr.md index 5088d39..8485d6f 100644 --- a/skills/llm-project-workflow/templates/review-pr.md +++ b/skills/llm-project-workflow/templates/review-pr.md @@ -23,6 +23,10 @@ Rules (llm-project-workflow): - You must NOT be the PR author. If the authenticated user == PR author, stop. A different LLM-Agent-SHA does NOT make you a different actor — only a different authenticated Gitea user does (docs/llm-agent-sha.md). +- Do not pivot from a reviewer queue task into author implementation unless + the operator explicitly retasks the run. If author namespace was used, the + final report must justify why; author mutations after reviewer queue work + without explicit authorization are a role-boundary violation. - Do not merge if any check fails. Steps: @@ -39,6 +43,10 @@ Steps: cannot evidence whether this session authored/touched the PR branch, report contamination as UNKNOWN (not contaminated, not clean) and choose another PR or stop. + Role-boundary claims must also be evidence-backed (#175): report whether + reviewer namespace, author namespace, author mutations, or review mutations + occurred. Use `review_proofs.assess_role_boundary`; if it is not clean, + downgrade or stop instead of claiming an A-level run. 5. scripts/worktree-review # detached, branches/review-* cd branches/review- 6. Checkout proof (#173) — prove and state, before any diff review or diff --git a/skills/llm-project-workflow/templates/start-issue.md b/skills/llm-project-workflow/templates/start-issue.md index 194d6aa..54933fc 100644 --- a/skills/llm-project-workflow/templates/start-issue.md +++ b/skills/llm-project-workflow/templates/start-issue.md @@ -26,8 +26,19 @@ Steps: cd branches/-issue-- 6. Implement the narrow scope only; add/update focused tests if behavior changes. 7. Checks: run the test suite, compile/lint changed files, git diff --check, - and scan the diff for secrets. -8. Commit (issue-linked message), push the branch, open a PR to master. + and scan the diff for secrets. Record the branch name and HEAD SHA at + validation time. +8. Branch proof before commit (#177) — prove and state: + - git branch --show-current == the intended issue branch from step 5 + - the branch is NOT master/main/develop/development/dev + - branch and HEAD unchanged since step 7 (another session can switch a + shared checkout mid-session; if drift is detected, STOP and reconcile + before committing) + If a commit accidentally lands on a protected branch: do NOT push; + report the accident and the exact repair steps — never silently continue. +9. Commit (issue-linked message). Branch proof before push (#177): local + branch == push target branch == intended issue branch, none protected. + Then push the branch and open a PR to master. *The PR body MUST use closing keywords like `Closes #N` or `Fixes #N` to close the issue; do NOT use `Implements #N` or `Refs #N` for closing, as Gitea will not auto-close it.* Include an "LLM Handoff Metadata" block in the PR body (attribution only; never an eligibility input — docs/llm-agent-sha.md): @@ -40,7 +51,7 @@ Steps: - Branch: - Worktree: - Self-review allowed: no -9. Stop before review/merge — you are the author. +10. Stop before review/merge — you are the author. Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md §K (compact; long form only on the high-risk triggers), including the author diff --git a/tests/test_author_proofs.py b/tests/test_author_proofs.py new file mode 100644 index 0000000..cc46ec6 --- /dev/null +++ b/tests/test_author_proofs.py @@ -0,0 +1,234 @@ +"""Tests for author-side branch-identity proofs (Issue #177). + +Issue #177 (author-side counterpart of the #173 reviewer proofs) requires +author workflows to *prove* local git state before staging, committing, or +pushing, instead of discovering drift after the fact: + +1. The current branch equals the intended feature branch and is never a + protected branch (master/main/develop/development/dev). +2. Branch or HEAD drift between validation and commit — including external + branch switches in a shared worktree — stops the workflow. +3. A push requires local branch, remote target branch, and intended issue + branch to all match. +4. An accidental commit on a protected branch must not be pushed and its + repair must be reported, never silently continued. + +These are the harness assertions from the issue's Required behavior 5. +""" +import sys +import unittest + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +from author_proofs import ( # noqa: E402 + PROTECTED_BRANCHES, + assess_protected_branch_commit, + build_commit_push_report, + detect_branch_drift, + verify_branch_for_commit, + verify_push_target, +) + +FEATURE = "feat/issue-177-branch-drift-proofs" +HEAD_1 = "64dc334a92685b7b6a1fdb7ffe363f02a69f5dbd" +HEAD_2 = "ccc5ef79dfe629853e144763238593bd808d57e0" + + +class TestProtectedBranches(unittest.TestCase): + def test_known_protected_names(self): + for name in ("master", "main", "develop", "development", "dev"): + self.assertIn(name, PROTECTED_BRANCHES) + + +class TestVerifyBranchForCommit(unittest.TestCase): + """Required behavior 1: prove the branch before staging/committing.""" + + def test_on_intended_feature_branch_is_proven(self): + proof = verify_branch_for_commit(FEATURE, FEATURE) + self.assertTrue(proof["proven"]) + self.assertFalse(proof["block"]) + + def test_commit_attempted_while_on_master_is_blocked(self): + # Harness assertion (behavior 5, bullet 1). + proof = verify_branch_for_commit("master", FEATURE) + self.assertFalse(proof["proven"]) + self.assertTrue(proof["block"]) + self.assertTrue(any("master" in r for r in proof["reasons"])) + + def test_every_protected_branch_is_blocked_as_current(self): + for name in PROTECTED_BRANCHES: + proof = verify_branch_for_commit(name, FEATURE) + self.assertTrue(proof["block"], name) + + def test_intended_branch_may_not_be_protected(self): + proof = verify_branch_for_commit("master", "master") + self.assertFalse(proof["proven"]) + self.assertTrue(proof["block"]) + + def test_wrong_feature_branch_is_blocked(self): + proof = verify_branch_for_commit("feat/issue-178-other-work", FEATURE) + self.assertFalse(proof["proven"]) + self.assertTrue(proof["block"]) + + def test_missing_current_branch_fails_closed(self): + proof = verify_branch_for_commit("", FEATURE) + self.assertTrue(proof["block"]) + + def test_missing_intended_branch_fails_closed(self): + proof = verify_branch_for_commit(FEATURE, None) + self.assertTrue(proof["block"]) + + +class TestBranchDrift(unittest.TestCase): + """Required behaviors 2 + 3: drift between validation and commit stops + the workflow.""" + + def test_no_drift_when_branch_and_head_unchanged(self): + drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1) + self.assertFalse(drift["drifted"]) + self.assertFalse(drift["block"]) + + def test_branch_drift_between_validation_and_commit_is_blocked(self): + # Harness assertion (behavior 5, bullet 2). + drift = detect_branch_drift(FEATURE, HEAD_1, "feat/other", HEAD_1) + self.assertTrue(drift["drifted"]) + self.assertTrue(drift["block"]) + + def test_shared_worktree_branch_switch_is_detected(self): + # Harness assertion (behavior 5, bullet 4): an external session + # switching the shared checkout to another branch (e.g. master) + # must be detected as drift, not treated as exceptional noise. + drift = detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1) + self.assertTrue(drift["drifted"]) + self.assertTrue(drift["block"]) + self.assertTrue(any("switch" in r.lower() for r in drift["reasons"])) + + def test_head_moved_since_validation_is_blocked(self): + drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_2) + self.assertTrue(drift["drifted"]) + self.assertTrue(drift["block"]) + self.assertTrue(any("HEAD" in r for r in drift["reasons"])) + + def test_missing_state_fails_closed(self): + drift = detect_branch_drift(FEATURE, HEAD_1, FEATURE, None) + self.assertTrue(drift["drifted"]) + self.assertTrue(drift["block"]) + + +class TestVerifyPushTarget(unittest.TestCase): + """Required behavior 1 (push leg) + acceptance: push needs proof that + local, remote, and intended branches all match.""" + + def test_matching_local_remote_and_intended_is_proven(self): + proof = verify_push_target(FEATURE, FEATURE, FEATURE) + self.assertTrue(proof["proven"]) + self.assertFalse(proof["block"]) + + def test_push_target_mismatch_is_blocked(self): + # Harness assertion (behavior 5, bullet 3). + proof = verify_push_target(FEATURE, "feat/issue-178-other-work", FEATURE) + self.assertFalse(proof["proven"]) + self.assertTrue(proof["block"]) + + def test_local_branch_differs_from_intended_is_blocked(self): + proof = verify_push_target("feat/other", FEATURE, FEATURE) + self.assertTrue(proof["block"]) + + def test_pushing_a_protected_branch_is_blocked(self): + proof = verify_push_target("master", "master", "master") + self.assertFalse(proof["proven"]) + self.assertTrue(proof["block"]) + + def test_missing_remote_target_fails_closed(self): + proof = verify_push_target(FEATURE, "", FEATURE) + self.assertTrue(proof["block"]) + + +class TestProtectedBranchAccident(unittest.TestCase): + """Required behavior 4: accidental protected-branch commits must not be + pushed and their repair must be reported.""" + + def test_feature_branch_commit_is_not_an_accident(self): + result = assess_protected_branch_commit(FEATURE) + self.assertFalse(result["accident"]) + self.assertEqual(result["violations"], []) + + def test_commit_on_master_is_an_accident_and_must_not_push(self): + result = assess_protected_branch_commit( + "master", pushed=False, repair_reported=True + ) + self.assertTrue(result["accident"]) + self.assertTrue(result["must_not_push"]) + self.assertEqual(result["violations"], []) + self.assertTrue(result["repair_required"]) + + def test_pushing_the_accident_is_a_violation(self): + result = assess_protected_branch_commit( + "master", pushed=True, repair_reported=True + ) + self.assertTrue(any("push" in v.lower() for v in result["violations"])) + + def test_silent_repair_is_a_violation(self): + # Harness assertion (behavior 5, bullet 5): the repair path must not + # silently continue without reporting. + result = assess_protected_branch_commit( + "master", pushed=False, repair_reported=False + ) + self.assertTrue(any("report" in v.lower() for v in result["violations"])) + + +class TestCommitPushReport(unittest.TestCase): + """Acceptance criteria: the final report includes branch proof before + commit and before push, and blocks instead of continuing.""" + + def _report(self, **overrides): + kwargs = { + "commit_proof": verify_branch_for_commit(FEATURE, FEATURE), + "drift": detect_branch_drift(FEATURE, HEAD_1, FEATURE, HEAD_1), + "push_proof": verify_push_target(FEATURE, FEATURE, FEATURE), + "accident": assess_protected_branch_commit(FEATURE), + } + kwargs.update(overrides) + return build_commit_push_report(**kwargs) + + def test_fully_proven_report_is_ok(self): + report = self._report() + self.assertEqual(report["status"], "ok") + self.assertTrue(report["branch_proof_before_commit"]) + self.assertTrue(report["branch_proof_before_push"]) + self.assertFalse(report["drift_detected"]) + self.assertEqual(report["violations"], []) + + def test_commit_proof_failure_blocks(self): + report = self._report( + commit_proof=verify_branch_for_commit("master", FEATURE) + ) + self.assertEqual(report["status"], "blocked") + self.assertFalse(report["branch_proof_before_commit"]) + + def test_drift_blocks(self): + report = self._report( + drift=detect_branch_drift(FEATURE, HEAD_1, "master", HEAD_1) + ) + self.assertEqual(report["status"], "blocked") + self.assertTrue(report["drift_detected"]) + + def test_push_proof_failure_blocks(self): + report = self._report( + push_proof=verify_push_target(FEATURE, "feat/other", FEATURE) + ) + self.assertEqual(report["status"], "blocked") + self.assertFalse(report["branch_proof_before_push"]) + + def test_accident_violations_block(self): + report = self._report( + accident=assess_protected_branch_commit( + "master", pushed=False, repair_reported=False + ) + ) + self.assertEqual(report["status"], "blocked") + self.assertTrue(report["violations"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_operator_guide.py b/tests/test_operator_guide.py index b01d978..36767aa 100644 --- a/tests/test_operator_guide.py +++ b/tests/test_operator_guide.py @@ -46,8 +46,8 @@ EXPECTED_SKILLS = [ "gitea-resolve-task-capability", "profile-switching", "redaction-security-review", - "jenkins-readonly", - "glitchtip-readonly", + "jenkins-mcp", + "glitchtip-mcp", "release-operator", ] @@ -234,8 +234,8 @@ class TestProjectSkills(GuideTestBase): with patch.dict(os.environ, AUTHOR_ENV, clear=True): r = mcp_list_project_skills() by_name = {s["name"]: s for s in r["skills"]} - self.assertNotEqual(by_name["jenkins-readonly"]["status"], "available") - self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available") + self.assertNotEqual(by_name["jenkins-mcp"]["status"], "available") + self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available") def test_no_urls_in_registry(self): with patch.dict(os.environ, AUTHOR_ENV, clear=True): @@ -245,6 +245,17 @@ class TestProjectSkills(GuideTestBase): self.assertNotIn("http://", blob) self.assertNotIn("keychain:", blob) + def test_enabled_but_no_usable_tools_negative_assertion(self): + """Negative assertion for 'enabled but no usable tools' (per issue #146).""" + with patch.dict(os.environ, AUTHOR_ENV, clear=True): + r = mcp_list_project_skills() + by_name = {s["name"]: s for s in r["skills"]} + # jenkins-mcp is designed-not-implemented; even if "enabled" in config, + # it should not be usable/available to current profile without tools. + self.assertIn("jenkins-mcp", by_name) + self.assertEqual(by_name["jenkins-mcp"]["status"], "designed-not-implemented") + self.assertFalse(by_name["jenkins-mcp"].get("available_to_current_profile", False)) + # --------------------------------------------------------------------------- # mcp_get_skill_guide diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 68d90d1..9570d5d 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -23,9 +23,11 @@ sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.par from review_proofs import ( # noqa: E402 assess_controller_handoff, assess_inventory_completeness, + assess_role_boundary, assess_self_review_contamination, assess_validation_report, build_final_report, + pr_inventory_trust_gate, resolve_repos_from_user_reference, verify_pinned_head_checkout, ) @@ -99,6 +101,21 @@ def _good_contamination(): ) +def _good_role_boundary(): + return assess_role_boundary( + { + "task_role": "reviewer", + "task_kind": "blind_pr_queue_review", + "reviewer_namespace_used": True, + "author_namespace_used": False, + "author_mutations": [], + "review_mutations": [], + "operator_authorized_author_work": False, + "scratch_evidence_claimed": False, + } + ) + + class TestCheckoutProof(unittest.TestCase): """Required behavior 1 + 2: prove HEAD == pinned PR head or stop.""" @@ -368,6 +385,74 @@ class TestSelfReviewContamination(unittest.TestCase): self.assertEqual(result["status"], "unknown") +class TestRoleBoundary(unittest.TestCase): + """Issue #175: reviewer queue tasks must not pivot into author work.""" + + def test_reviewer_queue_without_author_mutations_is_clean(self): + result = _good_role_boundary() + self.assertEqual(result["status"], "clean") + self.assertEqual(result["violations"], []) + + def test_reviewer_queue_author_mutation_without_authorization_violates(self): + result = assess_role_boundary( + { + "task_role": "reviewer", + "task_kind": "blind_pr_queue_review", + "reviewer_namespace_used": True, + "author_namespace_used": True, + "author_mutations": ["claim issue #171", "push branch"], + "operator_authorized_author_work": False, + "mixed_namespace_justification": ( + "author namespace was used for implementation" + ), + } + ) + self.assertEqual(result["status"], "violation") + self.assertTrue(any("pivot" in r for r in result["violations"])) + + def test_mixed_namespace_use_without_justification_is_warning(self): + result = assess_role_boundary( + { + "task_role": "reviewer", + "task_kind": "blind_pr_queue_review", + "reviewer_namespace_used": True, + "author_namespace_used": True, + "author_mutations": [], + } + ) + self.assertEqual(result["status"], "warning") + self.assertTrue( + any("mixed" in r.lower() for r in result["reasons"]) + ) + + def test_author_task_cannot_perform_review_mutations(self): + result = assess_role_boundary( + { + "task_role": "author", + "reviewer_namespace_used": False, + "author_namespace_used": True, + "review_mutations": ["approve PR"], + } + ) + self.assertEqual(result["status"], "violation") + self.assertTrue( + any("reviewer-only" in r for r in result["violations"]) + ) + + def test_scratch_only_notes_are_not_durable_evidence(self): + result = assess_role_boundary( + { + "task_role": "reviewer", + "task_kind": "blind_pr_queue_review", + "reviewer_namespace_used": True, + "scratch_evidence_claimed": True, + "scratch_evidence_durable": False, + } + ) + self.assertEqual(result["status"], "warning") + self.assertTrue(any("scratch-only" in r for r in result["reasons"])) + + class TestFinalReport(unittest.TestCase): """Required behavior 6 + acceptance criteria: the report must distinguish each proof, and only a fully proven run earns an "A".""" @@ -381,6 +466,7 @@ class TestFinalReport(unittest.TestCase): "identity_eligible": True, "merge_performed": False, "issue_status_verified": True, + "role_boundary": _good_role_boundary(), } kwargs.update(overrides) return build_final_report(**kwargs) @@ -393,6 +479,7 @@ class TestFinalReport(unittest.TestCase): self.assertTrue(report["identity_eligible"]) self.assertTrue(report["pr_author_distinct_from_reviewer"]) self.assertEqual(report["session_contamination"], "clean") + self.assertEqual(report["role_boundary"], "clean") self.assertTrue(report["validated_on_pinned_head"]) self.assertFalse(report["merge_performed"]) self.assertTrue(report["issue_status_verified"]) @@ -465,6 +552,37 @@ class TestFinalReport(unittest.TestCase): self.assertNotEqual(report["grade"], "A") self.assertFalse(report["merge_allowed"]) + def test_missing_role_boundary_downgrades_and_blocks_merge(self): + kwargs = { + "checkout_proof": _good_checkout(), + "inventory": _good_inventory(), + "validation": _good_validation(), + "contamination": _good_contamination(), + "identity_eligible": True, + "merge_performed": False, + "issue_status_verified": True, + } + report = build_final_report(**kwargs) + self.assertNotEqual(report["grade"], "A") + self.assertFalse(report["merge_allowed"]) + self.assertEqual(report["role_boundary"], "warning") + + def test_role_boundary_violation_blocks_report(self): + boundary = assess_role_boundary( + { + "task_role": "reviewer", + "task_kind": "blind_pr_queue_review", + "reviewer_namespace_used": True, + "author_namespace_used": True, + "author_mutations": ["create PR"], + "operator_authorized_author_work": False, + "mixed_namespace_justification": "implementation pivot", + } + ) + report = self._report(role_boundary=boundary) + self.assertEqual(report["grade"], "blocked") + self.assertFalse(report["merge_allowed"]) + class TestStdoutIsolation(unittest.TestCase): """Regression test for #178: tests must not close or corrupt stdout/stderr @@ -713,5 +831,87 @@ class TestControllerHandoff(unittest.TestCase): self.assertIn("issue #182", skill) +class TestPRInventoryTrustGate(unittest.TestCase): + """Issue #194: unit tests for the PR inventory trust gate.""" + + def setUp(self): + self.profile = { + "profile_name": "prgs-reviewer", + "allowed_operations": ["read", "gitea.read", "gitea.pr.approve"], + } + self.local_url = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git" + + def test_trusted_nonempty(self): + res = pr_inventory_trust_gate([{"number": 1}]) + self.assertEqual(res["status"], "trusted_nonempty") + self.assertFalse(res["corroborated"]) + + def test_inventory_error_none_or_not_list(self): + self.assertEqual(pr_inventory_trust_gate(None)["status"], "inventory_error") + self.assertEqual(pr_inventory_trust_gate("not a list")["status"], "inventory_error") + + def test_untrusted_empty_no_pagination_or_corroboration(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=None, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertIn("pagination finality not proven and open_pr_counter corroboration is missing or non-zero", res["reasons"]) + + def test_untrusted_empty_profile_permission_mismatch(self): + bad_profile = {"profile_name": "prgs-bad", "allowed_operations": ["write"]} + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=bad_profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertIn("authenticated profile lacks read permissions", res["reasons"]) + + def test_untrusted_empty_remote_url_mismatch(self): + bad_url = "https://gitea.prgs.cc/other-org/other-repo.git" + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=bad_url, user_context=None, + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertIn("local remote URL does not match target repository 'Scaled-Tech-Consulting/Gitea-Tools'", res["reasons"]) + + def test_untrusted_empty_user_context_indicates_prs(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context="please check open PR #181", + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "untrusted_empty") + self.assertTrue(any("user context indicates open PRs should exist" in r for r in res["reasons"])) + + def test_trusted_empty_with_corroboration(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=0, has_finality_metadata=False + ) + self.assertEqual(res["status"], "trusted_empty") + self.assertTrue(res["corroborated"]) + + def test_trusted_empty_with_finality_metadata(self): + res = pr_inventory_trust_gate( + [], remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", + state="open", authenticated_profile=self.profile, + local_remote_url=self.local_url, user_context=None, + corroboration_open_pr_counter=None, has_finality_metadata=True + ) + self.assertEqual(res["status"], "trusted_empty") + self.assertTrue(res["corroborated"]) + + if __name__ == "__main__": unittest.main()