feat(loop): port run_loop.sh to Python lib_py.loop with thin wrapper and strict parser (Phase 3)

This commit is contained in:
Godopu committed 2026-09-07 09:04:49 +09:00
1 parent e1203166d9
commit 8cd256cc7a
8 files changed
+1275 -7

No files matched your search

+125
View File
@@ -0,0 +1,125 @@
"""Strict parser for reviewer reports, verdicts, and workflow tags.
Issue #5 Phase 3.
Implements fence-aware parsing so markdown code blocks (e.g. documentation or
examples quoted inside reports) cannot trigger or spoof verdicts, escalations,
or rebuttals.
"""
from __future__ import annotations
import os
import re
from pathlib import Path
from typing import List, Optional
# Fences start with at least 3 backticks or tildes: ``` or ~~~
_FENCE_START_RE = re.compile(r"^[ \t]*(```+|~~~+)")
_VERDICT_RE = re.compile(r"^\[VERDICT:\s*(PASS|NOT PASS)\]\s*$")
def strip_code_fences(text: str) -> str:
"""Remove lines that are inside markdown code blocks (``` or ~~~)."""
clean_lines = []
in_fence = False
fence_char = ""
fence_len = 0
for line in text.splitlines():
match = _FENCE_START_RE.match(line)
if match:
marker = match.group(1)
char = marker[0]
length = len(marker)
if not in_fence:
in_fence = True
fence_char = char
fence_len = length
continue
else:
# Closing fence must match the fence character and have at least the opening length
if char == fence_char and length >= fence_len:
in_fence = False
fence_char = ""
fence_len = 0
continue
if not in_fence:
clean_lines.append(line)
return "\n".join(clean_lines)
def parse_verdict(content: str) -> Optional[str]:
"""Extract verdict ('PASS' or 'NOT PASS') from the last non-blank line outside code fences.
Returns None if no valid verdict line is found or if content is malformed.
"""
stripped = strip_code_fences(content)
lines = [line.strip() for line in stripped.splitlines() if line.strip()]
if not lines:
return None
last_line = lines[-1]
match = _VERDICT_RE.match(last_line)
if match:
return match.group(1)
return None
def has_verdict(file_or_content: str | Path, expected: str) -> bool:
"""Check whether a report file or content ends with [VERDICT: <expected>]."""
content = ""
if isinstance(file_or_content, Path) or (isinstance(file_or_content, str) and os.path.isfile(file_or_content)):
try:
content = Path(file_or_content).read_text(encoding="utf-8", errors="replace")
except OSError:
return False
else:
content = str(file_or_content)
verdict = parse_verdict(content)
return verdict == expected
def has_tag_line(file_or_content: str | Path, tag: str) -> bool:
"""Detect a standalone '[TAG]' line anywhere outside code fences."""
content = ""
if isinstance(file_or_content, Path) or (isinstance(file_or_content, str) and os.path.isfile(file_or_content)):
try:
content = Path(file_or_content).read_text(encoding="utf-8", errors="replace")
except OSError:
return False
else:
content = str(file_or_content)
stripped = strip_code_fences(content)
pattern = re.compile(rf"^\[{re.escape(tag)}\]\s*$", re.MULTILINE)
return bool(pattern.search(stripped))
def extract_tag_values(file_or_content: str | Path, key: str) -> List[str]:
"""Emit the trimmed payload of every standalone '[KEY: value]' line outside code fences.
Preserves appearance order and eliminates duplicates.
"""
content = ""
if isinstance(file_or_content, Path) or (isinstance(file_or_content, str) and os.path.isfile(file_or_content)):
try:
content = Path(file_or_content).read_text(encoding="utf-8", errors="replace")
except OSError:
return []
else:
content = str(file_or_content)
stripped = strip_code_fences(content)
pattern = re.compile(rf"^\[{re.escape(key)}:\s*([^\]]*)\]\s*$", re.MULTILINE)
results = []
seen = set()
for match in pattern.finditer(stripped):
val = match.group(1).strip()
if val and val not in seen:
seen.add(val)
results.append(val)
return results