File size: 5,544 Bytes
5bfc214 5849ef1 5bfc214 5849ef1 5bfc214 | 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 | """Strict parser for the joint output contract (spec §5.1).
Never raises: any malformed input yields `ParsedOutput(format_ok=False, ...)`
with `parse_errors` populated. This is required by the Phase 3 gate (fuzz-tested
on >=100 malformed inputs; must never crash) since parse failure feeds the R5
reward guard at training/eval time, not a Python exception.
"""
from __future__ import annotations
import re
from src.contracts.schema import ParsedOutput, TraceStep
_ANALYSIS_RE = re.compile(r"<analysis>(.*?)</analysis>", re.DOTALL)
_LABEL_RE = re.compile(r"<label>\s*(.*?)\s*</label>", re.DOTALL)
_CWE_RE = re.compile(r"<cwe>\s*(.*?)\s*</cwe>", re.DOTALL)
_STATEMENTS_RE = re.compile(r"<statements>\s*(.*?)\s*</statements>", re.DOTALL)
_TRACE_BLOCK_RE = re.compile(r"<trace>(.*?)</trace>", re.DOTALL)
_STEP_RE = re.compile(r"<step>\s*(.*?)\s*</step>", re.DOTALL)
_EXPLANATION_RE = re.compile(r"<explanation>\s*(.*?)\s*</explanation>", re.DOTALL)
_STEP_LINE_RE = re.compile(
r"^\s*line\s+(\d+)\s*:?\s*(.*?)\s*(?:\[\s*(.*?)\s*\])?\s*$", re.IGNORECASE
)
_VALID_LABELS = {"VULNERABLE", "SAFE"}
_CWE_VALUE_RE = re.compile(r"^(CWE-\d+|NONE)$", re.IGNORECASE)
_RANGE_RE = re.compile(r"^\s*(\d+)\s*(?:-\s*(\d+))?\s*$")
def parse_statement_lines(text: str) -> tuple[set[int], list[str]]:
"""Parses 'comma-separated line ranges, e.g. 12-14, 27' -> a set of line numbers."""
lines: set[int] = set()
errors: list[str] = []
text = (text or "").strip()
if not text:
return lines, errors
for chunk in text.split(","):
chunk = chunk.strip()
if not chunk:
continue
m = _RANGE_RE.match(chunk)
if not m:
errors.append(f"unparseable statement range token: {chunk!r}")
continue
start = int(m.group(1))
end = int(m.group(2)) if m.group(2) else start
if end < start:
errors.append(f"inverted range: {chunk!r}")
continue
if end - start > 10_000:
errors.append(f"implausibly large range (rejected): {chunk!r}")
continue
lines.update(range(start, end + 1))
return lines, errors
def _parse_trace(trace_block: str) -> tuple[list[TraceStep], list[str]]:
steps: list[TraceStep] = []
errors: list[str] = []
for step_text in _STEP_RE.findall(trace_block):
m = _STEP_LINE_RE.match(step_text)
if not m:
errors.append(f"unparseable trace step: {step_text!r}")
continue
line_no, code, tag = m.group(1), m.group(2) or "", m.group(3) or ""
try:
steps.append(TraceStep(line=int(line_no), code=code, tag=tag))
except (TypeError, ValueError):
errors.append(f"non-integer line in trace step: {step_text!r}")
return steps, errors
def parse_output(raw: str, contract_version: int = 1) -> ParsedOutput:
"""Parses one model completion against the joint output contract.
Returns format_ok=False (never raises) on any structural violation:
missing <analysis> block, missing/invalid <label>, missing/invalid <cwe>,
or an empty <explanation>. Statement/trace parse errors are logged but do
not by themselves flip format_ok to False, since partial localization or
trace output is still scoreable by R2/R3 on whatever did parse.
"""
if contract_version == 2:
from src.contracts.v2 import parse_v2
return parse_v2(raw)
if contract_version != 1:
return ParsedOutput(format_ok=False, parse_errors=["unsupported contract version"])
raw = raw if isinstance(raw, str) else ""
errors: list[str] = []
analysis_match = _ANALYSIS_RE.search(raw)
if not analysis_match:
return ParsedOutput(format_ok=False, raw=raw, parse_errors=["no <analysis> block found"])
body = analysis_match.group(1)
label_match = _LABEL_RE.search(body)
label = label_match.group(1).strip().upper() if label_match else None
if label not in _VALID_LABELS:
errors.append(f"invalid or missing <label>: {label!r}")
return ParsedOutput(format_ok=False, raw=raw, label=label, parse_errors=errors)
cwe_match = _CWE_RE.search(body)
cwe = cwe_match.group(1).strip().upper() if cwe_match else None
if not cwe or not _CWE_VALUE_RE.match(cwe):
errors.append(f"invalid or missing <cwe>: {cwe!r}")
return ParsedOutput(format_ok=False, raw=raw, label=label, cwe=cwe, parse_errors=errors)
statements_match = _STATEMENTS_RE.search(body)
statement_lines, stmt_errors = parse_statement_lines(
statements_match.group(1) if statements_match else ""
)
errors.extend(stmt_errors)
trace_block_match = _TRACE_BLOCK_RE.search(body)
trace, trace_errors = (
_parse_trace(trace_block_match.group(1)) if trace_block_match else ([], [])
)
errors.extend(trace_errors)
explanation_match = _EXPLANATION_RE.search(body)
explanation = explanation_match.group(1).strip() if explanation_match else ""
if not explanation:
errors.append("missing or empty <explanation>")
return ParsedOutput(
format_ok=False,
raw=raw,
label=label,
cwe=cwe,
statement_lines=statement_lines,
trace=trace,
parse_errors=errors,
)
return ParsedOutput(
format_ok=True,
raw=raw,
label=label,
cwe=cwe,
statement_lines=statement_lines,
trace=trace,
explanation=explanation,
parse_errors=errors,
)
|