fix(webfetch): keep targeted evidence refreshes bounded

harness/record_io.py

@@ -1,6 +1,6 @@
 """UTF-8 JSONL framing shared by execution, certification and report readers."""
-import json
 import hashlib
+import json
 from pathlib import Path
 
 
@@ -57,7 +57,14 @@ def parse_jsonl(raw: bytes | str, *, skip_blank=False, allow_incomplete_tail=Fal
 
 def load_jsonl(path, *, skip_blank=False):
     """Read a finalized file; every non-skipped record must be valid."""
-    return parse_jsonl(path.read_bytes(), skip_blank=skip_blank)
+    records = []
+    with Path(path).open('rb') as source:
+        for line in source:
+            line = line.removesuffix(b'\n')
+            if skip_blank and not line.strip():
+                continue
+            records.append(json.loads(line.decode('utf-8')))
+    return records
 
 
 def read_completed_records(path):

harness/web_fetch/assembly.py

@@ -235,9 +235,8 @@ def _selected_provider_replacement_rows(run, root, rows, receipt_path):
             raise ValueError('Invalid-call correction changed the provider binary')
         if item['category'] == 'implementation_correction':
             case = baseline_cases.get(item['case_id'])
-            if case is None or task_passed(
-                    bool(prior.get('ok')), prior.get('metrics', {}),
-                    case.get('task_kind', 'body')):
+            if (case is None or (prior.get('ok') is not False and task_passed(
+                    True, prior.get('metrics', {}), case.get('task_kind', 'body')))):
                 raise ValueError(
                     'Implementation correction baseline machine verdict is not a failure')
         seen.add(pair)

harness/web_fetch/reference_lifecycle.py

@@ -939,6 +939,16 @@ def finalize_reference_checks(run_dir: Path, observations_path: Path, now=None)
 
 
 def valid_scoring_revision(source) -> bool:
+    """Validate the receipt chain and its independent result reviews."""
+    return _valid_scoring_revision(source, validate_reviews=True)
+
+
+def valid_scoring_revision_chain(source) -> bool:
+    """Validate receipts and review bindings without re-entering verdict validation."""
+    return _valid_scoring_revision(source, validate_reviews=False)
+
+
+def _valid_scoring_revision(source, *, validate_reviews) -> bool:
     """Return true only for a continuous receipt chain ending at current evidence."""
     if source is None:
         return False
@@ -975,7 +985,12 @@ def valid_scoring_revision(source) -> bool:
             try:
                 raw = candidate.read_bytes()
                 review = json.loads(raw)
-                require_blind_result_review(root, dataset, records, review)
+                if validate_reviews:
+                    require_blind_result_review(root, dataset, records, review)
+                elif (review.get('schema_version') != 1
+                      or review.get('bindings') != {'dataset_sha256': current_dataset,
+                                                     'results_sha256': current_results}):
+                    continue
                 blind_review_hashes.add(sha256(raw))
             except (OSError, ValueError, TypeError, KeyError, StopIteration):
                 continue
@@ -985,7 +1000,8 @@ def valid_scoring_revision(source) -> bool:
             if dataset is None or records is None:
                 dataset = json.loads((root/'dataset.json').read_text())
                 records = load_jsonl(root/'results.jsonl')
-            assembled_review_valid = require_result_review(root, dataset, records) is not None
+            assembled_review_valid = (require_result_review(root, dataset, records) is not None
+                                      if validate_reviews else review_hash in blind_review_hashes)
         except (OSError, ValueError, TypeError, KeyError, StopIteration):
             assembled_review_valid = False
 

harness/web_fetch/result_review.py

@@ -5,7 +5,7 @@
 import tempfile
 from pathlib import Path
 
-from harness.record_io import parse_jsonl
+from harness.record_io import file_sha256, load_jsonl, parse_jsonl
 
 from .anomaly_review import require_anomaly_review
 from lexbench_webfetch.metrics import task_passed
@@ -51,7 +51,7 @@ def review_raw_artifacts(record):
 
 
 def _sha256(path):
-    return hashlib.sha256(path.read_bytes()).hexdigest()
+    return file_sha256(path)
 
 
 def review_input_identity(record):
@@ -88,6 +88,28 @@ def review_input_identity(record):
     }
 
 
