Implement submission rules. Be stricter with status at time limit

problemtools/checks/submissions.py

@@ -11,8 +11,9 @@
 from ..diagnostics import Diagnostics, pluralize
 from ..judge import SubmissionJudge, SubmissionResult, SubmissionsJudge
 from ..metadata import Metadata
-from ..model import Graders, LegacyPolicy, Submission, Submissions, TestCase, TestDataGroup
+from ..model import Graders, LegacyPolicy, Submission, Submissions, TestDataGroup
 from ..run import Program
+from ..submission_rules import FinalVerdictRule, Rule
 
 
 def check_submissions(
@@ -29,7 +30,8 @@ def check_submissions(
 ) -> None:
     """Run all checks on a problem's submissions."""
     policy = submissions.policy
-    known_submissions = _check_matches_policy(submissions, policy, diag)
+    rules_by_submission = _check_matches_policy(submissions, policy, testdata, metadata, diag)
+    known_submissions = list(rules_by_submission)
     included_submissions = [s for s in known_submissions if context.submission_filter.search(str(s.path))]
     ignored_submissions = len(known_submissions) - len(included_submissions)
     msg = f'Checking {pluralize(len(included_submissions), "submission")}'
@@ -59,9 +61,8 @@ def check_submissions(
     lower_bound_submissions = [sub for sub in included_submissions if policy.lower_bounds_time_limit(sub)]
     all_submission_results = _check_submission_group(
         lower_bound_submissions,
-        policy,
+        rules_by_submission,
         metadata,
-        testdata,
         submissions_judge,
         probdir,
         seen_oob_score_groups,
@@ -88,14 +89,16 @@ def check_submissions(
     # Run TLE submissions last (as they're presumably the slowest)
     rest = sorted(
         (sub for sub in included_submissions if sub not in lower_bound_submissions),
-        key=lambda sub: (policy.expected_verdict(sub) == 'TLE', str(sub.path)),
+        key=lambda sub: (
+            any(isinstance(r, FinalVerdictRule) and r.verdict == 'TLE' for r in rules_by_submission[sub]),
+            str(sub.path),
+        ),
     )
     all_submission_results.extend(
         _check_submission_group(
             rest,
-            policy,
+            rules_by_submission,
             metadata,
-            testdata,
             submissions_judge,
             probdir,
             seen_oob_score_groups,
@@ -115,23 +118,26 @@ def _check_has_accepted_submission(submissions: Submissions, diag: Diagnostics)
         diag.error('Require at least one "accepted" submission')
 
 
-def _check_matches_policy(submissions: Submissions, policy: LegacyPolicy, diag: Diagnostics) -> list[Submission]:
-    """Emit an error for, and exclude, any submission that doesn't match the policy at all
-    (i.e. sits in an unrecognized directory). Such a submission is never compiled or tested."""
-    matched = []
+def _check_matches_policy(
+    submissions: Submissions, policy: LegacyPolicy, testdata: TestDataGroup, metadata: Metadata, diag: Diagnostics
+) -> dict[Submission, list[Rule]]:
+    """Get the rules for every submission matching the policy. Emit an error for, and exclude, any
+    submission that doesn't match the policy at all (i.e. sits in an unrecognized directory). Such
+    a submission is never compiled or tested."""
+    rules_by_submission = {}
     for sub in submissions.submissions:
-        if policy.matches(sub):
-            matched.append(sub)
+        rules = policy.rules_for(sub, testdata, metadata)
+        if rules is not None:
+            rules_by_submission[sub] = rules
         else:
             diag.error(f'Submission {sub.path} does not match any known submissions directory; ignoring it')
-    return matched
+    return rules_by_submission
 
 
 def _check_submission_group(
     subs: list[Submission],
-    policy: LegacyPolicy,
+    rules_by_submission: dict[Submission, list[Rule]],
     metadata: Metadata,
-    testdata: TestDataGroup,
     submissions_judge: SubmissionsJudge,
     probdir: Path,
     seen_oob_score_groups: set[int],
@@ -164,7 +170,15 @@ def _check_submission_group(
 
         judge = submissions_judge.judges()[sub]
         sub_results = _check_submission(
-            sub, judge, policy, metadata, testdata, probdir, seen_oob_score_groups, timelim, timelim_high, diag
+            sub,
+            judge,
+            rules_by_submission[sub],
+            metadata,
+            probdir,
+            seen_oob_score_groups,
+            timelim,
+            timelim_high,
+            diag,
         )
         submission_results.append((sub, sub_results))
 
@@ -174,19 +188,15 @@ def _check_submission_group(
 def _check_submission(
     sub: Submission,
     judge: SubmissionJudge,
-    policy: LegacyPolicy,
+    rules: list[Rule],
     metadata: Metadata,
-    testdata: TestDataGroup,
     probdir: Path,
     seen_oob_score_groups: set[int],
     timelim: float,
     timelim_high: float,
     diag: Diagnostics,
 ) -> list[SubmissionResult]:
-    expected_verdict = policy.expected_verdict(sub)
-    assert expected_verdict is not None, '_check_submission called on a submission not matching the policy'
     partial = sub.directory == 'partially_accepted'
-    desc = str(sub)
 
     results_high = judge.judge(timelim_high)
     if not results_high:
@@ -202,39 +212,20 @@ def _check_submission(
             if r.score is not None and isinstance(r.test_node, TestDataGroup):
                 _check_score_in_bounds(r.test_node, sub.program, r.score, probdir, seen_oob_score_groups, diag)
 
-    # Warn if AC (but not PAC) submissions fail on samples. It's not uncommon for sample cases to be
-    # ignored, so failing on them could be silent otherwise. Skip warning if the result isn't AC -
-    # then something worse has gone wrong, and we'll error later.
-    if expected_verdict == 'AC' and not partial and result.verdict == 'AC':
-        if sample_failure := _find_sample_failure(results):
-            diag.warning(f'{desc} got {sample_failure.verdict} on sample: {sample_failure}')
-
     # Warn if a PAC submission would affect time limit, had it been use to compute the time limit. Only do this
     # if it gets AC on the computed time limit, otherwise we have other warnings below.
     if partial and result.verdict == 'AC':
-        _warn_pac_too_slow(judge, results, timelim, desc, metadata, diag)
+        _warn_pac_too_slow(judge, results, timelim, sub, metadata, diag)
 
     if result.verdict != result_high.verdict or result.score != result_high.score:
         diag.warning(
-            f'{desc} sensitive to time limit: limit of {timelim} secs -> {result}, limit of {timelim_high} secs -> {result_high}'
+            f'{sub} sensitive to time limit: limit of {timelim} secs -> {result}, limit of {timelim_high} secs -> {result_high}'
         )
 
-    if partial and _fully_accepted(result, testdata, metadata):
-        diag.warning(f'{desc} was fully accepted: {result}')
-    elif result.verdict == expected_verdict:
-        diag.msg(f'   {desc} OK: {result}')
-        if (
-            not partial
-            and expected_verdict == 'AC'
-            and not _fully_accepted(result, testdata, metadata)
-            and _full_score_finite(testdata, metadata)
-        ):
-            # For some heuristic problems, this is expected. Thus, only warn.
-            diag.warning(f'{desc} did not attain full score (consider moving it to partially_accepted)')
-    elif result_high.verdict == expected_verdict and not (partial and _fully_accepted(result_high, testdata, metadata)):
-        diag.msg(f'   {desc} OK with extra time: {result_high}')
-    else:
-        diag.error(f'{desc} got {result}', result_high.additional_info)
+    # Use a list rather than any() on a generator, so that every rule gets checked
+    reported_errors = [rule.check(sub, results, diag) for rule in rules]
+    if not any(reported_errors):
+        diag.msg(f'   {sub} OK: {result}')
 
     return results
 
@@ -259,23 +250,21 @@ def _check_score_in_bounds(
     )
 
 
-def _find_sample_failure(results: list[SubmissionResult]) -> SubmissionResult | None:
-    for r in results:
-        if r.verdict != 'AC' and isinstance(r.test_node, TestCase) and r.test_node.is_in_sample_group():
-            return r
-    return None
-
-
 def _warn_pac_too_slow(
-    judge: SubmissionJudge, results: list[SubmissionResult], timelim: float, desc: str, metadata: Metadata, diag: Diagnostics
+    judge: SubmissionJudge,
+    results: list[SubmissionResult],
+    timelim: float,
+    sub: Submission,
+    metadata: Metadata,
+    diag: Diagnostics,
 ) -> None:
     """Warn if a PAC submission is slow enough that it would have affected the time limit."""
     runtime_without_affecting_tl = timelim / metadata.limits.time_multipliers.ac_to_time_limit
     if judge.judge(runtime_without_affecting_tl)[-1].verdict == 'AC':
         return
     for t in sorted(r.runtime for r in results if r.runtime > runtime_without_affecting_tl):
         if judge.judge(t)[-1].verdict == 'AC':
-            diag.warning(f'{desc} is slower than all AC submissions. It needs {t:.2f}s to get AC')
+            diag.warning(f'{sub} is slower than all AC submissions. It needs {t:.2f}s to get AC')
             return
 
 
@@ -399,20 +388,6 @@ def _compute_time_limit(
     return timelim, timelim_high
 
 
-def _full_score_finite(testdata: TestDataGroup, metadata: Metadata) -> bool:
-    min_score, max_score = testdata.get_score_range()
-    if metadata.legacy_grading.objective == 'min':
-        return min_score != float('-inf')
-    else:
-        return max_score != float('inf')
-
-
-def _fully_accepted(result: SubmissionResult, testdata: TestDataGroup, metadata: Metadata) -> bool:
-    min_score, max_score = testdata.get_score_range()
-    best_score = min_score if metadata.legacy_grading.objective == 'min' else max_score
-    return result.verdict == 'AC' and (not metadata.is_scoring() or result.score == best_score)
-
-
 def _fmt_number(number: float | None) -> str:
     """Format a number with at most 3 decimals, dealing with None."""
     return f'{round(number, 3):g}' if number is not None else '-'

problemtools/judge/result.py

@@ -1,7 +1,14 @@
 from __future__ import annotations
 
+from collections.abc import Collection
+from typing import Literal
+
 from ..model import TestCase, TestDataGroup
 
+#: The optional parts of a formatted SubmissionResult.
+ResultField = Literal['score', 'reason', 'testcase', 'runtime']
+ALL_RESULT_FIELDS: tuple[ResultField, ...] = ('score', 'reason', 'testcase', 'runtime')
+
 
 class SubmissionResult:
     def __init__(
@@ -20,15 +27,20 @@ def __init__(
         self.runtime = -1.0
         self.validator_first = False  # Needed to work around interactive giving unreliable runtime on WA
 
-    def __str__(self) -> str:
+    def format(self, fields: Collection[ResultField] = ALL_RESULT_FIELDS) -> str:
+        """Format the verdict, followed by whichever of `fields` are available, e.g.
+        'AC (50) [CPU: 0.12s @ testcase secret/3]' or 'RTE [SIGABRT, testcase: testcase secret/1]'."""
         verdict = self.verdict
+        if 'score' in fields and verdict == 'AC' and self.score is not None:
+            verdict += f' ({self.score:g})'
         details = []
-        if verdict == 'AC' and self.score is not None:
-            verdict += f' ({self.score:.0f})'
-        if self.reason is not None:
+        if 'reason' in fields and self.reason is not None:
             details.append(self.reason)
-        if isinstance(self.test_node, TestCase):
+        if 'testcase' in fields and isinstance(self.test_node, TestCase):
             details.append(f'testcase: {self.test_node}')
-        if self.runtime != -1:
+        if 'runtime' in fields and self.runtime != -1:
             details.append(f'CPU: {self.runtime:.2f}s @ {self.runtime_testcase}')
         return verdict if not details else f'{verdict} [{", ".join(details)}]'
+
+    def __str__(self) -> str:
+        return self.format()

problemtools/model/submissions.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
+import math
 from dataclasses import dataclass, field
 from pathlib import Path
 
 from ..languages import Languages
+from ..metadata import Metadata
 from ..run import Program, find_programs
-from . import Verdict
+from ..submission_rules import FinalVerdictRule, PermittedRule, Rule, ScoreRule
 from .includes import Includes
 from .paths import RelativePath, relpath, resolve
+from .testdata import TestDataGroup
 
 
 @dataclass(frozen=True)
@@ -33,41 +36,63 @@ def __str__(self) -> str:
         return f'{self.directory}/{self.program}'
 
 
-_VERDICT_BY_DIRECTORY: dict[str, Verdict] = {
-    'accepted': 'AC',
-    'partially_accepted': 'AC',
-    'wrong_answer': 'WA',
-    'run_time_error': 'RTE',
-    'time_limit_exceeded': 'TLE',
-}
-
-
 @dataclass(frozen=True)
 class LegacyPolicy:
     """The directory-name-based policy for what's expected of a submission, used by problem
     formats that predate submissions.yaml: everything is inferred purely from which of the
     well-known directories (if any) a submission's path starts with.
 
-    This is a placeholder for the richer (and eventually per-testcase) policy that
-    submissions.yaml will bring.
+    Problemtools has traditionally only checked a submission's final verdict, so that's an
+    error, while violating the (stricter) per test case expectations is only a warning.
     """
 
-    def matches(self, submission: Submission) -> bool:
-        """Whether this submission is recognized by the policy at all, i.e. sits in a well-known directory."""
-        return self.expected_verdict(submission) is not None
-
-    def expected_verdict(self, submission: Submission) -> Verdict | None:
-        """The expected verdict for this submission, or None if it isn't in a well-known directory."""
-        return _VERDICT_BY_DIRECTORY.get(submission.directory)
+    def rules_for(self, submission: Submission, testdata: TestDataGroup, metadata: Metadata) -> list[Rule] | None:
+        """The rules for this submission, or None if it isn't in a well-known directory."""
+        match submission.directory:
+            case 'accepted':
+                # show_details=False: when the final verdict is wrong, the permitted rule shows the failing test case
+                rules: list[Rule] = [
+                    PermittedRule(verdicts=frozenset({'AC'}), is_error=False),
+                    FinalVerdictRule(verdict='AC', show_details=False),
+                ]
+                if metadata.is_scoring():
+                    best = _best_score(testdata, metadata)
+                    # For some heuristic problems, not attaining full score is expected. Thus, only warn.
+                    if math.isfinite(best):
+                        hint = 'if full score is attainable, consider moving it to partially_accepted'
+                        rules.append(ScoreRule(min_score=best, max_score=best, is_error=False, hint=hint))
+                return rules
+            case 'partially_accepted':
+                rules = [FinalVerdictRule(verdict='AC')]
+                if metadata.is_scoring():
+                    min_score, max_score = testdata.get_score_range()
+                    hint = 'consider moving it to accepted'
+                    # If the best score is infinite, so is the bound (inf - 0.01 == inf), and the rule accepts any score
+                    if metadata.legacy_grading.objective == 'min':
+                        rules.append(ScoreRule(min_score=min_score + 0.01, max_score=max_score, is_error=False, hint=hint))
+                    else:
+                        rules.append(ScoreRule(min_score=min_score, max_score=max_score - 0.01, is_error=False, hint=hint))
+                return rules
+            case 'wrong_answer':
+                return [
+                    PermittedRule(verdicts=frozenset({'AC', 'WA'}), is_error=False),
+                    FinalVerdictRule(verdict='WA', show_details=False),
+                ]
+            case 'time_limit_exceeded':
+                return [
+                    PermittedRule(verdicts=frozenset({'AC', 'WA', 'TLE'}), is_error=False),
+                    # Details are needed when the final verdict is WA, which the permitted rule allows
+                    FinalVerdictRule(verdict='TLE'),
+                ]
+            case 'run_time_error':
+                return [FinalVerdictRule(verdict='RTE')]
+            case _:
+                return None
 
     def lower_bounds_time_limit(self, submission: Submission) -> bool:
         """Whether this submission's runtime should be used to lower-bound the time limit."""
         return submission.directory == 'accepted'
 
-    def expects_full_score(self, submission: Submission) -> bool:
-        """Whether this submission is expected to achieve full score."""
-        return submission.directory == 'accepted'
-
 
 @dataclass(frozen=True)
 class Submissions:
@@ -88,3 +113,8 @@ def load_submissions(probdir: Path, language_config: Languages, includes: Includ
             for program in find_programs(str(entry), language_config=language_config, includes=includes):
                 submissions.append(Submission(program=program, path=relpath(Path(entry.name) / program.name)))
     return Submissions(submissions=submissions)
+
+
+def _best_score(testdata: TestDataGroup, metadata: Metadata) -> float:
+    min_score, max_score = testdata.get_score_range()
+    return min_score if metadata.legacy_grading.objective == 'min' else max_score