FEAT: repeat program cases and leave disagreeing runs undetermined

doc/code/scoring/7_program_execution_scorer.ipynb

@@ -111,7 +111,12 @@
     "## Replay without another execution\n",
     "\n",
     "Replay uses the saved outputs, not the Docker client. Expected results may change, but the\n",
-    "runtime profile name and invocation inputs must still match the saved run."
+    "runtime profile name and invocation inputs must still match the saved run.\n",
+    "\n",
+    "Replay repeats judgment, not execution, so a stored run cannot show that a program is\n",
+    "deterministic. To check, set `repetitions` on `ProgramBehaves`: each case runs that many times,\n",
+    "and a case whose runs reach different verdicts is undetermined. Output a case does not check\n",
+    "never counts as a difference."
    ]
   },
   {

doc/code/scoring/7_program_execution_scorer.py

@@ -76,6 +76,11 @@
 #
 # Replay uses the saved outputs, not the Docker client. Expected results may change, but the
 # runtime profile name and invocation inputs must still match the saved run.
+#
+# Replay repeats judgment, not execution, so a stored run cannot show that a program is
+# deterministic. To check, set `repetitions` on `ProgramBehaves`: each case runs that many times,
+# and a case whose runs reach different verdicts is undetermined. Output a case does not check
+# never counts as a difference.
 
 # %%
 memory = CentralMemory.get_memory_instance()

pyrit/models/score/condition.py

@@ -276,11 +276,20 @@ class ProgramBehaves(Condition):
     row. A scorer splits them before acquisition: the runtime profile and each invocation go to
     the source, and each expected result stays with the scorer's matcher. A run that times out
     or exceeds a resource limit fails its case.
+
+    ``repetitions`` runs each case that many times in one acquisition. A case is settled only when
+    every run reaches the same verdict; runs that disagree leave it undetermined. Only a difference
+    that changes a case's verdict counts as nondeterminism, such as output that depends on the
+    clock, randomness, or scheduling. Output the case does not check, and duration, never count.
+    Repetitions share one runtime profile, so they do not sample other platforms or architectures;
+    compare those under separate runtime profiles. Replay judges the stored runs again and never
+    repeats execution.
     """
 
     condition_type: Literal["program_behaves"] = "program_behaves"
     runtime_profile_name: _RuntimeProfileName
     cases: tuple[ProgramCase, ...]
+    repetitions: int = Field(default=1, ge=1)
 
     @model_validator(mode="after")
     def _validate_cases(self) -> ProgramBehaves:
@@ -299,8 +308,8 @@ def _validate_cases(self) -> ProgramBehaves:
 
     @property
     def invocations(self) -> tuple[_ProgramInvocation, ...]:
-        """The stimuli of every case, in case order, without their expected results."""
-        return tuple(case.invocation for case in self.cases)
+        """The stimuli of every case, in case order and repeated per case, without their expected results."""
+        return tuple(case.invocation for case in self.cases for _ in range(self.repetitions))
 
 
 def _parse_conditions(value: Any) -> Any:

pyrit/score/true_false/program_execution_scorer.py

@@ -87,6 +87,25 @@ def _evaluate_case(*, expected: ProgramResult, event: _ProgramExecutionEvent | N
     return _Evaluation(value=True, rationale="")
 
 
+def _evaluate_runs(*, expected: ProgramResult, events: list[_ProgramExecutionEvent | None]) -> _Evaluation:
+    """
+    Settle a repeated case only when every run reaches the same verdict.
+
+    Returns:
+        _Evaluation: The shared verdict, or an undetermined one when runs are missing or disagree.
+    """
+    evaluations = [_evaluate_case(expected=expected, event=event) for event in events]
+    values = {evaluation.value for evaluation in evaluations}
+    if len(values) == 1:
+        return evaluations[0]
+    if {True, False} <= values:
+        failed = sum(evaluation.value is False for evaluation in evaluations)
+        return _Evaluation(
+            value=None, rationale=f"gave different results across runs ({failed} of {len(events)} failed)"
+        )
+    return next(evaluation for evaluation in evaluations if evaluation.value is None)
+
+
 def _evaluate_program(
     *, condition: ProgramBehaves, payload: ProgramExecutionObservationPayload, acquisition: Acquisition
 ) -> _Evaluation:
@@ -103,12 +122,16 @@ def _evaluate_program(
             value=None, rationale=f"The program's behavior under {profile_name} could not be observed ({gaps})."
         )
     events = {event.invocation_index: event for event in payload.events}
+    runs = condition.repetitions
     evaluations = [
-        _evaluate_case(expected=case.expected, event=events.get(index)) for index, case in enumerate(condition.cases)
+        _evaluate_runs(expected=case.expected, events=[events.get(index * runs + run) for run in range(runs)])
+        for index, case in enumerate(condition.cases)
     ]
     if all(evaluation.value is True for evaluation in evaluations):
+        each = f" in each of {runs} runs" if runs > 1 else ""
         return _Evaluation(
-            value=True, rationale=f"All {len(evaluations)} case(s) produced the expected result under {profile_name}."
+            value=True,
+            rationale=f"All {len(evaluations)} case(s) produced the expected result{each} under {profile_name}.",
         )
     summary = "; ".join(
         f"case {index} {evaluation.rationale}"
@@ -134,7 +157,8 @@ class ProgramExecutionScorer(TrueFalseScorer):
     The source receives verbatim program text, a runtime profile name, and invocation inputs,
     but never expected results. Each case checks its expected exit and output; a timeout or
     resource limit fails the case. All cases passing yields true, any proven failure yields
-    false, and insufficient evidence leaves the verdict undetermined.
+    false, and insufficient evidence leaves the verdict undetermined. A repeated case whose runs
+    disagree is insufficient evidence.
 
     Every scoring call runs afresh. ``score_observation_async`` replays saved evidence without
     executing the program.

tests/unit/models/test_program.py

@@ -99,6 +99,31 @@ def test_program_behaves_invocations_drop_expected_results() -> None:
     assert condition.invocations == (_ProgramInvocation(arguments=("1",)), _ProgramInvocation(stdin="in"))
 
 
+def test_program_behaves_repeats_each_case_in_case_order() -> None:
+    first, second = _ProgramInvocation(arguments=("1",)), _ProgramInvocation(arguments=("2",))
+    condition = ProgramBehaves(
+        runtime_profile_name="python-3.12",
+        cases=(ProgramCase(invocation=first), ProgramCase(invocation=second)),
+        repetitions=3,
+    )
+    assert condition.invocations == (first, first, first, second, second, second)
+    assert Condition.model_validate(condition.model_dump(mode="json")) == condition
+
+
+def test_program_behaves_runs_each_case_once_by_default() -> None:
+    condition = ProgramBehaves(runtime_profile_name="python-3.12", cases=(ProgramCase(),))
+    assert condition.repetitions == 1
+    stored = condition.model_dump(mode="json")
+    del stored["repetitions"]
+    assert Condition.model_validate(stored) == condition
+
+
+@pytest.mark.parametrize("repetitions", [0, -1])
+def test_program_behaves_requires_a_positive_repetition_count(repetitions: int) -> None:
+    with pytest.raises(ValidationError, match="repetitions"):
+        ProgramBehaves(runtime_profile_name="python-3.12", cases=(ProgramCase(),), repetitions=repetitions)
+
+
 def test_program_case_defaults_expect_a_clean_exit_with_no_arguments() -> None:
     case = ProgramCase()
     assert case.invocation == _ProgramInvocation()

tests/unit/score/test_program_execution_scorer.py

@@ -167,6 +167,112 @@ def test_evaluate_program_needs_every_case() -> None:
         assert _evaluate_program(condition=condition, payload=empty, acquisition=acquisition).value is None
 
 
+@pytest.mark.parametrize(
+    ("outcomes", "result", "rationale"),
+    [
+        (("9", "9", "9"), True, "in each of 3 runs"),
+        (("8", "8", "8"), False, "case 0 output did not match"),
+        (("9", "8", "9"), None, "case 0 gave different results across runs (1 of 3 failed)"),
+        (("8", "9", "8"), None, "case 0 gave different results across runs (2 of 3 failed)"),
+        (("9", "9", None), None, "case 0 did not run"),
+        (("8", None, "8"), None, "case 0 did not run"),
+    ],
+)
+def test_evaluate_program_settles_a_repeated_case_only_when_every_run_agrees(
+    outcomes: tuple[str | None, ...], result: bool | None, rationale: str
+) -> None:
+    condition = ProgramBehaves(
+        runtime_profile_name=_PROFILE, cases=(_case("3", stdout=Equals(value="9")),), repetitions=3
+    )
+    events = tuple(
+        _event(invocation_index=index, stdout=stdout) for index, stdout in enumerate(outcomes) if stdout is not None
+    )
+    payload = ProgramExecutionObservationPayload(program_sha256="a" * 64, events=events)
+    acquisition = Acquisition.COMPLETE if len(events) == 3 else Acquisition.PARTIAL
+
+    evaluation = _evaluate_program(condition=condition, payload=payload, acquisition=acquisition)
+
+    assert evaluation.value is result
+    assert rationale in evaluation.rationale
+
+
+@pytest.mark.parametrize(
+    ("outputs", "result", "rationale"),
+    [
+        (("9", "9", "16", "16"), True, "All 2 case(s) produced the expected result in each of 2 runs"),
+        (("9", "9", "16", "15"), None, "case 1 gave different results across runs (1 of 2 failed)"),
+    ],
+)
+def test_evaluate_program_reads_each_case_from_its_own_runs(
+    outputs: tuple[str, ...], result: bool | None, rationale: str
+) -> None:
+    condition = ProgramBehaves(runtime_profile_name=_PROFILE, cases=_SQUARES, repetitions=2)
+    payload = ProgramExecutionObservationPayload(
+        program_sha256="a" * 64,
+        events=tuple(_event(invocation_index=index, stdout=stdout) for index, stdout in enumerate(outputs)),
+    )
+
+    evaluation = _evaluate_program(condition=condition, payload=payload, acquisition=Acquisition.COMPLETE)
+
+    assert evaluation.value is result
+    assert rationale in evaluation.rationale
+    assert "case 0" not in evaluation.rationale
+
+
+class _DriftingSandbox(_FakeSandbox):
+    """Prints a different value on every run, as a program reading the clock or a random source would."""
+
+    async def execute_async(
+        self, *, program: str, runtime_profile_name: str, invocations: tuple[_ProgramInvocation, ...]
+    ) -> _SandboxRun:
+        self.calls.append((program, runtime_profile_name, invocations))
+        return _SandboxRun(
+            events=tuple(
+                _event(invocation_index=index, stdout=f"seed {index}", duration_ms=index + 1)
+                for index in range(len(invocations))
+            )
+        )
+
+
+async def test_runs_that_disagree_leave_the_verdict_undetermined_async(sqlite_instance: SQLiteMemory) -> None:
+    sandbox = _DriftingSandbox()
+    scorer = _scorer(sandbox)
+    case = _case("1", stdout=Equals(value="seed 0"))
+    repeated = ScoringExpectation(
+        conditions=(ProgramBehaves(runtime_profile_name=_PROFILE, cases=(case,), repetitions=3),)
+    )
+
+    score = (await scorer.score_async(scorable=ContentScorable(value="x"), expectation=repeated))[0]
+
+    assert (score.status, score.score_value) == (ScoreStatus.UNDETERMINED, None)
+    assert "case 0 gave different results across runs (2 of 3 failed)" in score.score_rationale
+    assert sandbox.calls[0][2] == (case.invocation,) * 3
+    single = (await scorer.score_async(scorable=ContentScorable(value="x"), expectation=_expectation(case)))[0]
+    assert single.get_value() is True
+    observation = (await sqlite_instance.get_observations_async(observation_ids=score.observation_ids))[0]
+    replayed = (await scorer.score_observation_async(observation=observation, expectation=repeated))[0]
+    assert replayed.status is ScoreStatus.UNDETERMINED
+    assert len(sandbox.calls) == 2
+    with pytest.raises(NonReplayableObservationError, match="different invocations"):
+        await scorer.score_observation_async(observation=observation, expectation=_expectation(case))
+
+
+async def test_differences_a_case_does_not_check_are_not_nondeterminism_async() -> None:
+    sandbox = _DriftingSandbox()
+    expectation = ScoringExpectation(
+        conditions=(
+            ProgramBehaves(
+                runtime_profile_name=_PROFILE, cases=(_case("1", stdout=Contains(value="seed")),), repetitions=3
+            ),
+        )
+    )
+
+    score = (await _scorer(sandbox).score_async(scorable=ContentScorable(value="x"), expectation=expectation))[0]
+
+    assert score.get_value() is True
+    assert "in each of 3 runs" in score.score_rationale
+
+
 async def test_scores_true_when_every_case_passes_async(sqlite_instance: SQLiteMemory) -> None:
     sandbox = _FakeSandbox(outputs={"3": {"stdout": "9"}, "4": {"stdout": "16"}})
     program = ContentScorable(value="import sys\nprint(int(sys.argv[1]) ** 2)\n")