+def _machine_verdict(source, record, case, reviewed=None, *, revision_checks=None):
+    """Read current scores normally and frozen historical scores from their bound review."""
+    try:
+        return ('pass' if task_passed(
+            record['ok'], record['metrics'], case.get('task_kind', 'body')) else 'fail')
+    except ValueError as error:
+        if str(error) != (
+                'Historical scoring metrics require their frozen scorer or a complete offline rescore'):
+            raise
+        from .reference_lifecycle import valid_scoring_revision_chain
+        verdict = (reviewed or {}).get('machine_verdict')
+        if verdict not in {'pass', 'fail'}:
+            raise
+        if revision_checks is None:
+            revision_checks = {}
+        if source not in revision_checks:
+            revision_checks[source] = valid_scoring_revision_chain(source)
+        if not revision_checks[source]:
+            raise
+        return verdict
+
+
 def _require_result_review(source, dataset, records, review_document=None, *, anomaly_gate):
     required = dataset.get('comparison', {}).get('require_independent_result_review', False)
     path = source / 'result-review.json'
@@ -135,10 +157,11 @@ def _require_result_review(source, dataset, records, review_document=None, *, an
         raise ValueError(f'Result review has unresolved attempts: {len(unresolved)}')
     by_id = {record['attempt_id']: record for record in records}
     cases = {case['id']: case for case in dataset.get('cases', [])}
+    revision_checks = {}
     for row in attempts:
         record = by_id[row['attempt_id']]
-        machine = ('pass' if task_passed(record['ok'], record['metrics'],
-                   cases[record['case_id']].get('task_kind', 'body')) else 'fail')
+        machine = _machine_verdict(
+            source, record, cases[record['case_id']], reviewed=row, revision_checks=revision_checks)
         if row.get('machine_verdict') != machine:
             raise ValueError('Result review machine verdict no longer matches persisted scoring')
         if row.get('final_verdict') != machine:
@@ -289,14 +312,15 @@ def archive_superseded_result_review(path: Path, *, before_dataset: bytes,
 
 
 def _read_review_evidence(source, *, skip_blank=False):
-    dataset_bytes = contained_path(source, "dataset.json", reject_symlinks=True).read_bytes()
-    results_bytes = contained_path(source, "results.jsonl", reject_symlinks=True).read_bytes()
+    dataset_path = contained_path(source, "dataset.json", reject_symlinks=True)
+    results_path = contained_path(source, "results.jsonl", reject_symlinks=True)
+    dataset_bytes = dataset_path.read_bytes()
     return (
         json.loads(dataset_bytes),
-        parse_jsonl(results_bytes, skip_blank=skip_blank),
+        load_jsonl(results_path, skip_blank=skip_blank),
         {
             "dataset_sha256": hashlib.sha256(dataset_bytes).hexdigest(),
-            "results_sha256": hashlib.sha256(results_bytes).hexdigest(),
+            "results_sha256": file_sha256(results_path),
         },
     )
 
@@ -515,22 +539,20 @@ def merge_refresh_reviews(
         raise ValueError("Prior review must not contain accepted unreviewed passes")
 
     reusable = set()
+    revision_checks = {}
     for attempt_id, row in current.items():
         old = previous.get(attempt_id)
         case_id = row["case_id"]
-        current_machine = (
-            "pass"
-            if task_passed(
-                row["ok"], row["metrics"], current_cases[case_id].get("task_kind", "body")
-            )
-            else "fail"
-        )
+        reviewed = prior_attempts.get(attempt_id)
+        current_machine = _machine_verdict(
+            prior_evidence, row, current_cases[case_id], reviewed=reviewed, revision_checks=revision_checks)
         if (
             old is not None
             and old.get("case_id") == case_id
             and prior_cases.get(case_id) == current_cases[case_id]
             and review_input_identity(old) == review_input_identity(row)
-            and prior_attempts.get(attempt_id, {}).get("final_verdict") == current_machine
+            and reviewed is not None
+            and reviewed.get("final_verdict") == current_machine
         ):
             reusable.add(attempt_id)
     required = set(current) - reusable
@@ -581,15 +603,9 @@ def merge_refresh_reviews(
     # the current persisted score. Reviewer-visible inputs were checked above.
     normalized_attempts = []
     for record, reviewed in zip(current_records, attempts):
-        machine = (
-            "pass"
-            if task_passed(
-                record["ok"],
-                record["metrics"],
-                current_cases[record["case_id"]].get("task_kind", "body"),
-            )
-            else "fail"
-        )
+        source = prior_evidence if record['attempt_id'] in reusable else supplemental_evidence
+        machine = _machine_verdict(
+            source, record, current_cases[record['case_id']], reviewed=reviewed, revision_checks=revision_checks)
         reviewed = dict(reviewed)
         reviewed["machine_verdict"] = machine
         reviewed["final_verdict"] = machine

harness/web_fetch/safety.py

@@ -49,8 +49,13 @@ def evidence_tree_identity(root: Path) -> dict[str, str]:
     reject_evidence_symlinks(root)
     if not root.is_dir():
         raise ValueError('Evidence root must be an existing directory')
-    return {str(path.relative_to(root)): hashlib.sha256(path.read_bytes()).hexdigest()
-            for path in root.rglob('*') if path.is_file()}
+    identity = {}
+    for path in root.rglob('*'):
+        if path.is_file():
+            with path.open('rb') as stream:
+                identity[str(path.relative_to(root))] = hashlib.file_digest(
+                    stream, 'sha256').hexdigest()
+    return identity
 
 
 def stage_evidence_revision(source: Path, output: Path, operation):

tests/test_assembly_evidence.py

@@ -199,7 +199,8 @@ def test_implementation_correction_replaces_only_a_declared_failed_pair(
     (baseline/'dataset.json').write_text(json.dumps(dataset))
     (incoming/'dataset.json').write_text(json.dumps(dataset))
     prior = {'attempt_id': 'old-one', 'provider': 'p', 'case_id': 'one',
-             'ok': False, 'metrics': {'content_pass': False},
+             'ok': False,
+             'metrics': {'content_pass': False, 'forbidden_hits': []},
              'provider_metadata': {'binary_sha256': 'a'*64}}
     current = {'attempt_id': 'new-one', 'provider': 'p', 'case_id': 'one',
                'ok': selected_passes,

tests/test_record_io.py

@@ -1,7 +1,9 @@
 """JSONL records are framed by LF bytes, including during concurrent writes."""
-import json
 import hashlib
+import json
 from dataclasses import asdict
+from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 
@@ -91,6 +93,13 @@ def test_finalized_readers_preserve_unicode_and_reject_partial_records(tmp_path,
             load_jsonl(path)
 
 
+def test_finalized_reader_streams_records(tmp_path):
+    path = tmp_path / 'records.jsonl'
+    path.write_text('{"id":"one"}\n{"id":"two"}\n')
+    with patch.object(Path, 'read_bytes', side_effect=AssertionError('whole-file read')):
+        assert load_jsonl(path) == [{'id': 'one'}, {'id': 'two'}]
+
+
 def test_blank_line_policy_is_explicit_and_bare_cr_is_not_a_record_separator(tmp_path):
     path = tmp_path / 'records.jsonl'
     path.write_bytes(b'\n {"id":"one"}\r\n\t \n')

tests/test_report_inventory.py

@@ -1,10 +1,9 @@
 import copy
 import hashlib
 import json
-import runpy
+import tempfile
 import unittest
 from pathlib import Path
-from unittest.mock import patch
 
 ROOT = Path(__file__).resolve().parents[1]
 
@@ -28,35 +27,38 @@ def setUp(self):
         ]
 
     def check(self):
-        # Exercise the actual script with in-memory report files, no generated
-        # report rebuild, canonical data writes or environment/config loading.
-        root = Path('/__synthetic_inventory_report__')
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        directory = Path(temporary.name)
+        root = directory / 'report'
+        authority = directory / 'authority'
         files = {
             root/'dataset.json': json.dumps(self.dataset),
             root/'results.jsonl': '\n'.join(json.dumps(r, ensure_ascii=False) for r in self.records),
             root/'execution.json': json.dumps({'batches': [{'attempt_ids': [r['attempt_id'] for r in self.records]}]}),
             root/'index.html': '<h2>综合通过率总榜</h2> <h2>6 类任务概览</h2> 本类最高： 判分完整性 <h2>6 类分榜</h2>',
             root/'methodology/index.html': '计入耗时 不计入耗时 供应商上报的服务端耗时只作附加证据 形成可送入统一评分器的最终答案',
-            ROOT/'configs/evaluation-inventory.json': json.dumps(self.inventory),
+            authority/'configs/evaluation-inventory.json': json.dumps(self.inventory),
         }
         from harness.web_fetch.reliability import assess_reliability
         files[root/'reliability.json'] = json.dumps(assess_reliability(self.dataset,self.records))
-        read_text, read_bytes, rglob = Path.read_text, Path.read_bytes, Path.rglob
-
-        def text(path, *args, **kwargs):
-            return files[path] if path in files else read_text(path, *args, **kwargs)
-
-        def binary(path, *args, **kwargs):
-            if path in files:
-                return files[path].encode('utf-8')
-            return b'fixture' if path.is_relative_to(root) else read_bytes(path, *args, **kwargs)
-
-        def glob(path, *args, **kwargs):
-            return iter(()) if path == root else rglob(path, *args, **kwargs)
-
-        with patch.object(Path, 'read_text', text), patch.object(Path, 'read_bytes', binary), \
-                patch.object(Path, 'rglob', glob), patch('sys.argv', ['check_report.py', str(root)]):
-            runpy.run_path(str(ROOT/'scripts/check_report.py'), run_name='__main__')
+        for path, text in files.items():
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(text)
+        for record in self.records:
+            path = root / record['artifact_path']
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_bytes(b'fixture')
+        for case in self.dataset['cases']:
+            path = root / 'references' / (case['id'] + '.txt')
+            path.parent.mkdir(parents=True, exist_ok=True)
+            path.write_text(case['ground_truth_main_content'])
+        from harness.report_languages import english_notice
+        (root / 'report-languages.json').write_text(json.dumps({'primary_language': 'zh', 'languages': ['zh'], 'entrypoint': 'index.html'}))
+        (root / 'en').mkdir()
+        (root / 'en/index.html').write_text(english_notice())
+        from harness.web_fetch.report_check import main
+        main([str(root)], authority_root=authority)
 
     def test_reviewed_inventory_passes(self):
         self.check()

tests/test_result_review.py

@@ -145,6 +145,55 @@ def test_requires_exact_bound_attempt_complete_review(self):
             with self.assertRaisesRegex(ValueError,'different results'):
                 require_result_review(root,dataset,records)
 
+    def test_bound_review_preserves_historical_scorer_verdict(self):
+        from harness.web_fetch.reference_lifecycle import valid_scoring_revision, scoring_revision_history_sha256
+        from harness.web_fetch.scoring_revision import rebind_report_interpretation
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            dataset, records, review = self.fixture(root)
+            records[0]['metrics']['forbidden_hits'] = []
+            (root / 'results.jsonl').write_text(json.dumps(records[0]) + '\n')
+            receipt = {'kind': 'audited_scoring_rule_revision', 'provider_calls': 0,
+                       'changed_cases': [], 'task_decision_changes': [],
+                       'input_dataset_sha256': 'a' * 64, 'input_results_sha256': 'b' * 64,
+                       'output_dataset_sha256': hashlib.sha256((root / 'dataset.json').read_bytes()).hexdigest(),
+                       'output_results_sha256': hashlib.sha256((root / 'results.jsonl').read_bytes()).hexdigest()}
+            (root / 'scoring-revisions.json').write_text(json.dumps([receipt]))
+            self.assertTrue(valid_scoring_revision(root))
+            article = root / 'article.json'
+            article.write_text('{}')
+            rebind_report_interpretation(root, {'path': 'article.json',
+                'sha256': hashlib.sha256(article.read_bytes()).hexdigest()})
+            dataset = json.loads((root / 'dataset.json').read_text())
+            review['bindings'] = {name + '_sha256': hashlib.sha256(
+                (root / file).read_bytes()).hexdigest()
+                for name, file in [('dataset', 'dataset.json'), ('results', 'results.jsonl')]}
+            (root / 'result-review.json').write_text(json.dumps(review))
+            assembled = {**receipt, 'kind': 'audited_assembled_scoring_revision',
+                'input_dataset_sha256': receipt['output_dataset_sha256'],
+                'input_results_sha256': receipt['output_results_sha256'],
+                'source_scoring_receipts_sha256': scoring_revision_history_sha256([receipt]),
+                'independent_result_review_sha256': hashlib.sha256((root / 'result-review.json').read_bytes()).hexdigest()}
+            assembled['input_dataset_sha256'] = review['bindings']['dataset_sha256']
+            assembled['output_dataset_sha256'] = review['bindings']['dataset_sha256']
+            (root / 'scoring-revisions.json').write_text(json.dumps([receipt, assembled]))
+            self.assertTrue(valid_scoring_revision(root))
+            self.assertEqual(require_blind_result_review(root, dataset, records)['reviewed_attempts'], 1)
+            self.assertEqual(require_result_review(root, dataset, records)['reviewed_attempts'], 1)
+            review['attempts'][0]['final_verdict'] = 'fail'
+            (root / 'result-review.json').write_text(json.dumps(review))
+            self.assertFalse(valid_scoring_revision(root))
+            with self.assertRaises(ValueError):
+                require_result_review(root, dataset, records)
+            review['attempts'][0]['final_verdict'] = 'pass'
+            (root / 'result-review.json').write_text(json.dumps(review))
+            receipt['output_results_sha256'] = '0' * 64
+            (root / 'scoring-revisions.json').write_text(json.dumps([receipt]))
+            self.assertFalse(valid_scoring_revision(root))
+            with self.assertRaisesRegex(ValueError, 'Historical scoring metrics'):
+                require_blind_result_review(root, dataset, records)
+
     def test_missing_or_unresolved_review_blocks_required_report(self):
         with tempfile.TemporaryDirectory() as directory:
             root=Path(directory);dataset,records,review=self.fixture(root)