File size: 5,242 Bytes
8a23fcf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 | """
Rhodawk AI — Conviction Engine (Auto-Merge Gate)
================================================
Evaluates whether a successfully verified fix meets the conviction threshold
for autonomous merge without human review.
Conviction criteria (all must be met):
1. adversarial_confidence >= CONVICTION_CONFIDENCE_MIN (default 0.92)
2. adversarial_verdict == "APPROVE" (no conditional)
3. consensus_fraction >= CONVICTION_CONSENSUS_MIN (default 0.85 — 3/3 models agree)
4. Memory engine found a semantically identical past fix that was human-merged
(similarity >= CONVICTION_MEMORY_MIN, default 0.85)
5. test_attempts == 1 (fixed on first try — indicates clean, well-understood fix)
6. SAST findings == 0 (zero informational findings on the diff)
7. No new packages introduced in the diff
When all criteria pass, auto_merge() is called which uses the GitHub API to
merge the PR directly (no human required).
Enable with: RHODAWK_AUTO_MERGE=true
"""
import os
import time
import requests
CONVICTION_CONFIDENCE_MIN = float(os.getenv("RHODAWK_CONVICTION_CONFIDENCE", "0.92"))
CONVICTION_CONSENSUS_MIN = float(os.getenv("RHODAWK_CONVICTION_CONSENSUS", "0.85"))
CONVICTION_MEMORY_MIN = float(os.getenv("RHODAWK_CONVICTION_MEMORY_SIM", "0.85"))
AUTO_MERGE_ENABLED = os.getenv("RHODAWK_AUTO_MERGE", "false").lower() == "true"
def evaluate_conviction(
adversarial_review: dict,
similar_fixes: list[dict],
test_attempts: int,
sast_findings_count: int,
new_packages: list[str],
) -> tuple[bool, str]:
"""
Returns (should_auto_merge, reason_string).
All criteria must pass for auto-merge to be approved.
"""
if not AUTO_MERGE_ENABLED:
return False, "auto-merge disabled (RHODAWK_AUTO_MERGE != true)"
verdict = adversarial_review.get("verdict", "CONDITIONAL")
confidence = float(adversarial_review.get("confidence", 0.0))
consensus_fraction = float(adversarial_review.get("consensus_fraction", 0.0))
checks: list[tuple[bool, str]] = [
(verdict == "APPROVE",
f"adversarial_verdict must be APPROVE, got {verdict}"),
(confidence >= CONVICTION_CONFIDENCE_MIN,
f"adversarial_confidence {confidence:.3f} < {CONVICTION_CONFIDENCE_MIN}"),
(consensus_fraction >= CONVICTION_CONSENSUS_MIN,
f"consensus_fraction {consensus_fraction:.3f} < {CONVICTION_CONSENSUS_MIN}"),
(test_attempts == 1,
f"fixed in {test_attempts} attempt(s), require 1"),
(sast_findings_count == 0,
f"SAST has {sast_findings_count} finding(s), require 0"),
(not new_packages,
f"diff introduces new packages: {new_packages}"),
]
memory_check_passed = False
best_memory_sim = 0.0
for fix in similar_fixes:
sim = float(fix.get("similarity", 0.0))
if sim >= CONVICTION_MEMORY_MIN:
memory_check_passed = True
best_memory_sim = sim
break
checks.append((
memory_check_passed,
f"no human-merged memory match with similarity >= {CONVICTION_MEMORY_MIN} "
f"(best found: {best_memory_sim:.3f})"
))
failed = [(passed, reason) for passed, reason in checks if not passed]
if failed:
reasons = "; ".join(r for _, r in failed)
return False, f"conviction not met: {reasons}"
return True, (
f"all {len(checks)} conviction criteria passed "
f"(confidence={confidence:.3f}, consensus={consensus_fraction:.3f}, "
f"memory_sim={best_memory_sim:.3f})"
)
def auto_merge_pr(
repo: str,
pr_url: str,
token: str,
merge_method: str = "squash",
) -> tuple[bool, str]:
"""
Merge a PR via GitHub API.
merge_method: "merge" | "squash" | "rebase"
Returns (success, message).
"""
if not pr_url or "github.com" not in pr_url:
return False, f"invalid PR URL: {pr_url}"
try:
parts = pr_url.rstrip("/").split("/")
pr_number = int(parts[-1])
owner = parts[-4]
repo_name = parts[-3]
except (IndexError, ValueError) as e:
return False, f"could not parse PR URL {pr_url}: {e}"
headers = {
"Authorization": f"Bearer {token}",
"Accept": "application/vnd.github+json",
"X-GitHub-API-Version": "2022-11-28",
}
resp = requests.put(
f"https://api.github.com/repos/{owner}/{repo_name}/pulls/{pr_number}/merge",
headers=headers,
json={
"merge_method": merge_method,
"commit_title": f"[Rhodawk] Autonomous merge — conviction threshold met",
"commit_message": (
"This PR was autonomously merged by Rhodawk AI after meeting all "
"conviction criteria:\n"
"- Adversarial consensus review: APPROVED (3/3 models)\n"
"- Test verification: GREEN (1 attempt)\n"
"- SAST gate: CLEAN\n"
"- Memory match: CONFIRMED (prior human-merged fix)\n"
),
},
timeout=30,
)
if resp.status_code in (200, 201):
return True, f"auto-merged PR #{pr_number} via {merge_method}"
return False, f"GitHub merge API returned {resp.status_code}: {resp.text[:200]}"
|