Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dfcbca7355 | ||
|
|
9ea2707289 |
@@ -1,217 +0,0 @@
|
||||
"""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", []))
|
||||
),
|
||||
}
|
||||
+6
-6
@@ -2307,26 +2307,26 @@ _PROJECT_SKILLS = {
|
||||
"committed.",
|
||||
],
|
||||
},
|
||||
"jenkins-mcp": {
|
||||
"jenkins-readonly": {
|
||||
"description": "Read-only Jenkins CI inspection (jobs, builds, "
|
||||
"logs).",
|
||||
"logs). Actual server name: jenkins-mcp (see mcp-control-plane).",
|
||||
"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. 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).",
|
||||
"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).",
|
||||
"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-mcp": {
|
||||
"description": "Read-only GlitchTip error/event inspection.",
|
||||
"glitchtip-readonly": {
|
||||
"description": "Read-only GlitchTip error/event inspection. Actual server name: glitchtip-mcp (see mcp-control-plane).",
|
||||
"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. 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.",
|
||||
"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.",
|
||||
"steps": [
|
||||
"Confirm a GlitchTip MCP server is connected (glitchtip-mcp); if not, report "
|
||||
"SKIPPED.",
|
||||
|
||||
+65
-389
@@ -202,13 +202,15 @@ 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``.
|
||||
``justification``), ``canonical_command``, ``deviation_justification``,
|
||||
``is_stdout_capture_fix``, ``normal_pytest_summary``.
|
||||
|
||||
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).
|
||||
ignored paths, unexplained deviation from the canonical command, or
|
||||
missing normal summary on stdout capture fix).
|
||||
- 'strong' — full evidence.
|
||||
"""
|
||||
reasons = []
|
||||
@@ -252,6 +254,14 @@ 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}
|
||||
|
||||
@@ -330,99 +340,10 @@ 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, role_boundary=None,
|
||||
controller_handoff=None, capability_proof=None,
|
||||
sweep_proof=None):
|
||||
issue_status_verified, 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
|
||||
@@ -438,22 +359,6 @@ 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:
|
||||
@@ -473,26 +378,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")
|
||||
if not handoff["complete"]:
|
||||
downgrade_reasons.append("Controller Handoff missing or incomplete")
|
||||
|
||||
handoff = assess_controller_handoff(controller_handoff or "")
|
||||
if not handoff.get("present"):
|
||||
downgrade_reasons.append("Controller Handoff section missing or incomplete")
|
||||
downgrade_reasons.extend(handoff.get("reasons", []))
|
||||
if not capability_proof.get("proven"):
|
||||
downgrade_reasons.append("exact mutation capability proof missing")
|
||||
|
||||
if capability_proof and not capability_proof.get("proven"):
|
||||
downgrade_reasons.append("capability proof verification failed")
|
||||
downgrade_reasons.extend(capability_proof.get("reasons", []))
|
||||
if not sweep_proof.get("proven"):
|
||||
downgrade_reasons.append("exact secret/provenance sweep proof missing")
|
||||
|
||||
if sweep_proof and not sweep_proof.get("proven"):
|
||||
downgrade_reasons.append("secret/provenance sweep proof verification failed")
|
||||
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"
|
||||
)
|
||||
@@ -503,7 +408,6 @@ 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"
|
||||
@@ -520,7 +424,6 @@ 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":
|
||||
@@ -529,46 +432,36 @@ 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["present"],
|
||||
"controller_handoff_complete": handoff["complete"],
|
||||
"capability_proof": bool(capability_proof.get("proven")),
|
||||
"secret_sweep_proof": bool(sweep_proof.get("proven")),
|
||||
"controller_handoff_present": handoff.get("present", False),
|
||||
}
|
||||
|
||||
|
||||
def _normalize_controller_handoff_proof(controller_handoff):
|
||||
"""Normalize old boolean handoff proof and rich #182 handoff verdicts."""
|
||||
if controller_handoff is None:
|
||||
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:
|
||||
return {
|
||||
"present": False,
|
||||
"complete": False,
|
||||
"reasons": ["Controller Handoff proof not provided"],
|
||||
"reasons": [
|
||||
"final report missing exactly-titled 'Controller Handoff' section"
|
||||
],
|
||||
}
|
||||
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 []),
|
||||
}
|
||||
return {"present": True, "reasons": []}
|
||||
|
||||
|
||||
def assess_capability_proof(resolved_capabilities: dict) -> dict:
|
||||
"""Every mutation must have exact capability proof (#183)."""
|
||||
"""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.
|
||||
"""
|
||||
reasons = []
|
||||
if not resolved_capabilities:
|
||||
return {
|
||||
@@ -579,22 +472,20 @@ 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 (
|
||||
"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}
|
||||
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}
|
||||
|
||||
|
||||
def assess_secret_sweep(sweep_report: dict) -> dict:
|
||||
"""Secret/provenance sweeps must state exact command/method and scope."""
|
||||
"""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.
|
||||
"""
|
||||
if not sweep_report:
|
||||
return {
|
||||
"proven": False,
|
||||
@@ -605,24 +496,26 @@ 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": not reasons and sweep_report.get("clean") is True,
|
||||
"proven": proven,
|
||||
"verdict": verdict,
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
def assess_author_pr_report(pr_report: dict) -> dict:
|
||||
"""Author PR reports must include PR number, branch, and exact head SHA."""
|
||||
"""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).
|
||||
"""
|
||||
if not pr_report:
|
||||
return {
|
||||
"complete": False,
|
||||
@@ -642,225 +535,8 @@ 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": not reasons,
|
||||
"complete": complete,
|
||||
"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,
|
||||
}
|
||||
|
||||
@@ -150,33 +150,12 @@ 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).
|
||||
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:
|
||||
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:
|
||||
- why the PR merge path could not be used
|
||||
- exact commits pushed
|
||||
- PR metadata state
|
||||
@@ -217,27 +196,19 @@ 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. **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
|
||||
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
|
||||
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`).
|
||||
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`):
|
||||
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`):
|
||||
identity eligible; PR author different from reviewer; session
|
||||
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.
|
||||
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.
|
||||
|
||||
## G. Merge / cleanup workflow
|
||||
|
||||
@@ -327,43 +298,22 @@ 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, canonical field set per issue #182):
|
||||
Compact format (default):
|
||||
|
||||
```md
|
||||
## Controller Handoff
|
||||
|
||||
- Task:
|
||||
- Repo:
|
||||
- Role:
|
||||
- Identity:
|
||||
- Issue/PR:
|
||||
- Branch/SHA:
|
||||
- Files changed:
|
||||
- Repo/state:
|
||||
- Issues/PRs:
|
||||
- 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,11 +35,5 @@ 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: 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).
|
||||
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.
|
||||
```
|
||||
|
||||
@@ -23,10 +23,6 @@ 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:
|
||||
@@ -43,10 +39,6 @@ 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
|
||||
@@ -73,10 +65,7 @@ Steps:
|
||||
- MCP-Profile: <profile name>
|
||||
- Eligibility: passed/failed
|
||||
|
||||
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).
|
||||
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.
|
||||
```
|
||||
|
||||
@@ -26,19 +26,8 @@ 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. 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.
|
||||
and scan the diff for secrets.
|
||||
8. Commit (issue-linked message), push the branch, 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):
|
||||
@@ -51,12 +40,9 @@ Steps:
|
||||
- Branch: <branch>
|
||||
- Worktree: <worktree path>
|
||||
- Self-review allowed: no
|
||||
10. Stop before review/merge — you are the author.
|
||||
9. 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
|
||||
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).
|
||||
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".
|
||||
```
|
||||
|
||||
@@ -1,234 +0,0 @@
|
||||
"""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()
|
||||
@@ -46,8 +46,8 @@ EXPECTED_SKILLS = [
|
||||
"gitea-resolve-task-capability",
|
||||
"profile-switching",
|
||||
"redaction-security-review",
|
||||
"jenkins-mcp",
|
||||
"glitchtip-mcp",
|
||||
"jenkins-readonly",
|
||||
"glitchtip-readonly",
|
||||
"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-mcp"]["status"], "available")
|
||||
self.assertNotEqual(by_name["glitchtip-mcp"]["status"], "available")
|
||||
self.assertNotEqual(by_name["jenkins-readonly"]["status"], "available")
|
||||
self.assertNotEqual(by_name["glitchtip-readonly"]["status"], "available")
|
||||
|
||||
def test_no_urls_in_registry(self):
|
||||
with patch.dict(os.environ, AUTHOR_ENV, clear=True):
|
||||
@@ -245,17 +245,6 @@ 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
|
||||
|
||||
+83
-388
@@ -25,12 +25,10 @@ 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,
|
||||
)
|
||||
@@ -104,64 +102,6 @@ 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."""
|
||||
|
||||
@@ -378,6 +318,21 @@ 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."""
|
||||
@@ -431,74 +386,6 @@ 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"."""
|
||||
@@ -512,10 +399,7 @@ class TestFinalReport(unittest.TestCase):
|
||||
"identity_eligible": True,
|
||||
"merge_performed": False,
|
||||
"issue_status_verified": True,
|
||||
"role_boundary": _good_role_boundary(),
|
||||
"controller_handoff": _good_handoff(),
|
||||
"capability_proof": _good_capability_proof(),
|
||||
"sweep_proof": _good_secret_sweep(),
|
||||
"controller_handoff": "Controller Handoff\n- Task: test",
|
||||
}
|
||||
kwargs.update(overrides)
|
||||
return build_final_report(**kwargs)
|
||||
@@ -528,7 +412,6 @@ 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"])
|
||||
@@ -601,56 +484,21 @@ 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"])
|
||||
|
||||
def test_failed_capability_proof_downgrades(self):
|
||||
report = self._report(
|
||||
capability_proof={
|
||||
"proven": False,
|
||||
"reasons": ["task mark_issue not allowed"],
|
||||
}
|
||||
)
|
||||
cap_proof = {
|
||||
"proven": False,
|
||||
"reasons": ["task mark_issue not allowed"]
|
||||
}
|
||||
report = self._report(capability_proof=cap_proof)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertTrue(
|
||||
any("capability" in r for r in report["downgrade_reasons"])
|
||||
)
|
||||
self.assertTrue(any("capability" in r for r in report["downgrade_reasons"]))
|
||||
|
||||
def test_failed_sweep_proof_downgrades(self):
|
||||
report = self._report(
|
||||
sweep_proof={
|
||||
"proven": False,
|
||||
"reasons": ["method missing"],
|
||||
}
|
||||
)
|
||||
sweep_proof = {
|
||||
"proven": False,
|
||||
"reasons": ["method missing"]
|
||||
}
|
||||
report = self._report(sweep_proof=sweep_proof)
|
||||
self.assertNotEqual(report["grade"], "A")
|
||||
self.assertTrue(any("sweep" in r for r in report["downgrade_reasons"]))
|
||||
|
||||
@@ -767,194 +615,32 @@ class TestRepoNameDisambiguation(unittest.TestCase):
|
||||
|
||||
|
||||
class TestControllerHandoff(unittest.TestCase):
|
||||
"""Issue #182: final reports must end with a Controller Handoff."""
|
||||
"""#183: every final report must contain the Controller Handoff section."""
|
||||
|
||||
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_with_handoff_passes(self):
|
||||
report = "some details\n\nController Handoff\n- Task: foo"
|
||||
result = assess_controller_handoff(report)
|
||||
self.assertTrue(result["present"])
|
||||
|
||||
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"])
|
||||
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"]))
|
||||
|
||||
|
||||
class TestAuthorReporting(unittest.TestCase):
|
||||
"""Harness assertions for author reporting and capability/sweep proofs."""
|
||||
"""Harness assertions for author reporting and capability/sweep proofs (#183)."""
|
||||
|
||||
def test_good_capability_proof_passes(self):
|
||||
result = _good_capability_proof()
|
||||
resolved = {
|
||||
"mark_issue": {
|
||||
"requested_task": "mark_issue",
|
||||
"required_operation_permission": "gitea.issue.write",
|
||||
"allowed_in_current_session": True,
|
||||
}
|
||||
}
|
||||
result = assess_capability_proof(resolved)
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_missing_capability_proof_fails_closed(self):
|
||||
@@ -963,60 +649,69 @@ class TestAuthorReporting(unittest.TestCase):
|
||||
self.assertTrue(any("fail closed" in r for r in result["reasons"]))
|
||||
|
||||
def test_unresolved_capability_proof_fails_closed(self):
|
||||
result = assess_capability_proof({
|
||||
# unknown task resolving to None/unknown requested task
|
||||
resolved = {
|
||||
"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):
|
||||
result = _good_secret_sweep()
|
||||
sweep = {
|
||||
"method": "git diff | grep -iE 'token|secret'",
|
||||
"scope": "staged diff relative to master",
|
||||
"clean": True,
|
||||
}
|
||||
result = assess_secret_sweep(sweep)
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertEqual(result["verdict"], "strong")
|
||||
|
||||
def test_vague_secret_sweep_without_method_is_weak(self):
|
||||
result = _good_secret_sweep(method="")
|
||||
sweep = {
|
||||
"method": "",
|
||||
"scope": "staged diff",
|
||||
"clean": True,
|
||||
}
|
||||
result = assess_secret_sweep(sweep)
|
||||
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_weak(self):
|
||||
result = _good_secret_sweep(clean=False)
|
||||
def test_unconfirmed_secret_sweep_is_invalid(self):
|
||||
sweep = {
|
||||
"method": "git diff | grep",
|
||||
"scope": "staged diff",
|
||||
"clean": False,
|
||||
}
|
||||
result = assess_secret_sweep(sweep)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertEqual(result["verdict"], "weak")
|
||||
|
||||
def test_good_author_pr_report_passes(self):
|
||||
result = assess_author_pr_report({
|
||||
pr = {
|
||||
"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_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",
|
||||
def test_incomplete_author_pr_report_fails(self):
|
||||
pr = {
|
||||
"pr_number": 0,
|
||||
"branch": "",
|
||||
"head_sha": "e2bccba",
|
||||
})
|
||||
}
|
||||
result = assess_author_pr_report(pr)
|
||||
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__":
|
||||
|
||||
Reference in New Issue
Block a user