Compare commits

..
11 Commits
Author SHA1 Message Date
sysadmin cbaed49761 feat(author): harden reporting and claim capability per #183
- Add 'mark_issue' to TASK_MAP in resolve_task_capability for exact mutation proof (closes gap on unknown treated as allowed).
- Add assess_controller_handoff and integrate into build_final_report (downgrades missing handoff).
- Add tests for handoff and update existing.
- Update SKILL.md and review-pr.md to mandate exact 'Controller Handoff' section, exact sweep evidence, PR head SHA in reports.
- Author profile used throughout; no review/merge.

Refs #183
2026-07-05 16:46:48 -04:00
sysadmin c6fd0fd963 Merge pull request 'Add fail-closed PR inventory wall when list_prs returns an unexpected empty queue (Issue #194)' (#195) from feat/issue-194-fail-closed-pr-inventory-wall into master 2026-07-05 15:27:08 -05:00
sysadmin 4e8b6cc5a9 Merge pull request 'Add role-boundary proofs for reviewer queue workflows (Issue #175)' (#192) from feat/issue-175-role-pivot-wall into master 2026-07-05 15:24:45 -05:00
sysadmin 9a35b80e9a Merge pull request 'Fix MCP registration and naming for Jenkins/GlitchTip servers and add discoverability coverage (#146)' (#190) from feat/issue-150-add-integration-discoverability-coverage into master 2026-07-05 15:18:29 -05:00
sysadmin 2b6a60a189 Merge pull request 'Add fail-closed branch-identity proofs for author commit/push workflow (Issue #177)' (#181) from feat/issue-177-branch-drift-proofs into master 2026-07-05 15:13:24 -05:00
sysadmin b354c93710 Implement fail-closed PR inventory trust gate for Issue #194 2026-07-05 16:06:17 -04:00
sysadmin 4dcf8fdfe4 Merge pull request 'Require and validate Controller Handoff sections in workflow final reports (Issue #182)' (#186) from feat/issue-182-controller-handoff-enforcement into master 2026-07-05 14:54:07 -05:00
sysadmin 574a9ea7c1 Add role-boundary proofs for reviewer queue workflows 2026-07-05 15:42:35 -04:00
sysadmin 5b1f0be2be feat: fix MCP registration/naming for Jenkins/GlitchTip and add discoverability negative assertions (#146)
- Updated mcp_server.py operator guide entries to use jenkins-mcp / glitchtip-mcp names matching mcp-control-plane.
- Added reload/reconnect instructions for clients in notes.
- Added test for 'enabled but no usable tools' negative assertion.
- Updated EXPECTED_SKILLS and assertions in test_operator_guide.py.

Scoped to issue #146. Author profile.

Refs #146
2026-07-05 15:41:49 -04:00
sysadminandClaude Fable 5 45c5cac2bc Require and validate Controller Handoff sections in workflow final reports
- Upgrade SKILL.md §K compact format to the issue #182 canonical field set
  (Task/Repo/Role/Identity/Issue-PR/Branch-SHA/Files/Validation/Mutations/
  Current status/Blockers/Next/Safety) plus role-specific field lists for
  review/merge, author, and queue/inventory tasks.
- Point the review-pr, merge-pr, and start-issue template handoff lines at
  the exactly-titled Controller Handoff section with their role fields.
- Add review_proofs.assess_controller_handoff(): reports without the exact
  section are 'missing' (downgraded), present-but-partial are 'incomplete'
  with the absent fields listed, and role extras are enforced per role.
- Add TestControllerHandoff (8 tests) including a SKILL.md doc-contract
  test so the documented requirement cannot silently rot.

The handoff supplements the full report; full-report validation rules are
unchanged.

Closes #182

Co-Authored-By: Claude Fable 5 <[email protected]>
2026-07-05 15:36:57 -04:00
sysadminandClaude Fable 5 941ada38c2 feat(author-workflow): add fail-closed branch-identity proofs for commit/push (#177)
Adds author_proofs.py, the author-side counterpart of review_proofs.py
(#173), turning the branch-drift incident from PR #176 into fail-closed
gates:

- verify_branch_for_commit: the current branch must equal the intended
  feature branch and may never be a protected branch (master, main,
  develop, development, dev); missing state fails closed.
- detect_branch_drift: any branch or HEAD change between validation time
  and commit time — including an external branch switch in a shared
  worktree, which is treated as an expected event to detect — blocks the
  commit until reconciled.
- verify_push_target: a push requires the local branch, remote target
  branch, and intended issue branch to all match, none protected.
- assess_protected_branch_commit: an accidental protected-branch commit
  must never be pushed, requires repair, and the repair must be reported;
  pushing the accident or silently continuing is a violation.
- build_commit_push_report: final-report block carrying the branch proof
  before commit and before push; any failed proof, drift, or violation
  makes the status blocked.

tests/test_author_proofs.py (27 tests) covers the issue's harness
assertions: commit on master blocked; drift between validation and commit
blocked; push target mismatch blocked; shared-worktree branch switch
detected; repair path cannot silently continue without reporting.

SKILL.md section E and templates/start-issue.md now require recording the
validation-time branch/HEAD, the branch proof before commit, the branch
proof before push, and non-silent accident repair. No review/merge/
permission gate is weakened.

Closes #177

Co-Authored-By: Claude Fable 5 <[email protected]>
2026-07-05 15:16:36 -04:00
10 changed files with 1359 additions and 187 deletions
+217
View File
@@ -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 23: 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", []))
),
}
+6 -6
View File
@@ -2307,26 +2307,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.",
+389 -65
View File
@@ -202,15 +202,13 @@ def assess_validation_report(report):
*report* keys: ``command``, ``output_read``, ``result`` ('pass'/'fail'),
``passed``/``failed``/``skipped`` counts, ``ignored_paths`` (each with a
``justification``), ``canonical_command``, ``deviation_justification``,
``is_stdout_capture_fix``, ``normal_pytest_summary``.
``justification``), ``canonical_command``, ``deviation_justification``.
Verdicts:
- 'invalid' — the result may not be claimed at all (no command stated,
or the command output was never read).
- 'weak' — claimable but downgraded (missing counts, unjustified
ignored paths, unexplained deviation from the canonical command, or
missing normal summary on stdout capture fix).
ignored paths, unexplained deviation from the canonical command).
- 'strong' — full evidence.
"""
reasons = []
@@ -254,14 +252,6 @@ def assess_validation_report(report):
"command and the deviation is not justified"
)
# For stdout-capture fixes (e.g. Issue #178)
if report.get("is_stdout_capture_fix") is True:
if not report.get("normal_pytest_summary"):
reasons.append(
"stdout-capture fix validation is missing normal pytest summary output "
"or still requires junitxml workaround"
)
verdict = "strong" if not reasons else "weak"
return {"verdict": verdict, "claimable": True, "reasons": reasons}
@@ -340,10 +330,99 @@ 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, controller_handoff=None,
capability_proof=None, sweep_proof=None):
issue_status_verified, role_boundary=None,
controller_handoff=None, capability_proof=None,
sweep_proof=None):
"""Required behavior 6 + acceptance criteria: one report, distinct proofs.
Combines the individual proof verdicts into the final-report fields the
@@ -359,6 +438,22 @@ 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")
handoff = _normalize_controller_handoff_proof(controller_handoff)
capability_proof = capability_proof or {
"proven": False,
"reasons": ["capability proof not provided"],
}
sweep_proof = sweep_proof or {
"proven": False,
"verdict": "invalid",
"reasons": ["secret/provenance sweep proof not provided"],
}
downgrade_reasons = []
if not identity_eligible:
@@ -378,26 +473,26 @@ 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")
handoff = assess_controller_handoff(controller_handoff or "")
if not handoff.get("present"):
downgrade_reasons.append("Controller Handoff section missing or incomplete")
if not handoff["complete"]:
downgrade_reasons.append("Controller Handoff missing or incomplete")
downgrade_reasons.extend(handoff.get("reasons", []))
if capability_proof and not capability_proof.get("proven"):
downgrade_reasons.append("capability proof verification failed")
if not capability_proof.get("proven"):
downgrade_reasons.append("exact mutation capability proof missing")
downgrade_reasons.extend(capability_proof.get("reasons", []))
if sweep_proof and not sweep_proof.get("proven"):
downgrade_reasons.append("secret/provenance sweep proof verification failed")
if not sweep_proof.get("proven"):
downgrade_reasons.append("exact secret/provenance sweep proof missing")
downgrade_reasons.extend(sweep_proof.get("reasons", []))
merge_allowed = (
identity_eligible
and checkout_proven
and contamination_status == "clean"
and role_status == "clean"
and validation_claimable
and validation.get("verdict") != "invalid"
)
@@ -408,6 +503,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"
@@ -424,6 +520,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":
@@ -432,36 +529,46 @@ def build_final_report(checkout_proof, inventory, validation, contamination,
"merge_allowed": merge_allowed,
"merge_performed": bool(merge_performed),
"issue_status_verified": bool(issue_status_verified),
"controller_handoff_present": handoff.get("present", False),
"controller_handoff_present": handoff["present"],
"controller_handoff_complete": handoff["complete"],
"capability_proof": bool(capability_proof.get("proven")),
"secret_sweep_proof": bool(sweep_proof.get("proven")),
}
def assess_controller_handoff(report_text: str) -> dict:
"""Required for author and reviewer final reports (#183).
The report must contain an exactly titled 'Controller Handoff' section
(compact format preferred). Missing it downgrades the report.
"""
if not report_text:
return {"present": False, "reasons": ["no report text"]}
text = str(report_text)
if "Controller Handoff" not in text:
def _normalize_controller_handoff_proof(controller_handoff):
"""Normalize old boolean handoff proof and rich #182 handoff verdicts."""
if controller_handoff is None:
return {
"present": False,
"reasons": [
"final report missing exactly-titled 'Controller Handoff' section"
],
"complete": False,
"reasons": ["Controller Handoff proof not provided"],
}
return {"present": True, "reasons": []}
if isinstance(controller_handoff, str):
controller_handoff = assess_controller_handoff(controller_handoff)
if not isinstance(controller_handoff, dict):
return {
"present": False,
"complete": False,
"reasons": ["Controller Handoff proof has invalid type"],
}
if "verdict" in controller_handoff:
verdict = controller_handoff.get("verdict")
return {
"present": verdict in {"complete", "incomplete"},
"complete": verdict == "complete",
"reasons": list(controller_handoff.get("reasons") or []),
}
present = bool(controller_handoff.get("present"))
return {
"present": present,
"complete": present,
"reasons": list(controller_handoff.get("reasons") or []),
}
def assess_capability_proof(resolved_capabilities: dict) -> dict:
"""Required behavior: every mutation must have exact capability proof.
If a mutation task is unknown, unresolved, or lacks explicit resolver evidence,
the workflow must fail closed / be downgraded.
"""
"""Every mutation must have exact capability proof (#183)."""
reasons = []
if not resolved_capabilities:
return {
@@ -472,20 +579,22 @@ def assess_capability_proof(resolved_capabilities: dict) -> dict:
allowed = cap.get("allowed_in_current_session")
op = cap.get("required_operation_permission")
if not allowed:
reasons.append(f"task '{task}' requires permission '{op}' which is not allowed in current session")
# If the resolved result indicates an unknown task or missing flag
if "unknown" in str(cap.get("requested_task", "")).lower() or cap.get("allowed_in_current_session") is None:
reasons.append(f"task '{task}' could not be resolved to a known capability")
proven = not reasons
return {"proven": proven, "reasons": reasons}
reasons.append(
f"task '{task}' requires permission '{op}' which is not "
"allowed in current session"
)
if (
"unknown" in str(cap.get("requested_task", "")).lower()
or cap.get("allowed_in_current_session") is None
):
reasons.append(
f"task '{task}' could not be resolved to a known capability"
)
return {"proven": not reasons, "reasons": reasons}
def assess_secret_sweep(sweep_report: dict) -> dict:
"""Required behavior: secret/provenance sweeps must state exact command/method and scope.
*sweep_report* keys: ``method``, ``scope``, ``clean`` (bool).
Returns dict with ``proven`` (bool), ``verdict`` ('strong', 'weak', 'invalid'), and ``reasons`` list.
"""
"""Secret/provenance sweeps must state exact command/method and scope."""
if not sweep_report:
return {
"proven": False,
@@ -496,26 +605,24 @@ def assess_secret_sweep(sweep_report: dict) -> dict:
method = (sweep_report.get("method") or "").strip()
scope = (sweep_report.get("scope") or "").strip()
if not method:
reasons.append("secret sweep report missing exact scan command, pattern, or method")
reasons.append(
"secret sweep report missing exact scan command, pattern, or method"
)
if not scope:
reasons.append("secret sweep report missing scanned scope")
if sweep_report.get("clean") is not True:
reasons.append("secret sweep report did not confirm diff is clean")
verdict = "weak" if reasons else "strong"
proven = not reasons and sweep_report.get("clean") is True
return {
"proven": proven,
"proven": not reasons and sweep_report.get("clean") is True,
"verdict": verdict,
"reasons": reasons,
}
def assess_author_pr_report(pr_report: dict) -> dict:
"""Required behavior: author PR creation report must include PR number, branch, and exact head SHA.
*pr_report* keys: ``pr_number`` (int), ``branch`` (str), ``head_sha`` (str).
"""
"""Author PR reports must include PR number, branch, and exact head SHA."""
if not pr_report:
return {
"complete": False,
@@ -535,8 +642,225 @@ def assess_author_pr_report(pr_report: dict) -> dict:
elif not _FULL_SHA.match(head_sha):
reasons.append("PR head SHA is not a full 40-hex commit SHA")
complete = not reasons
return {
"complete": complete,
"complete": not reasons,
"reasons": reasons,
}
# ── Controller Handoff validation (Issue #182) ────────────────────────────────
#
# Every final report must end with a compact section titled exactly
# "Controller Handoff". Each required field is a (canonical name, aliases)
# pair; a field counts as present when any alias starts a bullet/label line
# inside the handoff section.
HANDOFF_HEADING = "Controller Handoff"
HANDOFF_BASE_FIELDS = (
("Task", ("task",)),
("Repo", ("repo", "repository", "repo/state")),
("Role", ("role",)),
("Identity", ("identity",)),
("Issue/PR", ("issue/pr", "issues/prs", "issue", "pr")),
("Branch/SHA", ("branch/sha", "branch", "head sha")),
("Files changed", ("files changed", "changed", "files")),
("Validation", ("validation",)),
("Mutations", ("mutations",)),
("Current status", ("current status", "status")),
("Blockers", ("blockers",)),
("Next", ("next",)),
("Safety", ("safety",)),
)
HANDOFF_ROLE_FIELDS = {
"review": (
("Selected PR", ("selected pr",)),
("Reviewer eligibility", ("reviewer eligibility", "eligibility")),
("Pinned reviewed head", ("pinned reviewed head", "pinned head")),
("Review decision", ("review decision", "decision")),
("Merge result", ("merge result",)),
("Linked issue status", ("linked issue status", "linked issue")),
("Cleanup status", ("cleanup status", "cleanup")),
),
"author": (
("Selected issue", ("selected issue",)),
("Claim/comment status", ("claim/comment status", "claim status",
"claim")),
("PR number opened", ("pr number opened", "pr opened", "pr number")),
("No review/merge confirmation", ("no review/merge",
"no review or merge")),
),
"inventory": (
("Repositories checked", ("repositories checked", "repos checked")),
("Open PR counts", ("open pr counts", "open pr count",
"open prs per repo")),
("Selected PR or reason", ("selected pr", "none selected",
"reason none selected")),
("Inventory completeness", ("inventory complete", "inventory scoped",
"inventory completeness")),
),
}
def _handoff_section_lines(report_text):
"""Return the lines of the Controller Handoff section, or None."""
lines = (report_text or "").splitlines()
start = None
for i, line in enumerate(lines):
bare = line.strip().lstrip("#").strip().rstrip(":")
if bare == HANDOFF_HEADING:
start = i + 1
break
if start is None:
return None
return lines[start:]
def assess_controller_handoff(report_text, role=None):
"""Issue #182: final reports without a Controller Handoff downgrade.
Verdicts:
- 'missing' — no exactly-titled section; the report is downgraded.
- 'incomplete' — section present but required fields absent (listed).
- 'complete' — all base fields plus the role-specific fields present.
*role* is 'review', 'author', 'inventory', or None (base fields only).
The handoff supplements the full report; this helper never validates
the full report body, only the continuation summary.
"""
section = _handoff_section_lines(report_text)
if section is None:
return {
"verdict": "missing",
"downgraded": True,
"missing_fields": [name for name, _ in HANDOFF_BASE_FIELDS],
"reasons": [
"final report has no section titled exactly "
f"'{HANDOFF_HEADING}'"
],
}
labels = []
for line in section:
stripped = line.strip().lstrip("-*").strip()
if ":" in stripped:
labels.append(stripped.split(":", 1)[0].strip().lower())
required = list(HANDOFF_BASE_FIELDS)
required.extend(HANDOFF_ROLE_FIELDS.get(role or "", ()))
missing = []
for name, aliases in required:
if not any(label.startswith(alias)
for label in labels for alias in aliases):
missing.append(name)
if missing:
return {
"verdict": "incomplete",
"downgraded": True,
"missing_fields": missing,
"reasons": [f"handoff missing required field: {m}"
for m in missing],
}
return {
"verdict": "complete",
"downgraded": False,
"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,
}
+68 -18
View File
@@ -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
@@ -298,22 +327,43 @@ The section title must be exactly "Controller Handoff" (or "Controller Handoff S
readability, not as a full human status report. PR bodies still carry the
full review detail — the handoff never replaces PR documentation.
Compact format (default):
Compact format (default, canonical field set per issue #182):
```md
## Controller Handoff
- Task:
- Repo/state:
- Issues/PRs:
- Changed:
- Repo:
- Role:
- Identity:
- Issue/PR:
- Branch/SHA:
- Files changed:
- Validation:
- Mutations:
- Current status:
- Blockers:
- Review:
- Next:
- Safety:
```
Role-specific fields (append to the compact block):
- review/merge tasks: `Selected PR:`, `Reviewer eligibility:`,
`Pinned reviewed head:`, `Review decision:`, `Merge result:`,
`Linked issue status:`, `Cleanup status:`
- author tasks: `Selected issue:`, `Claim/comment status:`,
`PR number opened:`, `No review/merge:` (explicit confirmation)
- queue/inventory tasks: `Repositories checked:`, `Open PR counts:`,
`Selected PR or reason none selected:`, `Inventory completeness:`
The section title must be exactly `Controller Handoff`.
`review_proofs.assess_controller_handoff()` validates this section; reports
missing it (or missing required fields) are downgraded. The handoff never
replaces the full report — it is the compact continuation summary at the end,
and the full report must still carry exact validation results and mutation
confirmation.
The `Safety:` line is never omitted; it is usually:
```text
@@ -35,5 +35,11 @@ Then run the cleanup template (worktree-cleanup.md):
- delete remote branch, remove local branch + worktree folder
- fetch/prune; confirm main checkout is clean and current (0 0).
Handoff: reviewer identity, merge result + commit, cleanup done, issue closed, PR metadata state/merged flag/hash, remote master hash, post-merge verification method used & verification results.
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
§K (long form — a merge is always high-risk), including the review/merge role
fields (Selected PR, Reviewer eligibility, Pinned reviewed head, Review
decision, Merge result, Linked issue status, Cleanup status) plus: merge
commit, PR metadata state/merged flag/hash, remote master hash, and the
post-merge verification method used & verification results. Reports missing
the handoff are downgraded (review_proofs.assess_controller_handoff).
```
@@ -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 <pr-head-branch> # detached, branches/review-*
cd branches/review-<pr-head-branch-slug>
6. Checkout proof (#173) — prove and state, before any diff review or
@@ -65,7 +73,10 @@ Steps:
- MCP-Profile: <profile name>
- Eligibility: passed/failed
Handoff: reviewer identity, PR author, scope verdict, checks + results, decision —
formatted per SKILL.md §K (compact by default; long form if a merge happened
or a gate blocked you); if you could not merge, name the exact gate.
Handoff: end with a section titled exactly `Controller Handoff` per SKILL.md
§K (compact by default; long form if a merge happened or a gate blocked you),
including the review/merge role fields: Selected PR, Reviewer eligibility,
Pinned reviewed head, Review decision, Merge result, Linked issue status,
Cleanup status. If you could not merge, name the exact gate. Reports missing
the handoff are downgraded (review_proofs.assess_controller_handoff).
```
@@ -26,8 +26,19 @@ Steps:
cd branches/<type>-issue-<n>-<slug>
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,9 +51,12 @@ Steps:
- Branch: <branch>
- Worktree: <worktree path>
- Self-review allowed: no
9. Stop before review/merge — you are the author.
10. Stop before review/merge — you are the author.
Handoff: issue #, branch, worktree path, files changed, checks + results, PR URL —
formatted as the compact Controller Handoff (SKILL.md §K; long form only on
the high-risk triggers); Review line: "Review needed — PR is open".
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
role fields: Selected issue, Claim/comment status, PR number opened, and an
explicit "No review/merge:" confirmation — plus branch, worktree path, files
changed, checks + results. Next line: "Review needed — PR is open". Reports
missing the handoff are downgraded (review_proofs.assess_controller_handoff).
```
+234
View File
@@ -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()
+15 -4
View File
@@ -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
+389 -84
View File
@@ -25,10 +25,12 @@ from review_proofs import ( # noqa: E402
assess_capability_proof,
assess_controller_handoff,
assess_inventory_completeness,
assess_role_boundary,
assess_secret_sweep,
assess_self_review_contamination,
assess_validation_report,
build_final_report,
pr_inventory_trust_gate,
resolve_repos_from_user_reference,
verify_pinned_head_checkout,
)
@@ -102,6 +104,64 @@ 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,
}
)
GOOD_HANDOFF = "\n".join([
"## Controller Handoff",
"",
"- Task: review PR #999",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: reviewer",
"- Identity: sysadmin / prgs-reviewer",
"- Issue/PR: #182 / PR #999",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: review only",
"- Current status: PR open",
"- Blockers: none",
"- Next: merge if approved",
"- Safety: no self-review; no self-merge; no secrets",
])
def _good_handoff():
return assess_controller_handoff(GOOD_HANDOFF)
def _good_capability_proof():
return assess_capability_proof({
"comment_issue": {
"requested_task": "comment_issue",
"required_operation_permission": "gitea.issue.comment",
"allowed_in_current_session": True,
}
})
def _good_secret_sweep(**overrides):
sweep = {
"method": "git diff prgs/master...HEAD | grep -iE 'token|secret'",
"scope": "full PR diff relative to prgs/master",
"clean": True,
}
sweep.update(overrides)
return assess_secret_sweep(sweep)
class TestCheckoutProof(unittest.TestCase):
"""Required behavior 1 + 2: prove HEAD == pinned PR head or stop."""
@@ -318,21 +378,6 @@ class TestValidationReporting(unittest.TestCase):
)
self.assertEqual(result["verdict"], "strong")
def test_stdout_capture_fix_without_summary_is_weak(self):
result = _good_validation(
is_stdout_capture_fix=True,
normal_pytest_summary=False
)
self.assertEqual(result["verdict"], "weak")
self.assertTrue(any("stdout-capture" in r for r in result["reasons"]))
def test_stdout_capture_fix_with_summary_is_strong(self):
result = _good_validation(
is_stdout_capture_fix=True,
normal_pytest_summary=True
)
self.assertEqual(result["verdict"], "strong")
class TestSelfReviewContamination(unittest.TestCase):
"""Required behavior 5: contamination claims need evidence."""
@@ -386,6 +431,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"."""
@@ -399,7 +512,10 @@ class TestFinalReport(unittest.TestCase):
"identity_eligible": True,
"merge_performed": False,
"issue_status_verified": True,
"controller_handoff": "Controller Handoff\n- Task: test",
"role_boundary": _good_role_boundary(),
"controller_handoff": _good_handoff(),
"capability_proof": _good_capability_proof(),
"sweep_proof": _good_secret_sweep(),
}
kwargs.update(overrides)
return build_final_report(**kwargs)
@@ -412,6 +528,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"])
@@ -484,21 +601,56 @@ class TestFinalReport(unittest.TestCase):
self.assertNotEqual(report["grade"], "A")
self.assertFalse(report["merge_allowed"])
def test_failed_capability_proof_downgrades(self):
cap_proof = {
"proven": False,
"reasons": ["task mark_issue not 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 = self._report(capability_proof=cap_proof)
report = build_final_report(**kwargs)
self.assertNotEqual(report["grade"], "A")
self.assertTrue(any("capability" in r for r in report["downgrade_reasons"]))
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"])
def test_failed_capability_proof_downgrades(self):
report = self._report(
capability_proof={
"proven": False,
"reasons": ["task mark_issue not allowed"],
}
)
self.assertNotEqual(report["grade"], "A")
self.assertTrue(
any("capability" in r for r in report["downgrade_reasons"])
)
def test_failed_sweep_proof_downgrades(self):
sweep_proof = {
"proven": False,
"reasons": ["method missing"]
}
report = self._report(sweep_proof=sweep_proof)
report = self._report(
sweep_proof={
"proven": False,
"reasons": ["method missing"],
}
)
self.assertNotEqual(report["grade"], "A")
self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
@@ -615,32 +767,194 @@ class TestRepoNameDisambiguation(unittest.TestCase):
class TestControllerHandoff(unittest.TestCase):
"""#183: every final report must contain the Controller Handoff section."""
"""Issue #182: final reports must end with a Controller Handoff."""
def test_report_with_handoff_passes(self):
report = "some details\n\nController Handoff\n- Task: foo"
result = assess_controller_handoff(report)
self.assertTrue(result["present"])
BASE_HANDOFF = "\n".join([
"## Controller Handoff",
"",
"- Task: implement issue #182",
"- Repo: Scaled-Tech-Consulting/Gitea-Tools",
"- Role: author",
"- Identity: jcwalker3 / prgs-author",
"- Issue/PR: #182 / PR #999",
"- Branch/SHA: feat/x @ 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Files changed: review_proofs.py",
"- Validation: 700 passed, 6 skipped",
"- Mutations: one PR opened",
"- Current status: PR open",
"- Blockers: none",
"- Next: review PR #999",
"- Safety: no self-review; no self-merge; no secrets",
])
def test_report_without_handoff_downgrades(self):
report = "long details without the section"
result = assess_controller_handoff(report)
self.assertFalse(result["present"])
self.assertTrue(any("Controller Handoff" in r for r in result["reasons"]))
def test_report_without_handoff_is_downgraded(self):
result = assess_controller_handoff("long report text, no handoff")
self.assertEqual(result["verdict"], "missing")
self.assertTrue(result["downgraded"])
def test_wrong_title_is_downgraded(self):
text = self.BASE_HANDOFF.replace(
"## Controller Handoff", "## Handoff Summary")
result = assess_controller_handoff(text)
self.assertEqual(result["verdict"], "missing")
def test_complete_base_handoff_passes(self):
result = assess_controller_handoff(
"full report body...\n\n" + self.BASE_HANDOFF)
self.assertEqual(result["verdict"], "complete")
self.assertFalse(result["downgraded"])
def test_missing_base_fields_are_listed(self):
text = "\n".join(
line for line in self.BASE_HANDOFF.splitlines()
if not line.startswith(("- Mutations:", "- Safety:")))
result = assess_controller_handoff(text)
self.assertEqual(result["verdict"], "incomplete")
self.assertTrue(result["downgraded"])
self.assertIn("Mutations", result["missing_fields"])
self.assertIn("Safety", result["missing_fields"])
def test_review_role_requires_review_fields(self):
result = assess_controller_handoff(self.BASE_HANDOFF, role="review")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("Pinned reviewed head", result["missing_fields"])
self.assertIn("Merge result", result["missing_fields"])
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected PR: #999",
"- Reviewer eligibility: passed",
"- Pinned reviewed head: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9",
"- Review decision: approve",
"- Merge result: merged",
"- Linked issue status: closed",
"- Cleanup status: branch deleted",
])
result = assess_controller_handoff(complete, role="review")
self.assertEqual(result["verdict"], "complete")
def test_author_role_requires_author_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Selected issue: #182",
"- Claim/comment status: comment-claimed",
"- PR number opened: #999",
"- No review/merge: confirmed",
])
result = assess_controller_handoff(complete, role="author")
self.assertEqual(result["verdict"], "complete")
result = assess_controller_handoff(self.BASE_HANDOFF, role="author")
self.assertEqual(result["verdict"], "incomplete")
self.assertIn("No review/merge confirmation", result["missing_fields"])
def test_inventory_role_requires_inventory_fields(self):
complete = self.BASE_HANDOFF + "\n" + "\n".join([
"- Repositories checked: Gitea-Tools, mcp-control-plane",
"- Open PR counts: 2 / 0",
"- Selected PR or reason: none eligible (self-authored)",
"- Inventory completeness: complete, no pagination needed",
])
result = assess_controller_handoff(complete, role="inventory")
self.assertEqual(result["verdict"], "complete")
def test_skill_doc_declares_handoff_requirement(self):
# Doc-contract: SKILL.md must keep requiring the exact section and
# naming this validator, or the convention silently rots.
skill = (
__import__("pathlib").Path(__file__).resolve().parent.parent
/ "skills" / "llm-project-workflow" / "SKILL.md"
).read_text(encoding="utf-8")
self.assertIn("## Controller Handoff", skill)
self.assertIn("assess_controller_handoff", skill)
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"])
class TestAuthorReporting(unittest.TestCase):
"""Harness assertions for author reporting and capability/sweep proofs (#183)."""
"""Harness assertions for author reporting and capability/sweep proofs."""
def test_good_capability_proof_passes(self):
resolved = {
"mark_issue": {
"requested_task": "mark_issue",
"required_operation_permission": "gitea.issue.write",
"allowed_in_current_session": True,
}
}
result = assess_capability_proof(resolved)
result = _good_capability_proof()
self.assertTrue(result["proven"])
def test_missing_capability_proof_fails_closed(self):
@@ -649,69 +963,60 @@ class TestAuthorReporting(unittest.TestCase):
self.assertTrue(any("fail closed" in r for r in result["reasons"]))
def test_unresolved_capability_proof_fails_closed(self):
# unknown task resolving to None/unknown requested task
resolved = {
result = assess_capability_proof({
"mark_issue": {
"requested_task": "unknown_task",
"required_operation_permission": None,
"allowed_in_current_session": None,
}
}
result = assess_capability_proof(resolved)
})
self.assertFalse(result["proven"])
self.assertTrue(any("could not be resolved" in r for r in result["reasons"]))
self.assertTrue(
any("could not be resolved" in r for r in result["reasons"])
)
def test_good_secret_sweep_passes(self):
sweep = {
"method": "git diff | grep -iE 'token|secret'",
"scope": "staged diff relative to master",
"clean": True,
}
result = assess_secret_sweep(sweep)
result = _good_secret_sweep()
self.assertTrue(result["proven"])
self.assertEqual(result["verdict"], "strong")
def test_vague_secret_sweep_without_method_is_weak(self):
sweep = {
"method": "",
"scope": "staged diff",
"clean": True,
}
result = assess_secret_sweep(sweep)
result = _good_secret_sweep(method="")
self.assertFalse(result["proven"])
self.assertEqual(result["verdict"], "weak")
self.assertTrue(any("method" in r or "scan" in r for r in result["reasons"]))
self.assertTrue(
any("method" in r or "scan" in r for r in result["reasons"])
)
def test_unconfirmed_secret_sweep_is_invalid(self):
sweep = {
"method": "git diff | grep",
"scope": "staged diff",
"clean": False,
}
result = assess_secret_sweep(sweep)
def test_unconfirmed_secret_sweep_is_weak(self):
result = _good_secret_sweep(clean=False)
self.assertFalse(result["proven"])
self.assertEqual(result["verdict"], "weak")
def test_good_author_pr_report_passes(self):
pr = {
result = assess_author_pr_report({
"pr_number": 185,
"branch": "feat/issue-184-repo-name-disambiguation",
"head_sha": "e2bccbafeeb93124ba068bfb06058d5aa7467cae",
}
result = assess_author_pr_report(pr)
})
self.assertTrue(result["complete"])
def test_incomplete_author_pr_report_fails(self):
pr = {
"pr_number": 0,
"branch": "",
"head_sha": "e2bccba",
}
result = assess_author_pr_report(pr)
def test_author_pr_report_without_head_sha_is_incomplete(self):
result = assess_author_pr_report({
"pr_number": 185,
"branch": "feat/issue-184-repo-name-disambiguation",
"head_sha": "",
})
self.assertFalse(result["complete"])
self.assertTrue(any("head SHA" in r for r in result["reasons"]))
def test_author_pr_report_rejects_short_sha(self):
result = assess_author_pr_report({
"pr_number": 185,
"branch": "feat/issue-184-repo-name-disambiguation",
"head_sha": "e2bccba",
})
self.assertFalse(result["complete"])
self.assertTrue(any("number" in r for r in result["reasons"]))
self.assertTrue(any("branch" in r for r in result["reasons"]))
self.assertTrue(any("SHA" in r for r in result["reasons"]))
if __name__ == "__main__":