scripts/directional_recommendation_validator.py
9.7 KB| 1 | #!/usr/bin/env python3 |
| 2 | """directional_recommendation_validator.py — Verify reflect output ends with discipline. |
| 3 | |
| 4 | Stdlib-only. Validates that a reflect-skill output: |
| 5 | |
| 6 | 1. Ends with a directional recommendation: Continue / Pivot / Pause |
| 7 | 2. The recommendation is SPECIFIC (not vague) |
| 8 | 3. Uses flowing prose (no markdown headers or bullet lists in the body) |
| 9 | 4. Cites specific evidence (turn references, specific decision points) |
| 10 | 5. Doesn't include manufactured-problem language without specific evidence |
| 11 | |
| 12 | Outputs PASS / WARN / FAIL with rule-by-rule findings. |
| 13 | |
| 14 | NO LLM CALLS. Pure regex + heuristic detection. |
| 15 | |
| 16 | Usage: |
| 17 | python directional_recommendation_validator.py --output /tmp/reflect_output.txt |
| 18 | python directional_recommendation_validator.py --sample-pass |
| 19 | python directional_recommendation_validator.py --sample-fail |
| 20 | """ |
| 21 | |
| 22 | import argparse |
| 23 | import json |
| 24 | import re |
| 25 | import sys |
| 26 | from pathlib import Path |
| 27 | from typing import Any, Dict, List |
| 28 | |
| 29 | |
| 30 | RECOMMENDATION_PATTERNS = { |
| 31 | "continue": [ |
| 32 | r"\bcontinue\b\.?\s*$", |
| 33 | r"\bcontinue\s+(this|the)\s", |
| 34 | r"\bkeep\s+going\b", |
| 35 | r"\bstay\s+(on|with)\s+this\b", |
| 36 | r"\bproceed\b", |
| 37 | ], |
| 38 | "pivot": [ |
| 39 | r"\bpivot\s+(to|toward|away from)\b", |
| 40 | r"\bchange\s+(direction|course|approach)\b", |
| 41 | r"\bredirect\b", |
| 42 | r"\bswitch\s+to\b", |
| 43 | ], |
| 44 | "pause": [ |
| 45 | r"\bpause\s+(for|to|until)\b", |
| 46 | r"\bstop\s+(to|and)\s+(answer|consider|address)\b", |
| 47 | r"\bhalt\s+(for|to|until)\b", |
| 48 | r"\bwait\s+(to|until|for)\s+(answer|resolve|clarify)\b", |
| 49 | ], |
| 50 | } |
| 51 | |
| 52 | VAGUE_REASSURANCE_PATTERNS = [ |
| 53 | r"\blooks good\b", |
| 54 | r"\bon the right track\b", |
| 55 | r"\bseems fine\b", |
| 56 | r"\bnothing major\b", |
| 57 | r"\bnot too bad\b", |
| 58 | r"\bgenerally okay\b", |
| 59 | ] |
| 60 | |
| 61 | MANUFACTURED_PROBLEM_HEDGES = [ |
| 62 | r"\bmight be worth\b", |
| 63 | r"\bcould consider\b", |
| 64 | r"\bperhaps reconsider\b", |
| 65 | r"\bsome (drift|issues?) (potentially|might)\b", |
| 66 | r"\bworth questioning\b", |
| 67 | r"\bsome assumptions\b", |
| 68 | ] |
| 69 | |
| 70 | HEADER_PATTERNS = [ |
| 71 | r"^#+\s", |
| 72 | r"^\*\*[A-Z][^*]+\*\*\s*$", |
| 73 | r"^[A-Z][A-Z ]+:\s*$", |
| 74 | ] |
| 75 | |
| 76 | BULLET_PATTERNS = [ |
| 77 | r"^\s*[-*+]\s", |
| 78 | r"^\s*\d+\.\s", |
| 79 | ] |
| 80 | |
| 81 | EVIDENCE_PATTERNS = [ |
| 82 | r"\b(turn|message|line)\s+\d+\b", |
| 83 | r"\bat\s+turn\s+\d+\b", |
| 84 | r"\bin\s+(turn|message)\s+\d+\b", |
| 85 | r"\bin\s+the\s+(first|second|third|fourth|fifth|earlier|later)\s+(turn|message|exchange)\b", |
| 86 | r"\boriginal\s+(goal|frame|framing)\b", |
| 87 | r"\bat\s+the\s+(start|beginning|outset)\b", |
| 88 | ] |
| 89 | |
| 90 | |
| 91 | def validate(output: str) -> Dict[str, Any]: |
| 92 | findings: List[Dict[str, str]] = [] |
| 93 | |
| 94 | def add(rule: str, level: str, message: str) -> None: |
| 95 | findings.append({"rule": rule, "level": level, "message": message}) |
| 96 | |
| 97 | # Rule 1: Closing recommendation present |
| 98 | output_lower = output.lower() |
| 99 | last_chunk = output[-400:] |
| 100 | last_chunk_lower = last_chunk.lower() |
| 101 | detected_recommendation = None |
| 102 | for rec_type, patterns in RECOMMENDATION_PATTERNS.items(): |
| 103 | for p in patterns: |
| 104 | if re.search(p, last_chunk_lower, re.IGNORECASE): |
| 105 | detected_recommendation = rec_type |
| 106 | break |
| 107 | if detected_recommendation: |
| 108 | break |
| 109 | |
| 110 | if detected_recommendation: |
| 111 | add("closing-recommendation", "PASS", f"Detected '{detected_recommendation}' recommendation in closing.") |
| 112 | else: |
| 113 | add("closing-recommendation", "FAIL", "No Continue / Pivot / Pause recommendation detected in closing 400 chars.") |
| 114 | |
| 115 | # Rule 2: Vague reassurance |
| 116 | vague_hits = sum(1 for p in VAGUE_REASSURANCE_PATTERNS if re.search(p, output_lower, re.IGNORECASE)) |
| 117 | if vague_hits >= 2: |
| 118 | add("vague-reassurance", "FAIL", f"Output contains {vague_hits} vague-reassurance phrases. Replace with specific reasoning.") |
| 119 | elif vague_hits == 1: |
| 120 | add("vague-reassurance", "WARN", f"Output contains 1 vague phrase. Consider replacing with specific reasoning.") |
| 121 | else: |
| 122 | add("vague-reassurance", "PASS", "No vague-reassurance phrases detected.") |
| 123 | |
| 124 | # Rule 3: Manufactured-problem hedging |
| 125 | hedge_hits = sum(1 for p in MANUFACTURED_PROBLEM_HEDGES if re.search(p, output_lower, re.IGNORECASE)) |
| 126 | if hedge_hits >= 3: |
| 127 | add("manufactured-problems", "WARN", f"{hedge_hits} hedge phrases detected ('might be worth', 'could consider', etc.). Verify each cites specific evidence.") |
| 128 | elif hedge_hits >= 1: |
| 129 | add("manufactured-problems", "PASS", f"{hedge_hits} hedge phrase(s). Verify each cites specific evidence.") |
| 130 | else: |
| 131 | add("manufactured-problems", "PASS", "No manufactured-problem hedge phrases.") |
| 132 | |
| 133 | # Rule 4: Headers detection (should NOT be present) |
| 134 | header_count = 0 |
| 135 | for p in HEADER_PATTERNS: |
| 136 | header_count += len(re.findall(p, output, re.MULTILINE)) |
| 137 | if header_count >= 2: |
| 138 | add("no-headers", "FAIL", f"Output contains {header_count} headers. Reflect output should be flowing prose, no headers.") |
| 139 | elif header_count == 1: |
| 140 | add("no-headers", "WARN", "One header detected. Verify it's part of a quote, not output structure.") |
| 141 | else: |
| 142 | add("no-headers", "PASS", "No headers in output (flowing prose confirmed).") |
| 143 | |
| 144 | # Rule 5: Bullet lists detection (should NOT be present in main body) |
| 145 | bullet_count = 0 |
| 146 | for p in BULLET_PATTERNS: |
| 147 | bullet_count += len(re.findall(p, output, re.MULTILINE)) |
| 148 | if bullet_count >= 3: |
| 149 | add("no-bullets", "FAIL", f"Output contains {bullet_count} bullet-list items. Reflect output should be flowing prose.") |
| 150 | elif bullet_count >= 1: |
| 151 | add("no-bullets", "WARN", f"{bullet_count} bullet items detected. Verify these are part of a recommendation list, not body structure.") |
| 152 | else: |
| 153 | add("no-bullets", "PASS", "No bullet lists in output (flowing prose confirmed).") |
| 154 | |
| 155 | # Rule 6: Specific evidence references |
| 156 | evidence_count = sum(len(re.findall(p, output_lower, re.IGNORECASE)) for p in EVIDENCE_PATTERNS) |
| 157 | if evidence_count >= 3: |
| 158 | add("specific-evidence", "PASS", f"{evidence_count} specific evidence references (turn numbers, original goal, etc.).") |
| 159 | elif evidence_count >= 1: |
| 160 | add("specific-evidence", "WARN", f"Only {evidence_count} specific evidence reference(s). Consider adding more for anchoring.") |
| 161 | else: |
| 162 | add("specific-evidence", "FAIL", "No specific evidence references (turn numbers, original goal anchors). Output is too vague.") |
| 163 | |
| 164 | return finalize(findings) |
| 165 | |
| 166 | |
| 167 | def finalize(findings: List[Dict[str, str]]) -> Dict[str, Any]: |
| 168 | counts = {"PASS": 0, "WARN": 0, "FAIL": 0} |
| 169 | for f in findings: |
| 170 | counts[f["level"]] += 1 |
| 171 | if counts["FAIL"] > 0: |
| 172 | verdict = "FAIL" |
| 173 | elif counts["WARN"] > 0: |
| 174 | verdict = "WARN" |
| 175 | else: |
| 176 | verdict = "PASS" |
| 177 | return {"verdict": verdict, "counts": counts, "findings": findings} |
| 178 | |
| 179 | |
| 180 | SAMPLE_PASS_OUTPUT = """Re-reading from the original goal at turn 3 — clarify the auth flow — the current direction is solid. Three specific reasons. |
| 181 | |
| 182 | First, the auth flow has been narrowed from generic OAuth to a specific Google plus GitHub combination at turn 9, which matches the user base stated at turn 3. The narrowing is principled, not arbitrary. |
| 183 | |
| 184 | Second, the bias check finds no anchoring — passwordless authentication was explicitly considered at turn 11 and rejected for reasons specific to the team's expertise. The rejection cites evidence (team has not deployed magic-link systems before) rather than dismissing the alternative without engagement. |
| 185 | |
| 186 | Third, the original goal at turn 3 connects directly to the current implementation work at turns 14-18. No drift has occurred. Recent decisions (rate limiting at turn 16, session storage at turn 17) are tactical refinements within the original strategic frame, not shifts away from it. |
| 187 | |
| 188 | Continue. |
| 189 | """ |
| 190 | |
| 191 | SAMPLE_FAIL_OUTPUT = """## Reflection |
| 192 | |
| 193 | Some things to consider: |
| 194 | |
| 195 | - The conversation might be drifting |
| 196 | - We could potentially reconsider some assumptions |
| 197 | - Some aspects look good |
| 198 | |
| 199 | Looks good overall! On the right track. |
| 200 | """ |
| 201 | |
| 202 | |
| 203 | def render_human(result: Dict[str, Any]) -> str: |
| 204 | out: List[str] = [] |
| 205 | out.append(f"Reflect-output validation verdict: {result['verdict']}") |
| 206 | c = result["counts"] |
| 207 | out.append(f" PASS: {c['PASS']} WARN: {c['WARN']} FAIL: {c['FAIL']}") |
| 208 | out.append("") |
| 209 | out.append("Findings:") |
| 210 | for f in result["findings"]: |
| 211 | marker = {"PASS": "[ok]", "WARN": "[warn]", "FAIL": "[FAIL]"}[f["level"]] |
| 212 | out.append(f" {marker} {f['rule']}: {f['message']}") |
| 213 | return "\n".join(out) |
| 214 | |
| 215 | |
| 216 | def main(argv: List[str]) -> int: |
| 217 | parser = argparse.ArgumentParser(description=__doc__.split("\n")[0]) |
| 218 | parser.add_argument("--output", help="Path to reflect-skill output text file") |
| 219 | parser.add_argument("--sample-pass", action="store_true", help="Validate embedded honest validation sample") |
| 220 | parser.add_argument("--sample-fail", action="store_true", help="Validate embedded vague-reassurance sample") |
| 221 | parser.add_argument("--output-format", choices=["human", "json"], default="human") |
| 222 | args = parser.parse_args(argv) |
| 223 | |
| 224 | if args.sample_pass: |
| 225 | text = SAMPLE_PASS_OUTPUT |
| 226 | elif args.sample_fail: |
| 227 | text = SAMPLE_FAIL_OUTPUT |
| 228 | elif args.output: |
| 229 | p = Path(args.output) |
| 230 | if not p.exists(): |
| 231 | print(f"error: {args.output} not found", file=sys.stderr) |
| 232 | return 2 |
| 233 | text = p.read_text(encoding="utf-8") |
| 234 | else: |
| 235 | parser.print_help() |
| 236 | return 0 |
| 237 | |
| 238 | result = validate(text) |
| 239 | if args.output_format == "json": |
| 240 | print(json.dumps(result, indent=2)) |
| 241 | else: |
| 242 | print(render_human(result)) |
| 243 | return 0 if result["verdict"] != "FAIL" else 1 |
| 244 | |
| 245 | |
| 246 | if __name__ == "__main__": |
| 247 | sys.exit(main(sys.argv[1:])) |
文件内容是平台检查时读出的文本，只供查看。平台不会运行包里的任何脚本或程序。
