File size: 5,041 Bytes
c32102b ef02f17 c32102b ffaeef0 c32102b ef02f17 c32102b ef02f17 c32102b ef02f17 ffaeef0 c32102b ef02f17 c32102b ef02f17 ffaeef0 c32102b | 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 144 145 146 | """
Rhodawk AI — Closed Verification Loop Engine
=============================================
This is the core capability that separates Rhodawk from every other AI CI tool.
Standard tools: AI generates fix → open PR (no idea if fix works)
Rhodawk: AI generates fix → re-run tests → if still failing, retry with
new failure context + what was tried → up to MAX_RETRIES rounds
The loop:
1. Run tests → get failure output
2. Dispatch Aider with failure context + memory-retrieved similar fixes
3. Re-run tests on the modified code
4. If GREEN → gate through adversarial review → open PR
5. If STILL RED → append new failure + what was tried → goto 2
6. After MAX_RETRIES → mark as FAILED, escalate
BUG-002 FIX: Removed hardcoded os.getenv("RHODAWK_REPO_DIR") — repo_dir is now
passed as a parameter to build_initial_prompt() and build_retry_prompt().
BUG-003 FIX: ADVERSARIAL_REJECTION_MULTIPLIER defaults to 2 (not 0) so adversarial
rejections get extra retry budget beyond MAX_RETRIES.
"""
import os
import time
from dataclasses import dataclass, field
from typing import Optional
from language_runtime import RuntimeFactory
MAX_RETRIES = int(os.getenv("RHODAWK_MAX_RETRIES", "5"))
ADVERSARIAL_REJECTION_MULTIPLIER = int(os.getenv("RHODAWK_ADVERSARIAL_REJECTION_MULTIPLIER", "2"))
RETRY_BACKOFF_SECONDS = 5
@dataclass
class VerificationAttempt:
attempt_number: int
prompt_hash: str
aider_exit_code: int
test_exit_code: int
test_output: str
diff_produced: str
timestamp: str = field(default_factory=lambda: time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
@dataclass
class VerificationResult:
success: bool
attempts: list[VerificationAttempt] = field(default_factory=list)
final_diff: str = ""
final_test_output: str = ""
failure_reason: str = ""
total_attempts: int = 0
def build_retry_prompt(
test_path: str,
src_file: str,
branch_name: str,
original_failure: str,
attempt_history: list[VerificationAttempt],
similar_fixes: list[dict],
repo_dir: str = "/data/repo",
) -> str:
"""
Build an increasingly rich prompt for each retry attempt.
Each retry includes:
- The original failure
- What was tried in previous attempts and why it failed
- Retrieved similar successful fixes from memory
"""
sections = []
sections.append(
f"The pytest test '{test_path}' is STILL FAILING. This is attempt "
f"{len(attempt_history) + 1} of {MAX_RETRIES}.\n"
)
sections.append(
f"ORIGINAL FAILURE:\n```\n{original_failure[:2000]}\n```\n"
)
if attempt_history:
sections.append("PREVIOUS ATTEMPTS THAT DID NOT WORK:")
for a in attempt_history:
sections.append(
f"\nAttempt {a.attempt_number}:\n"
f" Test output after fix:\n```\n{a.test_output[:800]}\n```\n"
f" Diff that was applied:\n```diff\n{a.diff_produced[:600]}\n```"
)
sections.append(
"\nDo NOT repeat the same fix approach. Analyze why previous attempts failed "
"and try a fundamentally different strategy.\n"
)
if similar_fixes:
sections.append("\nSIMILAR FIXES FROM MEMORY (from previously healed tests — use as guidance):")
for i, fix in enumerate(similar_fixes[:2], 1):
sections.append(
f"\nSimilar fix {i} (success rate: {fix.get('success_rate', 'unknown')}):\n"
f" Failure pattern: {fix.get('failure_signature', '')[:200]}\n"
f" Fix applied:\n```diff\n{fix.get('fix_diff', '')[:400]}\n```"
)
runtime = RuntimeFactory.for_repo(repo_dir)
sections.append("INSTRUCTIONS:\n" + runtime.get_fix_prompt_instructions(
test_path=test_path,
branch_name=branch_name,
src_hint=src_file,
))
return "\n".join(sections)
def build_initial_prompt(
test_path: str,
src_file: str,
branch_name: str,
failure_output: str,
similar_fixes: list[dict],
repo_dir: str = "/data/repo",
) -> str:
sections = []
sections.append(
f"The pytest test '{test_path}' is failing:\n\n"
f"```\n{failure_output[:3000]}\n```\n"
)
if similar_fixes:
sections.append("RELEVANT FIXES FROM MEMORY (similar past failures that were healed):")
for i, fix in enumerate(similar_fixes[:2], 1):
sections.append(
f"\nSimilar case {i} (success rate: {fix.get('success_rate', 'unknown')}):\n"
f" Failure: {fix.get('failure_signature', '')[:150]}\n"
f" What worked:\n```diff\n{fix.get('fix_diff', '')[:400]}\n```"
)
runtime = RuntimeFactory.for_repo(repo_dir)
sections.append("INSTRUCTIONS:\n" + runtime.get_fix_prompt_instructions(
test_path=test_path,
branch_name=branch_name,
src_hint=src_file,
))
return "\n".join(sections)
|