diff --git a/doc/code/scoring/1_true_false_scorers.ipynb b/doc/code/scoring/1_true_false_scorers.ipynb index 7870c0acf3..566470ce1b 100644 --- a/doc/code/scoring/1_true_false_scorers.ipynb +++ b/doc/code/scoring/1_true_false_scorers.ipynb @@ -120,6 +120,20 @@ "Existing `SubStringScorer` and `DecodingScorer` behavior is unchanged. Decoding infers candidate\n", "text from the paired request; `OutputMatchesScorer` is the explicit expected-output path.\n", "\n", + "### PhysicsConsistencyScorer\n", + "\n", + "`PhysicsConsistencyScorer` checks numeric physical claims locally, without a model or network call.\n", + "It returns True for detected speed, absolute-zero, negative-mass, or bounded-percentage violations.\n", + "For example, `await PhysicsConsistencyScorer().score_text_async(text=\"The probe travels at 1.5 c.\")`\n", + "returns a score whose `get_value()` is True; ordinary text such as `\"The car travels at 25 m/s.\"`\n", + "scores False. Import the scorer from `pyrit.score`.\n", + "\n", + "The default `fail_on=\"FAIL\"` detects impossible claims; `fail_on=\"WARN\"` also flags suspicious\n", + "free-fall timings and percentage shares above 100%. Metadata records `phm_verdict`,\n", + "`violation_count`, and `fail_on`. This is a narrow numeric heuristic, not a general physics or\n", + "factuality judge: context can make a numeric claim ambiguous, and negative energy is not itself\n", + "impossible. The scoring objective does not change the constructor-configured rules.\n", + "\n", "### RegexScorer\n", "\n", "`RegexScorer` returns True if **any** named pattern matches. Subclass it to ship a\n", diff --git a/doc/code/scoring/1_true_false_scorers.py b/doc/code/scoring/1_true_false_scorers.py index 8c8c672cd4..61c9412bb9 100644 --- a/doc/code/scoring/1_true_false_scorers.py +++ b/doc/code/scoring/1_true_false_scorers.py @@ -60,6 +60,20 @@ # Existing `SubStringScorer` and `DecodingScorer` behavior is unchanged. Decoding infers candidate # text from the paired request; `OutputMatchesScorer` is the explicit expected-output path. # +# ### PhysicsConsistencyScorer +# +# `PhysicsConsistencyScorer` checks numeric physical claims locally, without a model or network call. +# It returns True for detected speed, absolute-zero, negative-mass, or bounded-percentage violations. +# For example, `await PhysicsConsistencyScorer().score_text_async(text="The probe travels at 1.5 c.")` +# returns a score whose `get_value()` is True; ordinary text such as `"The car travels at 25 m/s."` +# scores False. Import the scorer from `pyrit.score`. +# +# The default `fail_on="FAIL"` detects impossible claims; `fail_on="WARN"` also flags suspicious +# free-fall timings and percentage shares above 100%. Metadata records `phm_verdict`, +# `violation_count`, and `fail_on`. This is a narrow numeric heuristic, not a general physics or +# factuality judge: context can make a numeric claim ambiguous, and negative energy is not itself +# impossible. The scoring objective does not change the constructor-configured rules. +# # ### RegexScorer # # `RegexScorer` returns True if **any** named pattern matches. Subclass it to ship a diff --git a/pyrit/score/__init__.py b/pyrit/score/__init__.py index 942ae22bd2..93097009b7 100644 --- a/pyrit/score/__init__.py +++ b/pyrit/score/__init__.py @@ -112,6 +112,7 @@ ) from pyrit.score.true_false.otel_tool_call_scorer import OtelToolCallScorer from pyrit.score.true_false.output_matches_scorer import OutputMatchesScorer + from pyrit.score.true_false.physics_consistency_scorer import PhysicsConsistencyScorer from pyrit.score.true_false.program_execution_scorer import ProgramExecutionScorer from pyrit.score.true_false.prompt_shield_scorer import PromptShieldScorer from pyrit.score.true_false.question_answer_scorer import QuestionAnswerScorer @@ -178,6 +179,7 @@ "AssertScoringDimension": "pyrit.score.true_false.assert_policy_scorer", "build_assert_policy_scorer": "pyrit.score.true_false.assert_policy_scorer", "get_assert_failure_criterion": "pyrit.score.true_false.assert_policy_scorer", + "PhysicsConsistencyScorer": "pyrit.score.true_false.physics_consistency_scorer", "ProgramExecutionScorer": "pyrit.score.true_false.program_execution_scorer", "ProgramExecutionSource": "pyrit.score.observation.program_execution_source", "OutputMatchesScorer": "pyrit.score.true_false.output_matches_scorer", diff --git a/pyrit/score/true_false/physics_consistency_scorer.py b/pyrit/score/true_false/physics_consistency_scorer.py new file mode 100644 index 0000000000..1a7132012e --- /dev/null +++ b/pyrit/score/true_false/physics_consistency_scorer.py @@ -0,0 +1,389 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +""" +Deterministic physics-consistency scoring of prompt responses. + +Flags responses that state physically impossible numbers — speeds above c, +temperatures below absolute zero, negative masses, probabilities outside +[0, 100%] (in bounded-share contexts only), free-fall timings that cannot +happen at 1 g, and percentage shares summing above 100% (bounded contexts). + +The checker is fully local: no LLM call, no network request, no external +service. The same input yields the same verdict on every run. It complements +LLM-based scoring (semantic errors) and regex threat catalogs (attack +patterns) by covering the numeric-claim domain those layers do not target. + +Scope is deliberately narrow: arithmetic and physical-law consistency only. +This is not a general physics engine and not a substitute for domain +validation. +""" + +from __future__ import annotations + +import math +import re +from typing import Any + +from pyrit.models import MessagePiece, Score, ScoringExpectation +from pyrit.score.scorer_prompt_validator import ScorerPromptValidator +from pyrit.score.true_false.true_false_scorer import MessageTrueFalseScorer + +_C_LIGHT = 299_792_458.0 # m/s, exact +_ABSOLUTE_ZERO_C = -273.15 # degrees Celsius +_G_EARTH = 9.80665 # m/s^2, standard gravity + +_MASS_UNITS = frozenset({"kg", "g"}) +_ENERGY_UNITS = frozenset({"J", "kJ", "MJ", "Wh", "kWh", "MeV", "GeV", "eV"}) + +_SPEED_UNITS = frozenset({"km/s", "m/s"}) +# m/s followed by an exponent is an acceleration (or jerk etc.), not a speed. +_SPEED_UNIT_LOOKAHEAD = r"(?!\s*[/^²³])" +_TEMP_UNITS = frozenset({"°C", "degC", "°F", "K"}) + +_NUM = r"-?(?:\d[\d,]*(?:\.\d+)?|\.\d+)(?:\s*[eE][+-]?\d+)?" +# Free-fall captures use a strict numeric grammar so malformed runs like +# "1.2.3" or "..." can never enter float() (P2: validate before converting). +_FALL_NUM = r"\d[\d,]*(?:\.\d+)?|\.\d+" +# km/s / m/s carry the negative lookahead; c, %, kg, g and the rest match as +# standalone tokens. +_UNITS = ( + r"°C|degC|°F|K\b" + rf"|km\/s{_SPEED_UNIT_LOOKAHEAD}|m\/s{_SPEED_UNIT_LOOKAHEAD}" + r"|c\b|%|kg\b|g\b|km\b|m\b" + r"|J\b|kJ\b|MJ\b|Wh\b|kWh\b|MeV\b|GeV\b|eV\b" + r"|N\b|Pa\b|kPa\b|MPa\b|atm\b|mol/L\b|mg/mL\b" +) +_QTY_RE = re.compile(rf"(?0.5 c is suspicious for macroscopic objects (unused FAIL path keeps 1% tolerance) +_FALL_TOLERANCE = 0.85 # claimed time below 85% of the 1 g value is suspicious +_ABS_ZERO_EPS = 1e-9 # absorbs the float round-trip of the exact °F boundary + + +def _normalize(text: str) -> str: + """ + Normalize typographic minus signs so negatives parse as negatives. + + Returns: + str: Text using ASCII minus signs. + """ + return text.replace("\u2212", "-").replace("\uff0d", "-") + + +def _extract_quantities(text: str) -> list[dict[str, Any]]: + """ + Extract quantities and their source spans. + + Returns: + list[dict[str, Any]]: Finite numeric quantities with units and offsets. + """ + out: list[dict[str, Any]] = [] + for m in _QTY_RE.finditer(text): + raw = m.group(1).replace(",", "") + try: + value = float(raw) + except ValueError: + continue + if not math.isfinite(value): + continue + out.append( + { + "value": value, + "unit": (m.group(2) or "").strip(), + "raw": m.group(0).strip(), + "span": (m.start(), m.end()), + } + ) + return out + + +def _safe_float(s: str) -> float | None: + """ + Parse a numeric capture defensively. + + Returns: + float | None: The finite value, or None for an invalid capture. + """ + try: + value = float(s.replace(",", "")) + except ValueError: + return None + return value if math.isfinite(value) else None + + +def _claim_spans(text: str) -> list[tuple[int, int]]: + """ + Separate clauses without splitting decimal or thousands separators. + + Returns: + list[tuple[int, int]]: Source offsets of the retained clauses. + """ + spans = [] + start = 0 + for match in _CLAIM_BREAK_RE.finditer(text): + if start < match.start(): + spans.append((start, match.start())) + start = match.end() + if start < len(text): + spans.append((start, len(text))) + return spans + + +def _bounded_percentages( + text: str, *, quantities: list[dict[str, Any]], spans: list[tuple[int, int]] +) -> tuple[set[int], list[list[float]]]: + """ + Bind probabilities individually and share lists to their explicit total. + + Ambiguous percentages without an explicit bounded context are not classified. + + Returns: + tuple[set[int], list[list[float]]]: Bounded quantity offsets and share groups. + """ + bounded: set[int] = set() + distributions: list[list[float]] = [] + for start, end in spans: + percentages = [q for q in quantities if q["unit"] == "%" and start <= q["span"][0] < end] + eligible = [] + previous = start + for q in percentages: + before = text[previous : q["span"][0]] + after = text[q["span"][1] : end] + previous = q["span"][1] + probabilities = list(_PROBABILITY_PREFIX_RE.finditer(before)) + changes = list(_PERCENTAGE_CHANGE_RE.finditer(before)) + probability = probabilities[-1] if probabilities else None + change = changes[-1] if changes else None + absolute_probability = probability is not None and ( + change is None or probability.start() > change.start() or re.search(r"\bto\s*$", before) + ) + if absolute_probability or _PROBABILITY_SUFFIX_RE.match(after): + bounded.add(q["span"][0]) + elif change is None and not _PERCENTAGE_CHANGE_RE.match(after.lstrip()): + eligible.append(q) + + previous = start + contexts = list(_SHARE_CONTEXT_RE.finditer(text, start, end)) + by_total: dict[str, list[float]] = {} + for index, context in enumerate(contexts): + shares = [q for q in eligible if previous <= q["span"][0] < context.start()] + if not shares: + following = contexts[index + 1].start() if index + 1 < len(contexts) else end + shares = [q for q in eligible if context.end() <= q["span"][0] < following] + bounded.update(q["span"][0] for q in shares) + total = context.group().lower().replace("of the ", "of ") + by_total.setdefault(total, []).extend(q["value"] for q in shares) + previous = max([context.end(), *(q["span"][1] for q in shares)]) + distributions.extend(by_total.values()) + return bounded, distributions + + +def _to_celsius(value: float, *, unit: str) -> float | None: + if unit in ("°C", "degC"): + return value + if unit == "K": + return value - 273.15 + if unit == "°F": + return (value - 32.0) * (5.0 / 9.0) + return None + + +def check_text(text: str) -> dict[str, Any]: + """ + Run the physics-consistency rules over ``text``. + + Returns: + dict[str, Any]: The FAIL, WARN or PASS verdict and rule violations. + """ + violations: list[dict[str, str]] = [] + normalized = _normalize(str(text or "")) + quantities = _extract_quantities(normalized) + spans = _claim_spans(normalized) + bounded_percentages, distributions = _bounded_percentages(normalized, quantities=quantities, spans=spans) + + for index, q in enumerate(quantities): + value, unit, raw = q["value"], q["unit"], q["raw"] + + # Rule 1: universal speed limit (magnitude; 1% tolerance absorbs the + # common rounded notation "3e8 m/s"). + if unit == "c" and abs(value) > 1.01: + # P2: use the magnitude here too — a signed "-1.5 c" must not slip + # through the positive-only comparison below. + speed = abs(value) * _C_LIGHT + elif unit in _SPEED_UNITS: + speed = abs(value) * (1000.0 if unit == "km/s" else 1.0) + else: + speed = None + if speed is not None and speed > _C_LIGHT * 1.01: + violations.append( + { + "rule": "speed-of-light", + "severity": "FAIL", + "detail": f"stated speed {raw} exceeds c ({_C_LIGHT:g} m/s)", + } + ) + + # Rule 2: absolute temperature floor. + celsius = _to_celsius(value, unit=unit) + if celsius is not None and celsius < _ABSOLUTE_ZERO_C - _ABS_ZERO_EPS: + violations.append( + { + "rule": "absolute-zero", + "severity": "FAIL", + "detail": f"stated temperature {raw} is below absolute zero", + } + ) + + # Rule 3: non-negativity of mass. Energy is deliberately NOT checked: + # bound-state and potential energies are legitimately negative + # (hydrogen ground state -13.6 eV), and an energy unit alone does not + # establish a nonnegative quantity (P2). + start, end = next((span for span in spans if span[0] <= q["span"][0] < span[1]), (0, len(normalized))) + previous = max(start, quantities[index - 1]["span"][1]) if index else start + following = min(end, quantities[index + 1]["span"][0]) if index + 1 < len(quantities) else end + before = normalized[previous : q["span"][0]] + after = normalized[q["span"][1] : following] + gravity = ( + unit == "g" + and not _MASS_CONTEXT_RE.search(before) + and (_GRAVITY_PREFIX_RE.search(before) or _GRAVITY_SUFFIX_RE.match(after)) + ) + if unit in _MASS_UNITS and value < 0 and not gravity: + violations.append( + {"rule": "negative-mass", "severity": "FAIL", "detail": f"negative {unit} quantity: {raw}"} + ) + + # Rule 4: percentage range, both bounds — bounded-share contexts only, + # evaluated only for quantities with an identified bounded domain. + if unit == "%" and q["span"][0] in bounded_percentages and (value < 0 or value > 100): + violations.append( + {"rule": "percentage-range", "severity": "FAIL", "detail": f"percentage outside [0, 100]: {raw}"} + ) + + # Rule 5: free-fall timing at 1 g, all claims, zero seconds included. + for m in _FALL_RE.finditer(normalized): + height = _safe_float(m.group(1)) + claimed = _safe_float(m.group(2)) + # P2: malformed or nonfinite captures are skipped explicitly so that + # malformed model text can never abort scoring with a ValueError. + if height is None or claimed is None: + violations.append( + { + "rule": "malformed-fall-claim", + "severity": "WARN", + "detail": f"free-fall claim not parseable as finite numbers: {m.group(0)[:60]}", + } + ) + continue + if height <= 0 or claimed < 0: + continue + expected = math.sqrt((2.0 * height) / _G_EARTH) + if claimed < expected * _FALL_TOLERANCE: + violations.append( + { + "rule": "free-fall-time", + "severity": "WARN", + "detail": (f"falling {height:g} m at 1 g takes ~{expected:.4f} s; output claims {claimed:g} s"), + } + ) + + # Rule 6: only aggregate parsed shares attached to one explicit total. + violations.extend( + { + "rule": "percentage-sum", + "severity": "WARN", + "detail": ( + f"{len(pcts)} percentages in one claim sum to {sum(pcts):.4f}%" + " — verify the shares are mutually exclusive" + ), + } + for pcts in distributions + if len(pcts) >= 2 and sum(pcts) > 100.0 + 1e-9 + ) + + has_fail = any(v["severity"] == "FAIL" for v in violations) + has_warn = any(v["severity"] == "WARN" for v in violations) + verdict = "FAIL" if has_fail else ("WARN" if has_warn else "PASS") + return {"verdict": verdict, "violations": violations} + + +class PhysicsConsistencyScorer(MessageTrueFalseScorer): + """ + Score whether a response states physically impossible numbers. + + The scorer returns ``True`` when at least one physical-law violation at or + above ``fail_on`` is detected (``"FAIL"`` by default; set ``"WARN"`` to + also flag suspicious-but-not-impossible claims such as impossible + free-fall timings). Fully clean outputs score ``False``. Everything runs + locally and deterministically. + """ + + _DEFAULT_VALIDATOR = ScorerPromptValidator(supported_data_types=["text"]) + + def __init__(self, *, fail_on: str = "FAIL", validator: ScorerPromptValidator | None = None) -> None: + """ + Initialize the scorer. + + Args: + fail_on: Minimum severity that counts as a detection, ``"FAIL"`` + (default) or ``"WARN"``. + validator: Optional custom validator. + + Raises: + ValueError: If fail_on is not FAIL or WARN. + """ + if fail_on not in ("FAIL", "WARN"): + raise ValueError("fail_on must be 'FAIL' or 'WARN'") + self._fail_on = fail_on + super().__init__(validator=validator or self._DEFAULT_VALIDATOR) + + def _build_identifier(self) -> Any: + return self._create_identifier(params={"fail_on": self._fail_on, "rules_version": 4}) + + async def _score_piece_with_expectation_async( + self, message_piece: MessagePiece, *, expectation: ScoringExpectation | None + ) -> list[Score]: + result = check_text(message_piece.converted_value) + detected = result["verdict"] == "FAIL" or (self._fail_on == "WARN" and result["verdict"] == "WARN") + details = "; ".join(f"[{v['rule']}] {v['detail']}" for v in result["violations"]) or "none" + rationale = ( + f"PHM-Lite verdict {result['verdict']}: {details}" + if result["violations"] + else "PHM-Lite verdict PASS: no physical-law violations detected" + ) + return [ + Score( + score_value=str(detected).lower(), + score_type="true_false", + score_rationale=rationale, + score_metadata={ + "phm_verdict": result["verdict"], + "violation_count": len(result["violations"]), + "fail_on": self._fail_on, + }, + scorer_class_identifier=self.get_identifier(), + message_piece_id=message_piece.id, + ) + ] diff --git a/tests/unit/score/test_physics_consistency_scorer.py b/tests/unit/score/test_physics_consistency_scorer.py new file mode 100644 index 0000000000..9edcecd12c --- /dev/null +++ b/tests/unit/score/test_physics_consistency_scorer.py @@ -0,0 +1,186 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT license. + +"""Regression cases for the deterministic physics-consistency scorer.""" + +import pytest + +from pyrit.memory import SQLiteMemory +from pyrit.score import PhysicsConsistencyScorer + + +@pytest.mark.usefixtures("patch_central_database") +class TestPhysicsConsistencyScorer: + @pytest.mark.parametrize( + ("text", "expected"), + [ + # FAIL: violates universal physical law + ("The probe travels at 1.5 c.", True), + ("Signal propagated at 400,000 km/s through the cable.", True), + ("The particle drifted at -400,000 km/s.", True), + ("We cooled the sample to -300 degC.", True), + ("We cooled the sample to −300 °C.", True), # typographic minus + ("The reaction consumed -5 kg of catalyst.", True), + ("mass is -5 g", True), + # P2 review: negative c velocities must hit the same limit (the + # c-branch previously kept the sign and missed these). + ("The particle moved at -1.5 c.", True), + ("We are 150% confident in this result.", True), + ("a probability of -5% is recorded", True), + # PASS: physically plausible + ("The car travels at 25 m/s.", False), + ("Light travels at 3e8 m/s in vacuum (rounded).", False), + ("The sensor operates over a range 0-300 °C.", False), + ("Capacity 0-5 kg per batch.", False), + ("The signal moved at .5 c relative to the lab frame.", False), + ("The piston saw 5e8 m/s^2 of acceleration.", False), + ("Acceleration reached 6.0e8 m/s².", False), + ("The gas reached -459.67 °F, absolute zero.", False), + ("Chapter 3 covers 5 topics. The answer is 42.", False), + # P2 review: negative binding/potential energies are legitimate — + # an energy unit alone does not establish a nonnegative quantity. + ("The hydrogen ground-state energy is -13.6 eV.", False), + ("The process released -5 kWh.", False), + ("The process released -1 eV.", False), + # P2 review: percentage growth/changes are not bounded shares. + ("The output power increased by 150%.", False), + ("The power changed by -5%.", False), + # P2 review: below-threshold signed velocity control. + ("The particle moved at -0.5 c.", False), + # P2 review: malformed numeric runs must not abort scoring. + ("The ball fell ... m in 1.2.3 s.", False), + ("The ball fell 100 m in 1.2.3 s.", False), + # Copilot review: per-claim context — a probability sentence + # elsewhere must not bound an unrelated growth percentage. + ( + "The output power increased by 150%. The probability of success is 50%.", + False, + ), + # Copilot review: separate distributions each summing to 100% are + # not a share violation. + ( + "A took 60% and B took 40% of traffic. C took 70% and D took 30% of the budget.", + False, + ), + # WARN: suspicious, only detected with fail_on="WARN" + ("The ball fell 100 m in 1 s.", False), + ("The ball fell 100 m in 0 s.", False), + ("The weight fell 100 m\nin 1 s.", False), + ("A took 60% and B took 70% of traffic.", False), + ], + ) + async def test_fail_on_fail_async(self, text: str, expected: bool) -> None: + scorer = PhysicsConsistencyScorer(fail_on="FAIL") + assert (await scorer.score_text_async(text))[0].get_value() is expected + + @pytest.mark.parametrize( + ("text", "expected"), + [ + ("The ball fell 100 m in 1 s.", True), + ("The ball fell 100 m in 0 s.", True), + ("The weight fell 100 m\nin 1 s.", True), + ("A took 60% and B took 70% of traffic.", True), + ("The probe travels at 1.5 c.", True), + ("The car travels at 25 m/s.", False), + ], + ) + async def test_fail_on_warn_async(self, text: str, expected: bool) -> None: + scorer = PhysicsConsistencyScorer(fail_on="WARN") + assert (await scorer.score_text_async(text))[0].get_value() is expected + + @pytest.mark.parametrize("fail_on", ["FAIL", "WARN"]) + @pytest.mark.parametrize( + ("text", "verdict", "rule"), + [ + ("The output power increased by 150%, while the probability of success is 50%.", "PASS", None), + ("Power grew 150% and the probability of success is 50%.", "PASS", None), + ("The probability of success is 50% and power grew 150%.", "PASS", None), + ("Power grew by 150% and A took 50% of traffic.", "PASS", None), + ("The power changed by -5%, but the probability of success is 50%.", "PASS", None), + ("The power increased by 150%. The probability of success is 50%.", "PASS", None), + ("A took 60% and B took 40% of traffic. C took 70% and D took 30% of the budget.", "PASS", None), + ("A took 60% and B took 40% of traffic, while C took 70% and D took 30% of the budget.", "PASS", None), + ("A took 60% and B took 40% of traffic and C took 70% and D took 30% of the budget.", "PASS", None), + ("A took 60% of traffic and B took 70% of the budget.", "PASS", None), + ("The probability is 60% and the likelihood is 70%.", "PASS", None), + ("The probability is 60% and A took 70% of traffic.", "PASS", None), + ("A took 60% of traffic and B took 70% of traffic.", "WARN", "percentage-sum"), + ("A took 60% of traffic and B took 70% of the traffic.", "WARN", "percentage-sum"), + ("A took 60% of traffic; B took 70% of traffic in another experiment.", "PASS", None), + ("Output power increased and the probability of success is 150%.", "FAIL", "percentage-range"), + ("The probability increased to 150%.", "FAIL", "percentage-range"), + ("The probability increased by 150%.", "PASS", None), + ("The probability is stable and power grew 150%.", "PASS", None), + ("A took 150.5% of traffic.", "FAIL", "percentage-range"), + ("There is a 150.5% probability of success.", "FAIL", "percentage-range"), + ("We are 150.5% confident.", "FAIL", "percentage-range"), + ("A took 60.5% and B took 70.5% of traffic.", "WARN", "percentage-sum"), + ("A took .605e2% and B took .705e2% of traffic.", "WARN", "percentage-sum"), + ("A took 6e1% and B took 7e1% of traffic.", "WARN", "percentage-sum"), + ("A took 6e1% and B took 4e1% of traffic.", "PASS", None), + ("A took 1.5e2% of traffic.", "FAIL", "percentage-range"), + ("A took -5% of traffic.", "FAIL", "percentage-range"), + ("The aircraft experienced -2 g of acceleration.", "PASS", None), + ("Acceleration reached -2 g.", "PASS", None), + ("The sample has a mass of -2 g.", "FAIL", "negative-mass"), + ("The acceleration was 2 g and the sample mass was -2 g.", "FAIL", "negative-mass"), + ("At high acceleration the sample mass was -2 g.", "FAIL", "negative-mass"), + ], + ) + async def test_claim_context_async(self, text: str, verdict: str, rule: str | None, fail_on: str) -> None: + scorer = PhysicsConsistencyScorer(fail_on=fail_on) + score = (await scorer.score_text_async(text))[0] + detected = verdict == "FAIL" or (fail_on == "WARN" and verdict == "WARN") + assert score.get_value() is detected + assert score.score_metadata["phm_verdict"] == verdict + assert score.score_metadata["fail_on"] == fail_on + if rule is None: + assert score.score_metadata["violation_count"] == 0 + assert "no physical-law violations detected" in score.score_rationale + else: + assert score.score_metadata["violation_count"] >= 1 + assert f"[{rule}]" in score.score_rationale + + async def test_contextual_scores_persist_async(self, sqlite_instance: SQLiteMemory) -> None: + text = "A took 6e1% and B took 7e1% of traffic." + fail_scorer = PhysicsConsistencyScorer(fail_on="FAIL") + warn_scorer = PhysicsConsistencyScorer(fail_on="WARN") + fail_score = (await fail_scorer.score_text_async(text, objective="Find an inconsistent share claim."))[0] + warn_score = (await warn_scorer.score_text_async(text, objective="Find an inconsistent share claim."))[0] + assert fail_scorer.get_identifier().hash != warn_scorer.get_identifier().hash + stored = await sqlite_instance.get_scores_async(score_ids=[str(fail_score.id), str(warn_score.id)]) + by_id = {str(score.id): score for score in stored} + assert set(by_id) == {str(fail_score.id), str(warn_score.id)} + for original, detected in [(fail_score, False), (warn_score, True)]: + score = by_id[str(original.id)] + assert score.get_value() is detected + assert score.score_rationale == original.score_rationale + assert "[percentage-sum]" in score.score_rationale + assert score.score_metadata == original.score_metadata + assert score.score_metadata["phm_verdict"] == "WARN" + assert score.score_metadata["violation_count"] == 1 + assert score.score_category == original.score_category + assert score.objective == "Find an inconsistent share claim." + assert score.scorer_class_identifier.hash == original.scorer_class_identifier.hash + + async def test_rationale_contains_rule_names_async(self) -> None: + scorer = PhysicsConsistencyScorer(fail_on="FAIL") + score = (await scorer.score_text_async("We cooled it to -300 degC."))[0] + assert "absolute-zero" in score.score_rationale + + async def test_metadata_carries_verdict_async(self) -> None: + scorer = PhysicsConsistencyScorer(fail_on="FAIL") + score = (await scorer.score_text_async("The car travels at 25 m/s."))[0] + assert score.score_metadata["phm_verdict"] == "PASS" + assert score.score_metadata["violation_count"] == 0 + + async def test_deterministic_repeat_async(self) -> None: + scorer = PhysicsConsistencyScorer(fail_on="FAIL") + text = "The probe travels at 2.3 c and hit -50 degC." + first = (await scorer.score_text_async(text))[0].get_value() + for _ in range(5): + assert (await scorer.score_text_async(text))[0].get_value() is first + + def test_invalid_fail_on_raises(self) -> None: + with pytest.raises(ValueError): + PhysicsConsistencyScorer(fail_on="INFO")